科技行者

行者学院 转型私董会 科技行者专题报道 网红大战科技行者

知识库

知识库 安全导航

至顶网软件频道应用软件客户端源码分析之三: StorageWrapper 类

客户端源码分析之三: StorageWrapper 类

  • 扫一扫
    分享文章到微信

  • 扫一扫
    关注官方公众号
    至顶头条

StorageWrapper 的作用:把文件片断进一步切割为子片断,并且为这些子片断发送 request消息。在获得子片断后,将数据写入磁盘。 请结合 Storage 类的分析来看。

作者:rstevens 来源:CSDN 2008年5月20日

关键字: 源码 客户端 python 软件

  • 评论
  • 分享微博
  • 分享邮件

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

请结合 Storage 类的分析来看。

 

几点说明:

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

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

3、 例如一个256k大小的片断,索引号是10,被划分为1616k大小的子片断。那么需要为这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 个数就要减1amount_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:

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

 

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

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

计算这段数据的消息摘要

                sp = sh.digest()

然后,更新 sh 这个 sha 对象,注意,是根据片断 i 剩下的数据来更新的。关于 sha::update() 的功能,请看 python的帮助。如果有两段数据 a b,那么

sh = sha(a)

sh.update(b),等效于

sh = sha(a+b)

所以,下面这个表达式等于

sh.update(self.storage.read(piece_size*i, self._piecelen(i)))

 

                sh.update(self.storage.read(piece_size * i + lastlen, self._piecelen(i) - lastlen))

所以,这次计算出来的就是片断i 的摘要

(原来的困惑:为什么不直接计算 i 的摘要,要这么绕一下了?后来分析清楚“空间分配算法”之后,这后面一段代码也就没有什么问题了。)

                s = sh.digest()

 

如果计算出来的摘要和 hashes[i] 一致(后者是从 torrent 文件中获得的),那么,这个片断有效且已经存在于磁盘上。

                if s == hashes[i]:

markgot(i, i)

                           

                elif targets.get(s)

and self._piecelen(i) == self._piecelen(targets[s][-1]):

                    markgot(targets[s].pop(), i)

                elif not self.have[len(hashes) - 1]

and sp == hashes[-1]

and (i == len(hashes) - 1 or not self._waspre(len(hashes) - 1)):

markgot(len(hashes) - 1, i)

 

                            else:

                    self.places[i] = i

                if flag.isSet():

                    return

                numchecked += 1

                statusfunc({'fractionDone': 1 - float(self.amount_left) / self.total_length})

 

# 如果所有片断都下载完了,那么结束。

        if self.amount_left == 0:

            finished()

 

 

# 检查某个片断,是否已经在磁盘上分配了空间,调用的是 Storage:: was_preallocated()

def _waspre(self, piece):

        return self.storage.was_preallocated(piece * self.piece_size,

 self._piecelen(piece))

 

    # 获取指定片断的长度,只有最后一个片断大小可能小于 piece_size

    def _piecelen(self, piece):

        if piece < len(self.hashes) - 1:

            return self.piece_size

        else:

return self.total_length - piece * self.piece_size

 

 

       # 返回剩余文件的大小

    def get_amount_left(self):

        return self.amount_left

 

       # 判断是否已经获得了一些文件片断

    def do_I_have_anything(self):

        return self.amount_left < self.total_length

 

       # 将指定片断切割为“子片断”

def _make_inactive(self, index):

 

       # 先获取该片断的长度

        length = min(self.piece_size, self.total_length - self.piece_size * index)

        l = []

        x = 0

 

# 为了获得更好的传输性能,BT把每个文件片断又分为更小的“子片断”,我们可以在 download.py 文件中 default 变量中,找到“子片断”大小的定义:

'download_slice_size', 2 ** 14,      "How many bytes to query for per request."

这里定义的“子片断”大小是16k

下面这个循环,就是将一个片断进一步切割为“子片断”的过程。

 

        while x + self.request_size < length:

            l.append((x, self.request_size))

x += self.request_size

        l.append((x, length - x))

 

# l 保存到 inactive_requests 这个列表中

        self.inactive_requests[index] = l

 

# 是否处于 endgame 模式,关于endgame模式,参加《Incentives Build Robustness in BitTorrent

    def is_endgame(self):

        return self.endgame

 

    def get_have_list(self):

        return self.have.tostring()

 

    def do_I_have(self, index):

        return self.have[index]

 

# 判断指定的片断,是否还有 request没有发出?如果有,那么返回 true,否则返回 false

    def do_I_have_requests(self, index):

        return not not self.inactive_requests[index]

 

为指定片断创建一个 request 消息,返回的是一个二元组,例如(32k, 16k),表示“子片断”的起始位置是 32k ,大小是 16k

    def new_request(self, index):

        # returns (begin, length)

# 如果还没有为该片断创建 request。,那么调用 _make_inactive() 创建 request列表。(inactive_requests[index] 初始化的值是1

        if self.inactive_requests[index] == 1:

self._make_inactive(index)

             

# numactive[index] 记录了已经为该片断发出了多少个 request

        self.numactive[index] += 1

        rs = self.inactive_requests[index]

 

       # inactive_request 中移出最小的那个request(也就是起始位置最小)。

        r = min(rs)

        rs.remove(r)

 

       # amount_inactive 记录了尚没有发出request的子片断总的大小。

        self.amount_inactive -= r[1]

 

       # 如果这是最后一个“子片断”,那么进入 endgame 模式

        if self.amount_inactive == 0:

self.endgame = T.rue

       # 返回这个 request

        return r

 

    def piece_came_in(self, index, begin, piece):

        try:

            return self._piece_came_in(index, begin, piece)

        except IOError, e:

            self.failed('IO Error ' + str(e))

            return True

 

如果获得了某个“子片断”,那么调用这个函数。

index:“子片断”所在片断的索引号,

begin:“子片断”在片断中的起始位置,

piece:实际数据

def _piece_came_in(self, index, begin, piece):

      

# 如果之前没有获得过该片断中任何“子片断”,那么首先需要在磁盘上为整个片断分配空间。

空间分配的算法如下:

假设一共是6个片断,现在已经为 014三个片断分配了空间,那么

holes[2, 3, 5]

places{0:0, 1:1, 4:4}

现在要为片断5分配空间,思路是把片断5的空间暂时先分配在片断2应该在的空间上。这样分配以后,

holes[3, 5]

places: {0:0, 1:1, 4:4, 5:2}

假设下一步为片断2分配空间,因为2的空间已经被5占用,所以把5的数据转移到3上,2才可以使用自己的空间。这样分配之后,

holes[5]

places{0:0, 1:1, 2:2, 4:4, 5:3}

最后,为3分配空间,因为3的空间被5占用,所以把5的数据转移到5自己的空间上,3就可以使用自己的空间了。这样分配之后,

holes[]

places{0:0, 1:1, 2:2, 3:3, 4:4, 5:5}

 

下面这段比较晦涩的代码,实现的就是这种空间分配算法。

 

        if not self.places.has_key(index):

n = self.holes.pop(0)

 

            if self.places.has_key(n):

                oldpos = self.places[n]

                old = self.storage.read(self.piece_size * oldpos, self._piecelen(n))

                if self.have[n] and sha(old).digest() != self.hashes[n]:

                    self.failed('data corrupted on disk - maybe you have two copies running?')

                    return True

                self.storage.write(self.piece_size * n, old)

                self.places[n] = n

                if index == oldpos or index in self.holes:

                    self.places[index] = oldpos

                else:

                    for p, v in self.places.items():

                        if v == index:

                            break

                    self.places[index] = index

                    self.places[p] = oldpos

                    old = self.storage.read(self.piece_size * index, self.piece_size)

self.storage.write(self.piece_size * oldpos, old)

                    

elif index in self.holes or index == n:

                if not self._waspre(n):

self.storage.write(self.piece_size * n,

self._piecelen(n) * chr(0xFF))

                self.places[index] = n

            else:

                for p, v in self.places.items():

                    if v == index:

                        break

                self.places[index] = index

                self.places[p] = n

                old = self.storage.read(self.piece_size * index, self._piecelen(n))

                self.storage.write(self.piece_size * n, old)

       

       # 调用 Stoarge::write() 将这个子片断写入磁盘,注意是写到 places[index] 所在的空间上。

self.storage.write(self.places[index] * self.piece_size + begin, piece)

 

# 既然获得了一个子片断,那么发出的request个数显然要减少一个。

        self.numactive[index] -= 1

       

# 如果既没有尚未发出的 request,而且也没有已发出的request(每当获得一个子片断,numactive[index]减少1numactive[index]0,说明所有发出的 request都已经接收到了响应的数据),那么显然整个片断已经全部获得了。

if not self.inactive_requests[index] and not self.numactive[index]:

       检查整个片断的有效性,如果通过检查

if sha(self.storage.read(self.piece_size * self.places[index],

 self._piecelen(index))).digest() == self.hashes[index]:

                    

#“我”已经拥有了这个片断

                self.have[index] = True

                self.inactive_requests[index] = None

                     # 也检查过了有效性

                self.waschecked[index] = True

                           

                self.amount_left -= self._piecelen(index)

                if self.amount_left == 0:

self.finished()

              如果没有通过有效性检查

            else:

                self.data_flunked(self._piecelen(index))

                     得丢弃这个片断

                self.inactive_requests[index] = 1

                self.amount_inactive += self._piecelen(index)

                return False

        return True

 

# 如果向某个 peer 发送的获取“子片断”的请求丢失了,那么调用此函数

    def request_lost(self, index, begin, length):

        self.inactive_requests[index].append((begin, length))

        self.amount_inactive += length

        self.numactive[index] -= 1

 

    def get_piece(self, index, begin, length):

        try:

            return self._get_piece(index, begin, length)

        except IOError, e:

            self.failed('IO Error ' + str(e))

            return None

 

    def _get_piece(self, index, begin, length):

        if not self.have[index]:

            return None

 

        if not self.waschecked[index]:

              # 检查片断的 hash值,如果错误,返回 None

if sha(self.storage.read(self.piece_size * self.places[index],

self._piecelen(index))).digest() != self.hashes[index]:

                self.failed('told file complete on start-up, but piece failed hash check')

                return None

              # 通过 hash 检查

self.waschecked[index] = True

 

       # 检查一下“子片断”长度是否越界

        if begin + length > self._piecelen(index):

return None

             

# 调用 Storage::read() ,将该“子片断”数据从磁盘上读出来,返回值就是这段数据。

        return self.storage.read(self.piece_size * self.places[index] + begin, length)

    • 评论
    • 分享微博
    • 分享邮件
    邮件订阅

    如果您非常迫切的想了解IT领域最新产品与技术信息,那么订阅至顶网技术邮件将是您的最佳途径之一。

    重磅专题
    往期文章
    最新文章