Merge lp:~allenap/launchpad/early-batching-bulk-methods-bug-509223 into lp:launchpad

[Gavin Panella]

New methods for IBugWatchSet - bulkSetStatus() and bulkAddActivity() - which will be used to improve performance in checkwatches.

This branch adds two new methods to BugWatchSet: bulkSetStatus() and bulkAddActivity().

In checkwatches it's often necessary to update many 100s (maybe 1000s) of bug watches at a time, especially when the remote bug tracker is down. Currently checkwatches iterates through the materialized bug watch objects (from Storm), updating their attributes and calling addActivity(), before committing. This is horrendously slow because (a) these bug watch objects may need reloading because they were materialized in a separate transaction and reloads are done one at a time, (b) they will be saved one at a time, and (c) the activity is added one row at a time. For even a modest 100 bug watches this will result in 100 SELECT, 100 UPDATE and 100 INSERT statements. This branch reduces this to 1 UPDATE and 1 INSERT statement.

These methods were conceived with checkwatches in mind, but have not been stitched in yet. Graham is doing some major refactoring in the checkwatches packages so I'm deferring that work until later.

[Graham Binns (gmb)]

<gmb> allenap, I think bulkSetStatus() is a little confusing, because it's called ...SetStatus() but it in fact mutates last_error_type. However, given that we should rename that at some point I'd settle for an XXX to the effect of "yes, this sounds confusing, it isn't"
 (and a bug)
<gmb> allenap, Other than that, the branch looks brilliant. r=me.

review: Approve (code)

