以太坊源码分析:fetcher模块和区块传播
刘艳琴
发表于 2022-12-21 01:53:48
162
0
0
当前代码是以太坊Release 1.8,如果版本不同,代码上可能存在差异。
总体过程和传播策略$ g4 ~. g( O. f! T3 O8 `
本节从宏观角度介绍,节点产生区块后,为了传播给远端节点做了啥,远端节点收到区块后又做了什么,每个节点都连接了很多Peer,它传播的策略是什么样的?
总体流程和策略可以总结为,传播给远端Peer节点,Peer验证区块无误后,加入到本地区块链,继续传播新区块信息。具体过程如下。$ ~& }; |1 U% A! H: `6 I
先看总体过程。产生区块后,miner模块会发布一个事件NewMinedBlockEvent,订阅事件的协程收到事件后,就会把新区块的消息,广播给它所有的peer,peer收到消息后,会交给自己的fetcher模块处理,fetcher进行基本的验证后,区块没问题,发现这个区块就是本地链需要的下一个区块,则交给blockChain进一步进行完整的验证,这个过程会执行区块所有的交易,无误后把区块加入到本地链,写入数据库,这个过程就是下面的流程图,图1。6 i6 T; X9 O1 e* F; n, Q
总体流程图,能看到有个分叉,是因为节点传播新区块是有策略的。它的传播策略为:
假如节点连接了N个Peer,它只向Peer列表的sqrt(N)个Peer广播完整的区块消息。向所有的Peer广播只包含区块Hash的消息。
策略图的效果如图2,红色节点将区块传播给黄色节点:
' a; j& Y+ s% T/ a
- @6 z2 F5 P' z& i
收到区块Hash的节点,需要从发送给它消息的Peer那里获取对应的完整区块,获取区块后就会按照图1的流程,加入到fetcher队列,最终插入本地区块链后,将区块的Hash值广播给和它相连,但还不知道这个区块的Peer。非产生区块节点的策略图,如图3,黄色节点将区块Hash传播给青色节点:
, q0 Q) ^, b) Z9 |
至此,可以看出以太坊采用以石击水的方式,像水纹一样,层层扩散新产生的区块。
Fetcher模块是干啥的) d/ k- _, L1 \3 b$ R0 g) R' o
fetcher模块的功能,就是收集其他Peer通知它的区块信息:1)完整的区块2)区块Hash消息。根据通知的消息,获取完整的区块,然后传递给eth模块把区块插入区块链。
如果是完整区块,就可以传递给eth插入区块,如果只有区块Hash,则需要从其他的Peer获取此完整的区块,然后再传递给eth插入区块
源码解读
本节介绍区块传播和处理的细节东西,方式仍然是先用图解释流程,再是代码流程。( U: ^$ Z) P, p E1 K! e
产块节点的传播新区块
节点产生区块后,广播的流程可以表示为图4:1 V* S% O/ j/ ?# X
发布事件事件处理函数选择要广播完整的Peer,然后将区块加入到它们的队列事件处理函数把区块Hash添加到所有Peer的另外一个通知队列每个Peer的广播处理函数,会遍历它的待广播区块队列和通知队列,把数据封装成消息,调用P2P接口发送出去' O+ q4 m& B/ Z& d6 C4 A
' \8 P4 b% x) N9 e" u/ q
再看下代码上的细节。" h$ f9 t) J% d; Z. Z
worker.wait()函数发布事件NewMinedBlockEvent。ProtocolManager.minedBroadcastLoop()是事件处理函数。它调用了2次pm.BroadcastBlock()。
$ [) e; Z+ p1 F8 y
// Mined broadcast loop
func (pm *ProtocolManager) minedBroadcastLoop() {
// automatically stops if unsubscribe6 s3 {0 r+ [- S" Q
for obj := range pm.minedBlockSub.Chan() {
switch ev := obj.Data.(type) {6 e4 A2 h/ b" g4 |
case core.NewMinedBlockEvent:
pm.BroadcastBlock(ev.Block, true) // First propagate block to peers
pm.BroadcastBlock(ev.Block, false) // Only then announce to the rest
}
}
}
pm.BroadcastBlock()的入参propagate为真时,向部分Peer广播完整的区块,调用peer.AsyncSendNewBlock(),否则向所有Peer广播区块头,调用peer.AsyncSendNewBlockHash(),这2个函数就是把数据放入队列,此处不再放代码。
// BroadcastBlock will either propagate a block to a subset of it's peers, or* v3 A* _$ P/ H% H( f6 B- T
// will only announce it's availability (depending what's requested).: t1 h6 i5 T% J! ^3 ^/ a# D
func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {) I1 \; Q+ G; ~5 b
hash := block.Hash()
peers := pm.peers.PeersWithoutBlock(hash)" }$ w* w9 R' T/ J" I( b- ~
// If propagation is requested, send to a subset of the peer$ p6 Q0 u' F1 U2 X
// 这种情况,要把区块广播给部分peer' p8 E y% D( J' G
if propagate {
// Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
// 计算新的总难度
var td *big.Int! M0 R! T; |+ }7 r" D7 F
if parent := pm.blockchain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {6 E6 ?$ ?! J" v- \& l: m
td = new(big.Int).Add(block.Difficulty(), pm.blockchain.GetTd(block.ParentHash(), block.NumberU64()-1))( I/ `2 N4 {; n$ G2 P# C
} else {
log.Error("Propagating dangling block", "number", block.Number(), "hash", hash)
return2 N; s: d) X: @% m7 z
}
// Send the block to a subset of our peers8 D L$ |( t8 _3 m
// 广播区块给部分peer$ M- M5 N1 k1 J6 _
transfer := peers[:int(math.Sqrt(float64(len(peers))))]" Y2 D% g3 m8 r8 y4 x! z$ ~1 z
for _, peer := range transfer {
peer.AsyncSendNewBlock(block, td)* x* _4 ?7 `9 C, f
}
log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
return4 H4 t4 c# F' N* O; M
} L* m# n6 i @3 u5 a6 V
// Otherwise if the block is indeed in out own chain, announce it
// 把区块hash值广播给所有peer
if pm.blockchain.HasBlock(hash, block.NumberU64()) {
for _, peer := range peers {4 Y8 o4 D$ I( @+ j, J& C1 M- k
peer.AsyncSendNewBlockHash(block)2 ~% L+ C9 }2 J
}4 K; F) {% p( B3 l/ ? s6 z. G
log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt))) M1 _ K |" t) [% d" S7 H1 r
}0 z \7 t1 e0 n" N2 V! ~. }" Y
}5 E# x- S5 c2 g/ g- t
peer.broadcase()是每个Peer连接的广播函数,它只广播3种消息:交易、完整的区块、区块的Hash,这样表明了节点只会主动广播这3中类型的数据,剩余的数据同步,都是通过请求-响应的方式。
// broadcast is a write loop that multiplexes block propagations, announcements
// and transaction broadcasts into the remote peer. The goal is to have an async# v% n' z2 W6 F3 K: e+ M l' L
// writer that does not lock up node internals.
func (p *peer) broadcast() {
for {
select {
// 广播交易% g# b9 d# d$ u0 ?; L7 U
case txs := 9 V4 H a! m2 c- F7 u
Peer节点处理新区块* F* }, Z) ?7 y5 w& n9 B; [
本节介绍远端节点收到2种区块同步消息的处理,其中NewBlockMsg的处理流程比较清晰,也简洁。NewBlockHashesMsg消息的处理就绕了2绕,从总体流程图1上能看出来,它需要先从给他发送消息Peer那里获取到完整的区块,剩下的流程和NewBlockMsg又一致了。
这部分涉及的模块多,画出来有种眼花缭乱的感觉,但只要抓住上面的主线,代码看起来还是很清晰的。通过图5先看下整体流程。+ O5 F! k+ _) `$ K& W! g' _
消息处理的起点是ProtocolManager.handleMsg,NewBlockMsg的处理流程是蓝色标记的区域,红色区域是单独的协程,是fetcher处理队列中区块的流程,如果从队列中取出的区块是当前链需要的,校验后,调用blockchian.InsertChain()把区块插入到区块链,最后写入数据库,这是黄色部分。最后,绿色部分是NewBlockHashesMsg的处理流程,代码流程上是比较复杂的,为了能通过图描述整体流程,我把它简化掉了。
仔细看看这幅图,掌握整体的流程后,接下来看每个步骤的细节。/ s K1 ^! t) [8 L" x9 T- Y
NewBlockMsg的处理: b( s3 m" O0 _; q, F
本节介绍节点收到完整区块的处理,流程如下:
首先进行RLP编解码,然后标记发送消息的Peer已经知道这个区块,这样本节点最后广播这个区块的Hash时,不会再发送给该Peer。; z* \) r( ` C1 f$ h
将区块存入到fetcher的队列,调用fetcher.Enqueue。
更新Peer的Head位置,然后判断本地链是否落后于Peer的链,如果是,则通过Peer更新本地链。" I0 J0 I. V0 k7 Z8 t
只看handle.Msg()的NewBlockMsg相关的部分。+ a. Q* c3 | t: x0 Z
case msg.Code == NewBlockMsg:: \* I$ ~7 O, E5 I E6 l
// Retrieve and decode the propagated block
// 收到新区块,解码,赋值接收数据+ c3 w( E# }: D1 q$ H/ i0 s9 L
var request newBlockData4 Y% J9 O( s( m4 Z# q
if err := msg.Decode(&request); err != nil {) x- Y" |: f% v* W6 z
return errResp(ErrDecode, "%v: %v", msg, err)9 C4 \2 f+ @" a+ ]" _
}7 e* Q; J7 V% |$ g x
request.Block.ReceivedAt = msg.ReceivedAt
request.Block.ReceivedFrom = p
// Mark the peer as owning the block and schedule it for import" B% \7 R; W& _
// 标记peer知道这个区块
p.MarkBlock(request.Block.Hash())' a. ~5 h( m7 Q0 u; ?. ]7 V. O, p W4 A
// 为啥要如队列?已经得到完整的区块了/ x+ n/ X' x4 p& c
// 答:存入fetcher的优先级队列,fetcher会从队列中选取当前高度需要的块+ Q# I% s. l+ _4 s7 A% R. `; u
pm.fetcher.Enqueue(p.id, request.Block)
// Assuming the block is importable by the peer, but possibly not yet done so,
// calculate the head hash and TD that the peer truly must have.
// 截止到parent区块的头和难度
var (
trueHead = request.Block.ParentHash()
trueTD = new(big.Int).Sub(request.TD, request.Block.Difficulty())$ q6 M w6 V+ z5 s) v: v
)4 \" t6 {' B2 W) B2 \ U5 n
// Update the peers total difficulty if better than the previous( M$ a- Q' G% {& ]1 `) n$ B* [
// 如果收到的块的难度大于peer之前的,以及自己本地的,就去和这个peer同步0 x, ] n d7 C! }8 x8 }% w, J
// 问题:就只用了一下块里的hash指,为啥不直接使用这个块呢,如果这个块不能用,干嘛不少发送些数据,减少网络负载呢。8 y& s9 y& T* k% j8 U) \0 Q1 x! |
// 答案:实际上,这个块加入到了优先级队列中,当fetcher的loop检查到当前下一个区块的高度,正是队列中有的,则不再向peer请求. L! q5 c9 K( G! q! ]3 o
// 该区块,而是直接使用该区块,检查无误后交给block chain执行insertChain8 O3 o) ]7 k2 m
if _, td := p.Head(); trueTD.Cmp(td) > 0 {; I# D% X4 c Y7 O
p.SetHead(trueHead, trueTD)& z z' w- C" W* m/ X9 |7 e' k0 e
// Schedule a sync if above ours. Note, this will not fire a sync for a gap of
// a singe block (as the true TD is below the propagated block), however this
// scenario should easily be covered by the fetcher.) ]# d" g& y' [7 h1 {
currentBlock := pm.blockchain.CurrentBlock()! N7 x' Z N q A8 V
if trueTD.Cmp(pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())) > 0 {8 u/ G) ^# Z. `
go pm.synchronise(p)
}
}
//------------------------ 以上 handleMsg
// Enqueue tries to fill gaps the the fetcher's future import queue.
// 发给inject通道,当前协程在handleMsg,通过通道发送给fetcher的协程处理
func (f *Fetcher) Enqueue(peer string, block *types.Block) error {
op := &inject{
origin: peer,
block: block,
}/ H u: l7 }% B6 I4 V
select {: k9 H3 h# C2 Z6 y9 `9 z
case f.inject blockLimit {0 R W" ~ F1 J/ D& P
log.Debug("Discarded propagated block, exceeded allowance", "peer", peer, "number", block.Number(), "hash", hash, "limit", blockLimit)9 m3 R6 [4 i& N# g, u4 s5 I4 B
propBroadcastDOSMeter.Mark(1)
f.forgetHash(hash)
return
}& w* q* Y4 r; t- y
// Discard any past or too distant blocks
// 高度检查:未来太远的块丢弃
if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist maxQueueDist {( F2 z T) F' q) I
log.Debug("Discarded propagated block, too far away", "peer", peer, "number", block.Number(), "hash", hash, "distance", dist)
propBroadcastDropMeter.Mark(1)7 L& |+ b" G! Y9 |
f.forgetHash(hash)2 F! D! v" h# t) X% E3 X
return
}
// Schedule the block for future importing0 o, C9 A% P) k) p j6 _
// 块先加入优先级队列,加入链之前,还有很多要做
if _, ok := f.queued[hash]; !ok {
op := &inject{- R2 j' }/ C2 B. N
origin: peer,, Q8 i3 K5 N1 \) @, S' Y! p+ W+ O! F
block: block,! \9 L5 R7 u7 v' k; i
}( S5 f+ @# ?$ U5 c5 F% V
f.queues[peer] = count$ ]& k7 }; Y5 i+ S6 i, n& m
f.queued[hash] = op
f.queue.Push(op, -float32(block.NumberU64()))9 N: W4 r& R0 g5 O. j/ J3 T0 R, P
if f.queueChangeHook != nil {. p$ w# p$ w. `/ u% |
f.queueChangeHook(op.block.Hash(), true)
}; s2 }9 I! g7 q" b$ |
log.Debug("Queued propagated block", "peer", peer, "number", block.Number(), "hash", hash, "queued", f.queue.Size())2 u6 i! @8 ?% C. L" J
}7 }9 |/ I% K T; y
}1 Q$ P4 y5 X# K: ^- |: n2 x3 y
fetcher队列处理/ O/ k4 } B# d1 A
本节我们看看,区块加入队列后,fetcher如何处理区块,为何不直接校验区块,插入到本地链?. i4 o7 @3 u) N1 N6 I% p3 J% x
由于以太坊又Uncle的机制,节点可能收到老一点的一些区块。另外,节点可能由于网络原因,落后了几个区块,所以可能收到“未来”的一些区块,这些区块都不能直接插入到本地链。. N4 G( U6 S/ ~* i7 n! M
区块入的队列是一个优先级队列,高度低的区块会被优先取出来。fetcher.loop是单独协程,不断运转,清理fecther中的事务和事件。首先会清理正在fetching的区块,但已经超时。然后处理优先级队列中的区块,判断高度是否是下一个区块,如果是则调用f.insert()函数,校验后调用BlockChain.InsertChain(),成功插入后,广播新区块的Hash。" I1 U2 ~5 o7 G- v" Z1 a! p
// Loop is the main fetcher loop, checking and processing various notification
// events.
func (f *Fetcher) loop() {
// Iterate the block fetching until a quit is requested- w+ T) ?$ V" L Z. k
fetchTimer := time.NewTimer(0)8 r& |8 w4 _: }
completeTimer := time.NewTimer(0)$ M0 z9 q4 I- N0 t: t9 _& ^
for {! Y1 e; [3 }7 Q* u. h' D% m
// Clean up any expired block fetches) M, c' J# X! {
// 清理过期的区块 A) z# a; o! q% d+ M
for hash, announce := range f.fetching {4 j* g' W3 G8 G m
if time.Since(announce.time) > fetchTimeout {, R# q- h- \- h( n3 F
f.forgetHash(hash)
}
}+ X5 Q, v# q' G) @$ i4 h
// Import any queued blocks that could potentially fit' P. I' A8 ]6 n/ H6 w& l
// 导入队列中合适的块+ J- r! x# i: t1 m5 T- H
height := f.chainHeight()$ {5 {0 \" q, N( X" |
for !f.queue.Empty() {
op := f.queue.PopItem().(*inject)
hash := op.block.Hash()1 N) u5 N4 b4 n# u- y6 o5 M6 o
if f.queueChangeHook != nil {
f.queueChangeHook(hash, false) K3 [/ q( ]7 b! r; k
}' R0 y/ @5 d' a% p4 n d7 Z4 S
// If too high up the chain or phase, continue later9 v1 F; F u; a4 L- `+ P7 _ M
// 块不是链需要的下一个块,再入优先级队列,停止循环
number := op.block.NumberU64()
if number > height+1 {+ X ~+ q7 s$ G8 ^5 Z! O
f.queue.Push(op, -float32(number)), _4 _) }8 C: O
if f.queueChangeHook != nil {
f.queueChangeHook(hash, true), m7 f* Z) ?# w/ @" ] M
}
break
}4 W9 B5 ~- p) c1 ^$ Q5 v
// Otherwise if fresh and still unknown, try and import
// 高度正好是我们想要的,并且链上也没有这个块) v- e% v( L2 c4 I8 K3 }
if number+maxUncleDist 5 R* }. o' i+ H: X
func (f *Fetcher) insert(peer string, block *types.Block) {6 {1 S4 N3 H& x. g/ j
hash := block.Hash()
// Run the import on a new thread
log.Debug("Importing propagated block", "peer", peer, "number", block.Number(), "hash", hash)2 Z$ r. U( u! O7 b
go func() {. W5 B3 D/ m! M# |7 g" O
defer func() { f.done + _+ B! I. Z3 r: I0 g
NewBlockHashesMsg的处理
本节介绍NewBlockHashesMsg的处理,其实,消息处理是简单的,而复杂一点的是从Peer哪获取完整的区块,下节再看。8 n/ E1 x% Z/ y# \# J; Q
流程如下:
对消息进行RLP解码,然后标记Peer已经知道此区块。寻找出本地区块链不存在的区块Hash值,把这些未知的Hash通知给fetcher。fetcher.Notify记录好通知信息,塞入notify通道,以便交给fetcher的协程。fetcher.loop()会对notify中的消息进行处理,确认区块并非DOS攻击,然后检查区块的高度,判断该区块是否已经在fetching或者comleting(代表已经下载区块头,在下载body),如果都没有,则加入到announced中,触发0s定时器,进行处理。" J: A" u$ ` Q2 d& b* @
关于announced下节再介绍。
// handleMsg()部分
case msg.Code == NewBlockHashesMsg:) v2 Y; e# }" A4 v W
var announces newBlockHashesData U9 U1 @6 F6 D. u2 J
if err := msg.Decode(&announces); err != nil {8 W2 e7 w/ H- ]
return errResp(ErrDecode, "%v: %v", msg, err)
}: t: [0 H& {% j, Z: c4 z
// Mark the hashes as present at the remote node3 ~( W( R# Z3 M* u! n7 y1 A2 n: X
for _, block := range announces {
p.MarkBlock(block.Hash)/ b% ` h3 d+ M
}0 Q* k. N' G3 ?
// Schedule all the unknown hashes for retrieval
// 把本地链没有的块hash找出来,交给fetcher去下载
unknown := make(newBlockHashesData, 0, len(announces))
for _, block := range announces {" ~1 g1 I2 g/ I; O# C
if !pm.blockchain.HasBlock(block.Hash, block.Number) {
unknown = append(unknown, block)& m3 [; k; V- C8 s1 X. k
}' g5 `' p& F# I/ S
}
for _, block := range unknown {
pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestOneHeader, p.RequestBodies)
}$ l- s3 {% D1 F, u% g
// Notify announces the fetcher of the potential availability of a new block in$ Q$ \+ P! K: x1 W& g5 J) ?( G% W% k
// the network.
// 通知fetcher(自己)有新块产生,没有块实体,有hash、高度等信息# i* P* ]2 M$ }, h. o! q. @ M- O" v
func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time.Time,$ A' ]% A& o) G
headerFetcher headerRequesterFn, bodyFetcher bodyRequesterFn) error {
block := &announce{
hash: hash,; L J; a2 O' q& D7 W' O
number: number,
time: time,& w( s( W( t( W3 K2 ^/ V! }4 i4 [) n
origin: peer,
fetchHeader: headerFetcher,
fetchBodies: bodyFetcher,$ |5 C$ ~5 k8 b( N
}. W9 G, e& l" b4 [- x. s
select {
case f.notify hashLimit {7 d+ Q4 o4 R$ V$ L5 \) X- i
log.Debug("Peer exceeded outstanding announces", "peer", notification.origin, "limit", hashLimit)
propAnnounceDOSMeter.Mark(1)
break
}
// If we have a valid block number, check that it's potentially useful
// 高度检查5 b" w& G4 s" K" g
if notification.number > 0 { d" Z+ w4 Z) s
if dist := int64(notification.number) - int64(f.chainHeight()); dist maxQueueDist {
log.Debug("Peer discarded announcement", "peer", notification.origin, "number", notification.number, "hash", notification.hash, "distance", dist)
propAnnounceDropMeter.Mark(1)
break; T& X' }1 f2 t7 Q! g ?' N6 `' H3 _
}
}9 Y4 X) k: t% d- n' T
// All is well, schedule the announce if block's not yet downloading
// 检查是否已经在下载,已下载则忽略: H( Q0 T8 L6 e3 j( {: z
if _, ok := f.fetching[notification.hash]; ok {
break9 i% ^7 _/ ~0 q, Y! Z$ e* {- l: ~
}
if _, ok := f.completing[notification.hash]; ok {1 e/ o- g: p: ^. G( P
break9 o8 n, X; ^9 ~4 {. Z t" t5 N
}3 m7 V. O8 n9 F. V# D5 d8 w& C
// 更新peer已经通知给我们的区块数量& j0 S, j# R7 G. ]* C
f.announces[notification.origin] = count
// 把通知信息加入到announced,供调度
f.announced[notification.hash] = append(f.announced[notification.hash], notification)
if f.announceChangeHook != nil && len(f.announced[notification.hash]) == 1 {+ t3 n9 c' S6 z3 E" l( n
f.announceChangeHook(notification.hash, true)
}
if len(f.announced) == 1 { i8 x" B9 c; O' m m
// 有通知放入到announced,则重设0s定时器,loop的另外一个分支会处理这些通知
f.rescheduleFetch(fetchTimer). M: C% k1 Q, r$ v( ^
}6 j# i5 s. L- D$ m4 K1 W) @
fetcher获取完整区块0 i+ [# o( y( \
本节介绍fetcher获取完整区块的过程,这也是fetcher最重要的功能,会涉及到fetcher至少80%的代码。单独拉放一大节吧。
Fetcher的大头2 i2 J8 W% U6 s. Q/ P& ]; s) T
Fetcher最主要的功能就是获取完整的区块,然后在合适的实际交给InsertChain去验证和插入到本地区块链。我们还是从宏观入手,看Fetcher是如何工作的,一定要先掌握好宏观,因为代码层面上没有这么清晰。
宏观0 d) o6 A/ {3 {& k- \( M
首先,看两个节点是如何交互,获取完整区块,使用时序图的方式看一下,见图6,流程很清晰不再文字介绍。( |. V4 t& l9 f# I; y- s
) z3 }/ w4 A1 d
再看下获取区块过程中,fetcher内部的状态转移,它使用状态来记录,要获取的区块在什么阶段,见图7。我稍微解释一下:
收到NewBlockHashesMsg后,相关信息会记录到announced,进入announced状态,代表了本节点接收了消息。announced由fetcher协程处理,经过校验后,会向给他发送消息的Peer发送请求,请求该区块的区块头,然后进入fetching状态。获取区块头后,如果区块头表示没有交易和uncle,则转移到completing状态,并且使用区块头合成完整的区块,加入到queued优先级队列。获取区块头后,如果区块头表示该区块有交易和uncle,则转移到fetched状态,然后发送请求,请求交易和uncle,然后转移到completing状态。收到交易和uncle后,使用头、交易、uncle这3个信息,生成完整的区块,加入到队列queued。" J/ Q) m$ a5 v0 k, C B; g
微观" K4 z; Y5 M$ ?/ K
接下来就是从代码角度看如何获取完整区块的流程了,有点多,看不懂的时候,再回顾下上面宏观的介绍图。
首先看Fetcher的定义,它存放了通信数据和状态管理,捡加注释的看,上文提到的状态,里面都有。; ^; F0 z0 B" F6 h. s; T
// Fetcher is responsible for accumulating block announcements from various peers
// and scheduling them for retrieval.
// 积累块通知,然后调度获取这些块
type Fetcher struct {% Z& e# K5 c; d4 y! Z& e9 |# |
// Various event channels
// 收到区块hash值的通道6 A( k, A9 t0 |
notify chan *announce" H x r: C1 D' t4 c
// 收到完整区块的通道5 ?5 U% m5 ^& ~, D* z) K3 x
inject chan *inject
blockFilter chan chan []*types.Block
// 过滤header的通道的通道
headerFilter chan chan *headerFilterTask2 n5 K+ \/ \. B9 u* j5 C' J
// 过滤body的通道的通道, R; g2 k( K4 c% Q% z6 ~" Q
bodyFilter chan chan *bodyFilterTask
done chan common.Hash* a+ [+ q+ a7 `( H# _1 W
quit chan struct{}
// Announce states
// Peer已经给了本节点多少区块头通知
announces map[string]int // Per peer announce counts to prevent memory exhaustion# V/ \; p( b# [) r0 o% D
// 已经announced的区块列表1 d( S0 s j0 Q. X* A8 ?
announced map[common.Hash][]*announce // Announced blocks, scheduled for fetching
// 正在fetching区块头的请求; n5 v4 R8 I& A* N$ W
fetching map[common.Hash]*announce // Announced blocks, currently fetching
// 已经fetch到区块头,还差body的请求,用来获取body; j$ m. v& S- P- h$ t
fetched map[common.Hash][]*announce // Blocks with headers fetched, scheduled for body retrieval
// 已经得到区块头的: K0 H9 Q# n# H: ^! E) i5 n+ B! F8 X
completing map[common.Hash]*announce // Blocks with headers, currently body-completing. u& c4 k x* T% e, t/ `" k4 e
// Block cache# ]9 c6 Z& y' C. C6 R
// queue,优先级队列,高度做优先级
// queues,统计peer通告了多少块3 @2 k% S- z- u( M
// queued,代表这个块如队列了,
queue *prque.Prque // Queue containing the import operations (block number sorted)
queues map[string]int // Per peer block counts to prevent memory exhaustion! B8 N1 W" E! V$ l$ R
queued map[common.Hash]*inject // Set of already queued blocks (to dedupe imports)3 b- ~ d( z' F ^2 l$ }$ V
// Callbacks; v$ q' t2 I( J( j& |6 y+ _3 P
getBlock blockRetrievalFn // Retrieves a block from the local chain
verifyHeader headerVerifierFn // Checks if a block's headers have a valid proof of work,验证区块头,包含了PoW验证) ]% @$ [8 l" e+ P" _! d
broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers,广播给peer
chainHeight chainHeightFn // Retrieves the current chain's height( @- S) B7 e/ w: B V3 W% {" `$ Z
insertChain chainInsertFn // Injects a batch of blocks into the chain,插入区块到链的函数
dropPeer peerDropFn // Drops a peer for misbehaving, X# ^+ U8 J0 y) }, ^8 i
// Testing hooks
announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list
queueChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a block from the import queue
fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch4 J( f. M- m$ T$ v
completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62)
importedHook func(*types.Block) // Method to call upon successful block import (both eth/61 and eth/62); F% O) {) T& L4 O+ ?
}% U3 Z* `3 m6 Z9 k+ U- g6 Q
NewBlockHashesMsg消息的处理前面的小节已经讲过了,不记得可向前翻看。这里从announced的状态处理说起。loop() 中,fetchTimer超时后,代表了收到了消息通知,需要处理,会从announced中选择出需要处理的通知,然后创建请求,请求区块头,由于可能有很多节点都通知了它某个区块的Hash,所以随机的从这些发送消息的Peer中选择一个Peer,发送请求的时候,为每个Peer都创建了单独的协程。
case arriveTimeout-gatherSlack {
// Pick a random peer to retrieve from, reset all others
// 可能有很多peer都发送了这个区块的hash值,随机选择一个peer' C5 O$ K3 x- O( S# e* v
announce := announces[rand.Intn(len(announces))]: }# K. v7 C; u
f.forgetHash(hash)
// If the block still didn't arrive, queue for fetching+ b3 d1 W0 R# T
// 本地还没有这个区块,创建获取区块的请求+ C3 c' m0 ]; o6 _( R6 l
if f.getBlock(hash) == nil {
request[announce.origin] = append(request[announce.origin], hash)
f.fetching[hash] = announce
}3 l& Z, D2 W6 U
}4 q/ d& I# k1 K# ?
}
// Send out all block header requests- m9 p# T; d; S* H! ]0 V
// 把所有的request发送出去
// 为每一个peer都创建一个协程,然后请求所有需要从该peer获取的请求
for peer, hashes := range request {
log.Trace("Fetching scheduled headers", "peer", peer, "list", hashes)
// Create a closure of the fetch and schedule in on a new thread
fetchHeader, hashes := f.fetching[hashes[0]].fetchHeader, hashes
go func() {; f0 c1 l& f' u) v
if f.fetchingHook != nil {
f.fetchingHook(hashes)
}
for _, hash := range hashes {: e& o& a t- O! ?
headerFetchMeter.Mark(1)
fetchHeader(hash) // Suboptimal, but protocol doesn't allow batch header retrievals
}1 q; D/ _ x/ K- p* Z; M
}()+ k* a$ c2 n0 A1 Z! X+ O
}
// Schedule the next fetch if blocks are still pending4 [' R0 a" K* D8 k5 j0 R
f.rescheduleFetch(fetchTimer)
从Notify的调用中,可以看出,fetcherHeader()的实际函数是RequestOneHeader(),该函数使用的消息是GetBlockHeadersMsg,可以用来请求多个区块头,不过fetcher只请求一个。
pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestOneHeader, p.RequestBodies)/ z! _, ]# l8 x7 h0 W3 Y/ d
// RequestOneHeader is a wrapper around the header query functions to fetch a
// single header. It is used solely by the fetcher.
func (p *peer) RequestOneHeader(hash common.Hash) error {
p.Log().Debug("Fetching single header", "hash", hash)
return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false}): r* l8 H: v& I1 n* w
}: I4 M' s" j! |" s
GetBlockHeadersMsg的处理如下:因为它是获取多个区块头的,所以处理起来比较“麻烦”,还好,fetcher只获取一个区块头,其处理在20行~33行,获取下一个区块头的处理逻辑,这里就不看了,最后调用SendBlockHeaders()将区块头发送给请求的节点,消息是BlockHeadersMsg。
···! D8 d% C+ u: {5 G M% R
// handleMsg()- [9 ?4 Q% \: i+ W. w+ H4 [
// Block header query, collect the requested headers and reply
case msg.Code == GetBlockHeadersMsg:( c% J& o H3 ^5 @1 q) R
// Decode the complex header query
var query getBlockHeadersData
if err := msg.Decode(&query); err != nil {
return errResp(ErrDecode, “%v: %v”, msg, err)8 c6 W3 L! L' N, ^0 U$ ]
}
hashMode := query.Origin.Hash != (common.Hash{})0 r7 |1 v! F( B" c
// Gather headers until the fetch or network limits is reached7 r! u9 U! ?& k& K7 d# m' p
// 收集区块头,直到达到限制4 q. G) Z2 s. ^6 A# y; h
var (
bytes common.StorageSize6 R& t+ @( `- Z5 U* I
headers []*types.Header
unknown bool4 j1 T; R1 b& _; V7 t- v& Q4 `
)) \, s* v. G: D8 I' Y1 E( G
// 自己已知区块 && 少于查询的数量 && 大小小于2MB && 小于能下载的最大数量
for !unknown && len(headers)
`BlockHeadersMsg`的处理很有意思,因为`GetBlockHeadersMsg`并不是fetcher独占的消息,downloader也可以调用,所以,响应消息的处理需要分辨出是fetcher请求的,还是downloader请求的。它的处理逻辑是:fetcher先过滤收到的区块头,如果fetcher不要的,那就是downloader的,在调用`fetcher.FilterHeaders`的时候,fetcher就将自己要的区块头拿走了。
// handleMsg()! t/ y+ Y! C3 t- s" }' ?& Y
case msg.Code == BlockHeadersMsg:& X; a% m9 _1 `
// A batch of headers arrived to one of our previous requests7 y, f. C/ i% V1 h/ B
var headers []*types.Header& U5 H* ^/ t: D, S: B& R8 b4 V; q
if err := msg.Decode(&headers); err != nil {
return errResp(ErrDecode, “msg %v: %v”, msg, err)/ E0 t S8 q( K
}! A1 S. v! a( e* g i* p3 l. j
// If no headers were received, but we’re expending a DAO fork check, maybe it’s that
// 检查是不是当前DAO的硬分叉
if len(headers) == 0 && p.forkDrop != nil {
// Possibly an empty reply to the fork header checks, sanity check TDs
verifyDAO := true* a5 _+ D4 }+ J- @6 m; A- b6 v
// If we already have a DAO header, we can check the peer's TD against it. If
// the peer's ahead of this, it too must have a reply to the DAO check2 f7 i( s8 v4 V6 u+ f
if daoHeader := pm.blockchain.GetHeaderByNumber(pm.chainconfig.DAOForkBlock.Uint64()); daoHeader != nil {
if _, td := p.Head(); td.Cmp(pm.blockchain.GetTd(daoHeader.Hash(), daoHeader.Number.Uint64())) >= 0 {1 k; @! y2 |- y, D- a
verifyDAO = false
}7 y/ P& V# i) ^* {2 M& q9 F6 ~" [
}
// If we're seemingly on the same chain, disable the drop timer- c' Y& f5 ^) m1 b
if verifyDAO {9 S5 m Y& s0 [2 e2 K9 }+ M
p.Log().Debug("Seems to be on the same side of the DAO fork")5 Y* W& g6 J1 V) y# j
p.forkDrop.Stop()/ k0 m& A2 N0 A( J L4 ?0 N# ^
p.forkDrop = nil: s# d7 i5 P( r9 t1 V0 H% y
return nil" i4 |. I/ M7 C& d1 b9 [# G" K* Z2 h
}6 Q$ |+ }/ P+ Z1 s3 n
}% \+ n/ p \* R- O+ W
// Filter out any explicitly requested headers, deliver the rest to the downloader) N* B* J6 Z$ H0 `
// 过滤是不是fetcher请求的区块头,去掉fetcher请求的区块头再交给downloader
filter := len(headers) == 1
if filter {
// If it's a potential DAO fork check, validate against the rules F1 z* |7 Z. m# g& T( D, p) o
// 检查是否硬分叉
if p.forkDrop != nil && pm.chainconfig.DAOForkBlock.Cmp(headers[0].Number) == 0 {
// Disable the fork drop timer
p.forkDrop.Stop()
p.forkDrop = nil+ V. w5 f3 s9 _
// Validate the header and either drop the peer or continue
if err := misc.VerifyDAOHeaderExtraData(pm.chainconfig, headers[0]); err != nil {* [$ I' D; v5 l2 Y- L; C
p.Log().Debug("Verified to be on the other side of the DAO fork, dropping")/ {/ @# B) L4 {+ W* H+ M' V
return err
}
p.Log().Debug("Verified to be on the same side of the DAO fork")
return nil- o- |3 Y& n$ F" y5 D
}- C: D% a6 v; v `' I
// Irrelevant of the fork checks, send the header to the fetcher just in case
// 使用fetcher过滤区块头4 _4 n9 @3 V- O/ x _5 U7 A
headers = pm.fetcher.FilterHeaders(p.id, headers, time.Now())$ R; w# v: _7 r8 C! {( \
}
// 剩下的区块头交给downloader
if len(headers) > 0 || !filter {
err := pm.downloader.DeliverHeaders(p.id, headers)
if err != nil {
log.Debug("Failed to deliver headers", "err", err)
}/ I1 x; u: p# L, Z% B7 A
}
`FilterHeaders()`是一个很有大智慧的函数,看起来耐人寻味,但实在妙。它要把所有的区块头,都传递给fetcher协程,还要获取fetcher协程处理后的结果。`fetcher.headerFilter`是存放通道的通道,而`filter`是存放包含区块头过滤任务的通道。它先把`filter`传递给了`headerFilter`,这样`fetcher`协程就在另外一段等待了,而后将`headerFilterTask`传入`filter`,`fetcher`就能读到数据了,处理后,再将数据写回`filter`而刚好被`FilterHeaders`函数处理了,该函数实际运行在`handleMsg()`的协程中。. A- K! H% f6 ^! m& C
每个Peer都会分配一个ProtocolManager然后处理该Peer的消息,但`fetcher`只有一个事件处理协程,如果不创建一个`filter`,fetcher哪知道是谁发给它的区块头呢?过滤之后,该如何发回去呢?
// FilterHeaders extracts all the headers that were explicitly requested by the fetcher,; C7 I$ G5 k0 F' j4 r$ ]
// returning those that should be handled differently.
// 寻找出fetcher请求的区块头! I6 s' L- u' P( U5 p9 D
func (f *Fetcher) FilterHeaders(peer string, headers []*types.Header, time time.Time) []*types.Header {% X$ B) n/ f( t
log.Trace(“Filtering headers”, “peer”, peer, “headers”, len(headers))
// Send the filter channel to the fetcher
// 任务通道
filter := make(chan *headerFilterTask)
select {
// 任务通道发送到这个通道
case f.headerFilter , X! l" c, r( d$ t
}
接下来要看f.headerFilter的处理,这段代码有90行,它做了一下几件事:# Q7 W7 b6 g7 h5 t, q& ]; i
1. 从`f.headerFilter`取出`filter`,然后取出过滤任务`task`。7 z' }% \) T. n1 r4 v
2. 它把区块头分成3类:`unknown`这不是分是要返回给调用者的,即`handleMsg()`, `incomplete`存放还需要获取body的区块头,`complete`存放只包含区块头的区块。遍历所有的区块头,填到到对应的分类中,具体的判断可看18行的注释,记住宏观中将的状态转移图。
3. 把`unknonw`中的区块返回给`handleMsg()`。
4. 把` incomplete`的区块头获取状态移动到`fetched`状态,然后触发定时器,以便去处理complete的区块。3 h8 S) c9 G6 R% m2 G
5. 把`compelete`的区块加入到`queued`。+ p4 ]. \5 Z- s$ W2 I- j
// fetcher.loop()
case filter :=
// Split the batch of headers into unknown ones (to return to the caller),' c& w. |0 k* ?$ d2 N
// known incomplete ones (requiring body retrievals) and completed blocks.; |+ H4 g6 x# e
// unknown的不是fetcher请求的,complete放没有交易和uncle的区块,有头就够了,incomplete放
// 还需要获取uncle和交易的区块
unknown, incomplete, complete := []*types.Header{}, []*announce{}, []*types.Block{}( N3 @! A0 o; \# X! S: R& |
// 遍历所有收到的header' x) j9 u. \0 o
for _, header := range task.headers {0 W. z( m" B4 [2 R6 t6 y7 ^
hash := header.Hash()
// Filter fetcher-requested headers from other synchronisation algorithms6 r& V B) ]& M" o5 a
// 是正在获取的hash,并且对应请求的peer,并且未fetched,未completing,未queued+ z. ?6 j" t" U( c& v* J: `
if announce := f.fetching[hash]; announce != nil && announce.origin == task.peer && f.fetched[hash] == nil && f.completing[hash] == nil && f.queued[hash] == nil {
// If the delivered header does not match the promised number, drop the announcer
// 高度校验,竟然不匹配,扰乱秩序,peer肯定是坏蛋。# j0 ~+ Z1 S1 S0 F$ M' \
if header.Number.Uint64() != announce.number {- R& ]! f7 S7 A* i$ \' q6 k$ ~* J
log.Trace("Invalid block number fetched", "peer", announce.origin, "hash", header.Hash(), "announced", announce.number, "provided", header.Number)& u( [/ d/ T, B3 _ C5 p- a' y; E
f.dropPeer(announce.origin)$ g- W3 F& d/ g# \' l0 c
f.forgetHash(hash)
continue! ^: D0 ]% z3 {1 a: X- m
}
// Only keep if not imported by other means' M. Y/ B; e$ w" y3 w
// 本地链没有当前区块( o$ o! o* p5 j, v, l
if f.getBlock(hash) == nil {
announce.header = header
announce.time = task.time7 Y% p7 _) y& y7 G3 B' W
// If the block is empty (header only), short circuit into the final import queue
// 如果区块没有交易和uncle,加入到complete
if header.TxHash == types.DeriveSha(types.Transactions{}) && header.UncleHash == types.CalcUncleHash([]*types.Header{}) {3 x' k h% _4 U1 I0 g" v2 z
log.Trace("Block empty, skipping body retrieval", "peer", announce.origin, "number", header.Number, "hash", header.Hash())
block := types.NewBlockWithHeader(header)
block.ReceivedAt = task.time6 C+ H/ w$ ~/ Z3 [
complete = append(complete, block)
f.completing[hash] = announce/ Z9 h8 G% Z. I: L$ K$ z* a
continue/ M+ o$ z* {$ X
}
// Otherwise add to the list of blocks needing completion
// 否则就是不完整的区块
incomplete = append(incomplete, announce)0 X, D* V! ~! V) {, E
} else {- ^! c& B" @1 L; k9 v9 Q2 q) O
log.Trace("Block already imported, discarding header", "peer", announce.origin, "number", header.Number, "hash", header.Hash())
f.forgetHash(hash)
}* P4 o( X9 L" F, P! H0 {+ w$ a
} else {; k& }! J4 N+ {6 d4 k0 {8 n3 n
// Fetcher doesn't know about it, add to the return list
// 没请求过的header/ O, o% X8 v: {- e! u; m
unknown = append(unknown, header)$ N' S: G# Y& f
}
}
// 把未知的区块头,再传递会filter, w6 N$ W! i& d9 Y! K
headerFilterOutMeter.Mark(int64(len(unknown)))
select {
case filter
跟随状态图的转义,剩下的工作是`fetched`转移到`completing` ,上面的流程已经触发了`completeTimer`定时器,超时后就会处理,流程与请求Header类似,不再赘述,此时发送的请求消息是`GetBlockBodiesMsg`,实际调的函数是`RequestBodies`。- a6 u0 [9 O* `3 N( Y0 Y
// fetcher.loop()+ e D; H! @- e7 @/ t7 E
case 3 ~/ M& F9 n2 T3 G' e# T
// 遍历所有待获取body的announce+ k, M3 f: O) s `
for hash, announces := range f.fetched {
// Pick a random peer to retrieve from, reset all others
// 随机选一个Peer发送请求,因为可能已经有很多Peer通知它这个区块了
announce := announces[rand.Intn(len(announces))]
f.forgetHash(hash), V$ F( S" W% Q5 n
// If the block still didn't arrive, queue for completion
// 如果本地没有这个区块,则放入到completing,创建请求
if f.getBlock(hash) == nil {
request[announce.origin] = append(request[announce.origin], hash)
f.completing[hash] = announce* ^7 T0 x+ N( |% a0 O. e# V
}, |& L# t) `) N9 q" V
}0 C6 Q+ Z" g4 ~* {9 I: o
// Send out all block body requests
// 发送所有的请求,获取body,依然是每个peer一个单独协程8 \% @0 g# d8 I9 F
for peer, hashes := range request {& O4 }. \) E. c( c3 E
log.Trace("Fetching scheduled bodies", "peer", peer, "list", hashes)
// Create a closure of the fetch and schedule in on a new thread: J5 n, U' i7 X9 F1 w8 f* N4 b, K
if f.completingHook != nil {
f.completingHook(hashes)
}
bodyFetchMeter.Mark(int64(len(hashes)))) R6 T8 g% _0 K7 x
go f.completing[hashes[0]].fetchBodies(hashes)
}
// Schedule the next fetch if blocks are still pending
f.rescheduleComplete(completeTimer)
`handleMsg()`处理该消息也是干净利落,直接获取RLP格式的body,然后发送响应消息。6 ] \' f$ g" k# s
// handleMsg()% {7 n7 ^% }4 X6 K) E) z& A
case msg.Code == GetBlockBodiesMsg:' l! d- [0 _6 Z. M1 @8 @1 `
// Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil {
return err
}
// Gather blocks until the fetch or network limits is reached
var (9 z' ?1 B2 F+ E) u* z! E
hash common.Hash
bytes int
bodies []rlp.RawValue
)
// 遍历所有请求
for bytes
响应消息`BlockBodiesMsg`的处理与处理获取header的处理原理相同,先交给fetcher过滤,然后剩下的才是downloader的。需要注意一点,响应消息里只包含交易列表和叔块列表。
// handleMsg()0 X( v. w1 i' i- L) W3 e) M. m
case msg.Code == BlockBodiesMsg:
// A batch of block bodies arrived to one of our previous requests
var request blockBodiesData
if err := msg.Decode(&request); err != nil {
return errResp(ErrDecode, “msg %v: %v”, msg, err)
}
// Deliver them all to the downloader for queuing% W( R4 G" u$ e
// 传递给downloader去处理. y6 \! A P1 D: ], Q( ^5 D5 G
transactions := make([][]*types.Transaction, len(request))
uncles := make([][]*types.Header, len(request))
for i, body := range request {5 Z. }1 Q3 B Z* p
transactions = body.Transactions
uncles = body.Uncles
}
// Filter out any explicitly requested bodies, deliver the rest to the downloader
// 先让fetcher过滤去fetcher请求的body,剩下的给downloader7 T& e" p% Y4 Q
filter := len(transactions) > 0 || len(uncles) > 03 r6 u% _" [1 K4 N+ `$ W
if filter {! r' a4 }0 r& R: c
transactions, uncles = pm.fetcher.FilterBodies(p.id, transactions, uncles, time.Now())2 y5 i! A- k& d. b# c" Z: M5 \
}
// 剩下的body交给downloader
if len(transactions) > 0 || len(uncles) > 0 || !filter {
err := pm.downloader.DeliverBodies(p.id, transactions, uncles)
if err != nil {8 a6 M- e5 U& [* U, b3 ^
log.Debug("Failed to deliver bodies", "err", err)1 n( ]# j( `, f) l$ s
}4 M- g X J: T2 ]1 o5 K
}+ P$ M% C% R6 k7 Q6 d _# f W
过滤函数的原理也与Header相同。5 J' D! ?- |7 |* ~, F2 m; a+ J
// FilterBodies extracts all the block bodies that were explicitly requested by6 x* q8 R: ^, m7 b, ~- A6 Y
// the fetcher, returning those that should be handled differently.9 E0 x7 c6 B# T$ ~& a
// 过去出fetcher请求的body,返回它没有处理的,过程类型header的处理7 r& P3 _2 ]5 F% \
func (f *Fetcher) FilterBodies(peer string, transactions [][]*types.Transaction, uncles [][]*types.Header, time time.Time) ([][]*types.Transaction, [][]*types.Header) {
log.Trace(“Filtering bodies”, “peer”, peer, “txs”, len(transactions), “uncles”, len(uncles))
// Send the filter channel to the fetcher
filter := make(chan *bodyFilterTask)
select {0 `! O6 F1 j1 R/ P' c+ I
case f.bodyFilter . c4 L v7 z8 U1 j( f8 a, S8 ^
}
实际过滤body的处理瞧一下,这和Header的处理是不同的。直接看不点:
1. 它要的区块,单独取出来存到`blocks`中,它不要的继续留在`task`中。
2. 判断是不是fetcher请求的方法:如果交易列表和叔块列表计算出的hash值与区块头中的一样,并且消息来自请求的Peer,则就是fetcher请求的。5 i7 I0 u& Y/ x7 i( } b
3. 将`blocks`中的区块加入到`queued`,终结。
case filter := ' V* z* _/ F. T$ z0 p9 U4 Q) o2 ^
blocks := []*types.Block{}
// 获取的每个body的txs列表和uncle列表# F! @5 z+ _# W7 Z5 C2 ~! T
// 遍历每个区块的txs列表和uncle列表,计算hash后判断是否是当前fetcher请求的body5 G5 P8 e+ H) z2 e1 {
for i := 0; i
}- c* b0 r0 H2 v+ b& r
: w& _" \/ K) Y/ I
至此,fetcher获取完整区块的流程讲完了,fetcher模块中80%的代码也都贴出来了,还有2个值得看看的函数:
1. `forgetHash(hash common.Hash)`:用于清空指定hash指的记/状态录信息。3 R& ?7 t" H; k( P) Z: [5 F! I
2. `forgetBlock(hash common.Hash)`:用于从队列中移除一个区块。. L6 S8 m5 {' Q& _+ [1 O1 ]1 o
最后了,再回到开始看看fetcher模块和新区块的传播流程,有没有豁然开朗。
成为第一个吐槽的人