Merge lp:~vishvananda/nova/no-db-messaging into lp:~hudson-openstack/nova/trunk

[Vish Ishaya]

This is an initial proposal for feedback. This branch is an attempt to start on this blueprint:
https://blueprints.launchpad.net/nova/+spec/no-db-messaging
which will allow for the implementation of this blueprint:
https://blueprints.launchpad.net/nova/+spec/separate-code-for-services
And will ultimately make it easy to replace our various services with external projects.

This prototype changes volume_create to pass data through the queue instead of writing information to the database and reading it on the other end. It attempts to make minimal changes. It includes:
 * a small change to model code to allow for conversion into dicts
 * scheduler uses call instead of cast
 * volume.manager provides a volume_get to query the state of a volume
 * volume.api creates volume with a call and then spawn greenthread to poll status
 * volume.manager returns initial data immediately and then spawn a greenthread to do further work

This will allow us to communicate with an external REST api very easily, by simply turning the two rpc calls into POST and GET. It also allows us to separate out the database for volumes, so the volume worker doesn't need to write to the shared zone database

Outstanding Issues for database split:
 * we have to use something like remote_id to refer to the volume when making requests to nova-volume because the id in our local database will not match. Ultimately we've discussed switching to UUIDs, so It may be good to do that change at the same time.
 * if the state of a volume changes after the initial poll has finished, we will never know about it. We ultimately need a way for api to be notified if the volume status changes so that we can update our local data, or we need a background worker that is polling the service for changes at an irregular interval.

I'm open to feedback about this approach. I tried a few other versions and this seems like the simplest change set to get what we want. If this looks good, I will modify the other volume commands to work the same way and propose a similar set of changes for compute.

[Vish Ishaya (vishvananda)]

accidentally pushed an old version. This is the right one.

[lp:~vishvananda/nova/no-db-messaging]

1078. By Eldar Nugaev on 2011-05-17

Improved error notification in network create

1079. By Eldar Nugaev on 2011-05-17

Added network_info into refresh_security_group_rules
That fixs https://bugs.launchpad.net/nova/+bug/773308

1080. By Mark Washenberger on 2011-05-17

Convert instance_type_ids in the instances table from strings to integers to enable joins with instance_types. This in particular fixes a problem when using postgresql.

[Dan Prince (dan-prince)]

Hey Vish,

I just smoke stacked this and got some errors. It looks like the changes to models.py aren't quite right.

I'm getting the following in nova-api.log:

2011-05-17 17:17:18,574 ERROR nova.api.openstack [-] Caught error: 'AuthToken' object has no attribute 'server_management_url'
(nova.api.openstack): TRACE: Traceback (most recent call last):
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/nova/api/openstack/__init__.py", line 59, in __call__
(nova.api.openstack): TRACE: return req.get_response(self.application)
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/webob/request.py", line 919, in get_response
(nova.api.openstack): TRACE: application, catch_exc_info=False)
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/webob/request.py", line 887, in call_application
(nova.api.openstack): TRACE: app_iter = application(self.environ, start_response)
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/webob/dec.py", line 147, in __call__
(nova.api.openstack): TRACE: resp = self.call_func(req, *args, **self.kwargs)
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/webob/dec.py", line 208, in call_func
(nova.api.openstack): TRACE: return self.func(req, *args, **kwargs)
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/nova/api/openstack/auth.py", line 54, in __call__
(nova.api.openstack): TRACE: return self.authenticate(req)
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/nova/api/openstack/auth.py", line 107, in authenticate
(nova.api.openstack): TRACE: token.server_management_url
(nova.api.openstack): TRACE: AttributeError: 'AuthToken' object has no attribute 'server_management_url'
--

Running a simple 'nova list' command (using the OSAPI) should allow you to reproduce this.

review: Needs Fixing

[Vish Ishaya (vishvananda)]

Thanks dan, I'll look at this. Also had some feedback from termie offline, and I'm moving where the polling is happening.

[lp:~vishvananda/nova/no-db-messaging]

1081. By Josh Kearney on 2011-05-17

Added missing metadata join to instance_get calls.

1082. By Matt Dietz on 2011-05-17

Fixes improper attribute naming around instance types that broke Resizes.

1083. By termie on 2011-05-17

Docstring cleanup and formatting (nova/network dir). Minor style fixes as well.

1084. By Vish Ishaya on 2011-05-17

Fixes the naming of the server_management_url in auth and tests.

1085. By Josh Kearney on 2011-05-18

Added missing xenhost plugin. This was causing warnings to pop up in the compute logs during periodic_task runs. It must have not been bzr add'd when this code was merged.

1086. By Matt Dietz on 2011-05-18

Implements a basic mechanism for pushing notifications out to interested parties. The rationale for implementing notifications this way is that the responsibility for them shouldn't fall to Nova. As such, we simply will be pushing messages to a queue where another worker entirely can be written to push messages around to subscribers.

1087. By Johannes Erdfelt on 2011-05-19

