Merge lp:~exarkun/ampoule/simpler-doWork-deferreds into lp:ampoule

[Jean-Paul Calderone]

Started off just trying to simplify _cb_doWork. This involved getting the tests to pass reliably, though. So, the branch implements one minor simplification to _cb_doWork, and changes a handful of tests in order to make them pass reliably on my desktop.

[dialtone (dialtone)]

review: Approve

=== modified file 'ampoule/main.py'
--- ampoule/main.py	2010-04-28 15:46:27 +0000
+++ ampoule/main.py	2010-04-29 04:10:34 +0000
@@ -189,7 +189,7 @@
         self.uid = uid
         self.gid = gid
         self.usePTY = usePTY
-        self.packages = packages
+        self.packages = ("ampoule",) + packages
         self.childReactor = childReactor
 
     def __repr__(self):

=== modified file 'ampoule/pool.py'
--- ampoule/pool.py	2010-04-28 16:39:58 +0000
+++ ampoule/pool.py	2010-04-29 04:10:34 +0000
@@ -157,8 +157,8 @@
         """
         if self._queue:
             _, (d, command, kwargs) = pop(self._queue)
-            self._cb_doWork(command, _d=d, **kwargs)
-    
+            self._cb_doWork(command, **kwargs).chainDeferred(d)
+
     def _handleTimeout(self, child):
         """
         One of the children went timeout, we need to deal with it
@@ -197,7 +197,7 @@
                                           ampChildArgs=self.ampChildArgs)
         return self._addProcess(child, finished)
     
