Merge lp:~soren/nova/consolidate-locking into lp:~hudson-openstack/nova/trunk

[Soren Hansen]

Move all types of locking into utils.synchronize decorator.

Convert all uses of semaphores to use this decorator.

[Vish Ishaya (vishvananda)]

Tests run just fine and seem complete to me. Nice to have a clean decorator implementation for locking. Nice job! lgtm

review: Approve

[Rick Harris (rconradharris)]

This looks really good-- great work Soren.

Brownie points for:

* Useful inline comments
* Added test
* YAGNI on the added complexity of a race-free sync (think it was probably wise to punt on that for now)

review: Approve

[Soren Hansen (soren)]

2011/3/23 Rick Harris <email address hidden>
> * YAGNI on the added complexity of a race-free sync (think it was probably wise to punt on that for now)

I actually started working on it, but when I realised the hilarity of
implementing an rwlock that worked with Eventlet in an effort to solve
a problem that specifically does not exist under Eventlet, I stopped
:)

[OpenStack Infra (hudson-openstack)]

The attempt to merge lp:~soren/nova/consolidate-locking into lp:nova failed. Below is the output from the failed tests.

AccountsTest
    test_account_create OK
    test_account_delete OK
    test_account_update OK
    test_get_account OK
AdminAPITest
    test_admin_disabled OK
    test_admin_enabled OK
APITest
    test_exceptions_are_converted_to_faults OK
Test
    test_authorize_token OK
    test_authorize_user OK
    test_bad_token OK
    test_bad_user_bad_key OK
    test_bad_user_good_key OK
    test_no_user OK
    test_token_expiry OK
TestFunctional
    test_token_doesnotexist OK
    test_token_expiry OK
TestLimiter
    test_authorize_token OK
LimiterTest
    test_limiter_custom_max_limit OK
    test_limiter_limit_and_offset OK
    test_limiter_limit_medium OK
    test_limiter_limit_over_max OK
    test_limiter_limit_zero OK
    test_limiter_negative_limit OK
    test_limiter_negative_offset OK
    test_limiter_nothing OK
    test_limiter_offset_bad OK
    test_limiter_offset_blank OK
    test_limiter_offset_medium OK
    test_limiter_offset_over_max OK
    test_limiter_offset_zero OK
TestFaults
    test_fault_parts OK
    test_raise OK
    test_retry_header OK
FlavorsTest
    test_get_flavor_by_id OK
    test_get_flavor_list OK
    test_get_flavor_list_detail OK
GlanceImageServiceTest
    test_create OK
    test_create_and_show_non_existing_image OK
    test_delete OK
    test_update OK
ImageControllerWithGlanceServiceTest
    test_get_image_details OK
    test_get_image_index OK
    test_show_image OK
LocalImageServiceTest
    test_create...

[Soren Hansen (soren)]

/me retries

=== modified file 'nova/network/linux_net.py'
--- nova/network/linux_net.py	2011-03-23 11:21:20 +0000
+++ nova/network/linux_net.py	2011-03-23 23:52:30 +0000
@@ -21,8 +21,6 @@
 import os
 import calendar
 
-from eventlet import semaphore
-
 from nova import db
 from nova import exception
 from nova import flags
@@ -272,37 +270,30 @@
         self.ipv4['nat'].add_chain('floating-snat')
         self.ipv4['nat'].add_rule('snat', '-j $floating-snat')
 
-        self.semaphore = semaphore.Semaphore()
-
-    @utils.synchronized('iptables')
+    @utils.synchronized('iptables', external=True)
     def apply(self):
         """Apply the current in-memory set of iptables rules
 
         This will blow away any rules left over from previous runs of the
         same component of Nova, and replace them with our current set of
         rules. This happens atomically, thanks to iptables-restore.
-
-        We wrap the call in a semaphore lock, so that we don't race with
-        ourselves. In the event of a race with another component running
-        an iptables-* command at the same time, we retry up to 5 times.
         """
-        with self.semaphore:
-            s = [('iptables', self.ipv4)]
-            if FLAGS.use_ipv6:
-                s += [('ip6tables', self.ipv6)]
+        s = [('iptables', self.ipv4)]
+        if FLAGS.use_ipv6:
+            s += [('ip6tables', self.ipv6)]
 
