mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-14 07:53:47 +00:00
eth, core: track block propagation (#1078)
* eth, core: track block announcement and import timings * eth: track total delay * eth: add more context and prefix in logs * eth/fetcher: fix invalid tracking of received at time * put logs behind log.enable-block-tracking flag * fix lint * improve logging * add comments * add comments
This commit is contained in:
parent
de45e3f54e
commit
cb23b9baa3
20 changed files with 160 additions and 85 deletions
|
|
@ -25,6 +25,7 @@ syncmode = "full"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
# maxpeers = 1
|
# maxpeers = 1
|
||||||
|
|
|
||||||
|
|
@ -263,6 +263,7 @@ type Block struct {
|
||||||
// inter-peer block relay.
|
// inter-peer block relay.
|
||||||
ReceivedAt time.Time
|
ReceivedAt time.Time
|
||||||
ReceivedFrom interface{}
|
ReceivedFrom interface{}
|
||||||
|
AnnouncedAt *time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
// "external" block encoding. used for eth protocol, etc.
|
// "external" block encoding. used for eth protocol, etc.
|
||||||
|
|
|
||||||
|
|
@ -24,10 +24,11 @@ devfakeauthor = false # Run miner without validator set authorization
|
||||||
"32000000" = "0x875500011e5eecc0c554f95d07b31cf59df4ca2505f4dbbfffa7d4e4da917c68"
|
"32000000" = "0x875500011e5eecc0c554f95d07b31cf59df4ca2505f4dbbfffa7d4e4da917c68"
|
||||||
|
|
||||||
[log]
|
[log]
|
||||||
vmodule = "" # Per-module verbosity: comma-separated list of <pattern>=<level> (e.g. eth/*=5,p2p=4)
|
vmodule = "" # Per-module verbosity: comma-separated list of <pattern>=<level> (e.g. eth/*=5,p2p=4)
|
||||||
json = false # Format logs with JSON
|
json = false # Format logs with JSON
|
||||||
backtrace = "" # Request a stack trace at a specific logging statement (e.g. "block.go:271")
|
backtrace = "" # Request a stack trace at a specific logging statement (e.g. "block.go:271")
|
||||||
debug = true # Prepends log messages with call-site location (file and line number) - {requires some effort}
|
debug = true # Prepends log messages with call-site location (file and line number)
|
||||||
|
enable-block-tracking = false # Enables additional logging of information collected while tracking block lifecycle
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 50 # Maximum number of network peers (network disabled if set to 0)
|
maxpeers = 50 # Maximum number of network peers (network disabled if set to 0)
|
||||||
|
|
|
||||||
|
|
@ -204,6 +204,8 @@ The ```bor server``` command runs the Bor client.
|
||||||
|
|
||||||
- ```log.debug```: Prepends log messages with call-site location (file and line number) (default: false)
|
- ```log.debug```: Prepends log messages with call-site location (file and line number) (default: false)
|
||||||
|
|
||||||
|
- ```log.enable-block-tracking```: Enables additional logging of information collected while tracking block lifecycle (default: false)
|
||||||
|
|
||||||
- ```log.json```: Format logs with JSON (default: false)
|
- ```log.json```: Format logs with JSON (default: false)
|
||||||
|
|
||||||
- ```vmodule```: Per-module verbosity: comma-separated list of <pattern>=<level> (e.g. eth/*=5,p2p=4)
|
- ```vmodule```: Per-module verbosity: comma-separated list of <pattern>=<level> (e.g. eth/*=5,p2p=4)
|
||||||
|
|
|
||||||
|
|
@ -272,18 +272,19 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
|
||||||
// Permit the downloader to use the trie cache allowance during fast sync
|
// Permit the downloader to use the trie cache allowance during fast sync
|
||||||
cacheLimit := cacheConfig.TrieCleanLimit + cacheConfig.TrieDirtyLimit + cacheConfig.SnapshotLimit
|
cacheLimit := cacheConfig.TrieCleanLimit + cacheConfig.TrieDirtyLimit + cacheConfig.SnapshotLimit
|
||||||
if eth.handler, err = newHandler(&handlerConfig{
|
if eth.handler, err = newHandler(&handlerConfig{
|
||||||
Database: chainDb,
|
Database: chainDb,
|
||||||
Chain: eth.blockchain,
|
Chain: eth.blockchain,
|
||||||
TxPool: eth.txPool,
|
TxPool: eth.txPool,
|
||||||
Merger: eth.merger,
|
Merger: eth.merger,
|
||||||
Network: config.NetworkId,
|
Network: config.NetworkId,
|
||||||
Sync: config.SyncMode,
|
Sync: config.SyncMode,
|
||||||
BloomCache: uint64(cacheLimit),
|
BloomCache: uint64(cacheLimit),
|
||||||
EventMux: eth.eventMux,
|
EventMux: eth.eventMux,
|
||||||
RequiredBlocks: config.RequiredBlocks,
|
RequiredBlocks: config.RequiredBlocks,
|
||||||
EthAPI: blockChainAPI,
|
EthAPI: blockChainAPI,
|
||||||
checker: checker,
|
checker: checker,
|
||||||
txArrivalWait: eth.p2pServer.TxArrivalWait,
|
txArrivalWait: eth.p2pServer.TxArrivalWait,
|
||||||
|
enableBlockTracking: eth.config.EnableBlockTracking,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -204,6 +204,9 @@ type Config struct {
|
||||||
|
|
||||||
// OverrideVerkle (TODO: remove after the fork)
|
// OverrideVerkle (TODO: remove after the fork)
|
||||||
OverrideVerkle *big.Int `toml:",omitempty"`
|
OverrideVerkle *big.Int `toml:",omitempty"`
|
||||||
|
|
||||||
|
// EnableBlockTracking allows logging of information collected while tracking block lifecycle
|
||||||
|
EnableBlockTracking bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// CreateConsensusEngine creates a consensus engine for the given chain configuration.
|
// CreateConsensusEngine creates a consensus engine for the given chain configuration.
|
||||||
|
|
|
||||||
|
|
@ -114,18 +114,20 @@ type blockAnnounce struct {
|
||||||
|
|
||||||
// headerFilterTask represents a batch of headers needing fetcher filtering.
|
// headerFilterTask represents a batch of headers needing fetcher filtering.
|
||||||
type headerFilterTask struct {
|
type headerFilterTask struct {
|
||||||
peer string // The source peer of block headers
|
peer string // The source peer of block headers
|
||||||
headers []*types.Header // Collection of headers to filter
|
headers []*types.Header // Collection of headers to filter
|
||||||
time time.Time // Arrival time of the headers
|
time time.Time // Arrival time of the headers
|
||||||
|
announcedTime time.Time // Announcement time of the availability of the block
|
||||||
}
|
}
|
||||||
|
|
||||||
// bodyFilterTask represents a batch of block bodies (transactions and uncles)
|
// bodyFilterTask represents a batch of block bodies (transactions and uncles)
|
||||||
// needing fetcher filtering.
|
// needing fetcher filtering.
|
||||||
type bodyFilterTask struct {
|
type bodyFilterTask struct {
|
||||||
peer string // The source peer of block bodies
|
peer string // The source peer of block bodies
|
||||||
transactions [][]*types.Transaction // Collection of transactions per block bodies
|
transactions [][]*types.Transaction // Collection of transactions per block bodies
|
||||||
uncles [][]*types.Header // Collection of uncles per block bodies
|
uncles [][]*types.Header // Collection of uncles per block bodies
|
||||||
time time.Time // Arrival time of the blocks' contents
|
time time.Time // Arrival time of the blocks' contents
|
||||||
|
announcedTime time.Time // Announcement time of the availability of the block
|
||||||
}
|
}
|
||||||
|
|
||||||
// blockOrHeaderInject represents a schedules import operation.
|
// blockOrHeaderInject represents a schedules import operation.
|
||||||
|
|
@ -197,34 +199,38 @@ type BlockFetcher struct {
|
||||||
fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch
|
fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch
|
||||||
completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62)
|
completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62)
|
||||||
importedHook func(*types.Header, *types.Block) // Method to call upon successful header or block import (both eth/61 and eth/62)
|
importedHook func(*types.Header, *types.Block) // Method to call upon successful header or block import (both eth/61 and eth/62)
|
||||||
|
|
||||||
|
// Logging
|
||||||
|
enableBlockTracking bool // Whether to log information collected while tracking block lifecycle
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewBlockFetcher creates a block fetcher to retrieve blocks based on hash announcements.
|
// NewBlockFetcher creates a block fetcher to retrieve blocks based on hash announcements.
|
||||||
func NewBlockFetcher(light bool, getHeader HeaderRetrievalFn, getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertHeaders headersInsertFn, insertChain chainInsertFn, dropPeer peerDropFn) *BlockFetcher {
|
func NewBlockFetcher(light bool, getHeader HeaderRetrievalFn, getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertHeaders headersInsertFn, insertChain chainInsertFn, dropPeer peerDropFn, enableBlockTracking bool) *BlockFetcher {
|
||||||
return &BlockFetcher{
|
return &BlockFetcher{
|
||||||
light: light,
|
light: light,
|
||||||
notify: make(chan *blockAnnounce),
|
notify: make(chan *blockAnnounce),
|
||||||
inject: make(chan *blockOrHeaderInject),
|
inject: make(chan *blockOrHeaderInject),
|
||||||
headerFilter: make(chan chan *headerFilterTask),
|
headerFilter: make(chan chan *headerFilterTask),
|
||||||
bodyFilter: make(chan chan *bodyFilterTask),
|
bodyFilter: make(chan chan *bodyFilterTask),
|
||||||
done: make(chan common.Hash),
|
done: make(chan common.Hash),
|
||||||
quit: make(chan struct{}),
|
quit: make(chan struct{}),
|
||||||
announces: make(map[string]int),
|
announces: make(map[string]int),
|
||||||
announced: make(map[common.Hash][]*blockAnnounce),
|
announced: make(map[common.Hash][]*blockAnnounce),
|
||||||
fetching: make(map[common.Hash]*blockAnnounce),
|
fetching: make(map[common.Hash]*blockAnnounce),
|
||||||
fetched: make(map[common.Hash][]*blockAnnounce),
|
fetched: make(map[common.Hash][]*blockAnnounce),
|
||||||
completing: make(map[common.Hash]*blockAnnounce),
|
completing: make(map[common.Hash]*blockAnnounce),
|
||||||
queue: prque.New[int64, *blockOrHeaderInject](nil),
|
queue: prque.New[int64, *blockOrHeaderInject](nil),
|
||||||
queues: make(map[string]int),
|
queues: make(map[string]int),
|
||||||
queued: make(map[common.Hash]*blockOrHeaderInject),
|
queued: make(map[common.Hash]*blockOrHeaderInject),
|
||||||
getHeader: getHeader,
|
getHeader: getHeader,
|
||||||
getBlock: getBlock,
|
getBlock: getBlock,
|
||||||
verifyHeader: verifyHeader,
|
verifyHeader: verifyHeader,
|
||||||
broadcastBlock: broadcastBlock,
|
broadcastBlock: broadcastBlock,
|
||||||
chainHeight: chainHeight,
|
chainHeight: chainHeight,
|
||||||
insertHeaders: insertHeaders,
|
insertHeaders: insertHeaders,
|
||||||
insertChain: insertChain,
|
insertChain: insertChain,
|
||||||
dropPeer: dropPeer,
|
dropPeer: dropPeer,
|
||||||
|
enableBlockTracking: enableBlockTracking,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -276,7 +282,7 @@ func (f *BlockFetcher) Enqueue(peer string, block *types.Block) error {
|
||||||
|
|
||||||
// FilterHeaders extracts all the headers that were explicitly requested by the fetcher,
|
// FilterHeaders extracts all the headers that were explicitly requested by the fetcher,
|
||||||
// returning those that should be handled differently.
|
// returning those that should be handled differently.
|
||||||
func (f *BlockFetcher) FilterHeaders(peer string, headers []*types.Header, time time.Time) []*types.Header {
|
func (f *BlockFetcher) FilterHeaders(peer string, headers []*types.Header, time time.Time, announcedAt time.Time) []*types.Header {
|
||||||
log.Trace("Filtering headers", "peer", peer, "headers", len(headers))
|
log.Trace("Filtering headers", "peer", peer, "headers", len(headers))
|
||||||
|
|
||||||
// Send the filter channel to the fetcher
|
// Send the filter channel to the fetcher
|
||||||
|
|
@ -289,7 +295,7 @@ func (f *BlockFetcher) FilterHeaders(peer string, headers []*types.Header, time
|
||||||
}
|
}
|
||||||
// Request the filtering of the header list
|
// Request the filtering of the header list
|
||||||
select {
|
select {
|
||||||
case filter <- &headerFilterTask{peer: peer, headers: headers, time: time}:
|
case filter <- &headerFilterTask{peer: peer, headers: headers, time: time, announcedTime: announcedAt}:
|
||||||
case <-f.quit:
|
case <-f.quit:
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -304,7 +310,7 @@ func (f *BlockFetcher) FilterHeaders(peer string, headers []*types.Header, time
|
||||||
|
|
||||||
// FilterBodies extracts all the block bodies that were explicitly requested by
|
// FilterBodies extracts all the block bodies that were explicitly requested by
|
||||||
// the fetcher, returning those that should be handled differently.
|
// the fetcher, returning those that should be handled differently.
|
||||||
func (f *BlockFetcher) FilterBodies(peer string, transactions [][]*types.Transaction, uncles [][]*types.Header, time time.Time) ([][]*types.Transaction, [][]*types.Header) {
|
func (f *BlockFetcher) FilterBodies(peer string, transactions [][]*types.Transaction, uncles [][]*types.Header, time time.Time, announcedAt time.Time) ([][]*types.Transaction, [][]*types.Header) {
|
||||||
log.Trace("Filtering bodies", "peer", peer, "txs", len(transactions), "uncles", len(uncles))
|
log.Trace("Filtering bodies", "peer", peer, "txs", len(transactions), "uncles", len(uncles))
|
||||||
|
|
||||||
// Send the filter channel to the fetcher
|
// Send the filter channel to the fetcher
|
||||||
|
|
@ -317,7 +323,7 @@ func (f *BlockFetcher) FilterBodies(peer string, transactions [][]*types.Transac
|
||||||
}
|
}
|
||||||
// Request the filtering of the body list
|
// Request the filtering of the body list
|
||||||
select {
|
select {
|
||||||
case filter <- &bodyFilterTask{peer: peer, transactions: transactions, uncles: uncles, time: time}:
|
case filter <- &bodyFilterTask{peer: peer, transactions: transactions, uncles: uncles, time: time, announcedTime: announcedAt}:
|
||||||
case <-f.quit:
|
case <-f.quit:
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
@ -480,7 +486,7 @@ func (f *BlockFetcher) loop() {
|
||||||
log.Trace("Fetching scheduled headers", "peer", peer, "list", hashes)
|
log.Trace("Fetching scheduled headers", "peer", peer, "list", hashes)
|
||||||
|
|
||||||
// Create a closure of the fetch and schedule in on a new thread
|
// Create a closure of the fetch and schedule in on a new thread
|
||||||
fetchHeader, hashes := f.fetching[hashes[0]].fetchHeader, hashes
|
fetchHeader, hashes, announcedAt := f.fetching[hashes[0]].fetchHeader, hashes, f.fetching[hashes[0]].time
|
||||||
go func(peer string) {
|
go func(peer string) {
|
||||||
if f.fetchingHook != nil {
|
if f.fetchingHook != nil {
|
||||||
f.fetchingHook(hashes)
|
f.fetchingHook(hashes)
|
||||||
|
|
@ -504,7 +510,7 @@ func (f *BlockFetcher) loop() {
|
||||||
select {
|
select {
|
||||||
case res := <-resCh:
|
case res := <-resCh:
|
||||||
res.Done <- nil
|
res.Done <- nil
|
||||||
f.FilterHeaders(peer, *res.Res.(*eth.BlockHeadersPacket), time.Now().Add(res.Time))
|
f.FilterHeaders(peer, *res.Res.(*eth.BlockHeadersPacket), time.Now(), announcedAt)
|
||||||
|
|
||||||
case <-timeout.C:
|
case <-timeout.C:
|
||||||
// The peer didn't respond in time. The request
|
// The peer didn't respond in time. The request
|
||||||
|
|
@ -547,6 +553,7 @@ func (f *BlockFetcher) loop() {
|
||||||
|
|
||||||
fetchBodies := f.completing[hashes[0]].fetchBodies
|
fetchBodies := f.completing[hashes[0]].fetchBodies
|
||||||
bodyFetchMeter.Mark(int64(len(hashes)))
|
bodyFetchMeter.Mark(int64(len(hashes)))
|
||||||
|
announcedAt := f.completing[hashes[0]].time
|
||||||
|
|
||||||
go func(peer string, hashes []common.Hash) {
|
go func(peer string, hashes []common.Hash) {
|
||||||
resCh := make(chan *eth.Response)
|
resCh := make(chan *eth.Response)
|
||||||
|
|
@ -565,7 +572,7 @@ func (f *BlockFetcher) loop() {
|
||||||
res.Done <- nil
|
res.Done <- nil
|
||||||
// Ignoring withdrawals here, since the block fetcher is not used post-merge.
|
// Ignoring withdrawals here, since the block fetcher is not used post-merge.
|
||||||
txs, uncles, _ := res.Res.(*eth.BlockBodiesPacket).Unpack()
|
txs, uncles, _ := res.Res.(*eth.BlockBodiesPacket).Unpack()
|
||||||
f.FilterBodies(peer, txs, uncles, time.Now())
|
f.FilterBodies(peer, txs, uncles, time.Now(), announcedAt)
|
||||||
|
|
||||||
case <-timeout.C:
|
case <-timeout.C:
|
||||||
// The peer didn't respond in time. The request
|
// The peer didn't respond in time. The request
|
||||||
|
|
@ -631,6 +638,7 @@ func (f *BlockFetcher) loop() {
|
||||||
|
|
||||||
block := types.NewBlockWithHeader(header)
|
block := types.NewBlockWithHeader(header)
|
||||||
block.ReceivedAt = task.time
|
block.ReceivedAt = task.time
|
||||||
|
block.AnnouncedAt = &task.announcedTime
|
||||||
|
|
||||||
complete = append(complete, block)
|
complete = append(complete, block)
|
||||||
f.completing[hash] = announce
|
f.completing[hash] = announce
|
||||||
|
|
@ -725,6 +733,7 @@ func (f *BlockFetcher) loop() {
|
||||||
if f.getBlock(hash) == nil {
|
if f.getBlock(hash) == nil {
|
||||||
block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i])
|
block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i])
|
||||||
block.ReceivedAt = task.time
|
block.ReceivedAt = task.time
|
||||||
|
block.AnnouncedAt = &task.announcedTime
|
||||||
blocks = append(blocks, block)
|
blocks = append(blocks, block)
|
||||||
} else {
|
} else {
|
||||||
f.forgetHash(hash)
|
f.forgetHash(hash)
|
||||||
|
|
@ -923,6 +932,31 @@ func (f *BlockFetcher) importBlocks(peer string, block *types.Block) {
|
||||||
log.Debug("Propagated block import failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
|
log.Debug("Propagated block import failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if f.enableBlockTracking {
|
||||||
|
// Log the insertion event
|
||||||
|
var (
|
||||||
|
msg string
|
||||||
|
delayInMs uint64
|
||||||
|
prettyDelay common.PrettyDuration
|
||||||
|
)
|
||||||
|
|
||||||
|
if block.AnnouncedAt != nil {
|
||||||
|
msg = "[block tracker] Inserted new block with announcement"
|
||||||
|
delayInMs = uint64(time.Since(*block.AnnouncedAt).Milliseconds())
|
||||||
|
prettyDelay = common.PrettyDuration(time.Since(*block.AnnouncedAt))
|
||||||
|
} else {
|
||||||
|
msg = "[block tracker] Inserted new block without announcement"
|
||||||
|
delayInMs = uint64(time.Since(block.ReceivedAt).Milliseconds())
|
||||||
|
prettyDelay = common.PrettyDuration(time.Since(block.ReceivedAt))
|
||||||
|
}
|
||||||
|
|
||||||
|
totalDelayInMs := uint64(time.Now().UnixMilli()) - block.Time()*1000
|
||||||
|
totalDelay := common.PrettyDuration(time.Millisecond * time.Duration(totalDelayInMs))
|
||||||
|
|
||||||
|
log.Info(msg, "number", block.Number().Uint64(), "hash", hash, "delay", prettyDelay, "delayInMs", delayInMs, "totalDelay", totalDelay, "totalDelayInMs", totalDelayInMs)
|
||||||
|
}
|
||||||
|
|
||||||
// If import succeeded, broadcast the block
|
// If import succeeded, broadcast the block
|
||||||
blockAnnounceOutTimer.UpdateSince(block.ReceivedAt)
|
blockAnnounceOutTimer.UpdateSince(block.ReceivedAt)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -104,7 +104,7 @@ func newTester(light bool) *fetcherTester {
|
||||||
blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis},
|
blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis},
|
||||||
drops: make(map[string]bool),
|
drops: make(map[string]bool),
|
||||||
}
|
}
|
||||||
tester.fetcher = NewBlockFetcher(light, tester.getHeader, tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertHeaders, tester.insertChain, tester.dropPeer)
|
tester.fetcher = NewBlockFetcher(light, tester.getHeader, tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertHeaders, tester.insertChain, tester.dropPeer, false)
|
||||||
tester.fetcher.Start()
|
tester.fetcher.Start()
|
||||||
|
|
||||||
return tester
|
return tester
|
||||||
|
|
|
||||||
|
|
@ -85,18 +85,19 @@ type txPool interface {
|
||||||
// handlerConfig is the collection of initialization parameters to create a full
|
// handlerConfig is the collection of initialization parameters to create a full
|
||||||
// node network handler.
|
// node network handler.
|
||||||
type handlerConfig struct {
|
type handlerConfig struct {
|
||||||
Database ethdb.Database // Database for direct sync insertions
|
Database ethdb.Database // Database for direct sync insertions
|
||||||
Chain *core.BlockChain // Blockchain to serve data from
|
Chain *core.BlockChain // Blockchain to serve data from
|
||||||
TxPool txPool // Transaction pool to propagate from
|
TxPool txPool // Transaction pool to propagate from
|
||||||
Merger *consensus.Merger // The manager for eth1/2 transition
|
Merger *consensus.Merger // The manager for eth1/2 transition
|
||||||
Network uint64 // Network identifier to adfvertise
|
Network uint64 // Network identifier to adfvertise
|
||||||
Sync downloader.SyncMode // Whether to snap or full sync
|
Sync downloader.SyncMode // Whether to snap or full sync
|
||||||
BloomCache uint64 // Megabytes to alloc for snap sync bloom
|
BloomCache uint64 // Megabytes to alloc for snap sync bloom
|
||||||
EventMux *event.TypeMux // Legacy event mux, deprecate for `feed`
|
EventMux *event.TypeMux // Legacy event mux, deprecate for `feed`
|
||||||
txArrivalWait time.Duration // Maximum duration to wait for an announced tx before requesting it
|
txArrivalWait time.Duration // Maximum duration to wait for an announced tx before requesting it
|
||||||
checker ethereum.ChainValidator
|
checker ethereum.ChainValidator
|
||||||
RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
|
RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
|
||||||
EthAPI *ethapi.BlockChainAPI // EthAPI to interact
|
EthAPI *ethapi.BlockChainAPI // EthAPI to interact
|
||||||
|
enableBlockTracking bool // Whether to log information collected while tracking block lifecycle
|
||||||
}
|
}
|
||||||
|
|
||||||
type handler struct {
|
type handler struct {
|
||||||
|
|
@ -126,6 +127,8 @@ type handler struct {
|
||||||
|
|
||||||
requiredBlocks map[uint64]common.Hash
|
requiredBlocks map[uint64]common.Hash
|
||||||
|
|
||||||
|
enableBlockTracking bool
|
||||||
|
|
||||||
// channels for fetcher, syncer, txsyncLoop
|
// channels for fetcher, syncer, txsyncLoop
|
||||||
quitSync chan struct{}
|
quitSync chan struct{}
|
||||||
|
|
||||||
|
|
@ -144,19 +147,20 @@ func newHandler(config *handlerConfig) (*handler, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
h := &handler{
|
h := &handler{
|
||||||
networkID: config.Network,
|
networkID: config.Network,
|
||||||
forkFilter: forkid.NewFilter(config.Chain),
|
forkFilter: forkid.NewFilter(config.Chain),
|
||||||
eventMux: config.EventMux,
|
eventMux: config.EventMux,
|
||||||
database: config.Database,
|
database: config.Database,
|
||||||
txpool: config.TxPool,
|
txpool: config.TxPool,
|
||||||
chain: config.Chain,
|
chain: config.Chain,
|
||||||
peers: newPeerSet(),
|
peers: newPeerSet(),
|
||||||
merger: config.Merger,
|
merger: config.Merger,
|
||||||
ethAPI: config.EthAPI,
|
ethAPI: config.EthAPI,
|
||||||
requiredBlocks: config.RequiredBlocks,
|
requiredBlocks: config.RequiredBlocks,
|
||||||
quitSync: make(chan struct{}),
|
enableBlockTracking: config.enableBlockTracking,
|
||||||
handlerDoneCh: make(chan struct{}),
|
quitSync: make(chan struct{}),
|
||||||
handlerStartCh: make(chan struct{}),
|
handlerDoneCh: make(chan struct{}),
|
||||||
|
handlerStartCh: make(chan struct{}),
|
||||||
}
|
}
|
||||||
if config.Sync == downloader.FullSync {
|
if config.Sync == downloader.FullSync {
|
||||||
// The database seems empty as the current block is the genesis. Yet the snap
|
// The database seems empty as the current block is the genesis. Yet the snap
|
||||||
|
|
@ -295,7 +299,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
|
||||||
|
|
||||||
return n, err
|
return n, err
|
||||||
}
|
}
|
||||||
h.blockFetcher = fetcher.NewBlockFetcher(false, nil, h.chain.GetBlockByHash, validator, h.BroadcastBlock, heighter, nil, inserter, h.removePeer)
|
h.blockFetcher = fetcher.NewBlockFetcher(false, nil, h.chain.GetBlockByHash, validator, h.BroadcastBlock, heighter, nil, inserter, h.removePeer, h.enableBlockTracking)
|
||||||
|
|
||||||
fetchTx := func(peer string, hashes []common.Hash) error {
|
fetchTx := func(peer string, hashes []common.Hash) error {
|
||||||
p := h.peers.peer(peer)
|
p := h.peers.peer(peer)
|
||||||
|
|
@ -688,6 +692,11 @@ func (h *handler) minedBroadcastLoop() {
|
||||||
|
|
||||||
for obj := range h.minedBlockSub.Chan() {
|
for obj := range h.minedBlockSub.Chan() {
|
||||||
if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
|
if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
|
||||||
|
if h.enableBlockTracking {
|
||||||
|
delayInMs := uint64(time.Now().UnixMilli()) - ev.Block.Time()*1000
|
||||||
|
delay := common.PrettyDuration(time.Millisecond * time.Duration(delayInMs))
|
||||||
|
log.Info("[block tracker] Broadcasting mined block", "number", ev.Block.NumberU64(), "hash", ev.Block.Hash(), "blockTime", ev.Block.Time(), "now", time.Now().Unix(), "delay", delay, "delayInMs", delayInMs)
|
||||||
|
}
|
||||||
h.BroadcastBlock(ev.Block, true) // First propagate block to peers
|
h.BroadcastBlock(ev.Block, true) // First propagate block to peers
|
||||||
h.BroadcastBlock(ev.Block, false) // Only then announce to the rest
|
h.BroadcastBlock(ev.Block, false) // Only then announce to the rest
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -382,8 +382,10 @@ func handleNewBlock(backend Backend, msg Decoder, peer *Peer) error {
|
||||||
return nil // TODO(karalabe): return error eventually, but wait a few releases
|
return nil // TODO(karalabe): return error eventually, but wait a few releases
|
||||||
}
|
}
|
||||||
|
|
||||||
|
msgTime := msg.Time()
|
||||||
ann.Block.ReceivedAt = msg.Time()
|
ann.Block.ReceivedAt = msg.Time()
|
||||||
ann.Block.ReceivedFrom = peer
|
ann.Block.ReceivedFrom = peer
|
||||||
|
ann.Block.AnnouncedAt = &msgTime
|
||||||
|
|
||||||
// Mark the peer as owning the block
|
// Mark the peer as owning the block
|
||||||
peer.markBlock(ann.Block.Hash())
|
peer.markBlock(ann.Block.Hash())
|
||||||
|
|
|
||||||
|
|
@ -154,6 +154,9 @@ type LoggingConfig struct {
|
||||||
// Prepends log messages with call-site location (file and line number)
|
// Prepends log messages with call-site location (file and line number)
|
||||||
Debug bool `hcl:"debug,optional" toml:"debug,optional"`
|
Debug bool `hcl:"debug,optional" toml:"debug,optional"`
|
||||||
|
|
||||||
|
// EnableBlockTracking allows logging of information collected while tracking block lifecycle
|
||||||
|
EnableBlockTracking bool `hcl:"enable-block-tracking,optional" toml:"enable-block-tracking,optional"`
|
||||||
|
|
||||||
// TODO - implement this
|
// TODO - implement this
|
||||||
// // Write execution trace to the given file
|
// // Write execution trace to the given file
|
||||||
// Trace string `hcl:"trace,optional" toml:"trace,optional"`
|
// Trace string `hcl:"trace,optional" toml:"trace,optional"`
|
||||||
|
|
@ -606,10 +609,11 @@ func DefaultConfig() *Config {
|
||||||
DBEngine: "leveldb",
|
DBEngine: "leveldb",
|
||||||
KeyStoreDir: "",
|
KeyStoreDir: "",
|
||||||
Logging: &LoggingConfig{
|
Logging: &LoggingConfig{
|
||||||
Vmodule: "",
|
Vmodule: "",
|
||||||
Json: false,
|
Json: false,
|
||||||
Backtrace: "",
|
Backtrace: "",
|
||||||
Debug: false,
|
Debug: false,
|
||||||
|
EnableBlockTracking: false,
|
||||||
},
|
},
|
||||||
RPCBatchLimit: 100,
|
RPCBatchLimit: 100,
|
||||||
RPCReturnDataLimit: 100000,
|
RPCReturnDataLimit: 100000,
|
||||||
|
|
@ -1186,6 +1190,8 @@ func (c *Config) buildEth(stack *node.Node, accountManager *accounts.Manager) (*
|
||||||
n.DatabaseFreezer = c.Ancient
|
n.DatabaseFreezer = c.Ancient
|
||||||
}
|
}
|
||||||
|
|
||||||
|
n.EnableBlockTracking = c.Logging.EnableBlockTracking
|
||||||
|
|
||||||
return &n, nil
|
return &n, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -146,6 +146,13 @@ func (c *Command) Flags(config *Config) *flagset.Flagset {
|
||||||
Default: c.cliConfig.Logging.Debug,
|
Default: c.cliConfig.Logging.Debug,
|
||||||
Group: "Logging",
|
Group: "Logging",
|
||||||
})
|
})
|
||||||
|
f.BoolFlag(&flagset.BoolFlag{
|
||||||
|
Name: "log.enable-block-tracking",
|
||||||
|
Usage: "Enables additional logging of information collected while tracking block lifecycle",
|
||||||
|
Value: &c.cliConfig.Logging.EnableBlockTracking,
|
||||||
|
Default: c.cliConfig.Logging.EnableBlockTracking,
|
||||||
|
Group: "Logging",
|
||||||
|
})
|
||||||
|
|
||||||
// heimdall
|
// heimdall
|
||||||
f.StringFlag(&flagset.StringFlag{
|
f.StringFlag(&flagset.StringFlag{
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ gcmode = "archive"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 50
|
maxpeers = 50
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ syncmode = "full"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 50
|
maxpeers = 50
|
||||||
|
|
|
||||||
|
|
@ -23,6 +23,7 @@ syncmode = "full"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 20
|
maxpeers = 20
|
||||||
|
|
|
||||||
|
|
@ -23,6 +23,7 @@ syncmode = "full"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 50
|
maxpeers = 50
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ gcmode = "archive"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 50
|
maxpeers = 50
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ syncmode = "full"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 50
|
maxpeers = 50
|
||||||
|
|
|
||||||
|
|
@ -23,6 +23,7 @@ syncmode = "full"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 1
|
maxpeers = 1
|
||||||
|
|
|
||||||
|
|
@ -23,6 +23,7 @@ syncmode = "full"
|
||||||
# json = false
|
# json = false
|
||||||
# backtrace = ""
|
# backtrace = ""
|
||||||
# debug = true
|
# debug = true
|
||||||
|
# enable-block-tracking = false
|
||||||
|
|
||||||
[p2p]
|
[p2p]
|
||||||
maxpeers = 50
|
maxpeers = 50
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue