Adds the ability to make a call that returns multiple times (a call returning a generator). This is also based on the work in rpc-improvements + a bunch of fixes Vish and I worked through to get all the tests to pass so the code is a bit all over the place.

The functionality is being added to support Vish's work on removing worker access to the database, this allows us to write multi-phase actions that yield state updates as they progress, letting the frontend update the db.

4 files changed, 169 insertions, 37 deletions

diff --git a/nova/tests/integrated/integrated_helpers.py b/nova/tests/integrated/integrated_helpers.py
index bc98921f0..7f590441e 100644
--- a/nova/tests/integrated/integrated_helpers.py
+++ b/nova/tests/integrated/integrated_helpers.py
@@ -154,10 +154,7 @@ class _IntegratedTestBase(test.TestCase):
         # set up services
         self.start_service('compute')
         self.start_service('volume')
-        # NOTE(justinsb): There's a bug here which is eluding me...
-        # If we start the network_service, all is good, but then subsequent
-        # tests fail: CloudTestCase.test_ajax_console in particular.
-        #self.start_service('network')
+        self.start_service('network')
         self.start_service('scheduler')
 
         self._start_api_service()
diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py
index 54c0454de..b64be662e 100644
--- a/nova/tests/test_cloud.py
+++ b/nova/tests/test_cloud.py
@@ -17,13 +17,9 @@
 #    under the License.
 
 from base64 import b64decode
-import json
 from M2Crypto import BIO
 from M2Crypto import RSA
 import os
-import shutil
-import tempfile
-import time
 
 from eventlet import greenthread
 
@@ -33,12 +29,10 @@ from nova import db
 from nova import flags
 from nova import log as logging
 from nova import rpc
-from nova import service
 from nova import test
 from nova import utils
 from nova import exception
 from nova.auth import manager
-from nova.compute import power_state
 from nova.api.ec2 import cloud
 from nova.api.ec2 import ec2utils
 from nova.image import local
@@ -79,14 +73,21 @@ class CloudTestCase(test.TestCase):
         self.stubs.Set(local.LocalImageService, 'show', fake_show)
         self.stubs.Set(local.LocalImageService, 'show_by_name', fake_show)
 
+        # NOTE(vish): set up a manual wait so rpc.cast has a chance to finish
+        rpc_cast = rpc.cast
+
+        def finish_cast(*args, **kwargs):
+            rpc_cast(*args, **kwargs)
+            greenthread.sleep(0.2)
+
+        self.stubs.Set(rpc, 'cast', finish_cast)
+
     def tearDown(self):
         network_ref = db.project_get_network(self.context,
                                              self.project.id)
         db.network_disassociate(self.context, network_ref['id'])
         self.manager.delete_project(self.project)
         self.manager.delete_user(self.user)
-        self.compute.kill()
-        self.network.kill()
         super(CloudTestCase, self).tearDown()
 
     def _create_key(self, name):
@@ -113,7 +114,6 @@ class CloudTestCase(test.TestCase):
         self.cloud.describe_addresses(self.context)
         self.cloud.release_address(self.context,
                                   public_ip=address)
-        greenthread.sleep(0.3)
         db.floating_ip_destroy(self.context, address)
 
     def test_associate_disassociate_address(self):
@@ -129,12 +129,10 @@ class CloudTestCase(test.TestCase):
         self.cloud.associate_address(self.context,
                                      instance_id=ec2_id,
                                      public_ip=address)
-        greenthread.sleep(0.3)
         self.cloud.disassociate_address(self.context,
                                         public_ip=address)
         self.cloud.release_address(self.context,
                                   public_ip=address)
-        greenthread.sleep(0.3)
         self.network.deallocate_fixed_ip(self.context, fixed)
         db.instance_destroy(self.context, inst['id'])
         db.floating_ip_destroy(self.context, address)
@@ -306,31 +304,25 @@ class CloudTestCase(test.TestCase):
                   'instance_type': instance_type,
                   'max_count': max_count}
         rv = self.cloud.run_instances(self.context, **kwargs)
-        greenthread.sleep(0.3)
         instance_id = rv['instancesSet'][0]['instanceId']
         output = self.cloud.get_console_output(context=self.context,
                                                instance_id=[instance_id])
         self.assertEquals(b64decode(output['output']), 'FAKE CONSOLE?OUTPUT')
         # TODO(soren): We need this until we can stop polling in the rpc code
         #              for unit tests.
-        greenthread.sleep(0.3)
         rv = self.cloud.terminate_instances(self.context, [instance_id])
-        greenthread.sleep(0.3)
 
     def test_ajax_console(self):
         kwargs = {'image_id': 'ami-1'}
         rv = self.cloud.run_instances(self.context, **kwargs)
         instance_id = rv['instancesSet'][0]['instanceId']
-        greenthread.sleep(0.3)
         output = self.cloud.get_ajax_console(context=self.context,
                                              instance_id=[instance_id])
         self.assertEquals(output['url'],
                           '%s/?token=FAKETOKEN' % FLAGS.ajax_console_proxy_url)
         # TODO(soren): We need this until we can stop polling in the rpc code
         #              for unit tests.
-        greenthread.sleep(0.3)
         rv = self.cloud.terminate_instances(self.context, [instance_id])
-        greenthread.sleep(0.3)
 
     def test_key_generation(self):
         result = self._create_key('test')
diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py
index 44d7c91eb..ffd748efe 100644
--- a/nova/tests/test_rpc.py
+++ b/nova/tests/test_rpc.py
@@ -31,7 +31,6 @@ LOG = logging.getLogger('nova.tests.rpc')
 
 
 class RpcTestCase(test.TestCase):
-    """Test cases for rpc"""
     def setUp(self):
         super(RpcTestCase, self).setUp()
         self.conn = rpc.Connection.instance(True)
@@ -43,14 +42,55 @@ class RpcTestCase(test.TestCase):
         self.context = context.get_admin_context()
 
     def test_call_succeed(self):
-        """Get a value through rpc call"""
         value = 42
         result = rpc.call(self.context, 'test', {"method": "echo",
                                                  "args": {"value": value}})
         self.assertEqual(value, result)
 
+    def test_call_succeed_despite_multiple_returns(self):
+        value = 42
+        result = rpc.call(self.context, 'test', {"method": "echo_three_times",
+                                                 "args": {"value": value}})
+        self.assertEqual(value + 2, result)
+
+    def test_call_succeed_despite_multiple_returns_yield(self):
+        value = 42
+        result = rpc.call(self.context, 'test',
+                          {"method": "echo_three_times_yield",
+                           "args": {"value": value}})
+        self.assertEqual(value + 2, result)
+
+    def test_multicall_succeed_once(self):
+        value = 42
+        result = rpc.multicall(self.context,
+                              'test',
+                              {"method": "echo",
+                               "args": {"value": value}})
+        for i, x in enumerate(result):
+            if i > 0:
+                self.fail('should only receive one response')
+            self.assertEqual(value + i, x)
+
+    def test_multicall_succeed_three_times(self):
+        value = 42
+        result = rpc.multicall(self.context,
+                              'test',
+                              {"method": "echo_three_times",
+                               "args": {"value": value}})
+        for i, x in enumerate(result):
+            self.assertEqual(value + i, x)
+
+    def test_multicall_succeed_three_times_yield(self):
+        value = 42
+        result = rpc.multicall(self.context,
+                              'test',
+                              {"method": "echo_three_times_yield",
+                               "args": {"value": value}})
+        for i, x in enumerate(result):
+            self.assertEqual(value + i, x)
+
     def test_context_passed(self):
