mirror of
https://gitlab.com/pulsechaincom/go-pulse.git
synced 2024-12-23 03:51:09 +00:00
691 lines
25 KiB
Go
691 lines
25 KiB
Go
// 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"
|
|
"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/beacon"
|
|
"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/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/params"
|
|
)
|
|
|
|
const (
|
|
// txChanSize is the size of channel listening to NewTxsEvent.
|
|
// The number is referenced from the size of tx pool.
|
|
txChanSize = 4096
|
|
)
|
|
|
|
var (
|
|
syncChallengeTimeout = 15 * time.Second // Time allowance for a node to reply to the sync progress challenge
|
|
)
|
|
|
|
// 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(enforceTips bool) map[common.Address]types.Transactions
|
|
|
|
// 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
|
|
Merger *consensus.Merger // The manager for eth1/2 transition
|
|
Network uint64 // Network identifier to adfvertise
|
|
Sync downloader.SyncMode // Whether to snap or full sync
|
|
BloomCache uint64 // Megabytes to alloc for snap sync bloom
|
|
EventMux *event.TypeMux // Legacy event mux, deprecate for `feed`
|
|
Checkpoint *params.TrustedCheckpoint // Hard coded checkpoint for sync challenges
|
|
RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
|
|
}
|
|
|
|
type handler struct {
|
|
networkID uint64
|
|
forkFilter forkid.Filter // Fork ID filter, constant across the lifetime of the node
|
|
|
|
snapSync uint32 // Flag whether snap sync is enabled (gets disabled if we already have blocks)
|
|
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
|
|
blockFetcher *fetcher.BlockFetcher
|
|
txFetcher *fetcher.TxFetcher
|
|
peers *peerSet
|
|
merger *consensus.Merger
|
|
|
|
eventMux *event.TypeMux
|
|
txsCh chan core.NewTxsEvent
|
|
txsSub event.Subscription
|
|
minedBlockSub *event.TypeMuxSubscription
|
|
|
|
requiredBlocks map[uint64]common.Hash
|
|
|
|
// channels for fetcher, syncer, txsyncLoop
|
|
quitSync chan struct{}
|
|
|
|
chainSync *chainSyncer
|
|
wg sync.WaitGroup
|
|
peerWG sync.WaitGroup
|
|
}
|
|
|
|
// 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(),
|
|
merger: config.Merger,
|
|
requiredBlocks: config.RequiredBlocks,
|
|
quitSync: make(chan struct{}),
|
|
}
|
|
if config.Sync == downloader.FullSync {
|
|
// The database seems empty as the current block is the genesis. Yet the snap
|
|
// block is ahead, so snap 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 snap sync node
|
|
// below the sync point.
|
|
// * the last snap 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 snap sync.
|
|
fullBlock, fastBlock := h.chain.CurrentBlock(), h.chain.CurrentFastBlock()
|
|
if fullBlock.NumberU64() == 0 && fastBlock.NumberU64() > 0 {
|
|
h.snapSync = uint32(1)
|
|
log.Warn("Switch sync mode from full sync to snap sync")
|
|
}
|
|
} else {
|
|
if h.chain.CurrentBlock().NumberU64() > 0 {
|
|
// Print warning log if database is not empty to run snap sync.
|
|
log.Warn("Switch sync mode from snap sync to full sync")
|
|
} else {
|
|
// If snap sync was requested and our database is empty, grant it
|
|
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
|
|
}
|
|
// If sync succeeds, pass a callback to potentially disable snap sync mode
|
|
// and enable transaction propagation.
|
|
success := func() {
|
|
// If we were running snap sync and it finished, disable doing another
|
|
// round on next sync cycle
|
|
if atomic.LoadUint32(&h.snapSync) == 1 {
|
|
log.Info("Snap sync complete, auto disabling")
|
|
atomic.StoreUint32(&h.snapSync, 0)
|
|
}
|
|
// If we've successfully finished a sync cycle and passed any required
|
|
// checkpoint, enable accepting transactions from the network
|
|
head := h.chain.CurrentBlock()
|
|
if head.NumberU64() >= h.checkpointNumber {
|
|
// Checkpoint passed, sanity check the timestamp to have a fallback mechanism
|
|
// for non-checkpointed (number = 0) private networks.
|
|
if head.Time() >= uint64(time.Now().AddDate(0, -1, 0).Unix()) {
|
|
atomic.StoreUint32(&h.acceptTxs, 1)
|
|
}
|
|
}
|
|
}
|
|
// Construct the downloader (long sync)
|
|
h.downloader = downloader.New(h.checkpointNumber, config.Database, h.eventMux, h.chain, nil, h.removePeer, success)
|
|
if ttd := h.chain.Config().TerminalTotalDifficulty; ttd != nil {
|
|
if h.chain.Config().TerminalTotalDifficultyPassed {
|
|
log.Info("Chain post-merge, sync via beacon client")
|
|
} else {
|
|
head := h.chain.CurrentBlock()
|
|
if td := h.chain.GetTd(head.Hash(), head.NumberU64()); td.Cmp(ttd) >= 0 {
|
|
log.Info("Chain post-TTD, sync via beacon client")
|
|
} else {
|
|
log.Warn("Chain pre-merge, sync via PoW (ensure beacon client is ready)")
|
|
}
|
|
}
|
|
} else if h.chain.Config().TerminalTotalDifficultyPassed {
|
|
log.Error("Chain configured post-merge, but without TTD. Are you debugging sync?")
|
|
}
|
|
// Construct the fetcher (short sync)
|
|
validator := func(header *types.Header) error {
|
|
// All the block fetcher activities should be disabled
|
|
// after the transition. Print the warning log.
|
|
if h.merger.PoSFinalized() {
|
|
log.Warn("Unexpected validation activity", "hash", header.Hash(), "number", header.Number)
|
|
return errors.New("unexpected behavior after transition")
|
|
}
|
|
// Reject all the PoS style headers in the first place. No matter
|
|
// the chain has finished the transition or not, the PoS headers
|
|
// should only come from the trusted consensus layer instead of
|
|
// p2p network.
|
|
if beacon, ok := h.chain.Engine().(*beacon.Beacon); ok {
|
|
if beacon.IsPoSHeader(header) {
|
|
return errors.New("unexpected post-merge header")
|
|
}
|
|
}
|
|
return h.chain.Engine().VerifyHeader(h.chain, header, true)
|
|
}
|
|
heighter := func() uint64 {
|
|
return h.chain.CurrentBlock().NumberU64()
|
|
}
|
|
inserter := func(blocks types.Blocks) (int, error) {
|
|
// All the block fetcher activities should be disabled
|
|
// after the transition. Print the warning log.
|
|
if h.merger.PoSFinalized() {
|
|
var ctx []interface{}
|
|
ctx = append(ctx, "blocks", len(blocks))
|
|
if len(blocks) > 0 {
|
|
ctx = append(ctx, "firsthash", blocks[0].Hash())
|
|
ctx = append(ctx, "firstnumber", blocks[0].Number())
|
|
ctx = append(ctx, "lasthash", blocks[len(blocks)-1].Hash())
|
|
ctx = append(ctx, "lastnumber", blocks[len(blocks)-1].Number())
|
|
}
|
|
log.Warn("Unexpected insertion activity", ctx...)
|
|
return 0, errors.New("unexpected behavior after transition")
|
|
}
|
|
// 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 snap sync is running, deny importing weird blocks. This is a problematic
|
|
// clause when starting up a new network, because snap-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.snapSync) == 1 {
|
|
log.Warn("Snap syncing, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
|
|
return 0, nil
|
|
}
|
|
if h.merger.TDDReached() {
|
|
// The blocks from the p2p network is regarded as untrusted
|
|
// after the transition. In theory block gossip should be disabled
|
|
// entirely whenever the transition is started. But in order to
|
|
// handle the transition boundary reorg in the consensus-layer,
|
|
// the legacy blocks are still accepted, but only for the terminal
|
|
// pow blocks. Spec: https://github.com/ethereum/EIPs/blob/master/EIPS/eip-3675.md#halt-the-importing-of-pow-blocks
|
|
for i, block := range blocks {
|
|
ptd := h.chain.GetTd(block.ParentHash(), block.NumberU64()-1)
|
|
if ptd == nil {
|
|
return 0, nil
|
|
}
|
|
td := new(big.Int).Add(ptd, block.Difficulty())
|
|
if !h.chain.Config().IsTerminalPoWBlock(ptd, td) {
|
|
log.Info("Filtered out non-termimal pow block", "number", block.NumberU64(), "hash", block.Hash())
|
|
return 0, nil
|
|
}
|
|
if err := h.chain.InsertBlockWithoutSetHead(block); err != nil {
|
|
return i, err
|
|
}
|
|
}
|
|
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 subsystems 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)
|
|
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)
|
|
return err
|
|
}
|
|
defer h.unregisterPeer(peer.ID())
|
|
|
|
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)
|
|
|
|
// Create a notification channel for pending requests if the peer goes down
|
|
dead := make(chan struct{})
|
|
defer close(dead)
|
|
|
|
// 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
|
|
resCh := make(chan *eth.Response)
|
|
|
|
req, err := peer.RequestHeadersByNumber(h.checkpointNumber, 1, 0, false, resCh)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Start a timer to disconnect if the peer doesn't reply in time
|
|
go func() {
|
|
// Ensure the request gets cancelled in case of error/drop
|
|
defer req.Close()
|
|
|
|
timeout := time.NewTimer(syncChallengeTimeout)
|
|
defer timeout.Stop()
|
|
|
|
select {
|
|
case res := <-resCh:
|
|
headers := ([]*types.Header)(*res.Res.(*eth.BlockHeadersPacket))
|
|
if len(headers) == 0 {
|
|
// If we're doing a snap sync, we must enforce the checkpoint
|
|
// block to avoid eclipse attacks. Unsynced nodes are welcome
|
|
// to connect after we're done joining the network.
|
|
if atomic.LoadUint32(&h.snapSync) == 1 {
|
|
peer.Log().Warn("Dropping unsynced node during sync", "addr", peer.RemoteAddr(), "type", peer.Name())
|
|
res.Done <- errors.New("unsynced node cannot serve sync")
|
|
return
|
|
}
|
|
res.Done <- nil
|
|
return
|
|
}
|
|
// Validate the header and either drop the peer or continue
|
|
if len(headers) > 1 {
|
|
res.Done <- errors.New("too many headers in checkpoint response")
|
|
return
|
|
}
|
|
if headers[0].Hash() != h.checkpointHash {
|
|
res.Done <- errors.New("checkpoint hash mismatch")
|
|
return
|
|
}
|
|
res.Done <- nil
|
|
|
|
case <-timeout.C:
|
|
peer.Log().Warn("Checkpoint challenge timed out, dropping", "addr", peer.RemoteAddr(), "type", peer.Name())
|
|
h.removePeer(peer.ID())
|
|
|
|
case <-dead:
|
|
// Peer handler terminated, abort all goroutines
|
|
}
|
|
}()
|
|
}
|
|
// If we have any explicit peer required block hashes, request them
|
|
for number, hash := range h.requiredBlocks {
|
|
resCh := make(chan *eth.Response)
|
|
|
|
req, err := peer.RequestHeadersByNumber(number, 1, 0, false, resCh)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
go func(number uint64, hash common.Hash, req *eth.Request) {
|
|
// Ensure the request gets cancelled in case of error/drop
|
|
defer req.Close()
|
|
|
|
timeout := time.NewTimer(syncChallengeTimeout)
|
|
defer timeout.Stop()
|
|
|
|
select {
|
|
case res := <-resCh:
|
|
headers := ([]*types.Header)(*res.Res.(*eth.BlockHeadersPacket))
|
|
if len(headers) == 0 {
|
|
// Required blocks are allowed to be missing if the remote
|
|
// node is not yet synced
|
|
res.Done <- nil
|
|
return
|
|
}
|
|
// Validate the header and either drop the peer or continue
|
|
if len(headers) > 1 {
|
|
res.Done <- errors.New("too many headers in required block response")
|
|
return
|
|
}
|
|
if headers[0].Number.Uint64() != number || headers[0].Hash() != hash {
|
|
peer.Log().Info("Required block mismatch, dropping peer", "number", number, "hash", headers[0].Hash(), "want", hash)
|
|
res.Done <- errors.New("required block mismatch")
|
|
return
|
|
}
|
|
peer.Log().Debug("Peer required block verified", "number", number, "hash", hash)
|
|
res.Done <- nil
|
|
case <-timeout.C:
|
|
peer.Log().Warn("Required block challenge timed out, dropping", "addr", peer.RemoteAddr(), "type", peer.Name())
|
|
h.removePeer(peer.ID())
|
|
}
|
|
}(number, hash, req)
|
|
}
|
|
// 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().Warn("Snapshot extension registration failed", "err", err)
|
|
return err
|
|
}
|
|
return handler(peer)
|
|
}
|
|
|
|
// removePeer requests disconnection of a peer.
|
|
func (h *handler) removePeer(id string) {
|
|
peer := h.peers.peer(id)
|
|
if peer != nil {
|
|
peer.Peer.Disconnect(p2p.DiscUselessPeer)
|
|
}
|
|
}
|
|
|
|
// unregisterPeer removes a peer from the downloader, fetchers and main peer set.
|
|
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)
|
|
}
|
|
}
|
|
|
|
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()
|
|
|
|
// broadcast mined blocks
|
|
h.wg.Add(1)
|
|
h.minedBlockSub = h.eventMux.Subscribe(core.NewMinedBlockEvent{})
|
|
go h.minedBroadcastLoop()
|
|
|
|
// start sync handlers
|
|
h.wg.Add(1)
|
|
go h.chainSync.loop()
|
|
}
|
|
|
|
func (h *handler) Stop() {
|
|
h.txsSub.Unsubscribe() // quits txBroadcastLoop
|
|
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")
|
|
}
|
|
|
|
// 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) {
|
|
// Disable the block propagation if the chain has already entered the PoS
|
|
// stage. The block propagation is delegated to the consensus layer.
|
|
if h.merger.PoSFinalized() {
|
|
return
|
|
}
|
|
// Disable the block propagation if it's the post-merge block.
|
|
if beacon, ok := h.chain.Engine().(*beacon.Beacon); ok {
|
|
if beacon.IsPoSHeader(block.Header()) {
|
|
return
|
|
}
|
|
}
|
|
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
|
|
for _, tx := range txs {
|
|
peers := h.peers.peersWithoutTransaction(tx.Hash())
|
|
// Send the tx unconditionally to a subset of our peers
|
|
numDirect := int(math.Sqrt(float64(len(peers))))
|
|
for _, peer := range peers[:numDirect] {
|
|
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())
|
|
}
|
|
}
|
|
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
|
|
}
|
|
}
|
|
}
|