plugeth/eth/handler.go
Marius van der Wijden 49c2816d54
eth: improve log message (#22146)
* eth: fixed typos

* eth: fixed log message
2021-01-11 12:53:13 +01:00

497 lines
17 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/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"
"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
)
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() (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
Whitelist map[uint64]common.Hash // Hard coded whitelist for sync challenged
}
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
whitelist map[uint64]common.Hash
// channels for fetcher, syncer, txsyncLoop
txsyncCh chan *txsync
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(),
whitelist: config.Whitelist,
txsyncCh: make(chan *txsync),
quitSync: make(chan struct{}),
}
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.
if atomic.LoadUint32(&h.fastSync) == 1 {
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.ethPeer(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
func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
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
}
// Ignore maxPeers if this is a trusted peer
if h.peers.Len() >= h.maxPeers && !peer.Peer.Info().Network.Trusted {
return p2p.DiscTooManyPeers
}
peer.Log().Debug("Ethereum peer connected", "name", peer.Name())
// Register the peer locally
if err := h.peers.registerEthPeer(peer); err != nil {
peer.Log().Error("Ethereum peer registration failed", "err", err)
return err
}
defer h.removePeer(peer.ID())
p := h.peers.ethPeer(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 {
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 whitelist block hashes, request them
for number := range h.whitelist {
if err := peer.RequestHeadersByNumber(number, 1, 0, false); err != nil {
return err
}
}
// Handle incoming messages until the connection is torn down
return handler(peer)
}
// runSnapPeer
func (h *handler) runSnapPeer(peer *snap.Peer, handler snap.Handler) error {
h.peerWG.Add(1)
defer h.peerWG.Done()
// Register the peer locally
if err := h.peers.registerSnapPeer(peer); err != nil {
peer.Log().Error("Snapshot peer registration failed", "err", err)
return err
}
defer h.removePeer(peer.ID())
if err := h.downloader.SnapSyncer.Register(peer); err != nil {
return err
}
// Handle incoming messages until the connection is torn down
return handler(peer)
}
func (h *handler) removePeer(id string) {
// Remove the eth peer if it exists
eth := h.peers.ethPeer(id)
if eth != nil {
log.Debug("Removing Ethereum peer", "peer", id)
h.downloader.UnregisterPeer(id)
h.txFetcher.Drop(id)
if err := h.peers.unregisterEthPeer(id); err != nil {
log.Error("Peer removal failed", "peer", id, "err", err)
}
}
// Remove the snap peer if it exists
snap := h.peers.snapPeer(id)
if snap != nil {
log.Debug("Removing Snapshot peer", "peer", id)
h.downloader.SnapSyncer.Unregister(id)
if err := h.peers.unregisterSnapPeer(id); err != nil {
log.Error("Peer removal failed", "peer", id, "err", err)
}
}
// Hard disconnect at the networking layer
if eth != nil {
eth.Peer.Disconnect(p2p.DiscUselessPeer)
}
if snap != nil {
snap.Peer.Disconnect(p2p.DiscUselessPeer)
}
}
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(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
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) {
hash := block.Hash()
peers := h.peers.ethPeersWithoutBlock(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 all peers which are not known to
// already have the given transaction.
func (h *handler) BroadcastTransactions(txs types.Transactions, propagate bool) {
var (
txset = make(map[*ethPeer][]common.Hash)
annos = make(map[*ethPeer][]common.Hash)
)
// Broadcast transactions to a batch of peers not knowing about it
if propagate {
for _, tx := range txs {
peers := h.peers.ethPeersWithoutTransaction(tx.Hash())
// Send the block to a subset of our peers
transfer := peers[:int(math.Sqrt(float64(len(peers))))]
for _, peer := range transfer {
txset[peer] = append(txset[peer], tx.Hash())
}
log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(transfer))
}
for peer, hashes := range txset {
peer.AsyncSendTransactions(hashes)
}
return
}
// Otherwise only broadcast the announcement to peers
for _, tx := range txs {
peers := h.peers.ethPeersWithoutTransaction(tx.Hash())
for _, peer := range peers {
annos[peer] = append(annos[peer], tx.Hash())
}
}
for peer, hashes := range annos {
if peer.Version() >= eth.ETH65 {
peer.AsyncSendPooledTransactionHashes(hashes)
} else {
peer.AsyncSendTransactions(hashes)
}
}
}
// 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, true) // First propagate transactions to peers
h.BroadcastTransactions(event.Txs, false) // Only then announce to the rest
case <-h.txsSub.Err():
return
}
}
}