剥开比原看代码05：如何从比原节点拿到区块数据？

时间  2019-11-26
标签 剥开 代码 如何 节点 拿到 区块 数据 栏目 比特币 繁體版
原文   原文链接

做者：freewindnode

比原项目仓库：react

Github地址：https://github.com/Bytom/bytomgit

Gitee地址：https://gitee.com/BytomBlockchain/bytomgithub

在前一篇中，咱们已经知道如何连上一个比原节点的p2p端口，并与对方完成身份验证。此时，双方结点已经创建起来了信任，而且链接也不会断开，下一步，二者就能够继续交换数据了。数组

那么，我首先想到的就是，如何才能让对方把它已有的区块数据全都发给我呢？函数

这其实能够分为三个问题：区块链

  1. 我须要发给它什么样的数据？
  2. 它在内部由是如何应答的呢？
  3. 我拿到数据以后，应该怎么处理？

因为这一块的逻辑仍是比较复杂的，因此在本篇咱们先回答第一个问题：fetch

咱们要发送什么样的数据请求，才能让比原节点把它持有的区块数据发给我？

找到发送请求的代码

首先咱们先要在代码中定位到，比原究竟是在何时来向对方节点发送请求的。ui

在前一篇讲的是如何创建链接并验证身份，那么发出数据请求的操做，必定在上次的代码以后。按照这个思路，咱们在SyncManager类中Switch启动以后，找到了一个叫BlockKeeper的类，相关的操做是在它里面完成的。atom

下面是老规矩，仍是从启动开始，可是会更简化一些：

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的逻辑比较复杂，它包含了：

  1. 根据本身持有的区块数据来计算须要同步的数据
  2. 向前面找到的最佳节点发送数据请求
  3. 拿到对方发过来的区块数据
  4. 对数据进行处理
  5. 广播新状态
  6. 处理各类出错状况，等等

因为本文中只关注“发送请求”，因此一些与之关系不大的逻辑我会忽略掉，留待之后再讲。

在“发送请求”这里，实际也包含了两种情形，一种简单的，一种复杂的：

  1. 简单的：假设不存在分叉，则直接检查本地高度最高的区块，而后请求下一个区块
  2. 复杂的：考虑分叉的状况，则当前本地的区块可能就存在分叉，那么到底应该请求哪一个区块，就须要慎重考虑

因为第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)这个方法，从逻辑上又能够分红两部分：

  1. 构造出请求的信息
  2. 把信息发送给对方节点

构造出请求的信息

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的做用。

为何须要这样的分发操做呢？这是由于，在比原中，节点之间交换数据，有多种不一样的方式：

  1. 一种是规定了详细的数据交互协议（好比有哪些信息类型，分别表明什么意思，什么状况下发哪一个，如何应答等），在ProtocolReactor中实现，它对应的chID是BlockchainChannel，值为byte(0x40)
  2. 另外一种使用了与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就是指的最高的那个区块的高度，而在复杂状况下，它就不必定了。这就留待之后咱们再详细讨论，本文的问题就算是回答完毕了。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程