Simple change to sort the list of controllers/methods before printing to make it easier to read

1088. By termie on 2011-05-19

add support to rpc for multicall

1089. By termie on 2011-05-19

make the test more expicit

1090. By termie on 2011-05-19

add commented out unworking code for yield-based returns

1091. By Chris Behrens on 2011-05-19

Add a connection pool for rpc cast/call
Use the same rabbit connection for all topic listening and wait to be notified vs doing a 0.1 second poll for each.

1092. By Chris Behrens on 2011-05-19

pep8 and comment fixes

1093. By Chris Behrens on 2011-05-19

convert fanout_cast to ConnectionPool

1094. By Chris Behrens on 2011-05-19

fakerabbit's declare_consumer should support more than 1 consumer. also: make fakerabbit Backend.consume be an iterator like it should be..

1095. By Vish Ishaya on 2011-05-19

fix consumers to actually be deleted and clean up cloud test

1096. By Chris Behrens on 2011-05-19

catch greenlet.GreenletExit when shutting service down

1097. By Chris Behrens on 2011-05-19

Always create Service consumers no matter if report_interval is 0
Fix tests to handle how Service loads Consumers now

1098. By Chris Behrens on 2011-05-19

Add rpc_conn_pool_size flag for the new connection pool

1099. By termie on 2011-05-19

bring back commits lost in merge

1100. By termie on 2011-05-19

almost everything working with fake_rabbit

1101. By termie on 2011-05-19

don't need to use a separate connection

1102. By Vish Ishaya on 2011-05-19

lots of fixes for rpc and extra imports

1103. By termie on 2011-05-20

make sure that using multicall on a call with a single result still functions

1104. By termie on 2011-05-20

cleanup the code for merging

1105. By Vish Ishaya on 2011-05-20

update manager to use multicall

1106. By Vish Ishaya on 2011-05-20

fix conversion of models to dicts

1107. By Vish Ishaya on 2011-05-20

change volume_api to have delayed_create

1108. By Vish Ishaya on 2011-05-20

make scheduler use multicall

1109. By Vish Ishaya on 2011-05-20

remove the unnecessary try except in manager

1110. By Vish Ishaya on 2011-05-20

pep8

[Vish Ishaya (vishvananda)]

reproposing this with a prereq branch.

Unmerged revisions

1123. By Vish Ishaya on 2011-06-22

merged trunk

1122. By Vish Ishaya on 2011-06-13

remove merge error calling failing test

1121. By Vish Ishaya on 2011-06-02

fix snapshot test

1120. By Vish Ishaya on 2011-06-02

return not yield in scheduler shortcut

1119. By Vish Ishaya on 2011-06-02

make sure to handle VolumeIsBusy

1118. By Vish Ishaya on 2011-06-01

merged trunk and removed conflicts

1117. By Vish Ishaya on 2011-05-26

lost some changes from rpc branch, bring them in manually

1116. By Vish Ishaya on 2011-05-26

use strtime for passing datetimes back and forth through the queue

1115. By Vish Ishaya on 2011-05-26

fix tests

1114. By Vish Ishaya on 2011-05-26

keep the database on the receiving end as well

=== modified file 'nova/db/sqlalchemy/models.py'
--- nova/db/sqlalchemy/models.py	2011-05-17 20:50:12 +0000
+++ nova/db/sqlalchemy/models.py	2011-05-20 01:34:25 +0000
@@ -77,17 +77,25 @@
         return getattr(self, key, default)
 
     def __iter__(self):
-        self._i = iter(object_mapper(self).columns)
+        # NOTE(vish): include name property in the iterator
+        columns = dict(object_mapper(self).columns).keys()
+        name = self.get('name')
+        if name:
+            columns.append('name')
+        self._i = iter(columns)
         return self
 
     def next(self):
-        n = self._i.next().name
+        n = self._i.next()
         return n, getattr(self, n)
 
     def update(self, values):
         """Make the model object behave like a dict"""
