eth共识算法分析,从本地节点挖到块开始分析

首先目前生产环境上面，肯定不是以CPU的形式挖矿的,那么就是remoteAgent这种形式,也就是矿机通过网络请求从以太的节点获取当前节点的出块任务,

然后矿机根据算出符合该块难度hash值,提交给节点,也就是对应的以下方法.

  1. func (a *RemoteAgent) SubmitWork(nonce types.BlockNonce, mixDigest, hash common.Hash) bool {
  2.     a.mu.Lock()
  3.     defer a.mu.Unlock()
  5.     // Make sure the work submitted is present
  6.     work := a.work[hash]
  7.     if work == nil {
  8.         log.Info("Work submitted but none pending", "hash", hash)
  9.         return false
  10.     }
  11.     // Make sure the Engine solutions is indeed valid
  12.     result := work.Block.Header()
  13.     result.Nonce = nonce
  14.     result.MixDigest = mixDigest
  16.     if err := a.engine.VerifySeal(a.chain, result); err != nil {
  17.         log.Warn("Invalid proof-of-work submitted", "hash", hash, "err", err)
  18.         return false
  19.     }
  20.     block := work.Block.WithSeal(result)
  22.     // Solutions seems to be valid, return to the miner and notify acceptance
  23.     a.returnCh <- &Result{work, block}
  24.     delete(a.work, hash)
  26.     return true
  27. }

该方法会校验提交过来的块的hash难度,如果是正常的话,则会将该生成的块写到管道中,管道接收的方法在/miner/worker.go/Wait方法中

  1. func (self *worker) wait() {
  2.     for {
  3.         mustCommitNewWork := true
  4.         for result := range self.recv {
  5.             atomic.AddInt32(&self.atWork, -1)
  7.             if result == nil {
  8.                 continue
  9.             }
  10.             block := result.Block
  11.             work := result.Work
  13.             // Update the block hash in all logs since it is now available and not when the
  14.             // receipt/log of individual transactions were created.
  15.             for _, r := range work.receipts {
  16.                 for _, l := range r.Logs {
  17.                     l.BlockHash = block.Hash()
  18.                 }
  19.             }
  20.             for _, log := range work.state.Logs() {
  21.                 log.BlockHash = block.Hash()
  22.             }
  23.             stat, err := self.chain.WriteBlockAndState(block, work.receipts, work.state)
  24.             if err != nil {
  25.                 log.Error("Failed writing block to chain", "err", err)
  26.                 continue
  27.             }
  28.             // check if canon block and write transactions
  29.             if stat == core.CanonStatTy {
  30.                 // implicit by posting ChainHeadEvent
  31.                 mustCommitNewWork = false
  32.             }
  33.             // Broadcast the block and announce chain insertion event
  34.             //　通过p2p的形式将块广播到连接的节点,走的还是channel
  35.             self.mux.Post(core.NewMinedBlockEvent{Block: block})
  36.             var (
  37.                 events []interface{}
  38.                 logs   = work.state.Logs()
  39.             )
  40.             events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs})
  41.             if stat == core.CanonStatTy {
  42.                 events = append(events, core.ChainHeadEvent{Block: block})
  43.             }
  44.             self.chain.PostChainEvents(events, logs)
  46.             // Insert the block into the set of pending ones to wait for confirmations
  47.             self.unconfirmed.Insert(block.NumberU64(), block.Hash())
  49.             if mustCommitNewWork {
  50.                 self.commitNewWork()
  51.             }
  52.         }
  53.     }
  54. }

这里发送了一个新挖到块的事件,接着跟,调用栈是

  1. /geth/main.go/geth　--> startNode　--> utils.StartNode(stack)
  2. --> stack.Start()　--> /node/node.go/Start() --> service.Start(running)
  3. --> /eth/backend.go/Start() --> /eth/handler.go/Start()

好了核心逻辑在handler.go/Start()里面

  1. func (pm *ProtocolManager) Start(maxPeers int) {
  2.     pm.maxPeers = maxPeers
  4.     // broadcast transactions
  5.     // 广播交易的通道。 txCh会作为txpool的TxPreEvent订阅通道。txpool有了这种消息会通知给这个txCh。 广播交易的goroutine会把这个消息广播出去。
  6.     pm.txCh = make(chan core.TxPreEvent, txChanSize)
  7.     // 订阅的回执
  8.     pm.txSub = pm.txpool.SubscribeTxPreEvent(pm.txCh)
  9.     go pm.txBroadcastLoop()
  11.     // 订阅挖矿消息。当新的Block被挖出来的时候会产生消息。 这个订阅和上面的那个订阅采用了两种不同的模式，这种是标记为Deprecated的订阅方式。
  12.     // broadcast mined blocks
  13.     pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
  14.     //　挖矿广播 goroutine 当挖出来的时候需要尽快的广播到网络上面去 本地挖出的块通过这种形式广播出去
  15.     go pm.minedBroadcastLoop()
  16.     // 同步器负责周期性地与网络同步，下载散列和块以及处理通知处理程序。
  17.     // start sync handlers
  18.     go pm.syncer()
  19.     // txsyncLoop负责每个新连接的初始事务同步。 当新的peer出现时，我们转发所有当前待处理的事务。 为了最小化出口带宽使用，我们一次只发送一个小包。
  20.     go pm.txsyncLoop()
  21. }

