launchpad-reviewers team mailing list archive

[Aaron Bentley <aaron@xxxxxxxxxxxxx>]

Aaron Bentley has proposed merging lp:~abentley/launchpad/simplify-twisted-runner-3 into lp:launchpad with lp:~abentley/launchpad/simplify-twisted-runner-2 as a prerequisite.

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)
Related bugs:
  Bug #833888 in Launchpad itself: "Twisted job runner is more complex than needed"
  https://bugs.launchpad.net/launchpad/+bug/833888

For more details, see:
https://code.launchpad.net/~abentley/launchpad/simplify-twisted-runner-3/+merge/74868

= Summary =
Fix issues invoking TwistedJobRunner

== Proposed fix ==
Inject the current sys.path into subprocesses as PYTHONPATH

== Pre-implementation notes ==
None

== Implementation details ==
This builds on previous work toward simplifying the TwistedJobRunner.

It fixes issues finding and launching the JobRunnerProcess class by injecting the current sys.path into subprocesses as PYTHONPATH.

It also simplifies the code further by using the inclineCallbacks decorator.

It also improves handling of process startup failures by handling the case when result is None.

It also outputs 'No jobs to run.' if no jobs were found, to clarify output.

It also raises 'Could not acquire lease...' to an INFO message, so that it's at the same level as 'Running ...' messages.

== Tests ==
bin/test test_runner

== Demo and Q/A ==
Create a branch and run cronscripts/scan_branches.py on qastaging.  It should complete successfully.


= 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
-- 
https://code.launchpad.net/~abentley/launchpad/simplify-twisted-runner-3/+merge/74868
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~abentley/launchpad/simplify-twisted-runner-3 into lp:launchpad.

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2011-09-09 20:05:32 +0000
+++ lib/lp/services/job/runner.py	2011-09-09 20:05:36 +0000
@@ -43,10 +43,14 @@
     reactor,
     )
 from twisted.internet.defer import (
+    inlineCallbacks,
     succeed,
     )
 from twisted.protocols import amp
-from twisted.python import log
+from twisted.python import (
+    failure,
+    log,
+    )
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
@@ -183,7 +187,7 @@
         try:
             job.acquireLease()
         except LeaseHeld:
-            self.logger.debug(
+            self.logger.info(
                 'Could not acquire lease for %s' % self.job_str(job))
             self.incomplete_jobs.append(job)
             return False
@@ -406,11 +410,10 @@
 
     def __init__(self, job_source, dbuser, logger=None, error_utility=None):
         env = {'PATH': os.environ['PATH']}
-        for name in ('PYTHONPATH', 'LPCONFIG'):
-            if name in os.environ:
-                env[name] = os.environ[name]
-        starter = main.ProcessStarter(
-            packages=('_pythonpath', 'twisted', 'ampoule'), env=env)
+        if 'LPCONFIG' in os.environ:
+            env['LPCONFIG'] = os.environ['LPCONFIG']
+        env['PYTHONPATH'] = os.pathsep.join(sys.path)
+        starter = main.ProcessStarter(env=env)
         super(TwistedJobRunner, self).__init__(logger, error_utility)
         self.job_source = job_source
         self.import_name = '%s.%s' % (
@@ -441,6 +444,10 @@
             RunJobCommand, job_id=job_id, _deadline=deadline)
 
         def update(response):
+            if response is None:
+                self.incomplete_jobs.append(job)
+                self.logger.debug('No response for %r', job)
+                return
             if response['success']:
                 self.completed_jobs.append(job)
                 self.logger.debug('Finished %r', job)
@@ -475,18 +482,20 @@
             oops = self._doOops(job, sys.exc_info())
             self._logOopsId(oops['id'])
 
+    @inlineCallbacks
     def runAll(self):
         """Run all ready jobs."""
         self.pool.start()
         try:
-            jobs = list(self.job_source.iterReady())
-            if len(jobs) == 0:
+            try:
+                job = None
+                for job in self.job_source.iterReady():
+                    yield self.runJobInSubprocess(job)
+                if job is None:
+                    self.logger.info('No jobs to run.')
                 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.failed(failure.Failure())
         except:
             self.terminated()
             raise