Merge lp:~facundo/ubuntuone-client/aq-better-waiting-structures into lp:ubuntuone-client

[Facundo Batista]

Better waiting structures for 'conditions' and 'inactive queue' (LP: #720844)

Better waiting structures for 'conditions' and 'inactive queue'

Both types of waiting needs are very different, so there's an explanation
for each.

Wait for conditions, this rarely happen, so there is no problem to create a deferred for each command to lock in this case. The problem here was that on each "check_conditions" called (that happens frequently, and will happen more frequently after a bug is fixed in VM), *all* commands were called to check_conditions, and this could be very expensive when a lot of commands were queued.

Now, a waiting structure is created for this case. The command just locks through it, and the check conditions are called only through this self-locked commands.

Wait for active queue: this happened a lot (every time the client disconnected), and each time it happened, a deferred were created for each command queued, which could be very expensive memory-wise. Furthermore, when the queue was run again, all commands were sequentially called to unlock that deferred, which was expensive cpu-wise.

Now, a single deferred is used for all commands (that live in the queue). All commands wait for that deferred, and when the queue runs, it just triggers that deferred.

Tests included for everything.

[Lucio Torre (lucio.torre)]

review: Approve

[Facundo Batista (facundo)]

Approving with one review

review: Approve

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2011-03-07 16:54:25 +0000
+++ tests/syncdaemon/test_action_queue.py	2011-03-08 13:52:39 +0000
@@ -63,7 +63,7 @@
     CreateShare, DeleteShare, GetPublicFiles, GetDelta, GetDeltaFromScratch,
     TRANSFER_PROGRESS_THRESHOLD, Unlink, Move, MakeFile, MakeDir, DeltaList,
     ZipQueue, DeferredMap, ThrottlingStorageClient, PathLockingTree,
-    InterruptibleDeferred, DeferredInterrupted,
+    InterruptibleDeferred, DeferredInterrupted, ConditionsLocker,
 )
 from ubuntuone.syncdaemon.event_queue import EventQueue, EVENTS
 from ubuntuone.syncdaemon.marker import MDMarker
@@ -105,7 +105,6 @@
 
     is_runnable = True
     paused = False
-    resumed = False
     conditions_checked = False
 
     def __init__(self, share_id=None, node_id=None):
@@ -120,10 +119,6 @@
         """Mark as paused."""
         self.paused = True
 
-    def resume(self):
-        """Mark as resumed."""
-        self.resumed = True
-
     @property
     def uniqueness(self):
         """Fake uniqueness."""
@@ -136,10 +131,6 @@
         """Cancel!"""
         self.cancelled = True
 
-    def check_conditions(self):
-        """Mark as checked."""
-        self.conditions_checked = True
-
 
 class FakedEventQueue(EventQueue):
     """Faked event queue."""
@@ -534,23 +525,20 @@
         """RQ borns not active."""
         self.assertFalse(self.rq.active)
 
+    def test_init_activedef(self):
+        """Just instanced queue has the deferred to take."""
+        self.assertTrue(isinstance(self.rq.active_deferred, defer.Deferred))
+
     def test_run_goes_active(self):
         """Activate on run."""
         self.rq.run()
         self.assertTrue(self.rq.active)
 
-    def test_run_resume_commands(self):
-        """Resume all queued command on run."""
-        # set up
-        cmd1 = FakeCommand()
-        cmd2 = FakeCommand()
-        self.rq.waiting.extend((cmd1, cmd2))
-        assert not cmd1.resumed and not cmd2.resumed
-
-        # run and check
+    def test_run_triggers_activedef(self):
+        """Trigger the active_deferred on run."""
+        assert not self.rq.active_deferred.called
         self.rq.run()
-        self.assertTrue(cmd1.resumed)
-        self.assertTrue(cmd2.resumed)
+        self.assertTrue(self.rq.active_deferred.called)
 
     def test_stop_goes_inactive(self):
         """Desactivate on stop."""
@@ -571,18 +559,24 @@
         self.assertTrue(cmd1.paused)
         self.assertTrue(cmd2.paused)
 
