Merge lp:~abentley/launchpad/memlimit-sendbranchmail into lp:launchpad

[Aaron Bentley]

Memory-limit branch mail jobs.

= Summary =
Fix bug #585126: sendbranchmail with lp:~vcs-imports/linux/trunk is eating memory

== Proposed fix ==
Introduce a memory limit to branch mail jobs.

== Pre-implementation notes ==
None

== Implementation details ==
The TwistedJobRunner now supports memory limits for jobs, which means that each job runs in a separate process, and raises a MemoryError if it exceeds allowable memory use. This branch uses that mechanism for the sendbranchmail jobs.

In order to use the memory limit, we must switch to the TwistedJobRunner. In order to use the TwistedJobRunner, we need a suitable job source, so BranchMailJobSource was implemented. In order to implement BranchMailJobSource, we changed BranchJobDerived.get so that it will return instances of any given type. In order to make BranchJobDerived.get to return instances of any given type, we used RegisteredSubclass as its metaclass and provided _subclass and _register_subclass.

There were several drivebys as well:
AsyncVirtualServer can now behave as a ContextManager, meaning that the following works:
with get_ro_server():
    server_ops

Old bzr workarounds that are no longer needed were deleted.

Lots of lint was fixed

== Tests ==

bin/test -vt sendbranchmail -t branchjob

== Demo and Q/A ==
Not sure whether this is QAable.

= Launchpad lint =

Checking for conflicts and issues in changed files.

Linting changed files:
  lib/lp/translations/tests/test_translationtemplatesbuildjob.py
  cronscripts/sendbranchmail.py
  lib/lp/registry/model/packagingjob.py
  lib/lp/codehosting/vfs/transport.py
  lib/lp/code/scripts/tests/test_sendbranchmail.py
  lib/lp/code/model/branchjob.py
  lib/lp/services/utils.py

[Gavin Panella (allenap)]

Looks good.

[1]

+ @staticmethod
+ def _register_subclass(cls):
+ """Register this class with its enumeration."""
+ # This would be a classmethod, except that subclasses (e.g.
+ # TranslationPackagingJob) need to be able to override it and call
+ # into it, and there's no syntax to call a base class's version of a
+ # classmethod with the subclass as the first parameter.

I think super(BranchJobDerived, cls)._register_subclass() might work.

[2]

+class BranchMailJobSource(BaseRunnableJobSource):
+ """Source of jobs that send mail about branches."""
+
+ memory_limit = 2 * (1024 ** 3)

2GB is quite a high limit. Is that correct?

review: Approve

[Aaron Bentley (abentley)]

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

On 11-06-23 12:36 PM, Gavin Panella wrote:
> [1]
>
> + @staticmethod
> + def _register_subclass(cls):
> + """Register this class with its enumeration."""
> + # This would be a classmethod, except that subclasses (e.g.
> + # TranslationPackagingJob) need to be able to override it and call
> + # into it, and there's no syntax to call a base class's version of a
> + # classmethod with the subclass as the first parameter.
>
> I think super(BranchJobDerived, cls)._register_subclass() might work.

This doesn't work because _register_subclass is called before
BranchJobDerived is part of the global namespace (because it's called by
the metaclass).

class Bar(type):

    def __init__(cls, name, bases, dict_):
        cls.upcall()

class Foo(object):

    __metaclass__ = Bar

    @classmethod
    def upcall(cls):
        super(Foo, cls)

$ python superme.py
Traceback (most recent call last):
  File "superme.py", line 7, in <module>
    class Foo(object):
  File "superme.py", line 4, in __init__
    cls.upcall()
  File "superme.py", line 13, in upcall
    super(Foo, cls)
NameError: global name 'Foo' is not defined

> [2]
>
> +class BranchMailJobSource(BaseRunnableJobSource):
> + """Source of jobs that send mail about branches."""
> +
> + memory_limit = 2 * (1024 ** 3)
>
> 2GB is quite a high limit. Is that correct?

I'm using the limit applied to cronscripts/scan_branches.py (in 12991)
as an example. sendbranchmail runs on the same machine and uses similar
technology, including bzrlib.

The bug reports that 5GB is problematic, so this cuts it by more than half.

If you're wondering whether bzrlib really needs that much, well, it has
been quite memory-hungry historically, but I understand bzr 2.4 will be
a vast improvement.

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

iEYEARECAAYFAk4DeNAACgkQ0F+nu1YWqI2WkQCfZri8hj8jxxSp76TSrDZ1cvAL
7vEAniM+dsU6qFgVa72NuZvtr/YxM9PU
=Mzw0
-----END PGP SIGNATURE-----