-        for k, v in values.iteritems():
-            setattr(self, k, v)
+        columns = dict(object_mapper(self).columns).keys()
+        for key, value in values.iteritems():
+            # NOTE(vish): don't update the 'name' property
+            if key in columns:
+                setattr(self, key, value)
 
     def iteritems(self):
         """Make the model object behave like a dict.
 
=== modified file 'nova/fakerabbit.py'
--- nova/fakerabbit.py	2011-02-22 23:05:48 +0000
+++ nova/fakerabbit.py	2011-05-20 01:34:25 +0000
@@ -31,6 +31,7 @@
 
 EXCHANGES = {}
 QUEUES = {}
+CONSUMERS = {}
 
 
 class Message(base.BaseMessage):
@@ -96,17 +97,29 @@
                 ' key %(routing_key)s') % locals())
         EXCHANGES[exchange].bind(QUEUES[queue].push, routing_key)
 
-    def declare_consumer(self, queue, callback, *args, **kwargs):
-        self.current_queue = queue
-        self.current_callback = callback
+    def declare_consumer(self, queue, callback, consumer_tag, *args, **kwargs):
+        global CONSUMERS
+        LOG.debug("Adding consumer %s", consumer_tag)
+        CONSUMERS[consumer_tag] = (queue, callback)
+
+    def cancel(self, consumer_tag):
+        global CONSUMERS
+        LOG.debug("Removing consumer %s", consumer_tag)
+        del CONSUMERS[consumer_tag]
 
     def consume(self, limit=None):
+        global CONSUMERS
+        num = 0
         while True:
-            item = self.get(self.current_queue)
-            if item:
-                self.current_callback(item)
-                raise StopIteration()
-            greenthread.sleep(0)
+            for (queue, callback) in CONSUMERS.itervalues():
+                item = self.get(queue)
+                if item:
+                    callback(item)
+                    num += 1
+                    yield
+                    if limit and num == limit:
+                        raise StopIteration()
+            greenthread.sleep(0.1)
 
     def get(self, queue, no_ack=False):
         global QUEUES
@@ -134,5 +147,7 @@
 def reset_all():
     global EXCHANGES
     global QUEUES
+    global CONSUMERS
     EXCHANGES = {}
     QUEUES = {}
+    CONSUMERS = {}
 
=== modified file 'nova/rpc.py'
--- nova/rpc.py	2011-04-20 19:08:22 +0000
+++ nova/rpc.py	2011-05-20 01:34:25 +0000
@@ -33,7 +33,9 @@
 from carrot import connection as carrot_connection
 from carrot import messaging
 from eventlet import greenpool
-from eventlet import greenthread
+from eventlet import pools
+from eventlet import queue
+import greenlet
 
 from nova import context
 from nova import exception
@@ -47,7 +49,10 @@
 
 
 FLAGS = flags.FLAGS
-flags.DEFINE_integer('rpc_thread_pool_size', 1024, 'Size of RPC thread pool')
+flags.DEFINE_integer('rpc_thread_pool_size', 1024,
+                     'Size of RPC thread pool')
+flags.DEFINE_integer('rpc_conn_pool_size', 30,
+                     'Size of RPC connection pool')
 
 
 class Connection(carrot_connection.BrokerConnection):
@@ -90,6 +95,17 @@
         return cls.instance()
 
 
+class Pool(pools.Pool):
+    """Class that implements a Pool of Connections."""
+
+    def create(self):
+        LOG.debug('Creating new connection')
+        return Connection.instance(new=True)
+
+
+ConnectionPool = Pool(max_size=FLAGS.rpc_conn_pool_size)
+
+
 class Consumer(messaging.Consumer):
     """Consumer base class.
 
@@ -131,7 +147,9 @@
                 self.connection = Connection.recreate()
                 self.backend = self.connection.create_backend()
                 self.declare()
-            super(Consumer, self).fetch(no_ack, auto_ack, enable_callbacks)
+            return super(Consumer, self).fetch(no_ack,
+                                               auto_ack,
+                                               enable_callbacks)
             if self.failed_connection:
                 LOG.error(_('Reconnected to queue'))
                 self.failed_connection = False
@@ -159,13 +177,13 @@
         self.pool = greenpool.GreenPool(FLAGS.rpc_thread_pool_size)
         super(AdapterConsumer, self).__init__(connection=connection,
                                               topic=topic)
-
-    def receive(self, *args, **kwargs):
-        self.pool.spawn_n(self._receive, *args, **kwargs)
-
-    @exception.wrap_exception
-    def _receive(self, message_data, message):
-        """Magically looks for a method on the proxy object and calls it.
+        self.register_callback(self.process_data)
+
+    def process_data(self, message_data, message):
+        """Consumer callback to call a method on a proxy object.
+
+        Parses the message for validity and fires off a thread to call the
+        proxy object method.
 
         Message data should be a dictionary with two keys:
             method: string representing the method to call
