Merge lp:~facundo/magicicada-server/dont-always-update-dbupload into lp:magicicada-server

[Facundo Batista]

Only save progress for resumable upload every accumulated STORAGE_CHUNK_SIZE, not for every chunk.

Only save progress for resumable upload every accumulated STORAGE_CHUNK_SIZE, not for every chunk.

[Natalia Bidart (nataliabidart)]

Branch looks good!

I would suggest renaming accumulated as "unsaved" or similar, but is just a nitpick.

review: Approve

[lp:~facundo/magicicada-server/dont-always-update-dbupload]

40. By Facundo Batista on 2015-12-05

Better variable name.

[Facundo Batista (facundo)]

review: Approve

=== modified file 'src/server/content.py'
--- src/server/content.py	2015-10-18 01:35:37 +0000
+++ src/server/content.py	2015-12-05 15:25:28 +0000
@@ -35,6 +35,7 @@
 
 from backends.filesync import errors as dataerrors
 from backends.filesync.models import Share
+from magicicada import settings
 from ubuntuone.storage.server import errors, upload
 from ubuntuone.storageprotocol import protocol_pb2
 
@@ -153,6 +154,9 @@
                  multipart_key, chunk_count, when_last_active):
         self.__dict__ = locals()
 
+        # will only update the DB with parts when accumulate over a trigger
+        self.unsaved_count = 0
+
     @classmethod
     def get(cls, user, volume_id, node_id, uploadjob_id, hash_value, crc32):
         """Get a multipart upload job."""
@@ -183,9 +187,16 @@
 
     def add_part(self, chunk_size):
         """Add a part to an upload job."""
-        kwargs = dict(user_id=self.user.id, volume_id=self.volume_id,
-                      uploadjob_id=self.uploadjob_id, chunk_size=chunk_size)
-        return self.user.rpc_dal.call('add_part_to_uploadjob', **kwargs)
+        self.unsaved_count += chunk_size
+        if self.unsaved_count >= settings.api_server.STORAGE_CHUNK_SIZE:
+            self.unsaved_count -= settings.api_server.STORAGE_CHUNK_SIZE
+            kwargs = dict(user_id=self.user.id, volume_id=self.volume_id,
+                          uploadjob_id=self.uploadjob_id,
+                          chunk_size=chunk_size)
+            d = self.user.rpc_dal.call('add_part_to_uploadjob', **kwargs)
+        else:
+            d = defer.succeed(True)
+        return d
 
     @defer.inlineCallbacks
     def delete(self):

=== modified file 'src/server/tests/test_content.py'
--- src/server/tests/test_content.py	2015-10-18 01:35:37 +0000
+++ src/server/tests/test_content.py	2015-12-05 15:25:28 +0000
@@ -2117,12 +2117,7 @@
         def auth(client):
 
             def raise_quota(_):
-
-                # we need a bigger quota for this test...
-                def _raise_quota(new_size):
-                    self.usr0.update(max_storage_bytes=new_size)
-                d = threads.deferToThread(_raise_quota, size * 2)
-                return d
+                self.usr0.update(max_storage_bytes=size * 2)
 
             def check_content(content):
                 self.assertEqual(content.data, deflated_data)
@@ -2186,6 +2181,43 @@
         self.assertFails(d, 'NOT_AVAILABLE')
         return d
 
+    def test_deferred_add_part_to_uj(self):
+        """Check that parts are added to upload job only after a limit."""
+        size = int(settings.api_server.STORAGE_CHUNK_SIZE * 2.5)
+        data = os.urandom(size)
+        deflated_data = zlib.compress(data)
+        hash_object = content_hash_factory()
+        hash_object.update(data)
+        hash_value = hash_object.content_hash()
+        crc32_value = crc32(data)
+        deflated_size = len(deflated_data)
+
+        recorded_calls = []
+        orig_call = self.service.rpcdal_client.call
+
+        def recording_call(method, **parameters):
+            if method == 'add_part_to_uploadjob':
+                recorded_calls.append(parameters)
+            return orig_call(method, **parameters)
+
+        self.service.rpcdal_client.call = recording_call
+
+        @defer.inlineCallbacks
+        def test(client):
+            yield client.dummy_authenticate("open sesame")
+            yield self.usr0.update(max_storage_bytes=size * 2)
+            root = yield client.get_root()
+            mkfile_req = yield client.make_file(request.ROOT, root, "hola")
+            putcontent_req = yield client.put_content_request(
+                request.ROOT, mkfile_req.new_id, NO_CONTENT_HASH, hash_value,
+                crc32_value, size, deflated_size, StringIO(deflated_data))
+            yield putcontent_req.deferred
+
+            # check calls; there should be only 2, as size == chunk size * 2.5
+            self.assertEqual(len(recorded_calls), 2)
+
+        return self.callback_test(test, add_default_callbacks=True)
+
 
 class UserTest(TestWithDatabase):
     """Test User functionality."""
@@ -3058,13 +3090,14 @@
     def test_add_part(self):
         """Test add_part method."""
         dbuj = yield self._make_uj()
-        yield dbuj.add_part('chunk_size')
+        chunk_size = int(settings.api_server.STORAGE_CHUNK_SIZE) + 1
+        yield dbuj.add_part(chunk_size)
 
         # check it called rpcdal correctly
         method, attribs = self.user.recorded
         self.assertEqual(method, 'add_part_to_uploadjob')
         should = dict(user_id='fake_user_id', uploadjob_id='uploadjob_id',
-                      chunk_size='chunk_size', volume_id='volume_id')
+                      chunk_size=chunk_size, volume_id='volume_id')
         self.assertEqual(attribs, should)
 
     @defer.inlineCallbacks