From 7c2af1c11722dc3175a98342c060afcfaf6a275f Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 16 Jun 2015 11:58:32 +0300 Subject: eth, eth/fetcher: separate notification sync mechanism --- eth/downloader/downloader.go | 1 + eth/fetcher/fetcher.go | 258 +++++++++++++++++++++++++++++++++++++++++++ eth/handler.go | 71 +++++------- eth/sync.go | 145 ++---------------------- 4 files changed, 293 insertions(+), 182 deletions(-) create mode 100644 eth/fetcher/fetcher.go diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 9866a5b46..18e5f50e8 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -1,3 +1,4 @@ +// Package downloader contains the manual full chain synchronisation. package downloader import ( diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go new file mode 100644 index 000000000..19c53048c --- /dev/null +++ b/eth/fetcher/fetcher.go @@ -0,0 +1,258 @@ +// Package fetcher contains the block announcement based synchonisation. +package fetcher + +import ( + "errors" + "math/rand" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" +) + +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 +) + +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 + +// 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 + +// announce is the hash notification of the availability of a new block in the +// network. +type announce struct { + hash common.Hash // Hash of the block being announced + time time.Time // Timestamp of the announcement + + origin string // Identifier of the peer originating the notification + fetch blockRequesterFn // Fetcher function to retrieve +} + +// Fetcher is responsible for accumulating block announcements from various peers +// and scheduling them for retrieval. +type Fetcher struct { + // Various event channels + notify chan *announce + filter chan chan []*types.Block + quit chan 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 +} + +// New creates a block fetcher to retrieve blocks based on hash announcements. +func New(hasBlock hashCheckFn, importBlock blockImporterFn) *Fetcher { + return &Fetcher{ + notify: make(chan *announce), + filter: make(chan chan []*types.Block), + quit: make(chan struct{}), + hasBlock: hasBlock, + importBlock: importBlock, + } +} + +// Start boots up the announcement based synchoniser, accepting and processing +// hash notifications and block fetches until termination requested. +func (f *Fetcher) Start() { + go f.loop() +} + +// Stop terminates the announcement based synchroniser, canceling all pending +// operations. +func (f *Fetcher) Stop() { + close(f.quit) +} + +// Notify announces the fetcher of the potential availability of a new block in +// the network. +func (f *Fetcher) Notify(peer string, hash common.Hash, time time.Time, fetcher blockRequesterFn) error { + block := &announce{ + hash: hash, + time: time, + origin: peer, + fetch: fetcher, + } + select { + case f.notify <- block: + return nil + case <-f.quit: + return errTerminated + } +} + +// Filter extracts all the blocks that were explicitly requested by the fetcher, +// returning those that should be handled differently. +func (f *Fetcher) Filter(blocks types.Blocks) types.Blocks { + // Send the filter channel to the fetcher + filter := make(chan []*types.Block) + + select { + case f.filter <- filter: + case <-f.quit: + return nil + } + // Request the filtering of the block list + select { + case filter <- blocks: + case <-f.quit: + return nil + } + // Retrieve the blocks remaining after filtering + select { + case blocks := <-filter: + return blocks + case <-f.quit: + return nil + } +} + +// Loop is the main fetcher loop, checking and processing various notification +// events. +func (f *Fetcher) loop() { + announced := make(map[common.Hash][]*announce) + fetching := make(map[common.Hash]*announce) + 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 { + if time.Since(announce.time) > fetchTimeout { + delete(announced, hash) + delete(fetching, hash) + } + } + // Wait for an outside event to occur + select { + case <-f.quit: + // Fetcher terminating, abort all operations + return + + case notification := <-f.notify: + // A block was announced, schedule if it's not yet downloading + glog.V(logger.Debug).Infof("Peer %s: scheduling %x", notification.origin, notification.hash[:4]) + if _, ok := fetching[notification.hash]; ok { + break + } + if len(announced) == 0 { + fetch.Reset(arriveTimeout) + } + announced[notification.hash] = append(announced[notification.hash], notification) + + case hash := <-done: + // A pending import finished, remove all traces of the notification + delete(announced, hash) + delete(fetching, hash) + + case <-fetch.C: + // At least one block's timer ran out, check for needing retrieval + request := make(map[string][]common.Hash) + + for hash, announces := range announced { + if time.Since(announces[0].time) > arriveTimeout { + announce := announces[rand.Intn(len(announces))] + if !f.hasBlock(hash) { + request[announce.origin] = append(request[announce.origin], hash) + fetching[hash] = announce + } + delete(announced, hash) + } + } + // Send out all block requests + for peer, hashes := range request { + glog.V(logger.Debug).Infof("Peer %s: explicitly fetching %d blocks", peer, len(hashes)) + go fetching[hashes[0]].fetch(hashes) + } + // Schedule the next fetch if blocks are still pending + if len(announced) > 0 { + nearest := time.Now() + for _, announces := range announced { + if nearest.Before(announces[0].time) { + nearest = announces[0].time + } + } + fetch.Reset(arriveTimeout + time.Since(nearest)) + } + + case filter := <-f.filter: + // Blocks arrived, extract any explicit fetches, return all else + var blocks types.Blocks + select { + case blocks = <-filter: + case <-f.quit: + return + } + + explicit, download := []*types.Block{}, []*types.Block{} + for _, block := range blocks { + hash := block.Hash() + + // Filter explicitly requested blocks from hash announcements + if _, ok := fetching[hash]; ok { + // Discard if already imported by other means + if !f.hasBlock(hash) { + explicit = append(explicit, block) + } else { + delete(fetching, hash) + } + } else { + download = append(download, block) + } + } + + select { + case filter <- download: + 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)) + for _, block := range explicit { + 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) + } + } + // 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/handler.go b/eth/handler.go index ec4f2d53a..99ac4ce68 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -7,6 +7,8 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/eth/fetcher" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" @@ -45,6 +47,7 @@ type ProtocolManager struct { txpool txPool chainman *core.ChainManager downloader *downloader.Downloader + fetcher *fetcher.Fetcher peers *peerSet SubProtocol p2p.Protocol @@ -54,11 +57,9 @@ type ProtocolManager struct { minedBlockSub event.Subscription // channels for fetcher, syncer, txsyncLoop - newPeerCh chan *peer - newHashCh chan []*blockAnnounce - newBlockCh chan chan []*types.Block - txsyncCh chan *txsync - quitSync chan struct{} + newPeerCh chan *peer + txsyncCh chan *txsync + quitSync chan struct{} // wait group is used for graceful shutdowns during downloading // and processing @@ -69,30 +70,33 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, chainman *core.ChainManager) *ProtocolManager { + // Create the protocol manager and initialize peer handlers manager := &ProtocolManager{ - eventMux: mux, - txpool: txpool, - chainman: chainman, - peers: newPeerSet(), - newPeerCh: make(chan *peer, 1), - newHashCh: make(chan []*blockAnnounce, 1), - newBlockCh: make(chan chan []*types.Block), - txsyncCh: make(chan *txsync), - quitSync: make(chan struct{}), + eventMux: mux, + txpool: txpool, + chainman: chainman, + peers: newPeerSet(), + newPeerCh: make(chan *peer, 1), + txsyncCh: make(chan *txsync), + quitSync: make(chan struct{}), } - manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) manager.SubProtocol = p2p.Protocol{ Name: "eth", Version: uint(protocolVersion), Length: ProtocolLength, Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { peer := manager.newPeer(protocolVersion, networkId, p, rw) - manager.newPeerCh <- peer - return manager.handle(peer) }, } + // Construct the different synchronisation mechanisms + manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) + + importer := func(peer string, block *types.Block) error { + return manager.importBlock(manager.peers.Peer(peer), block, nil) + } + manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer) return manager } @@ -126,7 +130,6 @@ func (pm *ProtocolManager) Start() { // start sync handlers go pm.syncer() - go pm.fetcher() go pm.txsyncLoop() } @@ -291,20 +294,9 @@ func (self *ProtocolManager) handleMsg(p *peer) error { glog.V(logger.Detail).Infoln("Decode error", err) blocks = nil } - // Filter out any explicitly requested blocks (cascading select to get blocking back to peer) - filter := make(chan []*types.Block) - select { - case <-self.quitSync: - case self.newBlockCh <- filter: - select { - case <-self.quitSync: - case filter <- blocks: - select { - case <-self.quitSync: - case blocks := <-filter: - self.downloader.DeliverBlocks(p.id, blocks) - } - } + // Filter out any explicitly requested blocks, deliver the rest to the downloader + if blocks := self.fetcher.Filter(blocks); len(blocks) > 0 { + self.downloader.DeliverBlocks(p.id, blocks) } case NewBlockHashesMsg: @@ -327,19 +319,8 @@ func (self *ProtocolManager) handleMsg(p *peer) error { unknown = append(unknown, hash) } } - announces := make([]*blockAnnounce, len(unknown)) - for i, hash := range unknown { - announces[i] = &blockAnnounce{ - hash: hash, - peer: p, - time: time.Now(), - } - } - if len(announces) > 0 { - select { - case self.newHashCh <- announces: - case <-self.quitSync: - } + for _, hash := range unknown { + self.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks) } case NewBlockMsg: diff --git a/eth/sync.go b/eth/sync.go index 751bc1a2a..82abb725f 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -12,11 +12,8 @@ import ( ) const ( - forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available - notifyCheckCycle = 100 * time.Millisecond // Time interval to allow hash notifies to fulfill before hard fetching - notifyArriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested - notifyFetchTimeout = 5 * time.Second // Maximum alloted time to return an explicitly requested block - minDesiredPeerCount = 5 // Amount of peers desired to start syncing + forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available + minDesiredPeerCount = 5 // Amount of peers desired to start syncing // This is the target size for the packs of transactions sent by txsyncLoop. // A pack can get larger than this if a single transactions exceeds this size. @@ -119,140 +116,15 @@ func (pm *ProtocolManager) txsyncLoop() { } } -// fetcher is responsible for collecting hash notifications, and periodically -// checking all unknown ones and individually fetching them. -func (pm *ProtocolManager) fetcher() { - announces := make(map[common.Hash][]*blockAnnounce) - request := make(map[*peer][]common.Hash) - pending := make(map[common.Hash]*blockAnnounce) - cycle := time.Tick(notifyCheckCycle) - done := make(chan common.Hash) - - // Iterate the block fetching until a quit is requested - for { - select { - case notifications := <-pm.newHashCh: - // A batch of hashes the notified, schedule them for retrieval - glog.V(logger.Debug).Infof("Scheduling %d hash announcements from %s", len(notifications), notifications[0].peer.id) - for _, announce := range notifications { - // Skip if it's already pending fetch - if _, ok := pending[announce.hash]; ok { - continue - } - // Otherwise queue up the peer as a potential source - announces[announce.hash] = append(announces[announce.hash], announce) - } - - case hash := <-done: - // A pending import finished, remove all traces - delete(pending, hash) - - case <-cycle: - // Clean up any expired block fetches - for hash, announce := range pending { - if time.Since(announce.time) > notifyFetchTimeout { - delete(pending, hash) - } - } - // Check if any notified blocks failed to arrive - for hash, all := range announces { - if time.Since(all[0].time) > notifyArriveTimeout { - announce := all[rand.Intn(len(all))] - if !pm.chainman.HasBlock(hash) { - request[announce.peer] = append(request[announce.peer], hash) - pending[hash] = announce - } - delete(announces, hash) - } - } - if len(request) == 0 { - break - } - // Send out all block requests - for peer, hashes := range request { - glog.V(logger.Debug).Infof("Explicitly fetching %d blocks from %s", len(hashes), peer.id) - go peer.requestBlocks(hashes) - } - request = make(map[*peer][]common.Hash) - - case filter := <-pm.newBlockCh: - // Blocks arrived, extract any explicit fetches, return all else - var blocks types.Blocks - select { - case blocks = <-filter: - case <-pm.quitSync: - return - } - - explicit, download := []*types.Block{}, []*types.Block{} - for _, block := range blocks { - hash := block.Hash() - - // Filter explicitly requested blocks from hash announcements - if _, ok := pending[hash]; ok { - // Discard if already imported by other means - if !pm.chainman.HasBlock(hash) { - explicit = append(explicit, block) - } else { - delete(pending, hash) - } - } else { - download = append(download, block) - } - } - - select { - case filter <- download: - case <-pm.quitSync: - return - } - // Create a closure with the retrieved blocks and origin peers - peers := make([]*peer, 0, len(explicit)) - blocks = make([]*types.Block, 0, len(explicit)) - for _, block := range explicit { - hash := block.Hash() - if announce := pending[hash]; announce != nil { - // Drop the block if it surely cannot fit - if pm.chainman.HasBlock(hash) || !pm.chainman.HasBlock(block.ParentHash()) { - // delete(pending, hash) // if we drop, it will re-fetch it, wait for timeout? - continue - } - // Otherwise accumulate for import - peers = append(peers, announce.peer) - blocks = append(blocks, block) - } - } - // 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 := pm.importBlock(peers[i], blocks[i], nil); err != nil { - glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err) - return - } - } - }() - } - - case <-pm.quitSync: - return - } - } -} - // syncer is responsible for periodically synchronising with the network, both -// downloading hashes and blocks as well as retrieving cached ones. +// downloading hashes and blocks as well as handling the announcement handler. func (pm *ProtocolManager) syncer() { - // Abort any pending syncs if we terminate + // Start and ensure cleanup of sync mechanisms + pm.fetcher.Start() + defer pm.fetcher.Stop() defer pm.downloader.Terminate() + // Wait for different events to fire synchronisation operations forceSync := time.Tick(forceSyncCycle) for { select { @@ -273,8 +145,7 @@ func (pm *ProtocolManager) syncer() { } } -// synchronise tries to sync up our local block chain with a remote peer, both -// adding various sanity checks as well as wrapping it with various log entries. +// synchronise tries to sync up our local block chain with a remote peer. func (pm *ProtocolManager) synchronise(peer *peer) { // Short circuit if no peers are available if peer == nil { -- cgit From 2a1b722d048d00401084c37a5ca12612f1dd5fcd Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 16 Jun 2015 14:02:43 +0300 Subject: eth/fetcher: fix timer reset bug, add initial tests --- eth/downloader/downloader_test.go | 8 +- eth/fetcher/fetcher.go | 8 +- eth/fetcher/fetcher_test.go | 202 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 212 insertions(+), 6 deletions(-) create mode 100644 eth/fetcher/fetcher_test.go diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 484cc3218..4fc4e1434 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -52,6 +52,8 @@ func copyBlock(block *types.Block) *types.Block { return createBlock(int(block.Number().Int64()), block.ParentHeaderHash, block.HeaderHash) } +// createBlocksFromHashes assembles a collection of blocks, each having a correct +// place in the given hash chain. func createBlocksFromHashes(hashes []common.Hash) map[common.Hash]*types.Block { blocks := make(map[common.Hash]*types.Block) for i := 0; i < len(hashes); i++ { @@ -64,6 +66,7 @@ func createBlocksFromHashes(hashes []common.Hash) map[common.Hash]*types.Block { return blocks } +// downloadTester is a test simulator for mocking out local block chain. type downloadTester struct { downloader *Downloader @@ -75,6 +78,7 @@ type downloadTester struct { maxHashFetch int // Overrides the maximum number of retrieved hashes } +// newTester creates a new downloader test mocker. func newTester() *downloadTester { tester := &downloadTester{ ownHashes: []common.Hash{knownHash}, @@ -82,9 +86,7 @@ func newTester() *downloadTester { peerHashes: make(map[string][]common.Hash), peerBlocks: make(map[string]map[common.Hash]*types.Block), } - var mux event.TypeMux - downloader := New(&mux, tester.hasBlock, tester.getBlock, tester.insertChain, tester.dropPeer) - tester.downloader = downloader + tester.downloader = New(new(event.TypeMux), tester.hasBlock, tester.getBlock, tester.insertChain, tester.dropPeer) return tester } diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 19c53048c..a8d1f2fb5 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -149,7 +149,8 @@ func (f *Fetcher) loop() { break } if len(announced) == 0 { - fetch.Reset(arriveTimeout) + glog.V(logger.Detail).Infof("Scheduling fetch in %v, at %v", arriveTimeout-time.Since(notification.time), notification.time.Add(arriveTimeout)) + fetch.Reset(arriveTimeout - time.Since(notification.time)) } announced[notification.hash] = append(announced[notification.hash], notification) @@ -181,11 +182,12 @@ func (f *Fetcher) loop() { if len(announced) > 0 { nearest := time.Now() for _, announces := range announced { - if nearest.Before(announces[0].time) { + if nearest.After(announces[0].time) { nearest = announces[0].time } } - fetch.Reset(arriveTimeout + time.Since(nearest)) + glog.V(logger.Detail).Infof("Rescheduling fetch in %v, at %v", arriveTimeout-time.Since(nearest), nearest.Add(arriveTimeout)) + fetch.Reset(arriveTimeout - time.Since(nearest)) } case filter := <-f.filter: diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go new file mode 100644 index 000000000..e29f9c7cd --- /dev/null +++ b/eth/fetcher/fetcher_test.go @@ -0,0 +1,202 @@ +package fetcher + +import ( + "encoding/binary" + "errors" + "math/big" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" +) + +var ( + knownHash = common.Hash{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1} + unknownHash = common.Hash{2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2} + bannedHash = common.Hash{3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3} + + genesis = createBlock(1, common.Hash{}, knownHash) +) + +// idCounter is used by the createHashes method the generate deterministic but unique hashes +var idCounter = int64(2) // #1 is the genesis block + +// createHashes generates a batch of hashes rooted at a specific point in the chain. +func createHashes(amount int, root common.Hash) (hashes []common.Hash) { + hashes = make([]common.Hash, amount+1) + hashes[len(hashes)-1] = root + + for i := 0; i < len(hashes)-1; i++ { + binary.BigEndian.PutUint64(hashes[i][:8], uint64(idCounter)) + idCounter++ + } + return +} + +// createBlock assembles a new block at the given chain height. +func createBlock(i int, parent, hash common.Hash) *types.Block { + header := &types.Header{Number: big.NewInt(int64(i))} + block := types.NewBlockWithHeader(header) + block.HeaderHash = hash + block.ParentHeaderHash = parent + return block +} + +// copyBlock makes a deep copy of a block suitable for local modifications. +func copyBlock(block *types.Block) *types.Block { + return createBlock(int(block.Number().Int64()), block.ParentHeaderHash, block.HeaderHash) +} + +// createBlocksFromHashes assembles a collection of blocks, each having a correct +// place in the given hash chain. +func createBlocksFromHashes(hashes []common.Hash) map[common.Hash]*types.Block { + blocks := make(map[common.Hash]*types.Block) + for i := 0; i < len(hashes); i++ { + parent := knownHash + if i < len(hashes)-1 { + parent = hashes[i+1] + } + blocks[hashes[i]] = createBlock(len(hashes)-i, parent, hashes[i]) + } + return blocks +} + +// fetcherTester is a test simulator for mocking out local block chain. +type fetcherTester struct { + fetcher *Fetcher + + ownHashes []common.Hash // Hash chain belonging to the tester + ownBlocks map[common.Hash]*types.Block // Blocks belonging to the tester +} + +// newTester creates a new fetcher test mocker. +func newTester() *fetcherTester { + tester := &fetcherTester{ + ownHashes: []common.Hash{knownHash}, + ownBlocks: map[common.Hash]*types.Block{knownHash: genesis}, + } + tester.fetcher = New(tester.hasBlock, tester.importBlock) + 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 { + _, ok := f.ownBlocks[hash] + return ok +} + +// importBlock injects a new blocks into the simulated chain. +func (f *fetcherTester) importBlock(peer string, block *types.Block) error { + if _, ok := f.ownBlocks[block.ParentHash()]; !ok { + return errors.New("unknown parent") + } + f.ownHashes = append(f.ownHashes, block.Hash()) + f.ownBlocks[block.Hash()] = block + return nil +} + +// 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 + closure := make(map[common.Hash]*types.Block) + for hash, block := range blocks { + closure[hash] = copyBlock(block) + } + // Create a function that returns blocks from the closure + return func(hashes []common.Hash) error { + // Gather the blocks to return + blocks := make([]*types.Block, 0, len(hashes)) + for _, hash := range hashes { + if block, ok := closure[hash]; ok { + blocks = append(blocks, block) + } + } + // Return on a new thread + go f.fetcher.Filter(blocks) + + return nil + } +} + +// Tests that a fetcher accepts block announcements and initiates retrievals for +// them, successfully importing into the local chain. +func TestSequentialAnnouncements(t *testing.T) { + // Create a chain of blocks to import + targetBlocks := 24 + hashes := createHashes(targetBlocks, knownHash) + blocks := createBlocksFromHashes(hashes) + + tester := newTester() + fetcher := tester.makeFetcher(blocks) + + // Iteratively announce blocks until all are imported + for i := len(hashes) - 1; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], 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) + } +} + +// Tests that if blocks are announced by multiple peers (or even the same buggy +// peer), they will only get downloaded at most once. +func TestConcurrentAnnouncements(t *testing.T) { + // Create a chain of blocks to import + targetBlocks := 24 + hashes := createHashes(targetBlocks, knownHash) + blocks := createBlocksFromHashes(hashes) + + // Assemble a tester with a built in counter for the requests + tester := newTester() + fetcher := tester.makeFetcher(blocks) + + counter := uint32(0) + wrapper := func(hashes []common.Hash) error { + atomic.AddUint32(&counter, uint32(len(hashes))) + return fetcher(hashes) + } + // Iteratively announce blocks until all are imported + for i := len(hashes) - 1; i >= 0; i-- { + tester.fetcher.Notify("first", hashes[i], time.Now().Add(-arriveTimeout), wrapper) + tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper) + tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper) + + 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) + } + // Make sure no blocks were retrieved twice + if int(counter) != targetBlocks { + t.Fatalf("retrieval count mismatch: have %v, want %v", counter, targetBlocks) + } +} + +// Tests that announcements arriving while a previous is being fetched still +// results in a valid import. +func TestOverlappingAnnouncements(t *testing.T) { + // Create a chain of blocks to import + targetBlocks := 24 + hashes := createHashes(targetBlocks, knownHash) + blocks := createBlocksFromHashes(hashes) + + tester := newTester() + fetcher := tester.makeFetcher(blocks) + + // Iteratively announce blocks, but overlap them continuously + delay, overlap := 50*time.Millisecond, time.Duration(5) + for i := len(hashes) - 1; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout+overlap*delay), fetcher) + time.Sleep(delay) + } + time.Sleep(overlap * delay) + + if imported := len(tester.ownBlocks); imported != targetBlocks+1 { + t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) + } +} -- cgit From 8b64e041d6c41d76994510fdd8bb42ad7c5be5aa Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 16 Jun 2015 15:35:15 +0300 Subject: eth/fetcher: add test to check for duplicate downloads --- eth/fetcher/fetcher_test.go | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index e29f9c7cd..44e99f30f 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -200,3 +200,41 @@ func TestOverlappingAnnouncements(t *testing.T) { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) } } + +// Tests that announces already being retrieved will not be duplicated. +func TestPendingDeduplication(t *testing.T) { + // Create a hash and corresponding block + hashes := createHashes(1, knownHash) + blocks := createBlocksFromHashes(hashes) + + // Assemble a tester with a built in counter and delayed fetcher + tester := newTester() + fetcher := tester.makeFetcher(blocks) + + delay := 50 * time.Millisecond + counter := uint32(0) + wrapper := func(hashes []common.Hash) error { + atomic.AddUint32(&counter, uint32(len(hashes))) + + // Simulate a long running fetch + go func() { + time.Sleep(delay) + fetcher(hashes) + }() + return nil + } + // Announce the same block many times until it's fetched (wait for any pending ops) + for !tester.hasBlock(hashes[0]) { + tester.fetcher.Notify("repeater", hashes[0], time.Now().Add(-arriveTimeout), wrapper) + time.Sleep(time.Millisecond) + } + time.Sleep(delay) + + // Check that all blocks were imported and none fetched twice + if imported := len(tester.ownBlocks); imported != 2 { + t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 2) + } + if int(counter) != 1 { + t.Fatalf("retrieval count mismatch: have %v, want %v", counter, 1) + } +} -- cgit From 057bc237adf9ed0adf615a72cc1e92d9aed15d9e Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 16 Jun 2015 17:39:04 +0300 Subject: eth, eth/fetcher: use an import queue to store out of order blocks --- eth/fetcher/fetcher.go | 78 ++++++++++++++++++++++++++++----------------- eth/fetcher/fetcher_test.go | 35 +++++++++++++++++++- eth/handler.go | 11 +++++-- 3 files changed, 91 insertions(+), 33 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) + } +} diff --git a/eth/handler.go b/eth/handler.go index 99ac4ce68..e5908dc88 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -1,6 +1,7 @@ package eth import ( + "errors" "fmt" "math" "math/big" @@ -94,9 +95,15 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) importer := func(peer string, block *types.Block) error { - return manager.importBlock(manager.peers.Peer(peer), block, nil) + if p := manager.peers.Peer(peer); p != nil { + return manager.importBlock(manager.peers.Peer(peer), block, nil) + } + return errors.New("unknown peer") + } + heighter := func() uint64 { + return manager.chainman.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer) + manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer, heighter) return manager } -- cgit From 11c8f83a583745b01a4c06cdd29529af1df364f6 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 16 Jun 2015 18:14:52 +0300 Subject: eth, eth/fetcher: cache future propagated blocks too --- eth/fetcher/fetcher.go | 21 +++++++++++++++++++++ eth/fetcher/fetcher_test.go | 28 ++++++++++++++++++++++++++++ eth/handler.go | 2 ++ 3 files changed, 51 insertions(+) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index e8a9cc093..34d368780 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -56,6 +56,7 @@ type inject struct { type Fetcher struct { // Various event channels notify chan *announce + insert chan *inject filter chan chan []*types.Block quit chan struct{} @@ -69,6 +70,7 @@ type Fetcher struct { func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHeightFn) *Fetcher { return &Fetcher{ notify: make(chan *announce), + insert: make(chan *inject), filter: make(chan chan []*types.Block), quit: make(chan struct{}), hasBlock: hasBlock, @@ -106,6 +108,20 @@ func (f *Fetcher) Notify(peer string, hash common.Hash, time time.Time, fetcher } } +// Enqueue tries to fill gaps the the fetcher's future import queue. +func (f *Fetcher) Enqueue(peer string, block *types.Block) error { + op := &inject{ + origin: peer, + block: block, + } + select { + case f.insert <- op: + return nil + case <-f.quit: + return errTerminated + } +} + // Filter extracts all the blocks that were explicitly requested by the fetcher, // returning those that should be handled differently. func (f *Fetcher) Filter(blocks types.Blocks) types.Blocks { @@ -192,6 +208,11 @@ func (f *Fetcher) loop() { } announced[notification.hash] = append(announced[notification.hash], notification) + 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()) + case hash := <-done: // A pending import finished, remove all traces of the notification delete(announced, hash) diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index cde4bb70a..7c975841c 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -271,3 +271,31 @@ func TestRandomArrivalImport(t *testing.T) { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) } } + +// Tests that direct block enqueues (due to block propagation vs. hash announce) +// are correctly schedule, filling and import queue gaps. +func TestQueueGapFill(t *testing.T) { + // Create a chain of blocks to import, and choose one to not announce at all + 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) + } + } + // Fill the missing block directly as if propagated + tester.fetcher.Enqueue("valid", blocks[hashes[skip]]) + 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) + } +} diff --git a/eth/handler.go b/eth/handler.go index e5908dc88..604983a7b 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -340,6 +340,8 @@ func (self *ProtocolManager) handleMsg(p *peer) error { } request.Block.ReceivedAt = msg.ReceivedAt + // Try to import the propagated block, also making it fill any fetcher gaps + self.fetcher.Enqueue(p.id, request.Block) if err := self.importBlock(p, request.Block, request.TD); err != nil { return err } -- cgit From 026ee40650bde909c962f15e32bdea6c1d6e978a Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 16 Jun 2015 18:43:58 +0300 Subject: eth/fetcher: deduplicate future blocks --- eth/fetcher/fetcher.go | 60 ++++++++++++++++++++++++++------------------- eth/fetcher/fetcher_test.go | 36 +++++++++++++++++++++++++++ 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) + } +} -- cgit From 497a7f1717a798e59c9550fbb58504b08fe13b21 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 16 Jun 2015 23:18:01 +0300 Subject: eth, eth/fetcher: define and enforce propagation boundaries --- eth/fetcher/fetcher.go | 55 ++++++++++++++++++++++++++------------------- eth/fetcher/fetcher_test.go | 28 +++++++++++++++++++++++ eth/handler.go | 1 - 3 files changed, 60 insertions(+), 24 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 207bd9323..76d3798cb 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -3,6 +3,7 @@ package fetcher import ( "errors" + "math" "math/rand" "time" @@ -60,6 +61,10 @@ type Fetcher struct { filter chan chan []*types.Block quit chan struct{} + // Block cache + queue *prque.Prque // Queue containing the import operations (block number sorted) + 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 @@ -73,6 +78,8 @@ func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHei insert: make(chan *inject), filter: make(chan chan []*types.Block), quit: make(chan struct{}), + queue: prque.New(), + queued: make(map[common.Hash]struct{}), hasBlock: hasBlock, importBlock: importBlock, chainHeight: chainHeight, @@ -154,23 +161,6 @@ func (f *Fetcher) loop() { announced := make(map[common.Hash][]*announce) fetching := make(map[common.Hash]*announce) - // 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) @@ -185,16 +175,16 @@ func (f *Fetcher) loop() { } // Import any queued blocks that could potentially fit height := f.chainHeight() - for !queue.Empty() { + for !f.queue.Empty() { // If too high up the chain, continue later - op := queue.PopItem().(*inject) + op := f.queue.PopItem().(*inject) if number := op.block.NumberU64(); number > height+1 { - queue.Push(op, -float32(op.block.NumberU64())) + f.queue.Push(op, -float32(op.block.NumberU64())) break } // Otherwise if not known yet, try and import hash := op.block.Hash() - delete(queued, hash) + delete(f.queued, hash) if f.hasBlock(hash) { continue } @@ -229,7 +219,7 @@ func (f *Fetcher) loop() { case op := <-f.insert: // A direct block insertion was requested, try and fill any pending gaps - enqueue(op.origin, op.block) + f.enqueue(op.origin, op.block) case hash := <-done: // A pending import finished, remove all traces of the notification @@ -301,9 +291,28 @@ func (f *Fetcher) loop() { // Schedule the retrieved blocks for ordered import for _, block := range explicit { if announce := fetching[block.Hash()]; announce != nil { - enqueue(announce.origin, block) + f.enqueue(announce.origin, block) } } } } } + +// enqueue schedules a new future import operation, if the block to be imported +// has not yet been seen. +func (f *Fetcher) enqueue(peer string, block *types.Block) { + // Make sure the block isn't in some weird place + if math.Abs(float64(f.chainHeight())-float64(block.NumberU64())) > maxQueueDist { + return + } + // Schedule the block for future importing + hash := block.Hash() + if _, ok := f.queued[hash]; !ok { + f.queued[hash] = struct{}{} + f.queue.Push(&inject{origin: peer, block: block}, -float32(block.NumberU64())) + + if glog.V(logger.Detail) { + glog.Infof("Peer %s: queued block %x, total %v", peer, hash.Bytes()[:4], f.queue.Size()) + } + } +} diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index e11a211a1..af2652a86 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -335,3 +335,31 @@ func TestImportDeduplication(t *testing.T) { t.Fatalf("import invocation count mismatch: have %v, want %v", counter, 2) } } + +// Tests that blocks with numbers much lower or higher than out current head get +// discarded no prevent wasting resources on useless blocks from faulty peers. +func TestDistantDiscarding(t *testing.T) { + // Create a long chain to import + hashes := createHashes(3*maxQueueDist, knownHash) + blocks := createBlocksFromHashes(hashes) + + head := hashes[len(hashes)/2] + + // Create a tester and simulate a head block being the middle of the above chain + tester := newTester() + tester.ownHashes = []common.Hash{head} + tester.ownBlocks = map[common.Hash]*types.Block{head: blocks[head]} + + // Ensure that a block with a lower number than the threshold is discarded + tester.fetcher.Enqueue("lower", blocks[hashes[0]]) + time.Sleep(10 * time.Millisecond) + if !tester.fetcher.queue.Empty() { + t.Fatalf("fetcher queued stale block") + } + // Ensure that a block with a higher number than the threshold is discarded + tester.fetcher.Enqueue("higher", blocks[hashes[len(hashes)-1]]) + time.Sleep(10 * time.Millisecond) + if !tester.fetcher.queue.Empty() { + t.Fatalf("fetcher queued future block") + } +} diff --git a/eth/handler.go b/eth/handler.go index 604983a7b..c3b58650d 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -9,7 +9,6 @@ import ( "time" "github.com/ethereum/go-ethereum/eth/fetcher" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" -- cgit From 2a7411bc9605dae9798ed29ed237e28c8fc98895 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Wed, 17 Jun 2015 00:19:09 +0300 Subject: eth/fetcher: fix premature queue cleanup, general polishes --- eth/fetcher/fetcher.go | 82 +++++++++++++++++++++++++++++--------------------- 1 file changed, 47 insertions(+), 35 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 76d3798cb..78d25f72e 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -61,6 +61,10 @@ type Fetcher struct { filter chan chan []*types.Block quit chan struct{} + // Announce states + announced map[common.Hash][]*announce // Announced blocks, scheduled for fetching + fetching map[common.Hash]*announce // Announced blocks, currently fetching + // Block cache queue *prque.Prque // Queue containing the import operations (block number sorted) queued map[common.Hash]struct{} // Presence set of already queued blocks (to dedup imports) @@ -78,6 +82,8 @@ func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHei insert: make(chan *inject), filter: make(chan chan []*types.Block), 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, @@ -158,19 +164,16 @@ func (f *Fetcher) Filter(blocks types.Blocks) types.Blocks { // Loop is the main fetcher loop, checking and processing various notification // events. func (f *Fetcher) loop() { - announced := make(map[common.Hash][]*announce) - fetching := make(map[common.Hash]*announce) - // Iterate the block fetching until a quit is requested fetch := time.NewTimer(0) done := make(chan common.Hash) for { // Clean up any expired block fetches - for hash, announce := range fetching { + for hash, announce := range f.fetching { if time.Since(announce.time) > fetchTimeout { - delete(announced, hash) - delete(fetching, hash) + delete(f.announced, hash) + delete(f.fetching, hash) } } // Import any queued blocks that could potentially fit @@ -184,17 +187,16 @@ func (f *Fetcher) loop() { } // Otherwise if not known yet, try and import hash := op.block.Hash() - delete(f.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, hash.Bytes()[:4]) + glog.V(logger.Debug).Infof("Peer %s: importing block #%d [%x]", op.origin, op.block.NumberU64(), hash.Bytes()[:4]) go func() { 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, hash.Bytes()[:4], err) + glog.V(logger.Detail).Infof("Peer %s: block #%d [%x] import failed: %v", op.origin, op.block.NumberU64(), hash.Bytes()[:4], err) return } }() @@ -208,14 +210,13 @@ func (f *Fetcher) loop() { case notification := <-f.notify: // A block was announced, schedule if it's not yet downloading glog.V(logger.Debug).Infof("Peer %s: scheduling %x", notification.origin, notification.hash[:4]) - if _, ok := fetching[notification.hash]; ok { + if _, ok := f.fetching[notification.hash]; ok { break } - if len(announced) == 0 { - glog.V(logger.Detail).Infof("Scheduling fetch in %v, at %v", arriveTimeout-time.Since(notification.time), notification.time.Add(arriveTimeout)) - fetch.Reset(arriveTimeout - time.Since(notification.time)) + f.announced[notification.hash] = append(f.announced[notification.hash], notification) + if len(f.announced) == 1 { + f.reschedule(fetch) } - announced[notification.hash] = append(announced[notification.hash], notification) case op := <-f.insert: // A direct block insertion was requested, try and fill any pending gaps @@ -223,39 +224,31 @@ func (f *Fetcher) loop() { case hash := <-done: // A pending import finished, remove all traces of the notification - delete(announced, hash) - delete(fetching, hash) + delete(f.announced, hash) + delete(f.fetching, hash) + delete(f.queued, hash) case <-fetch.C: // At least one block's timer ran out, check for needing retrieval request := make(map[string][]common.Hash) - for hash, announces := range announced { + for hash, announces := range f.announced { if time.Since(announces[0].time) > arriveTimeout { announce := announces[rand.Intn(len(announces))] if !f.hasBlock(hash) { request[announce.origin] = append(request[announce.origin], hash) - fetching[hash] = announce + f.fetching[hash] = announce } - delete(announced, hash) + delete(f.announced, hash) } } // Send out all block requests for peer, hashes := range request { glog.V(logger.Debug).Infof("Peer %s: explicitly fetching %d blocks", peer, len(hashes)) - go fetching[hashes[0]].fetch(hashes) + go f.fetching[hashes[0]].fetch(hashes) } // Schedule the next fetch if blocks are still pending - if len(announced) > 0 { - nearest := time.Now() - for _, announces := range announced { - if nearest.After(announces[0].time) { - nearest = announces[0].time - } - } - glog.V(logger.Detail).Infof("Rescheduling fetch in %v, at %v", arriveTimeout-time.Since(nearest), nearest.Add(arriveTimeout)) - fetch.Reset(arriveTimeout - time.Since(nearest)) - } + f.reschedule(fetch) case filter := <-f.filter: // Blocks arrived, extract any explicit fetches, return all else @@ -271,12 +264,12 @@ func (f *Fetcher) loop() { hash := block.Hash() // Filter explicitly requested blocks from hash announcements - if _, ok := fetching[hash]; ok { + if _, ok := f.fetching[hash]; ok { // Discard if already imported by other means if !f.hasBlock(hash) { explicit = append(explicit, block) } else { - delete(fetching, hash) + delete(f.fetching, hash) } } else { download = append(download, block) @@ -290,7 +283,7 @@ func (f *Fetcher) loop() { } // Schedule the retrieved blocks for ordered import for _, block := range explicit { - if announce := fetching[block.Hash()]; announce != nil { + if announce := f.fetching[block.Hash()]; announce != nil { f.enqueue(announce.origin, block) } } @@ -298,21 +291,40 @@ func (f *Fetcher) loop() { } } +// reschedule resets the specified fetch timer to the next announce timeout. +func (f *Fetcher) reschedule(fetch *time.Timer) { + // Short circuit if no blocks are announced + if len(f.announced) == 0 { + return + } + // Otherwise find the earliest expiring announcement + earliest := time.Now() + for _, announces := range f.announced { + if earliest.After(announces[0].time) { + earliest = announces[0].time + } + } + glog.V(logger.Detail).Infof("Scheduling next fetch in %v", arriveTimeout-time.Since(earliest)) + fetch.Reset(arriveTimeout - time.Since(earliest)) +} + // enqueue schedules a new future import operation, if the block to be imported // has not yet been seen. func (f *Fetcher) enqueue(peer string, block *types.Block) { + hash := block.Hash() + // Make sure the block isn't in some weird place if math.Abs(float64(f.chainHeight())-float64(block.NumberU64())) > maxQueueDist { + glog.Infof("Peer %s: discarded block #%d [%x] too far from head", peer, block.NumberU64(), hash.Bytes()[:4]) return } // Schedule the block for future importing - hash := block.Hash() if _, ok := f.queued[hash]; !ok { f.queued[hash] = struct{}{} f.queue.Push(&inject{origin: peer, block: block}, -float32(block.NumberU64())) if glog.V(logger.Detail) { - glog.Infof("Peer %s: queued block %x, total %v", peer, hash.Bytes()[:4], f.queue.Size()) + glog.Infof("Peer %s: queued block #%d [%x], total %v", peer, block.NumberU64(), hash.Bytes()[:4], f.queue.Size()) } } } -- cgit From 37c5ff392f71fddaa6acd92925f69e81876fe22e Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Wed, 17 Jun 2015 16:53:28 +0300 Subject: eth/fetcher: build longest chain until proven otherwise --- eth/fetcher/fetcher.go | 67 +++++++++++++++++++++++++-------------------- eth/fetcher/fetcher_test.go | 57 ++++++++++++++++++++++++++++++++++++++ eth/handler.go | 2 +- 3 files changed, 95 insertions(+), 31 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 78d25f72e..a70fcbeed 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -3,7 +3,6 @@ package fetcher import ( "errors" - "math" "math/rand" "time" @@ -57,8 +56,9 @@ type inject struct { type Fetcher struct { // Various event channels notify chan *announce - insert chan *inject + inject chan *inject filter chan chan []*types.Block + done chan common.Hash quit chan struct{} // Announce states @@ -79,8 +79,9 @@ type Fetcher struct { func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHeightFn) *Fetcher { return &Fetcher{ notify: make(chan *announce), - insert: make(chan *inject), + 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), @@ -128,7 +129,7 @@ func (f *Fetcher) Enqueue(peer string, block *types.Block) error { block: block, } select { - case f.insert <- op: + case f.inject <- op: return nil case <-f.quit: return errTerminated @@ -166,8 +167,6 @@ func (f *Fetcher) Filter(blocks types.Blocks) types.Blocks { func (f *Fetcher) loop() { // Iterate the block fetching until a quit is requested fetch := time.NewTimer(0) - done := make(chan common.Hash) - for { // Clean up any expired block fetches for hash, announce := range f.fetching { @@ -179,27 +178,19 @@ func (f *Fetcher) loop() { // Import any queued blocks that could potentially fit height := f.chainHeight() for !f.queue.Empty() { - // If too high up the chain, continue later op := f.queue.PopItem().(*inject) - if number := op.block.NumberU64(); number > height+1 { + number := op.block.NumberU64() + + // If too high up the chain or phase, continue later + if number > height+1 { f.queue.Push(op, -float32(op.block.NumberU64())) break } - // Otherwise if not known yet, try and import - hash := op.block.Hash() - if f.hasBlock(hash) { + // Otherwise if fresh and still unknown, try and import + if number <= height || f.hasBlock(op.block.Hash()) { continue } - // Block may just fit, try to import it - glog.V(logger.Debug).Infof("Peer %s: importing block #%d [%x]", op.origin, op.block.NumberU64(), hash.Bytes()[:4]) - go func() { - defer func() { done <- hash }() - - if err := f.importBlock(op.origin, op.block); err != nil { - glog.V(logger.Detail).Infof("Peer %s: block #%d [%x] import failed: %v", op.origin, op.block.NumberU64(), hash.Bytes()[:4], err) - return - } - }() + f.insert(op.origin, op.block) } // Wait for an outside event to occur select { @@ -209,7 +200,6 @@ func (f *Fetcher) loop() { case notification := <-f.notify: // A block was announced, schedule if it's not yet downloading - glog.V(logger.Debug).Infof("Peer %s: scheduling %x", notification.origin, notification.hash[:4]) if _, ok := f.fetching[notification.hash]; ok { break } @@ -218,11 +208,11 @@ func (f *Fetcher) loop() { f.reschedule(fetch) } - case op := <-f.insert: + case op := <-f.inject: // A direct block insertion was requested, try and fill any pending gaps f.enqueue(op.origin, op.block) - case hash := <-done: + case hash := <-f.done: // A pending import finished, remove all traces of the notification delete(f.announced, hash) delete(f.fetching, hash) @@ -243,8 +233,7 @@ func (f *Fetcher) loop() { } } // Send out all block requests - for peer, hashes := range request { - glog.V(logger.Debug).Infof("Peer %s: explicitly fetching %d blocks", peer, len(hashes)) + for _, hashes := range request { go f.fetching[hashes[0]].fetch(hashes) } // Schedule the next fetch if blocks are still pending @@ -304,7 +293,6 @@ func (f *Fetcher) reschedule(fetch *time.Timer) { earliest = announces[0].time } } - glog.V(logger.Detail).Infof("Scheduling next fetch in %v", arriveTimeout-time.Since(earliest)) fetch.Reset(arriveTimeout - time.Since(earliest)) } @@ -313,9 +301,9 @@ func (f *Fetcher) reschedule(fetch *time.Timer) { func (f *Fetcher) enqueue(peer string, block *types.Block) { hash := block.Hash() - // Make sure the block isn't in some weird place - if math.Abs(float64(f.chainHeight())-float64(block.NumberU64())) > maxQueueDist { - glog.Infof("Peer %s: discarded block #%d [%x] too far from head", peer, block.NumberU64(), hash.Bytes()[:4]) + // 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) return } // Schedule the block for future importing @@ -328,3 +316,22 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { } } } + +// insert spawns a new goroutine to run a block insertion into the chain. If the +// block's number is at the same height as the current import phase, if updates +// the phase states accordingly. +func (f *Fetcher) insert(peer string, block *types.Block) { + hash := block.Hash() + + // Run the import on a new thread + glog.V(logger.Debug).Infof("Peer %s: importing block #%d [%x]", peer, block.NumberU64(), hash[:4]) + go func() { + defer func() { f.done <- hash }() + + // Run the actual import and log any issues + if err := f.importBlock(peer, block); err != nil { + glog.V(logger.Detail).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err) + return + } + }() +} diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index af2652a86..00fad3498 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -91,9 +91,15 @@ func (f *fetcherTester) hasBlock(hash common.Hash) bool { // importBlock injects a new blocks into the simulated chain. func (f *fetcherTester) importBlock(peer string, block *types.Block) error { + // Make sure the parent in known if _, ok := f.ownBlocks[block.ParentHash()]; !ok { return errors.New("unknown parent") } + // Discard any new blocks if the same height already exists + if block.NumberU64() <= f.ownBlocks[f.ownHashes[len(f.ownHashes)-1]].NumberU64() { + return nil + } + // Otherwise build our current chain f.ownHashes = append(f.ownHashes, block.Hash()) f.ownBlocks[block.Hash()] = block return nil @@ -363,3 +369,54 @@ func TestDistantDiscarding(t *testing.T) { t.Fatalf("fetcher queued future block") } } + +// Tests that if multiple uncles (i.e. blocks at the same height) are queued for +// importing, then they will get inserted in phases, previous heights needing to +// complete before the next numbered blocks can begin. +func TestCompetingImports(t *testing.T) { + // Generate a few soft-forks for concurrent imports + hashesA := createHashes(16, knownHash) + hashesB := createHashes(16, knownHash) + hashesC := createHashes(16, knownHash) + + blocksA := createBlocksFromHashes(hashesA) + blocksB := createBlocksFromHashes(hashesB) + blocksC := createBlocksFromHashes(hashesC) + + // Create a tester, and override the import to check number reversals + tester := newTester() + + first := int32(1) + height := uint64(1) + tester.fetcher.importBlock = func(peer string, block *types.Block) 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) + } + atomic.StoreUint64(&height, block.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) + } + // Queue up everything but with a missing link + for i := 0; i < len(hashesA)-2; i++ { + tester.fetcher.Enqueue("chain A", blocksA[hashesA[i]]) + tester.fetcher.Enqueue("chain B", blocksB[hashesB[i]]) + tester.fetcher.Enqueue("chain C", blocksC[hashesC[i]]) + } + // Add the three missing links, and wait for a full import + tester.fetcher.Enqueue("chain A", blocksA[hashesA[len(hashesA)-2]]) + tester.fetcher.Enqueue("chain B", blocksB[hashesB[len(hashesB)-2]]) + tester.fetcher.Enqueue("chain C", blocksC[hashesC[len(hashesC)-2]]) + + start := time.Now() + for len(tester.ownHashes) != len(hashesA) && time.Since(start) < time.Second { + time.Sleep(50 * time.Millisecond) + } + if len(tester.ownHashes) != len(hashesA) { + t.Fatalf("chain length mismatch: have %v, want %v", len(tester.ownHashes), len(hashesA)) + } +} diff --git a/eth/handler.go b/eth/handler.go index c3b58650d..b62815532 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -8,11 +8,11 @@ import ( "sync" "time" - "github.com/ethereum/go-ethereum/eth/fetcher" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth/downloader" + "github.com/ethereum/go-ethereum/eth/fetcher" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" -- cgit From a9ada0b5baa9c47f8673cb63707797cbd08f0d61 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Wed, 17 Jun 2015 17:08:32 +0300 Subject: eth/fetcher: make tests thread safe --- eth/fetcher/fetcher_test.go | 56 +++++++++++++++++++++++++++------------------ 1 file changed, 34 insertions(+), 22 deletions(-) diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 00fad3498..4c6a1bf6a 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -4,6 +4,7 @@ import ( "encoding/binary" "errors" "math/big" + "sync" "sync/atomic" "testing" "time" @@ -67,15 +68,17 @@ func createBlocksFromHashes(hashes []common.Hash) map[common.Hash]*types.Block { type fetcherTester struct { fetcher *Fetcher - ownHashes []common.Hash // Hash chain belonging to the tester - ownBlocks map[common.Hash]*types.Block // Blocks belonging to the tester + hashes []common.Hash // Hash chain belonging to the tester + blocks map[common.Hash]*types.Block // Blocks belonging to the tester + + lock sync.RWMutex } // newTester creates a new fetcher test mocker. func newTester() *fetcherTester { tester := &fetcherTester{ - ownHashes: []common.Hash{knownHash}, - ownBlocks: map[common.Hash]*types.Block{knownHash: genesis}, + hashes: []common.Hash{knownHash}, + blocks: map[common.Hash]*types.Block{knownHash: genesis}, } tester.fetcher = New(tester.hasBlock, tester.importBlock, tester.chainHeight) tester.fetcher.Start() @@ -85,29 +88,38 @@ func newTester() *fetcherTester { // hasBlock checks if a block is pres ent in the testers canonical chain. func (f *fetcherTester) hasBlock(hash common.Hash) bool { - _, ok := f.ownBlocks[hash] + f.lock.RLock() + defer f.lock.RUnlock() + + _, ok := f.blocks[hash] 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.ownBlocks[block.ParentHash()]; !ok { + 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.ownBlocks[f.ownHashes[len(f.ownHashes)-1]].NumberU64() { + if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() { return nil } // Otherwise build our current chain - f.ownHashes = append(f.ownHashes, block.Hash()) - f.ownBlocks[block.Hash()] = block + f.hashes = append(f.hashes, block.Hash()) + f.blocks[block.Hash()] = block 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() + f.lock.RLock() + defer f.lock.RUnlock() + + return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() } // peerFetcher retrieves a fetcher associated with a simulated peer. @@ -149,7 +161,7 @@ func TestSequentialAnnouncements(t *testing.T) { tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) time.Sleep(50 * time.Millisecond) } - if imported := len(tester.ownBlocks); imported != targetBlocks+1 { + if imported := len(tester.blocks); imported != targetBlocks+1 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) } } @@ -179,7 +191,7 @@ func TestConcurrentAnnouncements(t *testing.T) { time.Sleep(50 * time.Millisecond) } - if imported := len(tester.ownBlocks); imported != targetBlocks+1 { + if imported := len(tester.blocks); imported != targetBlocks+1 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) } // Make sure no blocks were retrieved twice @@ -207,7 +219,7 @@ func TestOverlappingAnnouncements(t *testing.T) { } time.Sleep(overlap * delay) - if imported := len(tester.ownBlocks); imported != targetBlocks+1 { + if imported := len(tester.blocks); imported != targetBlocks+1 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) } } @@ -242,7 +254,7 @@ func TestPendingDeduplication(t *testing.T) { time.Sleep(delay) // Check that all blocks were imported and none fetched twice - if imported := len(tester.ownBlocks); imported != 2 { + if imported := len(tester.blocks); imported != 2 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 2) } if int(counter) != 1 { @@ -273,7 +285,7 @@ func TestRandomArrivalImport(t *testing.T) { tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher) time.Sleep(50 * time.Millisecond) - if imported := len(tester.ownBlocks); imported != targetBlocks+1 { + if imported := len(tester.blocks); imported != targetBlocks+1 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) } } @@ -301,7 +313,7 @@ func TestQueueGapFill(t *testing.T) { tester.fetcher.Enqueue("valid", blocks[hashes[skip]]) time.Sleep(50 * time.Millisecond) - if imported := len(tester.ownBlocks); imported != targetBlocks+1 { + if imported := len(tester.blocks); imported != targetBlocks+1 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) } } @@ -334,7 +346,7 @@ func TestImportDeduplication(t *testing.T) { tester.fetcher.Enqueue("valid", blocks[hashes[1]]) time.Sleep(50 * time.Millisecond) - if imported := len(tester.ownBlocks); imported != 3 { + if imported := len(tester.blocks); imported != 3 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 3) } if counter != 2 { @@ -353,8 +365,8 @@ func TestDistantDiscarding(t *testing.T) { // Create a tester and simulate a head block being the middle of the above chain tester := newTester() - tester.ownHashes = []common.Hash{head} - tester.ownBlocks = map[common.Hash]*types.Block{head: blocks[head]} + tester.hashes = []common.Hash{head} + tester.blocks = map[common.Hash]*types.Block{head: blocks[head]} // Ensure that a block with a lower number than the threshold is discarded tester.fetcher.Enqueue("lower", blocks[hashes[0]]) @@ -413,10 +425,10 @@ func TestCompetingImports(t *testing.T) { tester.fetcher.Enqueue("chain C", blocksC[hashesC[len(hashesC)-2]]) start := time.Now() - for len(tester.ownHashes) != len(hashesA) && time.Since(start) < time.Second { + for len(tester.hashes) != len(hashesA) && time.Since(start) < time.Second { time.Sleep(50 * time.Millisecond) } - if len(tester.ownHashes) != len(hashesA) { - t.Fatalf("chain length mismatch: have %v, want %v", len(tester.ownHashes), len(hashesA)) + if len(tester.hashes) != len(hashesA) { + t.Fatalf("chain length mismatch: have %v, want %v", len(tester.hashes), len(hashesA)) } } -- cgit 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/downloader/downloader.go | 2 +- eth/fetcher/fetcher.go | 55 ++++++++++++++-------- eth/fetcher/fetcher_test.go | 62 ++++++++++++++---------- eth/handler.go | 110 +++++++++++++------------------------------ 4 files changed, 106 insertions(+), 123 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 18e5f50e8..39976aae1 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -99,7 +99,7 @@ type Downloader struct { hasBlock hashCheckFn // Checks if a block is present in the chain getBlock blockRetrievalFn // Retrieves a block from the chain insertChain chainInsertFn // Injects a batch of blocks into the chain - dropPeer peerDropFn // Retrieved the TD of our own chain + dropPeer peerDropFn // Drops a peer for misbehaving // Status synchroniseMock func(id string, hash common.Hash) error // Replacement for synchronise during testing 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++ { diff --git a/eth/handler.go b/eth/handler.go index b62815532..eeb6a4f14 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -1,10 +1,8 @@ package eth import ( - "errors" "fmt" "math" - "math/big" "sync" "time" @@ -93,16 +91,10 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo // Construct the different synchronisation mechanisms manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) - importer := func(peer string, block *types.Block) error { - if p := manager.peers.Peer(peer); p != nil { - return manager.importBlock(manager.peers.Peer(peer), block, nil) - } - return errors.New("unknown peer") - } heighter := func() uint64 { return manager.chainman.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer, heighter) + manager.fetcher = fetcher.New(manager.chainman.HasBlock, manager.BroadcastBlock, heighter, manager.chainman.InsertChain, manager.removePeer) return manager } @@ -194,7 +186,7 @@ func (pm *ProtocolManager) handle(p *peer) error { return nil } -func (self *ProtocolManager) handleMsg(p *peer) error { +func (pm *ProtocolManager) handleMsg(p *peer) error { msg, err := p.rw.ReadMsg() if err != nil { return err @@ -224,7 +216,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { RemoteId: p.ID().String(), }) } - self.txpool.AddTransactions(txs) + pm.txpool.AddTransactions(txs) case GetBlockHashesMsg: var request getBlockHashesMsgData @@ -236,7 +228,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { request.Amount = uint64(downloader.MaxHashFetch) } - hashes := self.chainman.GetBlockHashesFromHash(request.Hash, request.Amount) + hashes := pm.chainman.GetBlockHashesFromHash(request.Hash, request.Amount) if glog.V(logger.Debug) { if len(hashes) == 0 { @@ -254,7 +246,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { if err := msgStream.Decode(&hashes); err != nil { break } - err := self.downloader.DeliverHashes(p.id, hashes) + err := pm.downloader.DeliverHashes(p.id, hashes) if err != nil { glog.V(logger.Debug).Infoln(err) } @@ -280,7 +272,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "msg %v: %v", msg, err) } - block := self.chainman.GetBlock(hash) + block := pm.chainman.GetBlock(hash) if block != nil { blocks = append(blocks, block) totalsize += block.Size() @@ -301,8 +293,8 @@ func (self *ProtocolManager) handleMsg(p *peer) error { blocks = nil } // Filter out any explicitly requested blocks, deliver the rest to the downloader - if blocks := self.fetcher.Filter(blocks); len(blocks) > 0 { - self.downloader.DeliverBlocks(p.id, blocks) + if blocks := pm.fetcher.Filter(blocks); len(blocks) > 0 { + pm.downloader.DeliverBlocks(p.id, blocks) } case NewBlockHashesMsg: @@ -321,15 +313,16 @@ func (self *ProtocolManager) handleMsg(p *peer) error { // Schedule all the unknown hashes for retrieval unknown := make([]common.Hash, 0, len(hashes)) for _, hash := range hashes { - if !self.chainman.HasBlock(hash) { + if !pm.chainman.HasBlock(hash) { unknown = append(unknown, hash) } } for _, hash := range unknown { - self.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks) + pm.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks) } case NewBlockMsg: + // Retrieve and decode the propagated block var request newBlockMsgData if err := msg.Decode(&request); err != nil { return errResp(ErrDecode, "%v: %v", msg, err) @@ -339,11 +332,24 @@ func (self *ProtocolManager) handleMsg(p *peer) error { } request.Block.ReceivedAt = msg.ReceivedAt - // Try to import the propagated block, also making it fill any fetcher gaps - self.fetcher.Enqueue(p.id, request.Block) - if err := self.importBlock(p, request.Block, request.TD); err != nil { - return err - } + // Mark the block's arrival for whatever reason + _, chainHead, _ := pm.chainman.Status() + jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{ + BlockHash: request.Block.Hash().Hex(), + BlockNumber: request.Block.Number(), + ChainHeadHash: chainHead.Hex(), + BlockPrevHash: request.Block.ParentHash().Hex(), + RemoteId: p.ID().String(), + }) + // Mark the peer as owning the block and schedule it for import + p.blockHashes.Add(request.Block.Hash()) + p.SetHead(request.Block.Hash()) + + pm.fetcher.Enqueue(p.id, request.Block) + + // TODO: Schedule a sync to cover potential gaps (this needs proto update) + p.SetTd(request.TD) + go pm.synchronise(p) default: return errResp(ErrInvalidMsgCode, "%v", msg.Code) @@ -351,61 +357,11 @@ func (self *ProtocolManager) handleMsg(p *peer) error { return nil } -// importBlocks injects a new block retrieved from the given peer into the chain -// manager. -func (pm *ProtocolManager) importBlock(p *peer, block *types.Block, td *big.Int) error { - hash := block.Hash() - - // Mark the block as present at the remote node (don't duplicate already held data) - p.blockHashes.Add(hash) - p.SetHead(hash) - if td != nil { - p.SetTd(td) - } - // Log the block's arrival - _, chainHead, _ := pm.chainman.Status() - jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{ - BlockHash: hash.Hex(), - BlockNumber: block.Number(), - ChainHeadHash: chainHead.Hex(), - BlockPrevHash: block.ParentHash().Hex(), - RemoteId: p.ID().String(), - }) - // If the block's already known or its difficulty is lower than ours, drop - if pm.chainman.HasBlock(hash) { - p.SetTd(pm.chainman.GetBlock(hash).Td) // update the peer's TD to the real value - return nil - } - if td != nil && pm.chainman.Td().Cmp(td) > 0 && new(big.Int).Add(block.Number(), big.NewInt(7)).Cmp(pm.chainman.CurrentBlock().Number()) < 0 { - glog.V(logger.Debug).Infof("[%s] dropped block %v due to low TD %v\n", p.id, block.Number(), td) - return nil - } - // Attempt to insert the newly received block and propagate to our peers - if pm.chainman.HasBlock(block.ParentHash()) { - if _, err := pm.chainman.InsertChain(types.Blocks{block}); err != nil { - glog.V(logger.Error).Infoln("removed peer (", p.id, ") due to block error", err) - return err - } - if td != nil && block.Td.Cmp(td) != 0 { - err := fmt.Errorf("invalid TD on block(%v) from peer(%s): block.td=%v, request.td=%v", block.Number(), p.id, block.Td, td) - glog.V(logger.Error).Infoln(err) - return err - } - pm.BroadcastBlock(hash, block) - return nil - } - // Parent of the block is unknown, try to sync with this peer if it seems to be good - if td != nil { - go pm.synchronise(p) - } - return nil -} - // BroadcastBlock will propagate the block to a subset of its connected peers, // only notifying the rest of the block's appearance. -func (pm *ProtocolManager) BroadcastBlock(hash common.Hash, block *types.Block) { +func (pm *ProtocolManager) BroadcastBlock(block *types.Block) { // Retrieve all the target peers and split between full broadcast or only notification - peers := pm.peers.PeersWithoutBlock(hash) + peers := pm.peers.PeersWithoutBlock(block.Hash()) split := int(math.Sqrt(float64(len(peers)))) transfer := peers[:split] @@ -413,7 +369,7 @@ func (pm *ProtocolManager) BroadcastBlock(hash common.Hash, block *types.Block) // Send out the data transfers and the notifications for _, peer := range notify { - peer.sendNewBlockHashes([]common.Hash{hash}) + peer.sendNewBlockHashes([]common.Hash{block.Hash()}) } glog.V(logger.Detail).Infoln("broadcast hash to", len(notify), "peers.") @@ -442,7 +398,7 @@ func (self *ProtocolManager) minedBroadcastLoop() { for obj := range self.minedBlockSub.Chan() { switch ev := obj.(type) { case core.NewMinedBlockEvent: - self.BroadcastBlock(ev.Block.Hash(), ev.Block) + self.BroadcastBlock(ev.Block) } } } -- cgit From 629705ad53ceac1bcc5f32f2d7ef51d7626aefef Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Thu, 18 Jun 2015 16:09:34 +0300 Subject: eth: clean the block request packet handling a bit --- eth/handler.go | 34 +++++++++++++++++----------------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index eeb6a4f14..f3fe5dfe3 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -252,33 +252,31 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } case GetBlocksMsg: - var blocks []*types.Block - + // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { return err } + // Gather blocks until the fetch or network limits is reached var ( - i int - totalsize common.StorageSize + hash common.Hash + bytes common.StorageSize + blocks []*types.Block ) for { - i++ - var hash common.Hash err := msgStream.Decode(&hash) if err == rlp.EOL { break } else if err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - - block := pm.chainman.GetBlock(hash) - if block != nil { + // Retrieve the requested block, stopping if enough was found + if block := pm.chainman.GetBlock(hash); block != nil { blocks = append(blocks, block) - totalsize += block.Size() - } - if i == downloader.MaxBlockFetch || totalsize > maxBlockRespSize { - break + bytes += block.Size() + if len(blocks) >= downloader.MaxBlockFetch || bytes > maxBlockRespSize { + break + } } } return p.sendBlocks(blocks) @@ -360,8 +358,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // BroadcastBlock will propagate the block to a subset of its connected peers, // only notifying the rest of the block's appearance. func (pm *ProtocolManager) BroadcastBlock(block *types.Block) { + hash := block.Hash() + // Retrieve all the target peers and split between full broadcast or only notification - peers := pm.peers.PeersWithoutBlock(block.Hash()) + peers := pm.peers.PeersWithoutBlock(hash) split := int(math.Sqrt(float64(len(peers)))) transfer := peers[:split] @@ -369,14 +369,14 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block) { // Send out the data transfers and the notifications for _, peer := range notify { - peer.sendNewBlockHashes([]common.Hash{block.Hash()}) + peer.sendNewBlockHashes([]common.Hash{hash}) } - glog.V(logger.Detail).Infoln("broadcast hash to", len(notify), "peers.") + glog.V(logger.Detail).Infof("broadcast hash %x to %d peers.", hash[:4], len(notify)) for _, peer := range transfer { peer.sendNewBlock(block) } - glog.V(logger.Detail).Infoln("broadcast block to", len(transfer), "peers. Total processing time:", time.Since(block.ReceivedAt)) + glog.V(logger.Detail).Infof("broadcast block %x to %d peers. Total processing time: %v", hash[:4], len(transfer), time.Since(block.ReceivedAt)) } // BroadcastTx will propagate the block to its connected peers. It will sort -- cgit From b91b581b80ec99dfa07b7206104faa919210fc4f Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Thu, 18 Jun 2015 18:00:19 +0300 Subject: eth, eth/fetcher: propagate after header verify, announce only on insert --- eth/backend.go | 2 +- eth/fetcher/fetcher.go | 43 +++++++++++++++++++++----------- eth/fetcher/fetcher_test.go | 18 +++++++------ eth/handler.go | 61 ++++++++++++++++++++++++++++++--------------- 4 files changed, 81 insertions(+), 43 deletions(-) diff --git a/eth/backend.go b/eth/backend.go index c621fa260..37fe66abf 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -313,7 +313,7 @@ func New(config *Config) (*Ethereum, error) { eth.blockProcessor = core.NewBlockProcessor(stateDb, extraDb, eth.pow, eth.chainManager, eth.EventMux()) eth.chainManager.SetProcessor(eth.blockProcessor) - eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.chainManager) + eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.pow, eth.chainManager) eth.miner = miner.New(eth, eth.EventMux(), eth.pow) eth.miner.SetGasPrice(config.GasPrice) 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) } diff --git a/eth/handler.go b/eth/handler.go index f3fe5dfe3..88e340256 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -6,6 +6,8 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/pow" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" @@ -67,7 +69,7 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. -func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, chainman *core.ChainManager) *ProtocolManager { +func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager) *ProtocolManager { // Create the protocol manager and initialize peer handlers manager := &ProtocolManager{ eventMux: mux, @@ -91,10 +93,13 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo // Construct the different synchronisation mechanisms manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) + validator := func(block *types.Block, parent *types.Block) error { + return core.ValidateHeader(pow, block.Header(), parent.Header(), true) + } heighter := func() uint64 { return manager.chainman.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(manager.chainman.HasBlock, manager.BroadcastBlock, heighter, manager.chainman.InsertChain, manager.removePeer) + manager.fetcher = fetcher.New(manager.chainman.GetBlock, validator, manager.BroadcastBlock, heighter, manager.chainman.InsertChain, manager.removePeer) return manager } @@ -261,6 +266,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { var ( hash common.Hash bytes common.StorageSize + hashes []common.Hash blocks []*types.Block ) for { @@ -270,6 +276,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } else if err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } + hashes = append(hashes, hash) + // Retrieve the requested block, stopping if enough was found if block := pm.chainman.GetBlock(hash); block != nil { blocks = append(blocks, block) @@ -279,6 +287,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } } + if glog.V(logger.Detail) && len(blocks) == 0 && len(hashes) > 0 { + list := "[" + for _, hash := range hashes { + list += fmt.Sprintf("%x, ", hash[:4]) + } + list = list[:len(list)-2] + "]" + + glog.Infof("Peer %s: no blocks found for requested hashes %s", p.id, list) + } return p.sendBlocks(blocks) case BlocksMsg: @@ -290,6 +307,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { glog.V(logger.Detail).Infoln("Decode error", err) blocks = nil } + // Update the receive timestamp of each block + for i:=0; i 0 { pm.downloader.DeliverBlocks(p.id, blocks) @@ -355,28 +376,27 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return nil } -// BroadcastBlock will propagate the block to a subset of its connected peers, -// only notifying the rest of the block's appearance. -func (pm *ProtocolManager) BroadcastBlock(block *types.Block) { +// BroadcastBlock will either propagate a block to a subset of it's peers, or +// will only announce it's availability (depending what's requested). +func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) { hash := block.Hash() - - // Retrieve all the target peers and split between full broadcast or only notification peers := pm.peers.PeersWithoutBlock(hash) - split := int(math.Sqrt(float64(len(peers)))) - - transfer := peers[:split] - notify := peers[split:] - // Send out the data transfers and the notifications - for _, peer := range notify { - peer.sendNewBlockHashes([]common.Hash{hash}) + // If propagation is requested, send to a subset of the peer + if propagate { + transfer := peers[:int(math.Sqrt(float64(len(peers))))] + for _, peer := range transfer { + peer.sendNewBlock(block) + } + glog.V(logger.Detail).Infof("propagated block %x to %d peers in %v", hash[:4], len(transfer), time.Since(block.ReceivedAt)) } - glog.V(logger.Detail).Infof("broadcast hash %x to %d peers.", hash[:4], len(notify)) - - for _, peer := range transfer { - peer.sendNewBlock(block) + // Otherwise if the block is indeed in out own chain, announce it + if pm.chainman.HasBlock(hash) { + for _, peer := range peers { + peer.sendNewBlockHashes([]common.Hash{hash}) + } + glog.V(logger.Detail).Infof("announced block %x to %d peers in %v", hash[:4], len(peers), time.Since(block.ReceivedAt)) } - glog.V(logger.Detail).Infof("broadcast block %x to %d peers. Total processing time: %v", hash[:4], len(transfer), time.Since(block.ReceivedAt)) } // BroadcastTx will propagate the block to its connected peers. It will sort @@ -398,7 +418,8 @@ func (self *ProtocolManager) minedBroadcastLoop() { for obj := range self.minedBlockSub.Chan() { switch ev := obj.(type) { case core.NewMinedBlockEvent: - self.BroadcastBlock(ev.Block) + self.BroadcastBlock(ev.Block, false) + self.BroadcastBlock(ev.Block, true) } } } -- cgit From 90d45f0397666cce42657849257a6f4f7cd2381c Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Thu, 18 Jun 2015 18:25:27 +0300 Subject: eth: fix test breakage from the previous commit --- eth/handler.go | 1 - eth/protocol_test.go | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index 88e340256..a5986b779 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -7,7 +7,6 @@ import ( "time" "github.com/ethereum/go-ethereum/pow" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 69d487c71..6e0eef59c 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -167,7 +167,7 @@ func newProtocolManagerForTesting(txAdded chan<- []*types.Transaction) *Protocol db, _ = ethdb.NewMemDatabase() chain, _ = core.NewChainManager(core.GenesisBlock(0, db), db, db, core.FakePow{}, em) txpool = &fakeTxPool{added: txAdded} - pm = NewProtocolManager(ProtocolVersion, 0, em, txpool, chain) + pm = NewProtocolManager(ProtocolVersion, 0, em, txpool, core.FakePow{}, chain) ) pm.Start() return pm -- cgit From ecd19919c5ec7118862fc88e2bfac19d4abbff53 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Thu, 18 Jun 2015 19:43:47 +0300 Subject: eth/fetcher: allow backward uncle imports too --- eth/fetcher/fetcher.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index d5ff5d77e..98170cf79 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -16,6 +16,7 @@ import ( 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 + maxUncleDist = 7 // Maximum allowed backward distance from the chain head maxQueueDist = 256 // Maximum allowed distance from the chain head to queue ) @@ -202,7 +203,7 @@ func (f *Fetcher) loop() { break } // Otherwise if fresh and still unknown, try and import - if number <= height || f.getBlock(op.block.Hash()) != nil { + if number+maxUncleDist < height || f.getBlock(op.block.Hash()) != nil { continue } f.insert(op.origin, op.block) @@ -317,7 +318,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { hash := block.Hash() // Discard any past or too distant blocks - if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist <= 0 || dist > maxQueueDist { + if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { glog.V(logger.Detail).Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist) return } -- cgit From 13c25036ea6afc0b13d1d07be6bc298f2ee24216 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Thu, 18 Jun 2015 20:10:07 +0300 Subject: eth/fetcher: since uncles are allowed, drop phase test --- eth/fetcher/fetcher_test.go | 51 --------------------------------------------- 1 file changed, 51 deletions(-) diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index cdf875c5c..0d069ac65 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -395,54 +395,3 @@ func TestDistantDiscarding(t *testing.T) { t.Fatalf("fetcher queued future block") } } - -// Tests that if multiple uncles (i.e. blocks at the same height) are queued for -// importing, then they will get inserted in phases, previous heights needing to -// complete before the next numbered blocks can begin. -func TestCompetingImports(t *testing.T) { - // Generate a few soft-forks for concurrent imports - hashesA := createHashes(16, knownHash) - hashesB := createHashes(16, knownHash) - hashesC := createHashes(16, knownHash) - - blocksA := createBlocksFromHashes(hashesA) - blocksB := createBlocksFromHashes(hashesB) - blocksC := createBlocksFromHashes(hashesC) - - // Create a tester, and override the import to check number reversals - tester := newTester() - - first := int32(1) - height := uint64(1) - tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) { - // Check for any phase reordering - if prev := atomic.LoadUint64(&height); blocks[0].NumberU64() < prev { - t.Errorf("phase reversal: have %v, want %v", blocks[0].NumberU64(), prev) - } - 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.insertChain(blocks) - } - // Queue up everything but with a missing link - for i := 0; i < len(hashesA)-2; i++ { - tester.fetcher.Enqueue("chain A", blocksA[hashesA[i]]) - tester.fetcher.Enqueue("chain B", blocksB[hashesB[i]]) - tester.fetcher.Enqueue("chain C", blocksC[hashesC[i]]) - } - // Add the three missing links, and wait for a full import - tester.fetcher.Enqueue("chain A", blocksA[hashesA[len(hashesA)-2]]) - tester.fetcher.Enqueue("chain B", blocksB[hashesB[len(hashesB)-2]]) - tester.fetcher.Enqueue("chain C", blocksC[hashesC[len(hashesC)-2]]) - - start := time.Now() - for len(tester.hashes) != len(hashesA) && time.Since(start) < time.Second { - time.Sleep(50 * time.Millisecond) - } - if len(tester.hashes) != len(hashesA) { - t.Fatalf("chain length mismatch: have %v, want %v", len(tester.hashes), len(hashesA)) - } -} -- cgit