@@ -175,8 +193,8 @@
 
         """
         LOG.debug(_('received %s') % message_data)
-        msg_id = message_data.pop('_msg_id', None)
-
+        # This will be popped off in _unpack_context
+        msg_id = message_data.get('_msg_id', None)
         ctxt = _unpack_context(message_data)
 
         method = message_data.get('method')
@@ -190,6 +208,13 @@
             LOG.warn(_('no method for message: %s') % message_data)
             msg_reply(msg_id, _('No method for message: %s') % message_data)
             return
+        self.pool.spawn_n(self._process_data, msg_id, ctxt, method, args)
+
+    @exception.wrap_exception
+    def _process_data(self, msg_id, ctxt, method, args):
+        """Thread that maigcally looks for a method on the proxy
+        object and calls it.
+        """
 
         node_func = getattr(self.proxy, str(method))
         node_args = dict((str(k), v) for k, v in args.iteritems())
@@ -197,7 +222,15 @@
         try:
             rval = node_func(context=ctxt, **node_args)
             if msg_id:
-                msg_reply(msg_id, rval, None)
+                # Check if the result was a generator
+                if hasattr(rval, 'send'):
+                    for x in rval:
+                        msg_reply(msg_id, x, None)
+                else:
+                    msg_reply(msg_id, rval, None)
+
+                # This final None tells multicall that it is done.
+                msg_reply(msg_id, None, None)
         except Exception as e:
             logging.exception('Exception during message handling')
             if msg_id:
@@ -205,11 +238,6 @@
         return
 
 
-class Publisher(messaging.Publisher):
-    """Publisher base class."""
-    pass
-
-
 class TopicAdapterConsumer(AdapterConsumer):
     """Consumes messages on a specific topic."""
 
@@ -242,6 +270,59 @@
                                     topic=topic, proxy=proxy)
 
 
+class ConsumerSet(object):
+    """Groups consumers to listen on together on a single connection."""
+
+    def __init__(self, conn, consumer_list):
+        self.consumer_list = set(consumer_list)
+        self.consumer_set = None
+        self.enabled = True
+        self.init(conn)
+
+    def init(self, conn):
+        if not conn:
+            conn = Connection.instance(new=True)
+        if self.consumer_set:
+            self.consumer_set.close()
+        self.consumer_set = messaging.ConsumerSet(conn)
+        for consumer in self.consumer_list:
+            consumer.connection = conn
+            # consumer.backend is set for us
+            self.consumer_set.add_consumer(consumer)
+
+    def reconnect(self):
+        self.init(None)
+
+    def wait(self, limit=None):
+        running = True
+        while running:
+            it = self.consumer_set.iterconsume(limit=limit)
+            if not it:
+                break
+            while True:
+                try:
+                    it.next()
+                except StopIteration:
+                    return
+                except greenlet.GreenletExit:
+                    running = False
+                    break
+                except Exception as e:
+                    LOG.error(_("Received exception %s " % type(e) + \
+                            "while processing consumer"))
+                    self.reconnect()
+                    # Break to outer loop
+                    break
+
+    def close(self):
+        self.consumer_set.close()
+
+
+class Publisher(messaging.Publisher):
+    """Publisher base class."""
+    pass
+
+
 class TopicPublisher(Publisher):
     """Publishes messages on a specific topic."""
 
@@ -306,16 +387,18 @@
         LOG.error(_("Returning exception %s to caller"), message)
         LOG.error(tb)
         failure = (failure[0].__name__, str(failure[1]), tb)
-    conn = Connection.instance()
-    publisher = DirectPublisher(connection=conn, msg_id=msg_id)
-    try:
-        publisher.send({'result': reply, 'failure': failure})
-    except TypeError:
-        publisher.send(
-                {'result': dict((k, repr(v))
-                                for k, v in reply.__dict__.iteritems()),
-                 'failure': failure})
-    publisher.close()
+
+    with ConnectionPool.item() as conn:
+        publisher = DirectPublisher(connection=conn, msg_id=msg_id)
+        try:
+            publisher.send({'result': reply, 'failure': failure})
+        except TypeError:
+            publisher.send(
+                    {'result': dict((k, repr(v))
+                                    for k, v in reply.__dict__.iteritems()),
+                     'failure': failure})
+
+        publisher.close()
 
 
 class RemoteError(exception.Error):
@@ -347,8 +430,9 @@
         if key.startswith('_context_'):
             value = msg.pop(key)
             context_dict[key[9:]] = value
+    context_dict['msg_id'] = msg.pop('_msg_id', None)
     LOG.debug(_('unpacked context: %s'), context_dict)
-    return context.RequestContext.from_dict(context_dict)
+    return RpcContext.from_dict(context_dict)
 
 
 def _pack_context(msg, context):
@@ -360,70 +444,110 @@
     for args at some point.
 
     """
-    context = dict([('_context_%s' % key, value)
-                   for (key, value) in context.to_dict().iteritems()])
-    msg.update(context)
-
-
-def call(context, topic, msg):
-    """Sends a message on a topic and wait for a response."""
+    context_d = dict([('_context_%s' % key, value)
+                      for (key, value) in context.to_dict().iteritems()])
+    msg.update(context_d)
+
+
+class RpcContext(context.RequestContext):
+    def __init__(self, *args, **kwargs):
+        msg_id = kwargs.pop('msg_id', None)
+        self.msg_id = msg_id
+        super(RpcContext, self).__init__(*args, **kwargs)
+
+    def reply(self, *args, **kwargs):
+        msg_reply(self.msg_id, *args, **kwargs)
+
+
+def multicall(context, topic, msg):
+    """Make a call that returns multiple times."""
     LOG.debug(_('Making asynchronous call on %s ...'), topic)
     msg_id = uuid.uuid4().hex
     msg.update({'_msg_id': msg_id})
     LOG.debug(_('MSG_ID is %s') % (msg_id))
     _pack_context(msg, context)
 