pm.minedBroadcastLoop()里面就有管道接收到上面post出来的出块消息,跟进去将会看到通过p2p网络发送给节点的逻辑

  1. // BroadcastBlock will either propagate a block to a subset of it's peers, or
  2. // will only announce it's availability (depending what's requested).
  3. func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
  4.     hash := block.Hash()
  5.     peers := pm.peers.PeersWithoutBlock(hash)
  7.     // If propagation is requested, send to a subset of the peer
  8.     if propagate {
  9.         // Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
  10.         var td *big.Int
  11.         if parent := pm.blockchain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {
  12.             td = new(big.Int).Add(block.Difficulty(), pm.blockchain.GetTd(block.ParentHash(), block.NumberU64()-1))
  13.         } else {
  14.             log.Error("Propagating dangling block", "number", block.Number(), "hash", hash)
  15.             return
  16.         }
  17.         // Send the block to a subset of our peers
  18.         transfer := peers[:int(math.Sqrt(float64(len(peers))))]
  19.         for _, peer := range transfer {
  20.             peer.SendNewBlock(block, td)
  21.         }
  22.         log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
  23.         return
  24.     }
  25.     // Otherwise if the block is indeed in out own chain, announce it
  26.     if pm.blockchain.HasBlock(hash, block.NumberU64()) {
  27.         for _, peer := range peers {
  28.             peer.SendNewBlockHashes([]common.Hash{hash}, []uint64{block.NumberU64()})
  29.         }
  30.         log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
  31.     }
  32. }

这里面会发送两种时间，一种是NewBlockMsg,另外一种是NewBlockHashesMsg,好了到此本地节点挖到的块就通过p2p网络的形式开始扩散出去了 接着看下一个重要的方法

  1. // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
  2. // with the ethereum network.
  3. func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, networkId uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database) (*ProtocolManager, error) {
  4.     // Create the protocol manager with the base fields
  5.     manager := &ProtocolManager{
  6.         networkId:   networkId,
  7.         eventMux:    mux,
  8.         txpool:      txpool,
  9.         blockchain:  blockchain,
  10.         chaindb:     chaindb,
  11.         chainconfig: config,
  12.         peers:       newPeerSet(),
  13.         newPeerCh:   make(chan *peer),
  14.         noMorePeers: make(chan struct{}),
  15.         txsyncCh:    make(chan *txsync),
  16.         quitSync:    make(chan struct{}),
  17.     }
  18.     // Figure out whether to allow fast sync or not
  19.     if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 {
  20.         log.Warn("Blockchain not empty, fast sync disabled")
  21.         mode = downloader.FullSync
  22.     }
  23.     if mode == downloader.FastSync {
  24.         manager.fastSync = uint32(1)
  25.     }
  26.     // Initiate a sub-protocol for every implemented version we can handle
  27.     manager.SubProtocols = make([]p2p.Protocol, 0, len(ProtocolVersions))
  28.     for i, version := range ProtocolVersions {
  29.         // Skip protocol version if incompatible with the mode of operation
  30.         if mode == downloader.FastSync && version < eth63 {
  31.             continue
  32.         }
  33.         // Compatible; initialise the sub-protocol
  34.         version := version // Closure for the run
  35.         manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{
  36.             Name:    ProtocolName,
  37.             Version: version,
  38.             Length:  ProtocolLengths[i],
  39.             Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
  40.                 peer := manager.newPeer(int(version), p, rw)
  41.                 select {
  42.                 case manager.newPeerCh <- peer:
  43.                     manager.wg.Add(1)
  44.                     defer manager.wg.Done()
  45.                     return manager.handle(peer)
  46.                 case <-manager.quitSync:
  47.                     return p2p.DiscQuitting
  48.                 }
  49.             },
  50.             NodeInfo: func() interface{} {
  51.                 return manager.NodeInfo()
  52.             },
  53.             PeerInfo: func(id discover.NodeID) interface{} {
  54.                 if p := manager.peers.Peer(fmt.Sprintf("%x", id[:8])); p != nil {
  55.                     return p.Info()
  56.                 }
  57.                 return nil
  58.             },
  59.         })
  60.     }
  61.     if len(manager.SubProtocols) == 0 {
  62.         return nil, errIncompatibleConfig
  63.     }
  64.     // downloader是负责从其他的peer来同步自身数据。
  65.     // downloader是全链同步工具
  66.     // Construct the different synchronisation mechanisms
  67.     manager.downloader = downloader.New(mode, chaindb, manager.eventMux, blockchain, nil, manager.removePeer)
  69.     validator := func(header *types.Header) error {
  70.         return engine.VerifyHeader(blockchain, header, true)
  71.     }
  72.     heighter := func() uint64 {
  73.         return blockchain.CurrentBlock().NumberU64()
  74.     }
  75.     inserter := func(blocks types.Blocks) (int, error) {
  76.         // If fast sync is running, deny importing weird blocks
  77.         if atomic.LoadUint32(&manager.fastSync) == 1 {
  78.             log.Warn("Discarded bad propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
  79.             return 0, nil
  80.         }
  81.         // 设置开始接收交易
  82.         atomic.StoreUint32(&manager.acceptTxs, 1) // Mark initial sync done on any fetcher import
  83.         return manager.blockchain.InsertChain(blocks)
  84.     }
  85.     // 生成一个fetcher
  86.     // Fetcher负责积累来自各个peer的区块通知并安排进行检索。
  87.     manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)
  89.     return manager, nil
  90. }

