storagewrapper.py

一个多点下载且源码公开的P2P软件

            self.bgalloc_enabled = True
            return True
        if self.alloc_type == 'background':
            self.bgalloc_enabled = True
        if self.blocked_moveout:
            return True
        return False


    def _allocfunc(self):
        while self.holes:
            n = self.holes.pop(0)
            if self.blocked[n]: # assume not self.blocked[index]
                if not self.blocked_movein:
                    self.blocked_holes.append(n)
                    continue
                if not self.places.has_key(n):
                    b = self.blocked_movein.pop(0)
                    oldpos = self._move_piece(b, n)
                    self.places[oldpos] = oldpos
                    return None
            if self.places.has_key(n):
                oldpos = self._move_piece(n, n)
                self.places[oldpos] = oldpos
                return None
            return n
        return None

    def allocfunc(self):
        if self.flag.isSet():
            return None
        
        if self.blocked_moveout:
            self.bgalloc_active = True
            n = self._allocfunc()
            if n is not None:
                if self.blocked_moveout.includes(n):
                    self.blocked_moveout.remove(n)
                    b = n
                else:
                    b = self.blocked_moveout.pop(0)
                oldpos = self._move_piece(b,n)
                self.places[oldpos] = oldpos
            return len(self.holes) / self.numholes

        if self.holes and self.bgalloc_enabled:
            self.bgalloc_active = True
            n = self._allocfunc()
            if n is not None:
                self.write_raw(n, 0, self.alloc_buf[:self._piecelen(n)])
                self.places[n] = n
            return len(self.holes) / self.numholes

        self.bgalloc_active = False
        return None

    def bgalloc(self):
        if self.bgalloc_enabled:
            if not self.holes and not self.blocked_moveout and self.backfunc:
                self.backfunc(self.storage.flush)
                # force a flush whenever the "finish allocation" button is hit
        self.bgalloc_enabled = True
        return False

    def _bgalloc(self):
        self.allocfunc()
        if self.config.get('alloc_rate',0) < 0.1:
            self.config['alloc_rate'] = 0.1
        self.backfunc( self._bgalloc,
              float(self.piece_size)/(self.config['alloc_rate']*1048576) )


    def _waspre(self, piece):
        return self.storage.was_preallocated(piece * self.piece_size, self._piecelen(piece))

    def _piecelen(self, piece):
        if piece < len(self.hashes) - 1:
            return self.piece_size
        else:
            return self.total_length - (piece * self.piece_size)

    def get_amount_left(self):
        return self.amount_left

    def do_I_have_anything(self):
        return self.amount_left < self.total_length

    def _make_inactive(self, index):
        length = self._piecelen(index)
        l = []
        x = 0
        while x + self.request_size < length:
            l.append((x, self.request_size))
            x += self.request_size
        l.append((x, length - x))
        self.inactive_requests[index] = l

    def is_endgame(self):
        return not self.amount_inactive

    def am_I_complete(self):
        return self.amount_obtained == self.amount_desired

    def reset_endgame(self, requestlist):
        for index, begin, length in requestlist:
            self.request_lost(index, begin, length)

    def get_have_list(self):
        return self.have.tostring()

    def get_have_list_cloaked(self):
        if self.have_cloaked_data is None:
            newhave = Bitfield(copyfrom = self.have)
            unhaves = []
            n = min(randrange(2,5),len(self.hashes))    # between 2-4 unless torrent is small
            while len(unhaves) < n:
                unhave = randrange(min(32,len(self.hashes)))    # all in first 4 bytes
                if not unhave in unhaves:
                    unhaves.append(unhave)
                    newhave[unhave] = False
            self.have_cloaked_data = (newhave.tostring(), unhaves)
        return self.have_cloaked_data

    def do_I_have(self, index):
        return self.have[index]

    def do_I_have_requests(self, index):
        return not not self.inactive_requests[index]

    def is_unstarted(self, index):
        return ( not self.have[index] and not self.numactive[index]
                 and not self.dirty.has_key(index) )

    def get_hash(self, index):
        return self.hashes[index]

    def get_stats(self):
        return self.amount_obtained, self.amount_desired

    def new_request(self, index):
        # returns (begin, length)
        if self.inactive_requests[index] == 1:
            self._make_inactive(index)
        self.numactive[index] += 1
        self.stat_active[index] = 1
        if not self.dirty.has_key(index):
            self.stat_new[index] = 1
        rs = self.inactive_requests[index]
