【比原鏈】如何從比原節點拿到區塊數據？
在前一篇中，我們已經知道如何連上一個比原節點的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 －－ 也就是本文的主角 －－ 的世界。
BlockKeeper
blockKeeper.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就是指的最高的那個區塊的高度，而在復雜情況下，它就不一定了。這就留待以後我們再詳細討論，本文的問題就算是回答完畢了。
【比原鏈】如何從比原節點拿到區塊數據？