该方法是用来管理以太坊协议下的多个子协议,其中的Run方法在每个节点启动的时候就会调用,可以看到是阻塞的,跟进handler方法能看到这样的一块关键代码

  1. for {
  2.         if err := pm.handleMsg(p); err != nil {
  3.             p.Log().Debug("Ethereum message handling failed", "err", err)
  4.             return err
  5.         }
  6.     }

死循环,处理p2p网络过来的消息,接着看handleMsg方法

  1. func (pm *ProtocolManager) handleMsg(p *peer) error {
  2.     // Read the next message from the remote peer, and ensure it's fully consumed
  3.     msg, err := p.rw.ReadMsg()
  4.     if err != nil {
  5.         return err
  6.     }
  7.     if msg.Size > ProtocolMaxMsgSize {
  8.         return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
  9.     }
  10.     defer msg.Discard()
  12.     // Handle the message depending on its contents
  13.     switch {
  14.     case msg.Code == StatusMsg:
  15.         // Status messages should never arrive after the handshake
  16.         return errResp(ErrExtraStatusMsg, "uncontrolled status message")
  18.     // Block header query, collect the requested headers and reply
  19.     case msg.Code == GetBlockHeadersMsg:
  20.         // Decode the complex header query
  21.         var query getBlockHeadersData
  22.         if err := msg.Decode(&query); err != nil {
  23.             return errResp(ErrDecode, "%v: %v", msg, err)
  24.         }
  25.         hashMode := query.Origin.Hash != (common.Hash{})
  27.         // Gather headers until the fetch or network limits is reached
  28.         var (
  29.             bytes   common.StorageSize
  30.             headers []*types.Header
  31.             unknown bool
  32.         )
  33.         for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit && len(headers) < downloader.MaxHeaderFetch {
  34.             // Retrieve the next header satisfying the query
  35.             var origin *types.Header
  36.             if hashMode {
  37.                 origin = pm.blockchain.GetHeaderByHash(query.Origin.Hash)
  38.             } else {
  39.                 origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number)
  40.             }
  41.             if origin == nil {
  42.                 break
  43.             }
  44.             number := origin.Number.Uint64()
  45.             headers = append(headers, origin)
  46.             bytes += estHeaderRlpSize
  48.             // Advance to the next header of the query
  49.             switch {
  50.             case query.Origin.Hash != (common.Hash{}) && query.Reverse:
  51.                 // Hash based traversal towards the genesis block
  52.                 for i := 0; i < int(query.Skip)+1; i++ {
  53.                     if header := pm.blockchain.GetHeader(query.Origin.Hash, number); header != nil {
  54.                         query.Origin.Hash = header.ParentHash
  55.                         number--
  56.                     } else {
  57.                         unknown = true
  58.                         break
  59.                     }
  60.                 }
  61.             case query.Origin.Hash != (common.Hash{}) && !query.Reverse:
  62.                 // Hash based traversal towards the leaf block
  63.                 var (
  64.                     current = origin.Number.Uint64()
  65.                     next    = current + query.Skip + 1
  66.                 )
  67.                 if next <= current {
  68.                     infos, _ := json.MarshalIndent(p.Peer.Info(), "", "  ")
  69.                     p.Log().Warn("GetBlockHeaders skip overflow attack", "current", current, "skip", query.Skip, "next", next, "attacker", infos)
  70.                     unknown = true
  71.                 } else {
  72.                     if header := pm.blockchain.GetHeaderByNumber(next); header != nil {
  73.                         if pm.blockchain.GetBlockHashesFromHash(header.Hash(), query.Skip+1)[query.Skip] == query.Origin.Hash {
  74.                             query.Origin.Hash = header.Hash()
  75.                         } else {
  76.                             unknown = true
  77.                         }
  78.                     } else {
  79.                         unknown = true
  80.                     }
  81.                 }
  82.             case query.Reverse:
  83.                 // Number based traversal towards the genesis block
  84.                 if query.Origin.Number >= query.Skip+1 {
  85.                     query.Origin.Number -= (query.Skip + 1)
  86.                 } else {
  87.                     unknown = true
  88.                 }
  90.             case !query.Reverse:
  91.                 // Number based traversal towards the leaf block
  92.                 query.Origin.Number += (query.Skip + 1)
  93.             }
  94.         }
  95.         return p.SendBlockHeaders(headers)
  97.     case msg.Code == BlockHeadersMsg:
  98.         // A batch of headers arrived to one of our previous requests
  99.         var headers []*types.Header
  100.         if err := msg.Decode(&headers); err != nil {
  101.             return errResp(ErrDecode, "msg %v: %v", msg, err)
  102.         }
  103.         // If no headers were received, but we're expending a DAO fork check, maybe it's that
  104.         if len(headers) == 0 && p.forkDrop != nil {
  105.             // Possibly an empty reply to the fork header checks, sanity check TDs
  106.             verifyDAO := true
  108.             // If we already have a DAO header, we can check the peer's TD against it. If
  109.             // the peer's ahead of this, it too must have a reply to the DAO check
  110.             if daoHeader := pm.blockchain.GetHeaderByNumber(pm.chainconfig.DAOForkBlock.Uint64()); daoHeader != nil {
  111.                 if _, td := p.Head(); td.Cmp(pm.blockchain.GetTd(daoHeader.Hash(), daoHeader.Number.Uint64())) >= 0 {
  112.                     verifyDAO = false
  113.                 }
  114.             }
  115.             // If we're seemingly on the same chain, disable the drop timer
  116.             if verifyDAO {
  117.                 p.Log().Debug("Seems to be on the same side of the DAO fork")
  118.                 p.forkDrop.Stop()
  119.                 p.forkDrop = nil
  120.                 return nil
  121.             }
  122.         }
  123.         // Filter out any explicitly requested headers, deliver the rest to the downloader
  124.         filter := len(headers) == 1
  125.         if filter {
  126.             // If it's a potential DAO fork check, validate against the rules
  127.             if p.forkDrop != nil && pm.chainconfig.DAOForkBlock.Cmp(headers[0].Number) == 0 {
  128.                 // Disable the fork drop timer
  129.                 p.forkDrop.Stop()
  130.                 p.forkDrop = nil
  132.                 // Validate the header and either drop the peer or continue
  133.                 if err := misc.VerifyDAOHeaderExtraData(pm.chainconfig, headers[0]); err != nil {
  134.                     p.Log().Debug("Verified to be on the other side of the DAO fork, dropping")
  135.                     return err
  136.                 }
  137.                 p.Log().Debug("Verified to be on the same side of the DAO fork")
  138.                 return nil
  139.             }
  140.             // Irrelevant of the fork checks, send the header to the fetcher just in case
  141.             headers = pm.fetcher.FilterHeaders(p.id, headers, time.Now())
  142.         }
  143.         if len(headers) > 0 || !filter {
  144.             err := pm.downloader.DeliverHeaders(p.id, headers)
  145.             if err != nil {
  146.                 log.Debug("Failed to deliver headers", "err", err)
  147.             }
  148.         }
  150.     case msg.Code == GetBlockBodiesMsg:
  151.         // Decode the retrieval message
  152.         msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
  153.         if _, err := msgStream.List(); err != nil {
  154.             return err
  155.         }
  156.         // Gather blocks until the fetch or network limits is reached
  157.         var (
  158.             hash   common.Hash
  159.             bytes  int
  160.             bodies []rlp.RawValue
  161.         )
  162.         for bytes < softResponseLimit && len(bodies) < downloader.MaxBlockFetch {
  163.             // Retrieve the hash of the next block
  164.             if err := msgStream.Decode(&hash); err == rlp.EOL {
  165.                 break
  166.             } else if err != nil {
  167.                 return errResp(ErrDecode, "msg %v: %v", msg, err)
  168.             }
  169.             // Retrieve the requested block body, stopping if enough was found
  170.             if data := pm.blockchain.GetBodyRLP(hash); len(data) != 0 {
  171.                 bodies = append(bodies, data)
  172.                 bytes += len(data)
  173.             }
  174.         }
  175.         return p.SendBlockBodiesRLP(bodies)
  177.     case msg.Code == BlockBodiesMsg:
  178.         // A batch of block bodies arrived to one of our previous requests
  179.         var request blockBodiesData
  180.         if err := msg.Decode(&request); err != nil {
  181.             return errResp(ErrDecode, "msg %v: %v", msg, err)
  182.         }
  183.         // Deliver them all to the downloader for queuing
  184.         trasactions := make([][]*types.Transaction, len(request))
  185.         uncles := make([][]*types.Header, len(request))
  187.         for i, body := range request {
  188.             trasactions[i] = body.Transactions
  189.             uncles[i] = body.Uncles
  190.         }
  191.         // Filter out any explicitly requested bodies, deliver the rest to the downloader
  192.         filter := len(trasactions) > 0 || len(uncles) > 0
  193.         if filter {
  194.             trasactions, uncles = pm.fetcher.FilterBodies(p.id, trasactions, uncles, time.Now())
  195.         }
  196.         if len(trasactions) > 0 || len(uncles) > 0 || !filter {
  197.             err := pm.downloader.DeliverBodies(p.id, trasactions, uncles)
  198.             if err != nil {
  199.                 log.Debug("Failed to deliver bodies", "err", err)
  200.             }
  201.         }
  203.     case p.version >= eth63 && msg.Code == GetNodeDataMsg:
  204.         // Decode the retrieval message
  205.         msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
  206.         if _, err := msgStream.List(); err != nil {
  207.             return err
  208.         }
  209.         // Gather state data until the fetch or network limits is reached
  210.         var (
  211.             hash  common.Hash
  212.             bytes int
  213.             data  [][]byte
  214.         )
  215.         for bytes < softResponseLimit && len(data) < downloader.MaxStateFetch {
  216.             // Retrieve the hash of the next state entry
  217.             if err := msgStream.Decode(&hash); err == rlp.EOL {
  218.                 break
  219.             } else if err != nil {
  220.                 return errResp(ErrDecode, "msg %v: %v", msg, err)
  221.             }
  222.             // Retrieve the requested state entry, stopping if enough was found
  223.             if entry, err := pm.chaindb.Get(hash.Bytes()); err == nil {
  224.                 data = append(data, entry)
  225.                 bytes += len(entry)
  226.             }
  227.         }
  228.         return p.SendNodeData(data)
  230.     case p.version >= eth63 && msg.Code == NodeDataMsg:
  231.         // A batch of node state data arrived to one of our previous requests
  232.         var data [][]byte
  233.         if err := msg.Decode(&data); err != nil {
  234.             return errResp(ErrDecode, "msg %v: %v", msg, err)
  235.         }
  236.         // Deliver all to the downloader
  237.         if err := pm.downloader.DeliverNodeData(p.id, data); err != nil {
  238.             log.Debug("Failed to deliver node state data", "err", err)
  239.         }
  241.     case p.version >= eth63 && msg.Code == GetReceiptsMsg:
  242.         // Decode the retrieval message
  243.         msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
  244.         if _, err := msgStream.List(); err != nil {
  245.             return err
  246.         }
  247.         // Gather state data until the fetch or network limits is reached
  248.         var (
  249.             hash     common.Hash
  250.             bytes    int
  251.             receipts []rlp.RawValue
  252.         )
  253.         for bytes < softResponseLimit && len(receipts) < downloader.MaxReceiptFetch {
  254.             // Retrieve the hash of the next block
  255.             if err := msgStream.Decode(&hash); err == rlp.EOL {
  256.                 break
  257.             } else if err != nil {
  258.                 return errResp(ErrDecode, "msg %v: %v", msg, err)
  259.             }
  260.             // Retrieve the requested block's receipts, skipping if unknown to us
  261.             results := core.GetBlockReceipts(pm.chaindb, hash, core.GetBlockNumber(pm.chaindb, hash))
  262.             if results == nil {
  263.                 if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash {
  264.                     continue
  265.                 }
  266.             }
  267.             // If known, encode and queue for response packet
  268.             if encoded, err := rlp.EncodeToBytes(results); err != nil {
  269.                 log.Error("Failed to encode receipt", "err", err)
  270.             } else {
  271.                 receipts = append(receipts, encoded)
  272.                 bytes += len(encoded)
  273.             }
  274.         }
  275.         return p.SendReceiptsRLP(receipts)
  277.     case p.version >= eth63 && msg.Code == ReceiptsMsg:
  278.         // A batch of receipts arrived to one of our previous requests
  279.         var receipts [][]*types.Receipt
  280.         if err := msg.Decode(&receipts); err != nil {
  281.             return errResp(ErrDecode, "msg %v: %v", msg, err)
  282.         }
  283.         // Deliver all to the downloader
  284.         if err := pm.downloader.DeliverReceipts(p.id, receipts); err != nil {
  285.             log.Debug("Failed to deliver receipts", "err", err)
  286.         }
  288.     case msg.Code == NewBlockHashesMsg:
  289.         var announces newBlockHashesData
  290.         if err := msg.Decode(&announces); err != nil {
  291.             return errResp(ErrDecode, "%v: %v", msg, err)
  292.         }
  293.         // Mark the hashes as present at the remote node
  294.         for _, block := range announces {
  295.             p.MarkBlock(block.Hash)
  296.         }
  297.         // Schedule all the unknown hashes for retrieval
  298.         unknown := make(newBlockHashesData, 0, len(announces))
  299.         for _, block := range announces {
  300.             if !pm.blockchain.HasBlock(block.Hash, block.Number) {
  301.                 unknown = append(unknown, block)
  302.             }
  303.         }
  304.         for _, block := range unknown {
  305.             pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestOneHeader, p.RequestBodies)
  306.         }
  308.     case msg.Code == NewBlockMsg:
  309.         // Retrieve and decode the propagated block
  310.         var request newBlockData
  311.         if err := msg.Decode(&request); err != nil {
  312.             return errResp(ErrDecode, "%v: %v", msg, err)
  313.         }
  314.         request.Block.ReceivedAt = msg.ReceivedAt
  315.         request.Block.ReceivedFrom = p
  317.         // Mark the peer as owning the block and schedule it for import
  318.         p.MarkBlock(request.Block.Hash())
  319.         pm.fetcher.Enqueue(p.id, request.Block)
  321.         // Assuming the block is importable by the peer, but possibly not yet done so,
  322.         // calculate the head hash and TD that the peer truly must have.
  323.         var (
  324.             trueHead = request.Block.ParentHash()
  325.             trueTD   = new(big.Int).Sub(request.TD, request.Block.Difficulty())
  326.         )
  327.         // Update the peers total difficulty if better than the previous
  328.         if _, td := p.Head(); trueTD.Cmp(td) > 0 {
  329.             p.SetHead(trueHead, trueTD)
  331.             // Schedule a sync if above ours. Note, this will not fire a sync for a gap of
  332.             // a singe block (as the true TD is below the propagated block), however this
  333.             // scenario should easily be covered by the fetcher.
  334.             currentBlock := pm.blockchain.CurrentBlock()
  335.             if trueTD.Cmp(pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())) > 0 {
  336.                 go pm.synchronise(p)
  337.             }
  338.         }
  340.     case msg.Code == TxMsg:
  341.         // Transactions arrived, make sure we have a valid and fresh chain to handle them
  342.         if atomic.LoadUint32(&pm.acceptTxs) == 0 {
  343.             break
  344.         }
  345.         // Transactions can be processed, parse all of them and deliver to the pool
  346.         var txs []*types.Transaction
  347.         if err := msg.Decode(&txs); err != nil {
  348.             return errResp(ErrDecode, "msg %v: %v", msg, err)
  349.         }
  350.         for i, tx := range txs {
  351.             // Validate and mark the remote transaction
  352.             if tx == nil {
  353.                 return errResp(ErrDecode, "transaction %d is nil", i)
  354.             }
  355.             p.MarkTransaction(tx.Hash())
  356.         }
  357.         pm.txpool.AddRemotes(txs)
  359.     default:
  360.         return errResp(ErrInvalidMsgCode, "%v", msg.Code)
  361.     }
  362.     return nil
  363. }

