美文网首页分布式
以太坊源码深入分析(5)-- Ethereum服务和以太坊P2P

以太坊源码深入分析(5)-- Ethereum服务和以太坊P2P

作者: 老鱼游啊游 | 来源:发表于2018-04-24 19:20 被阅读179次

    在“以太坊源码深入分析(2)”一文中,我们提到Ethereum作为一个service,被Node 注册进去。Node start的时候会启动其注册的所有服务,Ethereum service也是一样。

    一,ethereum service的初始化和启动
    初始化方法

    func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
        if config.SyncMode == downloader.LightSync {
            return nil, errors.New("can't run eth.Ethereum in light sync mode, use les.LightEthereum")
        }
        if !config.SyncMode.IsValid() {
            return nil, fmt.Errorf("invalid sync mode %d", config.SyncMode)
        }
        chainDb, err := CreateDB(ctx, config, "chaindata")
        if err != nil {
            return nil, err
        }
        stopDbUpgrade := upgradeDeduplicateData(chainDb)
        chainConfig, genesisHash, genesisErr := core.SetupGenesisBlock(chainDb, config.Genesis)
        if _, ok := genesisErr.(*params.ConfigCompatError); genesisErr != nil && !ok {
            return nil, genesisErr
        }
        log.Info("Initialised chain configuration", "config", chainConfig)
    
        eth := &Ethereum{
            config:         config,
            chainDb:        chainDb,
            chainConfig:    chainConfig,
            eventMux:       ctx.EventMux,
            accountManager: ctx.AccountManager,
            engine:         CreateConsensusEngine(ctx, &config.Ethash, chainConfig, chainDb),
            shutdownChan:   make(chan bool),
            stopDbUpgrade:  stopDbUpgrade,
            networkId:      config.NetworkId,
            gasPrice:       config.GasPrice,
            etherbase:      config.Etherbase,
            bloomRequests:  make(chan chan *bloombits.Retrieval),
            bloomIndexer:   NewBloomIndexer(chainDb, params.BloomBitsBlocks),
        }
    
        log.Info("Initialising Ethereum protocol", "versions", ProtocolVersions, "network", config.NetworkId)
    
        if !config.SkipBcVersionCheck {
            bcVersion := core.GetBlockChainVersion(chainDb)
            if bcVersion != core.BlockChainVersion && bcVersion != 0 {
                return nil, fmt.Errorf("Blockchain DB version mismatch (%d / %d). Run geth upgradedb.\n", bcVersion, core.BlockChainVersion)
            }
            core.WriteBlockChainVersion(chainDb, core.BlockChainVersion)
        }
        var (
            vmConfig    = vm.Config{EnablePreimageRecording: config.EnablePreimageRecording}
            cacheConfig = &core.CacheConfig{Disabled: config.NoPruning, TrieNodeLimit: config.TrieCache, TrieTimeLimit: config.TrieTimeout}
        )
        eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, eth.chainConfig, eth.engine, vmConfig)
        if err != nil {
            return nil, err
        }
        // Rewind the chain in case of an incompatible config upgrade.
        if compat, ok := genesisErr.(*params.ConfigCompatError); ok {
            log.Warn("Rewinding chain to upgrade configuration", "err", compat)
            eth.blockchain.SetHead(compat.RewindTo)
            core.WriteChainConfig(chainDb, genesisHash, chainConfig)
        }
        eth.bloomIndexer.Start(eth.blockchain)
    
        if config.TxPool.Journal != "" {
            config.TxPool.Journal = ctx.ResolvePath(config.TxPool.Journal)
        }
        eth.txPool = core.NewTxPool(config.TxPool, eth.chainConfig, eth.blockchain)
    
        if eth.protocolManager, err = NewProtocolManager(eth.chainConfig, config.SyncMode, config.NetworkId, eth.eventMux, eth.txPool, eth.engine, eth.blockchain, chainDb); err != nil {
            return nil, err
        }
        eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine)
        eth.miner.SetExtra(makeExtraData(config.ExtraData))
    
        eth.ApiBackend = &EthApiBackend{eth, nil}
        gpoParams := config.GPO
        if gpoParams.Default == nil {
            gpoParams.Default = config.GasPrice
        }
        eth.ApiBackend.gpo = gasprice.NewOracle(eth.ApiBackend, gpoParams)
    
        return eth, nil
    }
    

    1,如果config.SyncMode 是 downloader.LightSync,走的是les/backend.go的初始化方法。
    2,chainDb, err := CreateDB(ctx, config, "chaindata")打开leveldb,leveldb是eth存储数据库。
    3,stopDbUpgrade := upgradeDeduplicateData(chainDb) 检查chainDb版本,如果需要的话,启动后台进程进行升级。
    4,chainConfig, genesisHash, genesisErr := core.SetupGenesisBlock(chainDb, config.Genesis)装载创世区块。 根据节点条件判断是从数据库里面读取,还是从默认配置文件读取,还是从自定义配置文件读取,或者是从代码里面获取默认值。并返回区块链的config和创世块的hash。
    5,装载Etherum struct的各个成员。eventMux和accountManager 是Node 启动 eth service的时候传入的。eventMux可以认为是一个全局的事件多路复用器,accountManager认为是一个全局的账户管理器。engine创建共识引擎。etherbase 配置此Etherum的主账号地址。初始化bloomRequests 通道和bloom过滤器。
    6,判断客户端版本号和数据库版本号是否一致
    7,eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, eth.chainConfig, eth.engine, vmConfig) 初始化eth的blockchain,也就是eth的区块链
    8,eth.blockchain.SetHead(compat.RewindTo) 根据创始区块设置区块头
    9,eth.bloomIndexer.Start(eth.blockchain)启动bloomIndexer
    10,eth.txPool = core.NewTxPool(config.TxPool, eth.chainConfig, eth.blockchain) 初始化eth 区块链的交易池,存储本地生产的和P2P网络同步过来的交易。
    11,eth.protocolManager, err = NewProtocolManager(eth.chainConfig, config.SyncMode, config.NetworkId, eth.eventMux, eth.txPool, eth.engine, eth.blockchain, chainDb)初始化以太坊协议管理器,用于区块链P2P通讯
    12, miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine) 初始化矿工
    13,eth.ApiBackend.gpo = gasprice.NewOracle(eth.ApiBackend, gpoParams) 创建预言最新gasprice的预言机。

    ethereum service的初始化配置了不少东西,基本上涉及到了以太坊区块链系统的所有内容,后续一一分解各个模块。

    启动方法

    func (s *Ethereum) Start(srvr *p2p.Server) error {
        // Start the bloom bits servicing goroutines
        s.startBloomHandlers()
    
        // Start the RPC service
        s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.NetVersion())
    
        // Figure out a max peers count based on the server limits
        maxPeers := srvr.MaxPeers
        if s.config.LightServ > 0 {
            if s.config.LightPeers >= srvr.MaxPeers {
                return fmt.Errorf("invalid peer config: light peer count (%d) >= total peer count (%d)", s.config.LightPeers, srvr.MaxPeers)
            }
            maxPeers -= s.config.LightPeers
        }
        // Start the networking layer and the light server if requested
        s.protocolManager.Start(maxPeers)
        if s.lesServer != nil {
            s.lesServer.Start(srvr)
        }
        return nil
    }
    

    首先启动bloom过滤器
    eth 的net 相关Api 加入RPC 服务。
    s.protocolManager.Start(maxPeers) 设置最大同步节点数,并启动eth P2P通讯。
    如果ethereum service 出问题了才会启动lesServer。

    二,ProtocolManager 以太坊P2P通讯协议管理
    首先分析一下同在eth目录下的eth/handler.go。
    ProtocolManager 的初始化方法

    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) {
        // Create the protocol manager with the base fields
        manager := &ProtocolManager{
            networkId:   networkId,
            eventMux:    mux,
            txpool:      txpool,
            blockchain:  blockchain,
            chainconfig: config,
            peers:       newPeerSet(),
            newPeerCh:   make(chan *peer),
            noMorePeers: make(chan struct{}),
            txsyncCh:    make(chan *txsync),
            quitSync:    make(chan struct{}),
        }
        // Figure out whether to allow fast sync or not
        if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 {
            log.Warn("Blockchain not empty, fast sync disabled")
            mode = downloader.FullSync
        }
        if mode == downloader.FastSync {
            manager.fastSync = uint32(1)
        }
        // Initiate a sub-protocol for every implemented version we can handle
        manager.SubProtocols = make([]p2p.Protocol, 0, len(ProtocolVersions))
        for i, version := range ProtocolVersions {
            // Skip protocol version if incompatible with the mode of operation
            if mode == downloader.FastSync && version < eth63 {
                continue
            }
            // Compatible; initialise the sub-protocol
            version := version // Closure for the run
            manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{
                Name:    ProtocolName,
                Version: version,
                Length:  ProtocolLengths[i],
                Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
                    peer := manager.newPeer(int(version), p, rw)
                    select {
                    case manager.newPeerCh <- peer:
                        manager.wg.Add(1)
                        defer manager.wg.Done()
                        return manager.handle(peer)
                    case <-manager.quitSync:
                        return p2p.DiscQuitting
                    }
                },
                NodeInfo: func() interface{} {
                    return manager.NodeInfo()
                },
                PeerInfo: func(id discover.NodeID) interface{} {
                    if p := manager.peers.Peer(fmt.Sprintf("%x", id[:8])); p != nil {
                        return p.Info()
                    }
                    return nil
                },
            })
        }
        if len(manager.SubProtocols) == 0 {
            return nil, errIncompatibleConfig
        }
        // Construct the different synchronisation mechanisms
        manager.downloader = downloader.New(mode, chaindb, manager.eventMux, blockchain, nil, manager.removePeer)
    
        validator := func(header *types.Header) error {
            return engine.VerifyHeader(blockchain, header, true)
        }
        heighter := func() uint64 {
            return blockchain.CurrentBlock().NumberU64()
        }
        inserter := func(blocks types.Blocks) (int, error) {
            // If fast sync is running, deny importing weird blocks
            if atomic.LoadUint32(&manager.fastSync) == 1 {
                log.Warn("Discarded bad propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
                return 0, nil
            }
            atomic.StoreUint32(&manager.acceptTxs, 1) // Mark initial sync done on any fetcher import
            return manager.blockchain.InsertChain(blocks)
        }
        manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)
    
        return manager, nil
    }
    

    1,peers 为以太坊临近的同步网络节点,newPeerCh、noMorePeers、txsyncCh、quitSync对应同步的通知
    2,manager.SubProtocols 创建以太坊 P2P server 的 通讯协议,通常只有一个值。manager.SubProtocols,在Node start的时候传给以太坊 P2P server并同时start P2P server。协议里面三个函数指针(Run、NodeInfo、PeerInfo)非常重要,后面会用到。
    3,manager.downloader = downloader.New(mode, chaindb, manager.eventMux, blockchain, nil, manager.removePeer)
    创建了一个下载器,从远程网络节点中获取hashes和blocks。
    4,manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)收集网络其他以太坊节点发过来的同步通知,进行验证,并做出相应的处理。初始化传入的几个参数 都是用于处理同步区块链数据的函数指针

    Ethereum service 启动的时候会同时启动ProtocolManager,ProtocolManager的start()方法:

    func (pm *ProtocolManager) Start(maxPeers int) {
        pm.maxPeers = maxPeers
    
        // broadcast transactions
        pm.txCh = make(chan core.TxPreEvent, txChanSize)
        pm.txSub = pm.txpool.SubscribeTxPreEvent(pm.txCh)
        go pm.txBroadcastLoop()
    
        // broadcast mined blocks
        pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
        go pm.minedBroadcastLoop()
    
        // start sync handlers
        go pm.syncer()
        go pm.txsyncLoop()
    }
    

    1,创建一个新交易的订阅通道,并启动交易广播的goroutine
    2,创建一个挖坑的订阅通道,并启动挖坑广播的goroutine
    注:同为订阅通道为什么pm.txSub和pm.minedBlockSub的实现不一样?深入代码会发现pm.txSub用的是event/feed通知方式,pm.minedBlockSub用的是event/TypeMuxEvent通知方式,event/TypeMuxEvent方式将要被Deprecated。
    3,pm.syncer() 启动同步goroutine,定时的和网络其他节点同步,并处理网络节点的相关通知
    4,pm.txsyncLoop() 启动交易同步goroutine,把新的交易均匀的同步给网路节点

    三,ProtocolManager主动向网络节点广播
    ProtocolManager Start()方法里面的4个goroutine都是处理ProtocolManager向以太坊网络节点进行广播的。

    1,pm.txBroadcastLoop()方法

    func (self *ProtocolManager) txBroadcastLoop() {
        for {
            select {
            case event := <-self.txCh:
                self.BroadcastTx(event.Tx.Hash(), event.Tx)
    
            // Err() channel will be closed when unsubscribing.
            case <-self.txSub.Err():
                return
            }
        }
    }
    

    core/tx_pool.go 产生新的交易的时候会send self.txCh,这时候会激活
    self.BroadcastTx(event.Tx.Hash(), event.Tx)

    func (pm *ProtocolManager) BroadcastTx(hash common.Hash, tx *types.Transaction) {
        // Broadcast transaction to a batch of peers not knowing about it
        peers := pm.peers.PeersWithoutTx(hash)
        //FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))]
        for _, peer := range peers {
            peer.SendTransactions(types.Transactions{tx})
        }
        log.Trace("Broadcast transaction", "hash", hash, "recipients", len(peers))
    }
    

    向缓存的没有这个交易hash的网络节点广播此次交易。

    2,pm.minedBroadcastLoop()方法

    // Mined broadcast loop
    func (self *ProtocolManager) minedBroadcastLoop() {
        // automatically stops if unsubscribe
        for obj := range self.minedBlockSub.Chan() {
            switch ev := obj.Data.(type) {
            case core.NewMinedBlockEvent:
                self.BroadcastBlock(ev.Block, true)  // First propagate block to peers
                self.BroadcastBlock(ev.Block, false) // Only then announce to the rest
            }
        }
    }
    

    收到miner.go 里面NewMinedBlockEvent 挖到新区块的事件通知,激活self.BroadcastBlock(ev.Block, true)

    func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
        hash := block.Hash()
        peers := pm.peers.PeersWithoutBlock(hash)
    
        // If propagation is requested, send to a subset of the peer
        if propagate {
            // Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
            var td *big.Int
            if parent := pm.blockchain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {
                td = new(big.Int).Add(block.Difficulty(), pm.blockchain.GetTd(block.ParentHash(), block.NumberU64()-1))
            } else {
                log.Error("Propagating dangling block", "number", block.Number(), "hash", hash)
                return
            }
            // Send the block to a subset of our peers
            transfer := peers[:int(math.Sqrt(float64(len(peers))))]
            for _, peer := range transfer {
                peer.SendNewBlock(block, td)
            }
            log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
            return
        }
        // Otherwise if the block is indeed in out own chain, announce it
        if pm.blockchain.HasBlock(hash, block.NumberU64()) {
            for _, peer := range peers {
                peer.SendNewBlockHashes([]common.Hash{hash}, []uint64{block.NumberU64()})
            }
            log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
        }
    }
    

    如果propagate为true 向网络节点广播整个挖到的block,为false 只广播挖到的区块的hash值和number值。广播的区块还包括这个区块打包的所有交易。

    3,pm.syncer() 方法

    func (pm *ProtocolManager) syncer() {
        // Start and ensure cleanup of sync mechanisms
        pm.fetcher.Start()
        defer pm.fetcher.Stop()
        defer pm.downloader.Terminate()
    
        // Wait for different events to fire synchronisation operations
        forceSync := time.NewTicker(forceSyncCycle)
        defer forceSync.Stop()
    
        for {
            select {
            case <-pm.newPeerCh:
                // Make sure we have peers to select from, then sync
                if pm.peers.Len() < minDesiredPeerCount {
                    break
                }
                go pm.synchronise(pm.peers.BestPeer())
    
            case <-forceSync.C:
                // Force a sync even if not enough peers are present
                go pm.synchronise(pm.peers.BestPeer())
    
            case <-pm.noMorePeers:
                return
            }
        }
    }
    

    pm.fetcher.Start()启动 fetcher,辅助同步区块数据

    当P2P server执行 ProtocolManager 的p2p.Protocol 的Run指针的时候会send pm.newPeerCh,这时候选择最优的网络节点(TD 总难度最大的)启动pm.synchronise(pm.peers.BestPeer()) goroutine。

    // synchronise tries to sync up our local block chain with a remote peer.
    func (pm *ProtocolManager) synchronise(peer *peer) {
        // Short circuit if no peers are available
        if peer == nil {
            return
        }
        // Make sure the peer's TD is higher than our own
        currentBlock := pm.blockchain.CurrentBlock()
        td := pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())
    
        pHead, pTd := peer.Head()
        if pTd.Cmp(td) <= 0 {
            return
        }
        // Otherwise try to sync with the downloader
        mode := downloader.FullSync
        if atomic.LoadUint32(&pm.fastSync) == 1 {
            // Fast sync was explicitly requested, and explicitly granted
            mode = downloader.FastSync
        } else if currentBlock.NumberU64() == 0 && pm.blockchain.CurrentFastBlock().NumberU64() > 0 {
            // The database seems empty as the current block is the genesis. Yet the fast
            // block is ahead, so fast sync was enabled for this node at a certain point.
            // The only scenario where this can happen is if the user manually (or via a
            // bad block) rolled back a fast sync node below the sync point. In this case
            // however it's safe to reenable fast sync.
            atomic.StoreUint32(&pm.fastSync, 1)
            mode = downloader.FastSync
        }
        // Run the sync cycle, and disable fast sync if we've went past the pivot block
        if err := pm.downloader.Synchronise(peer.id, pHead, pTd, mode); err != nil {
            return
        }
        if atomic.LoadUint32(&pm.fastSync) == 1 {
            log.Info("Fast sync complete, auto disabling")
            atomic.StoreUint32(&pm.fastSync, 0)
        }
        atomic.StoreUint32(&pm.acceptTxs, 1) // Mark initial sync done
        if head := pm.blockchain.CurrentBlock(); head.NumberU64() > 0 {
            // We've completed a sync cycle, notify all peers of new state. This path is
            // essential in star-topology networks where a gateway node needs to notify
            // all its out-of-date peers of the availability of a new block. This failure
            // scenario will most often crop up in private and hackathon networks with
            // degenerate connectivity, but it should be healthy for the mainnet too to
            // more reliably update peers or the local TD state.
            go pm.BroadcastBlock(head, false)
        }
    }
    

    如果最优的网络节点的TD值大于本地最新区块的TD值,调用pm.downloader.Synchronise(peer.id, pHead, pTd, mode)进行同步。同步完成后再屌用go pm.BroadcastBlock(head, false),把自己最新的区块状态广播出去。

    4,pm.txsyncLoop()方法

    func (pm *ProtocolManager) txsyncLoop() {
        var (
            pending = make(map[discover.NodeID]*txsync)
            sending = false               // whether a send is active
            pack    = new(txsync)         // the pack that is being sent
            done    = make(chan error, 1) // result of the send
        )
    
        // send starts a sending a pack of transactions from the sync.
        send := func(s *txsync) {
            // Fill pack with transactions up to the target size.
            size := common.StorageSize(0)
            pack.p = s.p
            pack.txs = pack.txs[:0]
            for i := 0; i < len(s.txs) && size < txsyncPackSize; i++ {
                pack.txs = append(pack.txs, s.txs[i])
                size += s.txs[i].Size()
            }
            // Remove the transactions that will be sent.
            s.txs = s.txs[:copy(s.txs, s.txs[len(pack.txs):])]
            if len(s.txs) == 0 {
                delete(pending, s.p.ID())
            }
            // Send the pack in the background.
            s.p.Log().Trace("Sending batch of transactions", "count", len(pack.txs), "bytes", size)
            sending = true
            go func() { done <- pack.p.SendTransactions(pack.txs) }()
        }
    
        // pick chooses the next pending sync.
        pick := func() *txsync {
            if len(pending) == 0 {
                return nil
            }
            n := rand.Intn(len(pending)) + 1
            for _, s := range pending {
                if n--; n == 0 {
                    return s
                }
            }
            return nil
        }
    
        for {
            select {
            case s := <-pm.txsyncCh:
                pending[s.p.ID()] = s
                if !sending {
                    send(s)
                }
            case err := <-done:
                sending = false
                // Stop tracking peers that cause send failures.
                if err != nil {
                    pack.p.Log().Debug("Transaction send failed", "err", err)
                    delete(pending, pack.p.ID())
                }
                // Schedule the next send.
                if s := pick(); s != nil {
                    send(s)
                }
            case <-pm.quitSync:
                return
            }
        }
    }
    

    当从网络节点同步过来最新的交易数据后,本地也会把新同步下来的交易数据广播给网络中的其他节点。

    总结一下
    这四个goroutine 基本上就在不停的做广播区块、广播交易,同步到区块、同步到交易,再广播区块、广播交易。

    相关文章

      网友评论

      • 圣西罗球僮:LZ,我想问下“func (s *Ethereum) Start(srvr *p2p.Server) error”这个函数在具体在哪里被调用的啊?
        老鱼游啊游:@圣西罗球僮 node.go 的func (n *Node) Start()里面 service.Start(running)

      本文标题:以太坊源码深入分析(5)-- Ethereum服务和以太坊P2P

      本文链接:https://www.haomeiwen.com/subject/kqjllftx.html