Peer协议设计 - 走看看

Python边学边用--BT客户端实现之(三)Peer协议设计

与peer建立tcp连接后，首先发送handshake消息进行握手

handshake消息格式如下：

一个字节0x19 + 一个字符串'BitTorrent protocol' + 8 byte 保留字节默认值为0（draft中对保留字节有定义）

+ 种子文件中info 部分的sha1字，大小为20个字节 + 20个自己的peer id（从tracker获取到的peer信息大多没有peerid，这个可以使用本地的peer id）

如果handshake信息协商不上，tcp连接将被关闭。

BT标准BEP-3中定义了8种peer消息：消息格式为msg_len(4个字节) + msg_type(1一个字节) + payload

0 - choke --发送该消息表示本段发送阻塞，对端将不能获取到piece数据，payload 为 0

1 - unchoke --发送该消息表示解阻塞，对端可以开始发送请求获取piece数据，payload 为 0

2 - interested --发送该消息，表示对对端的pieces数据有兴趣，payload 为 0

3 - not interested ---发送该消息，表示对对端的pieces数据没有兴趣了，payload 为 0

4 - have ---发送该消息，通告对端本段拥有的pieces，payload为4个字节的piece index

5 - bitfield ---发送该消息，通告对端本段拥有的pieces，为bit map的方式表示每个piece index在本端是否拥有。piece index所在bit位为1，表示拥有。

该消息为handshake成功后的第一个消息。

6 - request ---piece请求消息，payload为： index, begin, and length，都是4个字节表示，length一般实现为0x8000, 最大不能超过0x20000。

7 - piece ---piece 数据，payload为： index, begin,data

8 - cancel ---发送该消息，表示本端取消了某个piece请求。payload为：index, begin, and length

使用python的同步socket接口实现，为了减少处理过程被socket阻塞，使用多个线程处理每个peer。

每个peer包括3个线程：request timeout timer ，socket send data thread， socket receive data thread

使用 RLock 保护临界数据。peer建立和piece request分配有download task线程处理。

发送数据数据时先写一个队列，然后通过set threading.Event 变量出发socket send data thread发送数据，保证发送数据的线程不阻塞

由于 python没有结束线程的接口，socket send data thread， socket receive data thread 需要依赖特殊变量的赋值，使socket处理进程结束。

使用同步调用来触发下载过程运转，尽量不使用timer轮询的方式，可以降低cpu使用率并加快下载过程。

但是，多线程间的同步调用由于锁的存在，会导致性能下降并容易引入信号量死锁的问题。需要仔细设计好多线程的运行轨迹避免死锁。

draft BEP中定义的功能暂未实现，peer的上传流控未实现，peer质量分级未实现。

10 2012 档案

Python边学边用--BT客户端实现之(三)Peer协议设计

摘要: 与peer建立tcp连接后，首先发送handshake消息进行握手handshake消息格式如下：一个字节0x19 + 一个字符串'BitTorrent protocol' + 8 byte 保留字节默认值为0（draft中对保留字节有定义）+ 种子文件中info 部分的sha1字，大小为20个字节 + 20个自己的peer id（从tracker获取到的peer信息大多没有peerid，这个可以使用本地的peer id）如果handshake信息协商不上，tcp连接将被关闭。BT标准BEP-3中定义了8种peer消息：消息格式为msg_len(4个字节) + msg_type阅读全文

posted @ 2012-10-07 23:03 duandetao 阅读(619) | 评论 (0) 编辑

Python边学边用--BT客户端实现之(二)Peers获取

摘要: 解析完torrent文件之后，需要从Tracker服务器获取拥有每个piece的peer列表。tracker服务器使用http协议提供支持，使用get的方式获取请求内容。也有使用udp协议的，暂时还没有分析。get 格式如下：announce-url?info_hash=xxxxxxxxxxxxxxxxxxxxx,peer_id=xxxxxxxxxxxxxxxxxxxx,ip=x.x.x.x,port=xxxx,uploaded=xx,downloaded=xx,left=xx,event=xurl中各参数需要经过url扰码处理。其中，info_hash为torrent文件中info属性的va阅读全文

