mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-24 13:46:43 +00:00
core, eth, trie: fix issues raised by reviewers
This commit is contained in:
parent
4fc1e1a914
commit
515ecc62cb
10 changed files with 67 additions and 74 deletions
|
|
@ -178,21 +178,21 @@ func (self *BlockChain) loadLastState() error {
|
||||||
fastTd := self.GetTd(self.currentFastBlock.Hash())
|
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("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("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
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetHead rewind the local chain to a new head entity. In the case of headers,
|
// SetHead rewinds the local chain to a new head. In the case of headers, everything
|
||||||
// everything above the new head will be deleted and the new one set. In the case
|
// above the new head will be deleted and the new one set. In the case of blocks
|
||||||
// of blocks though, the head may be further rewound if block bodies are missing
|
// though, the head may be further rewound if block bodies are missing (non-archive
|
||||||
// (non-archive nodes after a fast sync).
|
// nodes after a fast sync).
|
||||||
func (bc *BlockChain) SetHead(head uint64) {
|
func (bc *BlockChain) SetHead(head uint64) {
|
||||||
bc.mu.Lock()
|
bc.mu.Lock()
|
||||||
defer bc.mu.Unlock()
|
defer bc.mu.Unlock()
|
||||||
|
|
||||||
// Figure out the highest known canonical assignment
|
// Figure out the highest known canonical headers and/or blocks
|
||||||
height := uint64(0)
|
height := uint64(0)
|
||||||
if bc.currentHeader != nil {
|
if bc.currentHeader != nil {
|
||||||
if hh := bc.currentHeader.Number.Uint64(); hh > height {
|
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
|
// FastSyncCommitHead sets the current head block to the one defined by the hash
|
||||||
// irrelevant what the chain contents were prior.
|
// irrelevant what the chain contents were prior.
|
||||||
func (self *BlockChain) FastSyncCommitHead(hash common.Hash) error {
|
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)
|
block := self.GetBlock(hash)
|
||||||
if block == nil {
|
if block == nil {
|
||||||
return fmt.Errorf("non existent block [%x…]", hash[:4])
|
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
|
// 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 {
|
func (self *BlockChain) CurrentHeader() *types.Header {
|
||||||
self.mu.RLock()
|
self.mu.RLock()
|
||||||
defer self.mu.RUnlock()
|
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
|
// 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 {
|
func (self *BlockChain) CurrentBlock() *types.Block {
|
||||||
self.mu.RLock()
|
self.mu.RLock()
|
||||||
defer self.mu.RUnlock()
|
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
|
// 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 {
|
func (self *BlockChain) CurrentFastBlock() *types.Block {
|
||||||
self.mu.RLock()
|
self.mu.RLock()
|
||||||
defer self.mu.RUnlock()
|
defer self.mu.RUnlock()
|
||||||
|
|
@ -353,7 +353,7 @@ func (bc *BlockChain) ResetWithGenesisBlock(genesis *types.Block) {
|
||||||
bc.mu.Lock()
|
bc.mu.Lock()
|
||||||
defer bc.mu.Unlock()
|
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 {
|
if err := WriteTd(bc.chainDb, genesis.Hash(), genesis.Difficulty()); err != nil {
|
||||||
glog.Fatalf("failed to write genesis block TD: %v", err)
|
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
|
// 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
|
// 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
|
// 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!
|
// Note, this function assumes that the `mu` mutex is held!
|
||||||
func (bc *BlockChain) insert(block *types.Block) {
|
func (bc *BlockChain) insert(block *types.Block) {
|
||||||
|
|
@ -632,10 +632,10 @@ const (
|
||||||
|
|
||||||
// writeHeader writes a header into the local chain, given that its parent is
|
// 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
|
// 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
|
// 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
|
// without the real blocks. Hence, writing headers directly should only be done
|
||||||
// in two scenarios: pure-header mode of operation (light clients), or properly
|
// in two scenarios: pure-header mode of operation (light clients), or properly
|
||||||
// separated header/block phases (non-archive clients).
|
// separated header/block phases (non-archive clients).
|
||||||
|
|
@ -685,10 +685,9 @@ func (self *BlockChain) writeHeader(header *types.Header) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// InsertHeaderChain will attempt to insert the given header chain in to the
|
// InsertHeaderChain attempts to insert the given header chain in to the local
|
||||||
// local chain, possibly creating a fork. If an error is returned, it will
|
// chain, possibly creating a reorg. If an error is returned, it will return the
|
||||||
// return the index number of the failing header as well an error describing
|
// index number of the failing header as well an error describing what went wrong.
|
||||||
// what went wrong.
|
|
||||||
//
|
//
|
||||||
// The verify parameter can be used to fine tune whether nonce verification
|
// 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
|
// should be done or not. The reason behind the optional check is because some
|
||||||
|
|
@ -773,10 +772,6 @@ func (self *BlockChain) InsertHeaderChain(chain []*types.Header, checkFreq int)
|
||||||
pending.Wait()
|
pending.Wait()
|
||||||
|
|
||||||
// If anything failed, report
|
// 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 {
|
if failed > 0 {
|
||||||
for i, err := range errs {
|
for i, err := range errs {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -927,10 +922,6 @@ func (self *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain
|
||||||
pending.Wait()
|
pending.Wait()
|
||||||
|
|
||||||
// If anything failed, report
|
// 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 {
|
if failed > 0 {
|
||||||
for i, err := range errs {
|
for i, err := range errs {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -938,6 +929,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
|
// Update the head fast sync block if better
|
||||||
self.mu.Lock()
|
self.mu.Lock()
|
||||||
head := blockChain[len(errs)-1]
|
head := blockChain[len(errs)-1]
|
||||||
|
|
|
||||||
|
|
@ -26,14 +26,13 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/trie"
|
"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
|
// unknown state hashes to retrieve, accepts node data associated with said hashes
|
||||||
// and reconstructs the state database step by step until all is done.
|
// and reconstructs the state database step by step until all is done.
|
||||||
type StateSync trie.TrieSync
|
type StateSync trie.TrieSync
|
||||||
|
|
||||||
// NewStateSync create a new state trie download scheduler.
|
// NewStateSync create a new state trie download scheduler.
|
||||||
func NewStateSync(root common.Hash, database ethdb.Database) *StateSync {
|
func NewStateSync(root common.Hash, database ethdb.Database) *StateSync {
|
||||||
// Pre-declare the result syncer t
|
|
||||||
var syncer *trie.TrieSync
|
var syncer *trie.TrieSync
|
||||||
|
|
||||||
callback := func(leaf []byte, parent common.Hash) error {
|
callback := func(leaf []byte, parent common.Hash) error {
|
||||||
|
|
|
||||||
|
|
@ -67,7 +67,7 @@ func (r *Receipt) DecodeRLP(s *rlp.Stream) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// RlpEncode implements common.RlpEncode required for SHA derivation.
|
// RlpEncode implements common.RlpEncode required for SHA3 derivation.
|
||||||
func (r *Receipt) RlpEncode() []byte {
|
func (r *Receipt) RlpEncode() []byte {
|
||||||
bytes, err := rlp.EncodeToBytes(r)
|
bytes, err := rlp.EncodeToBytes(r)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -82,7 +82,7 @@ func (r *Receipt) String() string {
|
||||||
}
|
}
|
||||||
|
|
||||||
// ReceiptForStorage is a wrapper around a Receipt that flattens and parses the
|
// 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
|
type ReceiptForStorage Receipt
|
||||||
|
|
||||||
// EncodeRLP implements rlp.Encoder, and flattens all content fields of a 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})
|
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
|
// DecodeRLP implements rlp.Decoder, and loads both consensus and implementation
|
||||||
// from an RLP stream.
|
// fields of a receipt from an RLP stream.
|
||||||
func (r *ReceiptForStorage) DecodeRLP(s *rlp.Stream) error {
|
func (r *ReceiptForStorage) DecodeRLP(s *rlp.Stream) error {
|
||||||
var receipt struct {
|
var receipt struct {
|
||||||
PostState []byte
|
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.
|
// Receipts is a wrapper around a Receipt array to implement types.DerivableList.
|
||||||
type Receipts []*Receipt
|
type Receipts []*Receipt
|
||||||
|
|
||||||
// RlpEncode implements common.RlpEncode required for SHA derivation.
|
// RlpEncode implements common.RlpEncode required for SHA3 derivation.
|
||||||
func (r Receipts) RlpEncode() []byte {
|
func (r Receipts) RlpEncode() []byte {
|
||||||
bytes, err := rlp.EncodeToBytes(r)
|
bytes, err := rlp.EncodeToBytes(r)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
|
|
@ -66,6 +66,6 @@ func (l *Log) String() string {
|
||||||
type Logs []*Log
|
type Logs []*Log
|
||||||
|
|
||||||
// LogForStorage is a wrapper around a Log that flattens and parses the entire
|
// 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).
|
// the rlp interface methods).
|
||||||
type LogForStorage Log
|
type LogForStorage Log
|
||||||
|
|
|
||||||
|
|
@ -844,7 +844,7 @@ func (d *Downloader) fetchBlocks61(from uint64) error {
|
||||||
|
|
||||||
for _, peer := range idles {
|
for _, peer := range idles {
|
||||||
// Short circuit if throttling activated
|
// Short circuit if throttling activated
|
||||||
if d.queue.ThrottleBlocks() {
|
if d.queue.ShouldThrottleBlocks() {
|
||||||
throttled = true
|
throttled = true
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
@ -1234,7 +1234,7 @@ func (d *Downloader) fetchBodies(from uint64) error {
|
||||||
setIdle = func(p *peer) { p.SetBlocksIdle() }
|
setIdle = func(p *peer) { p.SetBlocksIdle() }
|
||||||
)
|
)
|
||||||
err := d.fetchParts(errCancelBodyFetch, d.bodyCh, deliver, d.bodyWakeCh, expire,
|
err := d.fetchParts(errCancelBodyFetch, d.bodyCh, deliver, d.bodyWakeCh, expire,
|
||||||
d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ThrottleBlocks, d.queue.ReserveBodies,
|
d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ShouldThrottleBlocks, d.queue.ReserveBodies,
|
||||||
d.bodyFetchHook, fetch, d.queue.CancelBodies, capacity, getIdles, setIdle, "Body")
|
d.bodyFetchHook, fetch, d.queue.CancelBodies, capacity, getIdles, setIdle, "Body")
|
||||||
|
|
||||||
glog.V(logger.Debug).Infof("Block body download terminated: %v", err)
|
glog.V(logger.Debug).Infof("Block body download terminated: %v", err)
|
||||||
|
|
@ -1258,7 +1258,7 @@ func (d *Downloader) fetchReceipts(from uint64) error {
|
||||||
setIdle = func(p *peer) { p.SetReceiptsIdle() }
|
setIdle = func(p *peer) { p.SetReceiptsIdle() }
|
||||||
)
|
)
|
||||||
err := d.fetchParts(errCancelReceiptFetch, d.receiptCh, deliver, d.receiptWakeCh, expire,
|
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")
|
d.receiptFetchHook, fetch, d.queue.CancelReceipts, capacity, d.peers.ReceiptIdlePeers, setIdle, "Receipt")
|
||||||
|
|
||||||
glog.V(logger.Debug).Infof("Receipt download terminated: %v", err)
|
glog.V(logger.Debug).Infof("Receipt download terminated: %v", err)
|
||||||
|
|
|
||||||
|
|
@ -20,7 +20,7 @@ package downloader
|
||||||
type SyncMode int
|
type SyncMode int
|
||||||
|
|
||||||
const (
|
const (
|
||||||
FullSync SyncMode = iota // Synchronise the entire block-chain history from full blocks
|
FullSync SyncMode = iota // Synchronise the entire blockchain history from full blocks
|
||||||
FastSync // Quikcly download the headers, full sync only at the chain head
|
FastSync // Quickly download the headers, full sync only at the chain head
|
||||||
LightSync // Download only the headers and terminate afterwards
|
LightSync // Download only the headers and terminate afterwards
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -196,35 +196,35 @@ func (p *peer) FetchNodeData(request *fetchRequest) error {
|
||||||
|
|
||||||
// SetBlocksIdle sets the peer to idle, allowing it to execute new retrieval requests.
|
// 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,
|
// 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() {
|
func (p *peer) SetBlocksIdle() {
|
||||||
p.setIdle(p.blockStarted, blockSoftTTL, blockHardTTL, MaxBlockFetch, &p.blockCapacity, &p.blockIdle)
|
p.setIdle(p.blockStarted, blockSoftTTL, blockHardTTL, MaxBlockFetch, &p.blockCapacity, &p.blockIdle)
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetBodiesIdle sets the peer to idle, allowing it to execute new retrieval requests.
|
// 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,
|
// 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() {
|
func (p *peer) SetBodiesIdle() {
|
||||||
p.setIdle(p.blockStarted, bodySoftTTL, bodyHardTTL, MaxBlockFetch, &p.blockCapacity, &p.blockIdle)
|
p.setIdle(p.blockStarted, bodySoftTTL, bodyHardTTL, MaxBlockFetch, &p.blockCapacity, &p.blockIdle)
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetReceiptsIdle sets the peer to idle, allowing it to execute new retrieval requests.
|
// 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,
|
// 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() {
|
func (p *peer) SetReceiptsIdle() {
|
||||||
p.setIdle(p.receiptStarted, receiptSoftTTL, receiptHardTTL, MaxReceiptFetch, &p.receiptCapacity, &p.receiptIdle)
|
p.setIdle(p.receiptStarted, receiptSoftTTL, receiptHardTTL, MaxReceiptFetch, &p.receiptCapacity, &p.receiptIdle)
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetNodeDataIdle sets the peer to idle, allowing it to execute new retrieval
|
// 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
|
// 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() {
|
func (p *peer) SetNodeDataIdle() {
|
||||||
p.setIdle(p.stateStarted, stateSoftTTL, stateSoftTTL, MaxStateFetch, &p.stateCapacity, &p.stateIdle)
|
p.setIdle(p.stateStarted, stateSoftTTL, stateSoftTTL, MaxStateFetch, &p.stateCapacity, &p.stateIdle)
|
||||||
}
|
}
|
||||||
|
|
||||||
// setIdle sets the peer to idle, allowing it to execute new retrieval requests.
|
// 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,
|
// 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) {
|
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
|
// Update the peer's download allowance based on previous performance
|
||||||
scale := 2.0
|
scale := 2.0
|
||||||
|
|
|
||||||
|
|
@ -56,9 +56,8 @@ type fetchRequest struct {
|
||||||
Time time.Time // Time when the request was made
|
Time time.Time // Time when the request was made
|
||||||
}
|
}
|
||||||
|
|
||||||
// fetchResult is the assembly collecting partial results from potentially more
|
// fetchResult is a struct collecting partial results from data fetchers until
|
||||||
// than one fetcher routines, until all outstanding retrievals complete and the
|
// all outstanding pieces complete and the result as a whole can be processed.
|
||||||
// result as a whole can be processed.
|
|
||||||
type fetchResult struct {
|
type fetchResult struct {
|
||||||
Pending int // Number of data fetches still pending
|
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
|
receiptPendPool map[string]*fetchRequest // [eth/63] Currently pending receipt retrieval operations
|
||||||
receiptDonePool map[common.Hash]struct{} // [eth/63] Set of the completed receipt fetches
|
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
|
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
|
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
|
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
|
stateDatabase ethdb.Database // [eth/63] Trie database to populate during state reassembly
|
||||||
stateScheduler *state.StateSync // [eth/63] State trie synchronisation scheduler and integrator
|
stateScheduler *state.StateSync // [eth/63] State trie synchronisation scheduler and integrator
|
||||||
stateProcessors int32 // [eth/63] Number of currently running state processors
|
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
|
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
|
lock sync.RWMutex
|
||||||
}
|
}
|
||||||
|
|
@ -233,9 +232,9 @@ func (q *queue) Idle() bool {
|
||||||
return (queued + pending + cached) == 0
|
return (queued + pending + cached) == 0
|
||||||
}
|
}
|
||||||
|
|
||||||
// ThrottleBlocks checks if the download should be throttled (active block (body)
|
// ShouldThrottleBlocks checks if the download should be throttled (active block (body)
|
||||||
// fetches exceed block cache).
|
// fetches exceed block cache).
|
||||||
func (q *queue) ThrottleBlocks() bool {
|
func (q *queue) ShouldThrottleBlocks() bool {
|
||||||
q.lock.RLock()
|
q.lock.RLock()
|
||||||
defer q.lock.RUnlock()
|
defer q.lock.RUnlock()
|
||||||
|
|
||||||
|
|
@ -248,9 +247,9 @@ func (q *queue) ThrottleBlocks() bool {
|
||||||
return pending >= len(q.resultCache)-len(q.blockDonePool)
|
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).
|
// fetches exceed block cache).
|
||||||
func (q *queue) ThrottleReceipts() bool {
|
func (q *queue) ShouldThrottleReceipts() bool {
|
||||||
q.lock.RLock()
|
q.lock.RLock()
|
||||||
defer q.lock.RUnlock()
|
defer q.lock.RUnlock()
|
||||||
|
|
||||||
|
|
@ -269,7 +268,7 @@ func (q *queue) Schedule61(hashes []common.Hash, fifo bool) []common.Hash {
|
||||||
q.lock.Lock()
|
q.lock.Lock()
|
||||||
defer q.lock.Unlock()
|
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))
|
inserts := make([]common.Hash, 0, len(hashes))
|
||||||
for _, hash := range hashes {
|
for _, hash := range hashes {
|
||||||
// Skip anything we already have
|
// Skip anything we already have
|
||||||
|
|
@ -297,10 +296,10 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
|
||||||
q.lock.Lock()
|
q.lock.Lock()
|
||||||
defer q.lock.Unlock()
|
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))
|
inserts := make([]*types.Header, 0, len(headers))
|
||||||
for _, header := range 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()
|
hash := header.Hash()
|
||||||
if header.Number == nil || header.Number.Uint64() != from {
|
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)
|
glog.V(logger.Warn).Infof("Header #%v [%x] broke chain ordering, expected %d", header.Number, hash[:4], from)
|
||||||
|
|
@ -439,8 +438,8 @@ func (q *queue) reserveHashes(p *peer, count int, taskQueue *prque.Prque, taskGe
|
||||||
q.lock.Lock()
|
q.lock.Lock()
|
||||||
defer q.lock.Unlock()
|
defer q.lock.Unlock()
|
||||||
|
|
||||||
// Short circuit if the peer's already downloading something (sanity check not
|
// Short circuit if the peer's already downloading something (sanity check to
|
||||||
// to corrupt state)
|
// not corrupt state)
|
||||||
if _, ok := pendPool[p.id]; ok {
|
if _, ok := pendPool[p.id]; ok {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -492,27 +491,27 @@ func (q *queue) reserveHashes(p *peer, count int, taskQueue *prque.Prque, taskGe
|
||||||
// previously failed downloads. Beside the next batch of needed fetches, it also
|
// previously failed downloads. Beside the next batch of needed fetches, it also
|
||||||
// returns a flag whether empty blocks were queued requiring processing.
|
// returns a flag whether empty blocks were queued requiring processing.
|
||||||
func (q *queue) ReserveBodies(p *peer, count int) (*fetchRequest, bool, error) {
|
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 header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash
|
||||||
}
|
}
|
||||||
return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, noop)
|
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
|
// ReserveReceipts reserves a set of receipt fetches for the given peer, skipping
|
||||||
// any previously failed downloads. Beside the next batch of needed fetches, it
|
// any previously failed downloads. Beside the next batch of needed fetches, it
|
||||||
// also returns a flag whether empty receipts were queued requiring importing.
|
// also returns a flag whether empty receipts were queued requiring importing.
|
||||||
func (q *queue) ReserveReceipts(p *peer, count int) (*fetchRequest, bool, error) {
|
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 header.ReceiptHash == types.EmptyRootHash
|
||||||
}
|
}
|
||||||
return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, noop)
|
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,
|
// reserveHeaders reserves a set of data download operations for a given peer,
|
||||||
// skipping any previously failed ones. This method is a generic version used
|
// skipping any previously failed ones. This method is a generic version used
|
||||||
// by the individual special reservation functions.
|
// by the individual special reservation functions.
|
||||||
func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
|
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) {
|
pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, isNoop func(*types.Header) bool) (*fetchRequest, bool, error) {
|
||||||
q.lock.Lock()
|
q.lock.Lock()
|
||||||
defer q.lock.Unlock()
|
defer q.lock.Unlock()
|
||||||
|
|
||||||
|
|
@ -537,7 +536,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++ {
|
for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ {
|
||||||
header := taskQueue.PopItem().(*types.Header)
|
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))
|
index := int(header.Number.Int64() - int64(q.resultOffset))
|
||||||
if index >= len(q.resultCache) || index < 0 {
|
if index >= len(q.resultCache) || index < 0 {
|
||||||
return nil, false, errInvalidChain
|
return nil, false, errInvalidChain
|
||||||
|
|
@ -553,7 +552,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 this fetch task is a noop, skip this fetch operation
|
||||||
if noop(header) {
|
if isNoop(header) {
|
||||||
donePool[header.Hash()] = struct{}{}
|
donePool[header.Hash()] = struct{}{}
|
||||||
delete(taskPool, header.Hash())
|
delete(taskPool, header.Hash())
|
||||||
|
|
||||||
|
|
@ -562,7 +561,7 @@ func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*typ
|
||||||
progress = true
|
progress = true
|
||||||
continue
|
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()) {
|
if p.ignored.Has(header.Hash()) {
|
||||||
skip = append(skip, header)
|
skip = append(skip, header)
|
||||||
} else {
|
} else {
|
||||||
|
|
@ -655,25 +654,25 @@ func (q *queue) Revoke(peerId string) {
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExpireBlocks checks for in flight requests that exceeded a timeout allowance,
|
// 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 {
|
func (q *queue) ExpireBlocks(timeout time.Duration) []string {
|
||||||
return q.expire(timeout, q.blockPendPool, q.hashQueue, blockTimeoutMeter)
|
return q.expire(timeout, q.blockPendPool, q.hashQueue, blockTimeoutMeter)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExpireBodies checks for in flight block body requests that exceeded a timeout
|
// 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 {
|
func (q *queue) ExpireBodies(timeout time.Duration) []string {
|
||||||
return q.expire(timeout, q.blockPendPool, q.blockTaskQueue, bodyTimeoutMeter)
|
return q.expire(timeout, q.blockPendPool, q.blockTaskQueue, bodyTimeoutMeter)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExpireReceipts checks for in flight receipt requests that exceeded a timeout
|
// 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 {
|
func (q *queue) ExpireReceipts(timeout time.Duration) []string {
|
||||||
return q.expire(timeout, q.receiptPendPool, q.receiptTaskQueue, receiptTimeoutMeter)
|
return q.expire(timeout, q.receiptPendPool, q.receiptTaskQueue, receiptTimeoutMeter)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExpireNodeData checks for in flight node data requests that exceeded a timeout
|
// 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 {
|
func (q *queue) ExpireNodeData(timeout time.Duration) []string {
|
||||||
return q.expire(timeout, q.statePendPool, q.stateTaskQueue, stateTimeoutMeter)
|
return q.expire(timeout, q.statePendPool, q.stateTaskQueue, stateTimeoutMeter)
|
||||||
}
|
}
|
||||||
|
|
@ -876,7 +875,7 @@ func (q *queue) DeliverNodeData(id string, data [][]byte, callback func(error, i
|
||||||
stateReqTimer.UpdateSince(request.Time)
|
stateReqTimer.UpdateSince(request.Time)
|
||||||
delete(q.statePendPool, id)
|
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 {
|
if len(data) == 0 {
|
||||||
for hash, _ := range request.Hashes {
|
for hash, _ := range request.Hashes {
|
||||||
request.Peer.ignored.Add(hash)
|
request.Peer.ignored.Add(hash)
|
||||||
|
|
|
||||||
|
|
@ -103,7 +103,7 @@ func NewProtocolManager(fastSync bool, networkId int, mux *event.TypeMux, txpool
|
||||||
if fastSync && version < eth63 {
|
if fastSync && version < eth63 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// Compatible, initialize the sub-protocol
|
// Compatible; initialise the sub-protocol
|
||||||
version := version // Closure for the run
|
version := version // Closure for the run
|
||||||
manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{
|
manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{
|
||||||
Name: "eth",
|
Name: "eth",
|
||||||
|
|
|
||||||
|
|
@ -31,7 +31,7 @@ type request struct {
|
||||||
object *node // Target node to populate with retrieved data (hashnode originally)
|
object *node // Target node to populate with retrieved data (hashnode originally)
|
||||||
|
|
||||||
parents []*request // Parent state nodes referencing this entry (notify all upon completion)
|
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
|
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
|
callback TrieSyncLeafCallback // Callback to invoke if a leaf node it reached on this branch
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue