torrent.py

bittorrent source by python. please enjoy

        except:
            e_str = str(e)
        if isinstance(e, BTFailure):
            self._activity = ( _("download failed: ") + e_str, 0)
        elif isinstance(e, IOError):
            if e.errno == errno.ENOSPC:
                msg = _("IO Error: No space left on disk, "
                        "or cannot create a file that large")
            self._activity = (_("killed by IO error: ") + e_str, 0)
        elif isinstance(e, OSError):
            self._activity = (_("killed by OS error: ") + e_str, 0)
        else:
            self._activity = (_("killed by internal exception: ") + e_str, 0)

        if isinstance(e, UserFailure):
            self.logger.log(severity, e_str )
        else:
            self.logger.log(severity, msg, exc_info=(type, e, stack))
        # steve wanted this too
        # Dave doesn't want it.
        #traceback.print_exception(type, e, stack, file=sys.stdout)

        self.failed(cannot_shutdown)

    def failed(self, cannot_shutdown=False):
        if cannot_shutdown:
            self.state = "failed"
        else:
            try:
                # this could complete later. sorry that's just the way it is.
                df = self.shutdown()
                def cb(*a):
                    self.state = "failed"
                df.addCallbacks(cb, cb)
            except:
                self.logger.exception(_("Additional error when closing down due to "
                                        "error: "))
        self.feedback.failed(self)

    def _error(self, level, text, exception=False, exc_info=None):
        if level > logging.WARNING:
            self.logger.log(level,
                            _('Error regarding "%s":\n')%self.metainfo.name + text,
                            exc_info=exc_info)
        if exception:
            self.feedback.exception(self, text)
        else:
            self.feedback.error(self, level, text)

    def finished(self, policy="auto", finished_this_session=True):
        assert self.state == "running"
        # because _finished() calls shutdown(), which invalidates the torrent
        # context, we need to use _rawserver.add_task directly here
        df = launch_coroutine(_wrap_task(self._rawserver.add_task), self._finished, policy=policy, finished_this_session=finished_this_session)
        df.addErrback(lambda e : self.got_exception(*e))
        return df

    def _finished(self, policy="auto", finished_this_session=True):
        if self.state != "running":
            return

        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()

        # don't bother trailing off the rate when we know we're done downloading
        self._downmeasure.rate = 0.0

        # If we haven't announced yet, normal first announce done later will
        # tell the tracker about seed status.
        if self._announced:
            self._rerequest.announce_finish()
        self._activity = (_("seeding"), 1)
        if self.config['check_hashes']:
            self._save_fastresume()

        # the old policy applied to downloading -- now that we are finished,
        # optionally reset it
        self.policy = policy

        self.feedback.finishing(self)

        config = self.config

        if finished_this_session:
            self.finished_this_session = True

        def move(working_path, destination_path):
            # this function is called from another thread, so don't do anything
            # that isn't thread safe in here

            self.logger.debug("deleting any file that might be in the way")
            try:
                os.remove(destination_path)
                self.logger.debug("successfully deleted file " +
                                  destination_path)
            except Exception, e:
                if os.path.exists(destination_path):
                    self.logger.debug(unicode(e.args[0]))

            self.logger.debug("deleting any directory that might be in the way")
            try:
                shutil.rmtree(destination_path)
                self.logger.debug("successfully deleted directory " +
                                  destination_path)
            except Exception, e:
                if os.path.exists(destination_path):
                    self.logger.debug(unicode(e.args[0]))

            self.logger.debug("actually moving file")
            shutil.move(working_path, destination_path)
            self.logger.debug("returned from move")


        if self.working_path != self.destination_path:
##            self.logger.debug("torrent finishing: shutting down, moving file, and restarting")
##            df = self.shutdown()
##            yield df
##            df.getResult()
            self.logger.debug("torrent finishing: pausing, moving file, and restarting")
            self.stop_download(pause=True)
            self._unregister_files()

            self.logger.debug("successfully paused torrent, moving file")

            self.state = "finishing"
            df = ThreadedDeferred(_wrap_task(self._rawserver.external_add_task),
                                  move, self.working_path, self.destination_path)
            yield df
            df.getResult()

            self.logger.debug("moved file, restarting")

            assert self.state == "finishing"
            self.working_path = self.destination_path
