multitorrent.py

bittorrent source by python. please enjoy

# The contents of this file are subject to the BitTorrent Open Source License
# Version 1.0 (the License).  You may not copy or use this file, in either
# source code or executable form, except in compliance with the License.  You
# may obtain a copy of the License at http://www.bittorrent.com/license/.
#
# Software distributed under the License is distributed on an AS IS basis,
# WITHOUT WARRANTY OF ANY KIND, either express or implied.  See the License
# for the specific language governing rights and limitations under the
# License.

# Author: Steve Hazel, Bram Cohen, and Uoti Urpala.

import os
import sys
import shutil
import socket
import cPickle
import logging
import traceback
from copy import copy
from BitTorrent.translation import _
from BitTorrent import GetTorrent
from BitTorrent.Choker import Choker
from BitTorrent.platform import bttime, encode_for_filesystem, old_broken_config_subencoding, get_filesystem_encoding
from BitTorrent.Torrent import Feedback, Torrent
from BitTorrent.bencode import bencode, bdecode
from BitTorrent.ConvertedMetainfo import ConvertedMetainfo
from BitTorrent.prefs import Preferences
from BitTorrent.NatTraversal import NatTraverser
from BitTorrent.BandwidthManager import BandwidthManager
from BitTorrent.InternetWatcher import InternetWatcher
from BitTorrent.Rerequester import Rerequester, DHTRerequester
from BitTorrent.NewRateLimiter import MultiRateLimiter as RateLimiter
from BitTorrent.ConnectionManager import SingleportListener
from BitTorrent.CurrentRateMeasure import Measure
from BitTorrent.Storage import FilePool
from BitTorrent.yielddefer import launch_coroutine, _wrap_task
from BitTorrent.defer import Deferred, DeferredEvent
from BitTorrent import BTFailure, InfoHashType
from BitTorrent import configfile
import BitTorrent
from khashmir.utkhashmir import UTKhashmir

class TorrentException(BTFailure):
    pass
class TorrentAlreadyInQueue(TorrentException):
    pass
class TorrentAlreadyRunning(TorrentException):
    pass
class TorrentNotInitialized(TorrentException):
    pass
class TorrentNotRunning(TorrentException):
    pass
class UnknownInfohash(TorrentException):
    pass
class TorrentShutdownFailed(TorrentException):
    pass
class TooManyTorrents(TorrentException):
    pass

#class DummyTorrent(object):
#    def __init__(self, infohash):
#        self.metainfo = object()
#        self.metainfo.infohash = infohash

BUTLE_INTERVAL = 1

