diff --git a/eth/fetcher/block_fetcher.go b/eth/fetcher/block_fetcher.go index 05e5833ce2..4819e9eab5 100644 --- a/eth/fetcher/block_fetcher.go +++ b/eth/fetcher/block_fetcher.go @@ -33,10 +33,12 @@ import ( ) const ( - lightTimeout = time.Millisecond // Time allowance before an announced header is explicitly requested - arriveTimeout = 500 * time.Millisecond // Time allowance before an announced block/transaction is explicitly requested - gatherSlack = 100 * time.Millisecond // Interval used to collate almost-expired announces with fetches - fetchTimeout = 5 * time.Second // Maximum allotted time to return an explicitly requested block/transaction + lightTimeout = time.Millisecond // Time allowance before an announced header is explicitly requested + arriveTimeout = 500 * time.Millisecond // Time allowance before an announced block/transaction is explicitly requested + gatherSlack = 100 * time.Millisecond // Interval used to collate almost-expired announces with fetches + fetchTimeout = 5 * time.Second // Maximum allotted time to return an explicitly requested block/transaction + reQueueBlockTimeout = 500 * time.Millisecond // Time allowance before blocks are requeued for import + ) const ( @@ -167,6 +169,8 @@ type BlockFetcher struct { done chan common.Hash quit chan struct{} + requeue chan *blockOrHeaderInject + // Announce states announces map[string]int // Per peer blockAnnounce counts to prevent memory exhaustion announced map[common.Hash][]*blockAnnounce // Announced blocks, scheduled for fetching @@ -207,6 +211,7 @@ func NewBlockFetcher(light bool, getHeader HeaderRetrievalFn, getBlock blockRetr bodyFilter: make(chan chan *bodyFilterTask), done: make(chan common.Hash), quit: make(chan struct{}), + requeue: make(chan *blockOrHeaderInject), announces: make(map[string]int), announced: make(map[common.Hash][]*blockAnnounce), fetching: make(map[common.Hash]*blockAnnounce), @@ -371,9 +376,9 @@ func (f *BlockFetcher) loop() { continue } if f.light { - f.importHeaders(op.origin, op.header) + f.importHeaders(op) } else { - f.importBlocks(op.origin, op.block) + f.importBlocks(op) } } // Wait for an outside event to occur @@ -416,6 +421,21 @@ func (f *BlockFetcher) loop() { f.rescheduleFetch(fetchTimer) } + case op := <-f.requeue: + // Re-queue blocks that have not been written due to fork block competition + number := int64(0) + hash := "" + if op.header != nil { + number = op.header.Number.Int64() + hash = op.header.Hash().String() + } else if op.block != nil { + number = op.block.Number().Int64() + hash = op.block.Hash().String() + } + + log.Info("Re-queue blocks", "number", number, "hash", hash) + f.enqueue(op.origin, op.header, op.block) + case op := <-f.inject: // A direct block insertion was requested, try and fill any pending gaps blockBroadcastInMeter.Mark(1) @@ -751,7 +771,9 @@ func (f *BlockFetcher) enqueue(peer string, header *types.Header, block *types.B // importHeaders spawns a new goroutine to run a header insertion into the chain. // If the header's number is at the same height as the current import phase, it // updates the phase states accordingly. -func (f *BlockFetcher) importHeaders(peer string, header *types.Header) { +func (f *BlockFetcher) importHeaders(op *blockOrHeaderInject) { + peer := op.origin + header := op.header hash := header.Hash() log.Debug("Importing propagated header", "peer", peer, "number", header.Number, "hash", hash) @@ -761,6 +783,8 @@ func (f *BlockFetcher) importHeaders(peer string, header *types.Header) { parent := f.getHeader(header.ParentHash) if parent == nil { log.Debug("Unknown parent of propagated header", "peer", peer, "number", header.Number, "hash", hash, "parent", header.ParentHash) + time.Sleep(reQueueBlockTimeout) + f.requeue <- op return } // Validate the header and if something went wrong, drop the peer @@ -784,7 +808,9 @@ func (f *BlockFetcher) importHeaders(peer string, header *types.Header) { // importBlocks spawns a new goroutine to run a block insertion into the chain. If the // block's number is at the same height as the current import phase, it updates // the phase states accordingly. -func (f *BlockFetcher) importBlocks(peer string, block *types.Block) { +func (f *BlockFetcher) importBlocks(op *blockOrHeaderInject) { + peer := op.origin + block := op.block hash := block.Hash() // Run the import on a new thread @@ -796,6 +822,8 @@ func (f *BlockFetcher) importBlocks(peer string, block *types.Block) { parent := f.getBlock(block.ParentHash()) if parent == nil { log.Debug("Unknown parent of propagated block", "peer", peer, "number", block.Number(), "hash", hash, "parent", block.ParentHash()) + time.Sleep(reQueueBlockTimeout) + f.requeue <- op return } // Quickly validate the header and propagate the block if it passes