[01;34mhunk[00m ./src/allmydata/immutable/downloader/fetcher.py 74
eventually(self.loop)
def no_more_shares(self):
+ log.msg("aaa %s.no_more_shares()" % (self,))
# ShareFinder tells us it's reached the end of its list
self._no_more_shares = True
eventually(self.loop)
[01;34mhunk[00m ./src/allmydata/immutable/downloader/fetcher.py 104
self._node.fetch_failed(self, f)
return
- #print "LOOP", self._blocks.keys(), "active:", self._active_share_map, "overdue:", self._overdue_share_map, "unused:", self._shares
+ log.msg( "aaa %s.loop(); blocks: %s, active: %s, overdue: %s, unused: %s" % (self, self._blocks.keys(), self._active_share_map, self._overdue_share_map, self._shares))
# Should we sent out more requests?
while len(set(self._blocks.keys())
| set(self._active_share_map.keys())
[01;34mhunk[00m ./src/allmydata/immutable/downloader/fetcher.py 130
# progress
self._ask_for_more_shares()
if self._no_more_shares:
+ self._no_shares_error() # this calls self.stop()
# But there are no more shares to be had. If we're going to
# succeed, it will be with the shares we've already seen.
# Will they be enough?
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 52
self._lp = log.msg(format="ShareFinder[si=%(si)s] starting",
si=self._si_prefix,
level=log.NOISY, parent=logparent, umid="2xjj2A")
+ log.msg("xxx %s.__init__(%s, %s, %s, %s, %s, %s)" % (self, storage_broker, verifycap, node, download_status, logparent, max_outstanding_requests))
def update_num_segments(self):
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 55
+ log.msg("xxx %s.update_num_segments()" % (self,))
(numsegs, authoritative) = self.node.get_num_segments()
assert authoritative
for cs in self._commonshares.values():
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 62
cs.set_authoritative_num_segments(numsegs)
def start_finding_servers(self):
+ log.msg("xxx %s.start_finding_servers()" % (self,))
# don't get servers until somebody uses us: creating the
# ImmutableFileNode should not cause work to happen yet. Test case is
# test_dirnode, which creates us with storage_broker=None
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 78
return log.msg(*args, **kwargs)
def stop(self):
+ log.msg("xxx %s.stop()" % (self,))
self.running = False
while self.overdue_timers:
req,t = self.overdue_timers.popitem()
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 86
# called by our parent CiphertextDownloader
def hungry(self):
+ log.msg("xxx %s.hungry()" % (self,))
self.log(format="ShareFinder[si=%(si)s] hungry",
si=self._si_prefix, level=log.NOISY, umid="NywYaQ")
self.start_finding_servers()
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 95
# internal methods
def loop(self):
+ log.msg("xxx %s.loop()" % (self,))
pending_s = ",".join([idlib.shortnodeid_b2a(rt.peerid)
for rt in self.pending_requests]) # sort?
self.log(format="ShareFinder loop: running=%(running)s"
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 123
self.send_request(server)
# we loop again to get parallel queries. The check above will
# prevent us from looping forever.
+ log.msg("xxx %s.loop() => loop again to get parallel queries" % (self,))
eventually(self.loop)
return
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 140
self.share_consumer.no_more_shares()
def send_request(self, server):
+ log.msg("xxx %s.send_request(%s)" % (self, server))
peerid, rref = server
req = RequestToken(peerid)
self.pending_requests.add(req)
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 163
d.addCallback(incidentally, eventually, self.loop)
def _request_retired(self, req):
+ log.msg("xxx %s._request_retired(%s)" % (self, req))
self.pending_requests.discard(req)
self.overdue_requests.discard(req)
if req in self.overdue_timers:
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 171
del self.overdue_timers[req]
def overdue(self, req):
+ log.msg("xxx %s.overdue(%s)" % (self, req))
del self.overdue_timers[req]
assert req in self.pending_requests # paranoia, should never be false
self.overdue_requests.add(req)
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 179
def _got_response(self, buckets, server_version, peerid, req, d_ev,
time_sent, lp):
+ log.msg("xxx %s._got_response(%s, %s, %s, %s, %s, %s, %s)" % (self, buckets, server_version, peerid, req, d_ev, time_sent, lp))
shnums = sorted([shnum for shnum in buckets])
time_received = now()
d_ev.finished(shnums, time_received)
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 201
self._deliver_shares(shares)
def _create_share(self, shnum, bucket, server_version, peerid, dyhb_rtt):
+ log.msg("xxx %s._create_share(%s, %s, %s, %s, %s)" % (self, shnum, bucket, server_version, peerid, dyhb_rtt))
if shnum in self._commonshares:
cs = self._commonshares[shnum]
else:
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 230
return s
def _deliver_shares(self, shares):
+ log.msg("xxx %s._deliver_shares(%s)" % (self, shares))
# they will call hungry() again if they want more
self._hungry = False
shares_s = ",".join([str(sh) for sh in shares])
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 239
eventually(self.share_consumer.got_shares, shares)
def _got_error(self, f, peerid, req, d_ev, lp):
+ log.msg("xxx %s._got_error(%s, %s, %s, %s, %s)" % (self, f, peerid, req, d_ev, lp))
d_ev.finished("error", now())
self.log(format="got error from [%(peerid)s]",
peerid=idlib.shortnodeid_b2a(peerid), failure=f,
[01;34mhunk[00m ./src/allmydata/immutable/downloader/finder.py 244
level=log.UNUSUAL, parent=lp, umid="zUKdCw")
-
-
[01;34mhunk[00m ./src/allmydata/immutable/downloader/node.py 224
# called by our child ShareFinder
def got_shares(self, shares):
+ log.msg("xxx %s.got_shares(%s)" % (self, shares))
self._shares.update(shares)
if self._active_segment:
self._active_segment.add_shares(shares)
[01;34mhunk[00m ./src/allmydata/immutable/downloader/node.py 229
def no_more_shares(self):
+ log.msg("xxx %s.no_more_shares() ; _active_segment: %s" % (self, self._active_segment))
self._no_more_shares = True
if self._active_segment:
self._active_segment.no_more_shares()
[01;34mhunk[00m ./src/allmydata/immutable/downloader/share.py 86
self._requested_blocks = [] # (segnum, set(observer2..))
ver = server_version["http://allmydata.org/tahoe/protocols/storage/v1"]
+ log.msg( "zzz ver: %r" % (ver,))
self._overrun_ok = ver["tolerates-immutable-read-overrun"]
# If _overrun_ok and we guess the offsets correctly, we can get
# everything in one RTT. If _overrun_ok and we guess wrong, we might
[01;34mhunk[00m ./src/allmydata/test/test_immutable.py 4
from allmydata.test import common
from allmydata.interfaces import NotEnoughSharesError
from allmydata.util.consumer import download_to_data
+from allmydata import uri
from twisted.internet import defer
from twisted.trial import unittest
[01;34mhunk[00m ./src/allmydata/test/test_immutable.py 7
+from allmydata.immutable.downloader.common import COMPLETE
import random
[01;34mhunk[00m ./src/allmydata/test/test_immutable.py 10
+from foolscap.api import eventually
+from allmydata.util import log
+
+from allmydata.immutable.downloader import fetcher, finder, share
+
+import mock
+
+class MockNode(object):
+ def __init__(self, check_reneging, check_fetch_failed):
+ self.got = 0
+ self.finished_d = defer.Deferred()
+ self.segment_size = 78
+ self.guessed_segment_size = 78
+ self._no_more_shares = False
+ self.check_reneging = check_reneging
+ self.check_fetch_failed = check_fetch_failed
+ self._si_prefix='aa'
+ self.have_UEB = True
+ self.share_hash_tree = mock.Mock()
+ self.share_hash_tree.needed_hashes.return_value = False
+ self.on_want_more_shares = None
+
+ def when_finished(self):
+ return self.finished_d
+ def get_num_segments(self):
+ return (5, True)
+ def _calculate_sizes(self, guessed_segment_size):
+ return {'block_size': 4, 'num_segments': 5}
+ def no_more_shares(self):
+ self._no_more_shares = True
+ def got_shares(self, shares):
+ if self.check_reneging:
+ if self._no_more_shares:
+ self.finished_d.errback(unittest.FailTest("The node was told by the share finder that it is destined to remain hungry, then was given another share."))
+ return
+ self.got += len(shares)
+ log.msg("yyy 3 %s.got_shares(%s) got: %s" % (self, shares, self.got))
+ if self.got == 3:
+ self.finished_d.callback(True)
+ def get_desired_ciphertext_hashes(self, *args, **kwargs):
+ return iter([])
+ def fetch_failed(self, *args, **kwargs):
+ if self.check_fetch_failed:
+ if self.finished_d:
+ self.finished_d.errback(unittest.FailTest("The node was told by the segment fetcher that the download failed."))
+ self.finished_d = None
+ def want_more_shares(self):
+ if self.on_want_more_shares:
+ self.on_want_more_shares()
+ def process_blocks(self, *args, **kwargs):
+ if self.finished_d:
+ self.finished_d.callback(None)
+
+class TestSegmentFetcher(unittest.TestCase):
+ def test_be_satisfied_with_ill_distributed_shares(self):
+ mocknode = MockNode(check_reneging=False, check_fetch_failed=True)
+ sf = fetcher.SegmentFetcher(mocknode, 1, 3, None)
+ mocknode.on_want_more_shares = lambda: sf.no_more_shares
+
+ rcap = uri.CHKFileURI('a'*32, 'a'*32, 3, 99, 100)
+ vcap = rcap.get_verify_cap()
+ server_version = {
+ 'http://allmydata.org/tahoe/protocols/storage/v1': {
+ "tolerates-immutable-read-overrun": True
+ }
+ }
+ ss = []
+ mockcommonshare = mock.Mock()
+ mockcommonshare.get_desired_block_hashes.return_value = iter([])
+
+ mockds0 = mock.Mock()
+ share0 = share.Share(mock.Mock(), server_version, vcap,
+ mockcommonshare, mocknode, mockds0, 's1', 0, 0,
+ None)
+ mockds0.add_request_sent.side_effect = \
+ eventually(sf._block_request_activity, share0, 0, COMPLETE)
+ ss.append(share0)
+
+ mockds1 = mock.Mock()
+ share1 = share.Share(mock.Mock(), server_version, vcap,
+ mockcommonshare, mocknode, mockds1, 's1', 1, 0,
+ None)
+ mockds1.add_request_sent.side_effect = \
+ eventually(sf._block_request_activity, share1, 1, COMPLETE)
+ ss.append(share1)
+
+ mockds2 = mock.Mock()
+ share2 = share.Share(mock.Mock(), server_version, vcap,
+ mockcommonshare, mocknode, mockds2, 's2', 2, 0,
+ None)
+ mockds2.add_request_sent.side_effect = \
+ eventually(sf._block_request_activity, share2, 2, COMPLETE)
+ ss.append(share2)
+
+ sf.add_shares(ss)
+ sf.no_more_shares()
+
+ return mocknode.when_finished()
+
+class TestShareFinder(unittest.TestCase):
+ def test_no_reneging_on_no_more_shares_ever(self):
+ # ticket #1191
+
+ # Suppose that K=3 and you send two DYHB requests, the first
+ # response offers two shares, and then the last offers one
+ # share. If you tell your share consumer "no more shares,
+ # ever", and then immediately tell them "oh, and here's
+ # another share", then you lose.
+
+ rcap = uri.CHKFileURI('a'*32, 'a'*32, 3, 99, 100)
+ vcap = rcap.get_verify_cap()
+
+ class MockServer(object):
+ def __init__(self, buckets):
+ self.version = {
+ 'http://allmydata.org/tahoe/protocols/storage/v1': {
+ "tolerates-immutable-read-overrun": True
+ }
+ }
+ self.buckets = buckets
+ self.d = defer.Deferred()
+ self.s = None
+ def callRemote(self, methname, *args, **kwargs):
+ log.msg("yyy 2 %s.callRemote(%s, %s, %s)" % (self, methname, args, kwargs))
+ d = defer.Deferred()
+
+ # Even after the 3rd answer we're still hungry because
+ # we're interested in finding a 3rd server so we don't
+ # have to download more than one share from the first
+ # server. This is actually necessary to trigger the
+ # bug.
+ def _give_buckets_and_hunger_again():
+ d.callback(self.buckets)
+ self.s.hungry()
+ eventually(_give_buckets_and_hunger_again)
+ return d
+
+ mockserver1 = MockServer({1: mock.Mock(), 2: mock.Mock()})
+ mockserver2 = MockServer({})
+ mockserver3 = MockServer({3: mock.Mock()})
+ mockstoragebroker = mock.Mock()
+ mockstoragebroker.get_servers_for_index.return_value = [ ('ms1', mockserver1), ('ms2', mockserver2), ('ms3', mockserver3), ]
+ mockdownloadstatus = mock.Mock()
+ mocknode = MockNode(check_reneging=True, check_fetch_failed=True)
+
+ s = finder.ShareFinder(mockstoragebroker, vcap, mocknode, mockdownloadstatus)
+
+ mockserver1.s = s
+ mockserver2.s = s
+ mockserver3.s = s
+
+ s.hungry()
+
+ return mocknode.when_finished()
+
class Test(common.ShareManglingMixin, common.ShouldFailMixin, unittest.TestCase):
def test_test_code(self):
# The following process of stashing the shares, running