-    class WaitMessage(object):
-        def __call__(self, data, message):
-            """Acks message and sets result."""
-            message.ack()
-            if data['failure']:
-                self.result = RemoteError(*data['failure'])
-            else:
-                self.result = data['result']
-
-    wait_msg = WaitMessage()
-    conn = Connection.instance()
-    consumer = DirectConsumer(connection=conn, msg_id=msg_id)
+    con_conn = ConnectionPool.get()
+    consumer = DirectConsumer(connection=con_conn, msg_id=msg_id)
+    wait_msg = MulticallWaiter(consumer)
     consumer.register_callback(wait_msg)
 
-    conn = Connection.instance()
-    publisher = TopicPublisher(connection=conn, topic=topic)
+    publisher = TopicPublisher(connection=con_conn, topic=topic)
     publisher.send(msg)
     publisher.close()
 
-    try:
-        consumer.wait(limit=1)
-    except StopIteration:
-        pass
-    consumer.close()
-    # NOTE(termie): this is a little bit of a change from the original
-    #               non-eventlet code where returning a Failure
-    #               instance from a deferred call is very similar to
-    #               raising an exception
-    if isinstance(wait_msg.result, Exception):
-        raise wait_msg.result
-    return wait_msg.result
+    return wait_msg
+
+
+class MulticallWaiter(object):
+    def __init__(self, consumer):
+        self._consumer = consumer
+        self._results = queue.Queue()
+        self._closed = False
+
+    def close(self):
+        self._closed = True
+        self._consumer.close()
+        ConnectionPool.put(self._consumer.connection)
+
+    def __call__(self, data, message):
+        """Acks message and sets result."""
+        message.ack()
+        if data['failure']:
+            self._results.put(RemoteError(*data['failure']))
+        else:
+            self._results.put(data['result'])
+
+    def __iter__(self):
+        return self.wait()
+
+    def wait(self):
+        while True:
+            rv = None
+            while rv is None and not self._closed:
+                try:
+                    rv = self._consumer.fetch(enable_callbacks=True)
+                except Exception:
+                    self.close()
+                    raise
+                time.sleep(0.01)
+
+            result = self._results.get()
+            if isinstance(result, Exception):
+                self.close()
+                raise result
+            if result == None:
+                self.close()
+                raise StopIteration
+            yield result
+
+
+def call(context, topic, msg):
+    """Sends a message on a topic and wait for a response."""
+    rv = multicall(context, topic, msg)
+    for x in rv:
+        rv.close()
+        return x
 
 
 def cast(context, topic, msg):
     """Sends a message on a topic without waiting for a response."""
     LOG.debug(_('Making asynchronous cast on %s...'), topic)
     _pack_context(msg, context)
-    conn = Connection.instance()
-    publisher = TopicPublisher(connection=conn, topic=topic)
-    publisher.send(msg)
-    publisher.close()
+    with ConnectionPool.item() as conn:
+        publisher = TopicPublisher(connection=conn, topic=topic)
+        publisher.send(msg)
+        publisher.close()
 
 
 def fanout_cast(context, topic, msg):
     """Sends a message on a fanout exchange without waiting for a response."""
     LOG.debug(_('Making asynchronous fanout cast...'))
     _pack_context(msg, context)
-    conn = Connection.instance()
-    publisher = FanoutPublisher(topic, connection=conn)
-    publisher.send(msg)
-    publisher.close()
+    with ConnectionPool.item() as conn:
+        publisher = FanoutPublisher(topic, connection=conn)
+        publisher.send(msg)
+        publisher.close()
 
 
 def generic_response(message_data, message):
@@ -459,6 +583,7 @@
 
     if wait:
         consumer.wait()
+        consumer.close()
 
 
 if __name__ == '__main__':
 
=== modified file 'nova/scheduler/manager.py'
--- nova/scheduler/manager.py	2011-05-05 14:35:44 +0000
+++ nova/scheduler/manager.py	2011-05-20 01:34:25 +0000
@@ -83,11 +83,13 @@
         except AttributeError:
             host = self.driver.schedule(elevated, topic, *args, **kwargs)
 
-        rpc.cast(context,
-                 db.queue_get_for(context, topic, host),
-                 {"method": method,
-                  "args": kwargs})
-        LOG.debug(_("Casting to %(topic)s %(host)s for %(method)s") % locals())
+        LOG.debug(_("Multicall %(topic)s %(host)s for %(method)s") % locals())
+        rvs = rpc.multicall(context,
+                            db.queue_get_for(context, topic, host),
+                            {"method": method,
+                             "args": kwargs})
+        for rv in rvs:
+            yield rv
 
     # NOTE (masumotok) : This method should be moved to nova.api.ec2.admin.
     #                    Based on bexar design summit discussion,
 
=== modified file 'nova/service.py'
--- nova/service.py	2011-04-20 19:08:22 +0000
+++ nova/service.py	2011-05-20 01:34:25 +0000
@@ -19,14 +19,11 @@
 
 """Generic Node baseclass for all workers that run on hosts."""
 
+import greenlet
 import inspect
 import os
-import sys
-import time
 
-from eventlet import event
 from eventlet import greenthread