该方法中就解码了p2p网络过来的消息，并且处理了NewBlockMsg和NewBlockHashesMsg这两种事件,如NewBlockMsg中的处理逻辑是直接通过管道发送到本地了,pm.fetcher.Enqueue(p.id, request.Block) ,对应的管道名是:f.inject,其中是一个队列,/fetcher.go/enqueue方法中写入了一个FIFO队列中

  1. func (f *Fetcher) enqueue(peer string, block *types.Block) {
  2.     hash := block.Hash()
  4.     // Ensure the peer isn't DOSing us
  5.     count := f.queues[peer] + 1
  6.     if count > blockLimit {
  7.         log.Debug("Discarded propagated block, exceeded allowance", "peer", peer, "number", block.Number(), "hash", hash, "limit", blockLimit)
  8.         propBroadcastDOSMeter.Mark(1)
  9.         f.forgetHash(hash)
  10.         return
  11.     }
  12.     // Discard any past or too distant blocks
  13.     if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist {
  14.         log.Debug("Discarded propagated block, too far away", "peer", peer, "number", block.Number(), "hash", hash, "distance", dist)
  15.         propBroadcastDropMeter.Mark(1)
  16.         f.forgetHash(hash)
  17.         return
  18.     }
  19.     // Schedule the block for future importing
  20.     if _, ok := f.queued[hash]; !ok {
  21.         op := &inject{
  22.             origin: peer,
  23.             block:  block,
  24.         }
  25.         f.queues[peer] = count
  26.         f.queued[hash] = op
  27.         f.queue.Push(op, -float32(block.NumberU64()))
  28.         if f.queueChangeHook != nil {
  29.             f.queueChangeHook(op.block.Hash(), true)
  30.         }
  31.         log.Debug("Queued propagated block", "peer", peer, "number", block.Number(), "hash", hash, "queued", f.queue.Size())
  32.     }
  33. }

