| 1 | |
|---|
| 2 | import time |
|---|
| 3 | now = time.time |
|---|
| 4 | from zope.interface import implements |
|---|
| 5 | from twisted.internet import defer |
|---|
| 6 | from twisted.internet.interfaces import IPushProducer |
|---|
| 7 | from foolscap.api import eventually |
|---|
| 8 | from allmydata.util import log |
|---|
| 9 | from allmydata.util.spans import overlap |
|---|
| 10 | |
|---|
| 11 | from common import BadSegmentNumberError, WrongSegmentError, DownloadStopped |
|---|
| 12 | |
|---|
| 13 | class Segmentation: |
|---|
| 14 | """I am responsible for a single offset+size read of the file. I handle |
|---|
| 15 | segmentation: I figure out which segments are necessary, request them |
|---|
| 16 | (from my CiphertextDownloader) in order, and trim the segments down to |
|---|
| 17 | match the offset+size span. I use the Producer/Consumer interface to only |
|---|
| 18 | request one segment at a time. |
|---|
| 19 | """ |
|---|
| 20 | implements(IPushProducer) |
|---|
| 21 | def __init__(self, node, offset, size, consumer, read_ev, logparent=None): |
|---|
| 22 | self._node = node |
|---|
| 23 | self._hungry = True |
|---|
| 24 | self._active_segnum = None |
|---|
| 25 | self._cancel_segment_request = None |
|---|
| 26 | # these are updated as we deliver data. At any given time, we still |
|---|
| 27 | # want to download file[offset:offset+size] |
|---|
| 28 | self._offset = offset |
|---|
| 29 | self._size = size |
|---|
| 30 | assert offset+size <= node._verifycap.size |
|---|
| 31 | self._consumer = consumer |
|---|
| 32 | self._read_ev = read_ev |
|---|
| 33 | self._start_pause = None |
|---|
| 34 | self._lp = logparent |
|---|
| 35 | |
|---|
| 36 | def start(self): |
|---|
| 37 | self._alive = True |
|---|
| 38 | self._deferred = defer.Deferred() |
|---|
| 39 | self._consumer.registerProducer(self, True) |
|---|
| 40 | self._maybe_fetch_next() |
|---|
| 41 | return self._deferred |
|---|
| 42 | |
|---|
| 43 | def _maybe_fetch_next(self): |
|---|
| 44 | if not self._alive or not self._hungry: |
|---|
| 45 | return |
|---|
| 46 | if self._active_segnum is not None: |
|---|
| 47 | return |
|---|
| 48 | self._fetch_next() |
|---|
| 49 | |
|---|
| 50 | def _fetch_next(self): |
|---|
| 51 | if self._size == 0: |
|---|
| 52 | # done! |
|---|
| 53 | self._alive = False |
|---|
| 54 | self._hungry = False |
|---|
| 55 | self._consumer.unregisterProducer() |
|---|
| 56 | self._deferred.callback(self._consumer) |
|---|
| 57 | return |
|---|
| 58 | n = self._node |
|---|
| 59 | have_actual_segment_size = n.segment_size is not None |
|---|
| 60 | guess_s = "" |
|---|
| 61 | if not have_actual_segment_size: |
|---|
| 62 | guess_s = "probably " |
|---|
| 63 | segment_size = n.segment_size or n.guessed_segment_size |
|---|
| 64 | if self._offset == 0: |
|---|
| 65 | # great! we want segment0 for sure |
|---|
| 66 | wanted_segnum = 0 |
|---|
| 67 | else: |
|---|
| 68 | # this might be a guess |
|---|
| 69 | wanted_segnum = self._offset // segment_size |
|---|
| 70 | log.msg(format="_fetch_next(offset=%(offset)d) %(guess)swants segnum=%(segnum)d", |
|---|
| 71 | offset=self._offset, guess=guess_s, segnum=wanted_segnum, |
|---|
| 72 | level=log.NOISY, parent=self._lp, umid="5WfN0w") |
|---|
| 73 | self._active_segnum = wanted_segnum |
|---|
| 74 | d,c = n.get_segment(wanted_segnum, self._lp) |
|---|
| 75 | self._cancel_segment_request = c |
|---|
| 76 | d.addBoth(self._request_retired) |
|---|
| 77 | d.addCallback(self._got_segment, wanted_segnum) |
|---|
| 78 | if not have_actual_segment_size: |
|---|
| 79 | # we can retry once |
|---|
| 80 | d.addErrback(self._retry_bad_segment) |
|---|
| 81 | d.addErrback(self._error) |
|---|
| 82 | |
|---|
| 83 | def _request_retired(self, res): |
|---|
| 84 | self._active_segnum = None |
|---|
| 85 | self._cancel_segment_request = None |
|---|
| 86 | return res |
|---|
| 87 | |
|---|
| 88 | def _got_segment(self, (segment_start,segment,decodetime), wanted_segnum): |
|---|
| 89 | self._cancel_segment_request = None |
|---|
| 90 | # we got file[segment_start:segment_start+len(segment)] |
|---|
| 91 | # we want file[self._offset:self._offset+self._size] |
|---|
| 92 | log.msg(format="Segmentation got data:" |
|---|
| 93 | " want [%(wantstart)d-%(wantend)d)," |
|---|
| 94 | " given [%(segstart)d-%(segend)d), for segnum=%(segnum)d", |
|---|
| 95 | wantstart=self._offset, wantend=self._offset+self._size, |
|---|
| 96 | segstart=segment_start, segend=segment_start+len(segment), |
|---|
| 97 | segnum=wanted_segnum, |
|---|
| 98 | level=log.OPERATIONAL, parent=self._lp, umid="32dHcg") |
|---|
| 99 | |
|---|
| 100 | o = overlap(segment_start, len(segment), self._offset, self._size) |
|---|
| 101 | # the overlap is file[o[0]:o[0]+o[1]] |
|---|
| 102 | if not o or o[0] != self._offset: |
|---|
| 103 | # we didn't get the first byte, so we can't use this segment |
|---|
| 104 | log.msg("Segmentation handed wrong data:" |
|---|
| 105 | " want [%d-%d), given [%d-%d), for segnum=%d," |
|---|
| 106 | " for si=%s" |
|---|
| 107 | % (self._offset, self._offset+self._size, |
|---|
| 108 | segment_start, segment_start+len(segment), |
|---|
| 109 | wanted_segnum, self._node._si_prefix), |
|---|
| 110 | level=log.UNUSUAL, parent=self._lp, umid="STlIiA") |
|---|
| 111 | # we may retry if the segnum we asked was based on a guess |
|---|
| 112 | raise WrongSegmentError("I was given the wrong data.") |
|---|
| 113 | offset_in_segment = self._offset - segment_start |
|---|
| 114 | desired_data = segment[offset_in_segment:offset_in_segment+o[1]] |
|---|
| 115 | |
|---|
| 116 | self._offset += len(desired_data) |
|---|
| 117 | self._size -= len(desired_data) |
|---|
| 118 | self._consumer.write(desired_data) |
|---|
| 119 | # the consumer might call our .pauseProducing() inside that write() |
|---|
| 120 | # call, setting self._hungry=False |
|---|
| 121 | self._read_ev.update(len(desired_data), 0, 0) |
|---|
| 122 | self._maybe_fetch_next() |
|---|
| 123 | |
|---|
| 124 | def _retry_bad_segment(self, f): |
|---|
| 125 | f.trap(WrongSegmentError, BadSegmentNumberError) |
|---|
| 126 | # we guessed the segnum wrong: either one that doesn't overlap with |
|---|
| 127 | # the start of our desired region, or one that's beyond the end of |
|---|
| 128 | # the world. Now that we have the right information, we're allowed to |
|---|
| 129 | # retry once. |
|---|
| 130 | assert self._node.segment_size is not None |
|---|
| 131 | return self._maybe_fetch_next() |
|---|
| 132 | |
|---|
| 133 | def _error(self, f): |
|---|
| 134 | log.msg("Error in Segmentation", failure=f, |
|---|
| 135 | level=log.WEIRD, parent=self._lp, umid="EYlXBg") |
|---|
| 136 | self._alive = False |
|---|
| 137 | self._hungry = False |
|---|
| 138 | self._consumer.unregisterProducer() |
|---|
| 139 | self._deferred.errback(f) |
|---|
| 140 | |
|---|
| 141 | def stopProducing(self): |
|---|
| 142 | self._hungry = False |
|---|
| 143 | self._alive = False |
|---|
| 144 | # cancel any outstanding segment request |
|---|
| 145 | if self._cancel_segment_request: |
|---|
| 146 | self._cancel_segment_request.cancel() |
|---|
| 147 | self._cancel_segment_request = None |
|---|
| 148 | e = DownloadStopped("our Consumer called stopProducing()") |
|---|
| 149 | self._deferred.errback(e) |
|---|
| 150 | |
|---|
| 151 | def pauseProducing(self): |
|---|
| 152 | self._hungry = False |
|---|
| 153 | self._start_pause = now() |
|---|
| 154 | def resumeProducing(self): |
|---|
| 155 | self._hungry = True |
|---|
| 156 | eventually(self._maybe_fetch_next) |
|---|
| 157 | if self._start_pause is not None: |
|---|
| 158 | paused = now() - self._start_pause |
|---|
| 159 | self._read_ev.update(0, 0, paused) |
|---|
| 160 | self._start_pause = None |
|---|