-    def test_check_conditions(self):
-        """Check all conditions on the commands."""
-        # set up
-        cmd1 = FakeCommand()
-        cmd2 = FakeCommand()
-        self.rq.waiting.extend((cmd1, cmd2))
-        assert not cmd1.conditions_checked and not cmd2.conditions_checked
-
-        # check conditions and test
-        self.rq.check_conditions()
-        self.assertTrue(cmd1.conditions_checked)
-        self.assertTrue(cmd2.conditions_checked)
+    def test_stop_pause_useful_activedef(self):
+        """Refresh the active_deferred before pausing."""
+        checked = defer.Deferred()
+
+        def fake_pause():
+            """Check that RQ has a useful active_deferred."""
+            self.assertTrue(isinstance(self.rq.active_deferred,
+                                       defer.Deferred))
+            self.assertFalse(self.rq.active_deferred.called)
+            checked.callback(True)
+
+        cmd = FakeCommand()
+        cmd.pause = fake_pause
+        self.rq.waiting.append(cmd)
+
+        # stop and test
+        self.rq.stop()
+        return checked
 
     def test_unqueue_remove(self):
         """Remove the command from queue on unqueue."""
@@ -1781,11 +1775,14 @@
         # run first time
         self.cmd.run()
         self.assertFalse(called)
+        self.assertTrue(self.handler.check_debug(
+                        'not running because of inactive queue'))
+        self.assertFalse(self.handler.check_debug('unblocked: queue active'))
 
         # active the queue
-        self.rq.active = True
-        self.cmd.resume()
+        self.rq.run()
         self.assertTrue(called)
+        self.assertTrue(self.handler.check_debug('unblocked: queue active'))
 
     def test_run_command_not_runnable(self):
         """Waiting cycle for command not runnable."""
@@ -1800,11 +1797,15 @@
         # run first time
         self.cmd.run()
         self.assertFalse(called)
+        self.assertTrue(self.handler.check_debug(
+                        'not running because of conditions'))
+        self.assertFalse(self.handler.check_debug('unblocked: conditions ok'))
 
         # active the command
         self.cmd.is_runnable = True
-        self.cmd.check_conditions()
+        self.action_queue.conditions_locker.check_conditions()
         self.assertTrue(called)
+        self.assertTrue(self.handler.check_debug('unblocked: conditions ok'))
 
     def test_run_notrunnable_inactivequeue(self):
         """Mixed behaviour between both stoppers."""
@@ -1821,19 +1822,17 @@
 
         # active the queue but inactive the command
         self.cmd.is_runnable = False
-        self.rq.active = True
-        self.cmd.resume()
+        self.rq.run()
         self.assertFalse(called)
 
         # active the command but inactive the queue again!
-        self.rq.active = False
+        self.rq.stop()
         self.cmd.is_runnable = True
-        self.cmd.check_conditions()
+        self.action_queue.conditions_locker.check_conditions()
         self.assertFalse(called)
 
         # finally resume the queue
-        self.rq.active = True
-        self.cmd.resume()
+        self.rq.run()
         self.assertTrue(called)
 
     def test_run_inactivequeue_cancel(self):
@@ -1851,8 +1850,7 @@
         self.cmd.cancel()
 
         # active the queue
-        self.rq.active = True
-        self.cmd.resume()
+        self.rq.run()
         self.assertFalse(called)
         self.assertTrue(self.handler.check_debug(
                         'cancelled before trying to run'))
@@ -1873,7 +1871,7 @@
 
         # active the command
         self.cmd.is_runnable = True