该队列的消费端在/fetcher.go/loop中,是一个死循环,核心代码

  1. for !f.queue.Empty() {
  2.             op := f.queue.PopItem().(*inject)
  3.             if f.queueChangeHook != nil {
  4.                 f.queueChangeHook(op.block.Hash(), false)
  5.             }
  6.             // If too high up the chain or phase, continue later
  7.             number := op.block.NumberU64()
  8.             if number > height+1 {
  9.                 f.queue.Push(op, -float32(op.block.NumberU64()))
  10.                 if f.queueChangeHook != nil {
  11.                     f.queueChangeHook(op.block.Hash(), true)
  12.                 }
  13.                 break
  14.             }
  15.             // Otherwise if fresh and still unknown, try and import
  16.             hash := op.block.Hash()
  17.             if number+maxUncleDist < height || f.getBlock(hash) != nil {
  18.                 f.forgetBlock(hash)
  19.                 continue
  20.             }
  21.             f.insert(op.origin, op.block)
  22.         }

从队列中取出,接着看insert方法

  1. func (f *Fetcher) insert(peer string, block *types.Block) {
  2.     hash := block.Hash()
  4.     // Run the import on a new thread
  5.     log.Debug("Importing propagated block", "peer", peer, "number", block.Number(), "hash", hash)
  6.     go func() {
  7.         defer func() { f.done <- hash }()
  9.         // If the parent's unknown, abort insertion
  10.         parent := f.getBlock(block.ParentHash())
  11.         if parent == nil {
  12.             log.Debug("Unknown parent of propagated block", "peer", peer, "number", block.Number(), "hash", hash, "parent", block.ParentHash())
  13.             return
  14.         }
  15.         // Quickly validate the header and propagate the block if it passes
  16.         switch err := f.verifyHeader(block.Header()); err {
  17.         case nil:
  18.             // All ok, quickly propagate to our peers
  19.             propBroadcastOutTimer.UpdateSince(block.ReceivedAt)
  20.             go f.broadcastBlock(block, true)
  22.         case consensus.ErrFutureBlock:
  23.             // Weird future block, don't fail, but neither propagate
  25.         default:
  26.             // Something went very wrong, drop the peer
  27.             log.Debug("Propagated block verification failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
  28.             f.dropPeer(peer)
  29.             return
  30.         }
  31.         // Run the actual import and log any issues
  32.         if _, err := f.insertChain(types.Blocks{block}); err != nil {
  33.             log.Debug("Propagated block import failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
  34.             return
  35.         }
  36.         // If import succeeded, broadcast the block
  37.         propAnnounceOutTimer.UpdateSince(block.ReceivedAt)
  38.         go f.broadcastBlock(block, false)
  40.         // Invoke the testing hook if needed
  41.         if f.importedHook != nil {
  42.             f.importedHook(block)
  43.         }
  44.     }()
  45. }

可以看到,该方法会调用verifyHeader方法去校验区块，如果没问题的话就通过p2p的形式广播出去,然后调用insertChain方法插入到本地的leveldb中,插入没问题的话，会再广播一次,不过这次只会广播block的hash, 如此，通过一个对等网络,只要块合法,那么就会被全网采纳,其中的verifyHeader,insertChain方法都是在/handler.go/NewProtocolManager中定义传过来的,所有启动的逻辑都是handler.go/Start方法中. fetch.go的start方法在syncer方法中用一个单独的协程触发的

/handler.go/handleMsg --> go pm.synchronise(p) --> pm.downloader.Synchronise(peer.id, pHead, pTd, mode) --> d.synchronise(id, head, td, mode) --> d.syncWithPeer(p, hash, td),让我们看下核心方法

  1. func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.Int) (err error) {
  2.     d.mux.Post(StartEvent{})
  3.     defer func() {
  4.         // reset on error
  5.         if err != nil {
  6.             d.mux.Post(FailedEvent{err})
  7.         } else {
  8.             d.mux.Post(DoneEvent{})
  9.         }
  10.     }()
  11.     if p.version < 62 {
  12.         return errTooOld
  13.     }
  15.     log.Debug("Synchronising with the network", "peer", p.id, "eth", p.version, "head", hash, "td", td, "mode", d.mode)
  16.     defer func(start time.Time) {
  17.         log.Debug("Synchronisation terminated", "elapsed", time.Since(start))
  18.     }(time.Now())
  20.     // Look up the sync boundaries: the common ancestor and the target block
  21.     latest, err := d.fetchHeight(p)
  22.     if err != nil {
  23.         return err
  24.     }
  25.     height := latest.Number.Uint64()
  27.     origin, err := d.findAncestor(p, height)
  28.     if err != nil {
  29.         return err
  30.     }
  31.     d.syncStatsLock.Lock()
  32.     if d.syncStatsChainHeight <= origin || d.syncStatsChainOrigin > origin {
  33.         d.syncStatsChainOrigin = origin
  34.     }
  35.     d.syncStatsChainHeight = height
  36.     d.syncStatsLock.Unlock()
  38.     // Initiate the sync using a concurrent header and content retrieval algorithm
  39.     pivot := uint64(0)
  40.     switch d.mode {
  41.     case LightSync:
  42.         pivot = height
  43.     case FastSync:
  44.         // Calculate the new fast/slow sync pivot point
  45.         if d.fsPivotLock == nil {
  46.             pivotOffset, err := rand.Int(rand.Reader, big.NewInt(int64(fsPivotInterval)))
  47.             if err != nil {
  48.                 panic(fmt.Sprintf("Failed to access crypto random source: %v", err))
  49.             }
  50.             if height > uint64(fsMinFullBlocks)+pivotOffset.Uint64() {
  51.                 pivot = height - uint64(fsMinFullBlocks) - pivotOffset.Uint64()
  52.             }
  53.         } else {
  54.             // Pivot point locked in, use this and do not pick a new one!
  55.             pivot = d.fsPivotLock.Number.Uint64()
  56.         }
  57.         // If the point is below the origin, move origin back to ensure state download
  58.         if pivot < origin {
  59.             if pivot > 0 {
  60.                 origin = pivot - 1
  61.             } else {
  62.                 origin = 0
  63.             }
  64.         }
  65.         log.Debug("Fast syncing until pivot block", "pivot", pivot)
  66.     }
  67.     d.queue.Prepare(origin+1, d.mode, pivot, latest)
  68.     if d.syncInitHook != nil {
  69.         d.syncInitHook(origin, height)
  70.     }
  72.     fetchers := []func() error{
  73.         func() error { return d.fetchHeaders(p, origin+1) }, // Headers are always retrieved
  74.         func() error { return d.fetchBodies(origin + 1) },   // Bodies are retrieved during normal and fast sync
  75.         func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during fast sync
  76.         func() error { return d.processHeaders(origin+1, td) },
  77.     }
  78.     if d.mode == FastSync {
  79.         fetchers = append(fetchers, func() error { return d.processFastSyncContent(latest) })
  80.     } else if d.mode == FullSync {
  81.         fetchers = append(fetchers, d.processFullSyncContent)
  82.     }
  83.     err = d.spawnSync(fetchers)
  84.     if err != nil && d.mode == FastSync && d.fsPivotLock != nil {
  85.         // If sync failed in the critical section, bump the fail counter.
  86.         atomic.AddUint32(&d.fsPivotFails, 1)
  87.     }
  88.     return err
  89. }

由于上述整个调用栈是在newBlockMsg的条件中触发的,这里的StartEvent会通过通道的形式传递到miner.go/update中

  1. func (self *Miner) update() {
  2.     events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
  3. out:
  4.     for ev := range events.Chan() {
  5.         switch ev.Data.(type) {
  6.         case downloader.StartEvent:
  7.             atomic.StoreInt32(&self.canStart, 0)
  8.             if self.Mining() {
  9.                 self.Stop()
  10.                 atomic.StoreInt32(&self.shouldStart, 1)
  11.                 log.Info("Mining aborted due to sync")
  12.             }
  13.         case downloader.DoneEvent, downloader.FailedEvent:
  14.             shouldStart := atomic.LoadInt32(&self.shouldStart) == 1
  16.             atomic.StoreInt32(&self.canStart, 1)
  17.             atomic.StoreInt32(&self.shouldStart, 0)
  18.             if shouldStart {
  19.                 self.Start(self.coinbase)
  20.             }
  21.             // unsubscribe. we're only interested in this event once
  22.             events.Unsubscribe()
  23.             // stop immediately and ignore all further pending events
  24.             break out
  25.         }
  26.     }
  27. }

可以看到接收到这个StartEvent就会通知所有的代理，调用stop停止当前相同块的挖矿,remote_Agent中的stop方法

最后再看一下新块如何广播给其他节点的,处理的方法在/eth/handle.go/BroadcastBlock

  1. func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
  2.     hash := block.Hash()
  3.     peers := pm.peers.PeersWithoutBlock(hash)
  5.     // If propagation is requested, send to a subset of the peer
  6.     if propagate {
  7.         // Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
  8.         var td *big.Int
  9.         if parent := pm.blockchain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {
  10.             td = new(big.Int).Add(block.Difficulty(), pm.blockchain.GetTd(block.ParentHash(), block.NumberU64()-1))
  11.         } else {
  12.             log.Error("Propagating dangling block", "number", block.Number(), "hash", hash)
  13.             return
  14.         }
  15.         // Send the block to a subset of our peers
  16.         transfer := peers[:int(math.Sqrt(float64(len(peers))))]
  17.         for _, peer := range transfer {
  18.             peer.SendNewBlock(block, td)
  19.         }
  20.         log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
  21.         return
  22.     }
  23.     // Otherwise if the block is indeed in out own chain, announce it
  24.     if pm.blockchain.HasBlock(hash, block.NumberU64()) {
  25.         for _, peer := range peers {
  26.             peer.SendNewBlockHashes([]common.Hash{hash}, []uint64{block.NumberU64()})
  27.         }
  28.         log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
  29.     }
  30. }

