123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855 |
- // Copyright 2015 The go-ethereum Authors
- // This file is part of the go-ethereum library.
- //
- // The go-ethereum library is free software: you can redistribute it and/or modify
- // it under the terms of the GNU Lesser General Public License as published by
- // the Free Software Foundation, either version 3 of the License, or
- // (at your option) any later version.
- //
- // The go-ethereum library is distributed in the hope that it will be useful,
- // but WITHOUT ANY WARRANTY; without even the implied warranty of
- // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- // GNU Lesser General Public License for more details.
- //
- // You should have received a copy of the GNU Lesser General Public License
- // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
- package eth
- import (
- "errors"
- "fmt"
- "math"
- "math/big"
- "sync"
- "sync/atomic"
- "time"
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/consensus"
- "github.com/ethereum/go-ethereum/consensus/clique"
- "github.com/ethereum/go-ethereum/consensus/ethash"
- "github.com/ethereum/go-ethereum/consensus/istanbul"
- "github.com/ethereum/go-ethereum/core"
- "github.com/ethereum/go-ethereum/core/forkid"
- "github.com/ethereum/go-ethereum/core/types"
- "github.com/ethereum/go-ethereum/crypto"
- "github.com/ethereum/go-ethereum/eth/downloader"
- "github.com/ethereum/go-ethereum/eth/fetcher"
- "github.com/ethereum/go-ethereum/eth/protocols/eth"
- "github.com/ethereum/go-ethereum/eth/protocols/snap"
- "github.com/ethereum/go-ethereum/ethdb"
- "github.com/ethereum/go-ethereum/event"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/p2p"
- "github.com/ethereum/go-ethereum/p2p/enode"
- "github.com/ethereum/go-ethereum/p2p/enr"
- "github.com/ethereum/go-ethereum/params"
- "github.com/ethereum/go-ethereum/qlight"
- "github.com/ethereum/go-ethereum/trie"
- )
- const (
- // txChanSize is the size of channel listening to NewTxsEvent.
- // The number is referenced from the size of tx pool.
- txChanSize = 4096
- // Quorum
- protocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message
- )
- var (
- syncChallengeTimeout = 15 * time.Second // Time allowance for a node to reply to the sync progress challenge
- // Quorum
- errMsgTooLarge = errors.New("message too long")
- )
- // txPool defines the methods needed from a transaction pool implementation to
- // support all the operations needed by the Ethereum chain protocols.
- type txPool interface {
- // Has returns an indicator whether txpool has a transaction
- // cached with the given hash.
- Has(hash common.Hash) bool
- // Get retrieves the transaction from local txpool with given
- // tx hash.
- Get(hash common.Hash) *types.Transaction
- // AddRemotes should add the given transactions to the pool.
- AddRemotes([]*types.Transaction) []error
- // Pending should return pending transactions.
- // The slice should be modifiable by the caller.
- Pending() (map[common.Address]types.Transactions, error)
- // SubscribeNewTxsEvent should return an event subscription of
- // NewTxsEvent and send events to the given channel.
- SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
- }
- // handlerConfig is the collection of initialization parameters to create a full
- // node network handler.
- type handlerConfig struct {
- Database ethdb.Database // Database for direct sync insertions
- Chain *core.BlockChain // Blockchain to serve data from
- TxPool txPool // Transaction pool to propagate from
- Network uint64 // Network identifier to adfvertise
- Sync downloader.SyncMode // Whether to fast or full sync
- BloomCache uint64 // Megabytes to alloc for fast sync bloom
- EventMux *event.TypeMux // Legacy event mux, deprecate for `feed`
- Checkpoint *params.TrustedCheckpoint // Hard coded checkpoint for sync challenges
- // Quorum
- AuthorizationList map[uint64]common.Hash // Hard coded authorization list for sync challenged
- Engine consensus.Engine
- RaftMode bool
- // Quorum QLight
- // client
- psi string
- privateClientCache qlight.PrivateClientCache
- tokenHolder *qlight.TokenHolder
- // server
- authProvider qlight.AuthProvider
- privateBlockDataResolver qlight.PrivateBlockDataResolver
- }
- type handler struct {
- networkID uint64
- forkFilter forkid.Filter // Fork ID filter, constant across the lifetime of the node
- fastSync uint32 // Flag whether fast sync is enabled (gets disabled if we already have blocks)
- snapSync uint32 // Flag whether fast sync should operate on top of the snap protocol
- acceptTxs uint32 // Flag whether we're considered synchronised (enables transaction processing)
- checkpointNumber uint64 // Block number for the sync progress validator to cross reference
- checkpointHash common.Hash // Block hash for the sync progress validator to cross reference
- database ethdb.Database
- txpool txPool
- chain *core.BlockChain
- maxPeers int
- downloader *downloader.Downloader
- stateBloom *trie.SyncBloom
- blockFetcher *fetcher.BlockFetcher
- txFetcher *fetcher.TxFetcher
- peers *peerSet
- eventMux *event.TypeMux
- txsCh chan core.NewTxsEvent
- txsSub event.Subscription
- minedBlockSub *event.TypeMuxSubscription
- authorizationList map[uint64]common.Hash
- // channels for fetcher, syncer, txsyncLoop
- txsyncCh chan *txsync
- quitSync chan struct{}
- chainSync *chainSyncer
- wg sync.WaitGroup
- peerWG sync.WaitGroup
- // Quorum
- raftMode bool
- engine consensus.Engine
- tokenHolder *qlight.TokenHolder
- // Test fields or hooks
- broadcastTxAnnouncesOnly bool // Testing field, disable transaction propagation
- // Quorum QLight
- // client
- psi string
- privateClientCache qlight.PrivateClientCache
- // server
- authProvider qlight.AuthProvider
- privateBlockDataResolver qlight.PrivateBlockDataResolver
- }
- // newHandler returns a handler for all Ethereum chain management protocol.
- func newHandler(config *handlerConfig) (*handler, error) {
- // Create the protocol manager with the base fields
- if config.EventMux == nil {
- config.EventMux = new(event.TypeMux) // Nicety initialization for tests
- }
- h := &handler{
- networkID: config.Network,
- forkFilter: forkid.NewFilter(config.Chain),
- eventMux: config.EventMux,
- database: config.Database,
- txpool: config.TxPool,
- chain: config.Chain,
- peers: newPeerSet(),
- txsyncCh: make(chan *txsync),
- quitSync: make(chan struct{}),
- // Quorum
- authorizationList: config.AuthorizationList,
- raftMode: config.RaftMode,
- engine: config.Engine,
- tokenHolder: config.tokenHolder,
- }
- // Quorum
- if handler, ok := h.engine.(consensus.Handler); ok {
- handler.SetBroadcaster(h)
- }
- // /Quorum
- if config.Sync == downloader.FullSync {
- // 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 scenarios where this can happen is
- // * if the user manually (or via a bad block) rolled back a fast sync node
- // below the sync point.
- // * the last fast sync is not finished while user specifies a full sync this
- // time. But we don't have any recent state for full sync.
- // In these cases however it's safe to reenable fast sync.
- fullBlock, fastBlock := h.chain.CurrentBlock(), h.chain.CurrentFastBlock()
- if fullBlock.NumberU64() == 0 && fastBlock.NumberU64() > 0 {
- h.fastSync = uint32(1)
- log.Warn("Switch sync mode from full sync to fast sync")
- }
- } else {
- if h.chain.CurrentBlock().NumberU64() > 0 {
- // Print warning log if database is not empty to run fast sync.
- log.Warn("Switch sync mode from fast sync to full sync")
- } else {
- // If fast sync was requested and our database is empty, grant it
- h.fastSync = uint32(1)
- if config.Sync == downloader.SnapSync {
- h.snapSync = uint32(1)
- }
- }
- }
- // If we have trusted checkpoints, enforce them on the chain
- if config.Checkpoint != nil {
- h.checkpointNumber = (config.Checkpoint.SectionIndex+1)*params.CHTFrequency - 1
- h.checkpointHash = config.Checkpoint.SectionHead
- }
- // Construct the downloader (long sync) and its backing state bloom if fast
- // sync is requested. The downloader is responsible for deallocating the state
- // bloom when it's done.
- // Note: we don't enable it if snap-sync is performed, since it's very heavy
- // and the heal-portion of the snap sync is much lighter than fast. What we particularly
- // want to avoid, is a 90%-finished (but restarted) snap-sync to begin
- // indexing the entire trie
- if atomic.LoadUint32(&h.fastSync) == 1 && atomic.LoadUint32(&h.snapSync) == 0 {
- h.stateBloom = trie.NewSyncBloom(config.BloomCache, config.Database)
- }
- h.downloader = downloader.New(h.checkpointNumber, config.Database, h.stateBloom, h.eventMux, h.chain, nil, h.removePeer)
- // Construct the fetcher (short sync)
- validator := func(header *types.Header) error {
- return h.chain.Engine().VerifyHeader(h.chain, header, true)
- }
- heighter := func() uint64 {
- return h.chain.CurrentBlock().NumberU64()
- }
- inserter := func(blocks types.Blocks) (int, error) {
- // If sync hasn't reached the checkpoint yet, deny importing weird blocks.
- //
- // Ideally we would also compare the head block's timestamp and similarly reject
- // the propagated block if the head is too old. Unfortunately there is a corner
- // case when starting new networks, where the genesis might be ancient (0 unix)
- // which would prevent full nodes from accepting it.
- if h.chain.CurrentBlock().NumberU64() < h.checkpointNumber {
- log.Warn("Unsynced yet, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
- return 0, nil
- }
- // If fast sync is running, deny importing weird blocks. This is a problematic
- // clause when starting up a new network, because fast-syncing miners might not
- // accept each others' blocks until a restart. Unfortunately we haven't figured
- // out a way yet where nodes can decide unilaterally whether the network is new
- // or not. This should be fixed if we figure out a solution.
- if atomic.LoadUint32(&h.fastSync) == 1 {
- log.Warn("Fast syncing, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
- return 0, nil
- }
- n, err := h.chain.InsertChain(blocks)
- if err == nil {
- atomic.StoreUint32(&h.acceptTxs, 1) // Mark initial sync done on any fetcher import
- }
- return n, err
- }
- h.blockFetcher = fetcher.NewBlockFetcher(false, nil, h.chain.GetBlockByHash, validator, h.BroadcastBlock, heighter, nil, inserter, h.removePeer)
- fetchTx := func(peer string, hashes []common.Hash) error {
- p := h.peers.peer(peer)
- if p == nil {
- return errors.New("unknown peer")
- }
- return p.RequestTxs(hashes)
- }
- h.txFetcher = fetcher.NewTxFetcher(h.txpool.Has, h.txpool.AddRemotes, fetchTx)
- h.chainSync = newChainSyncer(h)
- return h, nil
- }
- // runEthPeer registers an eth peer into the joint eth/snap peerset, adds it to
- // various subsistems and starts handling messages.
- func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
- // If the peer has a `snap` extension, wait for it to connect so we can have
- // a uniform initialization/teardown mechanism
- snap, err := h.peers.waitSnapExtension(peer)
- if err != nil {
- peer.Log().Error("Snapshot extension barrier failed", "err", err)
- return err
- }
- // TODO(karalabe): Not sure why this is needed
- if !h.chainSync.handlePeerEvent(peer) {
- return p2p.DiscQuitting
- }
- h.peerWG.Add(1)
- defer h.peerWG.Done()
- // Execute the Ethereum handshake
- var (
- genesis = h.chain.Genesis()
- head = h.chain.CurrentHeader()
- hash = head.Hash()
- number = head.Number.Uint64()
- td = h.chain.GetTd(hash, number)
- )
- forkID := forkid.NewID(h.chain.Config(), h.chain.Genesis().Hash(), h.chain.CurrentHeader().Number.Uint64())
- if err := peer.Handshake(h.networkID, td, hash, genesis.Hash(), forkID, h.forkFilter); err != nil {
- peer.Log().Debug("Ethereum handshake failed", "err", err)
- // Quorum
- // When the Handshake() returns an error, the Run method corresponding to `eth` protocol returns with the error, causing the peer to drop, signal subprotocol as well to exit the `Run` method
- peer.EthPeerDisconnected <- struct{}{}
- // End Quorum
- return err
- }
- reject := false // reserved peer slots
- if atomic.LoadUint32(&h.snapSync) == 1 {
- if snap == nil {
- // If we are running snap-sync, we want to reserve roughly half the peer
- // slots for peers supporting the snap protocol.
- // The logic here is; we only allow up to 5 more non-snap peers than snap-peers.
- if all, snp := h.peers.len(), h.peers.snapLen(); all-snp > snp+5 {
- reject = true
- }
- }
- }
- // Ignore maxPeers if this is a trusted peer
- if !peer.Peer.Info().Network.Trusted {
- if reject || h.peers.len() >= h.maxPeers {
- return p2p.DiscTooManyPeers
- }
- }
- peer.Log().Debug("Ethereum peer connected", "name", peer.Name())
- // Register the peer locally
- if err := h.peers.registerPeer(peer, snap); err != nil {
- peer.Log().Error("Ethereum peer registration failed", "err", err)
- // Quorum
- // When the Register() returns an error, the Run method corresponding to `eth` protocol returns with the error, causing the peer to drop, signal subprotocol as well to exit the `Run` method
- peer.EthPeerDisconnected <- struct{}{}
- // End Quorum
- return err
- }
- defer h.unregisterPeer(peer.ID()) // Quorum: changed by https://github.com/bnb-chain/bsc/pull/856
- p := h.peers.peer(peer.ID())
- if p == nil {
- return errors.New("peer dropped during handling")
- }
- // Register the peer in the downloader. If the downloader considers it banned, we disconnect
- if err := h.downloader.RegisterPeer(peer.ID(), peer.Version(), peer); err != nil {
- peer.Log().Error("Failed to register peer in eth syncer", "err", err)
- return err
- }
- if snap != nil {
- if err := h.downloader.SnapSyncer.Register(snap); err != nil {
- peer.Log().Error("Failed to register peer in snap syncer", "err", err)
- return err
- }
- }
- h.chainSync.handlePeerEvent(peer)
- // Propagate existing transactions. new transactions appearing
- // after this will be sent via broadcasts.
- h.syncTransactions(peer)
- // If we have a trusted CHT, reject all peers below that (avoid fast sync eclipse)
- if h.checkpointHash != (common.Hash{}) {
- // Request the peer's checkpoint header for chain height/weight validation
- if err := peer.RequestHeadersByNumber(h.checkpointNumber, 1, 0, false); err != nil {
- return err
- }
- // Start a timer to disconnect if the peer doesn't reply in time
- p.syncDrop = time.AfterFunc(syncChallengeTimeout, func() {
- peer.Log().Warn("Checkpoint challenge timed out, dropping", "addr", peer.RemoteAddr(), "type", peer.Name())
- h.removePeer(peer.ID())
- })
- // Make sure it's cleaned up if the peer dies off
- defer func() {
- if p.syncDrop != nil {
- p.syncDrop.Stop()
- p.syncDrop = nil
- }
- }()
- }
- // If we have any explicit authorized block hashes, request them
- for number := range h.authorizationList {
- if err := peer.RequestHeadersByNumber(number, 1, 0, false); err != nil {
- return err
- }
- }
- // Quorum notify other subprotocols that the eth peer is ready, and has been added to the peerset.
- p.EthPeerRegistered <- struct{}{}
- // Quorum
- // Handle incoming messages until the connection is torn down
- return handler(peer)
- }
- // runSnapExtension registers a `snap` peer into the joint eth/snap peerset and
- // starts handling inbound messages. As `snap` is only a satellite protocol to
- // `eth`, all subsystem registrations and lifecycle management will be done by
- // the main `eth` handler to prevent strange races.
- func (h *handler) runSnapExtension(peer *snap.Peer, handler snap.Handler) error {
- h.peerWG.Add(1)
- defer h.peerWG.Done()
- if err := h.peers.registerSnapExtension(peer); err != nil {
- peer.Log().Error("Snapshot extension registration failed", "err", err)
- return err
- }
- return handler(peer)
- }
- // removePeer requests disconnection of a peer.
- // Quorum: added by https://github.com/bnb-chain/bsc/pull/856
- func (h *handler) removePeer(id string) {
- peer := h.peers.peer(id)
- if peer != nil {
- // Hard disconnect at the networking layer. Handler will get an EOF and terminate the peer. defer unregisterPeer will do the cleanup task after then.
- peer.Peer.Disconnect(p2p.DiscUselessPeer)
- }
- }
- // unregisterPeer removes a peer from the downloader, fetchers and main peer set.
- // Quorum: changed by https://github.com/bnb-chain/bsc/pull/856
- func (h *handler) unregisterPeer(id string) {
- // Create a custom logger to avoid printing the entire id
- var logger log.Logger
- if len(id) < 16 {
- // Tests use short IDs, don't choke on them
- logger = log.New("peer", id)
- } else {
- logger = log.New("peer", id[:8])
- }
- // Abort if the peer does not exist
- peer := h.peers.peer(id)
- if peer == nil {
- logger.Error("Ethereum peer removal failed", "err", errPeerNotRegistered)
- return
- }
- // Remove the `eth` peer if it exists
- logger.Debug("Removing Ethereum peer", "snap", peer.snapExt != nil)
- // Remove the `snap` extension if it exists
- if peer.snapExt != nil {
- h.downloader.SnapSyncer.Unregister(id)
- }
- h.downloader.UnregisterPeer(id)
- h.txFetcher.Drop(id)
- if err := h.peers.unregisterPeer(id); err != nil {
- logger.Error("Ethereum peer removal failed", "err", err)
- }
- // Hard disconnect at the networking layer
- // peer.Peer.Disconnect(p2p.DiscUselessPeer) // Quorum: removed by https://github.com/bnb-chain/bsc/pull/856
- }
- func (h *handler) Start(maxPeers int) {
- h.maxPeers = maxPeers
- // broadcast transactions
- h.wg.Add(1)
- h.txsCh = make(chan core.NewTxsEvent, txChanSize)
- h.txsSub = h.txpool.SubscribeNewTxsEvent(h.txsCh)
- go h.txBroadcastLoop()
- // Quorum
- if !h.raftMode {
- // broadcast mined blocks
- h.wg.Add(1)
- h.minedBlockSub = h.eventMux.Subscribe(core.NewMinedBlockEvent{})
- go h.minedBroadcastLoop()
- } else {
- // We set this immediately in raft mode to make sure the miner never drops
- // incoming txes. Raft mode doesn't use the fetcher or downloader, and so
- // this would never be set otherwise.
- atomic.StoreUint32(&h.acceptTxs, 1)
- }
- // End Quorum
- // start sync handlers
- h.wg.Add(2)
- go h.chainSync.loop()
- go h.txsyncLoop64() // TODO(karalabe): Legacy initial tx echange, drop with eth/64.
- }
- func (h *handler) Stop() {
- h.txsSub.Unsubscribe() // quits txBroadcastLoop
- // quorum - ensure raft stops cleanly
- if h.minedBlockSub != nil {
- h.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
- }
- // Quit chainSync and txsync64.
- // After this is done, no new peers will be accepted.
- close(h.quitSync)
- h.wg.Wait()
- // Disconnect existing sessions.
- // This also closes the gate for any new registrations on the peer set.
- // sessions which are already established but not added to h.peers yet
- // will exit when they try to register.
- h.peers.close()
- h.peerWG.Wait()
- log.Info("Ethereum protocol stopped")
- }
- // Quorum
- func (h *handler) Enqueue(id string, block *types.Block) {
- h.blockFetcher.Enqueue(id, block)
- }
- // BroadcastBlock will either propagate a block to a subset of its peers, or
- // will only announce its availability (depending what's requested).
- func (h *handler) BroadcastBlock(block *types.Block, propagate bool) {
- hash := block.Hash()
- peers := h.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 := h.chain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {
- td = new(big.Int).Add(block.Difficulty(), h.chain.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.AsyncSendNewBlock(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 h.chain.HasBlock(hash, block.NumberU64()) {
- for _, peer := range peers {
- peer.AsyncSendNewBlockHash(block)
- }
- log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
- }
- }
- // BroadcastTransactions will propagate a batch of transactions
- // - To a square root of all peers
- // - And, separately, as announcements to all peers which are not known to
- // already have the given transaction.
- func (h *handler) BroadcastTransactions(txs types.Transactions) {
- var (
- annoCount int // Count of announcements made
- annoPeers int
- directCount int // Count of the txs sent directly to peers
- directPeers int // Count of the peers that were sent transactions directly
- txset = make(map[*ethPeer][]common.Hash) // Set peer->hash to transfer directly
- annos = make(map[*ethPeer][]common.Hash) // Set peer->hash to announce
- )
- // Broadcast transactions to a batch of peers not knowing about it
- // NOTE: Raft-based consensus currently assumes that geth broadcasts
- // transactions to all peers in the network. A previous comment here
- // indicated that this logic might change in the future to only send to a
- // subset of peers. If this change occurs upstream, a merge conflict should
- // arise here, and we should add logic to send to *all* peers in raft mode.
- for _, tx := range txs {
- peers := h.peers.peersWithoutTransaction(tx.Hash())
- // Send the tx unconditionally to a subset of our peers
- // Quorum changes for broadcasting to all peers not only Sqrt
- //numDirect := int(math.Sqrt(float64(len(peers))))
- for _, peer := range peers {
- txset[peer] = append(txset[peer], tx.Hash())
- }
- // For the remaining peers, send announcement only
- //for _, peer := range peers[numDirect:] {
- // annos[peer] = append(annos[peer], tx.Hash())
- //}
- log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(peers))
- }
- for peer, hashes := range txset {
- directPeers++
- directCount += len(hashes)
- peer.AsyncSendTransactions(hashes)
- }
- for peer, hashes := range txset {
- directPeers++
- directCount += len(hashes)
- peer.AsyncSendTransactions(hashes)
- }
- for peer, hashes := range annos {
- annoPeers++
- annoCount += len(hashes)
- peer.AsyncSendPooledTransactionHashes(hashes)
- }
- log.Debug("Transaction broadcast", "txs", len(txs),
- "announce packs", annoPeers, "announced hashes", annoCount,
- "tx packs", directPeers, "broadcast txs", directCount)
- }
- // minedBroadcastLoop sends mined blocks to connected peers.
- func (h *handler) minedBroadcastLoop() {
- defer h.wg.Done()
- for obj := range h.minedBlockSub.Chan() {
- if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
- h.BroadcastBlock(ev.Block, true) // First propagate block to peers
- h.BroadcastBlock(ev.Block, false) // Only then announce to the rest
- }
- }
- }
- // txBroadcastLoop announces new transactions to connected peers.
- func (h *handler) txBroadcastLoop() {
- defer h.wg.Done()
- for {
- select {
- case event := <-h.txsCh:
- h.BroadcastTransactions(event.Txs)
- case <-h.txsSub.Err():
- return
- }
- }
- }
- // NodeInfo represents a short summary of the Ethereum sub-protocol metadata
- // known about the host peer.
- type NodeInfo struct {
- Network uint64 `json:"network"` // Ethereum network ID (1=Frontier, 2=Morden, Ropsten=3, Rinkeby=4)
- Difficulty *big.Int `json:"difficulty"` // Total difficulty of the host's blockchain
- Genesis common.Hash `json:"genesis"` // SHA3 hash of the host's genesis block
- Config *params.ChainConfig `json:"config"` // Chain configuration for the fork rules
- Head common.Hash `json:"head"` // SHA3 hash of the host's best owned block
- Consensus string `json:"consensus"` // Consensus mechanism in use
- }
- // NodeInfo retrieves some protocol metadata about the running host node.
- func (h *handler) NodeInfo() *NodeInfo {
- currentBlock := h.chain.CurrentBlock()
- // //Quorum
- //
- // changes done to fetch maxCodeSize dynamically based on the
- // maxCodeSizeConfig changes
- // /Quorum
- chainConfig := h.chain.Config()
- chainConfig.MaxCodeSize = uint64(chainConfig.GetMaxCodeSize(h.chain.CurrentBlock().Number()) / 1024)
- return &NodeInfo{
- Network: h.networkID,
- Difficulty: h.chain.GetTd(currentBlock.Hash(), currentBlock.NumberU64()),
- Genesis: h.chain.Genesis().Hash(),
- Config: chainConfig,
- Head: currentBlock.Hash(),
- Consensus: h.getConsensusAlgorithm(),
- }
- }
- // Quorum
- func (h *handler) getConsensusAlgorithm() string {
- var consensusAlgo string
- if h.raftMode { // raft does not use consensus interface
- consensusAlgo = "raft"
- } else {
- switch h.engine.(type) {
- case consensus.Istanbul:
- consensusAlgo = "istanbul"
- case *clique.Clique:
- consensusAlgo = "clique"
- case *ethash.Ethash:
- consensusAlgo = "ethash"
- default:
- consensusAlgo = "unknown"
- }
- }
- return consensusAlgo
- }
- func (h *handler) FindPeers(targets map[common.Address]bool) map[common.Address]consensus.Peer {
- m := make(map[common.Address]consensus.Peer)
- h.peers.lock.RLock()
- defer h.peers.lock.RUnlock()
- for _, p := range h.peers.peers {
- pubKey := p.Node().Pubkey()
- addr := crypto.PubkeyToAddress(*pubKey)
- if targets[addr] {
- m[addr] = p
- }
- }
- return m
- }
- // makeQuorumConsensusProtocol is similar to eth/handler.go -> makeProtocol. Called from eth/handler.go -> Protocols.
- // returns the supported subprotocol to the p2p server.
- // The Run method starts the protocol and is called by the p2p server. The quorum consensus subprotocol,
- // leverages the peer created and managed by the "eth" subprotocol.
- // The quorum consensus protocol requires that the "eth" protocol is running as well.
- func (h *handler) makeQuorumConsensusProtocol(protoName string, version uint, length uint64, backend eth.Backend, network uint64, dnsdisc enode.Iterator) p2p.Protocol {
- log.Debug("registering qouorum protocol ", "protoName", protoName, "version", version)
- return p2p.Protocol{
- Name: protoName,
- Version: version,
- Length: length,
- // no new peer created, uses the "eth" peer, so no peer management needed.
- Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
- /*
- * 1. wait for the eth protocol to create and register an eth peer.
- * 2. get the associate eth peer that was registered by he "eth" protocol.
- * 3. add the rw protocol for the quorum subprotocol to the eth peer.
- * 4. start listening for incoming messages.
- * 5. the incoming message will be sent on the quorum specific subprotocol, e.g. "istanbul/100".
- * 6. send messages to the consensus engine handler.
- * 7. messages to other to other peers listening to the subprotocol can be sent using the
- * (eth)peer.ConsensusSend() which will write to the protoRW.
- */
- // wait for the "eth" protocol to create and register the peer (added to peerset)
- select {
- case <-p.EthPeerRegistered:
- // the ethpeer should be registered, try to retrieve it and start the consensus handler.
- p2pPeerId := fmt.Sprintf("%x", p.ID().Bytes()[:8])
- ethPeer := h.peers.peer(p2pPeerId)
- if ethPeer == nil {
- p2pPeerId = fmt.Sprintf("%x", p.ID().Bytes()) //TODO:BBO
- ethPeer = h.peers.peer(p2pPeerId)
- log.Warn("full p2p peer", "id", p2pPeerId, "ethPeer", ethPeer)
- }
- if ethPeer != nil {
- p.Log().Debug("consensus subprotocol retrieved eth peer from peerset", "ethPeer.id", p2pPeerId, "ProtoName", protoName)
- // add the rw protocol for the quorum subprotocol to the eth peer.
- ethPeer.AddConsensusProtoRW(rw)
- peer := eth.NewPeer(version, p, rw, h.txpool)
- return h.handleConsensusLoop(peer, rw, nil)
- }
- p.Log().Error("consensus subprotocol retrieved nil eth peer from peerset", "ethPeer.id", p2pPeerId)
- return errEthPeerNil
- case <-p.EthPeerDisconnected:
- return errEthPeerNotRegistered
- }
- },
- NodeInfo: func() interface{} {
- return eth.NodeInfoFunc(backend.Chain(), network)
- },
- PeerInfo: func(id enode.ID) interface{} {
- return backend.PeerInfo(id)
- },
- Attributes: []enr.Entry{eth.CurrentENREntry(backend.Chain())},
- DialCandidates: dnsdisc,
- }
- }
- func (h *handler) handleConsensusLoop(p *eth.Peer, protoRW p2p.MsgReadWriter, fallThroughBackend eth.Backend) error {
- // Handle incoming messages until the connection is torn down
- for {
- if err := h.handleConsensus(p, protoRW, fallThroughBackend); err != nil {
- // allow the P2P connection to remain active during sync (when the engine is stopped)
- if errors.Is(err, istanbul.ErrStoppedEngine) && h.downloader.Synchronising() {
- // should this be warn or debug
- p.Log().Debug("Ignoring `stopped engine` consensus error due to active sync.")
- continue
- }
- p.Log().Debug("Ethereum quorum message handling failed", "err", err)
- return err
- }
- }
- }
- // This is a no-op because the eth handleMsg main loop handle ibf message as well.
- func (h *handler) handleConsensus(p *eth.Peer, protoRW p2p.MsgReadWriter, fallThroughBackend eth.Backend) error {
- // Read the next message from the remote peer (in protoRW), and ensure it's fully consumed
- msg, err := protoRW.ReadMsg()
- if err != nil {
- return err
- }
- if msg.Size > protocolMaxMsgSize {
- return fmt.Errorf("%w: %v > %v", errMsgTooLarge, msg.Size, protocolMaxMsgSize)
- }
- defer msg.Discard()
- // See if the consensus engine protocol can handle this message, e.g. istanbul will check for message is
- // istanbulMsg = 0x11, and NewBlockMsg = 0x07.
- handled, err := h.handleConsensusMsg(p, msg)
- if handled {
- p.Log().Debug("consensus message was handled by consensus engine", "msg", msg.Code,
- "quorumConsensusProtocolName", quorumConsensusProtocolName, "err", err)
- return err
- }
- if fallThroughBackend != nil {
- var handlers = eth.ETH_65_FULL_SYNC
- p.Log().Trace("Message not handled by legacy sub-protocol", "msg", msg.Code)
- if handler := handlers[msg.Code]; handler != nil {
- p.Log().Debug("Found eth handler for msg", "msg", msg.Code)
- return handler(fallThroughBackend, msg, p)
- }
- }
- return nil
- }
- func (h *handler) handleConsensusMsg(p *eth.Peer, msg p2p.Msg) (bool, error) {
- if handler, ok := h.engine.(consensus.Handler); ok {
- pubKey := p.Node().Pubkey()
- addr := crypto.PubkeyToAddress(*pubKey)
- handled, err := handler.HandleMsg(addr, msg)
- return handled, err
- }
- return false, nil
- }
- // makeLegacyProtocol is basically a copy of the eth makeProtocol, but for legacy subprotocols, e.g. "istanbul/99" "istabnul/64"
- // If support legacy subprotocols is removed, remove this and associated code as well.
- // If quorum is using a legacy protocol then the "eth" subprotocol should not be available.
- func (h *handler) makeLegacyProtocol(protoName string, version uint, length uint64, backend eth.Backend, network uint64, dnsdisc enode.Iterator) p2p.Protocol {
- log.Debug("registering a legacy protocol ", "protoName", protoName, "version", version)
- return p2p.Protocol{
- Name: protoName,
- Version: version,
- Length: length,
- Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
- peer := eth.NewPeer(version, p, rw, h.txpool)
- return h.runEthPeer(peer, func(peer *eth.Peer) error {
- // We pass through the backend so that we can 'handle' messages that we can't handle
- return h.handleConsensusLoop(peer, rw, backend)
- })
- },
- NodeInfo: func() interface{} {
- return eth.NodeInfoFunc(backend.Chain(), network)
- },
- PeerInfo: func(id enode.ID) interface{} {
- return backend.PeerInfo(id)
- },
- Attributes: []enr.Entry{eth.CurrentENREntry(backend.Chain())},
- DialCandidates: dnsdisc,
- }
- }
- // End Quorum
|