Merge lp:~abentley/launchpad/efficient-universal-source into lp:launchpad

[Aaron Bentley]

Optimize UniversalJobSource

= Summary =
Make UniversalJobSource more efficient

== Proposed fix ==
Provide database user name, database class module and database class name as part of the job_id.

== Pre-implementation notes ==
None

== LOC Rationale ==
Removes code

== Implementation details ==
Provide this database username as part of the "ujob_id". This means that the correct database user can be set immediately, instead of first doing a lookup of the job to determine the database user. It also means that clearStore can be folded into the get method, since it's only done once.

Provide the name and module name of the database job class (one which refers to a Job, e.g. BranchJob) as part of the ujob_id. Combined with providing the dbuser above, this means that getUserAndBaseJob is obsolete. The database table cannot be used for this purpose, because AFAICT Storm permits multiple classes to use a given table.

Extract rawGet from _getDerived

Update getUserAndBaseJob test to use rawGet instead.

Implement getDBClass methods on all derived job classes. This is needed because a few job types don't have a 'context' member which is a database class.

== Tests ==
bin/test --layer=CeleryJobLayer

== Demo and Q/A ==
None

= Launchpad lint =

Checking for conflicts and issues in changed files.

Linting changed files:
  lib/lp/services/job/runner.py
  lib/lp/registry/tests/test_membership_notification_job.py
  lib/lp/translations/model/pofilestatsjob.py
  lib/lp/soyuz/model/packagecopyjob.py
  lib/lp/code/model/tests/test_branchjob.py
  lib/lp/translations/model/translationsharingjob.py
  lib/lp/soyuz/tests/test_packagecopyjob.py
  lib/lp/services/job/tests/test_job.py
  lib/lp/answers/model/questionjob.py
  lib/lp/registry/model/persontransferjob.py
  lib/lp/registry/tests/test_person_merge_job.py
  lib/lp/services/job/model/job.py
  lib/lp/answers/tests/test_questionjob.py

[j.c.sackett (jcsackett)]

This looks good; thanks for removing code.

review: Approve

=== modified file 'lib/lp/code/model/tests/test_branchjob.py'
--- lib/lp/code/model/tests/test_branchjob.py	2012-04-13 18:31:35 +0000
+++ lib/lp/code/model/tests/test_branchjob.py	2012-04-26 20:27:38 +0000
@@ -1247,11 +1247,11 @@
         commit.writeFile('foo.pot', 'gibberish')
         with person_logged_in(db_branch.owner):
             # wait for branch scan
-            with block_on_job():
+            with block_on_job(self):
                 commit.commit('message')
                 transaction.commit()
         series = self.factory.makeProductSeries(branch=db_branch)
-        with block_on_job():
+        with block_on_job(self):
             RosettaUploadJob.create(
                 commit.db_branch, NULL_REVISION,
                 force_translations_upload=True)

=== modified file 'lib/lp/services/job/model/job.py'
--- lib/lp/services/job/model/job.py	2012-04-26 20:27:32 +0000
+++ lib/lp/services/job/model/job.py	2012-04-26 20:27:38 +0000
@@ -38,7 +38,7 @@
 from zope.component import getUtility
 from zope.interface import implements
 
-from lp.services.config import config, dbconfig
+from lp.services.config import dbconfig
 from lp.services.database import bulk
 from lp.services.database.constants import UTC_NOW
 from lp.services.database.datetimecol import UtcDateTimeCol
@@ -258,59 +258,24 @@
     needs_init = True
 
     @staticmethod
-    def _getDerived(job_id, base_class):
-        store = IStore(base_class)
-        base_job = store.find(base_class, base_class.job == job_id).one()
-        if base_job is None:
-            return None, None, None
-        return base_job.makeDerived(), base_job.__class__, store
-
-    @classmethod
-    def getUserAndBaseJob(cls, job_id):
-        """Return the derived branch job associated with the job id."""
-        # Avoid circular imports.
-        from lp.bugs.model.apportjob import ApportJob
-        from lp.code.model.branchjob import (
-            BranchJob,
-            )
-        from lp.code.model.branchmergeproposaljob import (
-            BranchMergeProposalJob,
-            )
-        from lp.registry.model.persontransferjob import PersonTransferJob
-        from lp.answers.model.questionjob import QuestionJob
-        from lp.soyuz.model.distributionjob import DistributionJob
-        from lp.soyuz.model.packagecopyjob import PackageCopyJob
-        from lp.translations.model.pofilestatsjob import POFileStatsJob
-        from lp.translations.model.translationsharingjob import (
-            TranslationSharingJob,
-        )
-        dbconfig.override(
-            dbuser=config.launchpad.dbuser, isolation_level='read_committed')
-
-        for baseclass in [
-            ApportJob, BranchJob, BranchMergeProposalJob, DistributionJob,
-            PackageCopyJob, PersonTransferJob, POFileStatsJob, QuestionJob,
-            TranslationSharingJob,
-            ]:
-            derived, base_class, store = cls._getDerived(job_id, baseclass)
-            if derived is not None:
-                cls.clearStore(store)
-                return derived.config.dbuser, base_class
-        raise ValueError('No Job with job=%s.' % job_id)
-
-    @staticmethod
-    def clearStore(store):
-        transaction.abort()
-        getUtility(IZStorm).remove(store)
-        store.close()
-
-    @classmethod
-    def get(cls, job_id):
-        transaction.abort()
+    def rawGet(job_id, module_name, class_name):
+        bc_module = __import__(module_name, fromlist=[class_name])
+        db_class = getattr(bc_module, class_name)
+        store = IStore(db_class)
+        db_job = store.find(db_class, db_class.job == job_id).one()
+        if db_job is None:
+            return None
+        return db_job.makeDerived()
+
+    @classmethod
+    def get(cls, ujob_id):
         if cls.needs_init:
