# Written by John Hoffman # Modified by Cameron Dale # see LICENSE.txt for license information # # $Id$ """Manage the downloading of multiple torrents in one process. @type logger: C{logging.Logger} @var logger: the logger to send all log messages to for this module """ from DebTorrent import PSYCO if PSYCO.psyco: try: import psyco assert psyco.__version__ >= 0x010100f0 psyco.full() except: pass from download_bt1 import BT1Download from RawServer import RawServer from RateLimiter import RateLimiter from ServerPortHandler import MultiHandler from random import seed from socket import error as socketerror from threading import Event from sys import argv, exit import sys, os, binascii from clock import clock from __init__ import createPeerID, mapbase64, version from cStringIO import StringIO import logging from DebTorrent.BT1.AptListener import AptListener from DebTorrent.HTTPHandler import HTTPHandler from binascii import b2a_hex, a2b_hex logger = logging.getLogger('DebTorrent.launchmanycore') def fmttime(n): """Formats seconds into a human-readable time. Formats a given number of seconds into a human-readable time appropriate for display to the user. @type n: C{int} @param n: the number of seconds @rtype: C{string} @return: a displayable representation of the number of seconds """ try: n = int(n) # n may be None or too large assert n < 5184000 # 60 days except: return 'downloading' m, s = divmod(n, 60) h, m = divmod(m, 60) return '%d:%02d:%02d' % (h, m, s) class SingleDownload: """Manage a single torrent download. @type controller: L{LaunchMany} @ivar controller: the manager for all torrent downloads @type hash: C{string} @ivar hash: the info hash of the torrent @type identifier: C{string} @ivar identifier: the identifier of the torrent @type response: C{dictionary} @ivar response: the meta info for the torrent @type config: C{dictionary} @ivar config: the configuration parameters @type doneflag: C{threading.Event} @ivar doneflag: the flag that indicates when the torrent is to be shutdown @type waiting: C{boolean} @ivar waiting: whether the download is waiting for the hash check to complete before starting @type checking: C{boolean} @ivar checking: whether the hash check is in progress @type working: C{boolean} @ivar working: whether the download is under way @type seed: C{boolean} @ivar seed: whether this peer is a seed @type closed: C{boolean} @ivar closed: whether the download has been shutdown @type status_msg: C{string} @ivar status_msg: the current activity the torrent is engaged in @type status_err: C{list} of C{string} @ivar status_err: the list of errors that have occurred @type status_errtime: C{int} @ivar status_errtime: the time of the last error @type status_done: C{float} @ivar status_done: the fraction of the current activity that is complete @type rawserver: L{ServerPortHandler.SingleRawServer} @ivar rawserver: the simplified Server to use to handle this torrent @type d: L{download_bt1.BT1Download} @ivar d: the downloader for the torrent @type _hashcheckfunc: C{method} @ivar _hashcheckfunc: the method to call to hash check the torrent @type statsfunc: C{method} @ivar statsfunc: the method to call to get the statistics for the running download """ def __init__(self, controller, hash, identifier, response, config, myid): """Initialize the instance and start a new downloader. @type controller: L{LaunchMany} @param controller: the manager for all torrent downloads @type hash: C{string} @param hash: the info hash of the torrent @type identifier: C{string} @param identifier: the identifier of the torrent @type response: C{dictionary} @param response: the meta info for the torrent @type config: C{dictionary} @param config: the configuration parameters @type myid: C{string} @param myid: the peer ID to use """ self.controller = controller self.hash = hash self.identifier = identifier self.response = response self.config = config self.doneflag = Event() self.waiting = True self.checking = False self.working = False self.seed = False self.closed = False self.status_msg = '' self.status_err = [''] self.status_errtime = 0 self.status_done = 0.0 self.rawserver = controller.handler.newRawServer(hash, identifier, self.doneflag) d = BT1Download(self.update_status, self.finished, self.error, self.doneflag, config, response, identifier, myid, self.rawserver, controller.listen_port, self.controller.configdir) self.d = d def start(self): """Initialize the new torrent download and schedule it for hash checking.""" if not self.d.saveAs(self.saveAs): self._shutdown() return self._hashcheckfunc = self.d.initFiles() if not self._hashcheckfunc: self._shutdown() return self.controller.hashchecksched(self.hash) def saveAs(self, name, length, saveas, isdir): """Determine the location to save the torrent in. @type name: C{string} @param name: the name from the torrent's metainfo @type length: C{long} @param length: the total length of the torrent download (not used) @type saveas: C{string} @param saveas: the user specified location to save to @type isdir: C{boolean} @param isdir: whether the torrent needs a directory @rtype: C{string} @return: the location to save the torrent in """ return self.controller.saveAs(self.hash, name, saveas, isdir) def hashcheck_start(self, donefunc): """Start the hash checking of the torrent. @type donefunc: C{method} @param donefunc: the method to call when the hash checking is complete """ if self.is_dead(): self._shutdown() return self.waiting = False self.checking = True self._hashcheckfunc(donefunc) def hashcheck_callback(self): """Start the torrent running now that hash checking is complete.""" self.checking = False if self.is_dead(): self._shutdown() return if not self.d.startEngine(ratelimiter = self.controller.ratelimiter): self._shutdown() return self.d.startRerequester() self.statsfunc = self.d.startStats() self.rawserver.start_listening(self.d.getPortHandler()) self.working = True def is_dead(self): """Check if the torrent download has been shutdown. @rtype: C{boolean} @return: whether the torrent download has been shutdown """ return self.doneflag.isSet() def _shutdown(self): """Loudly shutdown the running torrent.""" self.shutdown(False) def shutdown(self, quiet=True): """Shutdown the running torrent. @type quiet: C{boolean} @param quiet: whether to announce the shutdown (optional, defaults to True) """ if self.closed: return logger.info('Download shutdown') self.doneflag.set() self.rawserver.shutdown() if self.checking or self.working: self.d.shutdown() self.waiting = False self.checking = False self.working = False self.closed = True self.controller.was_stopped(self.hash) if not quiet: self.controller.died(self.hash) def update_status(self, activity = None, fractionDone = None): """Update the current activity's status. Really only used by StorageWrapper now. @type activity: C{string} @param activity: the activity currently under way (optional, defaults to not changing the current activity) @type fractionDone: C{float} @param fractionDone: the fraction of the activity that is complete (optional, defaults to not changing the current fraction done) """ if activity: self.status_msg = activity if fractionDone is not None: self.status_done = float(fractionDone) def finished(self): """Indicate that the download has completed.""" self.seed = True def error(self, msg): """Add a new error to the list of errors that have occurred. @type msg: C{string} @param msg: the error message """ if self.doneflag.isSet(): self._shutdown() self.status_err.append(msg) self.status_errtime = clock() class LaunchMany: """Manage the collection of all single torrent downloads. @type config: C{dictionary} @ivar config: the configuration parameters @type configdir: L{ConfigDir.ConfigDir} @ivar configdir: the configuration and cache directory manager @type torrent_cache: C{dictionary} @ivar torrent_cache: the cache of known torrents, keys are info hashes @type file_cache: C{dictionary} @ivar file_cache: the files found in the parsing of the torrent directory @type blocked_files: C{dictionary} @ivar blocked_files: the torrents in the torrent directory that will not be run @type torrent_list: C{list} of C{string} @ivar torrent_list: the list of running torrents' info hashes @type downloads: C{dictionary} @ivar downloads: the currently running downloaders, keys are info hashes @type counter: C{int} @ivar counter: the number of torrents that have been started so far @type doneflag: C{threading.Event} @ivar doneflag: flag to indicate all is to be stopped @type hashcheck_queue: C{list} of C{string} @ivar hashcheck_queue: the list of torrent info hashes waiting to be hash checked @type hashcheck_current: C{string} @ivar hashcheck_current: the info hash of the torrent currently being hash checked @type rawserver: L{RawServer.RawServer} @ivar rawserver: the Server instance to use for the downloads @type listen_port: C{int} @ivar listen_port: the port to listen on for incoming torrent connections @type aptlistener: L{BT1.AptListener.AptListener} @ivar aptlistener: the AptListener instance used to listen for incoming connections from Apt @type ratelimiter: L{RateLimiter.RateLimiter} @ivar ratelimiter: the limiter used to cap the maximum upload rate @type handler: L{ServerPortHandler.MultiHandler} @ivar handler: the multi torrent port listener used to handle connections """ def __init__(self, config, configdir): """Initialize the instance. @type config: C{dictionary} @param config: the configuration parameters @type configdir: L{ConfigDir.ConfigDir} @param configdir: the configuration and cache directory manager """ self.config = config self.configdir = configdir self.torrent_cache = {} self.file_cache = {} self.blocked_files = {} self.torrent_list = [] self.downloads = {} self.counter = 0 self.doneflag = Event() self.hashcheck_queue = [] self.hashcheck_current = None def run(self): """Run the mutliple downloads. @rtype: C{boolean} @return: whether the server ended normally """ restart = False try: self.rawserver = RawServer(self.doneflag, self.config['timeout_check_interval'], self.config['timeout'], ipv6_enable = self.config['ipv6_enabled']) self.listen_port = self.rawserver.find_and_bind( self.config['minport'], self.config['maxport'], self.config['bind'], ipv6_socket_style = self.config['ipv6_binds_v4'], randomizer = self.config['random_port']) if self.config['log_dir']: logfile = os.path.join(self.config['log_dir'], 'apt-access.log') else: logfile = os.path.join(self.configdir.cache_dir, 'apt-access.log') self.aptlistener = AptListener(self, self.config, self.rawserver) self.rawserver.bind(self.config['apt_port'], self.config['apt_bind'], reuse = True, ipv6_socket_style = self.config['ipv6_binds_v4']) self.rawserver.set_handler(HTTPHandler(self.aptlistener.get, self.config['min_time_between_log_flushes'], logfile, self.config['hupmonitor'], 'HTTP/1.1'), self.config['apt_port']) self.ratelimiter = RateLimiter(self.rawserver.add_task, self.config['upload_unit_size']) self.ratelimiter.set_upload_rate(self.config['max_upload_rate']) self.handler = MultiHandler(self.rawserver, self.doneflag, self.config) seed(createPeerID()) # Restore the previous state of the downloads still_running = self.unpickle(self.configdir.getState()) # Expire any old cached files self.configdir.deleteOldCacheData(self.config['expire_cache_data'], still_running, True) restart = self.handler.listen_forever() # Save the current state of the downloads self.configdir.saveState(self.pickle()) logger.info('shutting down') self.hashcheck_queue = [] for hash in self.torrent_list: logger.info('dropped "'+self.torrent_cache[hash]['path']+'"') self.downloads[hash].shutdown() self.rawserver.shutdown() except: logger.exception('SYSTEM ERROR - EXCEPTION GENERATED') return restart def gather_stats(self): """Gather the statistics for the currently running torrents. Returns a list, one per running torrent, of tuples: (C{string}, C{string}, C{string}, C{string}, C{int}, C{int}, C{string}, C{float}, C{float}, C{float}, C{long}, C{long}, C{long}, C{float}, C{string}) Which are the name, info hash, current status, progress report, number of peers, number of seeds, seed message, number of distributed copies, upload rate, download rate, amount uploaded, amount downloaded, total length, time remaining, and latest error message. @rtype: C{list} @return: various statistics for the running torrents """ data = [] for hash in self.torrent_list: cache = self.torrent_cache[hash] id = cache['metainfo'].get('identifier', hash) if self.config['display_path']: name = cache['path'] else: name = cache['name'] size = 0 d = self.downloads[hash] progress = '0.0%' peers = 0 seeds = 0 seedsmsg = "S" dist = 0.0 uprate = 0.0 dnrate = 0.0 upamt = 0 dnamt = 0 httpdnamt = 0 t = 0 if d.is_dead(): status = 'stopped' elif d.waiting: status = 'waiting for hash check' elif d.checking: status = d.status_msg progress = '%.1f%%' % (d.status_done*100) else: stats = d.statsfunc() s = stats['stats'] if d.seed: status = 'seeding' progress = '100.0%' seeds = s.numOldSeeds seedsmsg = "s" dist = s.numCopies else: if s.numSeeds + s.numPeers: t = stats['time'] if t == 0: # unlikely t = 0.01 status = fmttime(t) else: t = -1 status = 'connecting to peers' progress = '%.1f%%' % (int(stats['frac']*1000)/10.0) seeds = s.numSeeds dist = s.numCopies2 dnrate = stats['down'] peers = s.numPeers uprate = stats['up'] upamt = s.upTotal dnamt = s.downTotal httpdnamt = s.httpDownTotal size = stats['wanted'] if d.is_dead() or d.status_errtime+300 > clock(): msg = d.status_err[-1] else: msg = '' data.append(( name, id, status, progress, peers, seeds, seedsmsg, dist, uprate, dnrate, upamt, dnamt, httpdnamt, size, t, msg )) return data def get_piecemap(self, id): """Return the piecemap (which peers have what) for a given torrent. @type id: C{string} @param id: Info hash for the torrent, as provided by gather_stats. Internal: The "id" argument should match the identifier that gather_stats provides in data[][1]. As of revision 373 it provides the long-lived torrent ID, not the hash. @rtype: C{list} @return: List of piecemaps. The first will be for the local peer, then the other peers in undefined order. """ hash = None for search_hash in self.torrent_list: if self.torrent_cache[search_hash]['metainfo'].get('identifier', search_hash) == id: hash = search_hash if hash == None: logger.warning('Could not find hash for id %s\n' % b2a_hex(id)) return None piecelist_list = [] piecelist_list.append(self.downloads[hash].d.storagewrapper.have) if (self.downloads[hash].d.downloader != None): for peer in self.downloads[hash].d.downloader.downloads: piecelist_list.append(peer.have) return piecelist_list def remove(self, hash): """Stop and remove a running torrent. @type hash: C{string} @param hash: the info hash of the torrent """ logger.info('Removing torrent: '+str(binascii.b2a_hex(hash))) self.torrent_list.remove(hash) self.downloads[hash].shutdown() del self.downloads[hash] def add(self, hash, data, save_cache = True): """Start a new torrent running, or add the data to the current one. @type hash: C{string} @param hash: the info hash of the torrent @type data: C{dictionary} @param data: various info about the torrent, including the metainfo @type save_cache: C{boolean} @param save_cache: whether to save the torrent metainfo to the cache """ # If the torrent is already cached, just add new deb_mirrors to it new_debmirrors = [] if hash in self.torrent_cache: for u in data['metainfo'].get('deb_mirrors', []): if u not in self.torrent_cache[hash]['metainfo'].get('deb_mirrors', []): new_debmirrors.append(u) self.torrent_cache[hash]['metainfo'].setdefault('deb_mirrors', []).append(u) else: self.torrent_cache[hash] = data # Save the new torrent file to the cache if save_cache: self.configdir.writeTorrent(self.torrent_cache[hash]['metainfo'], hash) # Check if the torrent is already running if hash in self.torrent_list: if not self.config['disable_http_downloader']: for u in new_debmirrors: self.downloads[hash].d.httpdownloader.make_download(u) else: # Stop any running previous versions of the same torrent same_names = [] for old_hash in self.torrent_list: if (self.torrent_cache[old_hash]['name'] == data['name'] or self.torrent_cache[old_hash]['metainfo'].get('identifier', old_hash) == data['metainfo'].get('identifier', hash)): same_names.append(old_hash) for old_hash in same_names: self.remove(old_hash) self.start(hash) def start(self, hash): """Start a cached torrent. @type hash: C{string} @param hash: the info hash of the torrent to start """ if hash not in self.torrent_cache: logger.error('Asked to start a torrent that is not in the cache') return # Assign the torrent identifier from the metainfo data if "identifier" in self.torrent_cache[hash]['metainfo']: id = self.torrent_cache[hash]['metainfo']['identifier'] logger.info('Starting torrent: '+str(binascii.b2a_hex(hash)) + ' identified by: '+str(binascii.b2a_hex(id))) else: id = hash logger.info('Starting torrent: '+str(binascii.b2a_hex(hash))) c = self.counter self.counter += 1 x = '' for i in xrange(3): x = mapbase64[c & 0x3F]+x c >>= 6 peer_id = createPeerID(x) d = SingleDownload(self, hash, id, self.torrent_cache[hash]['metainfo'], self.config, peer_id) self.torrent_list.append(hash) self.downloads[hash] = d d.start() def saveAs(self, hash, name, saveas, isdir): """Determine the location to save the torrent in. @type hash: C{string} @param hash: the info hash of the torrent @type name: C{string} @param name: the name from the torrent's metainfo @type saveas: C{string} @param saveas: the user specified location to save to @type isdir: C{boolean} @param isdir: whether the torrent needs a directory @rtype: C{string} @return: the location to save the torrent in """ x = self.torrent_cache[hash] style = self.config['saveas_style'] if style == 1: if saveas: saveas = os.path.join(saveas, name) else: saveas = os.path.join(self.configdir.home_dir, name) else: if saveas: saveas = os.path.join(saveas, x['mirror']) else: saveas = os.path.join(self.configdir.home_dir, x['mirror']) if isdir: if not os.path.exists(saveas): try: os.makedirs(saveas) except: logger.exception('error creating the saveas directory: '+saveas) raise OSError("couldn't create directory for "+x['path'] +" ("+saveas+")") else: if not os.path.exists(os.path.split(saveas)[0]): try: os.makedirs(os.path.split(saveas)[0]) except: raise OSError("couldn't create directory for "+x['path'] +" ("+os.path.split(saveas)[0]+")") return saveas def find_file(self, mirror, path): """Find which running torrent has the file. Checks the metainfo of each torrent in the cache to find one that has a file whose 'path' matches the given file's path. @type mirror: C{string} @param mirror: mirror name to find the download in @type path: C{list} of C{string} @param path: the path of the file to find @rtype: L{download_bt1.BT1Download}, C{int} @return: the running torrent that contains the file and the file's number (or None if no running torrents contain the file) """ file = '/'.join(path) logger.info('Trying to find file: '+file) found_torrents = [] # Check each torrent in the cache for hash, data in self.torrent_cache.items(): # Make sure this torrent is from the mirror in question if not data['metainfo']['name'].startswith(mirror): continue file_num = -1 for f in data['metainfo']['info']['files']: file_num += 1 # Check that the file ends with the desired file name if f['path'] and file.endswith('/'.join(f['path'])): logger.debug('Found file in: '+str(binascii.b2a_hex(hash))) found_torrents.append((hash, file_num)) if not found_torrents: logger.warning('Failed to find file: '+file) return None, None # Find a running torrent with the file, (also the newest non-running torrent) newest_mtime = 0 newest_torrent = '' for hash, file_num in found_torrents: if hash in self.torrent_list: logger.info('Using torrent: ' + str(binascii.b2a_hex(hash))) return self.downloads[hash].d, file_num else: if self.torrent_cache[hash]['time'] > newest_mtime: newest_mtime = self.torrent_cache[hash]['time'] newest_torrent = hash # Otherwise start the newest torrent found running if newest_torrent: logger.info('Starting torrent: ' + str(binascii.b2a_hex(hash))) self.start(newest_torrent) else: logger.warning('Could not find the newest torrent, just starting the last one: ' + str(binascii.b2a_hex(hash))) self.start(hash) logger.info('Using torrent: ' + str(binascii.b2a_hex(hash))) return self.downloads[hash].d, file_num def hashchecksched(self, hash = None): """Schedule a new torrent for hash checking. @type hash: C{string} @param hash: the info hash of the torrent to schedule (optional, default is to start the next torrent in the queue) """ if hash: self.hashcheck_queue.append(hash) if not self.hashcheck_current: self._hashcheck_start() def _hashcheck_start(self): """Start hash checking the next torrent in the queue.""" self.hashcheck_current = self.hashcheck_queue.pop(0) self.downloads[self.hashcheck_current].hashcheck_start(self.hashcheck_callback) def hashcheck_callback(self): """Start another torrent's hash check now that the current one is complete.""" self.downloads[self.hashcheck_current].hashcheck_callback() if self.hashcheck_queue: self._hashcheck_start() else: self.hashcheck_current = None def died(self, hash): """Inform the Output that the torrent has died. @type hash: C{string} @param hash: the info hash of the torrent """ if self.torrent_cache.has_key(hash): logger.error('DIED: "'+self.torrent_cache[hash]['path']+'"') def has_torrent(self, hash): """Determine whether there is a downloader for the torrent. @type hash: C{string} @param hash: the info hash of the torrent @rtype: C{boolean} @return: whether the torrent is in the cache of known torrents """ return self.torrent_cache.has_key(hash) def was_stopped(self, hash): """Remove the torrent from the hash check queue, even if it's already happening. @type hash: C{string} @param hash: the info hash of the torrent """ try: self.hashcheck_queue.remove(hash) except: pass if self.hashcheck_current == hash: self.hashcheck_current = None if self.hashcheck_queue: self._hashcheck_start() def pickle(self): """Save the current state of the downloads to a writable state. Pickled data format:: d['torrent cache'] = {info hash: C{dictionary}, ...} Contains the cached data for all running torrents, keys are the torrent's info hash, and the data is all the data saved in L{torrent_cache}. There is also a new key added "paused" for torrent's that were running but paused. @rtype: C{dictionary} @return: the saved state of the current downloads """ d = {} for hash in self.torrent_list: d[hash] = {} for k in self.torrent_cache[hash].keys(): if k != 'metainfo': d[hash][k] = self.torrent_cache[hash][k] if not self.downloads[hash].d.unpauseflag.isSet(): d[hash]['paused'] = 1 return {'torrent cache': d} def unpickle(self, data): """Restore the current state of the downloads. Reads the list of previosuly running torrents, loads their metainfo from the cache and starts the downloads running. Also pauses the download if it was previously paused. @type data: C{dictionary} @param data: the saved state of the previously running downloads downloads @rtype: C{list} of C{string} @return: the list of torrent hashes that are still running """ if data is None: return [] still_running = [] d = data['torrent cache'] for hash in d: still_running.append(hash) paused = d[hash].pop('paused', False) metainfo = self.configdir.getTorrent(hash) if metainfo: d[hash]['metainfo'] = metainfo self.add(hash, d[hash], False) if paused: self.downloads[hash].d.Pause() return still_running