Merge lp:~termie/nova/eventlet_merge into lp:~hudson-openstack/nova/trunk

[termie]

This branch removes most of the dependencies on twisted and moves towards the plan described by https://blueprints.launchpad.net/nova/+spec/unified-service-architecture

Tests are currently passing besides objectstore which is being skipped because it is heavily reliant on our twisted pieces, and I can run everything using the nova.sh

Additionally this adds nova-combined that covers everythign except for nova-objectstore, to test it what I've usually done is run nova.sh as usual

$ sudo ./eventlet_merge/contrib/nova.sh run ignored eventlet_merge

and then quit all the services except for nova-objectstore and then in one of the screens do

$ ./eventlet_merge/bin/nova-combined

And then run whatever manual testing you normally run.

Once objectstore has been deprecated and removed nova-combined can be expected to run the whole nova stack in a single process for testing and dev.

[Anthony Young (sleepsonthefloor)]

Some assorted notes from eday, termie, vishy, sleepsonthefloor, jesse

#11 - remove extra line above docstring
#32 remove main in nova-api
#93-95 duplicate flag api_port in
#199 xtra space
#548 fix space
#583 fix space
#634 s/return/pass
#938 pylint
#1398 - kill, and need to fix init scripts
#1440 use warn vs debug
#1442 remove section
#1452-1487 lots of asorted
#2467 can probably be removed
#2878 - hmm...
#3039 #3036 s/return/pass
#3125 - can just return timer_done
#3160 - _wait_for_timer
#3329 output, error
#3341 whitespace
#3427 delete
#3450 - fix comment - clarify what is meant by 'block'
#3425 maybe move to previous line?
#3450 move up to previous line
#3442-#3844 investigate - some funkiness going on. Ask armando?
#3831 - s/utis/utils
#3832 make sure timer stops
#3909 fix sleeping out of process greenthread.sleep
#3928 fix whitespace
#3938 fix whitespace
#3949 fix whitespace
#3999 greenthread.sleep
#4093 fix whitespace
#4114 fix whitespace
#4235 fix whitespace
#4253 remove debug code

review: Needs Fixing

[Jay Pipes (jaypipes)]

I have a feeling the git-bzr bridge is not working properly... I see that same issues with timestamps that I saw when Chris MacGown used the git-bzr bridge. Andy, I assume you are still using git with your git-bzr bridge?

Note that this merge request first *removes* nova/service.py (notice the 1970-01-01 timestamps, BTW) and then *adds* a file nova/service.py. I don't know why this is the case and why it doesn't simply show the diff of the modified nova/service.py.

For Chris' branch, I had to essentially re-construct his patches and submit them from bzr to Launchpad; trying to merge it as-is into trunk resulted in conflicts that could not be resolved. In fact, Chris' git-bzr original branch (lp:~chris-slicehost/glance/implements_swift_backend) still sits in the list of Glance's active branches (https://code.launchpad.net/glance/) despite it being marked as Merged. The reason is because something happens in the repository metadata when the git-bzr bridge produces a branch, and somehow history is discarded. You can see that timestamps of revisions coming from the branches managed with the git-bzr bridge are all 0000-00-00...

Anyway, I'm saying this because despite me liking some of the changes in this patch (and I certainly welcome the direction it takes), I fear it will not merge correctly. In addition, the patch is (IMHO) unnecessarily large. It would be a lot easier to review (and thus get into trunk quickly) if the patch was broken up and we could simply merge in changes in chunks. This patch does not need to be done all at once.

Just my 2 cents.
-jay

[Eric Day (eday)]

Hi Jay,

I stopped by to see the Anso guys on Friday and we did a group code review on this. The list above is what came out of it. I had the same questions about removing/adding some files, and the it sounds like it was done this way because of using tmp files for developing and then when it worked, moving them into place. Even though they are named the same, it's pretty different in content.

As for the size, I suppose it could have been done in a couple chunks, but then there would have been duplicate files/classes (like service.py) for both eventlet and twisted. Most of the changes are just removing defers/yields, so I'm okay withe the size.

I will approve once the issues above are resolved.

[Jay Pipes (jaypipes)]

OK, Eric, we'll see if it merges correctly, then...

[termie (termie)]

> I have a feeling the git-bzr bridge is not working properly... I see that same
> issues with timestamps that I saw when Chris MacGown used the git-bzr bridge.
> Andy, I assume you are still using git with your git-bzr bridge?
>

Not git-bzr was used on this branch, I have largely stopped using it because I need to maintain patches against both bzr and bzr-fastimport to keep it working. I don't know which timestamps you are referring to.

> Note that this merge request first *removes* nova/service.py (notice the
> 1970-01-01 timestamps, BTW) and then *adds* a file nova/service.py. I don't
> know why this is the case and why it doesn't simply show the diff of the
> modified nova/service.py.

I had another file for the first half of the development of this branch that I later overwrote service.py with, it is something of a shame to lose the development history of that single file but not worth dealing the bzr's poor tools to attempt to reconstruct it.

> For Chris' branch, I had to essentially re-construct his patches and submit
> them from bzr to Launchpad; trying to merge it as-is into trunk resulted in
> conflicts that could not be resolved. In fact, Chris' git-bzr original branch
> (lp:~chris-slicehost/glance/implements_swift_backend) still sits in the list
> of Glance's active branches (https://code.launchpad.net/glance/) despite it
> being marked as Merged. The reason is because something happens in the
> repository metadata when the git-bzr bridge produces a branch, and somehow
> history is discarded. You can see that timestamps of revisions coming from the
> branches managed with the git-bzr bridge are all 0000-00-00...
>
> Anyway, I'm saying this because despite me liking some of the changes in this
> patch (and I certainly welcome the direction it takes), I fear it will not
> merge correctly. In addition, the patch is (IMHO) unnecessarily large. It
> would be a lot easier to review (and thus get into trunk quickly) if the patch
> was broken up and we could simply merge in changes in chunks. This patch does
> not need to be done all at once.

I also fear it will not merge correctly simply because while merging from upstream a bunch of spurious conflicts were generated. This patch does need to be done all at once.

> Just my 2 cents.
> -jay

[Jay Pipes (jaypipes)]

On Tue, Dec 14, 2010 at 6:26 PM, termie <email address hidden> wrote:
>> I have a feeling the git-bzr bridge is not working properly... I see that same
>> issues with timestamps that I saw when Chris MacGown used the git-bzr bridge.
>> Andy, I assume you are still using git with your git-bzr bridge?
>>
>
> Not git-bzr was used on this branch, I have largely stopped using it because I need to maintain patches against both bzr and bzr-fastimport to keep it working. I don't know which timestamps you are referring to.

OK.

>> Note that this merge request first *removes* nova/service.py (notice the
>> 1970-01-01 timestamps, BTW) and then *adds* a file nova/service.py.  I don't
>> know why this is the case and why it doesn't simply show the diff of the
>> modified nova/service.py.
>
> I had another file for the first half of the development of this branch that I later overwrote service.py with, it is something of a shame to lose the development history of that single file but not worth dealing the bzr's poor tools to attempt to reconstruct it.

Because you are used to or prefer git doesn't mean that bzr has poor
tooling. If you overwrite a file in git, the same thing would happen.

<snip>
> I also fear it will not merge correctly simply because while merging from upstream a bunch of spurious conflicts were generated. This patch does need to be done all at once.

OK. However, I don't agree, but I can move forward with the review...

-jay

[termie (termie)]

Updated per review.

[Josh Kearney (jk0)]

For what it's worth, I'm running this branch in my development environment and it's been great.

[Eric Day (eday)]

Is there any reason why the virt_unittests were removed? Also, I know we discussed this, but I'd still like to find a way to get the objectstore unittests running again, even if it means creating the base twisted test classes again just for it.

[Sandy Walsh (sandy-walsh)]

Some notes on trying to get this branch to run. I'll update again just to be sure I have the latest.

1. logging is console only by default (assuming run as user). Needed to update service.Service to get file logging. Not a biggie.

2. Fails when trying to run a xenserver instance: http://paste.openstack.org/show/303/

3. Why does nova-objectstore have to be run the old way and not integrated with nova-combined?

I'll see if I can fix #2 in the meanwhile.

[Sandy Walsh (sandy-walsh)]

Oh, it should be noted that run-combined still needs to take the flagfile ... this threw me off while testing.

bin/nova-combined --flagfile=../nova.conf

[Sandy Walsh (sandy-walsh)]

And, I was using the latest branch re: the above comments.

[termie (termie)]

> Is there any reason why the virt_unittests were removed? Also, I know we
> discussed this, but I'd still like to find a way to get the objectstore
> unittests running again, even if it means creating the base twisted test
> classes again just for it.

It was in there twice. But looking at the diff apparently bzr has lied to me yet again and somehow had conflicts it didn't mention, sorting those out.

[termie (termie)]

ah, the conflicts are just in the diff view that launchpad creates, which is still unfortunate

[termie (termie)]

> Some notes on trying to get this branch to run. I'll update again just to be
> sure I have the latest.
>
> 1. logging is console only by default (assuming run as user). Needed to update
> service.Service to get file logging. Not a biggie.
>
> 2. Fails when trying to run a xenserver instance:
> http://paste.openstack.org/show/303/

Fixed the improt error, would be appreciative if you can test whether it runs, I currently don't have any way to test the xen stuff.

> 3. Why does nova-objectstore have to be run the old way and not integrated
> with nova-combined?

It is very reliant on twisted's implementations of things and as it is being deprecated I didn't want to spend time porting it over to eventlet.

>
> I'll see if I can fix #2 in the meanwhile.

[termie (termie)]

Oh, and re logging, it isn't supposed to log to a file, that is for your supervisor daemon to handle (init, upstart and daemonstools all expect output to stdout and handle the logging for you)

[termie (termie)]

> On Tue, Dec 14, 2010 at 6:26 PM, termie <email address hidden> wrote:
> >> I have a feeling the git-bzr bridge is not working properly... I see that
> same
> >> issues with timestamps that I saw when Chris MacGown used the git-bzr
> bridge.
> >> Andy, I assume you are still using git with your git-bzr bridge?
> >>
> >
> > Not git-bzr was used on this branch, I have largely stopped using it because
> I need to maintain patches against both bzr and bzr-fastimport to keep it
> working. I don't know which timestamps you are referring to.
>
> OK.
>
> >> Note that this merge request first *removes* nova/service.py (notice the
> >> 1970-01-01 timestamps, BTW) and then *adds* a file nova/service.py.  I
> don't
> >> know why this is the case and why it doesn't simply show the diff of the
> >> modified nova/service.py.
> >
> > I had another file for the first half of the development of this branch that
> I later overwrote service.py with, it is something of a shame to lose the
> development history of that single file but not worth dealing the bzr's poor
> tools to attempt to reconstruct it.
>
> Because you are used to or prefer git doesn't mean that bzr has poor
> tooling. If you overwrite a file in git, the same thing would happen.
>

I was referring to the tools to pull apart and remake the commit to fix the problem, some equivalent of git's rebase -i (which to my knowledge doesn't exist outside of git), not that it was bzr's fault that there is an issue to begin with. And apologies for the jab at bzr, "bzr's poor tools" would better read "without git's tools."

> <snip>
> > I also fear it will not merge correctly simply because while merging from
> upstream a bunch of spurious conflicts were generated. This patch does need to
> be done all at once.
>
> OK. However, I don't agree, but I can move forward with the review...
>

Sorry, some further explanation: it needs to be done all at once because once one call in a stack has switched from using deferreds to eventlet every other call in that stack has to as well otherwise things begin to return before initiating actual work or return incorrect values (usually a deferred where the result of a deferred was required).

> -jay

[termie (termie)]

yay, it was just a tiny conflict (not sure how it was introduced as it doesn't seem like it could have run the other way), resolved and merged from upstream

[Eric Day (eday)]

Based on the in-person review we did, and seeing that this branch works (minus a few xen issues we can iron out, as that code is changing too), I'm ok with it. I would still like to see the objectstore unittests enabled again, but if we are deprecating this before release it may not be important.

review: Approve

[Jay Pipes (jaypipes)]

Yeah, the conflict was the i18n patch that was merged recently...
sorry about that... looking forward to seeing this work get into trunk
:)

On Wed, Dec 15, 2010 at 2:04 PM, termie <email address hidden> wrote:
> yay, it was just a tiny conflict (not sure how it was introduced as it doesn't seem like it could have run the other way), resolved and merged from upstream
> --
> https://code.launchpad.net/~termie/nova/eventlet_merge/+merge/43383
> Your team Nova Core is requested to review the proposed merge of lp:~termie/nova/eventlet_merge into lp:nova.
>

[Jay Pipes (jaypipes)]

> > On Tue, Dec 14, 2010 at 6:26 PM, termie <email address hidden> wrote:

> I was referring to the tools to pull apart and remake the commit to fix the
> problem, some equivalent of git's rebase -i (which to my knowledge doesn't
> exist outside of git), not that it was bzr's fault that there is an issue to
> begin with. And apologies for the jab at bzr, "bzr's poor tools" would better
> read "without git's tools."

No worries. :)

You may be interested in bzr's rebase plugin ;)

http://doc.bazaar.canonical.com/plugins/en/rebase-plugin.html

> > <snip>
> > > I also fear it will not merge correctly simply because while merging from
> > upstream a bunch of spurious conflicts were generated. This patch does need
> to
> > be done all at once.
> >
> > OK. However, I don't agree, but I can move forward with the review...
> >
>
> Sorry, some further explanation: it needs to be done all at once because once
> one call in a stack has switched from using deferreds to eventlet every other
> call in that stack has to as well otherwise things begin to return before
> initiating actual work or return incorrect values (usually a deferred where
> the result of a deferred was required).

Gotcha. After looking through the code more thoroughly, I think it's a great improvement. I share the same concerns as Eric re: objectstore, but then again, if I can get Glance done, maybe that won't be much of an issue! :)

-jay
> > -jay

review: Approve

[Eric Day (eday)]

There are some pep8 issues with this you'll need to fix before merge will succeed. Just run:

pep8 -r bin/* nova

bin/nova-combined:53:1: W293 blank line contains whitespace
bin/nova-combined:66:1: W391 blank line at end of file
bin/nova-combined:66:1: W293 blank line contains whitespace
nova/service.py:102:1: W293 blank line contains whitespace
nova/service.py:125:20: W291 trailing whitespace
nova/service.py:195:1: W293 blank line contains whitespace
nova/service.py:210:1: W293 blank line contains whitespace
nova/service.py:230:1: W293 blank line contains whitespace
nova/test.py:58:1: E302 expected 2 blank lines, found 1
nova/utils.py:236:9: E301 expected 1 blank line, found 0
nova/utils.py:247:1: W293 blank line contains whitespace
nova/utils.py:251:1: W293 blank line contains whitespace
nova/utils.py:254:1: W293 blank line contains whitespace
nova/compute/manager.py:67:1: W293 blank line contains whitespace
nova/tests/rpc_unittest.py:70:34: W291 trailing whitespace
nova/tests/service_unittest.py:122:1: W293 blank line contains whitespace
nova/virt/libvirt_conn.py:208:1: W293 blank line contains whitespace
nova/virt/libvirt_conn.py:390:1: W293 blank line contains whitespace
nova/virt/libvirt_conn.py:442:1: W293 blank line contains whitespace
nova/virt/libvirt_conn.py:444:15: E111 indentation is not a multiple of four
nova/virt/xenapi_conn.py:203:64: W291 trailing whitespace

[OpenStack Infra (hudson-openstack)]

The attempt to merge lp:~termie/nova/eventlet_merge into lp:nova failed. Below is the output from the failed tests.

bin/nova-combined:53:1: W293 blank line contains whitespace

^
    JCR: Trailing whitespace is superfluous.
    FBM: Except when it occurs as part of a blank line (i.e. the line is
         nothing but whitespace). According to Python docs[1] a line with only
         whitespace is considered a blank line, and is to be ignored. However,
         matching a blank line to its indentation level avoids mistakenly
         terminating a multi-line statement (e.g. class declaration) when
         pasting code into the standard Python interpreter.

         [1] http://docs.python.org/reference/lexical_analysis.html#blank-lines

    The warning returned varies on whether the line itself is blank, for easier
    filtering for those who want to indent their blank lines.

    Okay: spam(1)
    W291: spam(1)\s
    W293: class Foo(object):\n \n bang = 12
bin/nova-combined:66:1: W391 blank line at end of file

^
    JCR: Trailing blank lines are superfluous.

    Okay: spam(1)
    W391: spam(1)\n
bin/nova-combined:66:1: W293 blank line contains whitespace

^
    JCR: Trailing whitespace is superfluous.
    FBM: Except when it occurs as part of a blank line (i.e. the line is
         nothing but whitespace). According to Python docs[1] a line with only
         whitespace is considered a blank line, and is to be ignored. However,
         matching a blank line to its indentation level avoids mistakenly
         terminating a multi-line statement (e.g. class declaration) when
         pasting code into the standard Python interpreter.

         [1] http://docs.python.org/reference/lexical_analysis.html#blank-lines

    The warning returned varies on whether the line itself is blank, for easier
    filtering for those who want to indent their blank lines.

    Okay: spam(1)
    W291: spam(1)\s
    W293: class Foo(object):\n \n bang = 12

[lp:~termie/nova/eventlet_merge]

400. By termie on 2010-12-16

pep8 fixes for bin

=== modified file 'bin/nova-api'
--- bin/nova-api	2010-12-11 20:10:24 +0000
+++ bin/nova-api	2010-12-16 20:49:10 +0000
@@ -17,9 +17,8 @@
 #    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
-"""
-Nova API daemon.
-"""
+
+"""Starter script for Nova API."""
 
 import gettext
 import os
@@ -35,9 +34,11 @@
 
 gettext.install('nova', unicode=1)
 
+from nova import api
 from nova import flags
 from nova import utils
-from nova import server
+from nova import wsgi
+
 
 FLAGS = flags.FLAGS
 flags.DEFINE_integer('osapi_port', 8774, 'OpenStack API port')
@@ -46,15 +47,10 @@
 flags.DEFINE_string('ec2api_host', '0.0.0.0', 'EC2 API host')
 
 
-def main(_args):
-    from nova import api
-    from nova import wsgi
+if __name__ == '__main__':
+    utils.default_flagfile()
+    FLAGS(sys.argv)
     server = wsgi.Server()
     server.start(api.API('os'), FLAGS.osapi_port, host=FLAGS.osapi_host)
     server.start(api.API('ec2'), FLAGS.ec2api_port, host=FLAGS.ec2api_host)
     server.wait()
-
-
-if __name__ == '__main__':
-    utils.default_flagfile()
-    server.serve('nova-api', main)

=== added file 'bin/nova-combined'
--- bin/nova-combined	1970-01-01 00:00:00 +0000
+++ bin/nova-combined	2010-12-16 20:49:10 +0000
@@ -0,0 +1,65 @@
+#!/usr/bin/env python
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Combined starter script for Nova services."""
+
+import eventlet
+eventlet.monkey_patch()
+
+import os
+import sys
+
+# If ../nova/__init__.py exists, add ../ to Python search path, so that
+# it will override what happens to be installed in /usr/(local/)lib/python...
+possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
+                                   os.pardir,
+                                   os.pardir))
+if os.path.exists(os.path.join(possible_topdir, 'nova', '__init__.py')):
+    sys.path.insert(0, possible_topdir)
+
+from nova import api
+from nova import flags
+from nova import service
+from nova import utils
+from nova import wsgi
+
+
+FLAGS = flags.FLAGS
+flags.DEFINE_integer('osapi_port', 8774, 'OpenStack API port')
+flags.DEFINE_string('osapi_host', '0.0.0.0', 'OpenStack API host')
+flags.DEFINE_integer('ec2api_port', 8773, 'EC2 API port')
+flags.DEFINE_string('ec2api_host', '0.0.0.0', 'EC2 API host')
+
+
+if __name__ == '__main__':
+    utils.default_flagfile()
+    FLAGS(sys.argv)
+
+    compute = service.Service.create(binary='nova-compute')
+    network = service.Service.create(binary='nova-network')
+    volume = service.Service.create(binary='nova-volume')
+    scheduler = service.Service.create(binary='nova-scheduler')
+    #objectstore = service.Service.create(binary='nova-objectstore')
+
+    service.serve(compute, network, volume, scheduler)
+
+    server = wsgi.Server()
+    server.start(api.API('os'), FLAGS.osapi_port, host=FLAGS.osapi_host)
+    server.start(api.API('ec2'), FLAGS.ec2api_port, host=FLAGS.ec2api_host)
+    server.wait()

=== modified file 'bin/nova-compute'
--- bin/nova-compute	2010-12-11 20:10:24 +0000
+++ bin/nova-compute	2010-12-16 20:49:10 +0000
@@ -17,9 +17,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-"""
-  Twistd daemon for the nova compute nodes.
-"""
+"""Starter script for Nova Compute."""
+
+import eventlet
+eventlet.monkey_patch()
 
 import gettext
 import os
@@ -36,13 +37,9 @@
 gettext.install('nova', unicode=1)
 
 from nova import service
-from nova import twistd
 from nova import utils
 
-
 if __name__ == '__main__':
     utils.default_flagfile()
-    twistd.serve(__file__)
-
-if __name__ == '__builtin__':
-    application = service.Service.create()  # pylint: disable=C0103
+    service.serve()
+    service.wait()

=== modified file 'bin/nova-network'
--- bin/nova-network	2010-12-14 23:22:03 +0000
+++ bin/nova-network	2010-12-16 20:49:10 +0000
@@ -17,9 +17,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-"""
-  Twistd daemon for the nova network nodes.
-"""
+"""Starter script for Nova Network."""
+
+import eventlet
+eventlet.monkey_patch()
 
 import gettext
 import os
@@ -36,13 +37,9 @@
 gettext.install('nova', unicode=1)
 
 from nova import service
-from nova import twistd
 from nova import utils
 
-
 if __name__ == '__main__':
     utils.default_flagfile()
-    twistd.serve(__file__)
-
-if __name__ == '__builtin__':
-    application = service.Service.create()  # pylint: disable-msg=C0103
+    service.serve()
+    service.wait()

=== modified file 'bin/nova-scheduler'
--- bin/nova-scheduler	2010-12-14 23:22:03 +0000
+++ bin/nova-scheduler	2010-12-16 20:49:10 +0000
@@ -17,9 +17,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-"""
-  Twistd daemon for the nova scheduler nodes.
-"""
+"""Starter script for Nova Scheduler."""
+
+import eventlet
+eventlet.monkey_patch()
 
 import gettext
 import os
@@ -36,13 +37,9 @@
 gettext.install('nova', unicode=1)
 
 from nova import service
-from nova import twistd
 from nova import utils
 
-
 if __name__ == '__main__':
     utils.default_flagfile()
-    twistd.serve(__file__)
-
-if __name__ == '__builtin__':
-    application = service.Service.create()
+    service.serve()
+    service.wait()

=== modified file 'bin/nova-volume'
--- bin/nova-volume	2010-12-14 23:22:03 +0000
+++ bin/nova-volume	2010-12-16 20:49:10 +0000
@@ -17,9 +17,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-"""
-  Twistd daemon for the nova volume nodes.
-"""
+"""Starter script for Nova Volume."""
+
+import eventlet
+eventlet.monkey_patch()
 
 import gettext
 import os
@@ -36,13 +37,9 @@
 gettext.install('nova', unicode=1)
 
 from nova import service
-from nova import twistd
 from nova import utils
 
-
 if __name__ == '__main__':
     utils.default_flagfile()
-    twistd.serve(__file__)
-
-if __name__ == '__builtin__':
-    application = service.Service.create()  # pylint: disable-msg=C0103
+    service.serve()
+    service.wait()

=== modified file 'nova/compute/disk.py'
--- nova/compute/disk.py	2010-11-16 18:49:18 +0000
+++ nova/compute/disk.py	2010-12-16 20:49:10 +0000
@@ -26,8 +26,6 @@
 import os
 import tempfile
 
-from twisted.internet import defer
-
 from nova import exception
 from nova import flags
 
@@ -39,7 +37,6 @@
                      'block_size to use for dd')
 
 
[e-mail]
 def partition(infile, outfile, local_bytes=0, resize=True,
               local_type='ext2', execute=None):
     """
@@ -64,10 +61,10 @@
     file_size = os.path.getsize(infile)
     if resize and file_size < FLAGS.minimum_root_size:
         last_sector = FLAGS.minimum_root_size / sector_size - 1
-        yield execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
-                      % (infile, last_sector, sector_size))
-        yield execute('e2fsck -fp %s' % infile, check_exit_code=False)
-        yield execute('resize2fs %s' % infile)
+        execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
+                % (infile, last_sector, sector_size))
+        execute('e2fsck -fp %s' % infile, check_exit_code=False)
+        execute('resize2fs %s' % infile)
         file_size = FLAGS.minimum_root_size
     elif file_size % sector_size != 0:
         logging.warn("Input partition size not evenly divisible by"
@@ -86,30 +83,29 @@
     last_sector = local_last  # e
 
     # create an empty file
-    yield execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
-                  % (outfile, mbr_last, sector_size))
+    execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
+            % (outfile, mbr_last, sector_size))
 
     # make mbr partition
-    yield execute('parted --script %s mklabel msdos' % outfile)
+    execute('parted --script %s mklabel msdos' % outfile)
 
     # append primary file
-    yield execute('dd if=%s of=%s bs=%s conv=notrunc,fsync oflag=append'
-                  % (infile, outfile, FLAGS.block_size))
+    execute('dd if=%s of=%s bs=%s conv=notrunc,fsync oflag=append'
+            % (infile, outfile, FLAGS.block_size))
 
     # make primary partition
-    yield execute('parted --script %s mkpart primary %ds %ds'
-                  % (outfile, primary_first, primary_last))
+    execute('parted --script %s mkpart primary %ds %ds'
+            % (outfile, primary_first, primary_last))
 
     if local_bytes > 0:
         # make the file bigger
-        yield execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
-                      % (outfile, last_sector, sector_size))
+        execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
+                % (outfile, last_sector, sector_size))
         # make and format local partition
-        yield execute('parted --script %s mkpartfs primary %s %ds %ds'
-                      % (outfile, local_type, local_first, local_last))
-
-
[e-mail]
+        execute('parted --script %s mkpartfs primary %s %ds %ds'
+                % (outfile, local_type, local_first, local_last))
+
+
 def inject_data(image, key=None, net=None, partition=None, execute=None):
     """Injects a ssh key and optionally net data into a disk image.
 
@@ -119,26 +115,26 @@
     If partition is not specified it mounts the image as a single partition.
 
     """
-    out, err = yield execute('sudo losetup -f --show %s' % image)
+    out, err = execute('sudo losetup -f --show %s' % image)
     if err:
         raise exception.Error('Could not attach image to loopback: %s' % err)
     device = out.strip()
     try:
         if not partition is None:
             # create partition
-            out, err = yield execute('sudo kpartx -a %s' % device)
+            out, err = execute('sudo kpartx -a %s' % device)
             if err:
                 raise exception.Error('Failed to load partition: %s' % err)
             mapped_device = '/dev/mapper/%sp%s' % (device.split('/')[-1],
                                                    partition)
         else:
             mapped_device = device
-        out, err = yield execute('sudo tune2fs -c 0 -i 0 %s' % mapped_device)
+        out, err = execute('sudo tune2fs -c 0 -i 0 %s' % mapped_device)
 
         tmpdir = tempfile.mkdtemp()
         try:
             # mount loopback to dir
-            out, err = yield execute(
+            out, err = execute(
                     'sudo mount %s %s' % (mapped_device, tmpdir))
             if err:
                 raise exception.Error('Failed to mount filesystem: %s' % err)
@@ -146,24 +142,23 @@
             try:
                 if key:
                     # inject key file
-                    yield _inject_key_into_fs(key, tmpdir, execute=execute)
+                    _inject_key_into_fs(key, tmpdir, execute=execute)
                 if net:
-                    yield _inject_net_into_fs(net, tmpdir, execute=execute)
+                    _inject_net_into_fs(net, tmpdir, execute=execute)
             finally:
                 # unmount device
-                yield execute('sudo umount %s' % mapped_device)
+                execute('sudo umount %s' % mapped_device)
         finally:
             # remove temporary directory
-            yield execute('rmdir %s' % tmpdir)
+            execute('rmdir %s' % tmpdir)
             if not partition is None:
                 # remove partitions
-                yield execute('sudo kpartx -d %s' % device)
+                execute('sudo kpartx -d %s' % device)
     finally:
         # remove loopback
-        yield execute('sudo losetup -d %s' % device)
-
-
[e-mail]
+        execute('sudo losetup -d %s' % device)
+
+
 def _inject_key_into_fs(key, fs, execute=None):
     """Add the given public ssh key to root's authorized_keys.
 
@@ -171,22 +166,21 @@
     fs is the path to the base of the filesystem into which to inject the key.
     """
     sshdir = os.path.join(os.path.join(fs, 'root'), '.ssh')
-    yield execute('sudo mkdir -p %s' % sshdir)  # existing dir doesn't matter
-    yield execute('sudo chown root %s' % sshdir)
-    yield execute('sudo chmod 700 %s' % sshdir)
+    execute('sudo mkdir -p %s' % sshdir)  # existing dir doesn't matter
+    execute('sudo chown root %s' % sshdir)
+    execute('sudo chmod 700 %s' % sshdir)
     keyfile = os.path.join(sshdir, 'authorized_keys')
-    yield execute('sudo tee -a %s' % keyfile, '\n' + key.strip() + '\n')
-
-
[e-mail]
+    execute('sudo tee -a %s' % keyfile, '\n' + key.strip() + '\n')
+
+
 def _inject_net_into_fs(net, fs, execute=None):
     """Inject /etc/network/interfaces into the filesystem rooted at fs.
 
     net is the contents of /etc/network/interfaces.
     """
     netdir = os.path.join(os.path.join(fs, 'etc'), 'network')
-    yield execute('sudo mkdir -p %s' % netdir)  # existing dir doesn't matter
-    yield execute('sudo chown root:root %s' % netdir)
-    yield execute('sudo chmod 755 %s' % netdir)
+    execute('sudo mkdir -p %s' % netdir)  # existing dir doesn't matter
+    execute('sudo chown root:root %s' % netdir)
+    execute('sudo chmod 755 %s' % netdir)
     netfile = os.path.join(netdir, 'interfaces')
-    yield execute('sudo tee %s' % netfile, net)
+    execute('sudo tee %s' % netfile, net)

=== modified file 'nova/compute/manager.py'
--- nova/compute/manager.py	2010-12-02 18:58:13 +0000
+++ nova/compute/manager.py	2010-12-16 20:49:10 +0000
@@ -37,8 +37,6 @@
 import datetime
 import logging
 
-from twisted.internet import defer
-
 from nova import exception
 from nova import flags
 from nova import manager
@@ -78,13 +76,11 @@
             state = power_state.NOSTATE
         self.db.instance_set_state(context, instance_id, state)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def refresh_security_group(self, context, security_group_id, **_kwargs):
         """This call passes stright through to the virtualization driver."""
-        yield self.driver.refresh_security_group(security_group_id)
+        self.driver.refresh_security_group(security_group_id)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def run_instance(self, context, instance_id, **_kwargs):
         """Launch a new instance with specified options."""
@@ -105,7 +101,7 @@
                                    'spawning')
 
         try:
-            yield self.driver.spawn(instance_ref)
+            self.driver.spawn(instance_ref)
             now = datetime.datetime.utcnow()
             self.db.instance_update(context,
                                     instance_id,
@@ -119,7 +115,6 @@
 
         self._update_state(context, instance_id)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def terminate_instance(self, context, instance_id):
         """Terminate an instance on this machine."""
@@ -134,12 +129,11 @@
             self.db.instance_destroy(context, instance_id)
             raise exception.Error('trying to destroy already destroyed'
                                   ' instance: %s' % instance_id)
-        yield self.driver.destroy(instance_ref)
+        self.driver.destroy(instance_ref)
 
         # TODO(ja): should we keep it in a terminated state for a bit?
         self.db.instance_destroy(context, instance_id)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def reboot_instance(self, context, instance_id):
         """Reboot an instance on this server."""
@@ -159,10 +153,9 @@
                                    instance_id,
                                    power_state.NOSTATE,
                                    'rebooting')
-        yield self.driver.reboot(instance_ref)
+        self.driver.reboot(instance_ref)
         self._update_state(context, instance_id)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def rescue_instance(self, context, instance_id):
         """Rescue an instance on this server."""
@@ -175,10 +168,9 @@
                                    instance_id,
                                    power_state.NOSTATE,
                                    'rescuing')
-        yield self.driver.rescue(instance_ref)
+        self.driver.rescue(instance_ref)
         self._update_state(context, instance_id)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def unrescue_instance(self, context, instance_id):
         """Rescue an instance on this server."""
@@ -191,7 +183,7 @@
                                    instance_id,
                                    power_state.NOSTATE,
                                    'unrescuing')
-        yield self.driver.unrescue(instance_ref)
+        self.driver.unrescue(instance_ref)
         self._update_state(context, instance_id)
 
     @exception.wrap_exception
@@ -203,7 +195,6 @@
 
         return self.driver.get_console_output(instance_ref)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def attach_volume(self, context, instance_id, volume_id, mountpoint):
         """Attach a volume to an instance."""
@@ -211,12 +202,12 @@
         logging.debug("instance %s: attaching volume %s to %s", instance_id,
             volume_id, mountpoint)
         instance_ref = self.db.instance_get(context, instance_id)
-        dev_path = yield self.volume_manager.setup_compute_volume(context,
-                                                                  volume_id)
+        dev_path = self.volume_manager.setup_compute_volume(context,
+                                                            volume_id)
         try:
-            yield self.driver.attach_volume(instance_ref['name'],
-                                            dev_path,
-                                            mountpoint)
+            self.driver.attach_volume(instance_ref['name'],
+                                      dev_path,
+                                      mountpoint)
             self.db.volume_attached(context,
                                     volume_id,
                                     instance_id,
@@ -227,12 +218,12 @@
             #             ecxception below.
             logging.exception("instance %s: attach failed %s, removing",
                               instance_id, mountpoint)
-            yield self.volume_manager.remove_compute_volume(context,
-                                                            volume_id)
+            self.volume_manager.remove_compute_volume(context,
+                                                      volume_id)
             raise exc
-        defer.returnValue(True)
-
-    @defer.inlineCallbacks
+
+        return True
+
     @exception.wrap_exception
     def detach_volume(self, context, instance_id, volume_id):
         """Detach a volume from an instance."""
@@ -246,8 +237,8 @@
             logging.warn("Detaching volume from unknown instance %s",
                          instance_ref['name'])
         else:
-            yield self.driver.detach_volume(instance_ref['name'],
-                                            volume_ref['mountpoint'])
-        yield self.volume_manager.remove_compute_volume(context, volume_id)
+            self.driver.detach_volume(instance_ref['name'],
+                                      volume_ref['mountpoint'])
+        self.volume_manager.remove_compute_volume(context, volume_id)
         self.db.volume_detached(context, volume_id)
-        defer.returnValue(True)
+        return True

=== modified file 'nova/flags.py'
--- nova/flags.py	2010-12-03 20:21:18 +0000
+++ nova/flags.py	2010-12-16 20:49:10 +0000
@@ -159,6 +159,7 @@
                 return str(val)
         raise KeyError(name)
 
+
 FLAGS = FlagValues()
 gflags.FLAGS = FLAGS
 gflags.DEFINE_flag(gflags.HelpFlag(), FLAGS)
@@ -183,6 +184,12 @@
 DEFINE_spaceseplist = _wrapper(gflags.DEFINE_spaceseplist)
 DEFINE_multistring = _wrapper(gflags.DEFINE_multistring)
 DEFINE_multi_int = _wrapper(gflags.DEFINE_multi_int)
+DEFINE_flag = _wrapper(gflags.DEFINE_flag)
+
+
+HelpFlag = gflags.HelpFlag
+HelpshortFlag = gflags.HelpshortFlag
+HelpXMLFlag = gflags.HelpXMLFlag
 
 
 def DECLARE(name, module_string, flag_values=FLAGS):

=== modified file 'nova/manager.py'
--- nova/manager.py	2010-12-01 17:24:39 +0000
+++ nova/manager.py	2010-12-16 20:49:10 +0000
@@ -55,7 +55,6 @@
 from nova import flags
 from nova.db import base
 
-from twisted.internet import defer
 
 FLAGS = flags.FLAGS
 
@@ -67,10 +66,9 @@
         self.host = host
         super(Manager, self).__init__(db_driver)
 
-    @defer.inlineCallbacks
     def periodic_tasks(self, context=None):
         """Tasks to be run at a periodic interval"""
-        yield
+        pass
 
     def init_host(self):
         """Do any initialization that needs to be run if this is a standalone

=== modified file 'nova/network/manager.py'
--- nova/network/manager.py	2010-11-23 23:56:26 +0000
+++ nova/network/manager.py	2010-12-16 20:49:10 +0000
@@ -49,7 +49,6 @@
 import math
 
 import IPy
-from twisted.internet import defer
 
 from nova import context
 from nova import db
@@ -399,10 +398,9 @@
     instances in its subnet.
     """
 
-    @defer.inlineCallbacks
     def periodic_tasks(self, context=None):
         """Tasks to be run at a periodic interval."""
-        yield super(VlanManager, self).periodic_tasks(context)
+        super(VlanManager, self).periodic_tasks(context)
         now = datetime.datetime.utcnow()
         timeout = FLAGS.fixed_ip_disassociate_timeout
         time = now - datetime.timedelta(seconds=timeout)

=== modified file 'nova/objectstore/image.py'
--- nova/objectstore/image.py	2010-12-14 00:20:27 +0000
+++ nova/objectstore/image.py	2010-12-16 20:49:10 +0000
@@ -267,6 +267,7 @@
         if err:
             raise exception.Error("Failed to decrypt initialization "
                                   "vector: %s" % err)
+
         _out, err = utils.execute(
                 'openssl enc -d -aes-128-cbc -in %s -K %s -iv %s -out %s'
                  % (encrypted_filename, key, iv, decrypted_filename),

=== removed file 'nova/process.py'
--- nova/process.py	2010-10-21 18:49:51 +0000
+++ nova/process.py	1970-01-01 00:00:00 +0000
@@ -1,209 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# Copyright 2010 FathomDB Inc.
-# All Rights Reserved.
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-"""
-Process pool using twisted threading
-"""
-
-import logging
-import StringIO
-
-from twisted.internet import defer
-from twisted.internet import error
-from twisted.internet import protocol
-from twisted.internet import reactor
-
-from nova import flags
-from nova.exception import ProcessExecutionError
-
-FLAGS = flags.FLAGS
-flags.DEFINE_integer('process_pool_size', 4,
-                     'Number of processes to use in the process pool')
-
-
-# This is based on _BackRelay from twister.internal.utils, but modified to
-#  capture both stdout and stderr, without odd stderr handling, and also to
-#  handle stdin
-class BackRelayWithInput(protocol.ProcessProtocol):
-    """
-    Trivial protocol for communicating with a process and turning its output
-    into the result of a L{Deferred}.
-
-    @ivar deferred: A L{Deferred} which will be called back with all of stdout
-        and all of stderr as well (as a tuple).  C{terminate_on_stderr} is true
-        and any bytes are received over stderr, this will fire with an
-        L{_ProcessExecutionError} instance and the attribute will be set to
-        C{None}.
-
-    @ivar onProcessEnded: If C{terminate_on_stderr} is false and bytes are
-        received over stderr, this attribute will refer to a L{Deferred} which
-        will be called back when the process ends.  This C{Deferred} is also
-        associated with the L{_ProcessExecutionError} which C{deferred} fires
-        with earlier in this case so that users can determine when the process
-        has actually ended, in addition to knowing when bytes have been
-        received via stderr.
-    """
-
-    def __init__(self, deferred, cmd, started_deferred=None,
-                 terminate_on_stderr=False, check_exit_code=True,
-                 process_input=None):
-        self.deferred = deferred
-        self.cmd = cmd
-        self.stdout = StringIO.StringIO()
-        self.stderr = StringIO.StringIO()
-        self.started_deferred = started_deferred
-        self.terminate_on_stderr = terminate_on_stderr
-        self.check_exit_code = check_exit_code
-        self.process_input = process_input
-        self.on_process_ended = None
-
-    def _build_execution_error(self, exit_code=None):
-        return ProcessExecutionError(cmd=self.cmd,
-                                     exit_code=exit_code,
-                                     stdout=self.stdout.getvalue(),
-                                     stderr=self.stderr.getvalue())
-
-    def errReceived(self, text):
-        self.stderr.write(text)
-        if self.terminate_on_stderr and (self.deferred is not None):
-            self.on_process_ended = defer.Deferred()
-            self.deferred.errback(self._build_execution_error())
-            self.deferred = None
-            self.transport.loseConnection()
-
-    def outReceived(self, text):
-        self.stdout.write(text)
-
-    def processEnded(self, reason):
-        if self.deferred is not None:
-            stdout, stderr = self.stdout.getvalue(), self.stderr.getvalue()
-            exit_code = reason.value.exitCode
-            if self.check_exit_code and exit_code != 0:
-                self.deferred.errback(self._build_execution_error(exit_code))
-            else:
-                try:
-                    if self.check_exit_code:
-                        reason.trap(error.ProcessDone)
-                    self.deferred.callback((stdout, stderr))
-                except:
-                    # NOTE(justinsb): This logic is a little suspicious to me.
-                    # If the callback throws an exception, then errback will
-                    # be called also. However, this is what the unit tests
-                    # test for.
-                    exec_error = self._build_execution_error(exit_code)
-                    self.deferred.errback(exec_error)
-        elif self.on_process_ended is not None:
-            self.on_process_ended.errback(reason)
-
-    def connectionMade(self):
-        if self.started_deferred:
-            self.started_deferred.callback(self)
-        if self.process_input:
-            self.transport.write(str(self.process_input))
-        self.transport.closeStdin()
-
-
-def get_process_output(executable, args=None, env=None, path=None,
-                       process_reactor=None, check_exit_code=True,
-                       process_input=None, started_deferred=None,
-                       terminate_on_stderr=False):
-    if process_reactor is None:
-        process_reactor = reactor
-    args = args and args or ()
-    env = env and env and {}
-    deferred = defer.Deferred()
-    cmd = executable
-    if args:
-        cmd = " ".join([cmd] + args)
-    logging.debug("Running cmd: %s", cmd)
-    process_handler = BackRelayWithInput(
-            deferred,
-            cmd,
-            started_deferred=started_deferred,
-            check_exit_code=check_exit_code,
-            process_input=process_input,
-            terminate_on_stderr=terminate_on_stderr)
-    # NOTE(vish): commands come in as unicode, but self.executes needs
-    #             strings or process.spawn raises a deprecation warning
-    executable = str(executable)
-    if not args is None:
-        args = [str(x) for x in args]
-    process_reactor.spawnProcess(process_handler, executable,
-                                 (executable,) + tuple(args), env, path)
-    return deferred
-
-
-class ProcessPool(object):
-    """ A simple process pool implementation using Twisted's Process bits.
-
-    This is pretty basic right now, but hopefully the API will be the correct
-    one so that it can be optimized later.
-    """
-    def __init__(self, size=None):
-        self.size = size and size or FLAGS.process_pool_size
-        self._pool = defer.DeferredSemaphore(self.size)
-
-    def simple_execute(self, cmd, **kw):
-        """ Weak emulation of the old utils.execute() function.
-
-        This only exists as a way to quickly move old execute methods to
-        this new style of code.
-
-        NOTE(termie): This will break on args with spaces in them.
-        """
-        parsed = cmd.split(' ')
-        executable, args = parsed[0], parsed[1:]
-        return self.execute(executable, args, **kw)
-
-    def execute(self, *args, **kw):
-        deferred = self._pool.acquire()
-
-        def _associate_process(proto):
-            deferred.process = proto.transport
-            return proto.transport
-
-        started = defer.Deferred()
-        started.addCallback(_associate_process)
-        kw.setdefault('started_deferred', started)
-
-        deferred.process = None
-        deferred.started = started
-
-        deferred.addCallback(lambda _: get_process_output(*args, **kw))
-        deferred.addBoth(self._release)
-        return deferred
-
-    def _release(self, retval=None):
-        self._pool.release()
-        return retval
-
-
-class SharedPool(object):
-    _instance = None
-
-    def __init__(self):
-        if SharedPool._instance is None:
-            self.__class__._instance = ProcessPool()
-
-    def __getattr__(self, key):
-        return getattr(self._instance, key)
-
-
-def simple_execute(cmd, **kwargs):
-    return SharedPool().simple_execute(cmd, **kwargs)

=== modified file 'nova/rpc.py'
--- nova/rpc.py	2010-11-22 22:48:44 +0000
+++ nova/rpc.py	2010-12-16 20:49:10 +0000
@@ -25,18 +25,18 @@
 import logging
 import sys
 import time
+import traceback
 import uuid
 
 from carrot import connection as carrot_connection
 from carrot import messaging
 from eventlet import greenthread
-from twisted.internet import defer
-from twisted.internet import task
 
+from nova import context
 from nova import exception
 from nova import fakerabbit
 from nova import flags
-from nova import context
+from nova import utils
 
 
 FLAGS = flags.FLAGS
@@ -128,17 +128,9 @@
 
     def attach_to_eventlet(self):
         """Only needed for unit tests!"""
-        def fetch_repeatedly():
-            while True:
-                self.fetch(enable_callbacks=True)
-                greenthread.sleep(0.1)
-        greenthread.spawn(fetch_repeatedly)
-
-    def attach_to_twisted(self):
-        """Attach a callback to twisted that fires 10 times a second"""
-        loop = task.LoopingCall(self.fetch, enable_callbacks=True)
-        loop.start(interval=0.1)
-        return loop
+        timer = utils.LoopingCall(self.fetch, enable_callbacks=True)
+        timer.start(0.1)
+        return timer
 
 
 class Publisher(messaging.Publisher):
@@ -196,11 +188,13 @@
         node_func = getattr(self.proxy, str(method))
         node_args = dict((str(k), v) for k, v in args.iteritems())
         # NOTE(vish): magic is fun!
-        # pylint: disable-msg=W0142
-        d = defer.maybeDeferred(node_func, context=ctxt, **node_args)
-        if msg_id:
-            d.addCallback(lambda rval: msg_reply(msg_id, rval, None))
-            d.addErrback(lambda e: msg_reply(msg_id, None, e))
+        try:
+            rval = node_func(context=ctxt, **node_args)
+            if msg_id:
+                msg_reply(msg_id, rval, None)
+        except Exception as e:
+            if msg_id:
+                msg_reply(msg_id, None, sys.exc_info())
         return
 
 
@@ -242,13 +236,15 @@
 def msg_reply(msg_id, reply=None, failure=None):
     """Sends a reply or an error on the channel signified by msg_id
 
-    failure should be a twisted failure object"""
+    failure should be a sys.exc_info() tuple.
+
+    """
     if failure:
-        message = failure.getErrorMessage()
-        traceback = failure.getTraceback()
+        message = str(failure[1])
+        tb = traceback.format_exception(*failure)
         logging.error("Returning exception %s to caller", message)
-        logging.error(traceback)
-        failure = (failure.type.__name__, str(failure.value), traceback)
+        logging.error(tb)
+        failure = (failure[0].__name__, str(failure[1]), tb)
     conn = Connection.instance()
     publisher = DirectPublisher(connection=conn, msg_id=msg_id)
     try:
@@ -313,7 +309,6 @@
     _pack_context(msg, context)
 
     class WaitMessage(object):
-
         def __call__(self, data, message):
             """Acks message and sets result."""
             message.ack()
@@ -337,41 +332,15 @@
     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
 
 
-def call_twisted(context, topic, msg):
-    """Sends a message on a topic and wait for a response"""
-    LOG.debug("Making asynchronous call...")
-    msg_id = uuid.uuid4().hex
-    msg.update({'_msg_id': msg_id})
-    LOG.debug("MSG_ID is %s" % (msg_id))
-    _pack_context(msg, context)
-
-    conn = Connection.instance()
-    d = defer.Deferred()
-    consumer = DirectConsumer(connection=conn, msg_id=msg_id)
-
-    def deferred_receive(data, message):
-        """Acks message and callbacks or errbacks"""
-        message.ack()
-        if data['failure']:
-            return d.errback(RemoteError(*data['failure']))
-        else:
-            return d.callback(data['result'])
-
-    consumer.register_callback(deferred_receive)
-    injected = consumer.attach_to_twisted()
-
-    # clean up after the injected listened and return x
-    d.addCallback(lambda x: injected.stop() and x or x)
-
-    publisher = TopicPublisher(connection=conn, topic=topic)
-    publisher.send(msg)
-    publisher.close()
-    return d
-
-
 def cast(context, topic, msg):
     """Sends a message on a topic without waiting for a response"""
     LOG.debug("Making asynchronous cast...")

=== removed file 'nova/server.py'
--- nova/server.py	2010-11-23 20:52:00 +0000
+++ nova/server.py	1970-01-01 00:00:00 +0000
@@ -1,151 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# All Rights Reserved.
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-"""
-Base functionality for nova daemons - gradually being replaced with twistd.py.
-"""
-
-import daemon
-from daemon import pidlockfile
-import logging
-import logging.handlers
-import os
-import signal
-import sys
-import time
-
-from nova import flags
-
-
-FLAGS = flags.FLAGS
-flags.DEFINE_bool('daemonize', False, 'daemonize this process')
-# NOTE(termie): right now I am defaulting to using syslog when we daemonize
-#               it may be better to do something else -shrug-
-# NOTE(Devin): I think we should let each process have its own log file
-#              and put it in /var/logs/nova/(appname).log
-#              This makes debugging much easier and cuts down on sys log
-#              clutter.
-flags.DEFINE_bool('use_syslog', True, 'output to syslog when daemonizing')
-flags.DEFINE_string('logfile', None, 'log file to output to')
-flags.DEFINE_string('logdir',  None, 'directory to keep log files in '
-                                     '(will be prepended to $logfile)')
-flags.DEFINE_string('pidfile', None, 'pid file to output to')
-flags.DEFINE_string('working_directory', './', 'working directory...')
-flags.DEFINE_integer('uid', os.getuid(), 'uid under which to run')
-flags.DEFINE_integer('gid', os.getgid(), 'gid under which to run')
-
-
-def stop(pidfile):
-    """
-    Stop the daemon
-    """
-    # Get the pid from the pidfile
-    try:
-        pid = int(open(pidfile, 'r').read().strip())
-    except IOError:
-        message = "pidfile %s does not exist. Daemon not running?\n"
-        sys.stderr.write(message % pidfile)
-        return
-
-    # Try killing the daemon process
-    try:
-        while 1:
-            os.kill(pid, signal.SIGTERM)
-            time.sleep(0.1)
-    except OSError, err:
-        err = str(err)
-        if err.find("No such process") > 0:
-            if os.path.exists(pidfile):
-                os.remove(pidfile)
-        else:
-            print str(err)
-            sys.exit(1)
-
-
-def serve(name, main):
-    """Controller for server"""
-    argv = FLAGS(sys.argv)
-
-    if not FLAGS.pidfile:
-        FLAGS.pidfile = '%s.pid' % name
-
-    logging.debug("Full set of FLAGS: \n\n\n")
-    for flag in FLAGS:
-        logging.debug("%s : %s", flag, FLAGS.get(flag, None))
-
-    action = 'start'
-    if len(argv) > 1:
-        action = argv.pop()
-
-    if action == 'stop':
-        stop(FLAGS.pidfile)
-        sys.exit()
-    elif action == 'restart':
-        stop(FLAGS.pidfile)
-    elif action == 'start':
-        pass
-    else:
-        print 'usage: %s [options] [start|stop|restart]' % argv[0]
-        sys.exit(1)
-    daemonize(argv, name, main)
-
-
-def daemonize(args, name, main):
-    """Does the work of daemonizing the process"""
-    logging.getLogger('amqplib').setLevel(logging.WARN)
-    files_to_keep = []
-    if FLAGS.daemonize:
-        logger = logging.getLogger()
-        formatter = logging.Formatter(
-                name + '(%(name)s): %(levelname)s %(message)s')
-        if FLAGS.use_syslog and not FLAGS.logfile:
-            syslog = logging.handlers.SysLogHandler(address='/dev/log')
-            syslog.setFormatter(formatter)
-            logger.addHandler(syslog)
-            files_to_keep.append(syslog.socket)
-        else:
-            if not FLAGS.logfile:
-                FLAGS.logfile = '%s.log' % name
-            if FLAGS.logdir:
-                FLAGS.logfile = os.path.join(FLAGS.logdir, FLAGS.logfile)
-            logfile = logging.FileHandler(FLAGS.logfile)
-            logfile.setFormatter(formatter)
-            logger.addHandler(logfile)
-            files_to_keep.append(logfile.stream)
-        stdin, stdout, stderr = None, None, None
-    else:
-        stdin, stdout, stderr = sys.stdin, sys.stdout, sys.stderr
-
-    if FLAGS.verbose:
-        logging.getLogger().setLevel(logging.DEBUG)
-    else:
-        logging.getLogger().setLevel(logging.WARNING)
-
-    with daemon.DaemonContext(
-            detach_process=FLAGS.daemonize,
-            working_directory=FLAGS.working_directory,
-            pidfile=pidlockfile.TimeoutPIDLockFile(FLAGS.pidfile,
-                                                   acquire_timeout=1,
-                                                   threaded=False),
-            stdin=stdin,
-            stdout=stdout,
-            stderr=stderr,
-            uid=FLAGS.uid,
-            gid=FLAGS.gid,
-            files_preserve=files_to_keep):
-        main(args)

=== added file 'nova/service.py'
--- nova/service.py	1970-01-01 00:00:00 +0000
+++ nova/service.py	2010-12-16 20:49:10 +0000
@@ -0,0 +1,234 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+Generic Node baseclass for all workers that run on hosts
+"""
+
+import inspect
+import logging
+import os
+import sys
+
+from eventlet import event
+from eventlet import greenthread
+from eventlet import greenpool
+
+from nova import context
+from nova import db
+from nova import exception
+from nova import flags
+from nova import rpc
+from nova import utils
+
+
+FLAGS = flags.FLAGS
+flags.DEFINE_integer('report_interval', 10,
+                     'seconds between nodes reporting state to datastore',
+                     lower_bound=1)
+
+flags.DEFINE_integer('periodic_interval', 60,
+                     'seconds between running periodic tasks',
+                     lower_bound=1)
+
+flags.DEFINE_string('pidfile', None,
+                    'pidfile to use for this service')
+
+
+flags.DEFINE_flag(flags.HelpFlag())
+flags.DEFINE_flag(flags.HelpshortFlag())
+flags.DEFINE_flag(flags.HelpXMLFlag())
+
+
+class Service(object):
+    """Base class for workers that run on hosts."""
+
+    def __init__(self, host, binary, topic, manager, report_interval=None,
+                 periodic_interval=None, *args, **kwargs):
+        self.host = host
+        self.binary = binary
+        self.topic = topic
+        self.manager_class_name = manager
+        self.report_interval = report_interval
+        self.periodic_interval = periodic_interval
+        super(Service, self).__init__(*args, **kwargs)
+        self.saved_args, self.saved_kwargs = args, kwargs
+        self.timers = []
+
+    def start(self):
+        manager_class = utils.import_class(self.manager_class_name)
+        self.manager = manager_class(host=self.host, *self.saved_args,
+                                                     **self.saved_kwargs)
+        self.manager.init_host()
+        self.model_disconnected = False
+        ctxt = context.get_admin_context()
+        try:
+            service_ref = db.service_get_by_args(ctxt,
+                                                 self.host,
+                                                 self.binary)
+            self.service_id = service_ref['id']
+        except exception.NotFound:
+            self._create_service_ref(ctxt)
+
+        conn1 = rpc.Connection.instance(new=True)
+        conn2 = rpc.Connection.instance(new=True)
+        if self.report_interval:
+            consumer_all = rpc.AdapterConsumer(
+                    connection=conn1,
+                    topic=self.topic,
+                    proxy=self)
+            consumer_node = rpc.AdapterConsumer(
+                    connection=conn2,
+                    topic='%s.%s' % (self.topic, self.host),
+                    proxy=self)
+
+            self.timers.append(consumer_all.attach_to_eventlet())
+            self.timers.append(consumer_node.attach_to_eventlet())
+
+            pulse = utils.LoopingCall(self.report_state)
+            pulse.start(interval=self.report_interval, now=False)
+            self.timers.append(pulse)
+
+        if self.periodic_interval:
+            periodic = utils.LoopingCall(self.periodic_tasks)
+            periodic.start(interval=self.periodic_interval, now=False)
+            self.timers.append(periodic)
+
+    def _create_service_ref(self, context):
+        service_ref = db.service_create(context,
+                                        {'host': self.host,
+                                         'binary': self.binary,
+                                         'topic': self.topic,
+                                         'report_count': 0})
+        self.service_id = service_ref['id']
+
+    def __getattr__(self, key):
+        manager = self.__dict__.get('manager', None)
+        return getattr(manager, key)
+
+    @classmethod
+    def create(cls,
+               host=None,
+               binary=None,
+               topic=None,
+               manager=None,
+               report_interval=None,
+               periodic_interval=None):
+        """Instantiates class and passes back application object.
+
+        Args:
+            host, defaults to FLAGS.host
+            binary, defaults to basename of executable
+            topic, defaults to bin_name - "nova-" part
+            manager, defaults to FLAGS.<topic>_manager
+            report_interval, defaults to FLAGS.report_interval
+            periodic_interval, defaults to FLAGS.periodic_interval
+        """
+        if not host:
+            host = FLAGS.host
+        if not binary:
+            binary = os.path.basename(inspect.stack()[-1][1])
+        if not topic:
+            topic = binary.rpartition("nova-")[2]
+        if not manager:
+            manager = FLAGS.get('%s_manager' % topic, None)
+        if not report_interval:
+            report_interval = FLAGS.report_interval
+        if not periodic_interval:
+            periodic_interval = FLAGS.periodic_interval
+        logging.warn("Starting %s node", topic)
+        service_obj = cls(host, binary, topic, manager,
+                          report_interval, periodic_interval)
+
+        return service_obj
+
+    def kill(self):
+        """Destroy the service object in the datastore"""
+        self.stop()
+        try:
+            db.service_destroy(context.get_admin_context(), self.service_id)
+        except exception.NotFound:
+            logging.warn("Service killed that has no database entry")
+
+    def stop(self):
+        for x in self.timers:
+            try:
+                x.stop()
+            except Exception:
+                pass
+        self.timers = []
+
+    def periodic_tasks(self):
+        """Tasks to be run at a periodic interval"""
+        self.manager.periodic_tasks(context.get_admin_context())
+
+    def report_state(self):
+        """Update the state of this service in the datastore."""
+        ctxt = context.get_admin_context()
+        try:
+            try:
+                service_ref = db.service_get(ctxt, self.service_id)
+            except exception.NotFound:
+                logging.debug("The service database object disappeared, "
+                              "Recreating it.")
+                self._create_service_ref(ctxt)
+                service_ref = db.service_get(ctxt, self.service_id)
+
+            db.service_update(ctxt,
+                             self.service_id,
+                             {'report_count': service_ref['report_count'] + 1})
+
+            # TODO(termie): make this pattern be more elegant.
+            if getattr(self, "model_disconnected", False):
+                self.model_disconnected = False
+                logging.error("Recovered model server connection!")
+
+        # TODO(vish): this should probably only catch connection errors
+        except Exception:  # pylint: disable-msg=W0702
+            if not getattr(self, "model_disconnected", False):
+                self.model_disconnected = True
+                logging.exception("model server went away")
+
+
+def serve(*services):
+    argv = FLAGS(sys.argv)
+
+    if not services:
+        services = [Service.create()]
+
+    name = '_'.join(x.binary for x in services)
+    logging.debug("Serving %s" % name)
+
+    logging.getLogger('amqplib').setLevel(logging.WARN)
+
+    if FLAGS.verbose:
+        logging.getLogger().setLevel(logging.DEBUG)
+    else:
+        logging.getLogger().setLevel(logging.WARNING)
+
+    logging.debug("Full set of FLAGS:")
+    for flag in FLAGS:
+        logging.debug("%s : %s" % (flag, FLAGS.get(flag, None)))
+
+    for x in services:
+        x.start()
+
+
+def wait():
+    while True:
+        greenthread.sleep(5)

=== removed file 'nova/service.py'
--- nova/service.py	2010-11-15 19:43:50 +0000
+++ nova/service.py	1970-01-01 00:00:00 +0000
@@ -1,195 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# All Rights Reserved.
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-"""
-A service is a very thin wrapper around a Manager object.  It exposes the
-manager's public methods to other components of the system via rpc.  It will
-report state periodically to the database and is responsible for initiating
-any periodic tasts that need to be executed on a given host.
-
-This module contains Service, a generic baseclass for all workers.
-"""
-
-import inspect
-import logging
-import os
-
-from twisted.internet import defer
-from twisted.internet import task
-from twisted.application import service
-
-from nova import context
-from nova import db
-from nova import exception
-from nova import flags
-from nova import rpc
-from nova import utils
-
-
-FLAGS = flags.FLAGS
-flags.DEFINE_integer('report_interval', 10,
-                     'seconds between nodes reporting state to datastore',
-                     lower_bound=1)
-
-flags.DEFINE_integer('periodic_interval', 60,
-                     'seconds between running periodic tasks',
-                     lower_bound=1)
-
-
-class Service(object, service.Service):
-    """Base class for workers that run on hosts."""
-
-    def __init__(self, host, binary, topic, manager, report_interval=None,
-                 periodic_interval=None, *args, **kwargs):
-        self.host = host
-        self.binary = binary
-        self.topic = topic
-        self.manager_class_name = manager
-        self.report_interval = report_interval
-        self.periodic_interval = periodic_interval
-        super(Service, self).__init__(*args, **kwargs)
-        self.saved_args, self.saved_kwargs = args, kwargs
-
-    def startService(self):  # pylint: disable-msg C0103
-        manager_class = utils.import_class(self.manager_class_name)
-        self.manager = manager_class(host=self.host, *self.saved_args,
-                                                     **self.saved_kwargs)
-        self.manager.init_host()
-        self.model_disconnected = False
-        ctxt = context.get_admin_context()
-        try:
-            service_ref = db.service_get_by_args(ctxt,
-                                                 self.host,
-                                                 self.binary)
-            self.service_id = service_ref['id']
-        except exception.NotFound:
-            self._create_service_ref(ctxt)
-
-        conn = rpc.Connection.instance()
-        if self.report_interval:
-            consumer_all = rpc.AdapterConsumer(
-                    connection=conn,
-                    topic=self.topic,
-                    proxy=self)
-            consumer_node = rpc.AdapterConsumer(
-                    connection=conn,
-                    topic='%s.%s' % (self.topic, self.host),
-                    proxy=self)
-
-            consumer_all.attach_to_twisted()
-            consumer_node.attach_to_twisted()
-
-            pulse = task.LoopingCall(self.report_state)
-            pulse.start(interval=self.report_interval, now=False)
-
-        if self.periodic_interval:
-            pulse = task.LoopingCall(self.periodic_tasks)
-            pulse.start(interval=self.periodic_interval, now=False)
-
-    def _create_service_ref(self, context):
-        service_ref = db.service_create(context,
-                                        {'host': self.host,
-                                         'binary': self.binary,
-                                         'topic': self.topic,
-                                         'report_count': 0})
-        self.service_id = service_ref['id']
-
-    def __getattr__(self, key):
-        manager = self.__dict__.get('manager', None)
-        return getattr(manager, key)
-
-    @classmethod
-    def create(cls,
-               host=None,
-               binary=None,
-               topic=None,
-               manager=None,
-               report_interval=None,
-               periodic_interval=None):
-        """Instantiates class and passes back application object.
-
-        Args:
-            host, defaults to FLAGS.host
-            binary, defaults to basename of executable
-            topic, defaults to bin_name - "nova-" part
-            manager, defaults to FLAGS.<topic>_manager
-            report_interval, defaults to FLAGS.report_interval
-            periodic_interval, defaults to FLAGS.periodic_interval
-        """
-        if not host:
-            host = FLAGS.host
-        if not binary:
-            binary = os.path.basename(inspect.stack()[-1][1])
-        if not topic:
-            topic = binary.rpartition("nova-")[2]
-        if not manager:
-            manager = FLAGS.get('%s_manager' % topic, None)
-        if not report_interval:
-            report_interval = FLAGS.report_interval
-        if not periodic_interval:
-            periodic_interval = FLAGS.periodic_interval
-        logging.warn("Starting %s node", topic)
-        service_obj = cls(host, binary, topic, manager,
-                          report_interval, periodic_interval)
-
-        # This is the parent service that twistd will be looking for when it
-        # parses this file, return it so that we can get it into globals.
-        application = service.Application(binary)
-        service_obj.setServiceParent(application)
-        return application
-
-    def kill(self):
-        """Destroy the service object in the datastore"""
-        try:
-            db.service_destroy(context.get_admin_context(), self.service_id)
-        except exception.NotFound:
-            logging.warn("Service killed that has no database entry")
-
-    @defer.inlineCallbacks
-    def periodic_tasks(self):
-        """Tasks to be run at a periodic interval"""
-        yield self.manager.periodic_tasks(context.get_admin_context())
-
-    @defer.inlineCallbacks
-    def report_state(self):
-        """Update the state of this service in the datastore."""
-        ctxt = context.get_admin_context()
-        try:
-            try:
-                service_ref = db.service_get(ctxt, self.service_id)
-            except exception.NotFound:
-                logging.debug("The service database object disappeared, "
-                              "Recreating it.")
-                self._create_service_ref(ctxt)
-                service_ref = db.service_get(ctxt, self.service_id)
-
-            db.service_update(ctxt,
-                             self.service_id,
-                             {'report_count': service_ref['report_count'] + 1})
-
-            # TODO(termie): make this pattern be more elegant.
-            if getattr(self, "model_disconnected", False):
-                self.model_disconnected = False
-                logging.error("Recovered model server connection!")
-
-        # TODO(vish): this should probably only catch connection errors
-        except Exception:  # pylint: disable-msg=W0702
-            if not getattr(self, "model_disconnected", False):
-                self.model_disconnected = True
-                logging.exception("model server went away")
-        yield

=== modified file 'nova/test.py'
--- nova/test.py	2010-10-26 15:48:20 +0000
+++ nova/test.py	2010-12-16 20:49:10 +0000
@@ -25,11 +25,12 @@
 import datetime
 import sys
 import time
+import unittest
 
 import mox
 import stubout
 from twisted.internet import defer
-from twisted.trial import unittest
+from twisted.trial import unittest as trial_unittest
 
 from nova import context
 from nova import db
@@ -55,7 +56,89 @@
     return _skipper
 
 
-class TrialTestCase(unittest.TestCase):
+class TestCase(unittest.TestCase):
+    """Test case base class for all unit tests"""
+    def setUp(self):
+        """Run before each test method to initialize test environment"""
+        super(TestCase, self).setUp()
+        # NOTE(vish): We need a better method for creating fixtures for tests
+        #             now that we have some required db setup for the system
+        #             to work properly.
+        self.start = datetime.datetime.utcnow()
+        ctxt = context.get_admin_context()
+        if db.network_count(ctxt) != 5:
+            network_manager.VlanManager().create_networks(ctxt,
+                                                          FLAGS.fixed_range,
+                                                          5, 16,
+                                                          FLAGS.vlan_start,
+                                                          FLAGS.vpn_start)
+
+        # emulate some of the mox stuff, we can't use the metaclass
+        # because it screws with our generators
+        self.mox = mox.Mox()
+        self.stubs = stubout.StubOutForTesting()
+        self.flag_overrides = {}
+        self.injected = []
+        self._monkey_patch_attach()
+        self._original_flags = FLAGS.FlagValuesDict()
+
+    def tearDown(self):
+        """Runs after each test method to finalize/tear down test
+        environment."""
+        try:
+            self.mox.UnsetStubs()
+            self.stubs.UnsetAll()
+            self.stubs.SmartUnsetAll()
+            self.mox.VerifyAll()
+            # NOTE(vish): Clean up any ips associated during the test.
+            ctxt = context.get_admin_context()
+            db.fixed_ip_disassociate_all_by_timeout(ctxt, FLAGS.host,
+                                                    self.start)
+            db.network_disassociate_all(ctxt)
+            rpc.Consumer.attach_to_eventlet = self.originalAttach
+            for x in self.injected:
+                try:
+                    x.stop()
+                except AssertionError:
+                    pass
+
+            if FLAGS.fake_rabbit:
+                fakerabbit.reset_all()
+
+            db.security_group_destroy_all(ctxt)
+            super(TestCase, self).tearDown()
+        finally:
+            self.reset_flags()
+
+    def flags(self, **kw):
+        """Override flag variables for a test"""
+        for k, v in kw.iteritems():
+            if k in self.flag_overrides:
+                self.reset_flags()
+                raise Exception(
+                        'trying to override already overriden flag: %s' % k)
+            self.flag_overrides[k] = getattr(FLAGS, k)
+            setattr(FLAGS, k, v)
+
+    def reset_flags(self):
+        """Resets all flag variables for the test.  Runs after each test"""
+        FLAGS.Reset()
+        for k, v in self._original_flags.iteritems():
+            setattr(FLAGS, k, v)
+
+    def _monkey_patch_attach(self):
+        self.originalAttach = rpc.Consumer.attach_to_eventlet
+
+        def _wrapped(innerSelf):
+            rv = self.originalAttach(innerSelf)
+            self.injected.append(rv)
+            return rv
+
+        _wrapped.func_name = self.originalAttach.func_name
+        rpc.Consumer.attach_to_eventlet = _wrapped
+
+
+class TrialTestCase(trial_unittest.TestCase):
     """Test case base class for all unit tests"""
     def setUp(self):
         """Run before each test method to initialize test environment"""
@@ -78,7 +161,6 @@
         self.stubs = stubout.StubOutForTesting()
         self.flag_overrides = {}
         self.injected = []
-        self._monkey_patch_attach()
         self._original_flags = FLAGS.FlagValuesDict()
 
     def tearDown(self):
@@ -94,7 +176,6 @@
             db.fixed_ip_disassociate_all_by_timeout(ctxt, FLAGS.host,
                                                     self.start)
             db.network_disassociate_all(ctxt)
-            rpc.Consumer.attach_to_twisted = self.originalAttach
             for x in self.injected:
                 try:
                     x.stop()
@@ -147,14 +228,3 @@
             return d
         _wrapped.func_name = func.func_name
         return _wrapped
-
-    def _monkey_patch_attach(self):
-        self.originalAttach = rpc.Consumer.attach_to_twisted
-
-        def _wrapped(innerSelf):
-            rv = self.originalAttach(innerSelf)
-            self.injected.append(rv)
-            return rv
-
-        _wrapped.func_name = self.originalAttach.func_name
-        rpc.Consumer.attach_to_twisted = _wrapped

=== modified file 'nova/tests/access_unittest.py'
--- nova/tests/access_unittest.py	2010-10-22 07:48:27 +0000
+++ nova/tests/access_unittest.py	2010-12-16 20:49:10 +0000
@@ -35,7 +35,7 @@
     pass
 
 
-class AccessTestCase(test.TrialTestCase):
+class AccessTestCase(test.TestCase):
     def setUp(self):
         super(AccessTestCase, self).setUp()
         um = manager.AuthManager()

=== modified file 'nova/tests/auth_unittest.py'
--- nova/tests/auth_unittest.py	2010-10-22 07:48:27 +0000
+++ nova/tests/auth_unittest.py	2010-12-16 20:49:10 +0000
@@ -326,12 +326,12 @@
             self.assertTrue(user.is_admin())
 
 
-class AuthManagerLdapTestCase(AuthManagerTestCase, test.TrialTestCase):
+class AuthManagerLdapTestCase(AuthManagerTestCase, test.TestCase):
     auth_driver = 'nova.auth.ldapdriver.FakeLdapDriver'
 
     def __init__(self, *args, **kwargs):
         AuthManagerTestCase.__init__(self)
-        test.TrialTestCase.__init__(self, *args, **kwargs)
+        test.TestCase.__init__(self, *args, **kwargs)
         import nova.auth.fakeldap as fakeldap
         FLAGS.redis_db = 8
         if FLAGS.flush_db:
@@ -343,7 +343,7 @@
                 self.skip = True
 
 
-class AuthManagerDbTestCase(AuthManagerTestCase, test.TrialTestCase):
+class AuthManagerDbTestCase(AuthManagerTestCase, test.TestCase):
     auth_driver = 'nova.auth.dbdriver.DbDriver'
 
 

=== modified file 'nova/tests/cloud_unittest.py'
--- nova/tests/cloud_unittest.py	2010-12-09 17:30:03 +0000
+++ nova/tests/cloud_unittest.py	2010-12-16 20:49:10 +0000
@@ -27,8 +27,6 @@
 import time
 
 from eventlet import greenthread
-from twisted.internet import defer
-import unittest
 from xml.etree import ElementTree
 
 from nova import context
@@ -53,7 +51,7 @@
 os.makedirs(IMAGES_PATH)
 
 
-class CloudTestCase(test.TrialTestCase):
+class CloudTestCase(test.TestCase):
     def setUp(self):
         super(CloudTestCase, self).setUp()
         self.flags(connection_type='fake', images_path=IMAGES_PATH)
@@ -199,7 +197,7 @@
         logging.debug("Need to watch instance %s until it's running..." %
                       instance['instance_id'])
         while True:
-            rv = yield defer.succeed(time.sleep(1))
+            greenthread.sleep(1)
             info = self.cloud._get_instance(instance['instance_id'])
             logging.debug(info['state'])
             if info['state'] == power_state.RUNNING:

=== modified file 'nova/tests/compute_unittest.py'
--- nova/tests/compute_unittest.py	2010-12-03 20:21:18 +0000
+++ nova/tests/compute_unittest.py	2010-12-16 20:49:10 +0000
@@ -22,8 +22,6 @@
 import datetime
 import logging
 
-from twisted.internet import defer
-
 from nova import context
 from nova import db
 from nova import exception
@@ -33,10 +31,11 @@
 from nova.auth import manager
 from nova.compute import api as compute_api
 
+
 FLAGS = flags.FLAGS
 
 
-class ComputeTestCase(test.TrialTestCase):
+class ComputeTestCase(test.TestCase):
     """Test case for compute"""
     def setUp(self):
         logging.getLogger().setLevel(logging.DEBUG)
@@ -94,24 +93,22 @@
             db.security_group_destroy(self.context, group['id'])
             db.instance_destroy(self.context, ref[0]['id'])
 
-    @defer.inlineCallbacks
     def test_run_terminate(self):
         """Make sure it is possible to  run and terminate instance"""
         instance_id = self._create_instance()
 
-        yield self.compute.run_instance(self.context, instance_id)
+        self.compute.run_instance(self.context, instance_id)
 
         instances = db.instance_get_all(context.get_admin_context())
         logging.info("Running instances: %s", instances)
         self.assertEqual(len(instances), 1)
 
-        yield self.compute.terminate_instance(self.context, instance_id)
+        self.compute.terminate_instance(self.context, instance_id)
 
         instances = db.instance_get_all(context.get_admin_context())
         logging.info("After terminating instances: %s", instances)
         self.assertEqual(len(instances), 0)
 
-    @defer.inlineCallbacks
     def test_run_terminate_timestamps(self):
         """Make sure timestamps are set for launched and destroyed"""
         instance_id = self._create_instance()
@@ -119,42 +116,40 @@
         self.assertEqual(instance_ref['launched_at'], None)
         self.assertEqual(instance_ref['deleted_at'], None)
         launch = datetime.datetime.utcnow()
-        yield self.compute.run_instance(self.context, instance_id)
+        self.compute.run_instance(self.context, instance_id)
         instance_ref = db.instance_get(self.context, instance_id)
         self.assert_(instance_ref['launched_at'] > launch)
         self.assertEqual(instance_ref['deleted_at'], None)
         terminate = datetime.datetime.utcnow()
-        yield self.compute.terminate_instance(self.context, instance_id)
+        self.compute.terminate_instance(self.context, instance_id)
         self.context = self.context.elevated(True)
         instance_ref = db.instance_get(self.context, instance_id)
         self.assert_(instance_ref['launched_at'] < terminate)
         self.assert_(instance_ref['deleted_at'] > terminate)
 
-    @defer.inlineCallbacks
     def test_reboot(self):
         """Ensure instance can be rebooted"""
         instance_id = self._create_instance()
-        yield self.compute.run_instance(self.context, instance_id)
-        yield self.compute.reboot_instance(self.context, instance_id)
-        yield self.compute.terminate_instance(self.context, instance_id)
+        self.compute.run_instance(self.context, instance_id)
+        self.compute.reboot_instance(self.context, instance_id)
+        self.compute.terminate_instance(self.context, instance_id)
 
-    @defer.inlineCallbacks
     def test_console_output(self):
         """Make sure we can get console output from instance"""
         instance_id = self._create_instance()
-        yield self.compute.run_instance(self.context, instance_id)
+        self.compute.run_instance(self.context, instance_id)
 
-        console = yield self.compute.get_console_output(self.context,
+        console = self.compute.get_console_output(self.context,
                                                         instance_id)
         self.assert_(console)
-        yield self.compute.terminate_instance(self.context, instance_id)
+        self.compute.terminate_instance(self.context, instance_id)
 
-    @defer.inlineCallbacks
     def test_run_instance_existing(self):
         """Ensure failure when running an instance that already exists"""
         instance_id = self._create_instance()
-        yield self.compute.run_instance(self.context, instance_id)
-        self.assertFailure(self.compute.run_instance(self.context,
-                                                     instance_id),
-                           exception.Error)
-        yield self.compute.terminate_instance(self.context, instance_id)
+        self.compute.run_instance(self.context, instance_id)
+        self.assertRaises(exception.Error,
+                          self.compute.run_instance,
+                          self.context,
+                          instance_id)
+        self.compute.terminate_instance(self.context, instance_id)

=== modified file 'nova/tests/flags_unittest.py'
--- nova/tests/flags_unittest.py	2010-10-22 07:48:27 +0000
+++ nova/tests/flags_unittest.py	2010-12-16 20:49:10 +0000
@@ -24,7 +24,7 @@
 flags.DEFINE_string('flags_unittest', 'foo', 'for testing purposes only')
 
 
-class FlagsTestCase(test.TrialTestCase):
+class FlagsTestCase(test.TestCase):
 
     def setUp(self):
         super(FlagsTestCase, self).setUp()

=== modified file 'nova/tests/misc_unittest.py'
--- nova/tests/misc_unittest.py	2010-12-15 10:57:56 +0000
+++ nova/tests/misc_unittest.py	2010-12-16 20:49:10 +0000
@@ -20,7 +20,7 @@
 from nova.utils import parse_mailmap, str_dict_replace
 
 
-class ProjectTestCase(test.TrialTestCase):
+class ProjectTestCase(test.TestCase):
     def test_authors_up_to_date(self):
         if os.path.exists('../.bzr'):
             contributors = set()

=== modified file 'nova/tests/network_unittest.py'
--- nova/tests/network_unittest.py	2010-11-17 02:23:20 +0000
+++ nova/tests/network_unittest.py	2010-12-16 20:49:10 +0000
@@ -33,7 +33,7 @@
 FLAGS = flags.FLAGS
 
 
-class NetworkTestCase(test.TrialTestCase):
+class NetworkTestCase(test.TestCase):
     """Test cases for network code"""
     def setUp(self):
         super(NetworkTestCase, self).setUp()

=== modified file 'nova/tests/objectstore_unittest.py'
--- nova/tests/objectstore_unittest.py	2010-10-22 07:48:27 +0000
+++ nova/tests/objectstore_unittest.py	2010-12-16 20:49:10 +0000
@@ -54,7 +54,7 @@
 os.makedirs(os.path.join(OSS_TEMPDIR, 'buckets'))
 
 
-class ObjectStoreTestCase(test.TrialTestCase):
+class ObjectStoreTestCase(test.TestCase):
     """Test objectstore API directly."""
 
     def setUp(self):
@@ -191,7 +191,7 @@
     protocol = TestHTTPChannel
 
 
-class S3APITestCase(test.TrialTestCase):
+class S3APITestCase(test.TestCase):
     """Test objectstore through S3 API."""
 
     def setUp(self):

=== removed file 'nova/tests/process_unittest.py'
--- nova/tests/process_unittest.py	2010-10-22 07:48:27 +0000
+++ nova/tests/process_unittest.py	1970-01-01 00:00:00 +0000
@@ -1,132 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# All Rights Reserved.
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-import logging
-from twisted.internet import defer
-from twisted.internet import reactor
-from xml.etree import ElementTree
-
-from nova import exception
-from nova import flags
-from nova import process
-from nova import test
-from nova import utils
-
-FLAGS = flags.FLAGS
-
-
-class ProcessTestCase(test.TrialTestCase):
-    def setUp(self):
-        logging.getLogger().setLevel(logging.DEBUG)
-        super(ProcessTestCase, self).setUp()
-
-    def test_execute_stdout(self):
-        pool = process.ProcessPool(2)
-        d = pool.simple_execute('echo test')
-
-        def _check(rv):
-            self.assertEqual(rv[0], 'test\n')
-            self.assertEqual(rv[1], '')
-
-        d.addCallback(_check)
-        d.addErrback(self.fail)
-        return d
-
-    def test_execute_stderr(self):
-        pool = process.ProcessPool(2)
-        d = pool.simple_execute('cat BAD_FILE', check_exit_code=False)
-
-        def _check(rv):
-            self.assertEqual(rv[0], '')
-            self.assert_('No such file' in rv[1])
-
-        d.addCallback(_check)
-        d.addErrback(self.fail)
-        return d
-
-    def test_execute_unexpected_stderr(self):
-        pool = process.ProcessPool(2)
-        d = pool.simple_execute('cat BAD_FILE')
-        d.addCallback(lambda x: self.fail('should have raised an error'))
-        d.addErrback(lambda failure: failure.trap(IOError))
-        return d
-
-    def test_max_processes(self):
-        pool = process.ProcessPool(2)
-        d1 = pool.simple_execute('sleep 0.01')
-        d2 = pool.simple_execute('sleep 0.01')
-        d3 = pool.simple_execute('sleep 0.005')
-        d4 = pool.simple_execute('sleep 0.005')
-
-        called = []
-
-        def _called(rv, name):
-            called.append(name)
-
-        d1.addCallback(_called, 'd1')
-        d2.addCallback(_called, 'd2')
-        d3.addCallback(_called, 'd3')
-        d4.addCallback(_called, 'd4')
-
-        # Make sure that d3 and d4 had to wait on the other two and were called
-        # in order
-        # NOTE(termie): there may be a race condition in this test if for some
-        #               reason one of the sleeps takes longer to complete
-        #               than it should
-        d4.addCallback(lambda x: self.assertEqual(called[2], 'd3'))
-        d4.addCallback(lambda x: self.assertEqual(called[3], 'd4'))
-        d4.addErrback(self.fail)
-        return d4
-
-    def test_kill_long_process(self):
-        pool = process.ProcessPool(2)
-
-        d1 = pool.simple_execute('sleep 1')
-        d2 = pool.simple_execute('sleep 0.005')
-
-        timeout = reactor.callLater(0.1, self.fail, 'should have been killed')
-
-        # kill d1 and wait on it to end then cancel the timeout
-        d2.addCallback(lambda _: d1.process.signalProcess('KILL'))
-        d2.addCallback(lambda _: d1)
-        d2.addBoth(lambda _: timeout.active() and timeout.cancel())
-        d2.addErrback(self.fail)
-        return d2
-
-    def test_process_exit_is_contained(self):
-        pool = process.ProcessPool(2)
-
-        d1 = pool.simple_execute('sleep 1')
-        d1.addCallback(lambda x: self.fail('should have errbacked'))
-        d1.addErrback(lambda fail: fail.trap(IOError))
-        reactor.callLater(0.05, d1.process.signalProcess, 'KILL')
-
-        return d1
-
-    def test_shared_pool_is_singleton(self):
-        pool1 = process.SharedPool()
-        pool2 = process.SharedPool()
-        self.assertEqual(id(pool1._instance), id(pool2._instance))
-
-    def test_shared_pool_works_as_singleton(self):
-        d1 = process.simple_execute('sleep 1')
-        d2 = process.simple_execute('sleep 0.005')
-        # lp609749: would have failed with
-        # exceptions.AssertionError: Someone released me too many times:
-        # too many tokens!
-        return d1

=== modified file 'nova/tests/quota_unittest.py'
--- nova/tests/quota_unittest.py	2010-11-24 22:52:10 +0000
+++ nova/tests/quota_unittest.py	2010-12-16 20:49:10 +0000
@@ -32,7 +32,7 @@
 FLAGS = flags.FLAGS
 
 
-class QuotaTestCase(test.TrialTestCase):
+class QuotaTestCase(test.TestCase):
     def setUp(self):
         logging.getLogger().setLevel(logging.DEBUG)
         super(QuotaTestCase, self).setUp()

=== modified file 'nova/tests/rpc_unittest.py'
--- nova/tests/rpc_unittest.py	2010-10-22 07:48:27 +0000
+++ nova/tests/rpc_unittest.py	2010-12-16 20:49:10 +0000
@@ -20,8 +20,6 @@
 """
 import logging
 
-from twisted.internet import defer
-
 from nova import context
 from nova import flags
 from nova import rpc
@@ -31,7 +29,7 @@
 FLAGS = flags.FLAGS
 
 
-class RpcTestCase(test.TrialTestCase):
+class RpcTestCase(test.TestCase):
     """Test cases for rpc"""
     def setUp(self):
         super(RpcTestCase, self).setUp()
@@ -40,23 +38,22 @@
         self.consumer = rpc.AdapterConsumer(connection=self.conn,
                                             topic='test',
                                             proxy=self.receiver)
-        self.consumer.attach_to_twisted()
+        self.consumer.attach_to_eventlet()
         self.context = context.get_admin_context()
 
     def test_call_succeed(self):
         """Get a value through rpc call"""
         value = 42
-        result = yield rpc.call_twisted(self.context,
-                                        'test', {"method": "echo",
+        result = rpc.call(self.context, 'test', {"method": "echo",
                                                  "args": {"value": value}})
         self.assertEqual(value, result)
 
     def test_context_passed(self):
         """Makes sure a context is passed through rpc call"""
         value = 42
-        result = yield rpc.call_twisted(self.context,
-                                        'test', {"method": "context",
-                                                 "args": {"value": value}})
+        result = rpc.call(self.context,
+                          'test', {"method": "context",
+                                   "args": {"value": value}})
         self.assertEqual(self.context.to_dict(), result)
 
     def test_call_exception(self):
@@ -67,14 +64,17 @@
         to an int in the test.
         """
         value = 42
-        self.assertFailure(rpc.call_twisted(self.context, 'test',
-                                            {"method": "fail",
-                                             "args": {"value": value}}),
-                           rpc.RemoteError)
+        self.assertRaises(rpc.RemoteError,
+                          rpc.call,
+                          self.context,
+                          'test',
+                          {"method": "fail",
+                           "args": {"value": value}})
         try:
-            yield rpc.call_twisted(self.context,
-                                   'test', {"method": "fail",
-                                            "args": {"value": value}})
+            rpc.call(self.context,
+                     'test',
+                     {"method": "fail",
+                      "args": {"value": value}})
             self.fail("should have thrown rpc.RemoteError")
         except rpc.RemoteError as exc:
             self.assertEqual(int(exc.value), value)
@@ -89,13 +89,13 @@
     def echo(context, value):
         """Simply returns whatever value is sent in"""
         logging.debug("Received %s", value)
-        return defer.succeed(value)
+        return value
 
     @staticmethod
     def context(context, value):
         """Returns dictionary version of context"""
         logging.debug("Received %s", context)
-        return defer.succeed(context.to_dict())
+        return context.to_dict()
 
     @staticmethod
     def fail(context, value):

=== modified file 'nova/tests/scheduler_unittest.py'
--- nova/tests/scheduler_unittest.py	2010-10-26 06:37:51 +0000
+++ nova/tests/scheduler_unittest.py	2010-12-16 20:49:10 +0000
@@ -44,7 +44,7 @@
         return 'named_host'
 
 
-class SchedulerTestCase(test.TrialTestCase):
+class SchedulerTestCase(test.TestCase):
     """Test case for scheduler"""
     def setUp(self):
         super(SchedulerTestCase, self).setUp()
@@ -73,7 +73,7 @@
         scheduler.named_method(ctxt, 'topic', num=7)
 
 
-class SimpleDriverTestCase(test.TrialTestCase):
+class SimpleDriverTestCase(test.TestCase):
     """Test case for simple driver"""
     def setUp(self):
         super(SimpleDriverTestCase, self).setUp()
@@ -122,12 +122,12 @@
                                    'nova-compute',
                                    'compute',
                                    FLAGS.compute_manager)
-        compute1.startService()
+        compute1.start()
         compute2 = service.Service('host2',
                                    'nova-compute',
                                    'compute',
                                    FLAGS.compute_manager)
-        compute2.startService()
+        compute2.start()
         hosts = self.scheduler.driver.hosts_up(self.context, 'compute')
         self.assertEqual(len(hosts), 2)
         compute1.kill()
@@ -139,12 +139,12 @@
                                    'nova-compute',
                                    'compute',
                                    FLAGS.compute_manager)
-        compute1.startService()
+        compute1.start()
         compute2 = service.Service('host2',
                                    'nova-compute',
                                    'compute',
                                    FLAGS.compute_manager)
-        compute2.startService()
+        compute2.start()
         instance_id1 = self._create_instance()
         compute1.run_instance(self.context, instance_id1)
         instance_id2 = self._create_instance()
@@ -162,12 +162,12 @@
                                    'nova-compute',
                                    'compute',
                                    FLAGS.compute_manager)
-        compute1.startService()
+        compute1.start()
         compute2 = service.Service('host2',
                                    'nova-compute',
                                    'compute',
                                    FLAGS.compute_manager)
-        compute2.startService()
+        compute2.start()
         instance_ids1 = []
         instance_ids2 = []
         for index in xrange(FLAGS.max_cores):
@@ -195,12 +195,12 @@
                                    'nova-volume',
                                    'volume',
                                    FLAGS.volume_manager)
-        volume1.startService()
+        volume1.start()
         volume2 = service.Service('host2',
                                    'nova-volume',
                                    'volume',
                                    FLAGS.volume_manager)
-        volume2.startService()
+        volume2.start()
         volume_id1 = self._create_volume()
         volume1.create_volume(self.context, volume_id1)
         volume_id2 = self._create_volume()
@@ -218,12 +218,12 @@
                                    'nova-volume',
                                    'volume',
                                    FLAGS.volume_manager)
-        volume1.startService()
+        volume1.start()
         volume2 = service.Service('host2',
                                    'nova-volume',
                                    'volume',
                                    FLAGS.volume_manager)
-        volume2.startService()
+        volume2.start()
         volume_ids1 = []
         volume_ids2 = []
         for index in xrange(FLAGS.max_gigabytes):

=== modified file 'nova/tests/service_unittest.py'
--- nova/tests/service_unittest.py	2010-10-26 06:08:57 +0000
+++ nova/tests/service_unittest.py	2010-12-16 20:49:10 +0000
@@ -22,9 +22,6 @@
 
 import mox
 
-from twisted.application.app import startApplication
-from twisted.internet import defer
-
 from nova import exception
 from nova import flags
 from nova import rpc
@@ -48,7 +45,7 @@
         return 'service'
 
 
-class ServiceManagerTestCase(test.TrialTestCase):
+class ServiceManagerTestCase(test.TestCase):
     """Test cases for Services"""
 
     def test_attribute_error_for_no_manager(self):
@@ -63,7 +60,7 @@
                                'test',
                                'test',
                                'nova.tests.service_unittest.FakeManager')
-        serv.startService()
+        serv.start()
         self.assertEqual(serv.test_method(), 'manager')
 
     def test_override_manager_method(self):
@@ -71,11 +68,11 @@
                                'test',
                                'test',
                                'nova.tests.service_unittest.FakeManager')
-        serv.startService()
+        serv.start()
         self.assertEqual(serv.test_method(), 'service')
 
 
-class ServiceTestCase(test.TrialTestCase):
+class ServiceTestCase(test.TestCase):
     """Test cases for Services"""
 
     def setUp(self):
@@ -94,8 +91,6 @@
         self.mox.StubOutWithMock(rpc,
                                  'AdapterConsumer',
                                  use_mock_anything=True)
-        self.mox.StubOutWithMock(
-                service.task, 'LoopingCall', use_mock_anything=True)
         rpc.AdapterConsumer(connection=mox.IgnoreArg(),
                             topic=topic,
                             proxy=mox.IsA(service.Service)).AndReturn(
@@ -106,19 +101,8 @@
                             proxy=mox.IsA(service.Service)).AndReturn(
                                     rpc.AdapterConsumer)
 
-        rpc.AdapterConsumer.attach_to_twisted()
-        rpc.AdapterConsumer.attach_to_twisted()
-
-        # Stub out looping call a bit needlessly since we don't have an easy
-        # way to cancel it (yet) when the tests finishes
-        service.task.LoopingCall(mox.IgnoreArg()).AndReturn(
-                        service.task.LoopingCall)
-        service.task.LoopingCall.start(interval=mox.IgnoreArg(),
-                                       now=mox.IgnoreArg())
-        service.task.LoopingCall(mox.IgnoreArg()).AndReturn(
-                        service.task.LoopingCall)
-        service.task.LoopingCall.start(interval=mox.IgnoreArg(),
-                                       now=mox.IgnoreArg())
+        rpc.AdapterConsumer.attach_to_eventlet()
+        rpc.AdapterConsumer.attach_to_eventlet()
 
         service_create = {'host': host,
                           'binary': binary,
@@ -136,14 +120,14 @@
                                   service_create).AndReturn(service_ref)
         self.mox.ReplayAll()
 
-        startApplication(app, False)
+        app.start()
+        app.stop()
         self.assert_(app)
 
     # We're testing sort of weird behavior in how report_state decides
     # whether it is disconnected, it looks for a variable on itself called
     # 'model_disconnected' and report_state doesn't really do much so this
     # these are mostly just for coverage
-    @defer.inlineCallbacks
     def test_report_state_no_service(self):
         host = 'foo'
         binary = 'bar'
@@ -173,10 +157,9 @@
                                binary,
                                topic,
                                'nova.tests.service_unittest.FakeManager')
-        serv.startService()
-        yield serv.report_state()
+        serv.start()
+        serv.report_state()
 
-    @defer.inlineCallbacks
     def test_report_state_newly_disconnected(self):
         host = 'foo'
         binary = 'bar'
@@ -204,11 +187,10 @@
                                binary,
                                topic,
                                'nova.tests.service_unittest.FakeManager')
-        serv.startService()
-        yield serv.report_state()
+        serv.start()
+        serv.report_state()
         self.assert_(serv.model_disconnected)
 
-    @defer.inlineCallbacks
     def test_report_state_newly_connected(self):
         host = 'foo'
         binary = 'bar'
@@ -238,8 +220,8 @@
                                binary,
                                topic,
                                'nova.tests.service_unittest.FakeManager')
-        serv.startService()
+        serv.start()
         serv.model_disconnected = True
-        yield serv.report_state()
+        serv.report_state()
 
         self.assert_(not serv.model_disconnected)

=== removed file 'nova/tests/validator_unittest.py'
--- nova/tests/validator_unittest.py	2010-10-22 07:48:27 +0000
+++ nova/tests/validator_unittest.py	1970-01-01 00:00:00 +0000
@@ -1,42 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# All Rights Reserved.
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-import logging
-import unittest
-
-from nova import flags
-from nova import test
-from nova import validate
-
-
-class ValidationTestCase(test.TrialTestCase):
-    def setUp(self):
-        super(ValidationTestCase, self).setUp()
-
-    def tearDown(self):
-        super(ValidationTestCase, self).tearDown()
-
-    def test_type_validation(self):
-        self.assertTrue(type_case("foo", 5, 1))
-        self.assertRaises(TypeError, type_case, "bar", "5", 1)
-        self.assertRaises(TypeError, type_case, None, 5, 1)
-
-
[e-mail](instanceid=str, size=int, number_of_instances=int)
-def type_case(instanceid, size, number_of_instances):
-    return True

=== modified file 'nova/tests/virt_unittest.py'
--- nova/tests/virt_unittest.py	2010-10-24 22:06:42 +0000
+++ nova/tests/virt_unittest.py	2010-12-16 20:49:10 +0000
@@ -30,7 +30,7 @@
 flags.DECLARE('instances_path', 'nova.compute.manager')
 
 
-class LibvirtConnTestCase(test.TrialTestCase):
+class LibvirtConnTestCase(test.TestCase):
     def setUp(self):
         super(LibvirtConnTestCase, self).setUp()
         self.manager = manager.AuthManager()
@@ -123,7 +123,7 @@
         self.manager.delete_user(self.user)
 
 
-class NWFilterTestCase(test.TrialTestCase):
+class NWFilterTestCase(test.TestCase):
 
     def setUp(self):
         super(NWFilterTestCase, self).setUp()
@@ -235,7 +235,7 @@
                                           'project_id': 'fake'})
         inst_id = instance_ref['id']
 
-        def _ensure_all_called(_):
+        def _ensure_all_called():
             instance_filter = 'nova-instance-%s' % instance_ref['name']
             secgroup_filter = 'nova-secgroup-%s' % self.security_group['id']
             for required in [secgroup_filter, 'allow-dhcp-server',
@@ -253,7 +253,6 @@
         instance = db.instance_get(self.context, inst_id)
 
         d = self.fw.setup_nwfilters_for_instance(instance)
-        d.addCallback(_ensure_all_called)
-        d.addCallback(lambda _: self.teardown_security_group())
-
+        _ensure_all_called()
+        self.teardown_security_group()
         return d

=== modified file 'nova/tests/volume_unittest.py'
--- nova/tests/volume_unittest.py	2010-11-03 21:38:14 +0000
+++ nova/tests/volume_unittest.py	2010-12-16 20:49:10 +0000
@@ -21,8 +21,6 @@
 """
 import logging
 
-from twisted.internet import defer
-
 from nova import context
 from nova import exception
 from nova import db
@@ -33,7 +31,7 @@
 FLAGS = flags.FLAGS
 
 
-class VolumeTestCase(test.TrialTestCase):
+class VolumeTestCase(test.TestCase):
     """Test Case for volumes."""
 
     def setUp(self):
@@ -56,51 +54,48 @@
         vol['attach_status'] = "detached"
         return db.volume_create(context.get_admin_context(), vol)['id']
 
-    @defer.inlineCallbacks
     def test_create_delete_volume(self):
         """Test volume can be created and deleted."""
         volume_id = self._create_volume()
-        yield self.volume.create_volume(self.context, volume_id)
+        self.volume.create_volume(self.context, volume_id)
         self.assertEqual(volume_id, db.volume_get(context.get_admin_context(),
                          volume_id).id)
 
-        yield self.volume.delete_volume(self.context, volume_id)
+        self.volume.delete_volume(self.context, volume_id)
         self.assertRaises(exception.NotFound,
                           db.volume_get,
                           self.context,
                           volume_id)
 
-    @defer.inlineCallbacks
     def test_too_big_volume(self):
         """Ensure failure if a too large of a volume is requested."""
         # FIXME(vish): validation needs to move into the data layer in
         #              volume_create
-        defer.returnValue(True)
+        return True
         try:
             volume_id = self._create_volume('1001')
-            yield self.volume.create_volume(self.context, volume_id)
+            self.volume.create_volume(self.context, volume_id)
             self.fail("Should have thrown TypeError")
         except TypeError:
             pass
 
-    @defer.inlineCallbacks
     def test_too_many_volumes(self):
         """Ensure that NoMoreTargets is raised when we run out of volumes."""
         vols = []
         total_slots = FLAGS.iscsi_num_targets
         for _index in xrange(total_slots):
             volume_id = self._create_volume()
-            yield self.volume.create_volume(self.context, volume_id)
+            self.volume.create_volume(self.context, volume_id)
             vols.append(volume_id)
         volume_id = self._create_volume()
-        self.assertFailure(self.volume.create_volume(self.context,
-                                                     volume_id),
-                           db.NoMoreTargets)
+        self.assertRaises(db.NoMoreTargets,
+                          self.volume.create_volume,
+                          self.context,
+                          volume_id)
         db.volume_destroy(context.get_admin_context(), volume_id)
         for volume_id in vols:
-            yield self.volume.delete_volume(self.context, volume_id)
+            self.volume.delete_volume(self.context, volume_id)
 
-    @defer.inlineCallbacks
     def test_run_attach_detach_volume(self):
         """Make sure volume can be attached and detached from instance."""
         inst = {}
@@ -115,15 +110,15 @@
         instance_id = db.instance_create(self.context, inst)['id']
         mountpoint = "/dev/sdf"
         volume_id = self._create_volume()
-        yield self.volume.create_volume(self.context, volume_id)
+        self.volume.create_volume(self.context, volume_id)
         if FLAGS.fake_tests:
             db.volume_attached(self.context, volume_id, instance_id,
                                mountpoint)
         else:
-            yield self.compute.attach_volume(self.context,
-                                             instance_id,
-                                             volume_id,
-                                             mountpoint)
+            self.compute.attach_volume(self.context,
+                                       instance_id,
+                                       volume_id,
+                                       mountpoint)
         vol = db.volume_get(context.get_admin_context(), volume_id)
         self.assertEqual(vol['status'], "in-use")
         self.assertEqual(vol['attach_status'], "attached")
@@ -131,25 +126,26 @@
         instance_ref = db.volume_get_instance(self.context, volume_id)
         self.assertEqual(instance_ref['id'], instance_id)
 
-        self.assertFailure(self.volume.delete_volume(self.context, volume_id),
-                           exception.Error)
+        self.assertRaises(exception.Error,
+                          self.volume.delete_volume,
+                          self.context,
+                          volume_id)
         if FLAGS.fake_tests:
             db.volume_detached(self.context, volume_id)
         else:
-            yield self.compute.detach_volume(self.context,
-                                             instance_id,
-                                             volume_id)
+            self.compute.detach_volume(self.context,
+                                       instance_id,
+                                       volume_id)
         vol = db.volume_get(self.context, volume_id)
         self.assertEqual(vol['status'], "available")
 
-        yield self.volume.delete_volume(self.context, volume_id)
+        self.volume.delete_volume(self.context, volume_id)
         self.assertRaises(exception.Error,
                           db.volume_get,
                           self.context,
                           volume_id)
         db.instance_destroy(self.context, instance_id)
 
-    @defer.inlineCallbacks
     def test_concurrent_volumes_get_different_targets(self):
         """Ensure multiple concurrent volumes get different targets."""
         volume_ids = []
@@ -164,15 +160,11 @@
             self.assert_(iscsi_target not in targets)
             targets.append(iscsi_target)
             logging.debug("Target %s allocated", iscsi_target)
-        deferreds = []
         total_slots = FLAGS.iscsi_num_targets
         for _index in xrange(total_slots):
             volume_id = self._create_volume()
             d = self.volume.create_volume(self.context, volume_id)
-            d.addCallback(_check)
-            d.addErrback(self.fail)
-            deferreds.append(d)
-        yield defer.DeferredList(deferreds)
+            _check(d)
         for volume_id in volume_ids:
             self.volume.delete_volume(self.context, volume_id)
 

=== modified file 'nova/utils.py'
--- nova/utils.py	2010-11-23 20:58:46 +0000
+++ nova/utils.py	2010-12-16 20:49:10 +0000
@@ -31,7 +31,8 @@
 import sys
 from xml.sax import saxutils
 
-from twisted.internet.threads import deferToThread
+from eventlet import event
+from eventlet import greenthread
 
 from nova import exception
 from nova import flags
@@ -75,7 +76,7 @@
 
 
 def execute(cmd, process_input=None, addl_env=None, check_exit_code=True):
-    logging.debug("Running cmd: %s", cmd)
+    logging.debug("Running cmd (subprocess): %s", cmd)
     env = os.environ.copy()
     if addl_env:
         env.update(addl_env)
@@ -95,6 +96,10 @@
                                         stdout=stdout,
                                         stderr=stderr,
                                         cmd=cmd)
+    # NOTE(termie): this appears to be necessary to let the subprocess call
+    #               clean something up in between calls, without it two
+    #               execute calls in a row hangs the second one
+    greenthread.sleep(0)
     return result
 
 
@@ -123,13 +128,7 @@
 
 def runthis(prompt, cmd, check_exit_code=True):
     logging.debug("Running %s" % (cmd))
-    exit_code = subprocess.call(cmd.split(" "))
-    logging.debug(prompt % (exit_code))
-    if check_exit_code and exit_code != 0:
-        raise ProcessExecutionError(exit_code=exit_code,
-                                    stdout=None,
-                                    stderr=None,
-                                    cmd=cmd)
+    rv, err = execute(cmd, check_exit_code=check_exit_code)
 
 
 def generate_uid(topic, size=8):
@@ -224,10 +223,41 @@
         return getattr(backend, key)
 
 
-def deferredToThread(f):
-    def g(*args, **kwargs):
-        return deferToThread(f, *args, **kwargs)
-    return g
+class LoopingCall(object):
+    def __init__(self, f=None, *args, **kw):
+        self.args = args
+        self.kw = kw
+        self.f = f
+        self._running = False
+
+    def start(self, interval, now=True):
+        self._running = True
+        done = event.Event()
+
+        def _inner():
+            if not now:
+                greenthread.sleep(interval)
+            try:
+                while self._running:
+                    self.f(*self.args, **self.kw)
+                    greenthread.sleep(interval)
+            except Exception:
+                logging.exception('in looping call')
+                done.send_exception(*sys.exc_info())
+                return
+
+            done.send(True)
+
+        self.done = done
+
+        greenthread.spawn(_inner)
+        return self.done
+
+    def stop(self):
+        self._running = False
+
+    def wait(self):
+        return self.done.wait()
 
 
 def xhtml_escape(value):

=== removed file 'nova/validate.py'
--- nova/validate.py	2010-10-21 18:49:51 +0000
+++ nova/validate.py	1970-01-01 00:00:00 +0000
@@ -1,94 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# All Rights Reserved.
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-"""Decorators for argument validation, courtesy of
-http://rmi.net/~lutz/rangetest.html"""
-
-
-def rangetest(**argchecks):
-    """Validate ranges for both + defaults"""
-
-    def onDecorator(func):
-        """onCall remembers func and argchecks"""
-        import sys
-        code = func.__code__ if sys.version_info[0] == 3 else func.func_code
-        allargs = code.co_varnames[:code.co_argcount]
-        funcname = func.__name__
-
-        def onCall(*pargs, **kargs):
-            # all pargs match first N args by position
-            # the rest must be in kargs or omitted defaults
-            positionals = list(allargs)
-            positionals = positionals[:len(pargs)]
-
-            for (argname, (low, high)) in argchecks.items():
-                # for all args to be checked
-                if argname in kargs:
-                    # was passed by name
-                    if float(kargs[argname]) < low or \
-                       float(kargs[argname]) > high:
-                        errmsg = '{0} argument "{1}" not in {2}..{3}'
-                        errmsg = errmsg.format(funcname, argname, low, high)
-                        raise TypeError(errmsg)
-
-                elif argname in positionals:
-                    # was passed by position
-                    position = positionals.index(argname)
-                    if float(pargs[position]) < low or \
-                       float(pargs[position]) > high:
-                        errmsg = '{0} argument "{1}" with value of {4} ' \
-                                 'not in {2}..{3}'
-                        errmsg = errmsg.format(funcname, argname, low, high,
-                                               pargs[position])
-                        raise TypeError(errmsg)
-                else:
-                    pass
-
-            return func(*pargs, **kargs)    # okay: run original call
-        return onCall
-    return onDecorator
-
-
-def typetest(**argchecks):
-    def onDecorator(func):
-        import sys
-        code = func.__code__ if sys.version_info[0] == 3 else func.func_code
-        allargs = code.co_varnames[:code.co_argcount]
-        funcname = func.__name__
-
-        def onCall(*pargs, **kargs):
-            positionals = list(allargs)[:len(pargs)]
-            for (argname, typeof) in argchecks.items():
-                if argname in kargs:
-                    if not isinstance(kargs[argname], typeof):
-                        errmsg = '{0} argument "{1}" not of type {2}'
-                        errmsg = errmsg.format(funcname, argname, typeof)
-                        raise TypeError(errmsg)
-                elif argname in positionals:
-                    position = positionals.index(argname)
-                    if not isinstance(pargs[position], typeof):
-                        errmsg = '{0} argument "{1}" with value of {2} ' \
-                                 'not of type {3}'
-                        errmsg = errmsg.format(funcname, argname,
-                                               pargs[position], typeof)
-                        raise TypeError(errmsg)
-                else:
-                    pass
-            return func(*pargs, **kargs)
-        return onCall
-    return onDecorator

=== modified file 'nova/virt/fake.py'
--- nova/virt/fake.py	2010-11-01 20:13:18 +0000
+++ nova/virt/fake.py	2010-12-16 20:49:10 +0000
@@ -25,8 +25,6 @@
 
 """
 
-from twisted.internet import defer
-
 from nova import exception
 from nova.compute import power_state
 
@@ -107,7 +105,6 @@
         fake_instance = FakeInstance()
         self.instances[instance.name] = fake_instance
         fake_instance._state = power_state.RUNNING
-        return defer.succeed(None)
 
     def reboot(self, instance):
         """
@@ -119,19 +116,19 @@
         The work will be done asynchronously.  This function returns a
         Deferred that allows the caller to detect when it is complete.
         """
-        return defer.succeed(None)
+        pass
 
     def rescue(self, instance):
         """
         Rescue the specified instance.
         """
-        return defer.succeed(None)
+        pass
 
     def unrescue(self, instance):
         """
         Unrescue the specified instance.
         """
-        return defer.succeed(None)
+        pass
 
     def destroy(self, instance):
         """
@@ -144,7 +141,6 @@
         Deferred that allows the caller to detect when it is complete.
         """
         del self.instances[instance.name]
-        return defer.succeed(None)
 
     def attach_volume(self, instance_name, device_path, mountpoint):
         """Attach the disk at device_path to the instance at mountpoint"""

=== modified file 'nova/virt/images.py'
--- nova/virt/images.py	2010-10-22 00:15:21 +0000
+++ nova/virt/images.py	2010-12-16 20:49:10 +0000
@@ -26,7 +26,7 @@
 import urlparse
 
 from nova import flags
-from nova import process
+from nova import utils
 from nova.auth import manager
 from nova.auth import signer
 from nova.objectstore import image
@@ -50,7 +50,7 @@
 
     # This should probably move somewhere else, like e.g. a download_as
     # method on User objects and at the same time get rewritten to use
-    # twisted web client.
+    # a web client.
     headers = {}
     headers['Date'] = time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime())
 
@@ -63,15 +63,16 @@
 
     cmd = ['/usr/bin/curl', '--fail', '--silent', url]
     for (k, v) in headers.iteritems():
-        cmd += ['-H', '%s: %s' % (k, v)]
+        cmd += ['-H', '"%s: %s"' % (k, v)]
 
     cmd += ['-o', path]
-    return process.SharedPool().execute(executable=cmd[0], args=cmd[1:])
+    cmd_out = ' '.join(cmd)
+    return utils.execute(cmd_out)
 
 
 def _fetch_local_image(image, path, user, project):
     source = _image_path('%s/image' % image)
-    return process.simple_execute('cp %s %s' % (source, path))
+    return utils.execute('cp %s %s' % (source, path))
 
 
 def _image_path(path):

=== modified file 'nova/virt/libvirt_conn.py'
--- nova/virt/libvirt_conn.py	2010-11-17 19:56:42 +0000
+++ nova/virt/libvirt_conn.py	2010-12-16 20:49:10 +0000
@@ -45,16 +45,15 @@
 import os
 import shutil
 
+from eventlet import event
+from eventlet import tpool
+
 import IPy
-from twisted.internet import defer
-from twisted.internet import task
-from twisted.internet import threads
 
 from nova import context
 from nova import db
 from nova import exception
 from nova import flags
-from nova import process
 from nova import utils
 #from nova.api import context
 from nova.auth import manager
@@ -184,14 +183,12 @@
         except Exception as _err:
             pass
             # If the instance is already terminated, we're still happy
-        d = defer.Deferred()
-        if cleanup:
-            d.addCallback(lambda _: self._cleanup(instance))
-        # FIXME: What does this comment mean?
-        # TODO(termie): short-circuit me for tests
-        # WE'LL save this for when we do shutdown,
+
+        done = event.Event()
+
+        # We'll save this for when we do shutdown,
         # instead of destroy - but destroy returns immediately
-        timer = task.LoopingCall(f=None)
+        timer = utils.LoopingCall(f=None)
 
         def _wait_for_shutdown():
             try:
@@ -200,17 +197,26 @@
                                       instance['id'], state)
                 if state == power_state.SHUTDOWN:
                     timer.stop()
-                    d.callback(None)
             except Exception:
                 db.instance_set_state(context.get_admin_context(),
                                       instance['id'],
                                       power_state.SHUTDOWN)
                 timer.stop()
-                d.callback(None)
 
         timer.f = _wait_for_shutdown
-        timer.start(interval=0.5, now=True)
-        return d
+        timer_done = timer.start(interval=0.5, now=True)
+
+        # NOTE(termie): this is strictly superfluous (we could put the
+        #               cleanup code in the timer), but this emulates the
+        #               previous model so I am keeping it around until
+        #               everything has been vetted a bit
+        def _wait_for_timer():
+            timer_done.wait()
+            self._cleanup(instance)
+            done.send()
+
+        greenthread.spawn(_wait_for_timer)
+        return done
 
     def _cleanup(self, instance):
         target = os.path.join(FLAGS.instances_path, instance['name'])
@@ -219,7 +225,6 @@
         if os.path.exists(target):
             shutil.rmtree(target)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def attach_volume(self, instance_name, device_path, mountpoint):
         virt_dom = self._conn.lookupByName(instance_name)
@@ -230,7 +235,6 @@
                      <target dev='%s' bus='virtio'/>
                  </disk>""" % (device_path, mount_device)
         virt_dom.attachDevice(xml)
-        yield
 
     def _get_disk_xml(self, xml, device):
         """Returns the xml for the disk mounted at device"""
@@ -252,7 +256,6 @@
             if doc != None:
                 doc.freeDoc()
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def detach_volume(self, instance_name, mountpoint):
         virt_dom = self._conn.lookupByName(instance_name)
@@ -261,17 +264,13 @@
         if not xml:
             raise exception.NotFound("No disk at %s" % mount_device)
         virt_dom.detachDevice(xml)
-        yield
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def reboot(self, instance):
-        yield self.destroy(instance, False)
+        self.destroy(instance, False)
         xml = self.to_xml(instance)
-        yield self._conn.createXML(xml, 0)
-
-        d = defer.Deferred()
-        timer = task.LoopingCall(f=None)
+        self._conn.createXML(xml, 0)
+        timer = utils.LoopingCall(f=None)
 
         def _wait_for_reboot():
             try:
@@ -281,33 +280,28 @@
                 if state == power_state.RUNNING:
                     logging.debug('instance %s: rebooted', instance['name'])
                     timer.stop()
-                    d.callback(None)
             except Exception, exn:
                 logging.error('_wait_for_reboot failed: %s', exn)
                 db.instance_set_state(context.get_admin_context(),
                                       instance['id'],
                                       power_state.SHUTDOWN)
                 timer.stop()
-                d.callback(None)
 
         timer.f = _wait_for_reboot
-        timer.start(interval=0.5, now=True)
-        yield d
+        return timer.start(interval=0.5, now=True)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def rescue(self, instance):
-        yield self.destroy(instance, False)
+        self.destroy(instance, False)
 
         xml = self.to_xml(instance, rescue=True)
         rescue_images = {'image_id': FLAGS.rescue_image_id,
                          'kernel_id': FLAGS.rescue_kernel_id,
                          'ramdisk_id': FLAGS.rescue_ramdisk_id}
-        yield self._create_image(instance, xml, 'rescue-', rescue_images)
-        yield self._conn.createXML(xml, 0)
+        self._create_image(instance, xml, 'rescue-', rescue_images)
+        self._conn.createXML(xml, 0)
 
-        d = defer.Deferred()
-        timer = task.LoopingCall(f=None)
+        timer = utils.LoopingCall(f=None)
 
         def _wait_for_rescue():
             try:
@@ -316,27 +310,22 @@
                 if state == power_state.RUNNING:
                     logging.debug('instance %s: rescued', instance['name'])
                     timer.stop()
-                    d.callback(None)
             except Exception, exn:
                 logging.error('_wait_for_rescue failed: %s', exn)
                 db.instance_set_state(None,
                                       instance['id'],
                                       power_state.SHUTDOWN)
                 timer.stop()
-                d.callback(None)
 
         timer.f = _wait_for_rescue
-        timer.start(interval=0.5, now=True)
-        yield d
+        return timer.start(interval=0.5, now=True)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def unrescue(self, instance):
         # NOTE(vish): Because reboot destroys and recreates an instance using
         #             the normal xml file, we can just call reboot here
-        yield self.reboot(instance)
+        self.reboot(instance)
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def spawn(self, instance):
         xml = self.to_xml(instance)
@@ -344,14 +333,12 @@
                               instance['id'],
                               power_state.NOSTATE,
                               'launching')
-        yield NWFilterFirewall(self._conn).\
-              setup_nwfilters_for_instance(instance)
-        yield self._create_image(instance, xml)
-        yield self._conn.createXML(xml, 0)
+        NWFilterFirewall(self._conn).setup_nwfilters_for_instance(instance)
+        self._create_image(instance, xml)
+        self._conn.createXML(xml, 0)
         logging.debug("instance %s: is running", instance['name'])
 
-        local_d = defer.Deferred()
-        timer = task.LoopingCall(f=None)
+        timer = utils.LoopingCall(f=None)
 
         def _wait_for_boot():
             try:
@@ -361,7 +348,6 @@
                 if state == power_state.RUNNING:
                     logging.debug('instance %s: booted', instance['name'])
                     timer.stop()
-                    local_d.callback(None)
             except:
                 logging.exception('instance %s: failed to boot',
                                   instance['name'])
@@ -369,10 +355,9 @@
                                       instance['id'],
                                       power_state.SHUTDOWN)
                 timer.stop()
-                local_d.callback(None)
+
         timer.f = _wait_for_boot
-        timer.start(interval=0.5, now=True)
-        yield local_d
+        return timer.start(interval=0.5, now=True)
 
     def _flush_xen_console(self, virsh_output):
         logging.info('virsh said: %r' % (virsh_output,))
@@ -380,10 +365,9 @@
 
         if virsh_output.startswith('/dev/'):
             logging.info('cool, it\'s a device')
-            d = process.simple_execute("sudo dd if=%s iflag=nonblock" %
-                                       virsh_output, check_exit_code=False)
-            d.addCallback(lambda r: r[0])
-            return d
+            out, err = utils.execute("sudo dd if=%s iflag=nonblock" %
+                                     virsh_output, check_exit_code=False)
+            return out
         else:
             return ''
 
@@ -403,21 +387,20 @@
     def get_console_output(self, instance):
         console_log = os.path.join(FLAGS.instances_path, instance['name'],
                                    'console.log')
-        d = process.simple_execute('sudo chown %d %s' % (os.getuid(),
-                                   console_log))
+
+        utils.execute('sudo chown %d %s' % (os.getuid(), console_log))
+
         if FLAGS.libvirt_type == 'xen':
-            # Xen is spethial
-            d.addCallback(lambda _:
-                process.simple_execute("virsh ttyconsole %s" %
-                                       instance['name']))
-            d.addCallback(self._flush_xen_console)
-            d.addCallback(self._append_to_file, console_log)
+            # Xen is special
+            virsh_output = utils.execute("virsh ttyconsole %s" %
+                                         instance['name'])
+            data = self._flush_xen_console(virsh_output)
+            fpath = self._append_to_file(data, console_log)
         else:
-            d.addCallback(lambda _: defer.succeed(console_log))
-        d.addCallback(self._dump_file)
-        return d
-
-    @defer.inlineCallbacks
+            fpath = console_log
+
+        return self._dump_file(fpath)
+
     def _create_image(self, inst, libvirt_xml, prefix='', disk_images=None):
         # syntactic nicety
         basepath = lambda fname = '', prefix = prefix: os.path.join(
@@ -426,8 +409,8 @@
                                                  prefix + fname)
 
         # ensure directories exist and are writable
-        yield process.simple_execute('mkdir -p %s' % basepath(prefix=''))
-        yield process.simple_execute('chmod 0777 %s' % basepath(prefix=''))
+        utils.execute('mkdir -p %s' % basepath(prefix=''))
+        utils.execute('chmod 0777 %s' % basepath(prefix=''))
 
         # TODO(termie): these are blocking calls, it would be great
         #               if they weren't.
@@ -448,19 +431,19 @@
                            'kernel_id': inst['kernel_id'],
                            'ramdisk_id': inst['ramdisk_id']}
         if not os.path.exists(basepath('disk')):
-            yield images.fetch(inst.image_id, basepath('disk-raw'), user,
-                               project)
+            images.fetch(inst.image_id, basepath('disk-raw'), user,
+                         project)
         if not os.path.exists(basepath('kernel')):
-            yield images.fetch(inst.kernel_id, basepath('kernel'), user,
-                               project)
+            images.fetch(inst.kernel_id, basepath('kernel'), user,
+                         project)
         if not os.path.exists(basepath('ramdisk')):
-            yield images.fetch(inst.ramdisk_id, basepath('ramdisk'), user,
-                               project)
+            images.fetch(inst.ramdisk_id, basepath('ramdisk'), user,
+                         project)
 
-        execute = lambda cmd, process_input = None, check_exit_code = True: \
-                  process.simple_execute(cmd=cmd,
-                                         process_input=process_input,
-                                         check_exit_code=check_exit_code)
+        def execute(cmd, process_input=None, check_exit_code=True):
+            return utils.execute(cmd=cmd,
+                                 process_input=process_input,
+                                 check_exit_code=check_exit_code)
 
         key = str(inst['key_data'])
         net = None
@@ -482,11 +465,11 @@
             if net:
                 logging.info('instance %s: injecting net into image %s',
                     inst['name'], inst.image_id)
-            yield disk.inject_data(basepath('disk-raw'), key, net,
-                                   execute=execute)
+            disk.inject_data(basepath('disk-raw'), key, net,
+                             execute=execute)
 
         if os.path.exists(basepath('disk')):
-            yield process.simple_execute('rm -f %s' % basepath('disk'))
+            utils.execute('rm -f %s' % basepath('disk'))
 
         local_bytes = (instance_types.INSTANCE_TYPES[inst.instance_type]
                                                     ['local_gb']
@@ -495,12 +478,11 @@
         resize = True
         if inst['instance_type'] == 'm1.tiny' or prefix == 'rescue-':
             resize = False
-        yield disk.partition(basepath('disk-raw'), basepath('disk'),
-                             local_bytes, resize, execute=execute)
+        disk.partition(basepath('disk-raw'), basepath('disk'),
+                       local_bytes, resize, execute=execute)
 
         if FLAGS.libvirt_type == 'uml':
-            yield process.simple_execute('sudo chown root %s' %
-                                         basepath('disk'))
+            utils.execute('sudo chown root %s' % basepath('disk'))
 
     def to_xml(self, instance, rescue=False):
         # TODO(termie): cache?
@@ -758,15 +740,15 @@
     def _define_filter(self, xml):
         if callable(xml):
             xml = xml()
-        d = threads.deferToThread(self._conn.nwfilterDefineXML, xml)
-        return d
+
+        # execute in a native thread and block current greenthread until done
+        tpool.execute(self._conn.nwfilterDefineXML, xml)
 
     @staticmethod
     def _get_net_and_mask(cidr):
         net = IPy.IP(cidr)
         return str(net.net()), str(net.netmask())
 
-    @defer.inlineCallbacks
     def setup_nwfilters_for_instance(self, instance):
         """
         Creates an NWFilter for the given instance. In the process,
@@ -774,10 +756,10 @@
         the base filter are all in place.
         """
 
-        yield self._define_filter(self.nova_base_ipv4_filter)
-        yield self._define_filter(self.nova_base_ipv6_filter)
-        yield self._define_filter(self.nova_dhcp_filter)
-        yield self._define_filter(self.nova_base_filter)
+        self._define_filter(self.nova_base_ipv4_filter)
+        self._define_filter(self.nova_base_ipv6_filter)
+        self._define_filter(self.nova_dhcp_filter)
+        self._define_filter(self.nova_base_filter)
 
         nwfilter_xml = "<filter name='nova-instance-%s' chain='root'>\n" \
                        "  <filterref filter='nova-base' />\n" % \
@@ -789,20 +771,19 @@
             net, mask = self._get_net_and_mask(network_ref['cidr'])
             project_filter = self.nova_project_filter(instance['project_id'],
                                                       net, mask)
-            yield self._define_filter(project_filter)
+            self._define_filter(project_filter)
 
             nwfilter_xml += "  <filterref filter='nova-project-%s' />\n" % \
                             instance['project_id']
 
         for security_group in instance.security_groups:
-            yield self.ensure_security_group_filter(security_group['id'])
+            self.ensure_security_group_filter(security_group['id'])
 
             nwfilter_xml += "  <filterref filter='nova-secgroup-%d' />\n" % \
                             security_group['id']
         nwfilter_xml += "</filter>"
 
-        yield self._define_filter(nwfilter_xml)
-        return
+        self._define_filter(nwfilter_xml)
 
     def ensure_security_group_filter(self, security_group_id):
         return self._define_filter(

=== modified file 'nova/virt/xenapi/network_utils.py'
--- nova/virt/xenapi/network_utils.py	2010-12-06 12:42:34 +0000
+++ nova/virt/xenapi/network_utils.py	2010-12-16 20:49:10 +0000
@@ -20,8 +20,6 @@
 their lookup functions.
 """
 
-from twisted.internet import defer
-
 
 class NetworkHelper():
     """
@@ -31,14 +29,12 @@
         return
 
     @classmethod
-    @defer.inlineCallbacks
     def find_network_with_bridge(cls, session, bridge):
-        """ Return the network on which the bridge is attached, if found """
+        """ Return the network on which the bridge is attached, if found."""
         expr = 'field "bridge" = "%s"' % bridge
-        networks = yield session.call_xenapi('network.get_all_records_where',
-                                           expr)
+        networks = session.call_xenapi('network.get_all_records_where', expr)
         if len(networks) == 1:
-            defer.returnValue(networks.keys()[0])
+            return networks.keys()[0]
         elif len(networks) > 1:
             raise Exception('Found non-unique network for bridge %s' % bridge)
         else:

=== modified file 'nova/virt/xenapi/vm_utils.py'
--- nova/virt/xenapi/vm_utils.py	2010-12-09 19:37:30 +0000
+++ nova/virt/xenapi/vm_utils.py	2010-12-16 20:49:10 +0000
@@ -21,19 +21,14 @@
 
 import logging
 import urllib
-
-from twisted.internet import defer
 from xml.dom import minidom
 
-from nova import flags
 from nova import utils
-
 from nova.auth.manager import AuthManager
 from nova.compute import instance_types
 from nova.compute import power_state
 from nova.virt import images
 
-FLAGS = flags.FLAGS
 
 XENAPI_POWER_STATE = {
     'Halted': power_state.SHUTDOWN,
@@ -42,6 +37,7 @@
     'Suspended': power_state.SHUTDOWN,  # FIXME
     'Crashed': power_state.CRASHED}
 
+
 XenAPI = None
 
 
@@ -64,7 +60,6 @@
             XenAPI = __import__('XenAPI')
 
     @classmethod
-    @defer.inlineCallbacks
     def create_vm(cls, session, instance, kernel, ramdisk):
         """Create a VM record.  Returns a Deferred that gives the new
         VM reference."""
@@ -102,12 +97,11 @@
             'other_config': {},
             }
         logging.debug('Created VM %s...', instance.name)
-        vm_ref = yield session.call_xenapi('VM.create', rec)
+        vm_ref = session.call_xenapi('VM.create', rec)
         logging.debug('Created VM %s as %s.', instance.name, vm_ref)
-        defer.returnValue(vm_ref)
+        return vm_ref
 
     @classmethod
-    @defer.inlineCallbacks
     def create_vbd(cls, session, vm_ref, vdi_ref, userdevice, bootable):
         """Create a VBD record.  Returns a Deferred that gives the new
         VBD reference."""
@@ -126,13 +120,12 @@
         vbd_rec['qos_algorithm_params'] = {}
         vbd_rec['qos_supported_algorithms'] = []
         logging.debug('Creating VBD for VM %s, VDI %s ... ', vm_ref, vdi_ref)
-        vbd_ref = yield session.call_xenapi('VBD.create', vbd_rec)
+        vbd_ref = session.call_xenapi('VBD.create', vbd_rec)
         logging.debug('Created VBD %s for VM %s, VDI %s.', vbd_ref, vm_ref,
                       vdi_ref)
-        defer.returnValue(vbd_ref)
+        return vbd_ref
 
     @classmethod
-    @defer.inlineCallbacks
     def create_vif(cls, session, vm_ref, network_ref, mac_address):
         """Create a VIF record.  Returns a Deferred that gives the new
         VIF reference."""
@@ -148,13 +141,12 @@
         vif_rec['qos_algorithm_params'] = {}
         logging.debug('Creating VIF for VM %s, network %s ... ', vm_ref,
                       network_ref)
-        vif_ref = yield session.call_xenapi('VIF.create', vif_rec)
+        vif_ref = session.call_xenapi('VIF.create', vif_rec)
         logging.debug('Created VIF %s for VM %s, network %s.', vif_ref,
                       vm_ref, network_ref)
-        defer.returnValue(vif_ref)
+        return vif_ref
 
     @classmethod
-    @defer.inlineCallbacks
     def fetch_image(cls, session, image, user, project, use_sr):
         """use_sr: True to put the image as a VDI in an SR, False to place
         it on dom0's filesystem.  The former is for VM disks, the latter for
@@ -171,12 +163,11 @@
         args['password'] = user.secret
         if use_sr:
             args['add_partition'] = 'true'
-        task = yield session.async_call_plugin('objectstore', fn, args)
-        uuid = yield session.wait_for_task(task)
-        defer.returnValue(uuid)
+        task = session.async_call_plugin('objectstore', fn, args)
+        uuid = session.wait_for_task(task)
+        return uuid
 
     @classmethod
-    @utils.deferredToThread
     def lookup(cls, session, i):
         """ Look the instance i up, and returns it if available """
         return VMHelper.lookup_blocking(session, i)
@@ -194,7 +185,6 @@
             return vms[0]
 
     @classmethod
-    @utils.deferredToThread
     def lookup_vm_vdis(cls, session, vm):
         """ Look for the VDIs that are attached to the VM """
         return VMHelper.lookup_vm_vdis_blocking(session, vm)

=== modified file 'nova/virt/xenapi/vmops.py'
--- nova/virt/xenapi/vmops.py	2010-12-09 17:08:24 +0000
+++ nova/virt/xenapi/vmops.py	2010-12-16 20:49:10 +0000
@@ -20,8 +20,6 @@
 
 import logging
 
-from twisted.internet import defer
-
 from nova import db
 from nova import context
 
@@ -49,10 +47,9 @@
         return [self._session.get_xenapi().VM.get_name_label(vm) \
                 for vm in self._session.get_xenapi().VM.get_all()]
 
-    @defer.inlineCallbacks
     def spawn(self, instance):
         """ Create VM instance """
-        vm = yield VMHelper.lookup(self._session, instance.name)
+        vm = VMHelper.lookup(self._session, instance.name)
         if vm is not None:
             raise Exception('Attempted to create non-unique name %s' %
                             instance.name)
@@ -60,66 +57,63 @@
         bridge = db.project_get_network(context.get_admin_context(),
                                       instance.project_id).bridge
         network_ref = \
-            yield NetworkHelper.find_network_with_bridge(self._session, bridge)
+            NetworkHelper.find_network_with_bridge(self._session, bridge)
 
         user = AuthManager().get_user(instance.user_id)
         project = AuthManager().get_project(instance.project_id)
-        vdi_uuid = yield VMHelper.fetch_image(self._session,
-            instance.image_id, user, project, True)
-        kernel = yield VMHelper.fetch_image(self._session,
-            instance.kernel_id, user, project, False)
-        ramdisk = yield VMHelper.fetch_image(self._session,
-            instance.ramdisk_id, user, project, False)
-        vdi_ref = yield self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid)
-        vm_ref = yield VMHelper.create_vm(self._session,
-                                          instance, kernel, ramdisk)
-        yield VMHelper.create_vbd(self._session, vm_ref, vdi_ref, 0, True)
+        vdi_uuid = VMHelper.fetch_image(
+                self._session, instance.image_id, user, project, True)
+        kernel = VMHelper.fetch_image(
+                self._session, instance.kernel_id, user, project, False)
+        ramdisk = VMHelper.fetch_image(
+                self._session, instance.ramdisk_id, user, project, False)
+        vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid)
+        vm_ref = VMHelper.create_vm(
+                self._session, instance, kernel, ramdisk)
+        VMHelper.create_vbd(self._session, vm_ref, vdi_ref, 0, True)
         if network_ref:
-            yield VMHelper.create_vif(self._session, vm_ref,
-                                      network_ref, instance.mac_address)
+            VMHelper.create_vif(self._session, vm_ref,
+                                network_ref, instance.mac_address)
         logging.debug('Starting VM %s...', vm_ref)
-        yield self._session.call_xenapi('VM.start', vm_ref, False, False)
+        self._session.call_xenapi('VM.start', vm_ref, False, False)
         logging.info('Spawning VM %s created %s.', instance.name,
                      vm_ref)
 
-    @defer.inlineCallbacks
     def reboot(self, instance):
         """ Reboot VM instance """
         instance_name = instance.name
-        vm = yield VMHelper.lookup(self._session, instance_name)
+        vm = VMHelper.lookup(self._session, instance_name)
         if vm is None:
             raise Exception('instance not present %s' % instance_name)
-        task = yield self._session.call_xenapi('Async.VM.clean_reboot', vm)
-        yield self._session.wait_for_task(task)
+        task = self._session.call_xenapi('Async.VM.clean_reboot', vm)
+        self._session.wait_for_task(task)
 
-    @defer.inlineCallbacks
     def destroy(self, instance):
         """ Destroy VM instance """
-        vm = yield VMHelper.lookup(self._session, instance.name)
+        vm = VMHelper.lookup(self._session, instance.name)
         if vm is None:
             # Don't complain, just return.  This lets us clean up instances
             # that have already disappeared from the underlying platform.
-            defer.returnValue(None)
+            return
         # Get the VDIs related to the VM
-        vdis = yield VMHelper.lookup_vm_vdis(self._session, vm)
+        vdis = VMHelper.lookup_vm_vdis(self._session, vm)
         try:
-            task = yield self._session.call_xenapi('Async.VM.hard_shutdown',
+            task = self._session.call_xenapi('Async.VM.hard_shutdown',
                                                    vm)
-            yield self._session.wait_for_task(task)
+            self._session.wait_for_task(task)
         except XenAPI.Failure, exc:
             logging.warn(exc)
         # Disk clean-up
         if vdis:
             for vdi in vdis:
                 try:
-                    task = yield self._session.call_xenapi('Async.VDI.destroy',
-                                                           vdi)
-                    yield self._session.wait_for_task(task)
+                    task = self._session.call_xenapi('Async.VDI.destroy', vdi)
+                    self._session.wait_for_task(task)
                 except XenAPI.Failure, exc:
                     logging.warn(exc)
         try:
-            task = yield self._session.call_xenapi('Async.VM.destroy', vm)
-            yield self._session.wait_for_task(task)
+            task = self._session.call_xenapi('Async.VM.destroy', vm)
+            self._session.wait_for_task(task)
         except XenAPI.Failure, exc:
             logging.warn(exc)
 
@@ -131,14 +125,13 @@
         rec = self._session.get_xenapi().VM.get_record(vm)
         return VMHelper.compile_info(rec)
 
-    @defer.inlineCallbacks
     def get_diagnostics(self, instance_id):
         """Return data about VM diagnostics"""
-        vm = yield VMHelper.lookup(self._session, instance_id)
+        vm = VMHelper.lookup(self._session, instance_id)
         if vm is None:
             raise Exception("instance not present %s" % instance_id)
-        rec = yield self._session.get_xenapi().VM.get_record(vm)
-        defer.returnValue(VMHelper.compile_diagnostics(self._session, rec))
+        rec = self._session.get_xenapi().VM.get_record(vm)
+        return VMHelper.compile_diagnostics(self._session, rec)
 
     def get_console_output(self, instance):
         """ Return snapshot of console """

=== modified file 'nova/virt/xenapi_conn.py'
--- nova/virt/xenapi_conn.py	2010-12-08 20:16:49 +0000
+++ nova/virt/xenapi_conn.py	2010-12-16 20:49:10 +0000
@@ -48,10 +48,11 @@
 """
 
 import logging
+import sys
 import xmlrpclib
 
-from twisted.internet import defer
-from twisted.internet import reactor
+from eventlet import event
+from eventlet import tpool
 
 from nova import utils
 from nova import flags
@@ -159,53 +160,51 @@
         """ Return the xenapi host """
         return self._session.xenapi.session.get_this_host(self._session.handle)
 
-    @utils.deferredToThread
     def call_xenapi(self, method, *args):
-        """Call the specified XenAPI method on a background thread.  Returns
-        a Deferred for the result."""
+        """Call the specified XenAPI method on a background thread."""
         f = self._session.xenapi
         for m in method.split('.'):
             f = f.__getattr__(m)
-        return f(*args)
+        return tpool.execute(f, *args)
 
-    @utils.deferredToThread
     def async_call_plugin(self, plugin, fn, args):
-        """Call Async.host.call_plugin on a background thread.  Returns a
-        Deferred with the task reference."""
-        return _unwrap_plugin_exceptions(
-            self._session.xenapi.Async.host.call_plugin,
-            self.get_xenapi_host(), plugin, fn, args)
+        """Call Async.host.call_plugin on a background thread."""
+        return tpool.execute(_unwrap_plugin_exceptions,
+                             self._session.xenapi.Async.host.call_plugin,
+                             self.get_xenapi_host(), plugin, fn, args)
 
     def wait_for_task(self, task):
         """Return a Deferred that will give the result of the given task.
         The task is polled until it completes."""
-        d = defer.Deferred()
-        reactor.callLater(0, self._poll_task, task, d)
-        return d
-
-    @utils.deferredToThread
-    def _poll_task(self, task, deferred):
+
+        done = event.Event()
+        loop = utils.LoopingCall(self._poll_task, task, done)
+        loop.start(FLAGS.xenapi_task_poll_interval, now=True)
+        rv = done.wait()
+        loop.stop()
+        return rv
+
+    def _poll_task(self, task, done):
         """Poll the given XenAPI task, and fire the given Deferred if we
         get a result."""
         try:
             #logging.debug('Polling task %s...', task)
             status = self._session.xenapi.task.get_status(task)
             if status == 'pending':
-                reactor.callLater(FLAGS.xenapi_task_poll_interval,
-                                  self._poll_task, task, deferred)
+                return
             elif status == 'success':
                 result = self._session.xenapi.task.get_result(task)
                 logging.info('Task %s status: success.  %s', task, result)
-                deferred.callback(_parse_xmlrpc_value(result))
+                done.send(_parse_xmlrpc_value(result))
             else:
                 error_info = self._session.xenapi.task.get_error_info(task)
                 logging.warn('Task %s status: %s.  %s', task, status,
                              error_info)
-                deferred.errback(XenAPI.Failure(error_info))
-            #logging.debug('Polling task %s done.', task)
+                done.send_exception(XenAPI.Failure(error_info))
+                #logging.debug('Polling task %s done.', task)
         except XenAPI.Failure, exc:
             logging.warn(exc)
-            deferred.errback(exc)
+            done.send_exception(*sys.exc_info())
 
 
 def _unwrap_plugin_exceptions(func, *args, **kwargs):

=== modified file 'nova/volume/driver.py'
--- nova/volume/driver.py	2010-11-03 22:06:00 +0000
+++ nova/volume/driver.py	2010-12-16 20:49:10 +0000
@@ -22,12 +22,10 @@
 
 import logging
 import os
-
-from twisted.internet import defer
+import time
 
 from nova import exception
 from nova import flags
-from nova import process
 from nova import utils
 
 
@@ -55,14 +53,13 @@
 
 class VolumeDriver(object):
     """Executes commands relating to Volumes."""
-    def __init__(self, execute=process.simple_execute,
+    def __init__(self, execute=utils.execute,
                  sync_exec=utils.execute, *args, **kwargs):
         # NOTE(vish): db is set by Manager
         self.db = None
         self._execute = execute
         self._sync_exec = sync_exec
 
-    @defer.inlineCallbacks
     def _try_execute(self, command):
         # NOTE(vish): Volume commands can partially fail due to timing, but
         #             running them a second time on failure will usually
@@ -70,15 +67,15 @@
         tries = 0
         while True:
             try:
-                yield self._execute(command)
-                defer.returnValue(True)
+                self._execute(command)
+                return True
             except exception.ProcessExecutionError:
                 tries = tries + 1
                 if tries >= FLAGS.num_shell_tries:
                     raise
                 logging.exception("Recovering from a failed execute."
                                   "Try number %s", tries)
-                yield self._execute("sleep %s" % tries ** 2)
+                time.sleep(tries ** 2)
 
     def check_for_setup_error(self):
         """Returns an error if prerequisites aren't met"""
@@ -86,53 +83,45 @@
             raise exception.Error("volume group %s doesn't exist"
                                   % FLAGS.volume_group)
 
-    @defer.inlineCallbacks
     def create_volume(self, volume):
         """Creates a logical volume."""
         if int(volume['size']) == 0:
             sizestr = '100M'
         else:
             sizestr = '%sG' % volume['size']
-        yield self._try_execute("sudo lvcreate -L %s -n %s %s" %
-                            (sizestr,
-                             volume['name'],
-                             FLAGS.volume_group))
+        self._try_execute("sudo lvcreate -L %s -n %s %s" %
+                          (sizestr,
+                           volume['name'],
+                           FLAGS.volume_group))
 
-    @defer.inlineCallbacks
     def delete_volume(self, volume):
         """Deletes a logical volume."""
-        yield self._try_execute("sudo lvremove -f %s/%s" %
-                                (FLAGS.volume_group,
-                                 volume['name']))
+        self._try_execute("sudo lvremove -f %s/%s" %
+                          (FLAGS.volume_group,
+                           volume['name']))
 
-    @defer.inlineCallbacks
     def local_path(self, volume):
-        yield  # NOTE(vish): stops deprecation warning
+        # NOTE(vish): stops deprecation warning
         escaped_group = FLAGS.volume_group.replace('-', '--')
         escaped_name = volume['name'].replace('-', '--')
-        defer.returnValue("/dev/mapper/%s-%s" % (escaped_group,
-                                                 escaped_name))
+        return "/dev/mapper/%s-%s" % (escaped_group, escaped_name)
 
     def ensure_export(self, context, volume):
         """Synchronously recreates an export for a logical volume."""
         raise NotImplementedError()
 
-    @defer.inlineCallbacks
     def create_export(self, context, volume):
         """Exports the volume."""
         raise NotImplementedError()
 
-    @defer.inlineCallbacks
     def remove_export(self, context, volume):
         """Removes an export for a logical volume."""
         raise NotImplementedError()
 
-    @defer.inlineCallbacks
     def discover_volume(self, volume):
         """Discover volume on a remote host."""
         raise NotImplementedError()
 
-    @defer.inlineCallbacks
     def undiscover_volume(self, volume):
         """Undiscover volume on a remote host."""
         raise NotImplementedError()
@@ -155,14 +144,13 @@
                 dev = {'shelf_id': shelf_id, 'blade_id': blade_id}
                 self.db.export_device_create_safe(context, dev)
 
-    @defer.inlineCallbacks
     def create_export(self, context, volume):
         """Creates an export for a logical volume."""
         self._ensure_blades(context)
         (shelf_id,
          blade_id) = self.db.volume_allocate_shelf_and_blade(context,
                                                              volume['id'])
-        yield self._try_execute(
+        self._try_execute(
                 "sudo vblade-persist setup %s %s %s /dev/%s/%s" %
                 (shelf_id,
                  blade_id,
@@ -176,33 +164,30 @@
         #             still works for the other volumes, so we
         #             just wait a bit for the current volume to
         #             be ready and ignore any errors.
-        yield self._execute("sleep 2")
-        yield self._execute("sudo vblade-persist auto all",
-                            check_exit_code=False)
-        yield self._execute("sudo vblade-persist start all",
-                            check_exit_code=False)
+        time.sleep(2)
+        self._execute("sudo vblade-persist auto all",
+                      check_exit_code=False)
+        self._execute("sudo vblade-persist start all",
+                      check_exit_code=False)
 
-    @defer.inlineCallbacks
     def remove_export(self, context, volume):
         """Removes an export for a logical volume."""
         (shelf_id,
          blade_id) = self.db.volume_get_shelf_and_blade(context,
                                                         volume['id'])
-        yield self._try_execute("sudo vblade-persist stop %s %s" %
-                                (shelf_id, blade_id))
-        yield self._try_execute("sudo vblade-persist destroy %s %s" %
-                                (shelf_id, blade_id))
+        self._try_execute("sudo vblade-persist stop %s %s" %
+                          (shelf_id, blade_id))
+        self._try_execute("sudo vblade-persist destroy %s %s" %
+                          (shelf_id, blade_id))
 
-    @defer.inlineCallbacks
     def discover_volume(self, _volume):
         """Discover volume on a remote host."""
-        yield self._execute("sudo aoe-discover")
-        yield self._execute("sudo aoe-stat", check_exit_code=False)
+        self._execute("sudo aoe-discover")
+        self._execute("sudo aoe-stat", check_exit_code=False)
 
-    @defer.inlineCallbacks
     def undiscover_volume(self, _volume):
         """Undiscover volume on a remote host."""
-        yield
+        pass
 
 
 class FakeAOEDriver(AOEDriver):
@@ -252,7 +237,6 @@
             target = {'host': host, 'target_num': target_num}
             self.db.iscsi_target_create_safe(context, target)
 
-    @defer.inlineCallbacks
     def create_export(self, context, volume):
         """Creates an export for a logical volume."""
         self._ensure_iscsi_targets(context, volume['host'])
@@ -261,61 +245,55 @@
                                                       volume['host'])
         iscsi_name = "%s%s" % (FLAGS.iscsi_target_prefix, volume['name'])
         volume_path = "/dev/%s/%s" % (FLAGS.volume_group, volume['name'])
-        yield self._execute("sudo ietadm --op new "
-                            "--tid=%s --params Name=%s" %
-                            (iscsi_target, iscsi_name))
-        yield self._execute("sudo ietadm --op new --tid=%s "
-                            "--lun=0 --params Path=%s,Type=fileio" %
-                            (iscsi_target, volume_path))
+        self._execute("sudo ietadm --op new "
+                      "--tid=%s --params Name=%s" %
+                      (iscsi_target, iscsi_name))
+        self._execute("sudo ietadm --op new --tid=%s "
+                      "--lun=0 --params Path=%s,Type=fileio" %
+                      (iscsi_target, volume_path))
 
-    @defer.inlineCallbacks
     def remove_export(self, context, volume):
         """Removes an export for a logical volume."""
         iscsi_target = self.db.volume_get_iscsi_target_num(context,
                                                            volume['id'])
-        yield self._execute("sudo ietadm --op delete --tid=%s "
-                            "--lun=0" % iscsi_target)
-        yield self._execute("sudo ietadm --op delete --tid=%s" %
-                            iscsi_target)
+        self._execute("sudo ietadm --op delete --tid=%s "
+                      "--lun=0" % iscsi_target)
+        self._execute("sudo ietadm --op delete --tid=%s" %
+                      iscsi_target)
 
-    @defer.inlineCallbacks
     def _get_name_and_portal(self, volume_name, host):
         """Gets iscsi name and portal from volume name and host."""
-        (out, _err) = yield self._execute("sudo iscsiadm -m discovery -t "
-                                         "sendtargets -p %s" % host)
+        (out, _err) = self._execute("sudo iscsiadm -m discovery -t "
+                                    "sendtargets -p %s" % host)
         for target in out.splitlines():
             if FLAGS.iscsi_ip_prefix in target and volume_name in target:
                 (location, _sep, iscsi_name) = target.partition(" ")
                 break
         iscsi_portal = location.split(",")[0]
-        defer.returnValue((iscsi_name, iscsi_portal))
+        return (iscsi_name, iscsi_portal)
 
-    @defer.inlineCallbacks
     def discover_volume(self, volume):
         """Discover volume on a remote host."""
-        (iscsi_name,
-         iscsi_portal) = yield self._get_name_and_portal(volume['name'],
-                                                         volume['host'])
-        yield self._execute("sudo iscsiadm -m node -T %s -p %s --login" %
-                            (iscsi_name, iscsi_portal))
-        yield self._execute("sudo iscsiadm -m node -T %s -p %s --op update "
-                            "-n node.startup -v automatic" %
-                            (iscsi_name, iscsi_portal))
-        defer.returnValue("/dev/iscsi/%s" % volume['name'])
+        iscsi_name, iscsi_portal = self._get_name_and_portal(volume['name'],
+                                                             volume['host'])
+        self._execute("sudo iscsiadm -m node -T %s -p %s --login" %
+                      (iscsi_name, iscsi_portal))
+        self._execute("sudo iscsiadm -m node -T %s -p %s --op update "
+                      "-n node.startup -v automatic" %
+                      (iscsi_name, iscsi_portal))
+        return "/dev/iscsi/%s" % volume['name']
 
-    @defer.inlineCallbacks
     def undiscover_volume(self, volume):
         """Undiscover volume on a remote host."""
-        (iscsi_name,
-         iscsi_portal) = yield self._get_name_and_portal(volume['name'],
-                                                         volume['host'])
-        yield self._execute("sudo iscsiadm -m node -T %s -p %s --op update "
-                            "-n node.startup -v manual" %
-                            (iscsi_name, iscsi_portal))
-        yield self._execute("sudo iscsiadm -m node -T %s -p %s --logout " %
-                            (iscsi_name, iscsi_portal))
-        yield self._execute("sudo iscsiadm -m node --op delete "
-                            "--targetname %s" % iscsi_name)
+        iscsi_name, iscsi_portal = self._get_name_and_portal(volume['name'],
+                                                             volume['host'])
+        self._execute("sudo iscsiadm -m node -T %s -p %s --op update "
+                      "-n node.startup -v manual" %
+                      (iscsi_name, iscsi_portal))
+        self._execute("sudo iscsiadm -m node -T %s -p %s --logout " %
+                      (iscsi_name, iscsi_portal))
+        self._execute("sudo iscsiadm -m node --op delete "
+                      "--targetname %s" % iscsi_name)
 
 
 class FakeISCSIDriver(ISCSIDriver):

=== modified file 'nova/volume/manager.py'
--- nova/volume/manager.py	2010-11-03 21:38:14 +0000
+++ nova/volume/manager.py	2010-12-16 20:49:10 +0000
@@ -45,7 +45,6 @@
 import logging
 import datetime
 
-from twisted.internet import defer
 
 from nova import context
 from nova import exception
@@ -86,7 +85,6 @@
         for volume in volumes:
             self.driver.ensure_export(ctxt, volume)
 
-    @defer.inlineCallbacks
     def create_volume(self, context, volume_id):
         """Creates and exports the volume."""
         context = context.elevated()
@@ -102,19 +100,18 @@
 
         logging.debug("volume %s: creating lv of size %sG",
                       volume_ref['name'], volume_ref['size'])
-        yield self.driver.create_volume(volume_ref)
+        self.driver.create_volume(volume_ref)
 
         logging.debug("volume %s: creating export", volume_ref['name'])
-        yield self.driver.create_export(context, volume_ref)
+        self.driver.create_export(context, volume_ref)
 
         now = datetime.datetime.utcnow()
         self.db.volume_update(context,
                               volume_ref['id'], {'status': 'available',
                                                  'launched_at': now})
         logging.debug("volume %s: created successfully", volume_ref['name'])
-        defer.returnValue(volume_id)
+        return volume_id
 
-    @defer.inlineCallbacks
     def delete_volume(self, context, volume_id):
         """Deletes and unexports volume."""
         context = context.elevated()
@@ -124,14 +121,13 @@
         if volume_ref['host'] != self.host:
             raise exception.Error("Volume is not local to this node")
         logging.debug("volume %s: removing export", volume_ref['name'])
-        yield self.driver.remove_export(context, volume_ref)
+        self.driver.remove_export(context, volume_ref)
         logging.debug("volume %s: deleting", volume_ref['name'])
-        yield self.driver.delete_volume(volume_ref)
+        self.driver.delete_volume(volume_ref)
         self.db.volume_destroy(context, volume_id)
         logging.debug("volume %s: deleted successfully", volume_ref['name'])
-        defer.returnValue(True)
+        return True
 
-    @defer.inlineCallbacks
     def setup_compute_volume(self, context, volume_id):
         """Setup remote volume on compute host.
 
@@ -139,17 +135,16 @@
         context = context.elevated()
         volume_ref = self.db.volume_get(context, volume_id)
         if volume_ref['host'] == self.host and FLAGS.use_local_volumes:
-            path = yield self.driver.local_path(volume_ref)
+            path = self.driver.local_path(volume_ref)
         else:
-            path = yield self.driver.discover_volume(volume_ref)
-        defer.returnValue(path)
+            path = self.driver.discover_volume(volume_ref)
+        return path
 
-    @defer.inlineCallbacks
     def remove_compute_volume(self, context, volume_id):
         """Remove remote volume on compute host."""
         context = context.elevated()
         volume_ref = self.db.volume_get(context, volume_id)
         if volume_ref['host'] == self.host and FLAGS.use_local_volumes:
-            defer.returnValue(True)
+            return True
         else:
-            yield self.driver.undiscover_volume(volume_ref)
+            self.driver.undiscover_volume(volume_ref)

=== modified file 'run_tests.py'
--- run_tests.py	2010-12-11 20:10:24 +0000
+++ run_tests.py	2010-12-16 20:49:10 +0000
@@ -39,6 +39,9 @@
 
 """
 
+import eventlet
+eventlet.monkey_patch()
+
 import __main__
 import gettext
 import os
@@ -59,15 +62,12 @@
 from nova.tests.flags_unittest import *
 from nova.tests.misc_unittest import *
 from nova.tests.network_unittest import *
-from nova.tests.objectstore_unittest import *
-from nova.tests.process_unittest import *
+#from nova.tests.objectstore_unittest import *
 from nova.tests.quota_unittest import *
 from nova.tests.rpc_unittest import *
 from nova.tests.scheduler_unittest import *
 from nova.tests.service_unittest import *
 from nova.tests.twistd_unittest import *
-from nova.tests.validator_unittest import *
-from nova.tests.virt_unittest import *
 from nova.tests.virt_unittest import *
 from nova.tests.volume_unittest import *