summaryrefslogtreecommitdiff
path: root/src/leap/soledad/client/_db
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/soledad/client/_db')
-rw-r--r--src/leap/soledad/client/_db/blobs.py23
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=''):