=== modified file 'lib/lp/bugs/interfaces/bugwatch.py'
--- lib/lp/bugs/interfaces/bugwatch.py	2010-03-23 12:55:05 +0000
+++ lib/lp/bugs/interfaces/bugwatch.py	2010-04-27 08:44:29 +0000
@@ -288,6 +288,37 @@
         :type bug_watch_ids: An iterable of `int`s, or `None`.
         """
 
+    # XXX: GavinPanella bug=570277 2010-04-26: In bulkSetError() the
+    # last_error_type argument accepts the same values as the result
+    # argument to bulkAddActivity(). Using different terms for
+    # essentially the same thing is confusing.
+
+    def bulkSetError(bug_watches, last_error_type=None):
+        """Efficiently update the status of the given bug watches.
+
+        Sets the `last_error_type` field as instructed, updates
+        `lastchecked` to now and resets `next_check` to None, all in
+        the most efficient way possible.
+
+        :param bug_watches: An iterable of `IBugWatch` objects or
+            primary keys for the same.
+        :param last_error_type: A member of `BugWatchActivityStatus`
+            or None.
+        """
+
+    def bulkAddActivity(bug_watches, result=None, message=None, oops_id=None):
+        """Efficiently add activity for the given bug watches.
+
+        Add `BugWatchActivity` records for the given bug watches in
+        the most efficient way possible.
+
+        :param bug_watches: An iterable of `IBugWatch` objects or
+            primary keys for the same.
+        :param result: See `IBugWatch.addActivity`.
+        :param message: See `IBugWatch.addActivity`.
+        :param oops_id: See `IBugWatch.addActivity`.
+        """
+
 
 class NoBugTrackerFound(Exception):
     """No bug tracker with the base_url is registered in Launchpad."""
 
=== modified file 'lib/lp/bugs/model/bugwatch.py'
--- lib/lp/bugs/model/bugwatch.py	2010-04-09 15:04:19 +0000
+++ lib/lp/bugs/model/bugwatch.py	2010-04-27 08:44:29 +0000
@@ -33,7 +33,7 @@
 from canonical.database.constants import UTC_NOW
 from canonical.database.datetimecol import UtcDateTimeCol
 from canonical.database.enumcol import EnumCol
-from canonical.database.sqlbase import SQLBase
+from canonical.database.sqlbase import SQLBase, sqlvalues
 from canonical.launchpad.database.message import Message
 from canonical.launchpad.helpers import shortlist
 from canonical.launchpad.interfaces.launchpad import ILaunchpadCelebrities
@@ -641,6 +641,34 @@
             query = query.find(In(BugWatch.id, bug_watch_ids))
         return query
 
+    def bulkSetError(self, bug_watches, last_error_type=None):
+        """See `IBugWatchSet`."""
+        bug_watch_ids = set(
+            (bug_watch.id if IBugWatch.providedBy(bug_watch) else bug_watch)
+            for bug_watch in bug_watches)
+        bug_watches_in_database = IStore(BugWatch).find(
+            BugWatch, In(BugWatch.id, list(bug_watch_ids)))
+        bug_watches_in_database.set(
+            lastchecked=UTC_NOW,
+            last_error_type=last_error_type,
+            next_check=None)
+
+    def bulkAddActivity(self, bug_watches, result=None, message=None,
+                        oops_id=None):
+        """See `IBugWatchSet`."""
+        bug_watch_ids = set(
+            (bug_watch.id if IBugWatch.providedBy(bug_watch) else bug_watch)
+            for bug_watch in bug_watches)
+        insert_activity_statement = (
+            "INSERT INTO BugWatchActivity"
+            " (bug_watch, result, message, oops_id) "
+            "SELECT BugWatch.id, %s, %s, %s FROM BugWatch"
+            " WHERE BugWatch.id IN %s"
+            )
+        IStore(BugWatch).execute(
+            insert_activity_statement % sqlvalues(
+                result, message, oops_id, bug_watch_ids))
+
 
 class BugWatchActivity(Storm):
     """See `IBugWatchActivity`."""
 
=== modified file 'lib/lp/bugs/tests/test_bugwatch.py'
--- lib/lp/bugs/tests/test_bugwatch.py	2010-04-13 01:49:42 +0000
+++ lib/lp/bugs/tests/test_bugwatch.py	2010-04-27 08:44:29 +0000
@@ -13,6 +13,8 @@
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
+from canonical.database.constants import UTC_NOW
+
 from canonical.launchpad.ftests import login, ANONYMOUS
 from canonical.launchpad.scripts.garbo import BugWatchActivityPruner
 from canonical.launchpad.scripts.logger import QuietFakeLogger
@@ -23,7 +25,8 @@
 from lp.bugs.interfaces.bugtask import BugTaskImportance, BugTaskStatus
 from lp.bugs.interfaces.bugtracker import BugTrackerType, IBugTrackerSet
 from lp.bugs.interfaces.bugwatch import (
-    IBugWatchSet, NoBugTrackerFound, UnrecognizedBugTrackerURL)
+    BugWatchActivityStatus, IBugWatchSet, NoBugTrackerFound,
+    UnrecognizedBugTrackerURL)
 from lp.bugs.scripts.checkwatches.scheduler import MAX_SAMPLE_SIZE
 from lp.registry.interfaces.person import IPersonSet
 
@@ -434,6 +437,108 @@
                         bug_watch.id for bug_watch in bug_watches_limited])))
 
 
+class TestBugWatchSetBulkOperations(TestCaseWithFactory):
+    """Tests for the bugwatch updating system."""
+
+    layer = LaunchpadZopelessLayer
+
+    def setUp(self):
+        super(TestBugWatchSetBulkOperations, self).setUp()
+        self.bug_watches = [
+            self.factory.makeBugWatch(remote_bug='alice'),
+            self.factory.makeBugWatch(remote_bug='bob'),
+            ]
+        for bug_watch in self.bug_watches:
+            bug_watch.lastchecked = None
+            bug_watch.last_error_type = None
+            bug_watch.next_check = UTC_NOW
+
+    def _checkStatusOfBugWatches(
+        self, last_checked_is_null, next_check_is_null, last_error_type):
+        for bug_watch in self.bug_watches:
+            self.failUnlessEqual(
+                last_checked_is_null, bug_watch.lastchecked is None)
+            self.failUnlessEqual(
+                next_check_is_null, bug_watch.next_check is None)
+            self.failUnlessEqual(
+                last_error_type, bug_watch.last_error_type)
+
+    def test_bulkSetError(self):
+        # Called with only bug watches, bulkSetError() marks the
+        # watches as checked without error, and unscheduled.
+        getUtility(IBugWatchSet).bulkSetError(self.bug_watches)
+        self._checkStatusOfBugWatches(False, True, None)
+
+    def test_bulkSetError_with_error(self):
+        # Called with bug watches and an error, bulkSetError() marks
+        # the watches with the given error, and unschedules them.
+        error = BugWatchActivityStatus.BUG_NOT_FOUND
+        getUtility(IBugWatchSet).bulkSetError(self.bug_watches, error)
+        self._checkStatusOfBugWatches(False, True, error)
+
+    def test_bulkSetError_with_id_list(self):
+        # The ids of bug watches to update can be passed in.
+        getUtility(IBugWatchSet).bulkSetError(
+            [bug_watch.id for bug_watch in self.bug_watches])
+        self._checkStatusOfBugWatches(False, True, None)
+
+    def test_bulkSetError_with_mixed_list(self):
+        # The list passed in can contain a mix of bug watches and
+        # their ids.
+        getUtility(IBugWatchSet).bulkSetError(
+            [bug_watch.id for bug_watch in self.bug_watches[::2]] +
+            [bug_watch for bug_watch in self.bug_watches[1::2]])
+        self._checkStatusOfBugWatches(False, True, None)
+
+    def test_bulkSetError_with_iterator(self):
+        # Any iterator can be passed in.
+        getUtility(IBugWatchSet).bulkSetError(
+            (bug_watch for bug_watch in self.bug_watches))
+        self._checkStatusOfBugWatches(False, True, None)
+
+    def _checkActivityForBugWatches(self, result, message, oops_id):
+        for bug_watch in self.bug_watches:
+            latest_activity = bug_watch.activity.first()
+            self.failUnlessEqual(result, latest_activity.result)
+            self.failUnlessEqual(message, latest_activity.message)
+            self.failUnlessEqual(oops_id, latest_activity.oops_id)
+
+    def test_bulkAddActivity(self):
+        # Called with only bug watches, bulkAddActivity() adds
+        # successful activity records for the given bug watches.
+        getUtility(IBugWatchSet).bulkAddActivity(self.bug_watches)
+        self._checkActivityForBugWatches(None, None, None)
+
+    def test_bulkAddActivity_with_error(self):
+        # Called with additional error information, bulkAddActivity()
+        # adds appropriate and identical activity records for each of
+        # the given bug watches.
+        error = BugWatchActivityStatus.PRIVATE_REMOTE_BUG
+        getUtility(IBugWatchSet).bulkAddActivity(
+            self.bug_watches, error, "Forbidden", "OOPS-1234")
+        self._checkActivityForBugWatches(error, "Forbidden", "OOPS-1234")
+
+    def test_bulkAddActivity_with_id_list(self):
+        # The ids of bug watches can be passed in.
+        getUtility(IBugWatchSet).bulkAddActivity(
+            [bug_watch.id for bug_watch in self.bug_watches])
+        self._checkActivityForBugWatches(None, None, None)
+
+    def test_bulkAddActivity_with_mixed_list(self):
+        # The list passed in can contain a mix of bug watches and
+        # their ids.
+        getUtility(IBugWatchSet).bulkAddActivity(
+            [bug_watch.id for bug_watch in self.bug_watches[::2]] +
+            [bug_watch for bug_watch in self.bug_watches[1::2]])
+        self._checkActivityForBugWatches(None, None, None)
+
+    def test_bulkAddActivity_with_iterator(self):
+        # Any iterator can be passed in.
+        getUtility(IBugWatchSet).bulkAddActivity(
+            (bug_watch for bug_watch in self.bug_watches))
+        self._checkActivityForBugWatches(None, None, None)
+
+
 class TestBugWatchBugTasks(TestCaseWithFactory):
 
     layer = DatabaseFunctionalLayer