=== modified file 'cronscripts/sendbranchmail.py'
--- cronscripts/sendbranchmail.py	2010-05-19 18:07:56 +0000
+++ cronscripts/sendbranchmail.py	2011-06-23 21:13:26 +0000
@@ -1,6 +1,6 @@
 #!/usr/bin/python -S
 #
-# Copyright 2009 Canonical Ltd.  This software is licensed under the
+# Copyright 2008-2011 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 # pylint: disable-msg=W0403
@@ -12,14 +12,18 @@
 
 __metaclass__ = type
 
+
+import logging
+
 import _pythonpath
-from zope.component import getUtility
 
 from canonical.config import config
-from lp.codehosting.vfs import get_ro_server
-from lp.services.job.runner import JobRunner
-from lp.code.interfaces.branchjob import (
-    IRevisionMailJobSource, IRevisionsAddedJobSource)
+from lp.services.job.runner import (
+    TwistedJobRunner,
+    )
+from lp.code.model.branchjob import (
+    BranchMailJobSource,
+    )
 from lp.services.scripts.base import LaunchpadCronScript
 from canonical.launchpad.webapp.errorlog import globalErrorUtility
 
@@ -29,15 +33,8 @@
 
     def main(self):
         globalErrorUtility.configure('sendbranchmail')
-        jobs = list(getUtility(IRevisionMailJobSource).iterReady())
-        jobs.extend(getUtility(IRevisionsAddedJobSource).iterReady())
-        runner = JobRunner(jobs, self.logger)
-        server = get_ro_server()
-        server.start_server()
-        try:
-            runner.runAll()
-        finally:
-            server.stop_server()
+        runner = TwistedJobRunner.runFromSource(
+            BranchMailJobSource, 'send-branch-mail', logging.getLogger())
         self.logger.info(
             'Ran %d RevisionMailJobs.' % len(runner.completed_jobs))
 

=== modified file 'lib/lp/code/model/branchjob.py'
--- lib/lp/code/model/branchjob.py	2011-06-20 20:07:33 +0000
+++ lib/lp/code/model/branchjob.py	2011-06-23 21:13:26 +0000
@@ -13,6 +13,7 @@
 ]
 
 import contextlib
+from itertools import chain
 import operator
 import os
 import shutil
@@ -106,7 +107,11 @@
 from lp.scripts.helpers import TransactionFreeOperation
 from lp.services.job.interfaces.job import JobStatus
 from lp.services.job.model.job import Job
-from lp.services.job.runner import BaseRunnableJob
+from lp.services.job.runner import (
+    BaseRunnableJob,
+    BaseRunnableJobSource,
+    )
+from lp.services.utils import RegisteredSubclass
 from lp.translations.interfaces.translationimportqueue import (
     ITranslationImportQueue,
     )
@@ -217,6 +222,20 @@
 
 class BranchJobDerived(BaseRunnableJob):
 
+    __metaclass__ = RegisteredSubclass
+    _subclass = {}
+
+    @staticmethod
+    def _register_subclass(cls):
+        """Register this class with its enumeration."""
+        # Why not a classmethod?  See RegisteredSubclass.__init__.
+        job_type = getattr(cls, 'class_job_type', None)
+        if job_type is not None:
+            value = cls._subclass.setdefault(job_type, cls)
+            assert value is cls, (
+                '%s already registered to %s.' % (
+                    job_type.name, value.__name__))
+
     delegates(IBranchJob)
 
     def __init__(self, branch_job):
@@ -251,19 +270,29 @@
             And(BranchJob.job_type == cls.class_job_type,
                 BranchJob.job == Job.id,
                 Job.id.is_in(Job.ready_jobs)))
