- More code cleanup. The idea is that an API user only has to deal with

  the Tribler.API directory and triblerAPI. All other stuff is in subdirs
  that should not concern him/her.
  



git-svn-id: https://svn.tribler.org/abc/branches/arno/d07-09-21-fastgui-from-mb-r5526@6072 001aeff7-3401-0410-a489-f7902fc005dd

Tribler/API/Impl/DownloadImpl.py

@@ -0,0 +1,291 @@
+# Written by Arno Bakker 
+# see LICENSE.txt for license information
+
+import sys
+import os
+import copy
+from threading import currentThread
+from traceback import print_exc,print_stack
+
+import triblerAPI
+from Tribler.API.simpledefs import *
+from Tribler.API.osutils import *
+from Tribler.API.launchmanycore import SingleDownload
+from Tribler.unicode import metainfoname2unicode
+
+
+DEBUG = True
+
+class DownloadImpl:
+ def __init__(self):
+ pass
+
+ def setup(self,dcfg=None,pstate=None,lmcreatedcallback=None,lmvodplayablecallback=None):
+ """
+ Create a Download object. Used internally by Session. Copies tdef and 
+ dcfg and binds them to this download.
+ 
+ in: 
+ tdef = unbound TorrentDef
+ dcfg = unbound DownloadStartupConfig or None (in which case 
+ DownloadStartupConfig.get_copy_of_default() is called and the result 
+ becomes the (bound) config of this Download.
+ """
+ try:
+ self.dllock.acquire() # not really needed, no other threads know of it
+
+ # See if internal tracker used
+ itrackerurl = self.session.get_internal_tracker_url()
+ infohash = self.tdef.get_infohash()
+ metainfo = self.tdef.get_metainfo()
+ usingitracker = False
+ if metainfo['announce'] == itrackerurl:
+ usingitracker = True
+ elif 'announce-list' in metainfo:
+ for tier in metainfo['announce-list']:
+ if itrackerurl in tier:
+ usingitracker = True
+ break
+
+ if usingitracker:
+ # Copy .torrent to state_dir/itracker so the tracker thread 
+ # finds it and accepts peer registrations for it.
+ # 
+ trackerdir = os.path.join(self.sessconfig['state_dir'],STATEDIR_ITRACKER_DIR)
+ basename = binascii.hexlify(infohash)+'.torrent' # ignore .tribe stuff, not vital
+ filename = os.path.join(trackerdir,basename)
+ self.tdef.save(filename)
+ # Bring to attention of Tracker thread
+ session.lm.tracker_rescan_dir()
+
+ # Copy dlconfig, from default if not specified
+ if dcfg is None:
+ cdcfg = DownloadStartupConfig()
+ else:
+ cdcfg = dcfg
+ self.dlconfig = copy.copy(cdcfg.dlconfig)
+
+ # TODO: copy sessconfig into dlconfig?
+
+
+ self.set_filepieceranges()
+
+ # Things that only exist at runtime
+ self.dlruntimeconfig= {}
+ # We want to remember the desired rates and the actual assigned quota
+ # rates by the RateManager
+ self.dlruntimeconfig['max_desired_upload_rate'] = self.dlconfig['max_upload_rate'] 
+ self.dlruntimeconfig['max_desired_download_rate'] = self.dlconfig['max_download_rate']
+
+
+ print >>sys.stderr,"Download: setup: get_max_desired",self.dlruntimeconfig['max_desired_upload_rate']
+
+ if pstate is None or pstate['dlstate']['status'] != DLSTATUS_STOPPED:
+ # Also restart on STOPPED_ON_ERROR, may have been transient
+ self.create_engine_wrapper(lmcreatedcallback,pstate,lmvodplayablecallback)
+
+ self.dllock.release()
+ except Exception,e:
+ print_exc()
+ self.set_error(e)
+ self.dllock.release()
+
+ def create_engine_wrapper(self,lmcreatedcallback,pstate,lmvodplayablecallback):
+ """ Called by any thread, assume dllock already acquired """
+ if DEBUG:
+ print >>sys.stderr,"Download: create_engine_wrapper()"
+
+ # all thread safe
+ infohash = self.get_def().get_infohash()
+ metainfo = copy.deepcopy(self.get_def().get_metainfo())
+
+ # H4xor this so the 'name' field is safe
+ namekey = metainfoname2unicode(metainfo)
+ self.correctedinfoname = fix_filebasename(metainfo['info'][namekey])
+ metainfo['info'][namekey] = metainfo['info']['name'] = self.correctedinfoname 
+
+ multihandler = self.session.lm.multihandler
+ listenport = self.session.get_listen_port()
+ vapath = self.session.get_video_analyser_path()
+
+ # Note: BT1Download is started with copy of d.dlconfig, not direct access
+ # Set IP to report to tracker. 
+ self.dlconfig['ip'] = self.session.lm.get_ext_ip()
+ kvconfig = copy.copy(self.dlconfig)
+
+ # Define which file to DL in VOD mode
+ if self.dlconfig['mode'] == DLMODE_VOD:
+ vod_usercallback_wrapper = lambda mimetype,stream,filename:self.session.uch.perform_vod_usercallback(self,self.dlconfig['vod_usercallback'],mimetype,stream,filename)
+
+ if 'files' in metainfo['info'] and len(self.dlconfig['selected_files']) == 0:
+ # Multi-file torrent, but no file selected
+ raise VODNoFileSelectedInMultifileTorrentException() 
+
+ if len(self.dlconfig['selected_files']) == 0:
+ # single-file torrent
+ file = self.get_def().get_name()
+ idx = -1
+ bitrate = self.get_def().get_bitrate(None)
+ else:
+ # multi-file torrent
+ file = self.dlconfig['selected_files'][0]
+ idx = self.get_def().get_index_of_file_in_files(file)
+ bitrate = self.get_def().get_bitrate(file)
+ vodfileindex = [idx,file,bitrate,None,vod_usercallback_wrapper]
+ else:
+ vodfileindex = [-1,None,0.0,None,None]
+
+ # Delegate creation of engine wrapper to network thread
+ network_create_engine_wrapper_lambda = lambda:self.network_create_engine_wrapper(infohash,metainfo,kvconfig,multihandler,listenport,vapath,vodfileindex,lmcreatedcallback,pstate,lmvodplayablecallback)
+ self.session.lm.rawserver.add_task(network_create_engine_wrapper_lambda,0) 
+
+
+ def network_create_engine_wrapper(self,infohash,metainfo,kvconfig,multihandler,listenport,vapath,vodfileindex,lmcallback,pstate,lmvodplayablecallback):
+ """ Called by network thread """
+ self.dllock.acquire()
+ try:
+ self.sd = SingleDownload(infohash,metainfo,kvconfig,multihandler,listenport,vapath,vodfileindex,self.set_error,pstate,lmvodplayablecallback)
+ sd = self.sd
+ exc = self.error
+ if lmcallback is not None:
+ lmcallback(self,sd,exc,pstate)
+ finally:
+ self.dllock.release()
+
+
+ def set_error(self,e):
+ self.dllock.acquire()
+ self.error = e
+ self.dllock.release()
+
+
+ def set_filepieceranges(self):
+ """ Determine which file maps to which piece ranges for progress info """
+
+ print >>sys.stderr,"Download: set_filepieceranges:",self.dlconfig['selected_files']
+
+ if len(self.dlconfig['selected_files']) > 0:
+ if 'files' not in self.tdef.metainfo['info']:
+ raise ValueError("Selected more than 1 file, but torrent is single-file torrent")
+
+ files = self.tdef.metainfo['info']['files']
+ piecesize = self.tdef.metainfo['info']['piece length']
+
+ total = 0L
+ for i in xrange(len(files)):
+ path = files[i]['path']
+ length = files[i]['length']
+ filename = pathlist2filename(path)
+
+ print >>sys.stderr,"Download: set_filepieceranges: Torrent file",filename,"in",self.dlconfig['selected_files']
+
+ if filename in self.dlconfig['selected_files'] and length > 0:
+
+ range = (offset2piece(total,piecesize), offset2piece(total + length,piecesize),filename)
+
+ print >>sys.stderr,"Download: set_filepieceranges: Torrent file range append",range
+
+ self.filepieceranges.append(range)
+ total += length
+ else:
+ self.filepieceranges = None 
+
+
+ def stop_remove(self,removestate=False,removecontent=False):
+ self.dllock.acquire()
+ try:
+ if self.sd is not None:
+ network_stop_lambda = lambda:self.network_stop(removestate,removecontent)
+ self.session.lm.rawserver.add_task(network_stop_lambda,0.0)
+ # No exception if already stopped, for convenience
+ finally:
+ self.dllock.release()
+
+
+ def network_get_state(self,usercallback,getpeerlist,sessioncalling=False):
+ """ Called by network thread """
+ self.dllock.acquire()
+ try:
+ if self.sd is None:
+ ds = triblerAPI.DownloadState(self,DLSTATUS_STOPPED,self.error,self.progressbeforestop)
+ else:
+ (status,stats,logmsgs) = self.sd.get_stats(getpeerlist)
+ ds = triblerAPI.DownloadState(self,status,self.error,None,stats=stats,filepieceranges=self.filepieceranges,logmsgs=logmsgs)
+ self.progressbeforestop = ds.get_progress()
+
+ #print >>sys.stderr,"STATS",stats
+
+ if sessioncalling:
+ return ds
+
+ # Invoke the usercallback function via a new thread.
+ # After the callback is invoked, the return values will be passed to
+ # the returncallback for post-callback processing.
+ self.session.uch.perform_getstate_usercallback(usercallback,ds,self.sesscb_get_state_returncallback)
+ finally:
+ self.dllock.release()
+
+
+ def sesscb_get_state_returncallback(self,usercallback,when,newgetpeerlist):
+ """ Called by SessionCallbackThread """
+ self.dllock.acquire()
+ try:
+ if when > 0.0:
+ # Schedule next invocation, either on general or DL specific
+ # TODO: ensure this continues when dl is stopped. Should be OK.
+ network_get_state_lambda = lambda:self.network_get_state(usercallback,newgetpeerlist)
+ if self.sd is None:
+ self.session.lm.rawserver.add_task(network_get_state_lambda,when)
+ else:
+ self.sd.dlrawserver.add_task(network_get_state_lambda,when)
+ finally:
+ self.dllock.release()
+
+ def network_stop(self,removestate,removecontent):
+ """ Called by network thread """
+ self.dllock.acquire()
+ try:
+ infohash = self.tdef.get_infohash() 
+ pstate = self.network_get_persistent_state() 
+ pstate['engineresumedata'] = self.sd.shutdown()
+
+ # Offload the removal of the content and other disk cleanup to another thread
+ if removestate:
+ self.session.uch.perform_removestate_callback(infohash,self.correctedinfoname,removecontent,self.dlconfig['saveas'])
+
+ return (infohash,pstate)
+ finally:
+ self.dllock.release()
+
+
+ def network_checkpoint(self):
+ """ Called by network thread """
+ self.dllock.acquire()
+ try:
+ pstate = self.network_get_persistent_state() 
+ pstate['engineresumedata'] = self.sd.checkpoint()
+ return (self.tdef.get_infohash(),pstate)
+ finally:
+ self.dllock.release()
+
+
+ def network_get_persistent_state(self):
+ """ Assume dllock already held """
+ pstate = {}
+ pstate['version'] = PERSISTENTSTATE_CURRENTVERSION
+ pstate['metainfo'] = self.tdef.get_metainfo() # assumed immutable
+ dlconfig = copy.copy(self.dlconfig)
+ # Reset unpicklable params
+ dlconfig['vod_usercallback'] = None
+ dlconfig['dlmode'] = DLMODE_NORMAL # no callback, no VOD
+ pstate['dscfg'] = DownloadStartupConfig(dlconfig=dlconfig)
+
+ pstate['dlstate'] = {}
+ ds = self.network_get_state(None,False,sessioncalling=True)
+ pstate['dlstate']['status'] = ds.get_status()
+ pstate['dlstate']['progress'] = ds.get_progress()
+
+ print >>sys.stderr,"Download: netw_get_pers_state: status",dlstatus_strings[ds.get_status()],"progress",ds.get_progress()
+
+ pstate['engineresumedata'] = None
+ return pstate