Spark的standalone源码分析（五）

承接上文，本节继续介绍broadcast机制中的BitTorrentBroadcast；

3. BitTorrentBroadcast

BitTorrentBroadcast即采用BitTorrent的方式来广播变量；

大致流程如下：
1. 首先，work节点通过GuideMultipleRequests服务向master节点获得selectedSources；master节点通过random select的方式从当前的listOfSources选出sources返回；当然，刚开始的时候master的listOfSources只有master节点，随着work节点的connect，listOfSources会不断增加；因此，BitTorrentBroadcast的work节点会间歇的与master节点通信以update master节点的listOfSources到本地；

class TalkToGuide(gInfo: SourceInfo)
extends Thread with Logging {
override def run() {

// Keep exchaning information until all blocks have been received
while (hasBlocks.get < totalBlocks) {
talkOnce
Thread.sleep(MultiTracker.ranGen.nextInt(
MultiTracker.MaxKnockInterval - MultiTracker.MinKnockInterval) +
MultiTracker.MinKnockInterval)
}

// Talk one more time to let the Guide know of reception completion
talkOnce
}

2. 然后，将selectedSources加入该work节点的listOfSources中，在pickPeerToTalkToRandom方法中，选择一个SourceInfo；pickPeerToTalkToRandom首先将已经connect的节点从候选节点中去除，从剩下的节点中选择拥有当前节点没有的blocks最多的那个节点；

// Select the peer that has the most blocks that this receiver does not
peersNotInUse.foreach { eachSource =>
var tempHasBlocksBitVector: BitSet = null
hasBlocksBitVector.synchronized {
tempHasBlocksBitVector = hasBlocksBitVector.clone.asInstanceOf[BitSet]
}
tempHasBlocksBitVector.flip(0, tempHasBlocksBitVector.size)
tempHasBlocksBitVector.and(eachSource.hasBlocksBitVector)

if (tempHasBlocksBitVector.cardinality > curMax) {
curPeer = eachSource
curMax = tempHasBlocksBitVector.cardinality
}
}

3. 最后，根据2选择出来的SourceInfo，通过TalkToPeer，获取blocks；TalkToPeer会开启一个threadpool，向newPeerToTalkTo节点请求blocks数据，为防止重复的download block数据，会维护一个blocksInRequestBitVector位图，让其他的thread知道该block已经被同步；

// Let other threads know that blockToAskFor is being requested
blocksInRequestBitVector.synchronized {
blocksInRequestBitVector.set(blockToAskFor)
}