#        r = min(rs)
#        rs.remove(r)
        r = rs.pop(0)
        self.amount_inactive -= r[1]
        return r


    def write_raw(self, index, begin, data):
        try:
            self.storage.write(self.piece_size * index + begin, data)
            return True
        except IOError, e:
            self.failed('IO Error: ' + str(e))
            return False


    def _write_to_buffer(self, piece, start, data):
        if not self.write_buf_max:
            return self.write_raw(self.places[piece], start, data)
        self.write_buf_size += len(data)
        while self.write_buf_size > self.write_buf_max:
            old = self.write_buf_list.pop(0)
            if not self._flush_buffer(old, True):
                return False
        if self.write_buf.has_key(piece):
            self.write_buf_list.remove(piece)
        else:
            self.write_buf[piece] = []
        self.write_buf_list.append(piece)
        self.write_buf[piece].append((start,data))
        return True

    def _flush_buffer(self, piece, popped = False):
        if not self.write_buf.has_key(piece):
            return True
        if not popped:
            self.write_buf_list.remove(piece)
        l = self.write_buf[piece]
        del self.write_buf[piece]
        l.sort()
        for start, data in l:
            self.write_buf_size -= len(data)
            if not self.write_raw(self.places[piece], start, data):
                return False
        return True

    def sync(self):
        spots = {}
        for p in self.write_buf_list:
            spots[self.places[p]] = p
        l = spots.keys()
        l.sort()
        for i in l:
            try:
                self._flush_buffer(spots[i])
            except:
                pass
        try:
            self.storage.sync()
        except IOError, e:
            self.failed('IO Error: ' + str(e))
        except OSError, e:
            self.failed('OS Error: ' + str(e))


    def _move_piece(self, index, newpos):
        oldpos = self.places[index]
        if DEBUG:
            print 'moving '+str(index)+' from '+str(oldpos)+' to '+str(newpos)
        assert oldpos != index
        assert oldpos != newpos
        assert index == newpos or not self.places.has_key(newpos)
        old = self.read_raw(oldpos, 0, self._piecelen(index))
        if old is None:
            return -1
        if not self.write_raw(newpos, 0, old):
            return -1
        self.places[index] = newpos
        if self.have[index] and (
                self.triple_check or (self.double_check and index == newpos) ):
            if self.triple_check:
                old.release()
                old = self.read_raw(newpos, 0, self._piecelen(index),
                                    flush_first = True)
                if old is None:
                    return -1
            if sha(old[:]).digest() != self.hashes[index]:
                self.failed('download corrupted; please restart and resume')
                return -1
        old.release()

        if self.blocked[index]:
            self.blocked_moveout.remove(index)
            if self.blocked[newpos]:
                self.blocked_movein.remove(index)
            else:
                self.blocked_movein.add(index)
        else:
            self.blocked_movein.remove(index)
            if self.blocked[newpos]:
                self.blocked_moveout.add(index)
            else:
                self.blocked_moveout.remove(index)
                    
        return oldpos
            
    def _clear_space(self, index):
        h = self.holes.pop(0)
        n = h
        if self.blocked[n]: # assume not self.blocked[index]
            if not self.blocked_movein:
                self.blocked_holes.append(n)
                return True    # repeat
            if not self.places.has_key(n):
                b = self.blocked_movein.pop(0)
                oldpos = self._move_piece(b, n)
                if oldpos < 0:
                    return False
                n = oldpos
        if self.places.has_key(n):
            oldpos = self._move_piece(n, n)
            if oldpos < 0:
                return False
            n = oldpos
        if index == n or index in self.holes:
            if n == h:
                self.write_raw(n, 0, self.alloc_buf[:self._piecelen(n)])
            self.places[index] = n
            if self.blocked[n]:
                # because n may be a spot cleared 10 lines above, it's possible
                # for it to be blocked.  While that spot could be left cleared
                # and a new spot allocated, this condition might occur several
                # times in a row, resulting in a significant amount of disk I/O,
                # delaying the operation of the engine.  Rather than do this,
                # queue the piece to be moved out again, which will be performed
                # by the background allocator, with which data movement is
                # automatically limited.
                self.blocked_moveout.add(index)
            return False
        for p, v in self.places.items():
            if v == index:
                break
        else:
            self.failed('download corrupted; please restart and resume')
            return False
        self._move_piece(p, n)
        self.places[index] = index
        return False


    def piece_came_in(self, index, begin, piece, source = None):
        assert not self.have[index]
        
        if not self.places.has_key(index):
            while self._clear_space(index):
                pass
            if DEBUG:
                print 'new place for '+str(index)+' at '+str(self.places[index])
        if self.flag.isSet():
            return

        if self.failed_pieces.has_key(index):
            old = self.read_raw(self.places[index], begin, len(piece))
            if old is None:
                return True
            if old[:].tostring() != piece:
                try:
                    self.failed_pieces[index][self.download_history[index][begin]] = 1
                except:
                    self.failed_pieces[index][None] = 1
            old.release()
        self.download_history.setdefault(index,{})[begin] = source
        
        if not self._write_to_buffer(index, begin, piece):
            return True
        
        self.amount_obtained += len(piece)
        self.dirty.setdefault(index,[]).append((begin, len(piece)))
        self.numactive[index] -= 1
        assert self.numactive[index] >= 0
        if not self.numactive[index]:
            del self.stat_active[index]
        if self.stat_new.has_key(index):
            del self.stat_new[index]

        if self.inactive_requests[index] or self.numactive[index]:
            return True
        
        del self.dirty[index]
        if not self._flush_buffer(index):
            return True
        length = self._piecelen(index)
        data = self.read_raw(self.places[index], 0, length,
                                 flush_first = self.triple_check)
        if data is None:
            return True
        hash = sha(data[:]).digest()
        data.release()
        if hash != self.hashes[index]:

            self.amount_obtained -= length