forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-25 14:16:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core, eth, trie: fix data races and review issues

Browse source
This commit is contained in:
Péter Szilágyi 2015-10-13 12:04:25 +03:00
parent cca476f573
commit 3eb98b6178
14 changed files with 251 additions and 124 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
core/blockchain.go
View file

@ -178,21 +178,21 @@ func (self *BlockChain) loadLastState() error {
fastTd := self.GetTd(self.currentFastBlock.Hash())
glog.V(logger.Info).Infof("Last header: #%d [%x…] TD=%v", self.currentHeader.Number, self.currentHeader.Hash().Bytes()[:4], headerTd)
glog.V(logger.Info).Infof("Fast block: #%d [%x…] TD=%v", self.currentFastBlock.Number(), self.currentFastBlock.Hash().Bytes()[:4], fastTd)
glog.V(logger.Info).Infof("Last block: #%d [%x…] TD=%v", self.currentBlock.Number(), self.currentBlock.Hash().Bytes()[:4], blockTd)
glog.V(logger.Info).Infof("Fast block: #%d [%x…] TD=%v", self.currentFastBlock.Number(), self.currentFastBlock.Hash().Bytes()[:4], fastTd)
return nil
}
// SetHead rewind the local chain to a new head entity. In the case of headers,
// everything above the new head will be deleted and the new one set. In the case
// of blocks though, the head may be further rewound if block bodies are missing
// (non-archive nodes after a fast sync).
// SetHead rewinds the local chain to a new head. In the case of headers, everything
// above the new head will be deleted and the new one set. In the case of blocks
// though, the head may be further rewound if block bodies are missing (non-archive
// nodes after a fast sync).
func (bc *BlockChain) SetHead(head uint64) {
bc.mu.Lock()
defer bc.mu.Unlock()
// Figure out the highest known canonical assignment
// Figure out the highest known canonical headers and/or blocks
height := uint64(0)
if bc.currentHeader != nil {
if hh := bc.currentHeader.Number.Uint64(); hh > height {
@ -266,7 +266,7 @@ func (bc *BlockChain) SetHead(head uint64) {
// FastSyncCommitHead sets the current head block to the one defined by the hash
// irrelevant what the chain contents were prior.
func (self *BlockChain) FastSyncCommitHead(hash common.Hash) error {
// Make sure that both the block as well at it's state trie exists
// Make sure that both the block as well at its state trie exists
block := self.GetBlock(hash)
if block == nil {
return fmt.Errorf("non existent block [%x…]", hash[:4])
@ -298,7 +298,7 @@ func (self *BlockChain) LastBlockHash() common.Hash {
}
// CurrentHeader retrieves the current head header of the canonical chain. The
// header is retrieved from the chain manager's internal cache.
// header is retrieved from the blockchain's internal cache.
func (self *BlockChain) CurrentHeader() *types.Header {
self.mu.RLock()
defer self.mu.RUnlock()
@ -307,7 +307,7 @@ func (self *BlockChain) CurrentHeader() *types.Header {
}
// CurrentBlock retrieves the current head block of the canonical chain. The
// block is retrieved from the chain manager's internal cache.
// block is retrieved from the blockchain's internal cache.
func (self *BlockChain) CurrentBlock() *types.Block {
self.mu.RLock()
defer self.mu.RUnlock()
@ -316,7 +316,7 @@ func (self *BlockChain) CurrentBlock() *types.Block {
}
// CurrentFastBlock retrieves the current fast-sync head block of the canonical
// chain. The block is retrieved from the chain manager's internal cache.
// chain. The block is retrieved from the blockchain's internal cache.
func (self *BlockChain) CurrentFastBlock() *types.Block {
self.mu.RLock()
defer self.mu.RUnlock()
@ -353,7 +353,7 @@ func (bc *BlockChain) ResetWithGenesisBlock(genesis *types.Block) {
bc.mu.Lock()
defer bc.mu.Unlock()
// Prepare the genesis block and reinitialize the chain
// Prepare the genesis block and reinitialise the chain
if err := WriteTd(bc.chainDb, genesis.Hash(), genesis.Difficulty()); err != nil {
glog.Fatalf("failed to write genesis block TD: %v", err)
}
@ -403,7 +403,7 @@ func (self *BlockChain) ExportN(w io.Writer, first uint64, last uint64) error {
// insert injects a new head block into the current block chain. This method
// assumes that the block is indeed a true head. It will also reset the head
// header and the head fast sync block to this very same block to prevent them
// from diverging on a different header chain.
// from pointing to a possibly old canonical chain (i.e. side chain by now).
//
// Note, this function assumes that the `mu` mutex is held!
func (bc *BlockChain) insert(block *types.Block) {
@ -625,10 +625,10 @@ const (
// writeHeader writes a header into the local chain, given that its parent is
// already known. If the total difficulty of the newly inserted header becomes
// greater than the old known TD, the canonical chain is re-routed.
// greater than the current known TD, the canonical chain is re-routed.
//
// Note: This method is not concurrent-safe with inserting blocks simultaneously
// into the chain, as side effects caused by reorganizations cannot be emulated
// into the chain, as side effects caused by reorganisations cannot be emulated
// without the real blocks. Hence, writing headers directly should only be done
// in two scenarios: pure-header mode of operation (light clients), or properly
// separated header/block phases (non-archive clients).
@ -678,10 +678,9 @@ func (self *BlockChain) writeHeader(header *types.Header) error {
return nil
}
// InsertHeaderChain will attempt to insert the given header chain in to the
// local chain, possibly creating a fork. If an error is returned, it will
// return the index number of the failing header as well an error describing
// what went wrong.
// InsertHeaderChain attempts to insert the given header chain in to the local
// chain, possibly creating a reorg. If an error is returned, it will return the
// index number of the failing header as well an error describing what went wrong.
//
// The verify parameter can be used to fine tune whether nonce verification
// should be done or not. The reason behind the optional check is because some
@ -766,10 +765,6 @@ func (self *BlockChain) InsertHeaderChain(chain []*types.Header, checkFreq int)
pending.Wait()
// If anything failed, report
if atomic.LoadInt32(&self.procInterrupt) == 1 {
glog.V(logger.Debug).Infoln("premature abort during receipt chain processing")
return 0, nil
}
if failed > 0 {
for i, err := range errs {
if err != nil {
@ -807,6 +802,9 @@ func (self *BlockChain) InsertHeaderChain(chain []*types.Header, checkFreq int)
// Rollback is designed to remove a chain of links from the database that aren't
// certain enough to be valid.
func (self *BlockChain) Rollback(chain []common.Hash) {
self.mu.Lock()
defer self.mu.Unlock()
for i := len(chain) - 1; i >= 0; i-- {
hash := chain[i]
@ -920,10 +918,6 @@ func (self *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain
pending.Wait()
// If anything failed, report
if atomic.LoadInt32(&self.procInterrupt) == 1 {
glog.V(logger.Debug).Infoln("premature abort during receipt chain processing")
return 0, nil
}
if failed > 0 {
for i, err := range errs {
if err != nil {
@ -931,6 +925,10 @@ func (self *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain
}
}
}
if atomic.LoadInt32(&self.procInterrupt) == 1 {
glog.V(logger.Debug).Infoln("premature abort during receipt chain processing")
return 0, nil
}
// Update the head fast sync block if better
self.mu.Lock()
head := blockChain[len(errs)-1]

3
core/state/sync.go
View file

@ -26,14 +26,13 @@ import (
"github.com/ethereum/go-ethereum/trie"
)
// StateSync is the main state synchronisation scheduler, which provides yet the
// StateSync is the main state synchronisation scheduler, which provides yet the
// unknown state hashes to retrieve, accepts node data associated with said hashes
// and reconstructs the state database step by step until all is done.
type StateSync trie.TrieSync
// NewStateSync create a new state trie download scheduler.
func NewStateSync(root common.Hash, database ethdb.Database) *StateSync {
// Pre-declare the result syncer t
var syncer *trie.TrieSync
callback := func(leaf []byte, parent common.Hash) error {

10
core/types/receipt.go
View file

@ -67,7 +67,7 @@ func (r *Receipt) DecodeRLP(s *rlp.Stream) error {
return nil
}
// RlpEncode implements common.RlpEncode required for SHA derivation.
// RlpEncode implements common.RlpEncode required for SHA3 derivation.
func (r *Receipt) RlpEncode() []byte {
bytes, err := rlp.EncodeToBytes(r)
if err != nil {
@ -82,7 +82,7 @@ func (r *Receipt) String() string {
}
// ReceiptForStorage is a wrapper around a Receipt that flattens and parses the
// entire content of a receipt, opposed to only the consensus fields originally.
// entire content of a receipt, as opposed to only the consensus fields originally.
type ReceiptForStorage Receipt
// EncodeRLP implements rlp.Encoder, and flattens all content fields of a receipt
@ -95,8 +95,8 @@ func (r *ReceiptForStorage) EncodeRLP(w io.Writer) error {
return rlp.Encode(w, []interface{}{r.PostState, r.CumulativeGasUsed, r.Bloom, r.TxHash, r.ContractAddress, logs, r.GasUsed})
}
// DecodeRLP implements rlp.Decoder, and loads the consensus fields of a receipt
// from an RLP stream.
// DecodeRLP implements rlp.Decoder, and loads both consensus and implementation
// fields of a receipt from an RLP stream.
func (r *ReceiptForStorage) DecodeRLP(s *rlp.Stream) error {
var receipt struct {
PostState []byte
@ -125,7 +125,7 @@ func (r *ReceiptForStorage) DecodeRLP(s *rlp.Stream) error {
// Receipts is a wrapper around a Receipt array to implement types.DerivableList.
type Receipts []*Receipt
// RlpEncode implements common.RlpEncode required for SHA derivation.
// RlpEncode implements common.RlpEncode required for SHA3 derivation.
func (r Receipts) RlpEncode() []byte {
bytes, err := rlp.EncodeToBytes(r)
if err != nil {

2
core/vm/log.go
View file

@ -66,6 +66,6 @@ func (l *Log) String() string {
type Logs []*Log
// LogForStorage is a wrapper around a Log that flattens and parses the entire
// content of a log, opposed to only the consensus fields originally (by hiding
// content of a log, as opposed to only the consensus fields originally (by hiding
// the rlp interface methods).
type LogForStorage Log

41
eth/downloader/downloader.go
View file

@ -19,6 +19,7 @@ package downloader
import (
"errors"
"fmt"
"math"
"math/big"
"strings"
@ -85,6 +86,7 @@ var (
errCancelHeaderFetch = errors.New("block header download canceled (requested)")
errCancelBodyFetch = errors.New("block body download canceled (requested)")
errCancelReceiptFetch = errors.New("receipt download canceled (requested)")
errCancelStateFetch = errors.New("state data download canceled (requested)")
errNoSyncActive = errors.New("no sync active")
)
@ -844,7 +846,7 @@ func (d *Downloader) fetchBlocks61(from uint64) error {
for _, peer := range idles {
// Short circuit if throttling activated
if d.queue.ThrottleBlocks() {
if d.queue.ShouldThrottleBlocks() {
throttled = true
break
}
@ -860,8 +862,13 @@ func (d *Downloader) fetchBlocks61(from uint64) error {
}
// Fetch the chunk and make sure any errors return the hashes to the queue
if err := peer.Fetch61(request); err != nil {
glog.V(logger.Error).Infof("%v: fetch failed, rescheduling", peer)
d.queue.CancelBlocks(request)
// Although we could try and make an attempt to fix this, this error really
// means that we've double allocated a fetch task to a peer. If that is the
// case, the internal state of the downloader and the queue is very wrong so
// better hard crash and note the error instead of silently accumulating into
// a much bigger issue.
panic(fmt.Sprintf("%v: fetch assignment failed, hard panic", peer))
d.queue.CancelBlocks(request) // noop for now
}
}
// Make sure that we have peers available for fetching. If all peers have been tried
@ -1230,12 +1237,11 @@ func (d *Downloader) fetchBodies(from uint64) error {
expire = func() []string { return d.queue.ExpireBodies(bodyHardTTL) }
fetch = func(p *peer, req *fetchRequest) error { return p.FetchBodies(req) }
capacity = func(p *peer) int { return p.BlockCapacity() }
getIdles = func() ([]*peer, int) { return d.peers.BodyIdlePeers() }
setIdle = func(p *peer) { p.SetBlocksIdle() }
setIdle = func(p *peer) { p.SetBodiesIdle() }
)
err := d.fetchParts(errCancelBodyFetch, d.bodyCh, deliver, d.bodyWakeCh, expire,
d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ThrottleBlocks, d.queue.ReserveBodies,
d.bodyFetchHook, fetch, d.queue.CancelBodies, capacity, getIdles, setIdle, "Body")
d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ShouldThrottleBlocks, d.queue.ReserveBodies,
d.bodyFetchHook, fetch, d.queue.CancelBodies, capacity, d.peers.BodyIdlePeers, setIdle, "Body")
glog.V(logger.Debug).Infof("Block body download terminated: %v", err)
return err
@ -1252,13 +1258,13 @@ func (d *Downloader) fetchReceipts(from uint64) error {
pack := packet.(*receiptPack)
return d.queue.DeliverReceipts(pack.peerId, pack.receipts)
}
expire = func() []string { return d.queue.ExpireReceipts(bodyHardTTL) }
expire = func() []string { return d.queue.ExpireReceipts(receiptHardTTL) }
fetch = func(p *peer, req *fetchRequest) error { return p.FetchReceipts(req) }
capacity = func(p *peer) int { return p.ReceiptCapacity() }
setIdle = func(p *peer) { p.SetReceiptsIdle() }
)
err := d.fetchParts(errCancelReceiptFetch, d.receiptCh, deliver, d.receiptWakeCh, expire,
d.queue.PendingReceipts, d.queue.InFlightReceipts, d.queue.ThrottleReceipts, d.queue.ReserveReceipts,
d.queue.PendingReceipts, d.queue.InFlightReceipts, d.queue.ShouldThrottleReceipts, d.queue.ReserveReceipts,
d.receiptFetchHook, fetch, d.queue.CancelReceipts, capacity, d.peers.ReceiptIdlePeers, setIdle, "Receipt")
glog.V(logger.Debug).Infof("Receipt download terminated: %v", err)
@ -1307,9 +1313,9 @@ func (d *Downloader) fetchNodeData() error {
capacity = func(p *peer) int { return p.NodeDataCapacity() }
setIdle = func(p *peer) { p.SetNodeDataIdle() }
)
err := d.fetchParts(errCancelReceiptFetch, d.stateCh, deliver, d.stateWakeCh, expire,
err := d.fetchParts(errCancelStateFetch, d.stateCh, deliver, d.stateWakeCh, expire,
d.queue.PendingNodeData, d.queue.InFlightNodeData, throttle, reserve, nil, fetch,
d.queue.CancelNodeData, capacity, d.peers.ReceiptIdlePeers, setIdle, "State")
d.queue.CancelNodeData, capacity, d.peers.NodeDataIdlePeers, setIdle, "State")
glog.V(logger.Debug).Infof("Node state data download terminated: %v", err)
return err
@ -1475,8 +1481,13 @@ func (d *Downloader) fetchParts(errCancel error, deliveryCh chan dataPack, deliv
fetchHook(request.Headers)
}
if err := fetch(peer, request); err != nil {
glog.V(logger.Error).Infof("%v: %s fetch failed, rescheduling", peer, strings.ToLower(kind))
cancel(request)
// Although we could try and make an attempt to fix this, this error really
// means that we've double allocated a fetch task to a peer. If that is the
// case, the internal state of the downloader and the queue is very wrong so
// better hard crash and note the error instead of silently accumulating into
// a much bigger issue.
panic(fmt.Sprintf("%v: %s fetch assignment failed, hard panic", peer, strings.ToLower(kind)))
cancel(request) // noop for now
}
running = true
}
@ -1556,7 +1567,7 @@ func (d *Downloader) process() {
blocks = append(blocks, types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles))
case d.mode == FastSync:
blocks = append(blocks, types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles))
if result.Header.Number.Uint64() <= d.queue.fastSyncPivot {
if result.Header.Number.Uint64() <= d.queue.FastSyncPivot() {
receipts = append(receipts, result.Receipts)
}
case d.mode == LightSync:
@ -1574,7 +1585,7 @@ func (d *Downloader) process() {
case len(receipts) > 0:
index, err = d.insertReceipts(blocks, receipts)
if err == nil && blocks[len(blocks)-1].NumberU64() == d.queue.fastSyncPivot {
if err == nil && blocks[len(blocks)-1].NumberU64() == d.queue.FastSyncPivot() {
index, err = len(blocks)-1, d.commitHeadBlock(blocks[len(blocks)-1].Hash())
}
default:

31
eth/downloader/downloader_test.go
View file

@ -43,6 +43,18 @@ var (
genesis = core.GenesisBlockForTesting(testdb, testAddress, big.NewInt(1000000000))
)
// This is a very ugly hack, but needed to catch a Heisenbug :P. The windows CI
// servers extremely rarely lock up in this test suite, and adding any logging
// or measurements simply hides the error completely. This snippet will ensure
// that in the case of a lockup, we'll at least have a cash dump of the goroutines
// to at least get a glimpse into what state the tests were in.
func init() {
go func() {
time.Sleep(8 * time.Minute)
panic("Test probably deadlocked")
}()
}
// makeChain creates a chain of n blocks starting at and including parent.
// the returned hash chain is ordered head->parent. In addition, every 3rd block
// contains a transaction and every 5th an uncle to allow testing correct block
@ -730,6 +742,7 @@ func testThrottling(t *testing.T, protocol int, mode SyncMode) {
for start := time.Now(); time.Since(start) < time.Second; {
time.Sleep(25 * time.Millisecond)
tester.lock.RLock()
tester.downloader.queue.lock.RLock()
cached = len(tester.downloader.queue.blockDonePool)
if mode == FastSync {
@ -739,16 +752,22 @@ func testThrottling(t *testing.T, protocol int, mode SyncMode) {
}
}
}
retrieved = len(tester.ownBlocks)
tester.downloader.queue.lock.RUnlock()
tester.lock.RUnlock()
if cached == blockCacheLimit || len(tester.ownBlocks)+cached+int(atomic.LoadUint32(&blocked)) == targetBlocks+1 {
if cached == blockCacheLimit || retrieved+cached+int(atomic.LoadUint32(&blocked)) == targetBlocks+1 {
break
}
}
// Make sure we filled up the cache, then exhaust it
time.Sleep(25 * time.Millisecond) // give it a chance to screw up
if cached != blockCacheLimit && len(tester.ownBlocks)+cached+int(atomic.LoadUint32(&blocked)) != targetBlocks+1 {
t.Fatalf("block count mismatch: have %v, want %v (owned %v, target %v)", cached, blockCacheLimit, len(tester.ownBlocks), targetBlocks+1)
tester.lock.RLock()
retrieved = len(tester.ownBlocks)
tester.lock.RUnlock()
if cached != blockCacheLimit && retrieved+cached+int(atomic.LoadUint32(&blocked)) != targetBlocks+1 {
t.Fatalf("block count mismatch: have %v, want %v (owned %v, blocked %v, target %v)", cached, blockCacheLimit, retrieved, atomic.LoadUint32(&blocked), targetBlocks+1)
}
// Permit the blocked blocks to import
if atomic.LoadUint32(&blocked) > 0 {
@ -921,12 +940,12 @@ func testMultiProtoSync(t *testing.T, protocol int, mode SyncMode) {
// Create peers of every type
tester := newTester(mode)
tester.newPeer("peer 61", 61, hashes, headers, blocks, receipts)
tester.newPeer("peer 62", 62, hashes, headers, blocks, receipts)
tester.newPeer("peer 61", 61, hashes, nil, blocks, nil)
tester.newPeer("peer 62", 62, hashes, headers, blocks, nil)
tester.newPeer("peer 63", 63, hashes, headers, blocks, receipts)
tester.newPeer("peer 64", 64, hashes, headers, blocks, receipts)
// Synchronise with the requestd peer and make sure all blocks were retrieved
// Synchronise with the requested peer and make sure all blocks were retrieved
if err := tester.sync(fmt.Sprintf("peer %d", protocol), nil); err != nil {
t.Fatalf("failed to synchronise blocks: %v", err)
}

4
eth/downloader/modes.go
View file

@ -20,7 +20,7 @@ package downloader
type SyncMode int
const (
FullSync SyncMode = iota // Synchronise the entire block-chain history from full blocks
FastSync // Quikcly download the headers, full sync only at the chain head
FullSync SyncMode = iota // Synchronise the entire blockchain history from full blocks
FastSync // Quickly download the headers, full sync only at the chain head
LightSync // Download only the headers and terminate afterwards
)

28
eth/downloader/peer.go
View file

@ -124,6 +124,10 @@ func (p *peer) Reset() {
// Fetch61 sends a block retrieval request to the remote peer.
func (p *peer) Fetch61(request *fetchRequest) error {
// Sanity check the protocol version
if p.version != 61 {
panic(fmt.Sprintf("block fetch [eth/61] requested on eth/%d", p.version))
}
// Short circuit if the peer is already fetching
if !atomic.CompareAndSwapInt32(&p.blockIdle, 0, 1) {
return errAlreadyFetching
@ -142,6 +146,10 @@ func (p *peer) Fetch61(request *fetchRequest) error {
// FetchBodies sends a block body retrieval request to the remote peer.
func (p *peer) FetchBodies(request *fetchRequest) error {
// Sanity check the protocol version
if p.version < 62 {
panic(fmt.Sprintf("body fetch [eth/62+] requested on eth/%d", p.version))
}
// Short circuit if the peer is already fetching
if !atomic.CompareAndSwapInt32(&p.blockIdle, 0, 1) {
return errAlreadyFetching
@ -160,6 +168,10 @@ func (p *peer) FetchBodies(request *fetchRequest) error {
// FetchReceipts sends a receipt retrieval request to the remote peer.
func (p *peer) FetchReceipts(request *fetchRequest) error {
// Sanity check the protocol version
if p.version < 63 {
panic(fmt.Sprintf("body fetch [eth/63+] requested on eth/%d", p.version))
}
// Short circuit if the peer is already fetching
if !atomic.CompareAndSwapInt32(&p.receiptIdle, 0, 1) {
return errAlreadyFetching
@ -178,6 +190,10 @@ func (p *peer) FetchReceipts(request *fetchRequest) error {
// FetchNodeData sends a node state data retrieval request to the remote peer.
func (p *peer) FetchNodeData(request *fetchRequest) error {
// Sanity check the protocol version
if p.version < 63 {
panic(fmt.Sprintf("node data fetch [eth/63+] requested on eth/%d", p.version))
}
// Short circuit if the peer is already fetching
if !atomic.CompareAndSwapInt32(&p.stateIdle, 0, 1) {
return errAlreadyFetching
@ -196,35 +212,35 @@ func (p *peer) FetchNodeData(request *fetchRequest) error {
// SetBlocksIdle sets the peer to idle, allowing it to execute new retrieval requests.
// Its block retrieval allowance will also be updated either up- or downwards,
// depending on whether the previous fetch completed in time or not.
// depending on whether the previous fetch completed in time.
func (p *peer) SetBlocksIdle() {
p.setIdle(p.blockStarted, blockSoftTTL, blockHardTTL, MaxBlockFetch, &p.blockCapacity, &p.blockIdle)
}
// SetBodiesIdle sets the peer to idle, allowing it to execute new retrieval requests.
// Its block body retrieval allowance will also be updated either up- or downwards,
// depending on whether the previous fetch completed in time or not.
// depending on whether the previous fetch completed in time.
func (p *peer) SetBodiesIdle() {
p.setIdle(p.blockStarted, bodySoftTTL, bodyHardTTL, MaxBlockFetch, &p.blockCapacity, &p.blockIdle)
p.setIdle(p.blockStarted, bodySoftTTL, bodyHardTTL, MaxBodyFetch, &p.blockCapacity, &p.blockIdle)
}
// SetReceiptsIdle sets the peer to idle, allowing it to execute new retrieval requests.
// Its receipt retrieval allowance will also be updated either up- or downwards,
// depending on whether the previous fetch completed in time or not.
// depending on whether the previous fetch completed in time.
func (p *peer) SetReceiptsIdle() {
p.setIdle(p.receiptStarted, receiptSoftTTL, receiptHardTTL, MaxReceiptFetch, &p.receiptCapacity, &p.receiptIdle)
}
// SetNodeDataIdle sets the peer to idle, allowing it to execute new retrieval
// requests. Its node data retrieval allowance will also be updated either up- or
// downwards, depending on whether the previous fetch completed in time or not.
// downwards, depending on whether the previous fetch completed in time.
func (p *peer) SetNodeDataIdle() {
p.setIdle(p.stateStarted, stateSoftTTL, stateSoftTTL, MaxStateFetch, &p.stateCapacity, &p.stateIdle)
}
// setIdle sets the peer to idle, allowing it to execute new retrieval requests.
// Its data retrieval allowance will also be updated either up- or downwards,
// depending on whether the previous fetch completed in time or not.
// depending on whether the previous fetch completed in time.
func (p *peer) setIdle(started time.Time, softTTL, hardTTL time.Duration, maxFetch int, capacity, idle *int32) {
// Update the peer's download allowance based on previous performance
scale := 2.0

125
eth/downloader/queue.go
View file

@ -56,9 +56,8 @@ type fetchRequest struct {
Time time.Time // Time when the request was made
}
// fetchResult is the assembly collecting partial results from potentially more
// than one fetcher routines, until all outstanding retrievals complete and the
// result as a whole can be processed.
// fetchResult is a struct collecting partial results from data fetchers until
// all outstanding pieces complete and the result as a whole can be processed.
type fetchResult struct {
Pending int // Number of data fetches still pending
@ -89,7 +88,7 @@ type queue struct {
receiptPendPool map[string]*fetchRequest // [eth/63] Currently pending receipt retrieval operations
receiptDonePool map[common.Hash]struct{} // [eth/63] Set of the completed receipt fetches
stateTaskIndex int // [eth/63] Counter indexing the added hashes to ensure prioritized retrieval order
stateTaskIndex int // [eth/63] Counter indexing the added hashes to ensure prioritised retrieval order
stateTaskPool map[common.Hash]int // [eth/63] Pending node data retrieval tasks, mapping to their priority
stateTaskQueue *prque.Prque // [eth/63] Priority queue of the hashes to fetch the node data for
statePendPool map[string]*fetchRequest // [eth/63] Currently pending node data retrieval operations
@ -97,10 +96,10 @@ type queue struct {
stateDatabase ethdb.Database // [eth/63] Trie database to populate during state reassembly
stateScheduler *state.StateSync // [eth/63] State trie synchronisation scheduler and integrator
stateProcessors int32 // [eth/63] Number of currently running state processors
stateSchedLock sync.RWMutex // [eth/63] Lock serializing access to the state scheduler
stateSchedLock sync.RWMutex // [eth/63] Lock serialising access to the state scheduler
resultCache []*fetchResult // Downloaded but not yet delivered fetch results
resultOffset uint64 // Offset of the first cached fetch result in the block-chain
resultOffset uint64 // Offset of the first cached fetch result in the block chain
lock sync.RWMutex
}
@ -233,9 +232,17 @@ func (q *queue) Idle() bool {
return (queued + pending + cached) == 0
}
// ThrottleBlocks checks if the download should be throttled (active block (body)
// FastSyncPivot retrieves the currently used fast sync pivot point.
func (q *queue) FastSyncPivot() uint64 {
q.lock.RLock()
defer q.lock.RUnlock()
return q.fastSyncPivot
}
// ShouldThrottleBlocks checks if the download should be throttled (active block (body)
// fetches exceed block cache).
func (q *queue) ThrottleBlocks() bool {
func (q *queue) ShouldThrottleBlocks() bool {
q.lock.RLock()
defer q.lock.RUnlock()
@ -248,9 +255,9 @@ func (q *queue) ThrottleBlocks() bool {
return pending >= len(q.resultCache)-len(q.blockDonePool)
}
// ThrottleReceipts checks if the download should be throttled (active receipt
// ShouldThrottleReceipts checks if the download should be throttled (active receipt
// fetches exceed block cache).
func (q *queue) ThrottleReceipts() bool {
func (q *queue) ShouldThrottleReceipts() bool {
q.lock.RLock()
defer q.lock.RUnlock()
@ -269,7 +276,7 @@ func (q *queue) Schedule61(hashes []common.Hash, fifo bool) []common.Hash {
q.lock.Lock()
defer q.lock.Unlock()
// Insert all the hashes prioritized in the arrival order
// Insert all the hashes prioritised in the arrival order
inserts := make([]common.Hash, 0, len(hashes))
for _, hash := range hashes {
// Skip anything we already have
@ -297,10 +304,10 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
q.lock.Lock()
defer q.lock.Unlock()
// Insert all the headers prioritized by the contained block number
// Insert all the headers prioritised by the contained block number
inserts := make([]*types.Header, 0, len(headers))
for _, header := range headers {
// Make sure chain order is honored and preserved throughout
// Make sure chain order is honoured and preserved throughout
hash := header.Hash()
if header.Number == nil || header.Number.Uint64() != from {
glog.V(logger.Warn).Infof("Header #%v [%x] broke chain ordering, expected %d", header.Number, hash[:4], from)
@ -411,6 +418,9 @@ func (q *queue) TakeResults() []*fetchResult {
// ReserveBlocks reserves a set of block hashes for the given peer, skipping any
// previously failed download.
func (q *queue) ReserveBlocks(p *peer, count int) *fetchRequest {
q.lock.Lock()
defer q.lock.Unlock()
return q.reserveHashes(p, count, q.hashQueue, nil, q.blockPendPool, len(q.resultCache)-len(q.blockDonePool))
}
@ -430,17 +440,21 @@ func (q *queue) ReserveNodeData(p *peer, count int) *fetchRequest {
}
}
}
q.lock.Lock()
defer q.lock.Unlock()
return q.reserveHashes(p, count, q.stateTaskQueue, generator, q.statePendPool, count)
}
// reserveHashes reserves a set of hashes for the given peer, skipping previously
// failed ones.
//
// Note, this method expects the queue lock to be already held for writing. The
// reason the lock is not obtained in here is because the parameters already need
// to access the queue, so they already need a lock anyway.
func (q *queue) reserveHashes(p *peer, count int, taskQueue *prque.Prque, taskGen func(int), pendPool map[string]*fetchRequest, maxPending int) *fetchRequest {
q.lock.Lock()
defer q.lock.Unlock()
// Short circuit if the peer's already downloading something (sanity check not
// to corrupt state)
// Short circuit if the peer's already downloading something (sanity check to
// not corrupt state)
if _, ok := pendPool[p.id]; ok {
return nil
}
@ -492,30 +506,37 @@ func (q *queue) reserveHashes(p *peer, count int, taskQueue *prque.Prque, taskGe
// previously failed downloads. Beside the next batch of needed fetches, it also
// returns a flag whether empty blocks were queued requiring processing.
func (q *queue) ReserveBodies(p *peer, count int) (*fetchRequest, bool, error) {
noop := func(header *types.Header) bool {
isNoop := func(header *types.Header) bool {
return header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash
}
return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, noop)
q.lock.Lock()
defer q.lock.Unlock()
return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, isNoop)
}
// ReserveReceipts reserves a set of receipt fetches for the given peer, skipping
// any previously failed downloads. Beside the next batch of needed fetches, it
// also returns a flag whether empty receipts were queued requiring importing.
func (q *queue) ReserveReceipts(p *peer, count int) (*fetchRequest, bool, error) {
noop := func(header *types.Header) bool {
isNoop := func(header *types.Header) bool {
return header.ReceiptHash == types.EmptyRootHash
}
return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, noop)
q.lock.Lock()
defer q.lock.Unlock()
return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, isNoop)
}
// reserveHeaders reserves a set of data download operations for a given peer,
// skipping any previously failed ones. This method is a generic version used
// by the individual special reservation functions.
//
// Note, this method expects the queue lock to be already held for writing. The
// reason the lock is not obtained in here is because the parameters already need
// to access the queue, so they already need a lock anyway.
func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, noop func(*types.Header) bool) (*fetchRequest, bool, error) {
q.lock.Lock()
defer q.lock.Unlock()
pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, isNoop func(*types.Header) bool) (*fetchRequest, bool, error) {
// Short circuit if the pool has been depleted, or if the peer's already
// downloading something (sanity check not to corrupt state)
if taskQueue.Empty() {
@ -537,7 +558,7 @@ func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*typ
for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ {
header := taskQueue.PopItem().(*types.Header)
// If we're the first to request this task, initialize the result container
// If we're the first to request this task, initialise the result container
index := int(header.Number.Int64() - int64(q.resultOffset))
if index >= len(q.resultCache) || index < 0 {
return nil, false, errInvalidChain
@ -553,7 +574,7 @@ func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*typ
}
}
// If this fetch task is a noop, skip this fetch operation
if noop(header) {
if isNoop(header) {
donePool[header.Hash()] = struct{}{}
delete(taskPool, header.Hash())
@ -562,7 +583,7 @@ func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*typ
progress = true
continue
}
// Otherwise if not a known unknown block, add to the retrieve list
// Otherwise unless the peer is known not to have the data, add to the retrieve list
if p.ignored.Has(header.Hash()) {
skip = append(skip, header)
} else {
@ -655,35 +676,48 @@ func (q *queue) Revoke(peerId string) {
}
// ExpireBlocks checks for in flight requests that exceeded a timeout allowance,
// canceling them and returning the responsible peers for penalization.
// canceling them and returning the responsible peers for penalisation.
func (q *queue) ExpireBlocks(timeout time.Duration) []string {
q.lock.Lock()
defer q.lock.Unlock()
return q.expire(timeout, q.blockPendPool, q.hashQueue, blockTimeoutMeter)
}
// ExpireBodies checks for in flight block body requests that exceeded a timeout
// allowance, canceling them and returning the responsible peers for penalization.
// allowance, canceling them and returning the responsible peers for penalisation.
func (q *queue) ExpireBodies(timeout time.Duration) []string {
q.lock.Lock()
defer q.lock.Unlock()
return q.expire(timeout, q.blockPendPool, q.blockTaskQueue, bodyTimeoutMeter)
}
// ExpireReceipts checks for in flight receipt requests that exceeded a timeout
// allowance, canceling them and returning the responsible peers for penalization.
// allowance, canceling them and returning the responsible peers for penalisation.
func (q *queue) ExpireReceipts(timeout time.Duration) []string {
q.lock.Lock()
defer q.lock.Unlock()
return q.expire(timeout, q.receiptPendPool, q.receiptTaskQueue, receiptTimeoutMeter)
}
// ExpireNodeData checks for in flight node data requests that exceeded a timeout
// allowance, canceling them and returning the responsible peers for penalization.
// allowance, canceling them and returning the responsible peers for penalisation.
func (q *queue) ExpireNodeData(timeout time.Duration) []string {
q.lock.Lock()
defer q.lock.Unlock()
return q.expire(timeout, q.statePendPool, q.stateTaskQueue, stateTimeoutMeter)
}
// expire is the generic check that move expired tasks from a pending pool back
// into a task pool, returning all entities caught with expired tasks.
//
// Note, this method expects the queue lock to be already held for writing. The
// reason the lock is not obtained in here is because the parameters already need
// to access the queue, so they already need a lock anyway.
func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest, taskQueue *prque.Prque, timeoutMeter metrics.Meter) []string {
q.lock.Lock()
defer q.lock.Unlock()
// Iterate over the expired requests and return each to the queue
peers := []string{}
for id, request := range pendPool {
@ -764,7 +798,7 @@ func (q *queue) DeliverBlocks(id string, blocks []*types.Block) error {
case len(errs) == 1 && (errs[0] == errInvalidChain || errs[0] == errInvalidBlock):
return errs[0]
case len(errs) == len(request.Headers):
case len(errs) == len(blocks):
return errStaleDelivery
default:
@ -774,6 +808,9 @@ func (q *queue) DeliverBlocks(id string, blocks []*types.Block) error {
// DeliverBodies injects a block body retrieval response into the results queue.
func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) error {
q.lock.Lock()
defer q.lock.Unlock()
reconstruct := func(header *types.Header, index int, result *fetchResult) error {
if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash || types.CalcUncleHash(uncleLists[index]) != header.UncleHash {
return errInvalidBody
@ -787,6 +824,9 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLi
// DeliverReceipts injects a receipt retrieval response into the results queue.
func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) error {
q.lock.Lock()
defer q.lock.Unlock()
reconstruct := func(header *types.Header, index int, result *fetchResult) error {
if types.DeriveSha(types.Receipts(receiptList[index])) != header.ReceiptHash {
return errInvalidReceipt
@ -798,11 +838,12 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) error
}
// deliver injects a data retrieval response into the results queue.
//
// Note, this method expects the queue lock to be already held for writing. The
// reason the lock is not obtained in here is because the parameters already need
// to access the queue, so they already need a lock anyway.
func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque, pendPool map[string]*fetchRequest,
donePool map[common.Hash]struct{}, reqTimer metrics.Timer, results int, reconstruct func(header *types.Header, index int, result *fetchResult) error) error {
q.lock.Lock()
defer q.lock.Unlock()
// Short circuit if the data was never requested
request := pendPool[id]
if request == nil {
@ -855,7 +896,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
case len(errs) == 1 && (errs[0] == errInvalidChain || errs[0] == errInvalidBody || errs[0] == errInvalidReceipt):
return errs[0]
case len(errs) == len(request.Headers):
case len(errs) == results:
return errStaleDelivery
default:
@ -876,7 +917,7 @@ func (q *queue) DeliverNodeData(id string, data [][]byte, callback func(error, i
stateReqTimer.UpdateSince(request.Time)
delete(q.statePendPool, id)
// If no data was retrieved, mark them as unavailable for the origin peer
// If no data was retrieved, mark their hashes as unavailable for the origin peer
if len(data) == 0 {
for hash, _ := range request.Hashes {
request.Peer.ignored.Add(hash)

26
eth/fetcher/fetcher.go
View file

@ -142,9 +142,11 @@ type Fetcher struct {
dropPeer peerDropFn // Drops a peer for misbehaving
// Testing hooks
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)
importedHook func(*types.Block) // Method to call upon successful block import (both eth/61 and eth/62)
announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list
queueChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a block from the import queue
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)
importedHook func(*types.Block) // Method to call upon successful block import (both eth/61 and eth/62)
}
// New creates a block fetcher to retrieve blocks based on hash announcements.
@ -324,11 +326,16 @@ func (f *Fetcher) loop() {
height := f.chainHeight()
for !f.queue.Empty() {
op := f.queue.PopItem().(*inject)
if f.queueChangeHook != nil {
f.queueChangeHook(op.block.Hash(), false)
}
// If too high up the chain or phase, continue later
number := op.block.NumberU64()
if number > height+1 {
f.queue.Push(op, -float32(op.block.NumberU64()))
if f.queueChangeHook != nil {
f.queueChangeHook(op.block.Hash(), true)
}
break
}
// Otherwise if fresh and still unknown, try and import
@ -372,6 +379,9 @@ func (f *Fetcher) loop() {
}
f.announces[notification.origin] = count
f.announced[notification.hash] = append(f.announced[notification.hash], notification)
if f.announceChangeHook != nil && len(f.announced[notification.hash]) == 1 {
f.announceChangeHook(notification.hash, true)
}
if len(f.announced) == 1 {
f.rescheduleFetch(fetchTimer)
}
@ -714,7 +724,9 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) {
f.queues[peer] = count
f.queued[hash] = op
f.queue.Push(op, -float32(block.NumberU64()))
if f.queueChangeHook != nil {
f.queueChangeHook(op.block.Hash(), true)
}
if glog.V(logger.Debug) {
glog.Infof("Peer %s: queued block #%d [%x…], total %v", peer, block.NumberU64(), hash.Bytes()[:4], f.queue.Size())
}
@ -781,7 +793,9 @@ func (f *Fetcher) forgetHash(hash common.Hash) {
}
}
delete(f.announced, hash)
if f.announceChangeHook != nil {
f.announceChangeHook(hash, false)
}
// Remove any pending fetches and decrement the DOS counters
if announce := f.fetching[hash]; announce != nil {
f.announces[announce.origin]--

49
eth/fetcher/fetcher_test.go
View file

@ -145,6 +145,9 @@ func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) {
// dropPeer is an emulator for the peer removal, simply accumulating the various
// peers dropped by the fetcher.
func (f *fetcherTester) dropPeer(peer string) {
f.lock.Lock()
defer f.lock.Unlock()
f.drops[peer] = true
}
@ -608,8 +611,11 @@ func TestDistantPropagationDiscarding(t *testing.T) {
// Create a tester and simulate a head block being the middle of the above chain
tester := newTester()
tester.lock.Lock()
tester.hashes = []common.Hash{head}
tester.blocks = map[common.Hash]*types.Block{head: blocks[head]}
tester.lock.Unlock()
// Ensure that a block with a lower number than the threshold is discarded
tester.fetcher.Enqueue("lower", blocks[hashes[low]])
@ -641,8 +647,11 @@ func testDistantAnnouncementDiscarding(t *testing.T, protocol int) {
// Create a tester and simulate a head block being the middle of the above chain
tester := newTester()
tester.lock.Lock()
tester.hashes = []common.Hash{head}
tester.blocks = map[common.Hash]*types.Block{head: blocks[head]}
tester.lock.Unlock()
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
@ -687,14 +696,22 @@ func testInvalidNumberAnnouncement(t *testing.T, protocol int) {
tester.fetcher.Notify("bad", hashes[0], 2, time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher)
verifyImportEvent(t, imported, false)
if !tester.drops["bad"] {
tester.lock.RLock()
dropped := tester.drops["bad"]
tester.lock.RUnlock()
if !dropped {
t.Fatalf("peer with invalid numbered announcement not dropped")
}
// Make sure a good announcement passes without a drop
tester.fetcher.Notify("good", hashes[0], 1, time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher)
verifyImportEvent(t, imported, true)
if tester.drops["good"] {
tester.lock.RLock()
dropped = tester.drops["good"]
tester.lock.RUnlock()
if dropped {
t.Fatalf("peer with valid numbered announcement dropped")
}
verifyImportDone(t, imported)
@ -752,9 +769,15 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
// Create a tester with instrumented import hooks
tester := newTester()
imported := make(chan *types.Block)
imported, announces := make(chan *types.Block), int32(0)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
tester.fetcher.announceChangeHook = func(hash common.Hash, added bool) {
if added {
atomic.AddInt32(&announces, 1)
} else {
atomic.AddInt32(&announces, -1)
}
}
// Create a valid chain and an infinite junk chain
targetBlocks := hashLimit + 2*maxQueueDist
hashes, blocks := makeChain(targetBlocks, 0, genesis)
@ -782,8 +805,8 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
tester.fetcher.Notify("attacker", attack[i], 1 /* don't distance drop */, time.Now(), nil, attackerHeaderFetcher, attackerBodyFetcher)
}
}
if len(tester.fetcher.announced) != hashLimit+maxQueueDist {
t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist)
if count := atomic.LoadInt32(&announces); count != hashLimit+maxQueueDist {
t.Fatalf("queued announce count mismatch: have %d, want %d", count, hashLimit+maxQueueDist)
}
// Wait for fetches to complete
verifyImportCount(t, imported, maxQueueDist)
@ -807,9 +830,15 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) {
// Create a tester with instrumented import hooks
tester := newTester()
imported := make(chan *types.Block)
imported, enqueued := make(chan *types.Block), int32(0)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
tester.fetcher.queueChangeHook = func(hash common.Hash, added bool) {
if added {
atomic.AddInt32(&enqueued, 1)
} else {
atomic.AddInt32(&enqueued, -1)
}
}
// Create a valid chain and a batch of dangling (but in range) blocks
targetBlocks := hashLimit + 2*maxQueueDist
hashes, blocks := makeChain(targetBlocks, 0, genesis)
@ -825,7 +854,7 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) {
tester.fetcher.Enqueue("attacker", block)
}
time.Sleep(200 * time.Millisecond)
if queued := tester.fetcher.queue.Size(); queued != blockLimit {
if queued := atomic.LoadInt32(&enqueued); queued != blockLimit {
t.Fatalf("queued block count mismatch: have %d, want %d", queued, blockLimit)
}
// Queue up a batch of valid blocks, and check that a new peer is allowed to do so
@ -833,7 +862,7 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) {
tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-3-i]])
}
time.Sleep(100 * time.Millisecond)
if queued := tester.fetcher.queue.Size(); queued != blockLimit+maxQueueDist-1 {
if queued := atomic.LoadInt32(&enqueued); queued != blockLimit+maxQueueDist-1 {
t.Fatalf("queued block count mismatch: have %d, want %d", queued, blockLimit+maxQueueDist-1)
}
// Insert the missing piece (and sanity check the import)

2
eth/handler.go
View file

@ -103,7 +103,7 @@ func NewProtocolManager(fastSync bool, networkId int, mux *event.TypeMux, txpool
if fastSync && version < eth63 {
continue
}
// Compatible, initialize the sub-protocol
// Compatible; initialise the sub-protocol
version := version // Closure for the run
manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{
Name: "eth",

2
eth/metrics.go
View file

@ -101,7 +101,7 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter
case rw.version >= eth62 && msg.Code == BlockHeadersMsg:
packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter
packets, traffic = reqHeaderInPacketsMeter, reqHeaderInTrafficMeter
case rw.version >= eth62 && msg.Code == BlockBodiesMsg:
packets, traffic = reqBodyInPacketsMeter, reqBodyInTrafficMeter

2
trie/sync.go
View file

@ -31,7 +31,7 @@ type request struct {
object *node // Target node to populate with retrieved data (hashnode originally)
parents []*request // Parent state nodes referencing this entry (notify all upon completion)
depth int // Depth level within the trie the node is located to prioritize DFS
depth int // Depth level within the trie the node is located to prioritise DFS
deps int // Number of dependencies before allowed to commit this node
callback TrieSyncLeafCallback // Callback to invoke if a leaf node it reached on this branch