[people/shoehn/ipfire.org.git] / www / pages / torrent / client / launchmanycore.py

#!/usr/bin/env python

# Written by John Hoffman
# see LICENSE.txt for license information

from BitTornado 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, UPnP_ERROR
from RateLimiter import RateLimiter
from ServerPortHandler import MultiHandler
from parsedir import parsedir
from natpunch import UPnP_test
from random import seed
from socket import error as socketerror
from threading import Event
from sys import argv, exit
import sys, os
from clock import clock
from __init__ import createPeerID, mapbase64, version
from cStringIO import StringIO
from traceback import print_exc

try:
    True
except:
    True = 1
    False = 0


def fmttime(n):
    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:
    def __init__(self, controller, hash, response, config, myid):
        self.controller = controller
        self.hash = hash
        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, self.doneflag)

        d = BT1Download(self.display, self.finished, self.error,
                        controller.exchandler, self.doneflag, config, response,
                        hash, myid, self.rawserver, controller.listen_port)
        self.d = d

    def start(self):
        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):
        return self.controller.saveAs(self.hash, name, saveas, isdir)

    def hashcheck_start(self, donefunc):
        if self.is_dead():
            self._shutdown()
            return
        self.waiting = False
        self.checking = True
        self._hashcheckfunc(donefunc)

    def hashcheck_callback(self):
        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):
        return self.doneflag.isSet()

    def _shutdown(self):
        self.shutdown(False)

    def shutdown(self, quiet=True):
        if self.closed:
            return
        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 display(self, activity = None, fractionDone = None):
        # really only used by StorageWrapper now
        if activity:
            self.status_msg = activity
        if fractionDone is not None:
            self.status_done = float(fractionDone)

    def finished(self):
        self.seed = True

    def error(self, msg):
        if self.doneflag.isSet():
            self._shutdown()
        self.status_err.append(msg)
        self.status_errtime = clock()