-        return (cls(job) for job in jobs)
-
-    @classmethod
-    def get(cls, key):
-        """Return the instance of this class whose key is supplied.
+        return (cls._getInstance(job) for job in jobs)
+
+    @classmethod
+    def _getInstance(cls, job):
+        return cls._subclass[job.job_type](job)
+
+    @classmethod
+    def get(cls, key, desired_classes=None):
+        """Return the instance of this class (or a subclass) whose key
+        is supplied.  Calling this method on a subclass returns only values
+        for that subclass.
 
         :raises: SQLObjectNotFound
         """
-        instance = IStore(BranchJob).get(BranchJob, key)
-        if instance is None or instance.job_type != cls.class_job_type:
-            raise SQLObjectNotFound(
-                'No occurrence of %s has key %s' % (cls.__name__, key))
-        return cls(instance)
+        if desired_classes is None:
+            desired_classes = cls
+        branchjob = IStore(BranchJob).get(BranchJob, key)
+        if branchjob is not None:
+            job = cls._getInstance(branchjob)
+            if isinstance(job, cls):
+                return job
+        raise SQLObjectNotFound(
+            'No occurrence of %s has key %s' % (cls.__name__, key))
 
     def getOopsVars(self):
         """See `IRunnableJob`."""
@@ -764,6 +793,25 @@
         return outf.getvalue()
 
 
+class BranchMailJobSource(BaseRunnableJobSource):
+    """Source of jobs that send mail about branches."""
+
+    memory_limit = 2 * (1024 ** 3)
+
+    @staticmethod
+    def contextManager():
+        return get_ro_server()
+
+    @staticmethod
+    def iterReady():
+        return chain(
+            RevisionMailJob.iterReady(), RevisionsAddedJob.iterReady())
+
+    @staticmethod
+    def get(key):
+        return BranchJobDerived.get(key, (RevisionMailJob, RevisionsAddedJob))
+
+
 class RosettaUploadJob(BranchJobDerived):
     """A Job that uploads translation files to Rosetta."""
 

=== modified file 'lib/lp/code/scripts/tests/test_sendbranchmail.py'
--- lib/lp/code/scripts/tests/test_sendbranchmail.py	2010-10-26 15:47:24 +0000
+++ lib/lp/code/scripts/tests/test_sendbranchmail.py	2011-06-23 21:13:26 +0000
@@ -20,7 +20,6 @@
     RevisionMailJob,
     RevisionsAddedJob,
     )
-from lp.services.osutils import override_environ
 from lp.testing import TestCaseWithFactory
 
 
@@ -41,8 +40,7 @@
         tree.add('foo')
         # XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
         # required to generate the revision-id.
-        with override_environ(BZR_EMAIL='[e-mail]'):
-            tree.commit('Added foo.', rev_id='rev1')
+        tree.commit('Added foo.', rev_id='rev1', committer='[e-mail]')
         return branch, tree
 
     def test_sendbranchmail(self):
@@ -55,7 +53,9 @@
         retcode, stdout, stderr = run_script(
             'cronscripts/sendbranchmail.py', [])
         self.assertEqual(
-            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
+            'INFO    Creating lockfile: '
+                '/var/lock/launchpad-sendbranchmail.lock\n'
+            'INFO    Running through Twisted.\n'
             'INFO    Ran 1 RevisionMailJobs.\n', stderr)
         self.assertEqual('', stdout)
         self.assertEqual(0, retcode)
@@ -70,7 +70,8 @@
         retcode, stdout, stderr = run_script(
             'cronscripts/sendbranchmail.py', [])
         self.assertIn(
-            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n',
+            'INFO    Creating lockfile: '
+                '/var/lock/launchpad-sendbranchmail.lock\n',
             stderr)
         self.assertIn('INFO    Job resulted in OOPS:', stderr)
         self.assertIn('INFO    Ran 0 RevisionMailJobs.\n', stderr)
@@ -82,17 +83,16 @@
         self.useBzrBranches()
         branch, tree = self.createBranch()
         tree.bzrdir.root_transport.put_bytes('foo', 'baz')
-        # XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
-        # required to generate the revision-id.
-        with override_environ(BZR_EMAIL='[e-mail]'):
-            tree.commit('Added foo.', rev_id='rev2')
+        tree.commit('Added foo.', rev_id='rev2', committer='[e-mail]')
         RevisionsAddedJob.create(
             branch, 'rev1', 'rev2', '[e-mail]')
         transaction.commit()
         retcode, stdout, stderr = run_script(
             'cronscripts/sendbranchmail.py', [])
         self.assertEqual(
-            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
+            'INFO    Creating lockfile:'
+                ' /var/lock/launchpad-sendbranchmail.lock\n'
+            'INFO    Running through Twisted.\n'
             'INFO    Ran 1 RevisionMailJobs.\n',
             stderr)
         self.assertEqual('', stdout)

=== modified file 'lib/lp/codehosting/vfs/transport.py'
--- lib/lp/codehosting/vfs/transport.py	2011-03-29 03:24:05 +0000
+++ lib/lp/codehosting/vfs/transport.py	2011-06-23 21:13:26 +0000
@@ -102,7 +102,7 @@
         """Return the absolute, escaped path to `relpath` without the schema.
         """
         return urlutils.joinpath(
-            self.base[len(self.server.get_url())-1:], relpath)
+            self.base[len(self.server.get_url()) - 1:], relpath)
 
     def _getUnderylingTransportAndPath(self, relpath):
         """Return the underlying transport and path for `relpath`."""
@@ -178,6 +178,7 @@
 
     def iter_files_recursive(self):
         deferred = self._getUnderylingTransportAndPath('.')
+
         @no_traceback_failures
         def iter_files((transport, path)):
             return transport.clone(path).iter_files_recursive()
@@ -186,6 +187,7 @@
 
     def listable(self):
         deferred = self._getUnderylingTransportAndPath('.')
+
         @no_traceback_failures
         def listable((transport, path)):
             return transport.listable()
@@ -427,3 +429,10 @@
             return
         self._is_started = False
         unregister_transport(self.get_url(), self._transportFactory)
+
+    def __enter__(self):
+        self.start_server()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.stop_server()

=== modified file 'lib/lp/registry/model/packagingjob.py'
--- lib/lp/registry/model/packagingjob.py	2011-03-23 16:28:51 +0000
+++ lib/lp/registry/model/packagingjob.py	2011-06-23 21:13:26 +0000
@@ -35,6 +35,7 @@
     JobStatus,
     )
 from lp.services.job.model.job import Job
+from lp.services.utils import RegisteredSubclass
 
 
 class PackagingJobType(DBEnumeratedType):
@@ -96,13 +97,6 @@
         self.productseries = productseries
 
 
-class RegisteredSubclass(type):
-    """Metaclass for when subclasses should be registered."""
-
-    def __init__(cls, name, bases, dict_):
-        cls._register_subclass(cls)
-
-
 class PackagingJobDerived:
     """Base class for specialized Packaging Job types."""
 
@@ -120,10 +114,7 @@
     @staticmethod
     def _register_subclass(cls):
         """Register this class with its enumeration."""
-        # This would be a classmethod, except that subclasses (e.g.
-        # TranslationPackagingJob) need to be able to override it and call
-        # into it, and there's no syntax to call a base class's version of a
-        # classmethod with the subclass as the first parameter.
+        # Why not a classmethod?  See RegisteredSubclass.__init__.
         job_type = getattr(cls, 'class_job_type', None)
         if job_type is not None:
             value = cls._subclass.setdefault(job_type, cls)

=== modified file 'lib/lp/services/utils.py'
--- lib/lp/services/utils.py	2011-04-01 16:30:57 +0000
+++ lib/lp/services/utils.py	2011-06-23 21:13:26 +0000
@@ -18,6 +18,7 @@
     'file_exists',
     'iter_list_chunks',
     'iter_split',
+    'RegisteredSubclass',
     'run_capturing_output',
     'synchronize',
     'text_delta',
@@ -129,7 +130,7 @@
     I'm amazed this isn't in itertools (mwhudson).
     """
     for i in range(0, len(a_list), size):
