Merge lp:~abentley/launchpad/rollback-13801 into lp:launchpad

[Aaron Bentley]

Roll back r13801

= Summary =
Fix bug #836900: twisted job runner cannot handle LeaseHeld

== Proposed fix ==
roll back r13801

== Pre-implementation notes ==

== Implementation details ==

== Tests ==

== Demo and Q/A ==

= Launchpad lint =

Checking for conflicts and issues in changed files.

Linting changed files:
  lib/lp/services/job/tests/test_runner.py
  lib/lp/services/job/runner.py

[Aaron Bentley (abentley)]

review: Approve

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2011-08-25 16:57:30 +0000
+++ lib/lp/services/job/runner.py	2011-08-29 18:40:15 +0000
@@ -40,6 +40,7 @@
 from lazr.delegates import delegates
 import transaction
 from twisted.internet import (
+    defer,
     reactor,
     )
 from twisted.protocols import amp
@@ -60,6 +61,10 @@
 from lp.services.mail.sendmail import MailController
 from lp.services.scripts.base import LaunchpadCronScript
 from lp.services.twistedsupport import run_reactor
+from lp.services.twistedsupport.task import (
+    ParallelLimitedTaskConsumer,
+    PollingTaskSource,
+    )
 
 
 class BaseRunnableJobSource:
@@ -467,21 +472,36 @@
             oops = self._doOops(job, sys.exc_info())
             self._logOopsId(oops['id'])
 
+    def getTaskSource(self):
+        """Return a task source for all jobs in job_source."""
+
+        def producer():
+            while True:
+                # XXX: JonathanLange bug=741204: If we're getting all of the
+                # jobs at the start anyway, we can use a DeferredSemaphore,
+                # instead of the more complex PollingTaskSource, which is
+                # better suited to cases where we don't know how much work
+                # there will be.
+                jobs = list(self.job_source.iterReady())
+                if len(jobs) == 0:
+                    yield None
+                for job in jobs:
+                    yield lambda: self.runJobInSubprocess(job)
+        return PollingTaskSource(5, producer().next)
+
+    def doConsumer(self):
+        """Create a ParallelLimitedTaskConsumer for this job type."""
+        # 1 is hard-coded for now until we're sure we'd get gains by running
+        # more than one at a time.  Note that several tests, including
+        # test_timeout, rely on this being 1.
+        consumer = ParallelLimitedTaskConsumer(1, logger=None)
+        return consumer.consume(self.getTaskSource())
+
     def runAll(self):
-        """Run all ready jobs."""
+        """Run all ready jobs, and any that become ready while running."""
         self.pool.start()
-        try:
-            jobs = list(self.job_source.iterReady())
-            if len(jobs) == 0:
-                self.terminated()
-                return
-            d = self.runJobInSubprocess(jobs[0])
-            for job in jobs[1:]:
-                d.addCallback(lambda ignored: self.runJobInSubprocess(job))
-            d.addCallbacks(self.terminated, self.failed)
-        except:
-            self.terminated()
-            raise
+        d = defer.maybeDeferred(self.doConsumer)
+        d.addCallbacks(self.terminated, self.failed)
 
     def terminated(self, ignored=None):
         """Callback to stop the processpool and reactor."""

=== modified file 'lib/lp/services/job/tests/test_runner.py'
--- lib/lp/services/job/tests/test_runner.py	2011-08-25 16:57:30 +0000
+++ lib/lp/services/job/tests/test_runner.py	2011-08-29 18:40:15 +0000
@@ -500,13 +500,6 @@
         self.x = '*' * (10 ** 6)
 
 
-class NoJobs(StaticJobSource):
-
-    done = False
-
-    jobs = []
-
-
 class TestTwistedJobRunner(ZopeTestInSubProcess, TestCaseWithFactory):
 
     layer = ZopelessDatabaseLayer
@@ -615,14 +608,6 @@
         oops = self.getOopsReport(runner, 0)
         self.assertEqual('MemoryError', oops.type)
 
-    def test_no_jobs(self):
-        logger = BufferLogger()
-        logger.setLevel(logging.INFO)
-        runner = TwistedJobRunner.runFromSource(
-            NoJobs, 'branchscanner', logger)
-        self.assertEqual(
-            (0, 0), (len(runner.completed_jobs), len(runner.incomplete_jobs)))
-
 
 class TestJobCronScript(ZopeTestInSubProcess, TestCaseWithFactory):