Ctorrent: 寻找peer 算法分析

Ctorrent:

寻找
peer
算法分析

前言

网上有一篇很经典的
ctorrent
分析文章，虽然是用的
1.3
的源码分析的，但是相对我现在看的
3.3.2
的源码，

也只有很少一部分有差异，在寻找
peer
部分比较明显。

这篇文档的作用是补充完整
BT
协议描述的实现寻找
peer
的算法。

首先描述算法再结合源码具体分析

BT
下载之所以性能出众是由
BT
协议所规定的一系列机制所保证的。判断一个
BT
下载软件性能优秀与否则是看这个软件对
BT
协议中下载机制的执行情况。
BT
协议主要规定了两大类机制保证其性能（详细信息请参照
”
Incentives Build Robustness in BitTorrent”
）：

3.4.1 P
iece
选择机制

3.4.1
.1
初始模式（
Initial
Mode
）：
Random First
Piece
。

当客户端刚开始运行时，它一个完整的
piece
也没有，这时需要尽快下载到一个
piece
以便可以提供上传服务。此时的算法为：第一个随机
piece
。客户端会随机找到一个
piece
，然后下载。

CTorrent
随机选择
piece
，而且更进一步采取了一种加速下载的办法：虽然此时客户端没有
piece
，但应该有向其它
peer
的申请
slice
的队列了。客户端只要比较这些队列哪个最短，优先下载最短的队列即可最快获得第一个
piece
。（这个是用的位图来实现的）

函数
PeerList::Who_Can_Duplicate()

实现了此算法的代码。

3.4.1
.2
一般模式（
Normal
Mode
）：
Strict
Priority
和
Rarest First
。

1
，严格优先（
Strict Priority
）

一旦某个
slice
被申请，则这个
slice
所在的
piece
中的其它
slice
会先于其它
piece
的
slice
被申请。这样做可以尽量使最先申请的
piece
最先被下载完毕。

这条规则看似简单而且公平，但实现起来非常困难：
BT
协议规定一个
piece
中的多个
slice
可以向多个
peer
申请，而客户端又同时从多个
peer
处申请了多个
piece
中的
slice
，这么多数据传输队列同时进行，要保证严格优先是非常困难的。

CTorrent
在设计时采取了一种比较简单的方法：
一旦向某个
peer
申请了某个
slice
，则这个
piece
中的所有
slice
均向这个
peer
申请。为了保证尽快将一个
piece
下载完成，
CTorrent
会找出当前正在与之通信的那个
peer
（正在通信的
peer
通常比较活跃），然后把所有
slice
请求队列中最慢的那个队列找出来，交给这个
peer
去下载。

函数
PeerList::Who_Can_Abandon()
实现了此算法的代码。

2
，最少优先（
Rarest First
）

客户端下载时选择所有
peer
拥有最少的那个
piece
优先下载。

函数
BitField::Random()
是有关
piece
选择机制的代码，但它只是随机选择了
piece
，没有实现最少优先。

3.4.1
.3
结束模式
(Endgame
Mode)

由于每一个
piece
只向一个
peer
申请
，当
peer
数大于还没有申请的
piece
数时，客户端便进入了结束模式。此时客户端可以向所有的
peer
发送还没有下载完毕的
slice
的请求，以便尽快下载完毕好做种子。

CTorrent
变相实现了这个算法，它会找出当前正在与之通信的那个
peer
（正在通信的
peer
通常比较活跃），然后把所有
slice
请求队列中最长的那个队列找出来，交给这个
peer
去下载。

函数
PeerList::Who_Can_Duplicate()
实现了此算法的代码。

函数
PeerList::Who_Can_Duplicate()
和
PeerList::Who_Can_Abandon()
的调用环境均是函数
btPeer::RequestPiece()
，应将这三个函数一起查看才能清楚
piece
选择机制的实现。

//
函数主要检查需要向
peer
请求哪一个
piece
,
分几种情况