-    def _cb_doWork(self, command, _d=None, _timeout=None, _deadline=None,
+    def _cb_doWork(self, command, _timeout=None, _deadline=None,
                    **kwargs):
         """
         Go and call the command.
@@ -239,12 +239,7 @@
             # the process might have received tons of calls already
             # which would make it run more calls than what is
             # configured to do.
-            if is_error:
-                _d.errback(result)
-                return _d
-            else:
-                _d.callback(result)
-                return _d
+            return result
         
         die = False
         child = self.ready.pop()
@@ -295,10 +290,8 @@
         
         @param kwargs: dictionary containing the arguments for the command.
         """
-        d = defer.Deferred()
         if self.ready: # there are unused processes, let's use them
-            self._cb_doWork(command, _d=d, **kwargs)
-            return d
+            return self._cb_doWork(command, **kwargs)
         else:
             if len(self.processes) < self.max:
                 # no unused but we can start some new ones
@@ -309,11 +302,11 @@
                 # Process pool with min=1, max=1, recycle_after=1
                 # [call(Command) for x in xrange(BIG_NUMBER)]
                 self.startAWorker()
-                self._cb_doWork(command, _d=d, **kwargs)
-                return d
+                return self._cb_doWork(command, **kwargs)
             else:
                 # No one is free... just queue up and wait for a process
                 # to start and pick up the first item in the queue.
+                d = defer.Deferred()
                 self._queue.append((count(), (d, command, kwargs)))
                 return d
     

=== modified file 'ampoule/test/test_process.py'
--- ampoule/test/test_process.py	2010-03-01 19:35:21 +0000
+++ ampoule/test/test_process.py	2010-04-29 04:10:34 +0000
@@ -309,12 +309,10 @@
         def checkBootstrap(response):
             cwd.append(response['cwd'])
             self.assertNotEquals(cwd, os.getcwd())
-        def assertNotExists(path):
-            self.assertFalse()
-        c.callRemote(GetCWD
-            ).addCallback(checkBootstrap
-            ).addCallback(lambda _: c.callRemote(commands.Shutdown)
-            ).addCallback(lambda _: self.assertFalse(os.path.exists(cwd[0])))
+        d = c.callRemote(GetCWD)
+        d.addCallback(checkBootstrap)
+        d.addCallback(lambda _: c.callRemote(commands.Shutdown))
+        finished.addCallback(lambda _: self.assertFalse(os.path.exists(cwd[0])))
         return finished
 
     def test_BootstrapContextInstance(self):
@@ -325,10 +323,10 @@
         def checkBootstrap(response):
             cwd.append(response['cwd'])
             self.assertTrue(cwd[0].endswith('/foo'))
-        c.callRemote(GetCWD
-            ).addCallback(checkBootstrap
-            ).addCallback(lambda _: c.callRemote(commands.Shutdown)
-            ).addCallback(lambda _: self.assertFalse(os.path.exists(cwd[0])))
+        d = c.callRemote(GetCWD)
+        d.addCallback(checkBootstrap)
+        d.addCallback(lambda _: c.callRemote(commands.Shutdown))
+        finished.addCallback(lambda _: self.assertFalse(os.path.exists(cwd[0])))
         return finished
 
     def test_startAMPAndParentProtocol(self):
@@ -500,11 +498,16 @@
         """
         Test that deferToAMPProcess works as expected.
         """
-        
+        def cleanupGlobalPool():
+            d = pool.pp.stop()
+            pool.pp = None
+            return d
+        self.addCleanup(cleanupGlobalPool)
+
         STRING = "CIAOOOO"
-        return pool.deferToAMPProcess(commands.Echo, data=STRING
-           ).addCallback(lambda result: self.assertEquals(result['response'], STRING)
-           ).addCallback(lambda _: pool.pp.stop())
+        d = pool.deferToAMPProcess(commands.Echo, data=STRING)
+        d.addCallback(self.assertEquals, {"response": STRING})
+        return d
 
     def test_checkStateInPool(self):
         """
@@ -620,6 +623,7 @@
         MIN = 1
         RECYCLE_AFTER = 1
         pp = pool.ProcessPool(ampChild=PidChild, min=MIN, max=MAX, recycleAfter=RECYCLE_AFTER)
+        self.addCleanup(pp.stop)
         
         def _checks(_):
             self.assertEquals(pp.started, True)
@@ -634,13 +638,11 @@
                 ).addCallback(lambda response: response['pid']
                 ).addCallback(self.assertNotEquals, pid)
         
-        def finish(reason):
-            return pp.stop().addCallback(lambda _: reason)
 
-        return pp.start(
-            ).addCallback(_checks
-            ).addCallback(_checks2
-            ).addCallback(finish)
+        d = pp.start()
+        d.addCallback(_checks)
+        d.addCallback(_checks2)
+        return d
     
     def test_recyclingWithQueueOverload(self):
         """
@@ -652,20 +654,29 @@
         RECYCLE_AFTER = 10
         CALLS = 60
         pp = pool.ProcessPool(ampChild=PidChild, min=MIN, max=MAX, recycleAfter=RECYCLE_AFTER)
+        self.addCleanup(pp.stop)
         
         def _check(results):
             s = set()
             for succeed, response in results:
                 s.add(response['pid'])
-            self.assertEquals(len(s), MAX*math.ceil(float(CALLS)/(MAX*RECYCLE_AFTER)))
-        
+
+            # For the first C{MAX} calls, each is basically guaranteed to go to
+            # a different child.  After that, though, there are no guarantees.
+            # All the rest might go to a single child, since the child to
+            # perform a job is selected arbitrarily from the "ready" set.  Fair
+            # distribution of jobs needs to be implemented; right now it's "set
+            # ordering" distribution of jobs.
+            self.assertTrue(len(s) > MAX)
+
         def _work(_):
             l = [pp.doWork(Pid) for x in xrange(CALLS)]
             d = defer.DeferredList(l)
             return d.addCallback(_check)
-        return pp.start(
-            ).addCallback(_work
-            ).addCallback(lambda _: pp.stop())
+        d = pp.start()
+        d.addCallback(_work)
+        return d
+
 
     def test_disableProcessRecycling(self):
         """