1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
# -*- coding: utf-8 -*-
# _blobs.py
# Copyright (C) 2017 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Synchronization between blobs client/server
"""
from twisted.internet import defer
from twisted.internet import reactor
from twisted.logger import Logger
from twisted.internet import error
from .sql import SyncStatus
from .errors import RetriableTransferError
logger = Logger()
def sleep(seconds):
d = defer.Deferred()
reactor.callLater(seconds, d.callback, None)
return d
MAX_WAIT = 60 # In seconds. Max time between retries
@defer.inlineCallbacks
def with_retry(func, *args, **kwargs):
retry_wait = 1
retriable_errors = (error.ConnectError, error.ConnectionClosed,
RetriableTransferError,)
while True:
try:
yield func(*args, **kwargs)
break
except retriable_errors:
yield sleep(retry_wait)
retry_wait = min(retry_wait + 10, MAX_WAIT)
class BlobsSynchronizer(object):
@defer.inlineCallbacks
def refresh_sync_status_from_server(self, namespace=''):
d1 = self.remote_list(namespace=namespace)
d2 = self.local_list(namespace=namespace)
remote_list, local_list = yield defer.gatherResults([d1, d2])
pending_download_ids = tuple(set(remote_list) - set(local_list))
yield self.local.update_batch_sync_status(
pending_download_ids,
SyncStatus.PENDING_DOWNLOAD,
namespace=namespace)
@defer.inlineCallbacks
def send_missing(self, namespace=''):
"""
Compare local and remote blobs and send what's missing in server.
:param namespace:
Optional parameter to restrict operation to a given namespace.
:type namespace: str
"""
missing = yield self.local.list_status(
SyncStatus.PENDING_UPLOAD, namespace)
total = len(missing)
logger.info("Will send %d blobs to server." % total)
deferreds = []
semaphore = defer.DeferredSemaphore(self.concurrent_transfers_limit)
for i in xrange(total):
blob_id = missing.pop()
d = semaphore.run(
with_retry, self.__send_one, blob_id, namespace, i, total)
deferreds.append(d)
yield defer.gatherResults(deferreds, consumeErrors=True)
@defer.inlineCallbacks
def __send_one(self, blob_id, namespace, i, total):
logger.info("Sending blob to server (%d/%d): %s"
% (i, total, blob_id))
fd = yield self.local.get(blob_id, namespace=namespace)
yield self._encrypt_and_upload(blob_id, fd)
yield self.local.update_sync_status(blob_id, SyncStatus.SYNCED)
@defer.inlineCallbacks
def fetch_missing(self, namespace=''):
"""
Compare local and remote blobs and fetch what's missing in local
storage.
:param namespace:
Optional parameter to restrict operation to a given namespace.
:type namespace: str
"""
# TODO: Use something to prioritize user requests over general new docs
d = self.local_list_status(SyncStatus.PENDING_DOWNLOAD, namespace)
docs_we_want = yield d
total = len(docs_we_want)
logger.info("Will fetch %d blobs from server." % total)
deferreds = []
semaphore = defer.DeferredSemaphore(self.concurrent_transfers_limit)
for i in xrange(len(docs_we_want)):
blob_id = docs_we_want.pop()
logger.info("Fetching blob (%d/%d): %s" % (i, total, blob_id))
d = semaphore.run(with_retry, self.get, blob_id, namespace)
deferreds.append(d)
yield defer.gatherResults(deferreds, consumeErrors=True)
@defer.inlineCallbacks
def sync(self, namespace=''):
try:
yield self.refresh_sync_status_from_server(namespace)
yield self.fetch_missing(namespace)
yield self.send_missing(namespace)
except defer.FirstError as e:
e.subFailure.raiseException()
|
