aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--eth/fetcher/fetcher.go60
-rw-r--r--eth/fetcher/fetcher_test.go36
2 files changed, 71 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)
}
}
}
diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go
index 7c975841c..e11a211a1 100644
--- a/eth/fetcher/fetcher_test.go
+++ b/eth/fetcher/fetcher_test.go
@@ -299,3 +299,39 @@ func TestQueueGapFill(t *testing.T) {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
}
}
+
+// Tests that blocks arriving from various sources (multiple propagations, hash
+// announces, etc) do not get scheduled for import multiple times.
+func TestImportDeduplication(t *testing.T) {
+ // Create two blocks to import (one for duplication, the other for stalling)
+ hashes := createHashes(2, knownHash)
+ blocks := createBlocksFromHashes(hashes)
+
+ // Create the tester and wrap the importer with a counter
+ tester := newTester()
+ fetcher := tester.makeFetcher(blocks)
+
+ counter := uint32(0)
+ tester.fetcher.importBlock = func(peer string, block *types.Block) error {
+ atomic.AddUint32(&counter, 1)
+ return tester.importBlock(peer, block)
+ }
+ // Announce the duplicating block, wait for retrieval, and also propagate directly
+ tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher)
+ time.Sleep(50 * time.Millisecond)
+
+ tester.fetcher.Enqueue("valid", blocks[hashes[0]])
+ tester.fetcher.Enqueue("valid", blocks[hashes[0]])
+ tester.fetcher.Enqueue("valid", blocks[hashes[0]])
+
+ // Fill the missing block directly as if propagated, and check import uniqueness
+ tester.fetcher.Enqueue("valid", blocks[hashes[1]])
+ time.Sleep(50 * time.Millisecond)
+
+ if imported := len(tester.ownBlocks); imported != 3 {
+ t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 3)
+ }
+ if counter != 2 {
+ t.Fatalf("import invocation count mismatch: have %v, want %v", counter, 2)
+ }
+}