-from eventlet import greenpool
 
 from nova import context
 from nova import db
@@ -91,27 +88,38 @@
         if 'nova-compute' == self.binary:
             self.manager.update_available_resource(ctxt)
 
-        conn1 = rpc.Connection.instance(new=True)
-        conn2 = rpc.Connection.instance(new=True)
-        conn3 = rpc.Connection.instance(new=True)
+        self.conn = rpc.Connection.instance(new=True)
+        logging.debug("Creating Consumer connection for Service %s" %
+                      self.topic)
+
+        # Share this same connection for these Consumers
+        consumer_all = rpc.TopicAdapterConsumer(
+                connection=self.conn,
+                topic=self.topic,
+                proxy=self)
+        consumer_node = rpc.TopicAdapterConsumer(
+                connection=self.conn,
+                topic='%s.%s' % (self.topic, self.host),
+                proxy=self)
+        fanout = rpc.FanoutAdapterConsumer(
+                connection=self.conn,
+                topic=self.topic,
+                proxy=self)
+
+        cset = rpc.ConsumerSet(self.conn, [consumer_all,
+                                           consumer_node,
+                                           fanout])
+
+        # Wait forever, processing these consumers
+        def _wait():
+            try:
+                cset.wait()
+            finally:
+                cset.close()
+
+        self.csetthread = greenthread.spawn(_wait)
+
         if self.report_interval:
-            consumer_all = rpc.TopicAdapterConsumer(
-                    connection=conn1,
-                    topic=self.topic,
-                    proxy=self)
-            consumer_node = rpc.TopicAdapterConsumer(
-                    connection=conn2,
-                    topic='%s.%s' % (self.topic, self.host),
-                    proxy=self)
-            fanout = rpc.FanoutAdapterConsumer(
-                    connection=conn3,
-                    topic=self.topic,
-                    proxy=self)
-
-            self.timers.append(consumer_all.attach_to_eventlet())
-            self.timers.append(consumer_node.attach_to_eventlet())
-            self.timers.append(fanout.attach_to_eventlet())
-
             pulse = utils.LoopingCall(self.report_state)
             pulse.start(interval=self.report_interval, now=False)
             self.timers.append(pulse)
@@ -167,7 +175,13 @@
 
     def kill(self):
         """Destroy the service object in the datastore."""
+        self.csetthread.kill()
+        try:
+            self.csetthread.wait()
+        except greenlet.GreenletExit:
+            pass
         self.stop()
+        rpc.ConnectionPool.put(self.conn)
         try:
             db.service_destroy(context.get_admin_context(), self.service_id)
         except exception.NotFound:
 
=== modified file 'nova/test.py'
--- nova/test.py	2011-04-20 19:08:22 +0000
+++ nova/test.py	2011-05-20 01:34:25 +0000
@@ -31,17 +31,15 @@
 import unittest
 
 import mox
-import shutil
 import stubout
 from eventlet import greenthread
 
-from nova import context
-from nova import db
 from nova import fakerabbit
 from nova import flags
 from nova import rpc
 from nova import service
 from nova import wsgi
+from nova.virt import fake
 
 
 FLAGS = flags.FLAGS
@@ -85,6 +83,7 @@
         self._monkey_patch_attach()
         self._monkey_patch_wsgi()
         self._original_flags = FLAGS.FlagValuesDict()
+        rpc.ConnectionPool = rpc.Pool(max_size=FLAGS.rpc_conn_pool_size)
 
     def tearDown(self):
         """Runs after each test method to tear down test environment."""
@@ -99,6 +98,10 @@
             if FLAGS.fake_rabbit:
                 fakerabbit.reset_all()
 
+            if FLAGS.connection_type == 'fake':
+                if hasattr(fake.FakeConnection, '_instance'):
+                    del fake.FakeConnection._instance
+
             # Reset any overriden flags
             self.reset_flags()
 
 
=== modified file 'nova/tests/integrated/integrated_helpers.py'
--- nova/tests/integrated/integrated_helpers.py	2011-03-30 01:13:04 +0000
+++ nova/tests/integrated/integrated_helpers.py	2011-05-20 01:34:25 +0000
@@ -154,10 +154,7 @@
         # 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.auth_url = self._start_api_service()
 
=== modified file 'nova/tests/test_cloud.py'
--- nova/tests/test_cloud.py	2011-05-16 20:30:40 +0000
+++ nova/tests/test_cloud.py	2011-05-20 01:34:25 +0000
@@ -17,13 +17,8 @@
 #    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 +28,10 @@
 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 +72,21 @@
         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 +113,6 @@
         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 +128,10 @@
         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 +303,25 @@
                   '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')
 
=== modified file 'nova/tests/test_rpc.py'
--- nova/tests/test_rpc.py	2011-02-23 22:41:11 +0000
+++ nova/tests/test_rpc.py	2011-05-20 01:34:25 +0000
@@ -49,6 +49,59 @@
                                                  "args": {"value": value}})
         self.assertEqual(value, result)
 
