src/allmydata/immutable/downloader/finder.py

   1
   2 import time
   3 now = time.time
   4 from foolscap.api import eventually
   5 from allmydata.util import base32, log
   6 from twisted.internet import reactor
   7
   8 from share import Share, CommonShare
   9
  10 def incidentally(res, f, *args, **kwargs):
  11     """Add me to a Deferred chain like this:
  12      d.addBoth(incidentally, func, arg)
  13     and I'll behave as if you'd added the following function:
  14      def _(res):
  15          func(arg)
  16          return res
  17     This is useful if you want to execute an expression when the Deferred
  18     fires, but don't care about its value.
  19     """
  20     f(*args, **kwargs)
  21     return res
  22
  23 class RequestToken:
  24     def __init__(self, server):
  25         self.server = server
  26
  27 class ShareFinder:
  28     OVERDUE_TIMEOUT = 10.0
  29
  30     def __init__(self, storage_broker, verifycap, node, download_status,
  31                  logparent=None, max_outstanding_requests=10):
  32         self.running = True # stopped by Share.stop, from Terminator
  33         self.verifycap = verifycap
  34         self._started = False
  35         self._storage_broker = storage_broker
  36         self.share_consumer = self.node = node
  37         self.max_outstanding_requests = max_outstanding_requests
  38         self._hungry = False
  39
  40         self._commonshares = {} # shnum to CommonShare instance
  41         self.pending_requests = set()
  42         self.overdue_requests = set() # subset of pending_requests
  43         self.overdue_timers = {}
  44
  45         self._storage_index = verifycap.storage_index
  46         self._si_prefix = base32.b2a_l(self._storage_index[:8], 60)
  47         self._node_logparent = logparent
  48         self._download_status = download_status
  49         self._lp = log.msg(format="ShareFinder[si=%(si)s] starting",
  50                            si=self._si_prefix,
  51                            level=log.NOISY, parent=logparent, umid="2xjj2A")
  52
  53     def update_num_segments(self):
  54         (numsegs, authoritative) = self.node.get_num_segments()
  55         assert authoritative
  56         for cs in self._commonshares.values():
  57             cs.set_authoritative_num_segments(numsegs)
  58
  59     def start_finding_servers(self):
  60         # don't get servers until somebody uses us: creating the
  61         # ImmutableFileNode should not cause work to happen yet. Test case is
  62         # test_dirnode, which creates us with storage_broker=None
  63         if not self._started:
  64             si = self.verifycap.storage_index
  65             servers = self._storage_broker.get_servers_for_psi(si)
  66             self._servers = iter(servers)
  67             self._started = True
  68
  69     def log(self, *args, **kwargs):
  70         if "parent" not in kwargs:
  71             kwargs["parent"] = self._lp
  72         return log.msg(*args, **kwargs)
  73
  74     def stop(self):
  75         self.running = False
  76         while self.overdue_timers:
  77             req,t = self.overdue_timers.popitem()
  78             t.cancel()
  79
  80     # called by our parent CiphertextDownloader
  81     def hungry(self):
  82         self.log(format="ShareFinder[si=%(si)s] hungry",
  83                  si=self._si_prefix, level=log.NOISY, umid="NywYaQ")
  84         self.start_finding_servers()
  85         self._hungry = True
  86         eventually(self.loop)
  87
  88     # internal methods
  89     def loop(self):
  90         pending_s = ",".join([rt.server.get_name()
  91                               for rt in self.pending_requests]) # sort?
  92         self.log(format="ShareFinder loop: running=%(running)s"
  93                  " hungry=%(hungry)s, pending=%(pending)s",
  94                  running=self.running, hungry=self._hungry, pending=pending_s,
  95                  level=log.NOISY, umid="kRtS4Q")
  96         if not self.running:
  97             return
  98         if not self._hungry:
  99             return
 100
 101         non_overdue = self.pending_requests - self.overdue_requests
 102         if len(non_overdue) >= self.max_outstanding_requests:
 103             # cannot send more requests, must wait for some to retire
 104             return
 105
 106         server = None
 107         try:
 108             if self._servers:
 109                 server = self._servers.next()
 110         except StopIteration:
 111             self._servers = None
 112
 113         if server:
 114             self.send_request(server)
 115             # we loop again to get parallel queries. The check above will
 116             # prevent us from looping forever.
 117             eventually(self.loop)
 118             return
 119
 120         if self.pending_requests:
 121             # no server, but there are still requests in flight: maybe one of
 122             # them will make progress
 123             return
 124
 125         self.log(format="ShareFinder.loop: no_more_shares, ever",
 126                  level=log.UNUSUAL, umid="XjQlzg")
 127         # we've run out of servers (so we can't send any more requests), and
 128         # we have nothing in flight. No further progress can be made. They
 129         # are destined to remain hungry.
 130         eventually(self.share_consumer.no_more_shares)
 131
 132     def send_request(self, server):
 133         req = RequestToken(server)
 134         self.pending_requests.add(req)
 135         lp = self.log(format="sending DYHB to [%(name)s]", name=server.get_name(),
 136                       level=log.NOISY, umid="Io7pyg")
 137         time_sent = now()
 138         d_ev = self._download_status.add_dyhb_request(server, time_sent)
 139         # TODO: get the timer from a Server object, it knows best
 140         self.overdue_timers[req] = reactor.callLater(self.OVERDUE_TIMEOUT,
 141                                                      self.overdue, req)
 142         d = server.get_rref().callRemote("get_buckets", self._storage_index)
 143         d.addBoth(incidentally, self._request_retired, req)
 144         d.addCallbacks(self._got_response, self._got_error,
 145                        callbackArgs=(server, req, d_ev, time_sent, lp),
 146                        errbackArgs=(server, req, d_ev, lp))
 147         d.addErrback(log.err, format="error in send_request",
 148                      level=log.WEIRD, parent=lp, umid="rpdV0w")
 149         d.addCallback(incidentally, eventually, self.loop)
 150
 151     def _request_retired(self, req):
 152         self.pending_requests.discard(req)
 153         self.overdue_requests.discard(req)
 154         if req in self.overdue_timers:
 155             self.overdue_timers[req].cancel()
 156             del self.overdue_timers[req]
 157
 158     def overdue(self, req):
 159         del self.overdue_timers[req]
 160         assert req in self.pending_requests # paranoia, should never be false
 161         self.overdue_requests.add(req)
 162         eventually(self.loop)
 163
 164     def _got_response(self, buckets, server, req, d_ev, time_sent, lp):
 165         shnums = sorted([shnum for shnum in buckets])
 166         time_received = now()
 167         d_ev.finished(shnums, time_received)
 168         dyhb_rtt = time_received - time_sent
 169         if not buckets:
 170             self.log(format="no shares from [%(name)s]", name=server.get_name(),
 171                      level=log.NOISY, parent=lp, umid="U7d4JA")
 172             return
 173         shnums_s = ",".join([str(shnum) for shnum in shnums])
 174         self.log(format="got shnums [%(shnums)s] from [%(name)s]",
 175                  shnums=shnums_s, name=server.get_name(),
 176                  level=log.NOISY, parent=lp, umid="0fcEZw")
 177         shares = []
 178         for shnum, bucket in buckets.iteritems():
 179             s = self._create_share(shnum, bucket, server, dyhb_rtt)
 180             shares.append(s)
 181         self._deliver_shares(shares)
 182
 183     def _create_share(self, shnum, bucket, server, dyhb_rtt):
 184         if shnum in self._commonshares:
 185             cs = self._commonshares[shnum]
 186         else:
 187             numsegs, authoritative = self.node.get_num_segments()
 188             cs = CommonShare(numsegs, self._si_prefix, shnum,
 189                              self._node_logparent)
 190             if authoritative:
 191                 cs.set_authoritative_num_segments(numsegs)
 192             # Share._get_satisfaction is responsible for updating
 193             # CommonShare.set_numsegs after we know the UEB. Alternatives:
 194             #  1: d = self.node.get_num_segments()
 195             #     d.addCallback(cs.got_numsegs)
 196             #   the problem is that the OneShotObserverList I was using
 197             #   inserts an eventual-send between _get_satisfaction's
 198             #   _satisfy_UEB and _satisfy_block_hash_tree, and the
 199             #   CommonShare didn't get the num_segs message before
 200             #   being asked to set block hash values. To resolve this
 201             #   would require an immediate ObserverList instead of
 202             #   an eventual-send -based one
 203             #  2: break _get_satisfaction into Deferred-attached pieces.
 204             #     Yuck.
 205             self._commonshares[shnum] = cs
 206         s = Share(bucket, server, self.verifycap, cs, self.node,
 207                   self._download_status, shnum, dyhb_rtt,
 208                   self._node_logparent)
 209         return s
 210
 211     def _deliver_shares(self, shares):
 212         # they will call hungry() again if they want more
 213         self._hungry = False
 214         shares_s = ",".join([str(sh) for sh in shares])
 215         self.log(format="delivering shares: %s" % shares_s,
 216                  level=log.NOISY, umid="2n1qQw")
 217         eventually(self.share_consumer.got_shares, shares)
 218
 219     def _got_error(self, f, server, req, d_ev, lp):
 220         d_ev.error(now())
 221         self.log(format="got error from [%(name)s]",
 222                  name=server.get_name(), failure=f,
 223                  level=log.UNUSUAL, parent=lp, umid="zUKdCw")
 224
 225