class LaunchMany:
    def __init__(self, config, Output):
        try:
            self.config = config
            self.Output = Output

            self.torrent_dir = config['torrent_dir']
            self.torrent_cache = {}
            self.file_cache = {}
            self.blocked_files = {}
            self.scan_period = config['parse_dir_interval']
            self.stats_period = config['display_interval']

            self.torrent_list = []
            self.downloads = {}
            self.counter = 0
            self.doneflag = Event()

            self.hashcheck_queue = []
            self.hashcheck_current = None
            
            self.rawserver = RawServer(self.doneflag, config['timeout_check_interval'],
                              config['timeout'], ipv6_enable = config['ipv6_enabled'],
                              failfunc = self.failed, errorfunc = self.exchandler)
            upnp_type = UPnP_test(config['upnp_nat_access'])
            while True:
                try:
                    self.listen_port = self.rawserver.find_and_bind(
                                    config['minport'], config['maxport'], config['bind'],
                                    ipv6_socket_style = config['ipv6_binds_v4'],
                                    upnp = upnp_type, randomizer = config['random_port'])
                    break
                except socketerror, e:
                    if upnp_type and e == UPnP_ERROR:
                        self.Output.message('WARNING: COULD NOT FORWARD VIA UPnP')
                        upnp_type = 0
                        continue
                    self.failed("Couldn't listen - " + str(e))
                    return

            self.ratelimiter = RateLimiter(self.rawserver.add_task,
                                           config['upload_unit_size'])
            self.ratelimiter.set_upload_rate(config['max_upload_rate'])

            self.handler = MultiHandler(self.rawserver, self.doneflag)
            seed(createPeerID())
            self.rawserver.add_task(self.scan, 0)
            self.rawserver.add_task(self.stats, 0)

            self.handler.listen_forever()

            self.Output.message('shutting down')
            self.hashcheck_queue = []
            for hash in self.torrent_list:
                self.Output.message('dropped "'+self.torrent_cache[hash]['path']+'"')
                self.downloads[hash].shutdown()
            self.rawserver.shutdown()

        except:
            data = StringIO()
            print_exc(file = data)
            Output.exception(data.getvalue())


    def scan(self):
        self.rawserver.add_task(self.scan, self.scan_period)
                                
        r = parsedir(self.torrent_dir, self.torrent_cache,
                     self.file_cache, self.blocked_files,
                     return_metainfo = True, errfunc = self.Output.message)

        ( self.torrent_cache, self.file_cache, self.blocked_files,
            added, removed ) = r

        for hash, data in removed.items():
            self.Output.message('dropped "'+data['path']+'"')
            self.remove(hash)
        for hash, data in added.items():
            self.Output.message('added "'+data['path']+'"')
            self.add(hash, data)

    def stats(self):            
        self.rawserver.add_task(self.stats, self.stats_period)
        data = []
        for hash in self.torrent_list:
            cache = self.torrent_cache[hash]
            if self.config['display_path']:
                name = cache['path']
            else:
                name = cache['name']
            size = cache['length']
            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
            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
                   
            if d.is_dead() or d.status_errtime+300 > clock():
                msg = d.status_err[-1]
            else:
                msg = ''

            data.append(( name, status, progress, peers, seeds, seedsmsg, dist,
                          uprate, dnrate, upamt, dnamt, size, t, msg ))
        stop = self.Output.display(data)
        if stop:
            self.doneflag.set()

    def remove(self, hash):
        self.torrent_list.remove(hash)
        self.downloads[hash].shutdown()
        del self.downloads[hash]
        
    def add(self, hash, data):
        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, data['metainfo'], self.config, peer_id)
        self.torrent_list.append(hash)
        self.downloads[hash] = d
        d.start()


    def saveAs(self, hash, name, saveas, isdir):
        x = self.torrent_cache[hash]
        style = self.config['saveas_style']
        if style == 1 or style == 3:
            if saveas:
                saveas = os.path.join(saveas,x['file'][:-1-len(x['type'])])
            else:
                saveas = x['path'][:-1-len(x['type'])]
            if style == 3:
                if not os.path.isdir(saveas):
                    try:
                        os.mkdir(saveas)
                    except:
                        raise OSError("couldn't create directory for "+x['path']
                                      +" ("+saveas+")")
                if not isdir:
                    saveas = os.path.join(saveas, name)
        else:
            if saveas:
                saveas = os.path.join(saveas, name)
            else:
                saveas = os.path.join(os.path.split(x['path'])[0], name)
                
        if isdir and not os.path.isdir(saveas):
            try:
                os.mkdir(saveas)
            except:
                raise OSError("couldn't create directory for "+x['path']
                                      +" ("+saveas+")")
        return saveas


    def hashchecksched(self, hash = None):
        if hash:
            self.hashcheck_queue.append(hash)
        if not self.hashcheck_current:
            self._hashcheck_start()

    def _hashcheck_start(self):
        self.hashcheck_current = self.hashcheck_queue.pop(0)
        self.downloads[self.hashcheck_current].hashcheck_start(self.hashcheck_callback)

    def hashcheck_callback(self):
        self.downloads[self.hashcheck_current].hashcheck_callback()
        if self.hashcheck_queue:
            self._hashcheck_start()
        else:
            self.hashcheck_current = None

    def died(self, hash):
        if self.torrent_cache.has_key(hash):
            self.Output.message('DIED: "'+self.torrent_cache[hash]['path']+'"')
        
    def was_stopped(self, hash):
        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 failed(self, s):
        self.Output.message('FAILURE: '+s)

    def exchandler(self, s):
        self.Output.exception(s)