class MultiTorrent(Feedback):
    """A MultiTorrent object represents a set of BitTorrent file transfers.
       It acts as a factory for Torrent objects, and it acts as
       the interface through which communication is performed to and from
       torrent file transfers.

       If you wish to instantiate MultiTorrent to download only a single
       torrent then pass is_single_torrent=True.
       
       If you want to avoid resuming from prior torrent config state then
       pass resume_from_torrent_config = False.
       It will still use fast resume if available.

       """

    def __init__(self, config, rawserver,
                 data_dir, listen_fail_ok=False, init_torrents=True,
                 is_single_torrent=False, resume_from_torrent_config=True):
        """
         @param config: program-wide configuration object.
         @param rawserver: object that manages main event loop and event
           scheduling.
         @param data_dir: where variable data such as fastresume information
           and GUI state is saved.
         @param listen_fail_ok: if false, a BTFailure is raised if
           a server socket cannot be opened to accept incoming peer
           connections.
         @param init_torrents: restore fast resume state from prior
           instantiations of MultiTorrent.
         @param is_single_torrent: if true then allow only one torrent
           at a time in this MultiTorrent. 
         @param resume_from_torrent_config: resume from ui_state files.
        """
        # is_single_torrent will go away when we move MultiTorrent into
        # a separate process, in which case, single torrent applications like
        # curses and console will act as a client to the MultiTorrent daemon.
        #   --Dave

        # init_torrents refers to fast resume rather than torrent config.
        # If init_torrents is set to False, the UI state file is still
        # read and the paths to existing downloads still used. This is
        # not what we want for launchmany.
        #
        # resume_from_torrent_config is separate from
        # is_single_torrent because launchmany must be able to have
        # multiple torrents while not resuming from torrent config
        # state.  If launchmany resumes from torrent config then it
        # saves or seeds from the path in the torrent config even if
        # the file has moved in the directory tree.  Because
        # launchmany has no mechanism for removing torrents other than
        # to change the directory tree, the only way for the user to
        # eliminate the old state is to wipe out the files in the
        # .bittorrent/launchmany-*/ui_state directory.  This is highly
        # counterintuitive.  Best to simply ignore the ui_state
        # directory altogether.  --Dave
        
        assert isinstance(config, Preferences)
        #assert isinstance(data_dir, unicode)  # temporarily commented -Dave
        assert isinstance(listen_fail_ok, bool)
        assert not (is_single_torrent and resume_from_torrent_config)

        self.config = config
        self.data_dir = data_dir
        self.last_save_time = 0
        self.policies = []
        self.torrents = {}
        self.running = {}
        self.log_root = "core.MultiTorrent"
        self.logger = logging.getLogger(self.log_root)
        self.is_single_torrent = is_single_torrent
        self.resume_from_torrent_config = resume_from_torrent_config
        self.auto_update_policy_index = None
        self.dht = None
        self.rawserver = rawserver
        nattraverser = NatTraverser(self.rawserver)
        self.internet_watcher = InternetWatcher(self.rawserver)
        self.singleport_listener = SingleportListener(self.rawserver,
                                                      nattraverser,
                                                      self.log_root)
        self.choker = Choker(self.config, self.rawserver.add_task)
        self.ratelimiter = RateLimiter(self.rawserver.add_task)
        self.ratelimiter.set_parameters(config['max_upload_rate'],
                                        config['upload_unit_size'])
        self.total_downmeasure = Measure(config['max_rate_period'])

        self._find_port(listen_fail_ok)

        self.filepool_doneflag = DeferredEvent()
        self.filepool = FilePool(self.filepool_doneflag,
                                 self.rawserver.add_task,
                                 self.rawserver.external_add_task,
                                 config['max_files_open'],
                                 config['num_disk_threads'])

        if self.resume_from_torrent_config:
            try:
                self._restore_state(init_torrents)
            except BTFailure:
                # don't be retarted.
                self.logger.exception("_restore_state failed")

        def no_dump_set_option(option, value):
            self.set_option(option, value, dump=False)

        self.bandwidth_manager = BandwidthManager(
            self.rawserver.external_add_task, config,
            no_dump_set_option, self.rawserver.get_remote_endpoints,
            get_rates=self.get_total_rates )

        self.rawserver.add_task(0, self.butle)
        #raise Exception("blah")


    def butle(self):
        policy = None
        try:
            for policy in self.policies:
                policy.butle()
        except:
            # You had something to hide, should have hidden it shouldn't you?
            self.logger.error("Butler error", exc_info=sys.exc_info())
            # Should we remove policies?
            #if policy:
            #    self.policies.remove(policy)
        self.rawserver.add_task(BUTLE_INTERVAL, self.butle)


    def _find_port(self, listen_fail_ok=True):
        """Run BitTorrent on the first available port found starting
           from minport in the range [minport, maxport]."""

        exc_info = None

        self.config['minport'] = max(1024, self.config['minport'])

        self.config['maxport'] = max(self.config['minport'],
                                     self.config['maxport'])

        e = (_("maxport less than minport - no ports to check") +
             (": %s %s" % (self.config['minport'], self.config['maxport'])))

        for port in xrange(self.config['minport'], self.config['maxport'] + 1):
            try:
                self.singleport_listener.open_port(port, self.config)
                if self.config['start_trackerless_client']:
                    self.dht = UTKhashmir(self.config['bind'],
                                   self.singleport_listener.get_port(),
                                   self.data_dir, self.rawserver,
                                   int(self.config['max_upload_rate'] * 0.01),
                                   rlcount=self.ratelimiter.increase_offset,
                                   config=self.config)
                break
            except socket.error, e:
                exc_info = sys.exc_info()
        else:
            if not listen_fail_ok:
                raise BTFailure, (_("Could not open a listening port: %s.") %
                                  unicode(e.args[0]) )
            self.global_error(logging.CRITICAL,
                              (_("Could not open a listening port: %s. ") % e) +
                              (_("Check your port range settings (%s:%s-%s).") %
                               (self.config['bind'], self.config['minport'],
                                self.config['maxport'])),
                              exc_info=exc_info)

    def shutdown(self):
        df = launch_coroutine(_wrap_task(self.rawserver.add_task), self._shutdown)
        df.addErrback(lambda e : self.logger.error('shutdown failed!',
                                                   exc_info=e))
        return df

    def _shutdown(self):
        self.choker.shutdown()
        self.singleport_listener.close_sockets()
        for t in self.torrents.itervalues():
            try:
                df = t.shutdown()
                yield df
                df.getResult()
                totals = t.get_total_transfer()
                t.uptotal = t.uptotal_old + totals[0]
                t.downtotal = t.downtotal_old + totals[1]
            except:
                t.logger.debug("Torrent shutdown failed in state: %s", t.state)
                print "Torrent shutdown failed in state:", t.state
                traceback.print_exc()

        # the filepool must be shut down after the torrents,
        # or pending ops could never complete
        self.filepool_doneflag.set()

        if self.resume_from_torrent_config:
            self._dump_torrents()


    def set_option(self, option, value, infohash=None, dump=True):
        if infohash is not None:
            t = self.get_torrent(infohash)
            t.config[option] = value
            if dump:
                t._dump_torrent_config()
        else:
            self.config[option] = value
            if dump:
                self._dump_global_config()

        if option in ['max_upload_rate', 'upload_unit_size']:
            self.ratelimiter.set_parameters(self.config['max_upload_rate'],
                                            self.config['upload_unit_size'])
        elif option == 'max_download_rate':
            pass # polled from the config automatically by MultiDownload
        elif option == 'max_files_open':
            self.filepool.set_max_files_open(value)