注册
关闭
区块链大帝

区块链大帝

发布于 1周前 阅读数 2448

精通Filecoin:Filecoin 源码之 Hello 协议

当调用 Filecoin 全节点的 start 方法启动全节点时,调用 hello 协议的 New 方法,这个方法的处理如下:

  1. 生成 Hello 对象。
    hello := &Handler{      host:              h,      genesis:           gen,      chainSyncCB:       syncCallback,      getHeaviestTipSet: getHeaviestTipSet,      net:               net,      commitSha:         commitSha,  }  

    其中 host 对象为底层 libp2 提供的 Host 对象;genesis 为创世区块的 CID;chainSyncCB 为全节点对象的 syncCallBack 函数,用于从远程节点同步区块;getHeaviestTipSetporcelain.API 对象的 ChainHead 方法(这个对象继承自 plumbing.API 对象,ChainHead 方法定义于后者),用于返回区块链头部的 tipset;net 表示当前的网络环境,比如测试网、正式网;

  2. 调用 host 对象的 SetStreamHandler 方法,设置自身的 handleNewStream 方法作为 /fil/hello/1.0.0 协议的处理器。
    h.SetStreamHandler(protocol, hello.handleNewStream)  

    当连接建立时,通过后面注册的通知者,从而调用 sayHello 方法,在这个方法中打开一个 hello 协议的流,并发送 hello 消息。

  3. 调用 Host 对象的网络对象(即 swarm 对象)的 Notify 方法,把自身作为被通知者注册到网络对象上。
    h.Network().Notify((*helloNotify)(hello))  

    在这一步,通过调用 host 对象的 Network 方法,返回底层的 swarm 对象,然后把 hello 对象转化为 helloNotify 对象,最后调用 swarm 对象的 Notify 方法,从而当底层的 swarm 对象有任何事件发生时都会通知 helloNotify 对象(即 hello 对象)。

    在 Hello 协议中我们只关心建立连接事件,所以 helloNotify 类型只实现了这个方法,其他方法都为空实现,具体如下:

    type helloNotify Handler

    func (hn *helloNotify) hello() *Handler {     return (*Handler)(hn) }

    const helloTimeout = time.Second * 10

    func (hn *helloNotify) Connected(n net.Network, c net.Conn) {     go func() {         ctx, cancel := context.WithTimeout(context.Background(), helloTimeout)         defer cancel()         p := c.RemotePeer()         if err := hn.hello().sayHello(ctx, p); err != nil {             log.Warningf("failed to send hello handshake to peer %s: %s", p, err)         }     }() }

    当节点作为客户端,拨号连接到远程对等节点时,底层的 swarm 对象会调用自身的 notifyAll 方法,通知所有的 Notify 对象有连接被打开,即调用所有 Notify 对象的 Connected 方法,包括前面我们注册的通知对象。当调用 helloNotify 对象的 Connected 方法时,这个方法内部调用自身的 hello 方法,后者返回自身并强制转化为 Handler 类型,然后调用它的 sayHello 方法,对我们当前连接的远程进行打招呼。

    与此同时,当远程节点作为服务器,接收到我们发送的连接请求生成连接时,它的 swarm 对象也会通知它的所有 Notify 对象,从而也会它的前面注册的通知对象,即调用服务器商的 Connected 方法,从而调用它的 sayHello 方法向我们发送它的区块情况;因为第二步中,我们把 Hello 对象的 handleNewStream 方法注册为 Hello 协议的处理器,所以当节点接收到远程节点发送区块情况时,就会调用这个方法进行处理,这个方法又会调用调用全节点的 syncCallBack 方法进行区块同步处理。

总体上来说,Hello 协议通过 sayHellohandleNewStream 开启了区块同步,前者把自身的区块情况发送到远程节点,后者处理远程节点发送的区块情况。

sayHello 方法处理如下:

  1. 调用 Host 对象的 NewStream,生成一个处理 Hello 协议的流对象。
    s, err := h.host.NewStream(ctx, p, protocol)  if err != nil {      return err  }  defer s.Close() // nolint: errcheck  
  2. 调用自身的 getOurHelloMessage 方法,获取自身区块链顶端的信息。
    msg := h.getOurHelloMessage()  

    这个方法内部执行流程如下:

    • 调用自身 getHeaviestTipSet 方法,获取区块链顶端的信息这个方法是 plumbing.API 对象 ChainHead 方法的引用。
    • 使用获取到的区块链信息,生成并返回消息对象 Message
  3. 通过流发送区块信息到远程节点。

handleNewStream 方法处理如下:

  1. 生成消息对象 Message,并从流中读取远程对等节点发送过来的内容到消息对象中。
    var hello Message  if err := cbu.NewMsgReader(s).ReadMsg(&hello); err != nil {      log.Debugf("bad hello message from peer %s: %s", from, err)      helloMsgErrCt.Inc(context.TODO(), 1)      s.Conn().Close() // nolint: errcheck      return  }  
  2. 调用自身的 processHelloMessage 方法,处理远程节点发送的消息。这个方法代码如下:
    func (h *Handler) processHelloMessage(from peer.ID, msg *Message) error {      if !msg.GenesisHash.Equals(h.genesis) {          return ErrBadGenesis      }      if (h.net == "devnet-test" || h.net == "devnet-user") && msg.CommitSha != h.commitSha {          return ErrWrongVersion      }
    h.chainSyncCB(from, msg.HeaviestTipSetCids, msg.HeaviestTipSetHeight)  return nil  
    }  

    它的处理逻辑比较简单:

    • 首先,检查远程节点发送的创世区块哈希是否自身的创世区块哈希相等。如果不等,直接返回错误。
    • 然后,检查网络类型。
    • 最终,调用自身的 chainSyncCB 方法,处理远程节点发送的区块信息。这个同步回调方法对象在全节点的启动方法 Start 中生成。它的主要生成就是根据远程节点发送的区块链最顶层的信息,生成一个 types/SortedCidSet 对象,然后调用 chain/syncer.go 中的 HandleNewTipset 方法来处理远程发送的区块信息。
  3. 根据前面处理消息的结果进行不同的处理。
    switch err := h.processHelloMessage(from, &hello); err {  case ErrBadGenesis:      log.Debugf("genesis cid: %s does not match: %s, disconnecting from peer: %s", &hello.GenesisHash, h.genesis, from)      genesisErrCt.Inc(context.TODO(), 1)      s.Conn().Close() // nolint: errcheck      return  case ErrWrongVersion:      log.Debugf("code not at same version: peer has version %s, daemon has version %s, disconnecting from peer: %s", hello.CommitSha, h.commitSha, from)      versionErrCt.Inc(context.TODO(), 1)      s.Conn().Close() // nolint: errcheck      return  case nil: // ok, noop  default:      log.Error(err)  }     
  • 0
区块链大帝
区块链大帝

0 条评论