Merge lp:~abentley/launchpad/ampoule-0.1.1 into lp:launchpad

[Aaron Bentley]

[Aaron Bentley (abentley)]

= Summary =
Update to Ampoule 0.2.0.

== Proposed fix ==
This branch updates ampoule and also adds a version for pyOpenSSL, which is a
new Ampoule dependency. This new version of ampoule has many changes that I
suggested, to push as much code upstream as I thought was architecturally
sound.

== Pre-implementation notes ==
None

== Implementation details ==
Instead of having a separate AMPChild subclass for every JobSource, a new
ampoule feature allows us to parameterize the JobRunnerProcess. We supply a
full path to the JobSource. In JobRunnerProcess, we import this and use it. I
used 'exec' for this, because the import module looked crazy complicated.

Instead of supplying our own BOOTSTRAP code in order to provide a bit more
startup code, a new ampoule feature lets us simply implement the context
manager protocol in JobRunnerProcess.

Instead of subclassing pool.ProcessPool to use SIGHUP, ampoule now lets us
parameterize the timout_signal.

Instead of converting our leases, which are datetimes, into timeout durations,
a new ampoule feature lets us supply them as deadlines.

job_class was renamed to job_source where this was a better description of its
usage.

== Tests ==
bin/test -vt test_runner -t update_preview_diffs

== Demo and Q/A ==
No visible change.

= Launchpad lint =

Checking for conflicts. and issues in doctests and templates.
Running jslint, xmllint, pyflakes, and pylint.
Using normal rules.

Linting changed files:
  lib/lp/services/job/tests/test_runner.py
  lib/lp/services/job/runner.py
  versions.cfg
  lib/lp/code/model/branchmergeproposaljob.py

== Pyflakes notices ==

lib/lp/services/job/runner.py
    240: undefined name 'job_source'

^^^ This is because of the exec.

== Pylint notices ==

lib/lp/services/job/runner.py
    35: [F0401] Unable to import 'lazr.delegates' (No module named delegates)
    239: [W0122, JobRunnerProcess.__init__] Use of the exec statement
    240: [E0602, JobRunnerProcess.__init__] Undefined variable 'job_source'

^^^ This is because of the exec.

lib/lp/code/model/branchmergeproposaljob.py
    22: [F0401] Unable to import 'lazr.delegates' (No module named delegates)
    23: [F0401] Unable to import 'lazr.enum' (No module named enum)

[Aaron Bentley (abentley)]

After submitting this, I remembered about __import__, so I've replaced exec with __import__.

[Gavin Panella (allenap)]

Hi Aaron,

I can't pretend to fully understand all of this, but the changes look
good. I have one tiny comment.

Gavin.

> === modified file 'lib/lp/services/job/runner.py'
> --- lib/lp/services/job/runner.py 2010-01-16 10:20:46 +0000
> +++ lib/lp/services/job/runner.py 2010-02-03 16:02:29 +0000
> @@ -15,6 +15,7 @@
> ]
>
>
> +from calendar import timegm
> import contextlib
> import logging
> import os
> @@ -22,9 +23,8 @@
> import sys
>
> from ampoule import child, pool, main
> -from twisted.internet import defer, error, reactor, stdio
> +from twisted.internet import defer, reactor
> from twisted.protocols import amp
> -from twisted.python import log, reflect
>
> from zope.component import getUtility
> from zope.security.proxy import removeSecurityProxy
> @@ -231,9 +231,25 @@
> class JobRunnerProcess(child.AMPChild):
> """Base class for processes that run jobs."""
>
> - def __init__(self):
> + def __init__(self, job_source_name):
> child.AMPChild.__init__(self)
> - self.context_manager = self.job_class.contextManager()
> + segments = job_source_name.split('.')
> + module = '.'.join(segments[:-1])
> + name = segments[-1]

Could you simplify this to:

  module, name = job_source_name.rsplit('.', 1)

?

Ah, I guess it wouldn't work if there are no "."s in job_source_name.

review: Approve

