BT客户端源码分析之三（1）：StorageWrapper 类

转载自：http://blueidea.bokee.com/1497172.html

作者：小马哥

日期：2004-6-30

StorageWrapper 的作用：把文件片断进一步切割为子片断，并且为这些子片断发送 request消息。在获得子片断后，将数据写入磁盘。

请结合 Storage 类的分析来看。

几点说明：

1、 为了获取传输性能，BT把文件片断切割为多个子片断。

2、 BT为获取一个子片断，需要向拥有该子片断的peer发送request消息（关于 request消息，参见《BT协议规范》）。

3、 例如一个256k大小的片断，索引号是10，被划分为16个16k大小的子片断。那么需要为这16个子片断分别产生一个 request 消息。这些request消息在发出之前，以list的形式保存在 inactive_requests 这个list中。例如对这个片断，就保存在inactive_requests下标为 10（片断的索引号）的地方，值是如下的 list：[(0,16k),(16k, 16k), (32k, 16k), (48k, 16k), (64k, 16k), (80k, 16k), (96k, 16k), (112k, 16k), (128k, 16k), (144k, 16k), (160k, 16k), (176k, 16k), (192k, 16k), (208k, 16k), (224k, 16k), (240k, 16k)]。这个处理过程在 _make_inactive() 函数中。因为这些request还没有发送出去，所以叫做 inactive request（未激活的请求）。如果一个 request 发送出去了，那么叫做 active request。为每个片断已经发送出去的request个数记录在 numactive 中。如果收到一个子片断，那么 active request 个数就要减1。amount_inactive 记录了尚没有发出request的子片断总的大小。

4、 每当获得一个子片段，都要写入磁盘。如果子片断所属的片断在磁盘上还没有分配空间，那么首先需要为整个片断分配空间。如何为片断分配空间？这正是 StorageWrapper 类中最难理解的一部分代码。这个“空间分配算法”说起来很简单，但是在没有任何注释的情况下去看代码，耗费了我好几天的时间。具体的算法分析，请看 _piece_came_in() 的注释。

class StorageWrapper:

def __init__(self, storage, request_size, hashes,

piece_size, finished, failed,

statusfunc = dummy_status, flag = Event(), check_hashes = True,

data_flunked = dummy_data_flunked):

self.storage = storage # Storage 对象

self.request_size = request_size #子片断大小

self.hashes = hashes # 文件片断摘要信息

self.piece_size = piece_size # 片断大小

self.data_flunked = data_flunked # 一个函数，用来检查片断的完整性

self.total_length = storage.get_total_length() # 文件总大小

self.amount_left = self.total_length # 未下载完的文件大小

# 文件总大小的有效性检查

# 因为最后一个片断长度可能小于 piece_size

if self.total_length <= piece_size * (len(hashes) - 1):

raise ValueError, 'bad data from tracker - total too small'

if self.total_length > piece_size * len(hashes):

raise ValueError, 'bad data from tracker - total too big'

# 两个事件，分布在下载完成和下载失败的时候设置

self.finished = finished

self.failed = failed

这几个变量的作用在前面已经介绍过了。

self.numactive = [0] * len(hashes)

inactive_request

inactive_requests 的值全部被初始化为1，这表示每个片断都需要发送 request。后面在对磁盘文件检查之后，那些已经获得的片断，在 inactive_requests中对应的是 None，表示不需要再为这些片断发送 request了。

self.inactive_requests = [1] * len(hashes)

self.amount_inactive = self.total_length

# 是否进入 EndGame 模式？关于 endgame 模式，在《Incentives Build Robustness in BitTorrent 》的“片断选择算法”中有介绍。后面可以看到，在为最后一个“子片断”产生请求后，进入 endgame 模式。

self.endgame = False

self.have = Bitfield(len(hashes))

# 该片是否检查了完整性

self.waschecked = [check_hashes] * len(hashes)

这两个变量用于“空间分配算法”

self.places = { }

self.holes = [ ]

if len(hashes) == 0:

finished()

return

targets = {}

total = len(hashes)

# 检查每一个片断，，，

for i in xrange(len(hashes)):

# 如果磁盘上，还没有完全为这个片断分配空间，那么这个片断需要被下载，在 targets 字典中添加一项（如果已经存在，就不用添加了），它的关键字（key）是该片断的摘要值，它的值（value）是一个列表， 这个片断的索引号被添加到这个列表中。

这里一度让我非常迷惑，因为一直以为不同的文件片断肯定具有不同的摘要值。后来才想明白了，那就是：两个不同的文件片断，可能拥有相同的摘要值。不是么？只要这两个片断的内容是一样的。

这一点，对后面的分析非常重要。

if not self._waspre(i):

targets.setdefault(hashes[i], []).append(i)

total -= 1

numchecked = 0.0

if total and check_hashes:

statusfunc({"activity" : 'checking existing file', "fractionDone" : 0})

# 这是一个内嵌在函数中的函数。在 c++ 中，可以有内部类，不过好像没有内部函数的说法。这个函数只能在 __init__() 内部使用。

这个函数在一个片段被确认获得后调用

# piece: 片断的索引号

# pos: 这个片断在磁盘上存储的位置

例如，片断5可能存储在片断2的位置上。请参看后面的“空间分配算法”

def markgot(piece, pos, self = self, check_hashes = check_hashes):

self.places[piece] = pos

self.have[piece] = True

self.amount_left -= self._piecelen(piece)

self.amount_inactive -= self._piecelen(piece)

不用再为这个片断发送 request消息了

self.inactive_requests[piece] = None

self.waschecked[piece] = check_hashes

lastlen = self._piecelen(len(hashes) - 1) # 最后一个片断的长度

# 对每一个片断

for i in xrange(len(hashes)):

#如果磁盘上，还没有完全为这个片断分配空间，那么在 holes 中添加该片断的索引号。

if not self._waspre(i):

self.holes.append(i)

# 否则，也就是空间已经分配。但是还是不能保证这个片断已经完全获得了，正如分析 Storage 时提到的那样，可能存在“空洞”

# 如果不需要进行有效性检查，那么简单调用 markgot() 表示已经获得了该片断。这显然是一种不负责任的做法。

elif not check_hashes:

markgot(i, i)

# 如果需要进行有效性检查

else:

sha是python内置的模块，它封装了 SHA-1摘要算法。SHA-1摘要算法对一段任意长的数据进行计算，得出一个160bit （也就是20个字节）长的消息摘要。在 torrent 文件中，保存了每个片断的消息摘要。接收方在收到一个文件片断之后，再计算一次消息摘要，然后跟 torrent 文件中对应的值进行比较，如果结果不一致，那么说明数据在传输过程中发生了变化，这样的数据应该被丢弃。

这里，首先，根据片断i的起始位置开始，lastlen长的一段数据构造一个 sha 对象。

sh = sha(self.storage.read(piece_size * i, lastlen))

计算这段数据的消息摘要

sp = sh.digest()