-            for cmd, tables in s:
-                for table in tables:
-                    current_table, _ = self.execute('sudo',
-                                                    '%s-save' % (cmd,),
-                                                    '-t', '%s' % (table,),
-                                                    attempts=5)
-                    current_lines = current_table.split('\n')
-                    new_filter = self._modify_rules(current_lines,
-                                                    tables[table])
-                    self.execute('sudo', '%s-restore' % (cmd,),
-                                 process_input='\n'.join(new_filter),
-                                 attempts=5)
+        for cmd, tables in s:
+            for table in tables:
+                current_table, _ = self.execute('sudo',
+                                                '%s-save' % (cmd,),
+                                                '-t', '%s' % (table,),
+                                                attempts=5)
+                current_lines = current_table.split('\n')
+                new_filter = self._modify_rules(current_lines,
+                                                tables[table])
+                self.execute('sudo', '%s-restore' % (cmd,),
+                             process_input='\n'.join(new_filter),
+                             attempts=5)
 
     def _modify_rules(self, current_lines, table, binary=None):
         unwrapped_chains = table.unwrapped_chains

=== modified file 'nova/tests/test_misc.py'
--- nova/tests/test_misc.py	2011-03-11 08:54:08 +0000
+++ nova/tests/test_misc.py	2011-03-23 23:52:30 +0000
@@ -18,8 +18,12 @@
 import os
 import select
 
+from eventlet import greenpool
+from eventlet import greenthread
+
 from nova import test
-from nova.utils import parse_mailmap, str_dict_replace, synchronized
+from nova import utils
+from nova.utils import parse_mailmap, str_dict_replace
 
 
 class ProjectTestCase(test.TestCase):
@@ -63,7 +67,7 @@
 
 class LockTestCase(test.TestCase):
     def test_synchronized_wrapped_function_metadata(self):
-        @synchronized('whatever')
+        @utils.synchronized('whatever')
         def foo():
             """Bar"""
             pass
@@ -72,11 +76,42 @@
         self.assertEquals(foo.__name__, 'foo', "Wrapped function's name "
                                                "got mangled")
 
-    def test_synchronized(self):
+    def test_synchronized_internally(self):
+        """We can lock across multiple green threads"""
+        saved_sem_num = len(utils._semaphores)
+        seen_threads = list()
+
+        @utils.synchronized('testlock2', external=False)
+        def f(id):
+            for x in range(10):
+                seen_threads.append(id)
+                greenthread.sleep(0)
+
+        threads = []
+        pool = greenpool.GreenPool(10)
+        for i in range(10):
+            threads.append(pool.spawn(f, i))
+
+        for thread in threads:
+            thread.wait()
+
+        self.assertEquals(len(seen_threads), 100)
+        # Looking at the seen threads, split it into chunks of 10, and verify
+        # that the last 9 match the first in each chunk.
+        for i in range(10):
+            for j in range(9):
+                self.assertEquals(seen_threads[i * 10],
+                                  seen_threads[i * 10 + 1 + j])
+
+        self.assertEqual(saved_sem_num, len(utils._semaphores),
+                         "Semaphore leak detected")
+
+    def test_synchronized_externally(self):
+        """We can lock across multiple processes"""
         rpipe1, wpipe1 = os.pipe()
         rpipe2, wpipe2 = os.pipe()
 
-        @synchronized('testlock')
+        @utils.synchronized('testlock1', external=True)
         def f(rpipe, wpipe):
             try:
                 os.write(wpipe, "foo")

=== modified file 'nova/tests/test_virt.py'
--- nova/tests/test_virt.py	2011-03-14 17:59:41 +0000
+++ nova/tests/test_virt.py	2011-03-23 23:52:30 +0000
@@ -77,13 +77,11 @@
         eventlet.sleep(0)
         try:
             self.assertFalse(done2.ready())
-            self.assertTrue('fname' in conn._image_sems)
         finally:
             wait1.send()
         done1.wait()
         eventlet.sleep(0)
         self.assertTrue(done2.ready())
-        self.assertFalse('fname' in conn._image_sems)
 
     def test_different_fname_concurrency(self):
         """Ensures that two different fname caches are concurrent"""

=== modified file 'nova/utils.py'
--- nova/utils.py	2011-03-17 21:11:58 +0000
+++ nova/utils.py	2011-03-23 23:52:30 +0000
@@ -41,6 +41,7 @@
 
 from eventlet import event
 from eventlet import greenthread
+from eventlet import semaphore
 from eventlet.green import subprocess
 None
 from nova import exception
@@ -531,17 +532,76 @@
     return json.loads(s)
 
 