##            self.state = "created"
##            df = self.initialize()
##            yield df
##            df.getResult()
            self.completed = True
            self.feedback.finished(self)

            self.state = "initializing"
            self._register_files()
            df = self._storage.initialize(self.working_path,
                                          zip(self._myfiles,
                                              self.metainfo.sizes))
            yield df
            df.getResult()

            # so we store new path names
            self._storagewrapper.fastresume_dirty = True
            self._statuscollector.files = self._myfiles

            self.state = "initialized"
            self.logger.debug("attempting restart")
            self.start_download()
            self.logger.debug("re-started torrent")
        else:
            self.completed = True
            self.feedback.finished(self)

        self._dump_torrent_config()

    def fastresume_file_path(self):
        # HEREDAVE: should probably be self.data_dir?
        return os.path.join(self.config['data_dir'], 'resume',
                            self.infohash.encode('hex'))

    def config_file_path(self):
        return os.path.join(self.data_dir, 'torrents',
                            self.metainfo.infohash.encode('hex'))

    def _periodic_save_fastresume(self):
        self._save_fastresume()
        if not self.finflag.isSet():
            self.add_task(30, self._periodic_save_fastresume)

    def _save_fastresume(self):
        if not self.is_initialized():
            return
        # HEREDAVE: should probably be self.data_dir?
        if not self.config['data_dir']:
            return
        filename = self.fastresume_file_path()
        if os.path.exists(filename) and not self._storagewrapper.fastresume_dirty:
            return
        resumefile = None
        try:
            resumefile = file(filename, 'wb')
            self._storagewrapper.write_fastresume(resumefile)
            resumefile.close()
        except Exception, e:
            self._error(logging.WARNING, _("Could not write fastresume data: ") +
                        unicode(e.args[0]))
            if resumefile is not None:
                resumefile.close()

    def _dump_torrent_config(self):
        d = self.config.getDict()
##        nd = {}
##        for k,v in d.iteritems():
##            # can't bencode floats!
##            if not isinstance(v, float):
##                if isinstance(v, unicode):
##                    # FIXME -- what is the right thing to do here?
##                    v = v.encode('utf8')
##                nd[k] = v
##        s = bencode(nd)
        s = cPickle.dumps(d)
        path = self.config_file_path()
        f = file(path+'.new', 'wb')
        f.write(s)
        f.close()
        shutil.move(path+'.new', path)

    def remove_state_files(self, del_files=False):
        assert self.state == "created"

        try:
            os.remove(self.config_file_path())
        except Exception, e:
            self.logger.debug("error removing config file: %s", unicode(e.args[0]))

        try:
            os.remove(self.fastresume_file_path())
        except Exception, e:
            self.logger.debug("error removing fastresume file: %s", unicode(e.args[0]))

        if del_files:
            try:
                for file in self._last_myfiles:
                    try:
                        os.remove(file)
                    except OSError:
                        pass
                    d, f = os.path.split(file)
                    try:
                        os.rmdir(d)
                    except OSError:
                        pass

                try:
                    os.rmdir(self.working_path)
                except OSError:
                    pass
            except Exception, e:
                self.logger.debug("error removing incomplete files: %s", unicode(e.args[0]))

    def get_downrate(self):
        if self.is_running():
            return self._downmeasure.get_rate()

    def get_uprate(self):
        if self.is_running():
            return self._upmeasure.get_rate()

    def get_rates(self):
        return (self.get_uprate(), self.get_downrate())

    def get_downtotal(self):
        if self.is_running():
            return self._downmeasure.get_total()

    def get_uptotal(self):
        if self.is_running():
            return self._upmeasure.get_total()

    def get_percent_complete(self):
        if self.is_initialized():
            if self.total_bytes > 0:
                r = 1 - self._ratemeasure.get_size_left() / self.total_bytes
            else:
                r = 1.0
        else:
            r = 0.0

        return r

    def get_num_connections(self):
        if self._connection_manager:
            return self._connection_manager.how_many_connections()
        return 0

    def get_connections(self):
        return self._connection_manager.complete_connections

    def get_avg_peer_downrate(self):
        cs = self._connection_manager.complete_connections

        if len(cs) == 0:
            return 0.0

        total = 0.0
        for c in cs:
            total += c.download.connection.download.peermeasure.get_rate()

        return total / len(cs)

    def get_status(self, spew = False, fileinfo=False):
        if self.is_initialized():
            r = self._statuscollector.get_statistics(spew, fileinfo)
            r['activity'] = self._activity[0]
            r['priority'] = self.priority
            if not self.is_running():
                r['timeEst'] = None
        else:
            r = dict(itertools.izip(('activity', 'fractionDone'), self._activity))
            r['pieceStates'] = (0, 0, {})
            r['priority'] = self.priority
        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.config.has_key(option) and self.config[option] == value:
            return
        self.config[option] = value

    def change_port(self, new_port = None):
        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.rescrewedported_port == r:
                return
        elif new_port is not None:
            r = new_port
            self.reserved_ports.remove(self.reported_port)
            self.reserved_ports.append(r)
        elif self._singleport_listener.port != self.reported_port:
            r = self._singleport_listener.get_port(self.change_port)
            self.reserved_ports.append(r)
        else:
            return
        self.reported_port = r
        self._myid = self._make_id()
        if self._connection_manager:
            self._connection_manager.my_id = self._myid
        if self._announced:
            self._rerequest.change_port(self._myid, r)

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

    def _make_id(self):
        return PeerID.make_id()