tests and semaphore fix for image caching

2 files changed, 76 insertions, 4 deletions

diff --git a/nova/tests/test_virt.py b/nova/tests/test_virt.py
index f151ae911..ec7498d72 100644
--- a/nova/tests/test_virt.py
+++ b/nova/tests/test_virt.py
@@ -14,12 +14,16 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import os
+
+import eventlet
 from xml.etree.ElementTree import fromstring as xml_to_tree
 from xml.dom.minidom import parseString as xml_to_dom
 
 from nova import context
 from nova import db
 from nova import flags
+from nova import log as logging
 from nova import test
 from nova import utils
 from nova.api.ec2 import cloud
@@ -30,6 +34,68 @@ FLAGS = flags.FLAGS
 flags.DECLARE('instances_path', 'nova.compute.manager')
 
 
+def _concurrency(wait, done, target):
+    wait.wait()
+    done.send()
+
+
+class CacheConcurrencyTestCase(test.TestCase):
+    def setUp(self):
+        super(CacheConcurrencyTestCase, self).setUp()
+
+        def fake_exists(fname):
+            basedir = os.path.join(FLAGS.instances_path, '_base')
+            if fname == basedir:
+                return True
+            return False
+
+        def fake_execute(*args, **kwargs):
+            pass
+
+        self.stubs.Set(os.path, 'exists', fake_exists)
+        self.stubs.Set(utils, 'execute', fake_execute)
+
+    def test_same_fname_concurrency(self):
+        """Ensures that the same fname cache runs at a sequentially"""
+        conn = libvirt_conn.get_connection(False)
+        wait1 = eventlet.event.Event()
+        done1 = eventlet.event.Event()
+        eventlet.spawn(conn._cache_image, _concurrency,
+                       'target', 'fname', False, wait1, done1)
+        wait2 = eventlet.event.Event()
+        done2 = eventlet.event.Event()
+        eventlet.spawn(conn._cache_image, _concurrency,
+                       'target', 'fname', False, wait2, done2)
+        wait2.send()
+        eventlet.sleep(0)
+        try:
+            self.assertFalse(done2.ready())
+        finally:
+            wait1.send()
+        done1.wait()
+        eventlet.sleep(0)
+        self.assertTrue(done2.ready())
+
+    def test_different_fname_concurrency(self):
+        """Ensures that two different fname caches are concurrent"""
+        conn = libvirt_conn.get_connection(False)
+        wait1 = eventlet.event.Event()
+        done1 = eventlet.event.Event()
+        eventlet.spawn(conn._cache_image, _concurrency,
+                       'target', 'fname2', False, wait1, done1)
+        wait2 = eventlet.event.Event()
+        done2 = eventlet.event.Event()
+        eventlet.spawn(conn._cache_image, _concurrency,
+                       'target', 'fname1', False, wait2, done2)
+        wait2.send()
+        eventlet.sleep(0)
+        try:
+            self.assertTrue(done2.ready())
+        finally:
+            wait1.send()
+            eventlet.sleep(0)
+
+
 class LibvirtConnTestCase(test.TestCase):
     def setUp(self):
         super(LibvirtConnTestCase, self).setUp()
diff --git a/nova/virt/libvirt_conn.py b/nova/virt/libvirt_conn.py
index 9f7315c17..1a1f146d4 100644
--- a/nova/virt/libvirt_conn.py
+++ b/nova/virt/libvirt_conn.py
@@ -44,9 +44,8 @@ import uuid
 from xml.dom import minidom
 
 
-from eventlet import greenthread
-from eventlet import event
 from eventlet import tpool
+from eventlet import semaphore
 
 import IPy
 
@@ -512,6 +511,8 @@ class LibvirtConnection(object):
         subprocess.Popen(cmd, shell=True)
         return {'token': token, 'host': host, 'port': port}
 
+    _image_semaphores = {}
+
     def _cache_image(self, fn, target, fname, cow=False, *args, **kwargs):
         """Wrapper for a method that creates an image that caches the image.
 
@@ -531,8 +532,13 @@ class LibvirtConnection(object):
             if not os.path.exists(base_dir):
                 os.mkdir(base_dir)
             base = os.path.join(base_dir, fname)
-            if not os.path.exists(base):
-                fn(target=base, *args, **kwargs)
+
+            if fname not in self._image_semaphores:
+                self._image_semaphores[fname] = semaphore.Semaphore()
+            with self._image_semaphores[fname]:
+                if not os.path.exists(base):
+                    fn(target=base, *args, **kwargs)
+
             if cow:
                 utils.execute('qemu-img create -f qcow2 -o '
                               'cluster_size=2M,backing_file=%s %s'