+    def test_call_succeed_despite_multiple_returns(self):
+        """Get a value through rpc call"""
+        value = 42
+        result = rpc.call(self.context, 'test', {"method": "echo_three_times",
+                                                 "args": {"value": value}})
+        self.assertEqual(value, result)
+
+    def test_call_succeed_despite_multiple_returns_yield(self):
+        """Get a value through rpc call"""
+        value = 42
+        result = rpc.call(self.context, 'test',
+                          {"method": "echo_three_times_yield",
+                           "args": {"value": value}})
+        self.assertEqual(value, result)
+
+    def test_multicall_succeed_once(self):
+        """Get a value through rpc call"""
+        value = 42
+        result = rpc.multicall(self.context,
+                              'test',
+                              {"method": "echo",
+                               "args": {"value": value}})
+        i = 0
+        for x in result:
+            if i > 0:
+                self.fail('should only receive one response')
+            self.assertEqual(value + i, x)
+            i += 1
+
+    def test_multicall_succeed_three_times(self):
+        """Get a value through rpc call"""
+        value = 42
+        result = rpc.multicall(self.context,
+                              'test',
+                              {"method": "echo_three_times",
+                               "args": {"value": value}})
+        i = 0
+        for x in result:
+            self.assertEqual(value + i, x)
+            i += 1
+
+    def test_multicall_succeed_three_times_yield(self):
+        """Get a value through rpc call"""
+        value = 42
+        result = rpc.multicall(self.context,
+                              'test',
+                              {"method": "echo_three_times_yield",
+                               "args": {"value": value}})
+        i = 0
+        for x in result:
+            self.assertEqual(value + i, x)
+            i += 1
+
     def test_context_passed(self):
         """Makes sure a context is passed through rpc call"""
         value = 42
@@ -127,6 +180,18 @@
         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"""
         raise Exception(value)
 
=== modified file 'nova/tests/test_service.py'
--- nova/tests/test_service.py	2011-03-17 13:35:00 +0000
+++ nova/tests/test_service.py	2011-05-20 01:34:25 +0000
@@ -106,7 +106,10 @@
 
         # 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 @@
         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,13 @@
                             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(mox.IgnoreArg(), mox.IsA(list)).AndReturn(mock_cset)
+        wait_func(mox.IgnoreArg())
 
         service_create = {'host': host,
                           'binary': binary,
@@ -287,8 +299,41 @@
         # 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(mox.IgnoreArg(), 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)
 
=== modified file 'nova/volume/api.py'
--- nova/volume/api.py	2011-03-31 08:39:00 +0000
+++ nova/volume/api.py	2011-05-20 01:34:25 +0000
@@ -20,14 +20,14 @@
 Handles all requests relating to volumes.
 """
 
-import datetime
+import eventlet
 
-from nova import db
 from nova import exception
 from nova import flags
 from nova import log as logging
 from nova import quota
 from nova import rpc
+from nova import utils
 from nova.db import base
 
 FLAGS = flags.FLAGS
@@ -57,26 +57,60 @@
             'display_name': name,
             'display_description': description}
 
-        volume = self.db.volume_create(context, options)
-        rpc.cast(context,
-                 FLAGS.scheduler_topic,
-                 {"method": "create_volume",
-                  "args": {"topic": FLAGS.volume_topic,
-                           "volume_id": volume['id']}})
-        return volume
+        volume_ref = self.db.volume_create(context, options)
+        volume_ref = utils.to_primitive(dict(volume_ref))
+
+        def delayed_create(volume_ref):
+            vid = volume_ref['id']
+            try:
+                rvs = rpc.multicall(context,
+                                    FLAGS.scheduler_topic,
+                                    {"method": "create_volume",
+                                     "args": {"topic": FLAGS.volume_topic,
+                                              "volume_ref": volume_ref}})
+                for volume_ref in rvs:
+                    self.db.volume_update(context, vid, volume_ref)
+                volume_ref['launched_at'] = utils.utcnow()
+                self.db.volume_update(context, vid, volume_ref)
+
+            except rpc.RemoteError:
+                self.db.volume_update(context, vid, {'status': 'error'})
+
+        eventlet.spawn_n(delayed_create, volume_ref)
+        return volume_ref
 
     def delete(self, context, volume_id):
-        volume = self.get(context, volume_id)
-        if volume['status'] != "available":
+        volume_ref = self.get(context, volume_id)
+        if volume_ref['status'] != "available":
             raise exception.ApiError(_("Volume status must be available"))