/*
首先检查
request_q
的空间是否足够，一种情况是查看
PENDINGQUEUE
中是否有可以向
peer
请求的数
?

2:
在不为初始化阶段是接受到
have
消息

3:

*/

int btPeer::RequestPiece()

{

size_t idx;

BitField tmpBitfield, *pfilter;

int endgame = 0;

size_t qsize = request_q.Qsize();

size_t psize = BTCONTENT.GetPieceLength() /
cfg_req_slice_size;

// See if there's room in the queue for a new
piece.

// Also, don't queue another piece if we
still have a full piece queued.

//
在里涉及了具体的请求队列的管理方法

if( cfg_req_queue_length - qsize < psize
|| qsize >= psize ){

m_req_send = m_req_out;
// don't come back until you receive
something.

return 0;

}

tmpBitfield = bitfield;

tmpBitfield.Except(*BTCONTENT.pBMasterFilter);

if( m_last_req_piece <
BTCONTENT.GetNPieces() && tmpBitfield.Count() > 1 )

tmpBitfield.UnSet(m_last_req_piece);

//
查看
PENDINGQUEUE
中是否有可以向
peer
请求的数据

if( (idx =
PENDINGQUEUE.ReAssign(&request_q, tmpBitfield)) <

BTCONTENT.GetNPieces() ){

if(arg_verbose)

CONSOLE.Debug("Assigning #%d to %p
from Pending", (int)idx, this);

/* If another peer has the same slice
requested first, move the proposer's

slice to the last position for the piece. */

//

if( BTCONTENT.pBMultPeer->IsSet(idx) )

WORLD.CompareRequest(this, idx);

BTCONTENT.pBMultPeer->Set(idx);

return SendRequest();

}

//

接收到
have

消息

不在初始化阶段

if( m_cached_idx <
BTCONTENT.CheckedPieces() && !BTCONTENT.pBF->IsEmpty() ){

// A HAVE msg already selected what we want
from this peer

// but ignore it in initial-piece mode.

idx = m_cached_idx;

m_cached_idx = BTCONTENT.GetNPieces();

/*

若客户端没有这个
HAVE
消息声明的
piece
，

且
PENDINGQUEUE
中没有这个
piece
，且还没有向别的
peer
请求这个
piece
，

则发送
request
消息

*/

if( !BTCONTENT.pBF->IsSet(idx)
&&

(!BTCONTENT.GetFilter() ||
!BTCONTENT.GetFilter()->IsSet(idx)) &&

!PENDINGQUEUE.Exist(idx) &&

!WORLD.AlreadyRequested(idx) ){

if(arg_verbose)
CONSOLE.Debug("Assigning #%d to %p", (int)idx, this);

return (request_q.CreateWithIdx(idx) <
0) ? -1 : SendRequest();

}

}

// If we didn't want the cached piece, select
another.

if( BTCONTENT.pBF->IsEmpty() ){

// If we don't have a complete piece yet,
try to get one that's already

// in progress.
(Initial-piece mode)

pfilter = BTCONTENT.GetFilter();

do{

tmpBitfield = bitfield;

if( pfilter ){

tmpBitfield.Except(*pfilter); // what
we can download

pfilter =
BTCONTENT.GetNextFilter(pfilter);

}

}while( pfilter &&
tmpBitfield.IsEmpty() );

if( m_latency < 60 ){

// Don't dup to very slow/high latency
peers.

if( m_last_req_piece <
BTCONTENT.GetNPieces() && tmpBitfield.Count() > 1 )

tmpBitfield.UnSet(m_last_req_piece);

//
则寻找拥有可以最快获得的
piece
的那个
peer

//
下面语句是选择是那个
piece
，

//
tmpbitfield : what we can
download

idx =
WORLD.What_Can_Duplicate(tmpBitfield, this, BTCONTENT.GetNPieces());

if( idx < BTCONTENT.GetNPieces() ){

if(arg_verbose) CONSOLE.Debug("Want
to dup #%d to %p", (int)idx, this);

//
选择是那个
peer

btPeer *peer = WORLD.WhoHas(idx);

if(peer){

if(arg_verbose)
CONSOLE.Debug("Duping #%d from %p to %p",

(int)idx, peer, this);

//
把这个
peer
的
request_q
拷贝给当前
peer

if(
request_q.CopyShuffle(&peer->request_q, idx) < 0 ) return -1;

WORLD.CompareRequest(this, idx);

BTCONTENT.pBMultPeer->Set(idx);

return SendRequest();

}

}else if(arg_verbose)
CONSOLE.Debug("Nothing to dup to %p", this);

}

}

// Doesn't have a piece that's already in
progress--choose another.

pfilter = BTCONTENT.GetFilter();

do{

tmpBitfield = bitfield;

tmpBitfield.Except(*BTCONTENT.pBF);

if( pfilter ){

tmpBitfield.Except(*pfilter);

pfilter =
BTCONTENT.GetNextFilter(pfilter);

}

// Don't go after pieces we might already
have (but don't know yet)

tmpBitfield.And(*BTCONTENT.pBChecked);

// tmpBitfield tells what we need from this
peer...

}while( pfilter && tmpBitfield.IsEmpty()
);

if( m_last_req_piece <
BTCONTENT.GetNPieces() && tmpBitfield.Count() > 1 )

tmpBitfield.UnSet(m_last_req_piece);

if( tmpBitfield.IsEmpty() ){

// We don't need to request anything from
the peer.

if( !Need_Remote_Data() )

return SetLocal(M_NOT_INTERESTED);

else if( m_last_req_piece <
BTCONTENT.GetNPieces() &&

!BTCONTENT.pBF->IsSet(m_last_req_piece) ){

// May have excluded the only viable
request; allow a retry.

m_last_req_piece = BTCONTENT.GetNPieces();

return 0;

// Allow another peer a shot at it first.

}else{

if(arg_verbose) CONSOLE.Debug("%p
standby", this);

m_standby = 1;
// nothing to do at the moment

return 0;

}

}

BitField tmpBitfield2 = tmpBitfield;

WORLD.CheckBitField(tmpBitfield2);

// [tmpBitfield2] ...that we haven't
requested from anyone.

if( tmpBitfield2.IsEmpty() ){

// Everything this peer has that I want,
I've already requested.

int endgame = WORLD.Endgame();

if( endgame && m_latency < 60 ){

// OK to duplicate a request, but not to
very slow/high latency peers.

//
idx = tmpBitfield.Random();

idx = 0;

// flag for Who_Can_Duplicate()

//
这里
tmpbitfield2
为空，所有
tmpbitfield3

也为空，

BitField tmpBitfield3 = tmpBitfield2;

idx =
WORLD.What_Can_Duplicate(tmpBitfield3, this, idx);

if( idx < BTCONTENT.GetNPieces() ){

if(arg_verbose)
CONSOLE.Debug("Want to dup #%d to %p",

(int)idx, this);

btPeer *peer = WORLD.WhoHas(idx);

if(peer){
// failsafe

if(arg_verbose)
CONSOLE.Debug("Duping #%d from %p to %p",

(int)idx, peer, this);

if(
request_q.CopyShuffle(&peer->request_q, idx) < 0 ) return -1;

WORLD.CompareRequest(this, idx);

BTCONTENT.pBMultPeer->Set(idx);

return SendRequest();

}

}

}

btPeer *peer;

if( request_q.IsEmpty() && (peer =
WORLD.Who_Can_Abandon(this)) ){

// Cancel a request to the slowest peer
& request it from this one.

idx =
peer->FindLastCommonRequest(bitfield);

if(arg_verbose)
CONSOLE.Debug("Reassigning #%d from %p to %p",

(int)idx, peer, this);

// RequestQueue class "moves"
rather than "copies" in assignment!

if(
request_q.Copy(&peer->request_q, idx) < 0 ) return -1;

WORLD.CompareRequest(this, idx);

BTCONTENT.pBMultPeer->Set(idx);

if( endgame ) peer->UnStandby();

if( peer->CancelPiece(idx) < 0 ) peer->CloseConnection();

return SendRequest();

}else if( BTCONTENT.CheckedPieces() >=
BTCONTENT.GetNPieces() ){

if(arg_verbose) CONSOLE.Debug("%p
standby", this);

m_standby = 1;
// nothing to do at the moment

}

}else{

// Request something that we haven't
requested yet (most common case).

// Try to make it something that has good
trade value.

BitField tmpBitfield3 = tmpBitfield2;

WORLD.FindValuedPieces(tmpBitfield3, this,
BTCONTENT.pBF->IsEmpty());

if( tmpBitfield3.IsEmpty() ) tmpBitfield3 =
tmpBitfield2;

idx = tmpBitfield3.Random();

if(arg_verbose)
CONSOLE.Debug("Assigning #%d to %p", (int)idx, this);

return (request_q.CreateWithIdx(idx) <
0) ? -1 : SendRequest();

}

return 0;

}

----------------------------------------------------------------------------------------------------------------------

/*

调用此函数的前提条件是客户端向
proposer
的索取队列
request_q
为空。

当客户端一个
piece
也没有时，需要尽快获取一个
piece
以便有数据提供给别的
peer
下载。这种状态称为
initial-piece mode
。此时与其新建一个
request_q
，不如检查
PeerList
链表中的所有
peer
的
request_q
，选取长度最短的那个
request_q
，将其拷贝（在函数体外调用
RequestQueue::CopyShuffle()
）给当前正在通信的
peer
（即
proposer
）的
request_q
，以便尽快下载数据。

BT
协议允许从不同的
peer
处获得同一个
piece
中的不同的
slice
，但是为了程序设计上的方便，
CTorrent
采取从一个
peer
获取整个
piece
的方法。所以长度最短的那个
request_q
表明这个
piece
中已经有最多的
slice
被下载下来了，此时复制这个索取队列可以尽快下载完剩下的
slice
以便获得一整个
piece
。

当客户端已经下载了很多
piece
，还需要得到的
piece
数小于
peer
数时（此时已无法向每一个
peer
请求一个不同的
piece
了，因为
peer
太多而需要的
piece
太少），程序进入
endgame mode
。此时可以向许多
peer
请求一个
slice
，这样可以加速完成下载从而开始做种。因此函数遍历所有
peer
，找出长度最大的那个
request_q
，将其拷贝（在函数体外调用
RequestQueue::CopyShuffle()
）给当前正在通信的
peer
（即
proposer
）的
request_q
，以便尽快下载数据。

长度最大的那个
request_q
表明
request_q
正在请求的这个
piece
的下载工作
"
落后
"
了，此时复制这个索取队列可以从其它
peer
获得帮助尽快下载完这个
piece
。

*/

// This
takes an index parameter to facilitate modification of the function to

// allow
targeting of a specific piece.
It's
currently only used as a flag to

//
specify endgame or initial-piece mode though.

//
bf

是
peer
的位图，是大家都要求的
piece
的位图

//
返回所选择的
piece
的
idex

size_t
PeerList::What_Can_Duplicate(BitField &bf, const btPeer *proposer,

size_t idx)

{

struct qdata {

size_t idx, qlen, count;

};

struct qdata *data;

int endgame, pass, i, mark;

PEERNODE *p;

PSLICE ps;

size_t slots, piece, qsize;

double work, best;

//
<
优先级高于

=

endgame = idx <
BTCONTENT.GetNPieces();
// else
initial-piece mode

//?

slots = endgame ? BTCONTENT.GetNPieces() -
BTCONTENT.pBF->Count() :

m_downloads * 2;

if( slots < m_dup_req_pieces + 2 ) slots =
m_dup_req_pieces + 2;

data = new struct qdata[slots];

#ifndef
WINDOWS

if( !data ) return BTCONTENT.GetNPieces();

#endif

// In initial mode, only dup a piece with
trade value.

// In endgame mode, dup any if there are no
pieces with trade value.

FindValuedPieces(bf, proposer, !endgame);

if( bf.IsEmpty() ){

if(endgame) bf = proposer->bitfield;

else return BTCONTENT.GetNPieces();

}

// initialize

data[0].idx = BTCONTENT.GetNPieces();

data[0].qlen = 0;

data[0].count = 0;//
表示相应相应的
slot
被几个不同的
piece

映射

for( i = 1; i < slots; i++ )

memcpy(data + i, data, sizeof(struct
qdata));

// measure applicable piece request queues

//
统计其他
peer
的的
request
的情况

for( p = m_head; p; p = p->next ){

if( !PEER_IS_SUCCESS(p->peer) ||
p->peer == proposer ||

p->peer->request_q.IsEmpty() )

continue;

piece = BTCONTENT.GetNPieces();

ps = p->peer->request_q.GetHead();

for( ; ps; ps = ps->next ){

//?

if( piece == ps->index ||

!bf.IsSet(ps->index) ||
proposer->request_q.HasIdx(ps->index) )

continue;

piece = ps->index;

qsize =
p->peer->request_q.Qlen(piece);

// insert queue data into array at (idx %
slots)

pass = 0;

i = piece % slots;

/*
对于每一个
piece
，根据
idx % slots
的值插入相应的
slot
，不同的
idx
的

可能会得到相同的值，举例
:

总共有
19
个
piece
,
索引为
0-18
,
假设

slot
的值为
5
，这样对

第四个
slot(data[3])
可能有
3
，
8
，
13
，
18
相同，

假如已经在里面插入了
idx
8
，当我们要插入
18
的时候就会跳到后面一个

slot(data[4])
看是否能够插入
(
注意这里的
data[4]
已经是最后一个
slot
了
)
，

如果还不能够插入
()
，会从头开始寻找
( data[0])
，最多寻找
2
遍

(
变量
pass
表示的
)

*/

while( data[i].idx <
BTCONTENT.GetNPieces() && pass < 2 ){

//
注意这里的
piece
== data[i].idx

和上面一个循环的
piece == ps->index
的区别

if( piece == data[i].idx ) break;

i++;

if( i >= slots ){

i = 0;

pass++;

}

}

if( pass < 2 ){

if( data[i].idx ==
BTCONTENT.GetNPieces() ){

data[i].idx = piece;

data[i].qlen = qsize;

}

//
这个只有当不同的
peer
中有相同的
piece
请求的时候才会
jia

data[i].count++;

}

}

} // end of measurement loop

/* Find the best workload for
initial/endgame.

In endgame mode, request the piece that
should take the longest.

In initial mode, request the piece that
should complete the fastest. */

//
初始化赋值

best = endgame ? 0 :
BTCONTENT.GetPieceLength() / cfg_req_slice_size + 2;

mark = slots;

/*
这里
work

的越大表示对相应的
piece
请求的
slice
的长度越长，向其它
peer
请求的

越少

，这样
work
越大，在
endgame

模式中越适合

*/

for( i = 0; i < slots; i++ ){

if( data[i].idx == BTCONTENT.GetNPieces() )
continue;

work = data[i].qlen /
(double)(data[i].count);

if( work > 1 && (endgame ? work
> best : work < best) ){

best = work;

mark = i;

}

}

if( mark < slots &&
data[mark].count == 1 ) m_dup_req_pieces++;

CONSOLE.Debug("%d dup req pieces",
(int)m_dup_req_pieces);

delete []data;

return (mark < slots) ? data[mark].idx :
BTCONTENT.GetNPieces();

}

//
处理输入入的参数
bf

:// bf is now pertinent pieces that not everyone has

/*

函数的作用
:
在初始化模式
(
将
bf
设置成为至少两个
peer
拥有，但是不是全部的
peer
都

拥有，其他情况则将
bf
设置成为只有
(proposer)
才有

*/

void
PeerList::FindValuedPieces(BitField &bf, const btPeer *proposer,

int initial) const

{

PEERNODE *p;

//
这里的
bf_others_have,
bf_only_he_has
的意思

// bf_others_have :

所有连接到
client
的
peer(
除去正在通信的那一个
)
，

//bf_only_he_has :
代表正在通信那一个
peer

BitField bf_all_have = bf, bf_int_have = bf,

bf_others_have, bf_only_he_has = bf,
*pbf_prefer;

for( p = m_head; p; p = p->next ){

if( !PEER_IS_SUCCESS(p->peer) ||
p->peer == proposer ) continue;

if( p->peer->Need_Remote_Data() )

bf_int_have.And(p->peer->bitfield);

bf_all_have.And(p->peer->bitfield);

if( !initial &&
!p->peer->bitfield.IsFull() )

bf_only_he_has.Except(p->peer->bitfield);

else
bf_others_have.Comb(p->peer->bitfield);

}

/* bf_all_have is now pertinent pieces that
all peers have

bf_int_have is pertinent pieces that all
peers in which I'm interested have

We prefer to get pieces that those peers
need, if we can.
Otherwise go

for pieces that any peer needs in hopes of
future reciprocation. */

if( !bf_int_have.IsFull() )

bf_all_have = bf_int_have;

bf_all_have.Invert();

bf.And(bf_all_have); // bf is now pertinent
pieces that not everyone has

pbf_prefer = initial ? &bf_others_have :
&bf_only_he_has;

BitField tmpBitField = bf;

tmpBitField.And(*pbf_prefer);

/* If initial mode, tmpBitField is now
pertinent pieces that more than one

peer has, but not everyone.

Otherwise, it's pertinent pieces that only
the proposer has (not

considering what other seeders have).

In either case if there are no such
pieces, revert to the simple answer.*/

if( !tmpBitField.IsEmpty() ) bf =
tmpBitField;

}

----------------------------------------------------------------------------------------------------------------------

/*

当程序工作在正常状态（既不处于initial piece mode也不处于endgame mode），

且需要向某一个pere（即proposer）发送request信息，

而这个peer的请求队列request_q又为空时，

调用Who_Can_Abandon()找出下载最慢的那个peer。

程序会在函数体外给这个peer发送cancel信息，

并把这个peer的request_q拷贝给proposer。

*/

/*

这个就是一般模式下的严格优先，最少优先也是实现了的

但是和注释是有出入的

*/

btPeer* PeerList::Who_Can_Abandon(btPeer *proposer)

{

PEERNODE *p;

btPeer *peer = (btPeer*) 0;

PSLICE ps;

size_t idx;

for( p = m_head; p; p = p->next ){

//寻找peer，条件 是与客服端有正常的连接并且不能够为proposfer

//并且该peer的索取队列不能够为空(我们寻找peer的目的就是要寻找

//下载速度最慢的peer，将它的索取队列复制到proposer 中)

if(!PEER_IS_SUCCESS(p->peer) || p->peer == proposer ||

p->peer->request_q.IsEmpty() ) continue;

//这里是判断速度的，为什么 要1.5?

if( (peer && p->peer->NominalDL() < peer->NominalDL()) ||

(!peer && p->peer->NominalDL() * 1.5 < proposer->NominalDL()) ){

//找到了满足条件的peer 并且返回请求表中第一个的piece 号

idx = p->peer->request_q.GetRequestIdx();

//如果proposer 中含有该piece 并且以前并没有request , 找到相应的peer

if( proposer->bitfield.IsSet(idx) && !proposer->request_q.HasIdx(idx) )

peer = p->peer;

// 如果请求队列的第一个piece 不满足上面的条件，那么就搜索请求队列的下一个piece，

// 找到并返回该peer

else{

ps = p->peer->request_q.GetHead();

for( ; ps; ps = ps->next ){

if( idx == ps->index ) continue;

idx = ps->index;

if( proposer->bitfield.IsSet(idx) &&

!proposer->request_q.HasIdx(idx) ){

peer = p->peer;

break;

}

}

}

}

} //end for

if( peer && arg_verbose )

CONSOLE.Debug("Abandoning %p (%d B/s) for %p (%d B/s)",

peer, peer->NominalDL(), proposer, proposer->NominalDL());

return peer;

}