死磕以太坊源码分析之区块和交易广播
ProtocolManager详解
@H_403_7@ProtocolManager，从字面上看是协议管理器，负责着@H_403_7@p2p通信协议的管理。它连接了@H_403_7@p2p的逻辑层@H_403_7@peer与顶层@H_403_7@peer之间的调用，从顶层将协议传递至逻辑层，再从逻辑层得到@H_403_7@message传递到顶层。
- @H_403_7@fastSync规定了同步的模式 ；
- @H_403_7@acceptTxs是节点是否接受交易的阀门，只有当@H_403_7@pm.acceptTxs == 1时，节点才会接受交易。这个操作只会在同步结束后再开始，即同步的时候节点是不会接受交易的；
- @H_403_7@SubProtocols中是以太坊的通讯协议，通常只有一个值，即@H_403_7@eth63。
- @H_403_7@downloader是一个下载器，用于主动从远程节点中获取@H_403_7@hashes和@H_403_7@blocks。
- @H_403_7@fetcher则被动的收集网络其他以太坊节点发过来的同步通知，进行验证，并做出相应的处理。
@H_403_7@ProtocolManager.Start()启动了四条@H_403_7@go程，分别是交易订阅广播协程（@H_403_7@txBroadcastLoop）、挖矿订阅协程（@H_403_7@minedBroadcastLoop）、节点定期同步协程（@H_403_7@syncer）和交易同步协程（txsyncLoop）
- txBroadcastLoop:广播新出现的交易对象。@H_403_7@txBroadcastLoop()会在@H_403_7@txCh通道的收端持续等待，一旦接收到有关新交易的事件，会立即调用@H_403_7@BroadcastTx()函数广播给那些尚无该交易对象的相邻个体。
- minedBroadcastLoop:广播新挖掘出的区块。@H_403_7@minedBroadcastLoop()持续等待本节点的新挖掘出区块事件，然后立即广播给需要的相邻个体。当不再订阅新挖掘区块事件时，这个函数才会结束等待并返回。
- syncer:定时的和网络其他节点同步，并处理网络节点的相关通知。定时与相邻个体进行区块全链的强制同步。syncer()首先启动fetcher成员，然后进入一个无限循环，每次循环中都会向相邻peer列表中“最优”的那个peer作一次区块全链同步。发起上述同步的理由分两种：如果有新登记(加入)的相邻个体，则在整个peer列表数目大于5时，发起之；如果没有新peer到达，则以10s为间隔定时的发起之。这里所谓"最优"指的是peer中所维护区块链的TotalDifficulty(td)最高，由于Td是全链中从创世块到最新头块的Difficulty值总和，所以Td值最高就意味着它的区块链是最新的，跟这样的peer作区块全链同步，显然改动量是最小的，此即"最优"。
- txsyncLoop：把新的交易均匀的同步给网路节点。
广播的情形
- @H_403_7@minedBroadcastLoop()监听到新区块事件后，把新区块和区块@H_403_7@hash分别广播出去；
- 从远程节点同步完成后，将@H_403_7@CurrentBlock广播出去，此时广播的是区块@H_403_7@hash；
- @H_403_7@txBlockcastLoop()监听到区块池的新增交易事件时会广播交易；
广播区块及区块哈希
广播区块的入口在@H_403_7@pm.minedBroadcastLoop(),进入到@H_403_7@BroadcastBlock,这里的参数为@H_403_7@bool值，如果传入的为true，则将区块block和总难度td发送给一部分节点，节点数为根号n；如果传入的为false，则将区块的hash发送给所有的节点。需要注意的是两个广播函数都执行。
进入到@H_403_7@true分支：代表只传播区块给一部分节点
①：首先计算一个临时的TD
if parent := pm.blockchain.GetBlock(block.ParentHash(),block.NumberU64()-1); parent != nil {
td = new(big.Int).Add(block.Difficulty(),pm.blockchain.GetTd(block.ParentHash(),block.NumberU64()-1))
}
②：发送块到peers的子集
对节点数进行开方，16开方得4，然后取前4个节点。
transferLen := int(math.Sqrt(float64(len(peers))))
if transferLen < minBroadcastPeers {
transferLen = minBroadcastPeers
}
if transferLen > len(peers) {
transferLen = len(peers)
}
transfer := peers[:transferLen]
for _,peer := range transfer {
peer.AsyncSendNewBlock(block,td) // 块传播
}
执行完之后直接return出去，再次执行此函数，此时不会走ture分支，直接判断判断本地是否有区块，如果有则发送区区块哈希给剩下的节点，如果没有，则不做发送哈希的操作。
如果本地存在这个要广播的区块(很可能就是出块节点，或者接受块的节点已经插入到区块链中),那就还要像其他没有被广播到区块的节点发送区块哈希。
如果本地不存在这个要广播的区块哈希(应该是还没接收到区块或者区块哈希的节点)，那它只要向它的节点列表里发送区块即可。
接下来就是重点分析@H_403_7@AsyncSendNewBlock和@H_403_7@AsyncSendNewBlockHash两个函数了。
AsyncSendNewBlock
发送块到需要广播的节点的广播队列中
select {
case p.queuedProps <- &propEvent{block: block,td: td}:
p.knownBlocks.Add(block.Hash())
for p.knownBlocks.Cardinality() >= maxKnownBlocks {
p.knownBlocks.Pop()
}
这里的@H_403_7@queuedProps是用来存放要广播的块的队列，同时，要把广播的块标记为已知，还不能超过1024（maxKnownBlocks）个。超过就会弹出队列第一个@H_403_7@propEvent() 。接下来就是处理队列中的块了。
在@H_403_7@eth/peer.go中，有个专门处理广播的循环@H_403_7@brodcast
func (p *peer) broadcast() {
for {
select {
case txs := <-p.queuedTxs:
if err := p.SendTransactions(txs); err != nil {
return
}
p.Log().Trace("Broadcast transactions","count",len(txs))
case prop := <-p.queuedProps:
if err := p.SendNewBlock(prop.block,prop.td); err != nil {
return
}
p.Log().Trace("Propagated block","number",prop.block.Number(),"hash",prop.block.Hash(),"td",prop.td)
case block := <-p.queuedAnns:
if err := p.SendNewBlockHashes([]common.Hash{block.Hash()},[]uint64{block.NumberU64()}); err != nil {
return
}
p.Log().Trace("Announced block",block.Number(),block.Hash())
case <-p.term:
return
}
}
}
广播新块到远程节点
p.SendNewBlock(prop.block,prop.td);
远程节点收到块后同样也会标记哈希存入队列，并且不会超过最大，同时发送一个@H_403_7@NewBlockMsg，@H_403_7@msgcode为@H_403_7@0x07,同时数据会被RLP编码。
p.knownBlocks.Add(block.Hash())
for p.knownBlocks.Cardinality() >= maxKnownBlocks {
p.knownBlocks.Pop()
}
return p2p.Send(p.rw,NewBlockMsg,[]interface{}{block,td})
func Send(w MsgWriter,msgcode uint64,data interface{}) error {
size,r,err := rlp.EncodeToReader(data)
if err != nil {
return err
}
return w.WriteMsg(Msg{Code: msgcode,Size: uint32(size),Payload: r})
}
到此广播区块的过程结束，交由远程节点去处理@H_403_7@NewBlockMsg消息。
AsyncSendNewBlockHash
广播哈希的过程跟广播区块的过程非常的类似，最终是由远程节点去处理@H_403_7@NewBlockHashesMsg消息。
广播区块的过程完毕之后，会直接进入下一个阶段，调用@H_403_7@fetcher模块去同步这些广播的区块，接下的文章会讲到。
广播交易
广播交易的入口在@H_403_7@pm.txBroadcastLoop()，直接进入到@H_403_7@pm.BroadcastTxs(event.Txs)，大概做了以下几件事：
①：将交易广播给一批没有这个交易的节点
for _,tx := range txs {
peers := pm.peers.PeersWithoutTx(tx.Hash())
for _,peer := range peers {
txset[peer] = append(txset[peer],tx)
}
log.Trace("Broadcast transaction",tx.Hash(),"recipients",len(peers))
}
②：异步发送交易给这些节点
for peer,txs := range txset {
peer.AsyncSendTransactions(txs)
}
接着进入到@H_403_7@AsyncSendTransactions:
将所有交易标记为已知交易，同时还要保证没有超过最大的已知交易（32768笔）
case p.queuedTxs <- txs:
for _,tx := range txs {
p.knownTxs.Add(tx.Hash())
}
for p.knownTxs.Cardinality() >= maxKnownTxs {
p.knownTxs.Pop()
}
case txs := <-p.queuedTxs:
if err := p.SendTransactions(txs); err != nil {
return
}
func (p *peer) SendTransactions(txs types.Transactions) error {
...
return p2p.Send(p.rw,TxMsg,txs)
}
发送交易最终会发送一个@H_403_7@TxMsg消息，接收到这个消息的节点会通过@H_403_7@pm.txpool.AddRemotes(txs)处理交易。
消息处理（handleMsg）
@H_403_7@handleMsg从对方连接中读取消息，根据消息码的不同进行处理,从而将广播和同步之间来回的消息进行处理。
func (pm *ProtocolManager) handleMsg(p *peer) error {
msg,err := p.rw.ReadMsg()
if err != nil {
return err
}
if msg.Size > ProtocolMaxMsgSize {
return errResp(ErrMsgTooLarge,"%v > %v",msg.Size,ProtocolMaxMsgSize)
}
defer msg.Discard()
switch {
case msg.Code == StatusMsg: ......
case msg.Code == GetBlockHeadersMsg: ......
case msg.Code == BlockHeadersMsg: ......
case msg.Code == GetBlockBodiesMsg: ......
case msg.Code == BlockBodiesMsg: ......
case p.version >= eth63 && msg.Code == GetNodeDataMsg: ......
case p.version >= eth63 && msg.Code == NodeDataMsg: ......
case p.version >= eth63 && msg.Code == GetReceiptsMsg: ......
case p.version >= eth63 && msg.Code == ReceiptsMsg: ......
case msg.Code == NewBlockHashesMsg: ......
case msg.Code == NewBlockMsg: ......
case msg.Code == TxMsg: ......
default: return errResp(ErrInvalidMsgCode,"%v",msg.Code)
}
return nil
}