-        now = datetime.datetime.utcnow()
-        self.db.volume_update(context, volume_id, {'status': 'deleting',
-                                                   'terminated_at': now})
-        host = volume['host']
-        rpc.cast(context,
-                 self.db.queue_get_for(context, FLAGS.volume_topic, host),
-                 {"method": "delete_volume",
-                  "args": {"volume_id": volume_id}})
+        if volume_ref['attach_status'] == "attached":
+            raise exception.Error(_("Volume is still attached"))
+
+        volume_ref['status'] = 'deleting'
+        volume_ref['terminated_at'] = utils.utcnow()
+        self.db.volume_update(context, volume_ref['id'], volume_ref)
+        volume_ref = utils.to_primitive(dict(volume_ref))
+
+        def delayed_delete(volume_ref):
+            vid = volume_ref['id']
+            try:
+                topic = self.db.queue_get_for(context,
+                                              FLAGS.volume_topic,
+                                              volume_ref['host'])
+                rvs = rpc.multicall(context,
+                                    topic,
+                                    {"method": "delete_volume",
+                                     "args": {"volume_ref": volume_ref}})
+                for volume_ref in rvs:
+                    self.db.volume_update(context, vid, volume_ref)
+
+                self.db.volume_destroy(context, vid)
+
+            except rpc.RemoteError:
+                self.db.volume_update(context, vid, {'status': 'err_delete'})
+
+        eventlet.spawn_n(delayed_delete, volume_ref)
+        return True
 
     def update(self, context, volume_id, fields):
         self.db.volume_update(context, volume_id, fields)
 
=== modified file 'nova/volume/manager.py'
--- nova/volume/manager.py	2011-03-17 13:35:00 +0000
+++ nova/volume/manager.py	2011-05-20 01:34:25 +0000
@@ -90,67 +90,40 @@
             else:
                 LOG.info(_("volume %s: skipping export"), volume['name'])
 
-    def create_volume(self, context, volume_id):
+    def create_volume(self, context, volume_ref):
         """Creates and exports the volume."""
-        context = context.elevated()
-        volume_ref = self.db.volume_get(context, volume_id)
         LOG.info(_("volume %s: creating"), volume_ref['name'])
 
-        self.db.volume_update(context,
-                              volume_id,
-                              {'host': self.host})
-        # NOTE(vish): so we don't have to get volume from db again
-        #             before passing it to the driver.
         volume_ref['host'] = self.host
-
-        try:
-            vol_name = volume_ref['name']
-            vol_size = volume_ref['size']
-            LOG.debug(_("volume %(vol_name)s: creating lv of"
-                    " size %(vol_size)sG") % locals())
-            model_update = self.driver.create_volume(volume_ref)
-            if model_update:
-                self.db.volume_update(context, volume_ref['id'], model_update)
-
-            LOG.debug(_("volume %s: creating export"), volume_ref['name'])
-            model_update = self.driver.create_export(context, volume_ref)
-            if model_update:
-                self.db.volume_update(context, volume_ref['id'], model_update)
-        except Exception:
-            self.db.volume_update(context,
-                                  volume_ref['id'], {'status': 'error'})
-            raise
-
-        now = datetime.datetime.utcnow()
-        self.db.volume_update(context,
-                              volume_ref['id'], {'status': 'available',
-                                                 'launched_at': now})
+        yield volume_ref
+
+        vol_name = volume_ref['name']
+        vol_size = volume_ref['size']
+        LOG.debug(_("volume %(vol_name)s: creating lv of"
+                " size %(vol_size)sG") % locals())
+        model_update = self.driver.create_volume(volume_ref)
+        if model_update:
+            volume_ref.update(model_update)
+            yield volume_ref
+
+        LOG.debug(_("volume %s: creating export"), volume_ref['name'])
+        model_update = self.driver.create_export(context, volume_ref)
+        if model_update:
+            volume_ref.update(model_update)
+            yield volume_ref
+
         LOG.debug(_("volume %s: created successfully"), volume_ref['name'])
-        return volume_id
+        volume_ref['status'] = 'available'
+        yield volume_ref
 
-    def delete_volume(self, context, volume_id):
+    def delete_volume(self, context, volume_ref):
         """Deletes and unexports volume."""
-        context = context.elevated()
-        volume_ref = self.db.volume_get(context, volume_id)
-        if volume_ref['attach_status'] == "attached":
-            raise exception.Error(_("Volume is still attached"))
-        if volume_ref['host'] != self.host:
-            raise exception.Error(_("Volume is not local to this node"))
-
-        try:
-            LOG.debug(_("volume %s: removing export"), volume_ref['name'])
-            self.driver.remove_export(context, volume_ref)
-            LOG.debug(_("volume %s: deleting"), volume_ref['name'])
-            self.driver.delete_volume(volume_ref)
-        except Exception:
-            self.db.volume_update(context,
-                                  volume_ref['id'],
-                                  {'status': 'error_deleting'})
-            raise
-
-        self.db.volume_destroy(context, volume_id)
+        LOG.debug(_("volume %s: removing export"), volume_ref['name'])
+        self.driver.remove_export(context, volume_ref)
+        LOG.debug(_("volume %s: deleting"), volume_ref['name'])
+        self.driver.delete_volume(volume_ref)
         LOG.debug(_("volume %s: deleted successfully"), volume_ref['name'])
-        return True
+        yield volume_ref
 
     def setup_compute_volume(self, context, volume_id):
         """Setup remote volume on compute host.