From 5ec6ecc511dc861c0395335f06883bc9ac650e4d Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Wed, 17 Jun 2015 18:25:23 +0300 Subject: eth, eth/fetcher: move propagated block import into fetcher --- eth/fetcher/fetcher.go | 55 ++++++++++++++++++++++++++-------------- eth/fetcher/fetcher_test.go | 62 ++++++++++++++++++++++++++------------------- 2 files changed, 72 insertions(+), 45 deletions(-) (limited to 'eth/fetcher') diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index a70fcbeed..c96471554 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -29,12 +29,18 @@ type hashCheckFn func(common.Hash) bool // blockRequesterFn is a callback type for sending a block retrieval request. 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 +// blockBroadcasterFn is a callback type for broadcasting a block to connected peers. +type blockBroadcasterFn func(block *types.Block) // chainHeightFn is a callback type to retrieve the current chain height. type chainHeightFn func() uint64 +// chainInsertFn is a callback type to insert a batch of blocks into the local chain. +type chainInsertFn func(types.Blocks) (int, error) + +// peerDropFn is a callback type for dropping a peer detected as malicious. +type peerDropFn func(id string) + // announce is the hash notification of the availability of a new block in the // network. type announce struct { @@ -70,26 +76,30 @@ 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 - importBlock blockImporterFn // Injects a block from an origin peer into the chain - chainHeight chainHeightFn // Retrieves the current chain's height + hasBlock hashCheckFn // Checks if a block is present in the chain + 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 + dropPeer peerDropFn // Drops a peer for misbehaving } // New creates a block fetcher to retrieve blocks based on hash announcements. -func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHeightFn) *Fetcher { +func New(hasBlock hashCheckFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { return &Fetcher{ - notify: make(chan *announce), - inject: make(chan *inject), - filter: make(chan chan []*types.Block), - done: make(chan common.Hash), - quit: make(chan struct{}), - announced: make(map[common.Hash][]*announce), - fetching: make(map[common.Hash]*announce), - queue: prque.New(), - queued: make(map[common.Hash]struct{}), - hasBlock: hasBlock, - importBlock: importBlock, - chainHeight: chainHeight, + notify: make(chan *announce), + inject: make(chan *inject), + filter: make(chan chan []*types.Block), + done: make(chan common.Hash), + quit: make(chan struct{}), + announced: make(map[common.Hash][]*announce), + fetching: make(map[common.Hash]*announce), + queue: prque.New(), + queued: make(map[common.Hash]struct{}), + hasBlock: hasBlock, + broadcastBlock: broadcastBlock, + chainHeight: chainHeight, + insertChain: insertChain, + dropPeer: dropPeer, } } @@ -328,10 +338,17 @@ func (f *Fetcher) insert(peer string, block *types.Block) { go func() { defer func() { f.done <- hash }() + // If the parent's unknown, abort insertion + if !f.hasBlock(block.ParentHash()) { + return + } // Run the actual import and log any issues - if err := f.importBlock(peer, block); err != nil { + 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) return } + // If import succeeded, broadcast the block + go f.broadcastBlock(block) }() } diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 4c6a1bf6a..3e8df1804 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -80,7 +80,7 @@ func newTester() *fetcherTester { hashes: []common.Hash{knownHash}, blocks: map[common.Hash]*types.Block{knownHash: genesis}, } - tester.fetcher = New(tester.hasBlock, tester.importBlock, tester.chainHeight) + tester.fetcher = New(tester.hasBlock, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer) tester.fetcher.Start() return tester @@ -95,23 +95,8 @@ func (f *fetcherTester) hasBlock(hash common.Hash) bool { return ok } -// importBlock injects a new blocks into the simulated chain. -func (f *fetcherTester) importBlock(peer string, block *types.Block) error { - f.lock.Lock() - defer f.lock.Unlock() - - // Make sure the parent in known - if _, ok := f.blocks[block.ParentHash()]; !ok { - return errors.New("unknown parent") - } - // Discard any new blocks if the same height already exists - if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() { - return nil - } - // Otherwise build our current chain - f.hashes = append(f.hashes, block.Hash()) - f.blocks[block.Hash()] = block - return nil +// broadcastBlock is a nop placeholder for the block broadcasting. +func (f *fetcherTester) broadcastBlock(block *types.Block) { } // chainHeight retrieves the current height (block number) of the chain. @@ -122,6 +107,31 @@ func (f *fetcherTester) chainHeight() uint64 { return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() } +// insertChain injects a new blocks into the simulated chain. +func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) { + f.lock.Lock() + defer f.lock.Unlock() + + for i, block := range blocks { + // Make sure the parent in known + if _, ok := f.blocks[block.ParentHash()]; !ok { + return i, errors.New("unknown parent") + } + // Discard any new blocks if the same height already exists + if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() { + return i, nil + } + // Otherwise build our current chain + f.hashes = append(f.hashes, block.Hash()) + f.blocks[block.Hash()] = block + } + return 0, nil +} + +// dropPeer is a nop placeholder for the peer removal. +func (f *fetcherTester) dropPeer(peer string) { +} + // 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 @@ -330,9 +340,9 @@ func TestImportDeduplication(t *testing.T) { 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) + tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) { + atomic.AddUint32(&counter, uint32(len(blocks))) + return tester.insertChain(blocks) } // Announce the duplicating block, wait for retrieval, and also propagate directly tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher) @@ -400,18 +410,18 @@ func TestCompetingImports(t *testing.T) { first := int32(1) height := uint64(1) - tester.fetcher.importBlock = func(peer string, block *types.Block) error { + tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) { // Check for any phase reordering - if prev := atomic.LoadUint64(&height); block.NumberU64() < prev { - t.Errorf("phase reversal: have %v, want %v", block.NumberU64(), prev) + if prev := atomic.LoadUint64(&height); blocks[0].NumberU64() < prev { + t.Errorf("phase reversal: have %v, want %v", blocks[0].NumberU64(), prev) } - atomic.StoreUint64(&height, block.NumberU64()) + atomic.StoreUint64(&height, blocks[0].NumberU64()) // Sleep a bit on the first import not to race with the enqueues if atomic.CompareAndSwapInt32(&first, 1, 0) { time.Sleep(50 * time.Millisecond) } - return tester.importBlock(peer, block) + return tester.insertChain(blocks) } // Queue up everything but with a missing link for i := 0; i < len(hashesA)-2; i++ { -- cgit