+            transaction.abort()
             scripts.execute_zcml_for_scripts(use_web_security=False)
             cls.needs_init = False
-        cls.clearStore(IStore(Job))
-        dbuser, base_class = cls.getUserAndBaseJob(job_id)
-        dbconfig.override(dbuser=dbuser, isolation_level='read_committed')
-        return cls._getDerived(job_id, base_class)[0]
+        transaction.abort()
+        store = IStore(Job)
+        getUtility(IZStorm).remove(store)
+        store.close()
+        dbconfig.override(dbuser=ujob_id[3], isolation_level='read_committed')
+        return cls.rawGet(*ujob_id[:3])

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2012-04-19 20:15:26 +0000
+++ lib/lp/services/job/runner.py	2012-04-26 20:27:38 +0000
@@ -200,7 +200,14 @@
             cls = CeleryRunJobIgnoreResult
         else:
             cls = CeleryRunJob
-        return cls.apply_async((self.job_id,), queue=self.task_queue)
+        db_class = self.getDBClass()
+        ujob_id = (
+            self.job_id, db_class.__module__, db_class.__name__,
+            self.config.dbuser)
+        return cls.apply_async((ujob_id,), queue=self.task_queue)
+
+    def getDBClass(self):
+        return self.context.__class__
 
     def celeryCommitHook(self, succeeded):
         """Hook function to call when a commit completes."""

=== modified file 'lib/lp/services/job/tests/test_job.py'
--- lib/lp/services/job/tests/test_job.py	2012-04-24 15:04:25 +0000
+++ lib/lp/services/job/tests/test_job.py	2012-04-26 20:27:38 +0000
@@ -9,13 +9,12 @@
 import pytz
 from lazr.jobrunner.jobrunner import LeaseHeld
 from storm.locals import Store
+from testtools.matchers import Equals
 import transaction
 
 from lp.code.model.branchmergeproposaljob import (
-    BranchMergeProposalJob,
     CodeReviewCommentEmailJob,
     )
-from lp.services.config import config
 from lp.services.database.constants import UTC_NOW
 from lp.services.database.lpstorm import IStore
 from lp.services.job.interfaces.job import (
@@ -29,10 +28,12 @@
     )
 from lp.services.webapp.testing import verifyObject
 from lp.testing import (
+    StormStatementRecorder,
     TestCase,
     TestCaseWithFactory,
     )
 from lp.testing.layers import ZopelessDatabaseLayer
+from lp.testing.matchers import HasQueryCount
 
 
 class TestJob(TestCaseWithFactory):
@@ -461,9 +462,14 @@
 
     layer = ZopelessDatabaseLayer
 
-    def test_getUserAndBaseJob_with_merge_proposal_job(self):
+    def test_rawGet_with_merge_proposal_job(self):
         comment = self.factory.makeCodeReviewComment()
         job = CodeReviewCommentEmailJob.create(comment)
-        dbuser, base_class = UniversalJobSource.getUserAndBaseJob(job.job_id)
-        self.assertEqual(dbuser, config.merge_proposal_jobs.dbuser)
-        self.assertEqual(base_class, BranchMergeProposalJob)
+        job_id = job.job_id
+        transaction.commit()
+        with StormStatementRecorder() as recorder:
+            got_job = UniversalJobSource.rawGet(
+                job_id, 'lp.code.model.branchmergeproposaljob',
+                 'BranchMergeProposalJob')
+        self.assertThat(recorder, HasQueryCount(Equals(1)))
+        self.assertEqual(got_job, job)

=== modified file 'lib/lp/translations/model/pofilestatsjob.py'
--- lib/lp/translations/model/pofilestatsjob.py	2012-04-25 14:15:32 +0000
+++ lib/lp/translations/model/pofilestatsjob.py	2012-04-26 20:27:38 +0000
@@ -115,6 +115,9 @@
         """
         return self
 
+    def getDBClass(self):
+        return self.__class__
+
 
 def schedule(pofile):
     """Schedule a job to update a POFile's stats."""

=== modified file 'lib/lp/translations/model/translationsharingjob.py'
--- lib/lp/translations/model/translationsharingjob.py	2012-04-24 18:26:06 +0000
+++ lib/lp/translations/model/translationsharingjob.py	2012-04-26 20:27:38 +0000
@@ -121,6 +121,9 @@
 
     delegates(ITranslationSharingJob, 'job')
 
+    def getDBClass(self):
+        return TranslationSharingJob
+
     _event_types = {}
 
     @property