Commit	Line	Data
879aa787 MT	1	#!/usr/bin/env python
	2
	3	# Written by John Hoffman
	4	# see LICENSE.txt for license information
	5
	6	from BitTornado import PSYCO
	7	if PSYCO.psyco:
	8	try:
	9	import psyco
	10	assert psyco.__version__ >= 0x010100f0
	11	psyco.full()
	12	except:
	13	pass
	14
	15	from download_bt1 import BT1Download
	16	from RawServer import RawServer, UPnP_ERROR
	17	from RateLimiter import RateLimiter
	18	from ServerPortHandler import MultiHandler
	19	from parsedir import parsedir
	20	from natpunch import UPnP_test
	21	from random import seed
	22	from socket import error as socketerror
	23	from threading import Event
	24	from sys import argv, exit
	25	import sys, os
	26	from clock import clock
	27	from __init__ import createPeerID, mapbase64, version
	28	from cStringIO import StringIO
	29	from traceback import print_exc
	30
	31	try:
	32	True
	33	except:
	34	True = 1
	35	False = 0
	36
	37
	38	def fmttime(n):
	39	try:
	40	n = int(n) # n may be None or too large
	41	assert n < 5184000 # 60 days
	42	except:
	43	return 'downloading'
	44	m, s = divmod(n, 60)
	45	h, m = divmod(m, 60)
	46	return '%d:%02d:%02d' % (h, m, s)
	47
	48	class SingleDownload:
	49	def __init__(self, controller, hash, response, config, myid):
	50	self.controller = controller
	51	self.hash = hash
	52	self.response = response
	53	self.config = config
	54
	55	self.doneflag = Event()
	56	self.waiting = True
	57	self.checking = False
	58	self.working = False
	59	self.seed = False
	60	self.closed = False
	61
	62	self.status_msg = ''
	63	self.status_err = ['']
	64	self.status_errtime = 0
