Teach StorageFarmBroker to fire a deferred when a connection threshold is reached...

Signed-off-by: Daira Hopwood <daira@jacaranda.org>

diff --git a/src/allmydata/client.py b/src/allmydata/client.py
index 4e3785ea33827b5cb1255a400c799957d25d8e5f..0a6c1eb60e0e36347f13a97b4e5bbc2f4ac87e8e 100644 (file)
--- 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.upload_ready_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.upload_ready_d, preferred_peers=preferred_peers)
         self.storage_broker = sb
 
         # load static server specifications from tahoe.cfg, if any.
@@ -505,6 +511,9 @@ class Client(node.Node, pollmixin.PollMixin):
                 s = drop_upload.DropUploader(self, upload_dircap, local_dir, dbfile)
                 s.setServiceParent(self)
                 s.startService()
+
+                # start processing the upload queue when we've connected to enough servers
+                self.upload_ready_d.addCallback(s.upload_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 7796efd8f860262a1d26af608c71373c73b7b918..ea561d9f287f720bc452a0f4e4af7c84d7dd6d77 100644 (file)
--- a/src/allmydata/frontends/drop_upload.py
+++ b/src/allmydata/frontends/drop_upload.py
@@ -30,6 +30,8 @@ class DropUploader(service.MultiService):
         self._local_path = to_filepath(self._local_dir)
         self._dbfile = dbfile
 
+        self.is_upload_ready = False
+
         if inotify is None:
             if sys.platform == "win32":
                 from allmydata.windows import inotify
@@ -84,6 +86,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 81dc00427cac452f5e2fa3d3d52841fcc23c7b37..8b6cb4e2c9bd9171c916321fe77e616e2bae2928 100644 (file)
--- 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
@@ -190,9 +200,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"])
@@ -295,6 +306,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 d030c12a40176b225a574008e09b31732be57728..0040a021af78d5bdd24feb6c49cb8f0e8922ddf1 100644 (file)
--- 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 f2b7a7a23b2a6317bcff946f30a7e57d82a0a1d0..bf842f8a19e407099111e068be2963b28deb63cd 100644 (file)
--- a/src/allmydata/test/test_client.py
+++ b/src/allmydata/test/test_client.py
@@ -251,7 +251,7 @@ class Basic(testutil.ReallyEqualMixin, testutil.NonASCIIPathMixin, unittest.Test
         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) }

diff --git a/src/allmydata/test/test_helper.py b/src/allmydata/test/test_helper.py
index e1fd54dbe853fde5e4bccfad0a49f7ea0f6dd7b4..7b6c53eee2c9a749fb03265629120eefaba99e35 100644 (file)
--- 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 c9711c6dc3ac5287a6406f746667ce0d5cc18bc5..0757182c88d1ec445e0240e74696bb6f0351cba6 100644 (file)
--- 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 090fa3bbe554773ea118a21dc0fc3a0b5bbca81e..9c82077af0f4cc4d1860155d6586d9081b6b45d2 100644 (file)
--- 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 5b151ecb6b6ce466c7627c07dbfcd9621104bd23..12e76e340c9b98c12cfe5a8541f067086819688f 100644 (file)
--- a/src/allmydata/test/test_web.py
+++ b/src/allmydata/test/test_web.py
@@ -246,7 +246,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(