aboutsummaryrefslogtreecommitdiffstats
path: root/eth/fetcher/fetcher.go
diff options
context:
space:
mode:
Diffstat (limited to 'eth/fetcher/fetcher.go')
-rw-r--r--eth/fetcher/fetcher.go60
1 files changed, 35 insertions, 25 deletions
diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go
index 34d368780..207bd9323 100644
--- a/eth/fetcher/fetcher.go
+++ b/eth/fetcher/fetcher.go
@@ -153,11 +153,28 @@ func (f *Fetcher) Filter(blocks types.Blocks) types.Blocks {
func (f *Fetcher) loop() {
announced := make(map[common.Hash][]*announce)
fetching := make(map[common.Hash]*announce)
- queued := prque.New()
+
+ // Create the priority queue and a matching presence set
+ queue := prque.New()
+ queued := make(map[common.Hash]struct{})
+ enqueue := func(peer string, block *types.Block) {
+ // Make sure the block isn't in some weird place
+ if f.chainHeight()+maxQueueDist < block.NumberU64() {
+ return
+ }
+ // If not, schedule the block for future import
+ hash := block.Hash()
+ if _, ok := queued[hash]; !ok {
+ queued[hash] = struct{}{}
+ queue.Push(&inject{origin: peer, block: block}, -float32(block.NumberU64()))
+
+ glog.V(logger.Detail).Infof("Peer %s: queued block %x, total %v", peer, hash.Bytes()[:4], queue.Size())
+ }
+ }
+ // Iterate the block fetching until a quit is requested
fetch := time.NewTimer(0)
done := make(chan common.Hash)
- // Iterate the block fetching until a quit is requested
for {
// Clean up any expired block fetches
for hash, announce := range fetching {
@@ -168,24 +185,26 @@ func (f *Fetcher) loop() {
}
// Import any queued blocks that could potentially fit
height := f.chainHeight()
- for !queued.Empty() {
- // Fetch the next block, and skip if already known
- op := queued.PopItem().(*inject)
- if f.hasBlock(op.block.Hash()) {
- continue
- }
- // If unknown, but too high up the chain, continue later
+ for !queue.Empty() {
+ // If too high up the chain, continue later
+ op := queue.PopItem().(*inject)
if number := op.block.NumberU64(); number > height+1 {
- queued.Push(op, -float32(op.block.NumberU64()))
+ queue.Push(op, -float32(op.block.NumberU64()))
break
}
+ // Otherwise if not known yet, try and import
+ hash := op.block.Hash()
+ delete(queued, hash)
+ if f.hasBlock(hash) {
+ continue
+ }
// Block may just fit, try to import it
- glog.V(logger.Debug).Infof("Peer %s: importing block %x", op.origin, op.block.Hash().Bytes()[:4])
+ glog.V(logger.Debug).Infof("Peer %s: importing block %x", op.origin, hash.Bytes()[:4])
go func() {
- defer func() { done <- op.block.Hash() }()
+ defer func() { done <- hash }()
if err := f.importBlock(op.origin, op.block); err != nil {
- glog.V(logger.Detail).Infof("Peer %s: block %x import failed: %v", op.origin, op.block.Hash().Bytes()[:4], err)
+ glog.V(logger.Detail).Infof("Peer %s: block %x import failed: %v", op.origin, hash.Bytes()[:4], err)
return
}
}()
@@ -210,8 +229,7 @@ func (f *Fetcher) loop() {
case op := <-f.insert:
// A direct block insertion was requested, try and fill any pending gaps
- queued.Push(op, -float32(op.block.NumberU64()))
- glog.V(logger.Detail).Infof("Peer %s: filled block %x, total %v", op.origin, op.block.Hash().Bytes()[:4], queued.Size())
+ enqueue(op.origin, op.block)
case hash := <-done:
// A pending import finished, remove all traces of the notification
@@ -281,17 +299,9 @@ func (f *Fetcher) loop() {
return
}
// Schedule the retrieved blocks for ordered import
- height := f.chainHeight()
for _, block := range explicit {
- // Skip any blocks too far into the future
- if height+maxQueueDist < block.NumberU64() {
- continue
- }
- // Otherwise if the announce is still pending, schedule
- hash := block.Hash()
- if announce := fetching[hash]; announce != nil {
- queued.Push(&inject{origin: announce.origin, block: block}, -float32(block.NumberU64()))
- glog.V(logger.Detail).Infof("Peer %s: scheduled block %x, total %v", announce.origin, hash[:4], queued.Size())
+ if announce := fetching[block.Hash()]; announce != nil {
+ enqueue(announce.origin, block)
}
}
}