diff options
Diffstat (limited to 'src/leap/soledad/client/_db/blobs.py')
| -rw-r--r-- | src/leap/soledad/client/_db/blobs.py | 23 |
1 files changed, 18 insertions, 5 deletions
diff --git a/src/leap/soledad/client/_db/blobs.py b/src/leap/soledad/client/_db/blobs.py index 60c369b2..ab9de8dd 100644 --- a/src/leap/soledad/client/_db/blobs.py +++ b/src/leap/soledad/client/_db/blobs.py @@ -185,6 +185,7 @@ class BlobManager(object): in local and remote storages. """ max_retries = 3 + concurrency_limit = 3 def __init__( self, local_path, remote, key, secret, user, token=None, @@ -286,9 +287,17 @@ class BlobManager(object): missing = [b_id for b_id in our_blobs if b_id not in server_blobs] logger.info("Amount of documents missing on server: %s" % len(missing)) # TODO: Send concurrently when we are able to stream directly from db - for blob_id in missing: - fd = yield self.local.get(blob_id, namespace) + while missing: + deferreds = [] + for _ in range(min(self.concurrency_limit, len(missing))): + blob_id = missing.pop() + deferreds.append(self.__send_one(blob_id, namespace)) + yield defer.gatherResults(deferreds) + + @defer.inlineCallbacks + def __send_one(self, blob_id, namespace): logger.info("Upload local blob: %s" % blob_id) + fd = yield self.local.get(blob_id, namespace) try: yield self._encrypt_and_upload(blob_id, fd) yield self.local.update_sync_status(blob_id, SyncStatus.SYNCED) @@ -316,9 +325,13 @@ class BlobManager(object): docs_we_want = [b_id for b_id in server_blobs if b_id not in our_blobs] logger.info("Fetching new docs from server: %s" % len(docs_we_want)) # TODO: Fetch concurrently when we are able to stream directly into db - for blob_id in docs_we_want: - logger.info("Fetching new doc: %s" % blob_id) - yield self.get(blob_id, namespace) + while docs_we_want: + deferreds = [] + for _ in range(min(self.concurrency_limit, len(docs_we_want))): + blob_id = docs_we_want.pop() + logger.info("Fetching new doc: %s" % blob_id) + deferreds.append(self.get(blob_id, namespace)) + yield defer.DeferredList(deferreds) @defer.inlineCallbacks def put(self, doc, size, namespace=''): |