-def synchronized(name):
+_semaphores = {}
+
+
+class _NoopContextManager(object):
+    def __enter__(self):
+        pass
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+
+def synchronized(name, external=False):
+    """Synchronization decorator
+
+    Decorating a method like so:
+    @synchronized('mylock')
+    def foo(self, *args):
+       ...
+
+    ensures that only one thread will execute the bar method at a time.
+
+    Different methods can share the same lock:
+    @synchronized('mylock')
+    def foo(self, *args):
+       ...
+
+    @synchronized('mylock')
+    def bar(self, *args):
+       ...
+
+    This way only one of either foo or bar can be executing at a time.
+
+    The external keyword argument denotes whether this lock should work across
+    multiple processes. This means that if two different workers both run a
+    a method decorated with @synchronized('mylock', external=True), only one
+    of them will execute at a time.
+    """
+
     def wrap(f):
         @functools.wraps(f)
         def inner(*args, **kwargs):
-            LOG.debug(_("Attempting to grab %(lock)s for method "
-                        "%(method)s..." % {"lock": name,
+            # NOTE(soren): If we ever go natively threaded, this will be racy.
+            #              See http://stackoverflow.com/questions/5390569/dyn\
+            #              amically-allocating-and-destroying-mutexes
+            if name not in _semaphores:
+                _semaphores[name] = semaphore.Semaphore()
+            sem = _semaphores[name]
+            LOG.debug(_('Attempting to grab semaphore "%(lock)s" for method '
+                      '"%(method)s"...' % {"lock": name,
                                            "method": f.__name__}))
-            lock = lockfile.FileLock(os.path.join(FLAGS.lock_path,
-                                                  'nova-%s.lock' % name))
-            with lock:
-                return f(*args, **kwargs)
+            with sem:
+                if external:
+                    LOG.debug(_('Attempting to grab file lock "%(lock)s" for '
+                                'method "%(method)s"...' %
+                                {"lock": name, "method": f.__name__}))
+                    lock_file_path = os.path.join(FLAGS.lock_path,
+                                                  'nova-%s.lock' % name)
+                    lock = lockfile.FileLock(lock_file_path)
+                else:
+                    lock = _NoopContextManager()
+
+                with lock:
+                    retval = f(*args, **kwargs)
+
+            # If no-one else is waiting for it, delete it.
+            # See note about possible raciness above.
+            if not sem.balance < 1:
+                del _semaphores[name]
+
+            return retval
         return inner
     return wrap
 

=== modified file 'nova/virt/libvirt_conn.py'
--- nova/virt/libvirt_conn.py	2011-03-23 05:29:32 +0000
+++ nova/virt/libvirt_conn.py	2011-03-23 23:52:30 +0000
@@ -48,7 +48,7 @@
 
 from eventlet import greenthread
 from eventlet import tpool
-from eventlet import semaphore
+
 import IPy
 
 from nova import context
@@ -556,13 +556,12 @@
                 os.mkdir(base_dir)
             base = os.path.join(base_dir, fname)
 
-            if fname not in LibvirtConnection._image_sems:
-                LibvirtConnection._image_sems[fname] = semaphore.Semaphore()
-            with LibvirtConnection._image_sems[fname]:
+            @utils.synchronized(fname)
+            def call_if_not_exists(base, fn, *args, **kwargs):
                 if not os.path.exists(base):
                     fn(target=base, *args, **kwargs)
-            if not LibvirtConnection._image_sems[fname].locked():
-                del LibvirtConnection._image_sems[fname]
+
+            call_if_not_exists(base, fn, *args, **kwargs)
 
             if cow:
                 utils.execute('qemu-img', 'create', '-f', 'qcow2', '-o',
@@ -1780,15 +1779,15 @@
         pass
 
     def refresh_security_group_rules(self, security_group):
-        # We use the semaphore to make sure noone applies the rule set
-        # after we've yanked the existing rules but before we've put in
-        # the new ones.
-        with self.iptables.semaphore:
-            for instance in self.instances.values():
-                self.remove_filters_for_instance(instance)
-                self.add_filters_for_instance(instance)
+        self.do_refresh_security_group_rules(security_group)
         self.iptables.apply()
 
+    @utils.synchronized('iptables', external=True)
+    def do_refresh_security_group_rules(self, security_group):
+        for instance in self.instances.values():
+            self.remove_filters_for_instance(instance)
+            self.add_filters_for_instance(instance)
+
     def _security_group_chain_name(self, security_group_id):
         return 'nova-sg-%s' % (security_group_id,)