Get rid of the 'connected_threshold' argument, too

- use 'when_connected_to', which takes a threshold

src/allmydata/client.py

@@ -131,7 +131,6 @@ 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()
@@ -373,14 +372,13 @@ 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,
-                                              preferred_peers=preferred_peers)
         self.storage_broker = sb
-        self.upload_ready_d = self.storage_broker.when_connected_enough()
+        self.upload_ready_d = sb.when_connected_to(connection_threshold)
 
         # load static server specifications from tahoe.cfg, if any.
         # Not quite ready yet.

src/allmydata/storage_client.py

@@ -55,6 +55,7 @@ from allmydata.util.hashutil import sha1
 # look like?
 #  don't pass signatures: only pass validated blessed-objects
 
+
 class StorageFarmBroker:
     implements(IStorageBroker)
     """I live on the client, and know about storage servers. For each server
@@ -63,12 +64,10 @@ 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, connected_threshold,
-                 preferred_peers=()):
+    def __init__(self, tub, permute_peers, preferred_peers=()):
         self.tub = tub
         assert permute_peers # False not implemented yet
         self.permute_peers = permute_peers
-        self.connected_threshold = connected_threshold
         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,19 +75,38 @@ class StorageFarmBroker:
         # them for it.
         self.servers = {}
         self.introducer_client = None
-        # becomes True if we've ever gone past connected_threshold
-        self._threshold_passed = False
-        self._connected_observer = OneShotObserverList()
+        # the most servers we've connected to at once
+        self._highest_connections = 0
+        # maps int -> OneShotObserverList, where the int is the threshold
+        self._connected_observers = dict()
 
-    def when_connected_enough(self):
+    def when_connected_to(self, threshold):
         """
-        :returns: a Deferred that fires when we have connected to enough
-        servers to cross the "connected_threshold". Once this happens
-        -- even if the number of connected servers goes back below the
-        threshold -- all calls to this method will immediately succeed
-        (i.e. return an already-fired Deferred).
+        :returns: a Deferred that fires if/when our high water mark for
+        number of connected servers becomes (or ever was) above
+        "threshold".
         """
-        return self._connected_observer.when_fired()
+        threshold = int(threshold)
+        if threshold <= 0:
+            raise ValueError("threshold must be positive")
+        if threshold <= self._highest_connections:
+            return defer.succeed(None)
+        try:
+            return self._connected_observers[threshold].when_fired()
+        except KeyError:
+            self._connected_observers[threshold] = OneShotObserverList()
+            return self._connected_observers[threshold].when_fired()
+
+    def check_enough_connected(self):
+        """
+        internal helper
+        """
+        num_servers = len(self.get_connected_servers())
+        self._highest_connections = max(num_servers, self._highest_connections)
+        try:
+            self._connected_observers[num_servers].fire_if_not_fired(None)
+        except KeyError:
+            pass
 
     # these two are used in unit tests
     def test_add_rref(self, serverid, rref, ann):
@@ -135,12 +153,6 @@ class StorageFarmBroker:
         for dsc in self.servers.values():
             dsc.try_to_connect()
 
-    def check_enough_connected(self):
-        if (not self._threshold_passed and
-            len(self.get_connected_servers()) >= self.connected_threshold):
-            self._threshold_passed = True
-            self._connected_observer.fire(None)
-
     def get_servers_for_psi(self, peer_selection_index):
         # return a list of server objects (IServers)
         assert self.permute_peers == True

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, 0)
+        sb = StorageFarmBroker(None, True)
         # 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.."

src/allmydata/test/test_client.py

@@ -236,7 +236,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, 0)
+        sb = StorageFarmBroker(None, True)
         for k in ["%d" % i for i in range(5)]:
             ann = {"anonymous-storage-FURL": "pb://abcde@nowhere/fake",
                    "permutation-seed-base32": base32.b2a(k) }
@@ -248,7 +248,7 @@ class Basic(testutil.ReallyEqualMixin, unittest.TestCase):
         self.failUnlessReallyEqual(self._permute(sb, "one"), [])
 
     def test_permute_with_preferred(self):
-        sb = StorageFarmBroker(None, True, 5, ['1','4'])
+        sb = StorageFarmBroker(None, True, ['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) }

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, 0)
+        self.s.storage_broker = StorageFarmBroker(None, True)
         self.s.secret_holder = client.SecretHolder("lease secret", "converge")
         self.s.startService()
 

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, 0)
+    storage_broker = StorageFarmBroker(None, True)
     for peerid in peerids:
         fss = FakeStorageServer(peerid, s)
         ann = {"anonymous-storage-FURL": "pb://%s@nowhere/fake" % base32.b2a(peerid),

src/allmydata/test/test_storage_client.py

@@ -41,8 +41,8 @@ class TestStorageFarmBroker(unittest.TestCase):
     def test_threshold_reached(self):
         tub = Mock()
         introducer = Mock()
-        broker = StorageFarmBroker(tub, True, 5)
-        done = broker.when_connected_enough()
+        broker = StorageFarmBroker(tub, True)
+        done = broker.when_connected_to(5)
         broker.use_introducer(introducer)
         # subscribes to "storage" to learn of new storage nodes
         subscribe = introducer.mock_calls[0]

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, True, 0)
+        self.storage_broker = StorageFarmBroker(None, permute_peers=True)
         for (serverid, rref) in servers:
             ann = {"anonymous-storage-FURL": "pb://%s@nowhere/fake" % base32.b2a(serverid),
                    "permutation-seed-base32": base32.b2a(serverid) }

src/allmydata/test/test_web.py

@@ -248,7 +248,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, True, 0)
+        self.storage_broker = StorageFarmBroker(None, permute_peers=True)
         # fake knowledge of another server
         self.storage_broker.test_add_server("other_nodeid",
             FakeDisplayableServer(