From bc496b72dc66a5e44b3b99bbe711e5ff523e1848 Mon Sep 17 00:00:00 2001 From: Daira Hopwood Date: Mon, 28 Dec 2015 15:28:05 +0000 Subject: [PATCH] Teach StorageFarmBroker to fire a deferred when a connection threshold is reached. refs #1449 Signed-off-by: Daira Hopwood --- src/allmydata/client.py | 11 ++++++++++- src/allmydata/frontends/drop_upload.py | 8 ++++++++ src/allmydata/storage_client.py | 20 ++++++++++++++++---- src/allmydata/test/test_checker.py | 4 ++-- src/allmydata/test/test_client.py | 4 ++-- src/allmydata/test/test_helper.py | 2 +- src/allmydata/test/test_mutable.py | 2 +- src/allmydata/test/test_upload.py | 2 +- src/allmydata/test/test_web.py | 2 +- 9 files changed, 42 insertions(+), 13 deletions(-) diff --git a/src/allmydata/client.py b/src/allmydata/client.py index 41840e85..24432787 100644 --- a/src/allmydata/client.py +++ b/src/allmydata/client.py @@ -130,6 +130,7 @@ class Client(node.Node, pollmixin.PollMixin): def __init__(self, basedir="."): node.Node.__init__(self, basedir) + self.connected_enough_d = defer.Deferred() self.started_timestamp = time.time() self.logSource="Client" self.encoding_params = self.DEFAULT_ENCODING_PARAMETERS.copy() @@ -346,7 +347,12 @@ class Client(node.Node, pollmixin.PollMixin): # (and everybody else who wants to use storage servers) ps = self.get_config("client", "peers.preferred", "").split(",") preferred_peers = tuple([p.strip() for p in ps if p != ""]) - sb = storage_client.StorageFarmBroker(self.tub, permute_peers=True, preferred_peers=preferred_peers) + + connection_threshold = min(self.encoding_params["k"], + self.encoding_params["happy"] + 1) + + sb = storage_client.StorageFarmBroker(self.tub, True, connection_threshold, + self.connected_enough_d, preferred_peers=preferred_peers) self.storage_broker = sb # load static server specifications from tahoe.cfg, if any. @@ -502,6 +508,9 @@ class Client(node.Node, pollmixin.PollMixin): s = drop_upload.DropUploader(self, upload_dircap, local_dir_utf8) s.setServiceParent(self) s.startService() + + # start processing the upload queue when we've connected to enough servers + self.connected_enough_d.addCallback(s.ready) except Exception, e: self.log("couldn't start drop-uploader: %r", args=(e,)) diff --git a/src/allmydata/frontends/drop_upload.py b/src/allmydata/frontends/drop_upload.py index 42500771..5814c112 100644 --- a/src/allmydata/frontends/drop_upload.py +++ b/src/allmydata/frontends/drop_upload.py @@ -35,6 +35,8 @@ class DropUploader(service.MultiService): self._convergence = client.convergence self._local_path = FilePath(local_dir) + self.is_upload_ready = False + if inotify is None: from twisted.internet import inotify self._inotify = inotify @@ -68,6 +70,12 @@ class DropUploader(service.MultiService): self._stats_provider.count('drop_upload.dirs_monitored', 1) return d + def upload_ready(self): + """upload_ready is used to signal us to start + processing the upload items... + """ + self.is_upload_ready = True + def _notify(self, opaque, path, events_mask): self._log("inotify event %r, %r, %r\n" % (opaque, path, ', '.join(self._inotify.humanReadableMask(events_mask)))) diff --git a/src/allmydata/storage_client.py b/src/allmydata/storage_client.py index dd9780f2..09148932 100644 --- a/src/allmydata/storage_client.py +++ b/src/allmydata/storage_client.py @@ -62,10 +62,13 @@ class StorageFarmBroker: I'm also responsible for subscribing to the IntroducerClient to find out about new servers as they are announced by the Introducer. """ - def __init__(self, tub, permute_peers, preferred_peers=()): + def __init__(self, tub, permute_peers, connected_threshold, connected_d, + preferred_peers=()): self.tub = tub assert permute_peers # False not implemented yet self.permute_peers = permute_peers + self.connected_threshold = connected_threshold + self.connected_d = connected_d self.preferred_peers = preferred_peers # self.servers maps serverid -> IServer, and keeps track of all the # storage servers that we've heard about. Each descriptor manages its @@ -76,7 +79,7 @@ class StorageFarmBroker: # these two are used in unit tests def test_add_rref(self, serverid, rref, ann): - s = NativeStorageServer(serverid, ann.copy()) + s = NativeStorageServer(serverid, ann.copy(), self) s.rref = rref s._is_connected = True self.servers[serverid] = s @@ -93,7 +96,7 @@ class StorageFarmBroker: precondition(isinstance(key_s, str), key_s) precondition(key_s.startswith("v0-"), key_s) assert ann["service-name"] == "storage" - s = NativeStorageServer(key_s, ann) + s = NativeStorageServer(key_s, ann, self) serverid = s.get_serverid() old = self.servers.get(serverid) if old: @@ -119,6 +122,13 @@ class StorageFarmBroker: for dsc in self.servers.values(): dsc.try_to_connect() + def check_enough_connected(self): + if (self.connected_d is not None and + len(self.get_connected_servers()) >= self.connected_threshold): + d = self.connected_d + self.connected_d = None + d.callback(None) + def get_servers_for_psi(self, peer_selection_index): # return a list of server objects (IServers) assert self.permute_peers == True @@ -191,9 +201,10 @@ class NativeStorageServer: "application-version": "unknown: no get_version()", } - def __init__(self, key_s, ann): + def __init__(self, key_s, ann, broker): self.key_s = key_s self.announcement = ann + self.broker = broker assert "anonymous-storage-FURL" in ann, ann furl = str(ann["anonymous-storage-FURL"]) @@ -294,6 +305,7 @@ class NativeStorageServer: default = self.VERSION_DEFAULTS d = add_version_to_remote_reference(rref, default) d.addCallback(self._got_versioned_service, lp) + d.addCallback(lambda ign: self.broker.check_enough_connected()) d.addErrback(log.err, format="storageclient._got_connection", name=self.get_name(), umid="Sdq3pg") diff --git a/src/allmydata/test/test_checker.py b/src/allmydata/test/test_checker.py index d030c12a..0040a021 100644 --- a/src/allmydata/test/test_checker.py +++ b/src/allmydata/test/test_checker.py @@ -22,7 +22,7 @@ class FakeClient: class WebResultsRendering(unittest.TestCase, WebRenderingMixin): def create_fake_client(self): - sb = StorageFarmBroker(None, True) + sb = StorageFarmBroker(None, True, 0, None) # s.get_name() (the "short description") will be "v0-00000000". # s.get_longname() will include the -long suffix. # s.get_peerid() (i.e. tubid) will be "aaa.." or "777.." or "ceir.." @@ -41,7 +41,7 @@ class WebResultsRendering(unittest.TestCase, WebRenderingMixin): "my-version": "ver", "oldest-supported": "oldest", } - s = NativeStorageServer(key_s, ann) + s = NativeStorageServer(key_s, ann, sb) sb.test_add_server(peerid, s) # XXX: maybe use key_s? c = FakeClient() c.storage_broker = sb diff --git a/src/allmydata/test/test_client.py b/src/allmydata/test/test_client.py index 1819fa4a..0e646edf 100644 --- a/src/allmydata/test/test_client.py +++ b/src/allmydata/test/test_client.py @@ -251,7 +251,7 @@ class Basic(testutil.ReallyEqualMixin, unittest.TestCase): return [ s.get_longname() for s in sb.get_servers_for_psi(key) ] def test_permute(self): - sb = StorageFarmBroker(None, True) + sb = StorageFarmBroker(None, True, 0, None) for k in ["%d" % i for i in range(5)]: ann = {"anonymous-storage-FURL": "pb://abcde@nowhere/fake", "permutation-seed-base32": base32.b2a(k) } @@ -263,7 +263,7 @@ class Basic(testutil.ReallyEqualMixin, unittest.TestCase): self.failUnlessReallyEqual(self._permute(sb, "one"), []) def test_permute_with_preferred(self): - sb = StorageFarmBroker(None, True, ['1','4']) + sb = StorageFarmBroker(None, True, 0, None, ['1','4']) for k in ["%d" % i for i in range(5)]: ann = {"anonymous-storage-FURL": "pb://abcde@nowhere/fake", "permutation-seed-base32": base32.b2a(k) } diff --git a/src/allmydata/test/test_helper.py b/src/allmydata/test/test_helper.py index e1fd54db..7b6c53ee 100644 --- a/src/allmydata/test/test_helper.py +++ b/src/allmydata/test/test_helper.py @@ -116,7 +116,7 @@ class AssistedUpload(unittest.TestCase): timeout = 240 # It takes longer than 120 seconds on Francois's arm box. def setUp(self): self.s = FakeClient() - self.s.storage_broker = StorageFarmBroker(None, True) + self.s.storage_broker = StorageFarmBroker(None, True, 0, None) self.s.secret_holder = client.SecretHolder("lease secret", "converge") self.s.startService() diff --git a/src/allmydata/test/test_mutable.py b/src/allmydata/test/test_mutable.py index c9711c6d..0757182c 100644 --- a/src/allmydata/test/test_mutable.py +++ b/src/allmydata/test/test_mutable.py @@ -234,7 +234,7 @@ def make_storagebroker(s=None, num_peers=10): s = FakeStorage() peerids = [tagged_hash("peerid", "%d" % i)[:20] for i in range(num_peers)] - storage_broker = StorageFarmBroker(None, True) + storage_broker = StorageFarmBroker(None, True, 0, None) for peerid in peerids: fss = FakeStorageServer(peerid, s) ann = {"anonymous-storage-FURL": "pb://%s@nowhere/fake" % base32.b2a(peerid), diff --git a/src/allmydata/test/test_upload.py b/src/allmydata/test/test_upload.py index 090fa3bb..9c82077a 100644 --- a/src/allmydata/test/test_upload.py +++ b/src/allmydata/test/test_upload.py @@ -198,7 +198,7 @@ class FakeClient: mode = dict([i,mode] for i in range(num_servers)) servers = [ ("%20d"%fakeid, FakeStorageServer(mode[fakeid])) for fakeid in range(self.num_servers) ] - self.storage_broker = StorageFarmBroker(None, permute_peers=True) + self.storage_broker = StorageFarmBroker(None, True, 0, None) for (serverid, rref) in servers: ann = {"anonymous-storage-FURL": "pb://%s@nowhere/fake" % base32.b2a(serverid), "permutation-seed-base32": base32.b2a(serverid) } diff --git a/src/allmydata/test/test_web.py b/src/allmydata/test/test_web.py index 723ae7ac..6368f5b9 100644 --- a/src/allmydata/test/test_web.py +++ b/src/allmydata/test/test_web.py @@ -239,7 +239,7 @@ class FakeClient(Client): self._secret_holder = SecretHolder("lease secret", "convergence secret") self.helper = None self.convergence = "some random string" - self.storage_broker = StorageFarmBroker(None, permute_peers=True) + self.storage_broker = StorageFarmBroker(None, True, 0, None) # fake knowledge of another server self.storage_broker.test_add_server("other_nodeid", FakeDisplayableServer("other_nodeid", u"other_nickname \u263B")) -- 2.37.2