[Aaron Bentley (abentley)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

Gavin Panella wrote:
>> - def __init__(self):
>> + def __init__(self, job_source_name):
>> child.AMPChild.__init__(self)
>> - self.context_manager = self.job_class.contextManager()
>> + segments = job_source_name.split('.')
>> + module = '.'.join(segments[:-1])
>> + name = segments[-1]
>
> Could you simplify this to:
>
> module, name = job_source_name.rsplit('.', 1)

Done. Thanks for the tip.

> Ah, I guess it wouldn't work if there are no "."s in job_source_name.

If there are no dots, the import will fail anyhow. I can't imagine any
time when it would be sensible to supply a job_source_name with no dots
in it, though.

Aaron
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (GNU/Linux)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org

iEYEARECAAYFAktppBUACgkQ0F+nu1YWqI2AyACeM7qOYIfkj+04WACDGO5Mt+QA
QUQAni+O5BLOSNU31EOpbUauFrVgX3JF
=ieiV
-----END PGP SIGNATURE-----

=== modified file 'lib/lp/bugs/windmill/tests/test_bug_inline_subscriber.py'
--- lib/lp/bugs/windmill/tests/test_bug_inline_subscriber.py	2010-02-01 18:37:00 +0000
+++ lib/lp/bugs/windmill/tests/test_bug_inline_subscriber.py	2010-02-03 22:24:19 +0000
@@ -18,7 +18,8 @@
     layer = BugsWindmillLayer
     suite_name = 'Inline bug page subscribers test'
 
-    def test_inline_subscriber(self):
+    def DISABLED_test_inline_subscriber(self):
+        # This test fails intermittently.  See bug #516781.
         """Test inline subscribing on bugs pages.
 
         This test makes sure that subscribing and unsubscribing
 
=== modified file 'lib/lp/code/model/branchmergeproposaljob.py'
--- lib/lp/code/model/branchmergeproposaljob.py	2010-01-11 20:07:17 +0000
+++ lib/lp/code/model/branchmergeproposaljob.py	2010-02-03 22:24:19 +0000
@@ -51,7 +51,7 @@
 from lp.codehosting.vfs import get_multi_server, get_scanner_server
 from lp.services.job.model.job import Job
 from lp.services.job.interfaces.job import IRunnableJob
-from lp.services.job.runner import BaseRunnableJob, JobRunnerProcess
+from lp.services.job.runner import BaseRunnableJob
 
 
 class BranchMergeProposalJobType(DBEnumeratedType):
@@ -283,14 +283,6 @@
         self.branch_merge_proposal.preview_diff = preview
 
 
-class UpdatePreviewDiffProcess(JobRunnerProcess):
-    """A process that runs UpdatePreviewDiffJobs"""
-    job_class = UpdatePreviewDiffJob
-
-
-UpdatePreviewDiffJob.amp = UpdatePreviewDiffProcess
-
-
 class CreateMergeProposalJob(BaseRunnableJob):
     """See `ICreateMergeProposalJob` and `ICreateMergeProposalJobSource`."""
 
 
=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2010-02-03 11:17:12 +0000
+++ lib/lp/services/job/runner.py	2010-02-03 22:24:18 +0000
@@ -15,6 +15,7 @@
     ]
 
 
+from calendar import timegm
 import contextlib
 import logging
 import os
@@ -22,9 +23,8 @@
 import sys
 
 from ampoule import child, pool, main
-from twisted.internet import defer, error, reactor, stdio
+from twisted.internet import defer, reactor
 from twisted.protocols import amp
-from twisted.python import log, reflect
 
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
@@ -242,9 +242,23 @@
 class JobRunnerProcess(child.AMPChild):
     """Base class for processes that run jobs."""
 
-    def __init__(self):
+    def __init__(self, job_source_name):
         child.AMPChild.__init__(self)
-        self.context_manager = self.job_class.contextManager()
+        module, name = job_source_name.rsplit('.', 1)
+        source_module = __import__(module, fromlist=[name])
+        self.job_source = getattr(source_module, name)
+        self.context_manager = self.job_source.contextManager()
+
+    @staticmethod
+    def __enter__():
+        def handler(signum, frame):
+            raise TimeoutError
+        scripts.execute_zcml_for_scripts(use_web_security=False)
+        signal(SIGHUP, handler)
+
+    @staticmethod
+    def __exit__(exc_type, exc_val, exc_tb):
+        pass
 
     def makeConnection(self, transport):
         """The Job context is entered on connect."""
@@ -258,9 +272,9 @@
 
     @RunJobCommand.responder
     def runJobCommand(self, job_id):
-        """Run a job of this job_class according to its job id."""
+        """Run a job from this job_source according to its job id."""
         runner = BaseJobRunner()
-        job = self.job_class.get(job_id)
+        job = self.job_source.get(job_id)
         oops = runner.runJobHandleError(job)
         if oops is None:
             oops_id = ''
@@ -269,28 +283,22 @@
         return {'success': len(runner.completed_jobs), 'oops_id': oops_id}
 
 
-class HUPProcessPool(pool.ProcessPool):
-    """A ProcessPool that kills with HUP."""
-
-    def _handleTimeout(self, child):
-        try:
-            child.transport.signalProcess(SIGHUP)
-        except error.ProcessExitedAlready:
-            pass
-
-
 class TwistedJobRunner(BaseJobRunner):
     """Run Jobs via twisted."""
 
-    def __init__(self, job_source, job_amp, logger=None, error_utility=None):
+    def __init__(self, job_source, logger=None, error_utility=None):
         starter = main.ProcessStarter(
-            bootstrap=BOOTSTRAP, packages=('twisted', 'ampoule'),
+            packages=('twisted', 'ampoule'),
             env={'PYTHONPATH': os.environ['PYTHONPATH'],
             'PATH': os.environ['PATH'],
             'LPCONFIG': os.environ['LPCONFIG']})
         super(TwistedJobRunner, self).__init__(logger, error_utility)
         self.job_source = job_source
-        self.pool = HUPProcessPool(job_amp, starter=starter, min=0)
+        import_name = '%s.%s' % (
+            removeSecurityProxy(job_source).__module__, job_source.__name__)
+        self.pool = pool.ProcessPool(
+            JobRunnerProcess, ampChildArgs=[import_name], starter=starter,
+            min=0, timeout_signal=SIGHUP)
 
     def runJobInSubprocess(self, job):
         """Run the job_class with the specified id in the process pool.
@@ -304,12 +312,9 @@
             self.incomplete_jobs.append(job)
             return
         job_id = job.id
-        timeout = job.getTimeout()
-        # work around ampoule bug
-        if timeout == 0:
-            timeout = 0.0000000000001
+        deadline = timegm(job.lease_expires.timetuple())
         deferred = self.pool.doWork(
-            RunJobCommand, job_id = job_id, _timeout=timeout)
+            RunJobCommand, job_id = job_id, _deadline=deadline)
         def update(response):
             if response['success']:
                 self.completed_jobs.append(job)