posted @ 2012-10-02 20:09 duandetao 阅读(221) | 评论 (0) 编辑

09 2012 档案

Python边学边用--BT客户端实现之（一）BitTorrent文件解析

摘要: BitTorrent文件解析：BitTorrent文件使用bencode编码，其中包括了4种数据类型：'d' 开头表示是dict类型，'e'表示结束'l' （小写字母L）开头表示是list类型，'e'表示结束'i'开头表示是integer类型，'e'表示结束，可以表示负数以数字开头表示string类型，数字为string长度，长度与string内容以':'分割默认所有text类型的属性为utf-8编码，但是大多数BitTorrent包含codepage 和 encoding属性，指定阅读全文

posted @ 2012-09-29 23:17 duandetao 阅读(1038) | 评论 (2) 编辑

  1 '''
  2 Created on 2012-9-29
  3 
  4 @author: ddt
  5 '''
  6 import torrent_file
  7 import bcodec
  8 import random
  9 import threading
 10 from  trackers_info import *
 11 import socket
 12 from peer_connect import PeerConnect
 13 import os
 14 import time
 15 import log_info
 16 
 17 class DownLoadTask(object):
 18     '''
 19     TODO: UPNP support
 20     TODO: local listening support
 21     TODO: threading safety
 22     TODO: DHT support
 23     TODO: flow control support
 24     '''
 25     def __init__(self):
 26         '''
 27         Constructor
 28         '''
 29         self.__host_ip, self.__host_port = _get_listening_addr();
 30         self.__bt_file = torrent_file.TorrentFile()
 31         self.__peer_id = ''
 32         self.__download_state = {}
 33         self.__files = []
 34         self.__info_hash = None
 35         self.__piece_length = 0
 36         
 37         self.__trackers_info = None
 38         
 39         self.__pieces_refresh_intvl = 30 # second
 40         self.__pieces_refresh_timer = None
 41         self.__piece_max_unread_time = 20
 42         
 43         self.__unchoked_peers = {}
 44         self.__choked_peers = {}
 45         self.__max_pieces_per_peer = 50
 46         
 47         self.__rlock_common = threading.RLock()
 48         
 49         self.__base_path = '.\\'
 50         
 51     pass
 52     
 53     def start_task(self, bt_file_name):
 54         with self.__rlock_common:
 55             try:
 56                 self.__bt_file.read_file(bt_file_name)
 57             except (IOError,torrent_file.BTFormatError), reason:
 58                 self.__write_log("Read bittorrent file error! Error:%s" %str(reason))
 59                 return False
 60     
 61             self.__peer_id = _generate_peerid()
 62             self.__files = self.__bt_file.get_files()
 63             self.__info_hash = self.__bt_file.get_info_hash()
 64             
 65             self.__piece_length = self.__bt_file.get_piece_length()
 66             
 67             self.__download_state['downloaded'] = 0
 68             self.__download_state['uploaded'] = 0
 69             self.__download_state['left'] = self.__get_total_length()
 70             self.__download_state['event'] = 'started'
 71             self.__download_state['pieces-num'] = self.__bt_file.get_pieces_num()
 72             self.__download_state['downloading-pieces'] = {}
 73             self.__download_state['waiting-pieces'] = {}
 74             self.__download_state['completed-pieces'] = {}
 75             
 76             self.__writingfile_pieces = []
 77             
 78             self.__trackers_info = TrackersInfo(self)
 79             
 80             self.__init_files()
 81             
 82             self.__init_pieces_state()
 83             
 84             self.__trackers_info.start_trackers_refresh()
 85             
 86             self.__start_pieces_refresh()
 87 
 88         return True
 89     
 90     def tacker_get_peers(self, peers):
 91         with self.__rlock_common:
 92             self.__try_new_peers(peers)
 93         self.__dispatch_pieces() 
 94     
 95     def get_announce_list(self):
 96         return self.__bt_file.get_announces()
 97     
 98     def get_download_stat(self):
 99         with self.__rlock_common:
100             downloaded = self.__download_state['downloaded']
101             uploaded = self.__download_state['uploaded']
102             left = self.__download_state['left']
103             event = self.__download_state['event']
104         return (downloaded,uploaded,left,event)
105     
106     def get_host_info(self):
107         host_info = (self.__peer_id, self.__host_ip, self.__host_port)
108         return  host_info
109              
110     def get_info_hash(self):
111         return self.__info_hash
112 
113     def get_local_id(self):
114         return self.__peer_id
115         
116     def get_local_have_pieces(self):
117         with self.__rlock_common:
118             return self.__download_state['completed-pieces'].keys()
119     
120     def get_pieces_num(self):
121         return self.__bt_file.get_pieces_num()
122     
123     def peer_pieces_canceled(self, pieces):
124         print 'peer_pieces_canceled:%s' %str(pieces)
125 
126         with self.__rlock_common:
127             for piece_index in pieces:
128                 if piece_index in self.__download_state['downloading-pieces'].keys():
129                     self.__undownload_piece(piece_index)
130  
131         self.__dispatch_pieces()
132     
133     def peer_have_pieces(self,peer_con, pieces):
134         self.__write_log('peer_have_pieces:%s' %str(pieces))
135         with self.__rlock_common:
136             self.__dispatch_peer_pieces(peer_con, pieces)
137             
138     def peer_pieces_completed(self, peer_con):
139         self.__write_log('peer_pieces_completed:%s' %str(peer_con.get_peer_addr()))
140         with self.__rlock_common:
141             self.__dispatch_peer_pieces(peer_con)
142             
143     def peer_choked(self, peer_con):
144         with self.__rlock_common:
145             peer_addr = peer_con.get_peer_addr()
146             if peer_addr in self.__unchoked_peers.keys():
147                 self.__unchoked_peers.pop(peer_addr)
148             self.__choked_peers[peer_con.get_peer_addr()] = peer_con
149 
150             #peeding_pieces = peer_con.get_local_pending_pieces()
151             #peer_con.cancel_pieces(peeding_pieces)
152             #self.peer_pieces_canceled(peeding_pieces)
153 
154     def peer_unchoked(self, peer_con):
155         with self.__rlock_common:
156             peer_addr = peer_con.get_peer_addr()
157             if peer_addr in self.__choked_peers.keys():
158                 self.__choked_peers.pop(peer_addr)
159             self.__unchoked_peers[peer_con.get_peer_addr()] = peer_con
160             self.__dispatch_peer_pieces(peer_con)
161     
162     def peer_dead(self, peer_con):
163         with self.__rlock_common:
164             peer_addr = peer_con.get_peer_addr()
165             if peer_addr in self.__choked_peers.keys():
166                 self.__choked_peers.pop(peer_addr)
167                 
168             if peer_addr in self.__unchoked_peers.keys():
169                 self.__unchoked_peers.pop(peer_addr)
170             
171     def write_piecedata(self, piece_index, offset, data):
172         with self.__rlock_common:
173             if piece_index in self.__download_state['downloading-pieces'].keys():
174                 piece_state = self.__download_state['downloading-pieces'][piece_index]
175 
176                 if offset == piece_state['waiting-offset']:
177                     self.__piece_install_data(piece_state, data)
178                     
179                     while len(piece_state['disorder-data']) != 0:
180                         curr_offset = piece_state['waiting-offset']
181                         if curr_offset in piece_state['disorder-data'].keys():
182                             self.__piece_install_data(piece_state, piece_state['disorder-data'][curr_offset])
183                         else:
184                             break
185                 else:
186                     piece_state['disorder-data'][offset] = data
187                     
188                 return
189                 
190         self.__write_log('unexpected piece data: %s' %str((piece_index,offset,data)))
191               
192     def read_piecedata(self, piece_index, offset, length):
193         with self.__rlock_common:
194             if piece_index in self.__download_state['completed-pieces'].keys():
195                 piece_state = self.__download_state['completed-pieces'][piece_index]
196                 
197                 if piece_state['waiting-offset'] == 0:
198                     self.__read_from_file(piece_state)
199     
200                 if offset + length <= piece_state['waiting-offset']:
201                     piece_state['unread-time'] = 0
202                     return piece_state['piece-data'][offset:offset+length]
203             
204         self.__write_log('read piece date but not hava:%s' %str((piece_index, offset, length)))
205     
206     def __piece_install_data(self, piece_state, data):
207         piece_state['waiting-offset'] += len(data)
208         piece_state['piece-data'] += data
209         
210         if piece_state['waiting-offset'] >= self.__piece_length:
211             piece_state['disorder-data'].clear()
212             self.__complete_piece(piece_state['piece-index'])
213             
214         if piece_state['waiting-offset'] > self.__piece_length:
215             piece_state['piece-data'] =  piece_state['piece-data'][0:self.__piece_length]
216             piece_state['waiting-offset'] = self.__piece_length
217             self.__write_log('piece data is too much:%s' %str(piece_state['piece-index']))
218             
219         self.__download_state['downloaded'] += piece_state['waiting-offset']
220         self.__download_state['left'] -= piece_state['waiting-offset']
221         
222     def __read_from_file(self, piece_state):
223         piece_index = piece_state['piece-index']
224         files = piece_state['file-info']  
225         one_file = files[0]
226         first_piece = one_file['first-piece']
227         last_piece = one_file['last-piece']
228         if first_piece[0] <= piece_index  or piece_index <= last_piece[0]:
229             file_offset = (piece_index - first_piece[0])*self.__piece_length - first_piece[1]
230             file_handle = open(self.__base_path+'\\'.join(one_file['name']),'rb')
231             file_handle.seek(file_offset)
232             data = file_handle.read(self.__piece_length)
233             file_handle.close()
234             piece_state['piece-data'] = data
235             piece_state['waiting-offset'] = self.__piece_length
236             piece_state['unread-time'] = 0
237         
238     def __write_piece_to_file(self, piece_state):
239         self.__write_log('__write_piece_to_file')
240         
241         piece_index = piece_state['piece-index']
242         data = piece_state['piece-data']
243         
244         for one_file in piece_state['file-info']:
245             first_piece = one_file['first-piece']
246             last_piece = one_file['last-piece']
247             if first_piece[0] <= piece_index  or piece_index <= last_piece[0]:
248                 file_offset = (piece_index - first_piece[0])*self.__piece_length - first_piece[1]
249             else:
250                 continue
251             piece_offset = 0
252             if first_piece[0] == piece_index:
253                 piece_offset = first_piece[1]
254             
255             write_file_len = self.__piece_length - piece_offset
256             if last_piece[0] == piece_index:
257                 write_file_len = last_piece[1] - piece_offset
258               
259             file_path = self.__base_path+'\\'.join(one_file['name'])
260             file_handle = open(file_path,'rb+')
261             file_handle.seek(file_offset)
262             data = file_handle.write(data[piece_offset:piece_offset+write_file_len])
263             file_handle.close()
264             self.__write_log('write data to file:%s' %str((file_path, file_offset, write_file_len)))
265 
266         self.__write_log('__write_piece_to_file end')
267     
268     def __get_total_length(self):
269         total_length  = 0
270         for one_file in self.__files:
271             total_length += one_file['length']
272         return total_length
273     
274     def __init_files(self):
275         self.__write_log( '__init_files')
276         for one_file in self.__files:
277             
278             path = self.__base_path
279             for dir in one_file['name'][:-1]:
280                 path += dir+'.\\'
281                 if not os.path.exists(path):
282                     os.mkdir(path)
283             
284             file_path = self.__base_path + '\\'.join(one_file['name'])
285             
286             if os.path.exists(file_path):
287                 break
288             
289             file_length = one_file['length']
290             file_handle = open(file_path,'ab+')
291             dumy_data = chr(0)*0x4000
292             for i in range(0,file_length, 0x4000):
293                 write_len = 0x4000
294                 if i+write_len > file_length:
295                     write_len = file_length - i
296                 file_handle.write(dumy_data[0:write_len])
297             file_handle.close()
298         pass
299     
300     def __init_pieces_state(self):
301         self.__write_log( '__init_pieces_state')
302         pieces_state = {}
303         for one_file in self.__files:
304             for piece_index in range(one_file['first-piece'][0], one_file['last-piece'][0] + 1):
305                 if piece_index not in pieces_state.keys():
306                     piece_state = {}
307                     piece_state['connect-info'] = None
308                     piece_state['file-info'] = [one_file]
309 
310                     piece_state['waiting-offset'] = 0
311                     piece_state['piece-data'] = ''
312                     piece_state['disorder-data'] = {}
313                     
314                     piece_state['piece-index'] = piece_index
315                     
316                     piece_state['unread-time'] = 0
317                     pieces_state[piece_index] = piece_state
318                 else:
319                     pieces_state[piece_index]['file-info'] += one_file
320         self.__download_state['waiting-pieces'] = pieces_state
321         pass
322     
323     def __start_pieces_refresh(self):
324         if self.__pieces_refresh_timer != None:
325             self.__pieces_refresh_timer.cancel()
326                 
327         refresh_thread = threading.Thread(target=DownLoadTask.__refresh_pieces_state, args=(self,))
328         refresh_thread.start()
329     
330     def __refresh_pieces_state(self):
331         self.__write_log( '__refresh_pieces_state')
332         
333         with self.__rlock_common:
334             for piece_state in self.__download_state['completed-pieces'].values():
335                 if piece_state['waiting-offset'] != 0:   
336                     piece_state['unread-time'] += 1 
337                 
338                 if len(piece_state['file-info']) >= 2 and piece_state['unread-time'] > self.__piece_max_unread_time:
339                     piece_state['waiting-offset'] = 0
340                     piece_state['piece-data'] = ''
341                     piece_state['unread-time'] = 0
342             
343             self.__check_files_completed()
344                        
345             if self.__pieces_refresh_timer != None:
346                 self.__pieces_refresh_timer.cancel()
347                 
348             self.__pieces_refresh_timer = threading.Timer(self.__pieces_refresh_intvl,\
349                                                           DownLoadTask.__refresh_pieces_state, [self,])
350             self.__pieces_refresh_timer.start()
351         
352         self.__dispatch_pieces()
353         self.__write_log( '__refresh_pieces_state end')
354     
355     def __notify_peers_have(self, have_pieces):
356         self.__write_log('__notify_peers_have')
357         with self.__rlock_common:
358             for peer_con in self.__unchoked_peers.values():
359                 peer_con.notify_local_have_pieces(have_pieces) 
360                 
361             for peer_con in self.__choked_peers.values():
362                 peer_con.notify_local_have_pieces(have_pieces) 
363     
364     def __try_new_peers(self, peers):
365         with self.__rlock_common:
366             for peer in peers:
367                 if peer not in self.__unchoked_peers.keys() and peer not in self.__choked_peers.keys():
368                     peer_con = PeerConnect(peer[0],peer[1],self)
369                     self.__choked_peers[peer] = peer_con
370                     peer_con.start()
371     
372     def __check_files_completed(self):
373         self.__write_log( '__check_files_completed')
374         completed_pieces = self.__download_state['completed-pieces'].keys()
375         completed_pieces = set(completed_pieces)
376         writingfile_pieces = set(self.__writingfile_pieces)
377         
378         for one_file in self.__files:
379             file_path = self.__base_path + '.\\'.join(one_file['name'])
380 
381             first_piece = one_file['first-piece']
382             last_piece = one_file['last-piece']
383             
384             file_pieces = range(first_piece[0], last_piece[0]+1)
385             file_pieces = set(file_pieces)
386 
387             need_writefile = file_pieces & writingfile_pieces
388             if len(need_writefile) != 0:
389                 file_handle = open(file_path, 'ab+')
390                 for piece_index in need_writefile:
391                     
392                     piece_state = self.__download_state['completed-pieces'][piece_index]
393                     if len(piece_state['file-info']) > 1:
394                         self.__write_piece_to_file(piece_state)
395                         writingfile_pieces.remove(piece_index)
396                         continue
397                     
398                     data = piece_state['piece-data']
399                     file_offset = (piece_index - first_piece[0])*self.__piece_length - first_piece[1]
400                     file_handle.seek(file_offset)
401                     file_handle.write(data)
402                     self.__write_log( 'writing data to file:%s,%s' %(piece_index,file_path))
403                 file_handle.close()
404                 
405             if file_pieces <= completed_pieces:
406                 one_file['completed'] = True
407                 self.__write_log( 'file:%s completed!' %file_path)
408                 
409         self.__writingfile_pieces = []
410            
411     def __dispatch_pieces(self):
412         self.__write_log( '__dispatch_pieces')
413         for peer_con in self.__unchoked_peers.values():
414             if len(self.__download_state['waiting-pieces'].keys()) == 0:
415                 break
416             
417             self.__dispatch_peer_pieces(peer_con)
418         
419         for peer_con in self.__choked_peers.values():
420             if len(self.__download_state['waiting-pieces'].keys()) == 0:
421                 break
422             
423             self.__dispatch_peer_pieces(peer_con)
424 
425     def __dispatch_peer_pieces(self, peer_con, peer_have_pieces = None):
426         self.__write_log( '__dispatch_peer_pieces')
427 
428         peer_pending_pieces = peer_con.get_local_pending_pieces()
429         dispatching_num = self.__max_pieces_per_peer - len(peer_pending_pieces)
430         if dispatching_num <= 0:
431             return
432         
433         if peer_have_pieces == None:
434             peer_have_pieces = peer_con.get_peer_have_pieces()
435         
436         peer_have_pieces = set(peer_have_pieces)
437 
438         if len(peer_have_pieces) == 0:
439             return
440 
441         waiting_pieces = set(self.__download_state['waiting-pieces'].keys())
442 
443         if len(waiting_pieces) == 0:
444             return
445         
446         dispatching_pieces = peer_have_pieces & waiting_pieces
447 
448         dispatching_pieces = list(dispatching_pieces)
449         dispatching_pieces = dispatching_pieces[0:dispatching_num]
450         if len(dispatching_pieces) == 0:
451             return
452         
453         if peer_con.dispatch_pieces(dispatching_pieces, self.__piece_length):
454             for piece_index in dispatching_pieces:
455                 self.__download_piece(piece_index, peer_con)
456             self.__write_log( 'dispatch_pieces:%s,%s' %(str(dispatching_pieces),str(peer_con.get_peer_addr())))
457         pass
458     
459     def __complete_piece(self, piece_index):
460         piece_state = self.__download_state['downloading-pieces'].pop(piece_index, None)
461         if piece_state == None:
462             return
463         self.__download_state['completed-pieces'][piece_index] = piece_state
464         piece_state['connect-info'] = None
465         self.__notify_peers_have([piece_index,])
466         self.__writingfile_pieces.append(piece_index)
467 
468     def __undownload_piece(self, piece_index):
469         piece_state = self.__download_state['downloading-pieces'].pop(piece_index, None)
470         if piece_state == None:
471             return
472         self.__download_state['waiting-pieces'][piece_index] = piece_state
473         
474         self.__download_state['left'] += piece_state['waiting-offset']
475         self.__download_state['downloaded'] -= piece_state['waiting-offset']
476         piece_state['waiting-offset'] = 0
477         piece_state['piece-data'] = ''
478         piece_state['connect-info'] = None
479 
480     def __download_piece(self, piece_index, peer_con):
481         piece_state = self.__download_state['waiting-pieces'].pop(piece_index, None)
482         if piece_state == None:
483             return
484         
485         self.__download_state['downloading-pieces'][piece_index] = piece_state
486         piece_state['waiting-offset'] = 0
487         piece_state['piece-data'] = ''
488         piece_state['connect-info'] = peer_con
489         
490     def __write_log(self,log):
491         log_info.write_log(log)
492         pass
493         
494 def _generate_peerid():
495     peerid = ''
496     for i in range(0,20): 
497         peerid += chr(random.randrange(ord('a'),ord('z')))
498     return peerid
499 
500 def _get_listening_addr():
501     host_ip = socket.gethostbyname(socket.gethostname())
502     host_port = 6881
503     return host_ip,host_port          
504                 
505 if __name__ == '__main__':
506     new_task = DownLoadTask()
507     new_task.start_task(r".\narodo.torrent")
508     time.sleep(100000)
509