摘要:作者比原項目倉庫地址地址在前一篇中,我們已經知道如何連上一個比原節點的端口,并與對方完成身份驗證。代碼如下可以看到,首先是從眾多的中,找到最合適的那個。到這里,我們其實已經知道比原是如何向其它節點請求區塊數據,以及何時把信息發送出去。
作者:freewind
比原項目倉庫:
Github地址:https://github.com/Bytom/bytom
Gitee地址:https://gitee.com/BytomBlockc...
在前一篇中,我們已經知道如何連上一個比原節點的p2p端口,并與對方完成身份驗證。此時,雙方結點已經建立起來了信任,并且連接也不會斷開,下一步,兩者就可以繼續交換數據了。
那么,我首先想到的就是,如何才能讓對方把它已有的區塊數據全都發給我呢?
這其實可以分為三個問題:
我需要發給它什么樣的數據?
它在內部由是如何應答的呢?
我拿到數據之后,應該怎么處理?
由于這一塊的邏輯還是比較復雜的,所以在本篇我們先回答第一個問題:
我們要發送什么樣的數據請求,才能讓比原節點把它持有的區塊數據發給我? 找到發送請求的代碼首先我們先要在代碼中定位到,比原到底是在什么時候來向對方節點發送請求的。
在前一篇講的是如何建立連接并驗證身份,那么發出數據請求的操作,一定在上次的代碼之后。按照這個思路,我們在SyncManager類中Switch啟動之后,找到了一個叫BlockKeeper的類,相關的操作是在它里面完成的。
下面是老規矩,還是從啟動開始,但是會更簡化一些:
cmd/bytomd/main.go#L54
func main() { cmd := cli.PrepareBaseCmd(commands.RootCmd, "TM", os.ExpandEnv(config.DefaultDataDir())) cmd.Execute() }
cmd/bytomd/commands/run_node.go#L41
func runNode(cmd *cobra.Command, args []string) error { n := node.NewNode(config) if _, err := n.Start(); err != nil { // ... }
node/node.go#L169
func (n *Node) OnStart() error { // ... n.syncManager.Start() // ... }
netsync/handle.go#L141
func (sm *SyncManager) Start() { go sm.netStart() // ... go sm.syncer() }
注意sm.netStart(),我們在一篇中建立連接并驗證身份的操作,就是在它里面完成的。而這次的這個問題,是在下面的sm.syncer()中完成的。
另外注意,由于這兩個函數調用都使用了goroutine,所以它們是同時進行的。
sm.syncer()的代碼如下:
netsync/sync.go#L46
func (sm *SyncManager) syncer() { sm.fetcher.Start() defer sm.fetcher.Stop() // ... for { select { case <-sm.newPeerCh: log.Info("New peer connected.") // Make sure we have peers to select from, then sync if sm.sw.Peers().Size() < minDesiredPeerCount { break } go sm.synchronise() // .. } }
這里混入了一個叫fetcher的奇怪的東西,名字看起來好像是專門去抓取數據的,我們要找的是它嗎?
可惜不是,fetcher的作用是從多個peer那里拿到了區塊數據之后,對數據進行整理,把有用的放到本地鏈上。我們在以后會研究它,所以這里不展開討論。
接著是一個for循環,當發現通道newPeerCh有了新數據(也就是有了新的節點連接上了),會判斷一下當前自己連著的節點是否夠多(大于等于minDesiredPeerCount,值為5),夠多的話,就會進入sm.synchronise(),進行數據同步。
這里為什么要多等幾個節點,而不是一連上就馬上同步呢?我想這是希望有更多選擇的機會,找到一個數據夠多的節點。
sm.synchronise()還是屬于SyncManager的方法。在真正調用到BlockKeeper的方法之前,它還做了一些比如清理已經斷開的peer,找到最適合同步數據的peer等。其中“清理peer”的工作涉及到不同的對象持有的peer集合間的同步,略有些麻煩,但對當前問題幫助不大,所以我打算把它們放在以后的某個問題中回答(比如“當一個節點斷開了,比原會有什么樣的處理”),這里就先省略。
sm.synchronise()代碼如下:
netsync/sync.go#L77
func (sm *SyncManager) synchronise() { log.Info("bk peer num:", sm.blockKeeper.peers.Len(), " sw peer num:", sm.sw.Peers().Size(), " ", sm.sw.Peers().List()) // ... peer, bestHeight := sm.peers.BestPeer() // ... if bestHeight > sm.chain.BestBlockHeight() { // ... sm.blockKeeper.BlockRequestWorker(peer.Key, bestHeight) } }
可以看到,首先是從眾多的peers中,找到最合適的那個。什么叫Best呢?看一下BestPeer()的定義:
netsync/peer.go#L266
func (ps *peerSet) BestPeer() (*p2p.Peer, uint64) { // ... for _, p := range ps.peers { if bestPeer == nil || p.height > bestHeight { bestPeer, bestHeight = p.swPeer, p.height } } return bestPeer, bestHeight }
其實就是持有區塊鏈數據最長的那個。
找到了BestPeer之后,就調用sm.blockKeeper.BlockRequestWorker(peer.Key, bestHeight)方法,從這里,正式進入BlockKeeper -- 也就是本文的主角 -- 的世界。
BlockKeeperblockKeeper.BlockRequestWorker的邏輯比較復雜,它包含了:
根據自己持有的區塊數據來計算需要同步的數據
向前面找到的最佳節點發送數據請求
拿到對方發過來的區塊數據
對數據進行處理
廣播新狀態
處理各種出錯情況,等等
由于本文中只關注“發送請求”,所以一些與之關系不大的邏輯我會忽略掉,留待以后再講。
在“發送請求”這里,實際也包含了兩種情形,一種簡單的,一種復雜的:
簡單的:假設不存在分叉,則直接檢查本地高度最高的區塊,然后請求下一個區塊
復雜的:考慮分叉的情況,則當前本地的區塊可能就存在分叉,那么到底應該請求哪個區塊,就需要慎重考慮
由于第2種情況對于本文來說過于復雜(因為需要深刻理解比原鏈中分叉的處理邏輯),所以在本文中將把問題簡化,只考慮第1種。而分叉的處理,將放在以后講解。
下面是把blockKeeper.BlockRequestWorker中的代碼簡化成了只包含第1種情況:
netsync/block_keeper.go#L72
func (bk *blockKeeper) BlockRequestWorker(peerID string, maxPeerHeight uint64) error { num := bk.chain.BestBlockHeight() + 1 reqNum := uint64(0) reqNum = num // ... bkPeer, ok := bk.peers.Peer(peerID) swPeer := bkPeer.getPeer() // ... block, err := bk.BlockRequest(peerID, reqNum) // ... }
在這種情況下,我們可以認為bk.chain.BestBlockHeight()中的Best,指的是本地持有的不帶分叉的區塊鏈高度最高的那個。(需要提醒的是,如果存在分叉情況,則Best不一定是高度最高的那個)
那么我們就可以直接向最佳peer請求下一個高度的區塊,它是通過bk.BlockRequest(peerID, reqNum)實現的:
netsync/block_keeper.go#L152
func (bk *blockKeeper) BlockRequest(peerID string, height uint64) (*types.Block, error) { var block *types.Block if err := bk.blockRequest(peerID, height); err != nil { return nil, errReqBlock } // ... for { select { case pendingResponse := <-bk.pendingProcessCh: block = pendingResponse.block // ... return block, nil // ... } } }
在上面簡化后的代碼中,主要分成了兩個部分。一個是發送請求bk.blockRequest(peerID, height),這是本文的重點;它下面的for-select部分,已經是在等待并處理對方節點的返回數據了,這部分我們今天先略過不講。
bk.blockRequest(peerID, height)這個方法,從邏輯上又可以分成兩部分:
構造出請求的信息
把信息發送給對方節點
構造出請求的信息bk.blockRequest(peerID, height)經過一連串的方法調用之后,使用height構造出了一個BlockRequestMessage對象,代碼如下:
netsync/block_keeper.go#L148
func (bk *blockKeeper) blockRequest(peerID string, height uint64) error { return bk.peers.requestBlockByHeight(peerID, height) }
netsync/peer.go#L332
func (ps *peerSet) requestBlockByHeight(peerID string, height uint64) error { peer, ok := ps.Peer(peerID) // ... return peer.requestBlockByHeight(height) }
netsync/peer.go#L73
func (p *peer) requestBlockByHeight(height uint64) error { msg := &BlockRequestMessage{Height: height} p.swPeer.TrySend(BlockchainChannel, struct{ BlockchainMessage }{msg}) return nil }
到這里,終于構造出了所需要的BlockRequestMessage,其實主要就是把height告訴peer。
然后,通過Peer的TrySend()把該信息發出去。
發送請求在TrySend中,主要是通過github.com/tendermint/go-wire庫將其序列化,再發送給對方。看起來應該是很簡單的操作吧,先預個警,還是挺繞的。
當我們進入TrySend()后:
p2p/peer.go#L242
func (p *Peer) TrySend(chID byte, msg interface{}) bool { if !p.IsRunning() { return false } return p.mconn.TrySend(chID, msg) }
發現它把鍋丟給了p.mconn.TrySend方法,那么mconn是什么?chID又是什么?
mconn是MConnection的實例,它是從哪兒來的?它應該在之前的某個地方初始化了,否則我們沒法直接調用它。所以我們先來找到它初始化的地方。
經過一番尋找,發現原來是在前一篇之后,即比原節點與另一個節點完成了身份驗證之后,具體的位置在Switch類啟動的地方。
我們這次直接從Swtich的OnStart作為起點:
p2p/switch.go#L186
func (sw *Switch) OnStart() error { //... // Start listeners for _, listener := range sw.listeners { go sw.listenerRoutine(listener) } return nil }
p2p/switch.go#L498
func (sw *Switch) listenerRoutine(l Listener) { for { inConn, ok := <-l.Connections() // ... err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig) // ... } }
p2p/switch.go#L645
func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error { // ... peer, err := newInboundPeerWithConfig(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config) // ... }
p2p/peer.go#L87
func newInboundPeerWithConfig(conn net.Conn, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), ourNodePrivKey crypto.PrivKeyEd25519, config *PeerConfig) (*Peer, error) { return newPeerFromConnAndConfig(conn, false, reactorsByCh, chDescs, onPeerError, ourNodePrivKey, config) }
p2p/peer.go#L91
func newPeerFromConnAndConfig(rawConn net.Conn, outbound bool, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), ourNodePrivKey crypto.PrivKeyEd25519, config *PeerConfig) (*Peer, error) { conn := rawConn // ... if config.AuthEnc { // ... conn, err = MakeSecretConnection(conn, ourNodePrivKey) // ... } // Key and NodeInfo are set after Handshake p := &Peer{ outbound: outbound, conn: conn, config: config, Data: cmn.NewCMap(), } p.mconn = createMConnection(conn, p, reactorsByCh, chDescs, onPeerError, config.MConfig) p.BaseService = *cmn.NewBaseService(nil, "Peer", p) return p, nil }
終于找到了。上面方法中的MakeSecretConnection就是與對方節點交換公鑰并進行身份驗證的地方,下面的p.mconn = createMConnection(...)就是創建mconn的地方。
繼續進去:
p2p/peer.go#L292
func createMConnection(conn net.Conn, p *Peer, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), config *MConnConfig) *MConnection { onReceive := func(chID byte, msgBytes []byte) { reactor := reactorsByCh[chID] if reactor == nil { if chID == PexChannel { return } else { cmn.PanicSanity(cmn.Fmt("Unknown channel %X", chID)) } } reactor.Receive(chID, p, msgBytes) } onError := func(r interface{}) { onPeerError(p, r) } return NewMConnectionWithConfig(conn, chDescs, onReceive, onError, config) }
原來mconn是MConnection的實例,它是通過NewMConnectionWithConfig創建的。
看了上面的代碼,發現這個MConnectionWithConfig與普通的net.Conn并沒有太大的區別,只不過是當收到了對方發來的數據后,會根據指定的chID調用相應的Reactor的Receive方法來處理。所以它起到了將數據分發給Reactor的作用。
為什么需要這樣的分發操作呢?這是因為,在比原中,節點之間交換數據,有多種不同的方式:
一種是規定了詳細的數據交互協議(比如有哪些信息類型,分別代表什么意思,什么情況下發哪個,如何應答等),在ProtocolReactor中實現,它對應的chID是BlockchainChannel,值為byte(0x40)
另一種使用了與BitTorrent類似的文件共享協議,叫PEX,在PEXReactor中實現,它對應的chID是PexChannel,值為byte(0x00)
所以節點之間發送信息的時候,需要知道對方發過來的數據對應的是哪一種方式,然后轉交給相應的Reactor去處理。
在比原中,前者是主要的方式,后者起到輔助作用。我們目前的文章中涉及到的都是前者,后者將在以后專門研究。
p.mconn.TrySend當我們知道了p.mconn.TrySend中的mconn是什么,并且在什么時候初始化以后,下面就可以進入它的TrySend方法了。
p2p/connection.go#L243
func (c *MConnection) TrySend(chID byte, msg interface{}) bool { // ... channel, ok := c.channelsIdx[chID] // ... ok = channel.trySendBytes(wire.BinaryBytes(msg)) if ok { // Wake up sendRoutine if necessary select { case c.send <- struct{}{}: default: } } return ok }
可以看到,它找到相應的channel后(在這里應該是ProtocolReactor對應的channel),調用channel的trySendBytes方法。在發送數據的時候,使用了github.com/tendermint/go-wire庫,將msg序列化為二進制數組。
p2p/connection.go#L602
func (ch *Channel) trySendBytes(bytes []byte) bool { select { case ch.sendQueue <- bytes: atomic.AddInt32(&ch.sendQueueSize, 1) return true default: return false } }
原來它是把要發送的數據,放到了該channel對應的sendQueue中,交由別人來發送。具體是由誰來發送,我們馬上要就找到它。
細心的同學會發現,Channel除了trySendBytes方法外,還有一個sendBytes(在本文中沒有用上):
p2p/connection.go#L589
func (ch *Channel) sendBytes(bytes []byte) bool { select { case ch.sendQueue <- bytes: atomic.AddInt32(&ch.sendQueueSize, 1) return true case <-time.After(defaultSendTimeout): return false } }
它們兩個的區別是,前者嘗試把待發送數據bytes放入ch.sendQueue時,如果能放進去,則返回true,否則馬上失敗,返回false,所以它是非阻塞的。而后者,如果放不進去(sendQueue已滿,那邊還沒處理完),則等待defaultSendTimeout(值為10秒),然后才會失敗。另外,sendQueue的容量默認為1。
到這里,我們其實已經知道比原是如何向其它節點請求區塊數據,以及何時把信息發送出去。
本想在本篇中就把真正發送數據的代碼也一起講了,但是發現它的邏輯也相當復雜,所以就另開一篇講吧。
再回到本文問題,再強調一下,我們前面說了,對于向peer請求區塊數據,有兩種情況:一種是簡單的不考慮分叉的,另一種是復雜的考慮分叉的。在本文只考慮了簡單的情況,在這種情況下,所謂的bestHeight就是指的最高的那個區塊的高度,而在復雜情況下,它就不一定了。這就留待以后我們再詳細討論,本文的問題就算是回答完畢了。
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/24157.html
摘要:所以本文本來是想去研究一下,當別的節點把區塊數據發給我們之后,我們應該怎么處理,現在換成研究比原的是怎么做出來的。進去后會看到大量的與相關的配置。它的功能主要是為了在訪問與的函數之間增加了一層轉換。 作者:freewind 比原項目倉庫: Github地址:https://github.com/Bytom/bytom Gitee地址:https://gitee.com/BytomBlo...
摘要:到這里,我們總算能夠完整的理解清楚,當我們向一個比原節點請求區塊數據,我們這邊需要怎么做,對方節點又需要怎么做了。 作者:freewind 比原項目倉庫: Github地址:https://github.com/Bytom/bytom Gitee地址:https://gitee.com/BytomBlockc... 在上一篇,我們知道了比原是如何把請求區塊數據的信息BlockReque...
摘要:作者比原項目倉庫地址地址在前一篇中,我們說到,當比原向其它節點請求區塊數據時,會發送一個把需要的區塊告訴對方,并把該信息對應的二進制數據放入對應的通道中,等待發送。這個就是真正與連接對象綁定的一個緩存區,寫入到它里面的數據,會被發送出去。 作者:freewind 比原項目倉庫: Github地址:https://github.com/Bytom/bytom Gitee地址:https:...
摘要:如果傳的是,就會在內部使用默認的隨機數生成器生成隨機數并生成密鑰。使用的是,生成的是一個形如這樣的全球唯一的隨機數把密鑰以文件形式保存在硬盤上。 作者:freewind 比原項目倉庫: Github地址:https://github.com/Bytom/bytom Gitee地址:https://gitee.com/BytomBlockc... 在前一篇,我們探討了從瀏覽器的dashb...
摘要:所以在今天我打算通過源代碼分析一下比原的挖礦流程,但是考慮到它肯定會涉及到比原的核心,所以太復雜的地方我就會先跳過,那些地方時機成熟的時候會徹底研究一下。 作者:freewind 比原項目倉庫: Github地址:https://github.com/Bytom/bytom Gitee地址:https://gitee.com/BytomBlockc... 當我們以bytom init ...
閱讀 1319·2021-11-24 09:38
閱讀 3256·2021-11-22 12:03
閱讀 4158·2021-11-11 10:59
閱讀 2317·2021-09-28 09:36
閱讀 1032·2021-09-09 09:32
閱讀 3411·2021-08-05 10:00
閱讀 2528·2021-07-23 15:30
閱讀 2973·2019-08-30 13:12