-        self.cmd.check_conditions()
+        self.action_queue.conditions_locker.check_conditions()
         self.assertFalse(called)
         self.handler.debug = True
         self.assertTrue(self.handler.check_debug(
@@ -1967,29 +1965,23 @@
         called = []
         self.cmd.finish = lambda: called.append(True)
         self.cmd.markers_resolved_deferred = defer.succeed(True)
+        self.rq.waiting.append(self.cmd)
         assert self.rq.active
 
         # deferreds, first one stucks, the second allows to continue
         deferreds = [defer.Deferred(), defer.succeed(True)]
-
-        def fake_run():
-            """Set the queue inactive to avoid retry loop and fail."""
-            self.rq.active = False
-            return deferreds.pop(0)
-
-        # set up and test
-        self.cmd._run = fake_run
+        self.cmd._run = lambda: deferreds.pop(0)
 
         # run and check finish was not called
         self.cmd.run()
         self.assertFalse(called)
 
         # pause, still nothing called
-        self.cmd.pause()
+        self.rq.stop()
+        self.assertFalse(called)
 
         # resume, now it finished!
-        self.rq.active = True
-        self.cmd.resume()
+        self.rq.run()
         self.assertTrue(called)
 
     @defer.inlineCallbacks
@@ -2050,36 +2042,6 @@
         self.assertTrue(self.handler.check_debug("pausing"))
         self.assertTrue(called)
 
-    def test_resume(self):
-        """Trigger the deferred only if there."""
-        # nothing called when no deferred
-        assert self.cmd.wait_for_queue is None
-        self.cmd.resume()
-        self.assertFalse(self.handler.check_debug('resuming'))
-
-        # the deferred is triggered if there
-        d = defer.Deferred()
-        self.cmd.wait_for_queue = d
-        self.cmd.resume()
-        self.assertIdentical(self.cmd.wait_for_queue, None)
-        self.assertTrue(d.called)
-        self.assertTrue(self.handler.check_debug('resuming'))
-
-    def test_check_conditions(self):
-        """Trigger the deferred only if there."""
-        # nothing called when no deferred
-        assert self.cmd.wait_for_conditions is None
-        self.cmd.check_conditions()
-        self.assertFalse(self.handler.check_debug('unblocking conditions'))
-
-        # the deferred is triggered if there
-        d = defer.Deferred()
-        self.cmd.wait_for_conditions = d
-        self.cmd.check_conditions()
-        self.assertIdentical(self.cmd.wait_for_conditions, None)
-        self.assertTrue(d.called)
-        self.assertTrue(self.handler.check_debug('unblocking conditions'))
-
     def test_cancel_works(self):
         """Do default cleaning."""
         called = []
@@ -2093,18 +2055,9 @@
         self.assertTrue(self.handler.check_debug('cancelled'))
 
     def test_cancel_releases_conditions(self):
-        """Cancel unlocks the conditions deferred."""
-        self.cmd.finish = lambda: None # don't try to unqueue!
-        d = defer.Deferred()
-        self.cmd.wait_for_conditions = d
-        self.cmd.cancel()
-        self.assertTrue(d.called)
-
-    def test_cancel_releases_queue(self):
-        """Cancel unlocks the wait-for-queue deferred."""
-        self.cmd.finish = lambda: None # don't try to unqueue!
-        d = defer.Deferred()
-        self.cmd.wait_for_queue = d
+        """Cancel calls the conditions locker for the command."""
+        self.cmd.finish = lambda: None # don't try to unqueue!
+        d = self.action_queue.conditions_locker.get_lock(self.cmd)
         self.cmd.cancel()
         self.assertTrue(d.called)
 
@@ -4896,7 +4849,7 @@
 
         # fix conditions and check them
         self.cmd.is_runnable = True
-        self.queue.check_conditions()
+        self.action_queue.conditions_locker.check_conditions()
 
         # all check
         self.assertEqual(called, ['run', 'finish'])
@@ -4920,7 +4873,7 @@
         self.cmd.go()
 
         # before the command finishes, all conditions are checked
-        self.queue.check_conditions()
+        self.action_queue.conditions_locker.check_conditions()
 
         # command finished
         d.callback(2)
@@ -5051,7 +5004,7 @@
 
         # fix conditions
         self.cmd.is_runnable = True
-        self.queue.check_conditions()
+        self.action_queue.conditions_locker.check_conditions()
 
         # need to wait the callLater
         yield finished
@@ -5127,7 +5080,7 @@
     def test_cancel_while_waiting_queue(self):
         """Cancel the command while waiting for queue."""
         # stop the queue, and fake the pathlock to test releasing
-        self.queue.active = False
+        self.queue.stop()
         released = []
         self.cmd._acquire_pathlock = lambda: defer.succeed(
                                                 lambda: released.append(True))
@@ -5137,6 +5090,10 @@
         self.cmd.go()
         self.cmd.cancel()
 
+        # now, set the queue active again, it should release everything
+        # even if was cancelled before
+        self.queue.run()
+
         # all check
         self._check_finished_ok()
         self.assertTrue(released)
@@ -5311,3 +5268,99 @@
 
         # further callback to original deferred is harmless
         origdef.errback(ValueError('foo'))
+
+
+class ConditionsLockerTests(TwistedTestCase):
+    """Test the ConditionsLocker."""
+
+    def setUp(self):
+        """Set up."""
+        self.cl = ConditionsLocker()
+
+    def test_get_locking_deferred_returns_deferred(self):
+        """The locking is done by a deferred."""
+        d = self.cl.get_lock('command')
+        d.callback(True)
+        return d
+
+    def test_get_locking_different_commands_different_deferreds(self):
+        """Asked by two commands, get two deferreds."""
+        d1 = self.cl.get_lock('command1')
+        d2 = self.cl.get_lock('command2')
+        self.assertNotIdentical(d1, d2)
+
+    def test_get_locking_same_command_same_deferred(self):
+        """If asked twice by the same command, return the same deferred.
+
+        This is more a safe guard than a feature; if misused by the same
+        command we're assuring than we will not overwrite a second deferred
+        over the first one (so, never releasing the first one).
+        """
+        d1 = self.cl.get_lock('command')
+        d2 = self.cl.get_lock('command')
+        self.assertIdentical(d1, d2)
+
+    def test_check_conditions_simple_runnable(self):
+        """Release the command."""
+        cmd = FakeCommand()
+        locking_d = self.cl.get_lock(cmd)
+        self.assertFalse(locking_d.called)
+        self.assertIn(cmd, self.cl.locked)
+
+        # release it!
+        assert cmd.is_runnable
+        self.cl.check_conditions()
+        self.assertTrue(locking_d.called)
+        self.assertNotIn(cmd, self.cl.locked)
+
+    def test_check_conditions_simple_notrunnable_then_ok(self):
+        """First don't release the command, then release it."""
+        cmd = FakeCommand()
+        locking_d = self.cl.get_lock(cmd)
+        self.assertFalse(locking_d.called)
+
+        # check for conditions, do not release
+        cmd.is_runnable = False
+        self.cl.check_conditions()
+        self.assertFalse(locking_d.called)
+
+        # conditions are ok now, release
+        cmd.is_runnable = True
+        self.cl.check_conditions()
+        self.assertTrue(locking_d.called)
+
+    def test_check_conditions_mixed(self):
+        """Several commands, mixed situation."""
+        cmd1 = FakeCommand()
+        cmd1.is_runnable = False
+        cmd2 = FakeCommand()
+        assert cmd2.is_runnable
+
+        # get lock for both, and check conditions
+        locking_d1 = self.cl.get_lock(cmd1)
+        locking_d2 = self.cl.get_lock(cmd2)
+        self.cl.check_conditions()
+
+        # one should be released, the other should not
+        self.assertFalse(locking_d1.called)
+        self.assertTrue(locking_d2.called)
+
+    def test_cancel_command_nothold(self):
+        """It's ok to cancel a command not there."""
+        self.cl.cancel_command('command')
+
+    def test_cancel_releases_cancelled_command(self):
+        """It releases the cancelled command, even not runnable."""
+        cmd1 = FakeCommand()
+        cmd1.is_runnable = False
+        cmd2 = FakeCommand()
+        assert cmd2.is_runnable
+
+        # get lock for both, and cancel only 1
+        locking_d1 = self.cl.get_lock(cmd1)
+        locking_d2 = self.cl.get_lock(cmd2)
+        self.cl.cancel_command(cmd1)
+
+        # 1 should be released, 2 should not (even with conditions ok)
+        self.assertTrue(locking_d1.called)
+        self.assertFalse(locking_d2.called)

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2011-03-07 16:54:25 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2011-03-08 13:52:39 +0000
@@ -455,6 +455,7 @@
         self.hashed_waiting = {}
         self.active = False
         self.transfers_semaphore = defer.DeferredSemaphore(SIMULT_TRANSFERS)
+        self.active_deferred = defer.Deferred()
 
     def __len__(self):
         """Return the length of the waiting queue."""
@@ -488,20 +489,15 @@
         if len(self.waiting) == 0:
             self.action_queue.event_queue.push('SYS_QUEUE_DONE')
 
-    def check_conditions(self):
-        """Check conditions on which the commands may be waiting."""
-        for command in self.waiting[:]:
-            command.check_conditions()
-
     def run(self):
         """Go active and run all commands in the queue."""
         self.active = True
-        for command in self.waiting[:]:
-            command.resume()
+        self.active_deferred.callback(True)
 
     def stop(self):
         """Stop the pool and cleanup the running commands."""
         self.active = False
+        self.active_deferred = defer.Deferred()
         for command in self.waiting:
             command.pause()
 
@@ -557,6 +553,36 @@
                 d.errback(failure)
 
 
+class ConditionsLocker(object):
+    """Structure to hold commands waiting because of conditions.
+
+    On each call to lock it will return a deferred for the received
+    command. When check_conditions is called, it will trigger each
+    command deferred if it's runnable.
+    """
+    def __init__(self):
+        self.locked = {}
+
+    def get_lock(self, command):
+        """Return the deferred that will lock the command."""
+        if command not in self.locked:
+            self.locked[command] = defer.Deferred()
+        return self.locked[command]
+
+    def check_conditions(self):
+        """Check for all commands' conditions, and release accordingly."""
+        for cmd in self.locked.keys():
+            if cmd.is_runnable:
+                deferred = self.locked.pop(cmd)
+                deferred.callback(True)
+
+    def cancel_command(self, command):
+        """The command was cancelled, if lock hold, release it and clean."""
+        if command in self.locked:
+            deferred = self.locked.pop(command)
+            deferred.callback(True)
+
+
 class UploadProgressWrapper(object):
     """A wrapper around the file-like object used for Uploads.
 
@@ -630,13 +656,14 @@
         self.pathlock = PathLockingTree()
         self.uuid_map = DeferredMap()
         self.zip_queue = ZipQueue()
+        self.conditions_locker = ConditionsLocker()
 
         self.estimated_free_space = {}
         event_queue.subscribe(self)
 
     def check_conditions(self):
-        """Poll conditions on which running actions may be waiting."""
-        self.queue.check_conditions()
+        """Check conditions in the locker, to release all the waiting ops."""
+        self.conditions_locker.check_conditions()
 
     def have_sufficient_space_for_upload(self, share_id, upload_size):
         """Returns True if we have sufficient space for the given upload."""
@@ -1099,7 +1126,7 @@
 
     __slots__ = ('_queue', 'running', 'pathlock_release', 'log',
                  'markers_resolved_deferred', 'action_queue', 'cancelled',
-                 'wait_for_queue', 'wait_for_conditions', 'running_deferred')
+                 'running_deferred')
 
     def __init__(self, request_queue):
         """Initialize a command instance."""
@@ -1110,9 +1137,6 @@
         self.markers_resolved_deferred = defer.Deferred()
         self.pathlock_release = None
         self.cancelled = False
-
-        self.wait_for_queue = None
-        self.wait_for_conditions = None
         self.running_deferred = None
 
     def to_dict(self):
@@ -1196,20 +1220,6 @@
             self.running_deferred.interrupt()
         self.cleanup()
 
-    def resume(self):
-        """Unlock the command because the queue is back alive."""
-        if self.wait_for_queue is not None:
-            self.log.debug('resuming')
-            self.wait_for_queue.callback(True)
-            self.wait_for_queue = None
-
-    def check_conditions(self):
-        """If conditions are ok, run the command again."""
-        if self.is_runnable and self.wait_for_conditions is not None:
-            self.log.debug('unblocking conditions')
-            self.wait_for_conditions.callback(True)
-            self.wait_for_conditions = None
-
     @defer.inlineCallbacks
     def go(self):
         """Execute all the steps for a command."""
@@ -1261,14 +1271,14 @@
             # if queue not active, wait for it and check again
             if not self._queue.active:
                 self.log.debug('not running because of inactive queue')
-                self.wait_for_queue = defer.Deferred()
-                yield self.wait_for_queue
+                yield self._queue.active_deferred
+                self.log.debug('unblocked: queue active')
                 continue
 
             if not self.is_runnable:
                 self.log.debug('not running because of conditions')
-                self.wait_for_conditions = defer.Deferred()
-                yield self.wait_for_conditions
+                yield self.action_queue.conditions_locker.get_lock(self)
+                self.log.debug('unblocked: conditions ok')
                 continue
 
             try:
@@ -1313,8 +1323,7 @@
     def cancel(self):
         """Cancel the command.
 
-        Also trigger the wait_for_condition and wait_for_queue deferreds, to
-        unlock the command and finally release the pathlock.
+        Also cancel the command in the conditions locker.
 
         Do nothing if already cancelled (as cancellation can come from other
         thread, it can come at any time, so we need to support double
@@ -1327,12 +1336,7 @@
 
         self.cancelled = True
         self.log.debug('cancelled')
-        if self.wait_for_conditions is not None:
-            self.wait_for_conditions.callback(True)
-            self.wait_for_conditions = None
-        if self.wait_for_queue is not None:
-            self.wait_for_queue.callback(True)
-            self.wait_for_queue = None
+        self.action_queue.conditions_locker.cancel_command(self)
         self.cleanup()
         self.finish()
         return True