aboutsummaryrefslogtreecommitdiffstats
path: root/eth/fetcher
diff options
context:
space:
mode:
authorPéter Szilágyi <peterke@gmail.com>2015-06-16 22:39:04 +0800
committerPéter Szilágyi <peterke@gmail.com>2015-06-18 20:56:07 +0800
commit057bc237adf9ed0adf615a72cc1e92d9aed15d9e (patch)
tree20b852d95206d71a71ddb20489cd8f6068e4546b /eth/fetcher
parent8b64e041d6c41d76994510fdd8bb42ad7c5be5aa (diff)
downloaddexon-057bc237adf9ed0adf615a72cc1e92d9aed15d9e.tar
dexon-057bc237adf9ed0adf615a72cc1e92d9aed15d9e.tar.gz
dexon-057bc237adf9ed0adf615a72cc1e92d9aed15d9e.tar.bz2
dexon-057bc237adf9ed0adf615a72cc1e92d9aed15d9e.tar.lz
dexon-057bc237adf9ed0adf615a72cc1e92d9aed15d9e.tar.xz
dexon-057bc237adf9ed0adf615a72cc1e92d9aed15d9e.tar.zst
dexon-057bc237adf9ed0adf615a72cc1e92d9aed15d9e.zip
eth, eth/fetcher: use an import queue to store out of order blocks
Diffstat (limited to 'eth/fetcher')
-rw-r--r--eth/fetcher/fetcher.go78
-rw-r--r--eth/fetcher/fetcher_test.go35
2 files changed, 82 insertions, 31 deletions
diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go
index a8d1f2fb5..e8a9cc093 100644
--- a/eth/fetcher/fetcher.go
+++ b/eth/fetcher/fetcher.go
@@ -10,11 +10,13 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
+ "gopkg.in/karalabe/cookiejar.v2/collections/prque"
)
const (
arriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested
fetchTimeout = 5 * time.Second // Maximum alloted time to return an explicitly requested block
+ maxQueueDist = 256 // Maximum allowed distance from the chain head to queue
)
var (
@@ -30,6 +32,9 @@ type blockRequesterFn func([]common.Hash) error
// blockImporterFn is a callback type for trying to inject a block into the local chain.
type blockImporterFn func(peer string, block *types.Block) error
+// chainHeightFn is a callback type to retrieve the current chain height.
+type chainHeightFn func() uint64
+
// announce is the hash notification of the availability of a new block in the
// network.
type announce struct {
@@ -40,6 +45,12 @@ type announce struct {
fetch blockRequesterFn // Fetcher function to retrieve
}
+// inject represents a schedules import operation.
+type inject struct {
+ origin string
+ block *types.Block
+}
+
// Fetcher is responsible for accumulating block announcements from various peers
// and scheduling them for retrieval.
type Fetcher struct {
@@ -51,16 +62,18 @@ type Fetcher struct {
// Callbacks
hasBlock hashCheckFn // Checks if a block is present in the chain
importBlock blockImporterFn // Injects a block from an origin peer into the chain
+ chainHeight chainHeightFn // Retrieves the current chain's height
}
// New creates a block fetcher to retrieve blocks based on hash announcements.
-func New(hasBlock hashCheckFn, importBlock blockImporterFn) *Fetcher {
+func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHeightFn) *Fetcher {
return &Fetcher{
notify: make(chan *announce),
filter: make(chan chan []*types.Block),
quit: make(chan struct{}),
hasBlock: hasBlock,
importBlock: importBlock,
+ chainHeight: chainHeight,
}
}
@@ -124,6 +137,7 @@ 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()
fetch := time.NewTimer(0)
done := make(chan common.Hash)
@@ -136,6 +150,30 @@ func (f *Fetcher) loop() {
delete(fetching, hash)
}
}
+ // 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
+ if number := op.block.NumberU64(); number > height+1 {
+ queued.Push(op, -float32(op.block.NumberU64()))
+ break
+ }
+ // 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])
+ go func() {
+ defer func() { done <- op.block.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)
+ return
+ }
+ }()
+ }
// Wait for an outside event to occur
select {
case <-f.quit:
@@ -221,40 +259,20 @@ func (f *Fetcher) loop() {
case <-f.quit:
return
}
- // Create a closure with the retrieved blocks and origin peers
- peers := make([]string, 0, len(explicit))
- blocks = make([]*types.Block, 0, len(explicit))
+ // 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 {
- // Drop the block if it surely cannot fit
- if f.hasBlock(hash) || !f.hasBlock(block.ParentHash()) {
- // delete(fetching, hash) // if we drop, it will re-fetch it, wait for timeout?
- continue
- }
- // Otherwise accumulate for import
- peers = append(peers, announce.origin)
- blocks = append(blocks, block)
+ 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 any explicit fetches were replied to, import them
- if count := len(blocks); count > 0 {
- glog.V(logger.Debug).Infof("Importing %d explicitly fetched blocks", len(blocks))
- go func() {
- // Make sure all hashes are cleaned up
- for _, block := range blocks {
- hash := block.Hash()
- defer func() { done <- hash }()
- }
- // Try and actually import the blocks
- for i := 0; i < len(blocks); i++ {
- if err := f.importBlock(peers[i], blocks[i]); err != nil {
- glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err)
- return
- }
- }
- }()
- }
}
}
}
diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go
index 44e99f30f..cde4bb70a 100644
--- a/eth/fetcher/fetcher_test.go
+++ b/eth/fetcher/fetcher_test.go
@@ -77,7 +77,7 @@ func newTester() *fetcherTester {
ownHashes: []common.Hash{knownHash},
ownBlocks: map[common.Hash]*types.Block{knownHash: genesis},
}
- tester.fetcher = New(tester.hasBlock, tester.importBlock)
+ tester.fetcher = New(tester.hasBlock, tester.importBlock, tester.chainHeight)
tester.fetcher.Start()
return tester
@@ -99,6 +99,11 @@ func (f *fetcherTester) importBlock(peer string, block *types.Block) error {
return nil
}
+// chainHeight retrieves the current height (block number) of the chain.
+func (f *fetcherTester) chainHeight() uint64 {
+ return f.ownBlocks[f.ownHashes[len(f.ownHashes)-1]].NumberU64()
+}
+
// peerFetcher retrieves a fetcher associated with a simulated peer.
func (f *fetcherTester) makeFetcher(blocks map[common.Hash]*types.Block) blockRequesterFn {
// Copy all the blocks to ensure they are not tampered with
@@ -238,3 +243,31 @@ func TestPendingDeduplication(t *testing.T) {
t.Fatalf("retrieval count mismatch: have %v, want %v", counter, 1)
}
}
+
+// Tests that announcements retrieved in a random order are cached and eventually
+// imported when all the gaps are filled in.
+func TestRandomArrivalImport(t *testing.T) {
+ // Create a chain of blocks to import, and choose one to delay
+ targetBlocks := 24
+ hashes := createHashes(targetBlocks, knownHash)
+ blocks := createBlocksFromHashes(hashes)
+ skip := targetBlocks / 2
+
+ tester := newTester()
+ fetcher := tester.makeFetcher(blocks)
+
+ // Iteratively announce blocks, skipping one entry
+ for i := len(hashes) - 1; i >= 0; i-- {
+ if i != skip {
+ tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+ time.Sleep(50 * time.Millisecond)
+ }
+ }
+ // Finally announce the skipped entry and check full import
+ tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher)
+ time.Sleep(50 * time.Millisecond)
+
+ if imported := len(tester.ownBlocks); imported != targetBlocks+1 {
+ t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
+ }
+}