aboutsummaryrefslogtreecommitdiffstats
path: root/eth/fetcher
diff options
context:
space:
mode:
authorPéter Szilágyi <peterke@gmail.com>2015-06-18 23:00:19 +0800
committerPéter Szilágyi <peterke@gmail.com>2015-06-18 23:00:19 +0800
commitb91b581b80ec99dfa07b7206104faa919210fc4f (patch)
treedb9e452976af523e192c454148956cb15872fe9f /eth/fetcher
parent629705ad53ceac1bcc5f32f2d7ef51d7626aefef (diff)
downloaddexon-b91b581b80ec99dfa07b7206104faa919210fc4f.tar
dexon-b91b581b80ec99dfa07b7206104faa919210fc4f.tar.gz
dexon-b91b581b80ec99dfa07b7206104faa919210fc4f.tar.bz2
dexon-b91b581b80ec99dfa07b7206104faa919210fc4f.tar.lz
dexon-b91b581b80ec99dfa07b7206104faa919210fc4f.tar.xz
dexon-b91b581b80ec99dfa07b7206104faa919210fc4f.tar.zst
dexon-b91b581b80ec99dfa07b7206104faa919210fc4f.zip
eth, eth/fetcher: propagate after header verify, announce only on insert
Diffstat (limited to 'eth/fetcher')
-rw-r--r--eth/fetcher/fetcher.go43
-rw-r--r--eth/fetcher/fetcher_test.go18
2 files changed, 39 insertions, 22 deletions
diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go
index c96471554..d5ff5d77e 100644
--- a/eth/fetcher/fetcher.go
+++ b/eth/fetcher/fetcher.go
@@ -23,14 +23,17 @@ var (
errTerminated = errors.New("terminated")
)
-// hashCheckFn is a callback type for verifying a hash's presence in the local chain.
-type hashCheckFn func(common.Hash) bool
+// blockRetrievalFn is a callback type for retrieving a block from the local chain.
+type blockRetrievalFn func(common.Hash) *types.Block
// blockRequesterFn is a callback type for sending a block retrieval request.
type blockRequesterFn func([]common.Hash) error
+// blockValidatorFn is a callback type to verify a block's header for fast propagation.
+type blockValidatorFn func(block *types.Block, parent *types.Block) error
+
// blockBroadcasterFn is a callback type for broadcasting a block to connected peers.
-type blockBroadcasterFn func(block *types.Block)
+type blockBroadcasterFn func(block *types.Block, propagate bool)
// chainHeightFn is a callback type to retrieve the current chain height.
type chainHeightFn func() uint64
@@ -76,7 +79,8 @@ type Fetcher struct {
queued map[common.Hash]struct{} // Presence set of already queued blocks (to dedup imports)
// Callbacks
- hasBlock hashCheckFn // Checks if a block is present in the chain
+ getBlock blockRetrievalFn // Retrieves a block from the local chain
+ validateBlock blockValidatorFn // Checks if a block's headers have a valid proof of work
broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers
chainHeight chainHeightFn // Retrieves the current chain's height
insertChain chainInsertFn // Injects a batch of blocks into the chain
@@ -84,7 +88,7 @@ type Fetcher struct {
}
// New creates a block fetcher to retrieve blocks based on hash announcements.
-func New(hasBlock hashCheckFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher {
+func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher {
return &Fetcher{
notify: make(chan *announce),
inject: make(chan *inject),
@@ -95,7 +99,8 @@ func New(hasBlock hashCheckFn, broadcastBlock blockBroadcasterFn, chainHeight ch
fetching: make(map[common.Hash]*announce),
queue: prque.New(),
queued: make(map[common.Hash]struct{}),
- hasBlock: hasBlock,
+ getBlock: getBlock,
+ validateBlock: validateBlock,
broadcastBlock: broadcastBlock,
chainHeight: chainHeight,
insertChain: insertChain,
@@ -197,7 +202,7 @@ func (f *Fetcher) loop() {
break
}
// Otherwise if fresh and still unknown, try and import
- if number <= height || f.hasBlock(op.block.Hash()) {
+ if number <= height || f.getBlock(op.block.Hash()) != nil {
continue
}
f.insert(op.origin, op.block)
@@ -235,7 +240,7 @@ func (f *Fetcher) loop() {
for hash, announces := range f.announced {
if time.Since(announces[0].time) > arriveTimeout {
announce := announces[rand.Intn(len(announces))]
- if !f.hasBlock(hash) {
+ if f.getBlock(hash) == nil {
request[announce.origin] = append(request[announce.origin], hash)
f.fetching[hash] = announce
}
@@ -265,7 +270,7 @@ func (f *Fetcher) loop() {
// Filter explicitly requested blocks from hash announcements
if _, ok := f.fetching[hash]; ok {
// Discard if already imported by other means
- if !f.hasBlock(hash) {
+ if f.getBlock(hash) == nil {
explicit = append(explicit, block)
} else {
delete(f.fetching, hash)
@@ -313,7 +318,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) {
// Discard any past or too distant blocks
if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist <= 0 || dist > maxQueueDist {
- glog.Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist)
+ glog.V(logger.Detail).Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist)
return
}
// Schedule the block for future importing
@@ -321,7 +326,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) {
f.queued[hash] = struct{}{}
f.queue.Push(&inject{origin: peer, block: block}, -float32(block.NumberU64()))
- if glog.V(logger.Detail) {
+ if glog.V(logger.Debug) {
glog.Infof("Peer %s: queued block #%d [%x], total %v", peer, block.NumberU64(), hash.Bytes()[:4], f.queue.Size())
}
}
@@ -339,16 +344,24 @@ func (f *Fetcher) insert(peer string, block *types.Block) {
defer func() { f.done <- hash }()
// If the parent's unknown, abort insertion
- if !f.hasBlock(block.ParentHash()) {
+ parent := f.getBlock(block.ParentHash())
+ if parent == nil {
+ return
+ }
+ // Quickly validate the header and propagate the block if it passes
+ if err := f.validateBlock(block, parent); err != nil {
+ glog.V(logger.Debug).Infof("Peer %s: block #%d [%x] verification failed: %v", peer, block.NumberU64(), hash[:4], err)
+ f.dropPeer(peer)
return
}
+ go f.broadcastBlock(block, true)
+
// Run the actual import and log any issues
if _, err := f.insertChain(types.Blocks{block}); err != nil {
- glog.V(logger.Detail).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err)
- f.dropPeer(peer)
+ glog.V(logger.Warn).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err)
return
}
// If import succeeded, broadcast the block
- go f.broadcastBlock(block)
+ go f.broadcastBlock(block, false)
}()
}
diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go
index 3e8df1804..cdf875c5c 100644
--- a/eth/fetcher/fetcher_test.go
+++ b/eth/fetcher/fetcher_test.go
@@ -80,23 +80,27 @@ func newTester() *fetcherTester {
hashes: []common.Hash{knownHash},
blocks: map[common.Hash]*types.Block{knownHash: genesis},
}
- tester.fetcher = New(tester.hasBlock, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer)
+ tester.fetcher = New(tester.getBlock, tester.verifyBlock, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer)
tester.fetcher.Start()
return tester
}
-// hasBlock checks if a block is pres ent in the testers canonical chain.
-func (f *fetcherTester) hasBlock(hash common.Hash) bool {
+// getBlock retrieves a block from the tester's block chain.
+func (f *fetcherTester) getBlock(hash common.Hash) *types.Block {
f.lock.RLock()
defer f.lock.RUnlock()
- _, ok := f.blocks[hash]
- return ok
+ return f.blocks[hash]
+}
+
+// verifyBlock is a nop placeholder for the block header verification.
+func (f *fetcherTester) verifyBlock(block *types.Block, parent *types.Block) error {
+ return nil
}
// broadcastBlock is a nop placeholder for the block broadcasting.
-func (f *fetcherTester) broadcastBlock(block *types.Block) {
+func (f *fetcherTester) broadcastBlock(block *types.Block, propagate bool) {
}
// chainHeight retrieves the current height (block number) of the chain.
@@ -257,7 +261,7 @@ func TestPendingDeduplication(t *testing.T) {
return nil
}
// Announce the same block many times until it's fetched (wait for any pending ops)
- for !tester.hasBlock(hashes[0]) {
+ for tester.getBlock(hashes[0]) == nil {
tester.fetcher.Notify("repeater", hashes[0], time.Now().Add(-arriveTimeout), wrapper)
time.Sleep(time.Millisecond)
}