download.py

BT的原代码

                     self.infohash, errorfunc, resumefile)
            except:
                backthread_exception.append(sys.exc_info())
            self._contfunc()
        thread = threading.Thread(target = hashcheck)
        thread.setDaemon(False)
        self._hashcheck_thread = thread
        thread.start()
        yield None
        self._hashcheck_thread = None
        if resumefile is not None:
            resumefile.close()
        if backthread_exception:
            a, b, c = backthread_exception[0]
            raise a, b, c

        if self._storagewrapper.amount_left == 0:
            self._finished()
        choker = Choker(config, schedfunc, self.finflag.isSet)
        upmeasure = Measure(config['max_rate_period'])
        upmeasure_seedtime = Measure(config['max_rate_period_seedtime'])
        downmeasure = Measure(config['max_rate_period'])
        self._upmeasure = upmeasure
        self._downmeasure = downmeasure
        self._ratemeasure = RateMeasure(self._storagewrapper.
                                        amount_left_with_partials)
        picker = PiecePicker(len(metainfo.hashes), config)
        for i in xrange(len(metainfo.hashes)):
            if self._storagewrapper.do_I_have(i):
                picker.complete(i)
        for i in self._storagewrapper.stat_dirty:
            picker.requested(i)
        def kickpeer(connection):
            def kick():
                connection.close()
            schedfunc(kick, 0)
        def banpeer(ip):
            self._encoder.ban(ip)
        downloader = Downloader(config, self._storagewrapper, picker,
            len(metainfo.hashes), downmeasure, self._ratemeasure.data_came_in,
                                kickpeer, banpeer)
        def make_upload(connection):
            return Upload(connection, self._ratelimiter, upmeasure,
                        upmeasure_seedtime, choker, self._storagewrapper,
                        config['max_slice_length'], config['max_rate_period'])
        self._encoder = Encoder(make_upload, downloader, choker,
                     len(metainfo.hashes), self._ratelimiter, self._rawserver,
                     config, myid, schedfunc, self.infohash, self)
        self.reported_port = self.config['forwarded_port']
        if not self.reported_port:
            self.reported_port = self._singleport_listener.get_port()
            self.reserved_ports.append(self.reported_port)
        self._singleport_listener.add_torrent(self.infohash, self._encoder)
        self._listening = True
        self._rerequest = Rerequester(metainfo.announce, config,
            schedfunc, self._encoder.how_many_connections,
            self._encoder.start_connection, externalsched,
            self._storagewrapper.get_amount_left, upmeasure.get_total,
            downmeasure.get_total, self.reported_port, myid,
            self.infohash, self._error, self.finflag, upmeasure.get_rate,
            downmeasure.get_rate, self._encoder.ever_got_incoming,
            self.internal_shutdown, self._announce_done)
        self._statuscollecter = DownloaderFeedback(choker, upmeasure.get_rate,
            upmeasure_seedtime.get_rate, downmeasure.get_rate,
            upmeasure.get_total, downmeasure.get_total,
            self._ratemeasure.get_time_left, self._ratemeasure.get_size_left,
            self.total_bytes, self.finflag, downloader, self._myfiles)

        self._announced = True
        self._rerequest.begin()
        self.started = True
        if not self.finflag.isSet():
            self._activity = ('downloading', 0)
        self.feedback.started(self)

    def got_exception(self, e):
        is_external = False
        if isinstance(e, BTShutdown):
            self._error(ERROR, str(e))
            is_external = True
        elif isinstance(e, BTFailure):
            self._error(CRITICAL, str(e))
            self._activity = ('download failed: ' + str(e), 0)
        elif isinstance(e, IOError):
            msg = 'IO Error ' + str(e)
            if e.errno == errno.ENOSPC:
                msg = 'IO Error: No space left on disk, '\
                      'or cannot create a file that large:' + str(e)
            self._error(CRITICAL, msg)
            self._activity = ('killed by IO error: ' + str(e), 0)
        elif isinstance(e, OSError):
            self._error(CRITICAL, 'OS Error ' + str(e))
            self._activity = ('killed by OS error: ' + str(e), 0)
        else:
            data = StringIO()
            print_exc(file=data)
            self._error(CRITICAL, data.getvalue(), True)
            self._activity = ('killed by internal exception: ' + str(e), 0)
        try:
            self._close()
        except Exception, e:
            self._error(ERROR, 'Additional error when closing down due to '
                        'error: ' + str(e))
        if is_external:
            self.feedback.failed(self, True)
            return
        if self.config['data_dir'] and self._storage is not None:
            filename = os.path.join(self.config['data_dir'], 'resume',
                                    self.infohash.encode('hex'))
            if os.path.exists(filename):
                try:
                    os.remove(filename)
                except Exception, e:
                    self._error(WARNING, 'Could not remove fastresume file '
                                'after failure:' + str(e))
        self.feedback.failed(self, False)

    def _finished(self):
        self.finflag.set()
        # Call self._storage.close() to flush buffers and change files to
        # read-only mode (when they're possibly reopened). Let exceptions
        # from self._storage.close() kill the torrent since files might not
        # be correct on disk if file.close() failed.
        self._storage.close()
        # If we haven't announced yet, normal first announce done later will
        # tell the tracker about seed status.
        self.is_seed = True
        if self._announced:
            self._rerequest.announce_finish()
        self._activity = ('seeding', 1)
        if self.config['check_hashes']:
            self._save_fastresume(True)
        self.feedback.finished(self)

    def _save_fastresume(self, on_finish=False):
        if not on_finish and (self.finflag.isSet() or not self.started):
            return
        if not self.config['data_dir']:
            return
        if on_finish:    # self._ratemeasure might not exist yet
            amount_done = self.total_bytes
        else:
            amount_done = self.total_bytes - self._ratemeasure.get_size_left()
        filename = os.path.join(self.config['data_dir'], 'resume',
                                self.infohash.encode('hex'))
        resumefile = None
        try:
            resumefile = file(filename, 'wb')
            self._storage.write_fastresume(resumefile, amount_done)
            self._storagewrapper.write_fastresume(resumefile)
            resumefile.close()
        except Exception, e:
            self._error(WARNING, 'Could not write fastresume data: ' + str(e))
            if resumefile is not None:
                resumefile.close()

    def shutdown(self):
        if self.closed:
            return
        try:
            self._close()
            self._save_fastresume()
            self._activity = ('shut down', 0)
        except Exception, e:
            self.got_exception(e)

    def internal_shutdown(self, level, text):
        # This is only called when announce fails with no peers,
        # don't try to announce again telling we're leaving the torrent
        self._announced = False
        self._error(level, text)
        self.shutdown()
        self.feedback.failed(self, True)

    def _close(self):
        if self.closed:
            return
        self.closed = True
        self._rawserver.remove_context(self)
        self._doneflag.set()
        if self._announced:
            self._rerequest.announce_stop()
            self._rerequest.cleanup()
        if self._hashcheck_thread is not None:
            self._hashcheck_thread.join() # should die soon after doneflag set
        if self._myfiles is not None:
            self._filepool.remove_files(self._myfiles)
        if self._listening:
            self._singleport_listener.remove_torrent(self.infohash)
        for port in self.reserved_ports:
            self._singleport_listener.release_port(port)
        if self._encoder is not None:
            self._encoder.close_connections()
        if self._storage is not None:
            self._storage.close()
        self._ratelimiter.clean_closed()
        self._rawserver.add_task(gc.collect, 0, None)

    def get_status(self, spew = False, fileinfo=False):
        if self.started and not self.closed:
            r = self._statuscollecter.get_statistics(spew, fileinfo)
            r['activity'] = self._activity[0]
        else:
            r = dict(zip(('activity', 'fractionDone'), self._activity))
        return r

    def get_total_transfer(self):
        if self._upmeasure is None:
            return (0, 0)
        return (self._upmeasure.get_total(), self._downmeasure.get_total())

    def set_option(self, option, value):
        if self.closed:
            return
        if option not in self.config or self.config[option] == value:
            return
        if option not in 'min_uploads max_uploads max_initiate max_allow_in '\
           'data_dir ip max_upload_rate retaliate_to_garbled_data'.split():
            return
        # max_upload_rate doesn't affect upload rate here, just auto uploads
        self.config[option] = value
        self._set_auto_uploads()

    def change_port(self):
        if not self._listening:
            return
        r = self.config['forwarded_port']
        if r:
            for port in self.reserved_ports:
                self._singleport_listener.release_port(port)
            del self.reserved_ports[:]
            if self.reported_port == r:
                return
        elif self._singleport_listener.port != self.reported_port:
            r = self._singleport_listener.get_port()
            self.reserved_ports.append(r)
        else:
            return
        self.reported_port = r
        myid = self._make_id()
        self._encoder.my_id = myid
        self._rerequest.change_port(myid, r)

    def _announce_done(self):
        for port in self.reserved_ports[:-1]:
            self._singleport_listener.release_port(port)
        del self.reserved_ports[:-1]

    def _make_id(self):
        myid = 'M' + version.split()[0].replace('.', '-')
        myid = myid + ('-' * (8-len(myid)))+sha(repr(time())+ ' ' +
                                     str(getpid())).digest()[-6:].encode('hex')
        return myid

    def _set_auto_uploads(self):
        uploads = self.config['max_uploads']
        rate = self.config['max_upload_rate']
        if uploads > 0:
            pass
        elif rate <= 0:
            uploads = 7 # unlimited, just guess something here...
        elif rate < 9:
            uploads = 2
        elif rate < 15:
            uploads = 3
        elif rate < 42:
            uploads = 4
        else:
            uploads = int(sqrt(rate * .6))
        self.config['max_uploads_internal'] = uploads

    def _error(self, level, text, exception=False):
        self.errors.append((time(), level, text))
        if exception:
            self.feedback.exception(self, text)
        else:
            self.feedback.error(self, level, text)