-        yield a_list[i:i+size]
+        yield a_list[i:i + size]
 
 
 def synchronize(source, target, add, remove):
@@ -239,7 +240,7 @@
     then reassemble.
     """
     # Make sure there is at least one newline so the split works.
-    first, rest = (s+'\n').split('\n', 1)
+    first, rest = (s + '\n').split('\n', 1)
     return (first + '\n' + dedent(rest)).strip()
 
 
@@ -285,3 +286,20 @@
     variables, and helps to avoid typos.
     """
     sys._getframe(1).f_locals["__traceback_info__"] = info
+
+
+class RegisteredSubclass(type):
+    """Metaclass for when subclasses should be registered."""
+
+    def __init__(cls, name, bases, dict_):
+        # _register_subclass must be a static method to permit upcalls.
+        #
+        # We cannot use super(Class, cls) to do the upcalls, because Class
+        # isn't fully defined yet.  (Remember, we're calling this from a
+        # metaclass.)
+        #
+        # Without using super, a classmethod that overrides another
+        # classmethod has no reasonable way to call the overridden version AND
+        # provide its class as first parameter (i.e. "cls").  Therefore, we
+        # must use a static method.
+        cls._register_subclass(cls)

=== modified file 'lib/lp/translations/model/translationpackagingjob.py'
--- lib/lp/translations/model/translationpackagingjob.py	2011-05-12 20:21:58 +0000
+++ lib/lp/translations/model/translationpackagingjob.py	2011-06-23 21:13:26 +0000
@@ -52,6 +52,7 @@
 
     @staticmethod
     def _register_subclass(cls):
+        # Why not a classmethod?  See RegisteredSubclass.__init__.
         PackagingJobDerived._register_subclass(cls)
         job_type = getattr(cls, 'class_job_type', None)
         if job_type is not None:

=== modified file 'lib/lp/translations/tests/test_translationtemplatesbuildjob.py'
--- lib/lp/translations/tests/test_translationtemplatesbuildjob.py	2011-05-27 21:12:25 +0000
+++ lib/lp/translations/tests/test_translationtemplatesbuildjob.py	2011-06-23 21:13:26 +0000
@@ -156,6 +156,8 @@
     behavior by setting an attribute of the class (not an object!) at
     the beginning of every test.
     """
+    class_job_type = None
+
     # Fake _hasPotteryCompatibleSetup, and if so, make it give what
     # answer?
     fake_pottery_compatibility = None