Merge lp:~dobey/ubuntuone-client/queue-limit-3-0 into lp:ubuntuone-client/stable-3-0

[dobey]

Use the correct comparison to decide in which queue (LP: #978903).

[Eric Casteleijn (thisfred)]

review: Approve

[Zachery Bir (urbanape)]

review: Approve

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2012-04-09 20:07:05 +0000
+++ tests/syncdaemon/test_action_queue.py	2012-04-24 19:00:28 +0000
@@ -31,6 +31,7 @@
 from __future__ import with_statement
 
 import base64
+import collections
 import inspect
 import logging
 import operator
@@ -120,6 +121,26 @@
         return self.check(logger.NOTE, *msgs)
 
 
+class FakeOffloadQueue(object):
+    """Fake replacemente for offload_queue."""
+    def __init__(self):
+        self.queue = collections.deque()
+
+    def push(self, item):
+        """Push it."""
+        self.queue.append(item)
+
+    def pop(self):
+        """Pop it."""
+        return self.queue.popleft()
+
+    def __len__(self):
+        return len(self.queue)
+
+    def __getitem__(self, idx):
+        return self.queue[idx]
+
+
 class FakeMagicHash(object):
     """Fake magic hash."""
     _magic_hash = '666'
@@ -5858,3 +5879,93 @@
         self.assertEqual(called[0], ((FakeCommand, 'arg0'), {'bar': 'baz'}))
         self.assertEqual(called[1], ((FakeCommand, 'arg1'), {'foo': 'bar'}))
         self.assertEqual(called[2], ((FakeCommand, 'arg2'), {}))
+
+    def test_execute_pushing_popping(self):
+        """Exercise the limits when pushing/popping to disk."""
+        aq = self.action_queue
+        aq.memory_pool_limit = 2
+
+        def _fake_execute(_, cmd):
+            """Don't really execute, but store and return deferred.
+
+            It also handles the queue.
+            """
+            d = defer.Deferred()
+            commands.append((cmd, d))
+            aq.queue.append(cmd)
+
+            def remove(_):
+                aq.queue.remove(cmd)
+                commands.remove((cmd, d))
+
+            d.addCallback(remove)
+            return d
+
+        commands = []
+        self.patch(aq, '_really_execute', _fake_execute)
+        aq.disk_queue = FakeOffloadQueue()
+        aq.queue = []
+        aq.commands[FakeCommand.__name__] = FakeCommand
+
+        # send two commands, they should be executed right away
+        aq.execute(FakeCommand, 1)
+        aq.execute(FakeCommand, 2)
+        self.assertEqual(commands[0][0], 1)
+        self.assertEqual(commands[1][0], 2)
+
+        # send a third and fourth commands, they should be offloaded
+        aq.execute(FakeCommand, 3)
+        aq.execute(FakeCommand, 4)
+        self.assertEqual(len(commands), 2)
+        self.assertEqual(len(aq.disk_queue), 2)
+        self.assertEqual(aq.disk_queue[0], ('FakeCommand', (3,), {}))
+        self.assertEqual(aq.disk_queue[1], ('FakeCommand', (4,), {}))
+
+        # finish command 1, it should pop and execute command 3
+        commands[0][1].callback(True)
+        self.assertEqual(len(commands), 2)
+        self.assertEqual(commands[0][0], 2)
+        self.assertEqual(commands[1][0], 3)
+        self.assertEqual(len(aq.disk_queue), 1)
+        self.assertEqual(aq.disk_queue[0], ('FakeCommand', (4,), {}))
+
+        # other command should go offload
+        aq.execute(FakeCommand, 5)
+        self.assertEqual(len(commands), 2)
+        self.assertEqual(len(aq.disk_queue), 2)
+        self.assertEqual(aq.disk_queue[0], ('FakeCommand', (4,), {}))
+        self.assertEqual(aq.disk_queue[1], ('FakeCommand', (5,), {}))
+
+        # finish commands 2 and 3... 4 and 5 should go in
+        commands[0][1].callback(True)
+        commands[0][1].callback(True)
+        self.assertEqual(len(commands), 2)
+        self.assertEqual(commands[0][0], 4)
+        self.assertEqual(commands[1][0], 5)
+        self.assertEqual(len(aq.disk_queue), 0)
+
+        # even in the edge, another command should be offloaded
+        aq.execute(FakeCommand, 6)
+        self.assertEqual(len(commands), 2)
+        self.assertEqual(len(aq.disk_queue), 1)
+        self.assertEqual(aq.disk_queue[0], ('FakeCommand', (6,), {}))
+
+        # finish 4 and 5, we should only have 6 left
+        commands[0][1].callback(True)
+        commands[0][1].callback(True)
+        self.assertEqual(len(commands), 1)
+        self.assertEqual(commands[0][0], 6)
+        self.assertEqual(len(aq.disk_queue), 0)
+
+        # one below the limit, next op should be executed right away
+        aq.execute(FakeCommand, 7)
+        self.assertEqual(len(commands), 2)
+        self.assertEqual(commands[0][0], 6)
+        self.assertEqual(commands[1][0], 7)
+        self.assertEqual(len(aq.disk_queue), 0)
+
+        # finish 6 and 7, all clean
+        commands[0][1].callback(True)
+        commands[0][1].callback(True)
+        self.assertEqual(len(commands), 0)
+        self.assertEqual(len(aq.disk_queue), 0)

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2012-04-09 20:08:42 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2012-04-24 19:00:28 +0000
@@ -1095,8 +1095,9 @@
     @defer.inlineCallbacks
     def execute(self, command_class, *args, **kwargs):
         """Execute a command only if there's room in memory to handle it."""
-        if len(self.queue) > self.memory_pool_limit:
-            # not enough room in memory, store it in the offloaded queue
+        if len(self.queue) >= self.memory_pool_limit:
+            # already in the limit, can't go further as we don't have
+            # more room in memory, store it in the offloaded queue
             logger.debug('offload push: %s %s %s', command_class.__name__, args, kwargs)
             self.disk_queue.push((command_class.__name__, args, kwargs))
             return