65	self.status_done = 0.0
66
67	self.rawserver = controller.handler.newRawServer(hash, self.doneflag)
68
69	d = BT1Download(self.display, self.finished, self.error,
70	controller.exchandler, self.doneflag, config, response,
71	hash, myid, self.rawserver, controller.listen_port)
72	self.d = d
73
74	def start(self):
75	if not self.d.saveAs(self.saveAs):
76	self._shutdown()
77	return
78	self._hashcheckfunc = self.d.initFiles()
79	if not self._hashcheckfunc:
80	self._shutdown()
81	return
82	self.controller.hashchecksched(self.hash)
83
84
85	def saveAs(self, name, length, saveas, isdir):
86	return self.controller.saveAs(self.hash, name, saveas, isdir)
87
88	def hashcheck_start(self, donefunc):
89	if self.is_dead():
90	self._shutdown()
91	return
92	self.waiting = False
93	self.checking = True
94	self._hashcheckfunc(donefunc)
95
96	def hashcheck_callback(self):
97	self.checking = False
98	if self.is_dead():
99	self._shutdown()
100	return
101	if not self.d.startEngine(ratelimiter = self.controller.ratelimiter):
102	self._shutdown()
103	return
104	self.d.startRerequester()
105	self.statsfunc = self.d.startStats()
106	self.rawserver.start_listening(self.d.getPortHandler())
107	self.working = True
108
109	def is_dead(self):
110	return self.doneflag.isSet()
111
112	def _shutdown(self):
113	self.shutdown(False)
114
115	def shutdown(self, quiet=True):
116	if self.closed:
117	return
118	self.doneflag.set()
119	self.rawserver.shutdown()
120	if self.checking or self.working:
121	self.d.shutdown()
122	self.waiting = False
123	self.checking = False
124	self.working = False
125	self.closed = True
126	self.controller.was_stopped(self.hash)
127	if not quiet:
128	self.controller.died(self.hash)
129
130
131	def display(self, activity = None, fractionDone = None):
132	# really only used by StorageWrapper now
133	if activity:
134	self.status_msg = activity
135	if fractionDone is not None:
136	self.status_done = float(fractionDone)
137
138	def finished(self):
139	self.seed = True
140
141	def error(self, msg):
142	if self.doneflag.isSet():
143	self._shutdown()
144	self.status_err.append(msg)
145	self.status_errtime = clock()
146
147
148	class LaunchMany:
149	def __init__(self, config, Output):
150	try:
151	self.config = config
152	self.Output = Output
153
154	self.torrent_dir = config['torrent_dir']
155	self.torrent_cache = {}
156	self.file_cache = {}
157	self.blocked_files = {}
158	self.scan_period = config['parse_dir_interval']
159	self.stats_period = config['display_interval']
160
161	self.torrent_list = []
162	self.downloads = {}
163	self.counter = 0
164	self.doneflag = Event()
165
166	self.hashcheck_queue = []
167	self.hashcheck_current = None
168
169	self.rawserver = RawServer(self.doneflag, config['timeout_check_interval'],
170	config['timeout'], ipv6_enable = config['ipv6_enabled'],
171	failfunc = self.failed, errorfunc = self.exchandler)
172	upnp_type = UPnP_test(config['upnp_nat_access'])
173	while True:
174	try:
175	self.listen_port = self.rawserver.find_and_bind(
176	config['minport'], config['maxport'], config['bind'],
177	ipv6_socket_style = config['ipv6_binds_v4'],
178	upnp = upnp_type, randomizer = config['random_port'])
179	break
180	except socketerror, e:
181	if upnp_type and e == UPnP_ERROR:
182	self.Output.message('WARNING: COULD NOT FORWARD VIA UPnP')
183	upnp_type = 0
184	continue
185	self.failed("Couldn't listen - " + str(e))
186	return
187
188	self.ratelimiter = RateLimiter(self.rawserver.add_task,
189	config['upload_unit_size'])
190	self.ratelimiter.set_upload_rate(config['max_upload_rate'])
191
192	self.handler = MultiHandler(self.rawserver, self.doneflag)
193	seed(createPeerID())
194	self.rawserver.add_task(self.scan, 0)
195	self.rawserver.add_task(self.stats, 0)
196
197	self.handler.listen_forever()
198
199	self.Output.message('shutting down')
200	self.hashcheck_queue = []
201	for hash in self.torrent_list:
202	self.Output.message('dropped "'+self.torrent_cache[hash]['path']+'"')
203	self.downloads[hash].shutdown()
204	self.rawserver.shutdown()
205
206	except:
207	data = StringIO()
208	print_exc(file = data)
209	Output.exception(data.getvalue())
210
211
212	def scan(self):
213	self.rawserver.add_task(self.scan, self.scan_period)
214
215	r = parsedir(self.torrent_dir, self.torrent_cache,
216	self.file_cache, self.blocked_files,
217	return_metainfo = True, errfunc = self.Output.message)
218
219	( self.torrent_cache, self.file_cache, self.blocked_files,
220	added, removed ) = r
221
222	for hash, data in removed.items():
223	self.Output.message('dropped "'+data['path']+'"')
224	self.remove(hash)
225	for hash, data in added.items():
226	self.Output.message('added "'+data['path']+'"')
227	self.add(hash, data)
228
229	def stats(self):
230	self.rawserver.add_task(self.stats, self.stats_period)
231	data = []
232	for hash in self.torrent_list:
233	cache = self.torrent_cache[hash]
234	if self.config['display_path']:
235	name = cache['path']
236	else:
237	name = cache['name']
238	size = cache['length']
239	d = self.downloads[hash]
240	progress = '0.0%'
241	peers = 0
242	seeds = 0
243	seedsmsg = "S"
244	dist = 0.0
245	uprate = 0.0
246	dnrate = 0.0
247	upamt = 0
248	dnamt = 0
249	t = 0
250	if d.is_dead():
251	status = 'stopped'
252	elif d.waiting:
253	status = 'waiting for hash check'
254	elif d.checking:
255	status = d.status_msg
256	progress = '%.1f%%' % (d.status_done*100)
257	else:
258	stats = d.statsfunc()
259	s = stats['stats']
260	if d.seed:
261	status = 'seeding'
262	progress = '100.0%'
263	seeds = s.numOldSeeds
264	seedsmsg = "s"
265	dist = s.numCopies
266	else:
267	if s.numSeeds + s.numPeers:
268	t = stats['time']
269	if t == 0: # unlikely
270	t = 0.01
271	status = fmttime(t)
272	else:
273	t = -1
274	status = 'connecting to peers'
275	progress = '%.1f%%' % (int(stats['frac']*1000)/10.0)
276	seeds = s.numSeeds
277	dist = s.numCopies2
278	dnrate = stats['down']
279	peers = s.numPeers
280	uprate = stats['up']
281	upamt = s.upTotal
282	dnamt = s.downTotal
283
284	if d.is_dead() or d.status_errtime+300 > clock():
285	msg = d.status_err[-1]
286	else:
287	msg = ''
288
289	data.append(( name, status, progress, peers, seeds, seedsmsg, dist,
290	uprate, dnrate, upamt, dnamt, size, t, msg ))
291	stop = self.Output.display(data)
292	if stop:
293	self.doneflag.set()
294
295	def remove(self, hash):
296	self.torrent_list.remove(hash)
297	self.downloads[hash].shutdown()
298	del self.downloads[hash]
299
300	def add(self, hash, data):
301	c = self.counter
302	self.counter += 1
303	x = ''
304	for i in xrange(3):
305	x = mapbase64[c & 0x3F]+x
306	c >>= 6
307	peer_id = createPeerID(x)
308	d = SingleDownload(self, hash, data['metainfo'], self.config, peer_id)
309	self.torrent_list.append(hash)
310	self.downloads[hash] = d
311	d.start()
312
313
314	def saveAs(self, hash, name, saveas, isdir):
315	x = self.torrent_cache[hash]
316	style = self.config['saveas_style']
317	if style == 1 or style == 3:
318	if saveas:
319	saveas = os.path.join(saveas,x['file'][:-1-len(x['type'])])
320	else:
321	saveas = x['path'][:-1-len(x['type'])]
322	if style == 3:
323	if not os.path.isdir(saveas):
324	try:
325	os.mkdir(saveas)
326	except:
327	raise OSError("couldn't create directory for "+x['path']
328	+" ("+saveas+")")
329	if not isdir:
330	saveas = os.path.join(saveas, name)
331	else:
332	if saveas:
333	saveas = os.path.join(saveas, name)
334	else:
335	saveas = os.path.join(os.path.split(x['path'])[0], name)
336
337	if isdir and not os.path.isdir(saveas):
338	try:
339	os.mkdir(saveas)
340	except:
341	raise OSError("couldn't create directory for "+x['path']
342	+" ("+saveas+")")
343	return saveas
344
345
346	def hashchecksched(self, hash = None):
347	if hash:
348	self.hashcheck_queue.append(hash)
349	if not self.hashcheck_current:
350	self._hashcheck_start()
351
352	def _hashcheck_start(self):
353	self.hashcheck_current = self.hashcheck_queue.pop(0)
354	self.downloads[self.hashcheck_current].hashcheck_start(self.hashcheck_callback)
355
356	def hashcheck_callback(self):
357	self.downloads[self.hashcheck_current].hashcheck_callback()
358	if self.hashcheck_queue:
359	self._hashcheck_start()
360	else:
361	self.hashcheck_current = None
362
363	def died(self, hash):
364	if self.torrent_cache.has_key(hash):
365	self.Output.message('DIED: "'+self.torrent_cache[hash]['path']+'"')
366
367	def was_stopped(self, hash):
368	try:
369	self.hashcheck_queue.remove(hash)
370	except:
371	pass
372	if self.hashcheck_current == hash:
373	self.hashcheck_current = None
374	if self.hashcheck_queue:
375	self._hashcheck_start()
376
377	def failed(self, s):
378	self.Output.message('FAILURE: '+s)
379
380	def exchandler(self, s):
381	self.Output.exception(s)