@@ -362,8 +367,7 @@
     def runFromSource(cls, job_source, logger, error_utility=None):
         """Run all ready jobs provided by the specified source."""
         logger.info("Running through Twisted.")
-        runner = cls(job_source, removeSecurityProxy(job_source).amp, logger,
-                     error_utility)
+        runner = cls(job_source, logger, error_utility)
         reactor.callWhenRunning(runner.runAll)
         handler = getsignal(SIGCHLD)
         try:
@@ -396,22 +400,3 @@
 
     def __init__(self):
         Exception.__init__(self, "Job ran too long.")
-
-
-BOOTSTRAP = """\
-import sys
-from twisted.application import reactors
-reactors.installReactor(sys.argv[-2])
-from lp.services.job.runner import bootstrap
-bootstrap(sys.argv[-1])
-"""
-
-def bootstrap(ampChildPath):
-    def handler(signum, frame):
-        raise TimeoutError
-    signal(SIGHUP, handler)
-    log.startLogging(sys.stderr)
-    ampChild = reflect.namedAny(ampChildPath)
-    stdio.StandardIO(ampChild(), 3, 4)
-    scripts.execute_zcml_for_scripts(use_web_security=False)
-    reactor.run()
 
=== modified file 'lib/lp/services/job/tests/test_runner.py'
--- lib/lp/services/job/tests/test_runner.py	2010-01-16 10:20:46 +0000
+++ lib/lp/services/job/tests/test_runner.py	2010-02-03 22:24:18 +0000
@@ -18,7 +18,7 @@
 
 from lp.testing.mail_helpers import pop_notifications
 from lp.services.job.runner import (
-    JobRunner, BaseRunnableJob, JobRunnerProcess, TwistedJobRunner
+    JobRunner, BaseRunnableJob, TwistedJobRunner
 )
 from lp.services.job.interfaces.job import JobStatus, IRunnableJob
 from lp.services.job.model.job import Job
@@ -283,14 +283,6 @@
             sleep(30)
 
 
-class StuckJobProcess(JobRunnerProcess):
-
-    job_class = StuckJob
-
-
-StuckJob.amp = StuckJobProcess
-
-
 class ListLogger:
 
     def __init__(self):
 
=== modified file 'versions.cfg'
--- versions.cfg	2010-02-03 10:57:21 +0000
+++ versions.cfg	2010-02-03 22:24:18 +0000
@@ -4,11 +4,7 @@
 [versions]
 # Alphabetical, case-insensitive, please! :-)
 
-# from -r 3:lp:~launchpad/ampoule/launchpad-tweaked
-# To reproduce:
-# bzr export ampoule-0.1.0-lp-1.tar.gz lp:~launchpad/ampoule/launchpad-tweaked\
-#     -r 3
-ampoule = 0.1.0-lp-1
+ampoule = 0.2.0
 # Non-released bzr version from bzr+ssh://bazaar.launchpad.net/~mwhudson/bzr/2.1.0b4-lp2
 bzr = 2.1b4-lp2
 chameleon.core = 1.0b35
@@ -50,6 +46,7 @@
 PasteDeploy = 1.3.3
 pyasn1 = 0.0.9a
 pycrypto = 2.0.1
+pyOpenSSL = 0.10
 python-memcached = 1.45
 python-openid = 2.2.1
 pytz = 2009l