-        """Makes sure a context is passed through rpc call"""
+        """Makes sure a context is passed through rpc call."""
         value = 42
         result = rpc.call(self.context,
                           'test', {"method": "context",
@@ -58,11 +98,12 @@ class RpcTestCase(test.TestCase):
         self.assertEqual(self.context.to_dict(), result)
 
     def test_call_exception(self):
-        """Test that exception gets passed back properly
+        """Test that exception gets passed back properly.
 
         rpc.call returns a RemoteError object.  The value of the
         exception is converted to a string, so we convert it back
         to an int in the test.
+
         """
         value = 42
         self.assertRaises(rpc.RemoteError,
@@ -81,7 +122,7 @@ class RpcTestCase(test.TestCase):
             self.assertEqual(int(exc.value), value)
 
     def test_nested_calls(self):
-        """Test that we can do an rpc.call inside another call"""
+        """Test that we can do an rpc.call inside another call."""
         class Nested(object):
             @staticmethod
             def echo(context, queue, value):
@@ -108,25 +149,80 @@ class RpcTestCase(test.TestCase):
                                               "value": value}})
         self.assertEqual(value, result)
 
+    def test_connectionpool_single(self):
+        """Test that ConnectionPool recycles a single connection."""
+        conn1 = rpc.ConnectionPool.get()
+        rpc.ConnectionPool.put(conn1)
+        conn2 = rpc.ConnectionPool.get()
+        rpc.ConnectionPool.put(conn2)
+        self.assertEqual(conn1, conn2)
+
+    def test_connectionpool_double(self):
+        """Test that ConnectionPool returns and reuses separate connections.
+
+        When called consecutively we should get separate connections and upon
+        returning them those connections should be reused for future calls
+        before generating a new connection.
+
+        """
+        conn1 = rpc.ConnectionPool.get()
+        conn2 = rpc.ConnectionPool.get()
+
+        self.assertNotEqual(conn1, conn2)
+        rpc.ConnectionPool.put(conn1)
+        rpc.ConnectionPool.put(conn2)
+
+        conn3 = rpc.ConnectionPool.get()
+        conn4 = rpc.ConnectionPool.get()
+        self.assertEqual(conn1, conn3)
+        self.assertEqual(conn2, conn4)
+
+    def test_connectionpool_limit(self):
+        """Test connection pool limit and connection uniqueness."""
+        max_size = FLAGS.rpc_conn_pool_size
+        conns = []
+
+        for i in xrange(max_size):
+            conns.append(rpc.ConnectionPool.get())
+
+        self.assertFalse(rpc.ConnectionPool.free_items)
+        self.assertEqual(rpc.ConnectionPool.current_size,
+                rpc.ConnectionPool.max_size)
+        self.assertEqual(len(set(conns)), max_size)
+
 
 class TestReceiver(object):
-    """Simple Proxy class so the consumer has methods to call
+    """Simple Proxy class so the consumer has methods to call.
+
+    Uses static methods because we aren't actually storing any state.
 
-    Uses static methods because we aren't actually storing any state"""
+    """
 
     @staticmethod
     def echo(context, value):
-        """Simply returns whatever value is sent in"""
+        """Simply returns whatever value is sent in."""
         LOG.debug(_("Received %s"), value)
         return value
 
     @staticmethod
     def context(context, value):
-        """Returns dictionary version of context"""
+        """Returns dictionary version of context."""
         LOG.debug(_("Received %s"), context)
         return context.to_dict()
 
     @staticmethod
+    def echo_three_times(context, value):
+        context.reply(value)
+        context.reply(value + 1)
+        context.reply(value + 2)
+
+    @staticmethod
+    def echo_three_times_yield(context, value):
+        yield value
+        yield value + 1
+        yield value + 2
+
+    @staticmethod
     def fail(context, value):
-        """Raises an exception with the value sent in"""
+        """Raises an exception with the value sent in."""
         raise Exception(value)
diff --git a/nova/tests/test_service.py b/nova/tests/test_service.py
index d48de2057..d1cc8bd61 100644
--- a/nova/tests/test_service.py
+++ b/nova/tests/test_service.py
@@ -106,7 +106,10 @@ class ServiceTestCase(test.TestCase):
 
         # NOTE(vish): Create was moved out of mox replay to make sure that
         #             the looping calls are created in StartService.
-        app = service.Service.create(host=host, binary=binary)
+        app = service.Service.create(host=host, binary=binary, topic=topic)
+
+        self.mox.StubOutWithMock(service.rpc.Connection, 'instance')
+        service.rpc.Connection.instance(new=mox.IgnoreArg())
 
         self.mox.StubOutWithMock(rpc,
                                  'TopicAdapterConsumer',
@@ -114,6 +117,11 @@ class ServiceTestCase(test.TestCase):
         self.mox.StubOutWithMock(rpc,
                                  'FanoutAdapterConsumer',
                                  use_mock_anything=True)
+
+        self.mox.StubOutWithMock(rpc,
+                                 'ConsumerSet',
+                                 use_mock_anything=True)
+
         rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
                             topic=topic,
                             proxy=mox.IsA(service.Service)).AndReturn(
@@ -129,9 +137,14 @@ class ServiceTestCase(test.TestCase):
                             proxy=mox.IsA(service.Service)).AndReturn(
                                     rpc.FanoutAdapterConsumer)
 
-        rpc.TopicAdapterConsumer.attach_to_eventlet()
-        rpc.TopicAdapterConsumer.attach_to_eventlet()
-        rpc.FanoutAdapterConsumer.attach_to_eventlet()
+        def wait_func(self, limit=None):
+            return None
+
+        mock_cset = self.mox.CreateMock(rpc.ConsumerSet,
+                {'wait': wait_func})
+        rpc.ConsumerSet(connection=mox.IgnoreArg(),
+                        consumer_list=mox.IsA(list)).AndReturn(mock_cset)
+        wait_func(mox.IgnoreArg())
 
         service_create = {'host': host,
                           'binary': binary,
@@ -287,8 +300,42 @@ class ServiceTestCase(test.TestCase):
         # Creating mocks
         self.mox.StubOutWithMock(service.rpc.Connection, 'instance')
         service.rpc.Connection.instance(new=mox.IgnoreArg())
-        service.rpc.Connection.instance(new=mox.IgnoreArg())
-        service.rpc.Connection.instance(new=mox.IgnoreArg())
+
+        self.mox.StubOutWithMock(rpc,
+                                 'TopicAdapterConsumer',
+                                 use_mock_anything=True)
+        self.mox.StubOutWithMock(rpc,
+                                 'FanoutAdapterConsumer',
+                                 use_mock_anything=True)
+
+        self.mox.StubOutWithMock(rpc,
+                                 'ConsumerSet',
+                                 use_mock_anything=True)
+
+        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
+                            topic=topic,
+                            proxy=mox.IsA(service.Service)).AndReturn(
+                                    rpc.TopicAdapterConsumer)
+
+        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
+                            topic='%s.%s' % (topic, host),
+                            proxy=mox.IsA(service.Service)).AndReturn(
+                                    rpc.TopicAdapterConsumer)
+
+        rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(),
+                            topic=topic,
+                            proxy=mox.IsA(service.Service)).AndReturn(
+                                    rpc.FanoutAdapterConsumer)
+
+        def wait_func(self, limit=None):
+            return None
+
+        mock_cset = self.mox.CreateMock(rpc.ConsumerSet,
+                {'wait': wait_func})
+        rpc.ConsumerSet(connection=mox.IgnoreArg(),
+                        consumer_list=mox.IsA(list)).AndReturn(mock_cset)
+        wait_func(mox.IgnoreArg())
+
         self.mox.StubOutWithMock(serv.manager.driver,
                                  'update_available_resource')
         serv.manager.driver.update_available_resource(mox.IgnoreArg(), host)