可以看到该方法中循环每个连接的peer节点,调用peer.SendNewBlock发送产块消息过去

  1. func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error {
  2.     p.knownBlocks.Add(block.Hash())
  3.     return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td})
  4. }
  1. func Send(w MsgWriter, msgcode uint64, data interface{}) error {
  2.     size, r, err := rlp.EncodeToReader(data)
  3.     if err != nil {
  4.         return err
  5.     }
  6.     return w.WriteMsg(Msg{Code: msgcode, Size: uint32(size), Payload: r})
  7. }

可以看到通过writeMsg写入该节点里,该方法的实现是rw *netWrapper) WriteMsg(msg Msg)

  1. func (rw *netWrapper) WriteMsg(msg Msg) error {
  2.     rw.wmu.Lock()
  3.     defer rw.wmu.Unlock()
  4.     rw.conn.SetWriteDeadline(time.Now().Add(rw.wtimeout))
  5.     return rw.wrapped.WriteMsg(msg)
  6. }

该方法设置了一个超时时间,底层调用了net.go的Write(b []byte) (n int, err error),通过网络写给对应的节点了,然后接收端的方法为ReadMsg

  1. func (pm *ProtocolManager) handleMsg(p *peer) error {
  2.     // Read the next message from the remote peer, and ensure it's fully consumed
  3.     msg, err := p.rw.ReadMsg()
  4.     if err != nil {
  5.         return err
  6.     }
  7.     if msg.Size > ProtocolMaxMsgSize {
  8.         return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
  9.     }
  10.     defer msg.Discard()

可以看到在这边读取网络写入来的消息,然后根据不同的msgCode作不同的处理,由于handleMsg是在一个死循环中调用的,所以就能一直接收到节点广播过来的消息

  1. //eth/handler.go
  2. func (pm *ProtocolManager) handle(p *peer) error {
  3.     td, head, genesis := pm.blockchain.Status()
  4.     p.Handshake(pm.networkId, td, head, genesis)
  6.     if rw, ok := p.rw.(*meteredMsgReadWriter); ok {
  7.         rm.Init(p.version)
  8.     }
  10.     pm.peers.Register(p)
  11.     defer pm.removePeer(p.id)
  13.     pm.downloader.RegisterPeer(p.id, p.version, p)
  15.     pm.syncTransactions(p)
  16.     ...
  17.     for {
  18.         if err := pm.handleMsg(p); err != nil {
  19.             return err
  20.         }
  21.     }
  22. }

handle()函数针对一个新peer做了如下几件事： 握手，与对方peer沟通己方的区块链状态 初始化一个读写通道，用以跟对方peer相互数据传输。 注册对方peer，存入己方peer列表；只有handle()函数退出时，才会将这个peer移除出列表。 Downloader成员注册这个新peer；Downloader会自己维护一个相邻peer列表。 调用syncTransactions()，用当前txpool中新累计的tx对象组装成一个txsync{}对象，推送到内部通道txsyncCh。还记得Start()启动的四个函数么? 其中第四项txsyncLoop()中用以等待txsync{}数据的通道txsyncCh，正是在这里被推入txsync{}的。 在无限循环中启动handleMsg()，当对方peer发出任何msg时，handleMsg()可以捕捉相应类型的消息并在己方进行处理。

参考资料