From 7c2af1c11722dc3175a98342c060afcfaf6a275f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 16 Jun 2015 11:58:32 +0300 Subject: eth, eth/fetcher: separate notification sync mechanism --- eth/fetcher/fetcher.go | 258 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 258 insertions(+) create mode 100644 eth/fetcher/fetcher.go (limited to 'eth/fetcher') 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 + } + } + }() + } + } + } +} -- cgit v1.2.3 From 2a1b722d048d00401084c37a5ca12612f1dd5fcd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 16 Jun 2015 14:02:43 +0300 Subject: eth/fetcher: fix timer reset bug, add initial tests --- eth/fetcher/fetcher.go | 8 +- eth/fetcher/fetcher_test.go | 202 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 207 insertions(+), 3 deletions(-) create mode 100644 eth/fetcher/fetcher_test.go (limited to 'eth/fetcher') 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 v1.2.3 From 8b64e041d6c41d76994510fdd8bb42ad7c5be5aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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(+) (limited to 'eth/fetcher') 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 v1.2.3 From 057bc237adf9ed0adf615a72cc1e92d9aed15d9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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 +++++++++++++++++++- 2 files changed, 82 insertions(+), 31 deletions(-) (limited to 'eth/fetcher') 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) + } +} -- cgit v1.2.3 From 11c8f83a583745b01a4c06cdd29529af1df364f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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 ++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+) (limited to 'eth/fetcher') 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) + } +} -- cgit v1.2.3 From 026ee40650bde909c962f15e32bdea6c1d6e978a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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(-) (limited to 'eth/fetcher') 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 v1.2.3 From 497a7f1717a798e59c9550fbb58504b08fe13b21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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 +++++++++++++++++++++++ 2 files changed, 60 insertions(+), 23 deletions(-) (limited to 'eth/fetcher') 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") + } +} -- cgit v1.2.3 From 2a7411bc9605dae9798ed29ed237e28c8fc98895 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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(-) (limited to 'eth/fetcher') 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 v1.2.3 From 37c5ff392f71fddaa6acd92925f69e81876fe22e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 30 deletions(-) (limited to 'eth/fetcher') 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)) + } +} -- cgit v1.2.3 From a9ada0b5baa9c47f8673cb63707797cbd08f0d61 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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(-) (limited to 'eth/fetcher') 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 v1.2.3 From 5ec6ecc511dc861c0395335f06883bc9ac650e4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 17 Jun 2015 18:25:23 +0300 Subject: eth, eth/fetcher: move propagated block import into fetcher --- eth/fetcher/fetcher.go | 55 ++++++++++++++++++++++++++-------------- eth/fetcher/fetcher_test.go | 62 ++++++++++++++++++++++++++------------------- 2 files changed, 72 insertions(+), 45 deletions(-) (limited to 'eth/fetcher') diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index a70fcbeed..c96471554 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -29,12 +29,18 @@ type hashCheckFn func(common.Hash) bool // blockRequesterFn is a callback type for sending a block retrieval request. type blockRequesterFn func([]common.Hash) error -// blockImporterFn is a callback type for trying to inject a block into the local chain. -type blockImporterFn func(peer string, block *types.Block) error +// blockBroadcasterFn is a callback type for broadcasting a block to connected peers. +type blockBroadcasterFn func(block *types.Block) // chainHeightFn is a callback type to retrieve the current chain height. type chainHeightFn func() uint64 +// chainInsertFn is a callback type to insert a batch of blocks into the local chain. +type chainInsertFn func(types.Blocks) (int, error) + +// peerDropFn is a callback type for dropping a peer detected as malicious. +type peerDropFn func(id string) + // announce is the hash notification of the availability of a new block in the // network. type announce struct { @@ -70,26 +76,30 @@ type Fetcher struct { queued map[common.Hash]struct{} // Presence set of already queued blocks (to dedup imports) // Callbacks - hasBlock hashCheckFn // Checks if a block is present in the chain - importBlock blockImporterFn // Injects a block from an origin peer into the chain - chainHeight chainHeightFn // Retrieves the current chain's height + hasBlock hashCheckFn // Checks if a block is present in the chain + broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers + chainHeight chainHeightFn // Retrieves the current chain's height + insertChain chainInsertFn // Injects a batch of blocks into the chain + dropPeer peerDropFn // Drops a peer for misbehaving } // New creates a block fetcher to retrieve blocks based on hash announcements. -func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHeightFn) *Fetcher { +func New(hasBlock hashCheckFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { return &Fetcher{ - notify: make(chan *announce), - inject: make(chan *inject), - filter: make(chan chan []*types.Block), - done: make(chan common.Hash), - quit: make(chan struct{}), - announced: make(map[common.Hash][]*announce), - fetching: make(map[common.Hash]*announce), - queue: prque.New(), - queued: make(map[common.Hash]struct{}), - hasBlock: hasBlock, - importBlock: importBlock, - chainHeight: chainHeight, + notify: make(chan *announce), + inject: make(chan *inject), + filter: make(chan chan []*types.Block), + done: make(chan common.Hash), + quit: make(chan struct{}), + announced: make(map[common.Hash][]*announce), + fetching: make(map[common.Hash]*announce), + queue: prque.New(), + queued: make(map[common.Hash]struct{}), + hasBlock: hasBlock, + broadcastBlock: broadcastBlock, + chainHeight: chainHeight, + insertChain: insertChain, + dropPeer: dropPeer, } } @@ -328,10 +338,17 @@ func (f *Fetcher) insert(peer string, block *types.Block) { go func() { defer func() { f.done <- hash }() + // If the parent's unknown, abort insertion + if !f.hasBlock(block.ParentHash()) { + return + } // Run the actual import and log any issues - if err := f.importBlock(peer, block); err != nil { + if _, err := f.insertChain(types.Blocks{block}); err != nil { glog.V(logger.Detail).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err) + f.dropPeer(peer) return } + // If import succeeded, broadcast the block + go f.broadcastBlock(block) }() } diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 4c6a1bf6a..3e8df1804 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -80,7 +80,7 @@ func newTester() *fetcherTester { hashes: []common.Hash{knownHash}, blocks: map[common.Hash]*types.Block{knownHash: genesis}, } - tester.fetcher = New(tester.hasBlock, tester.importBlock, tester.chainHeight) + tester.fetcher = New(tester.hasBlock, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer) tester.fetcher.Start() return tester @@ -95,23 +95,8 @@ func (f *fetcherTester) hasBlock(hash common.Hash) bool { return ok } -// importBlock injects a new blocks into the simulated chain. -func (f *fetcherTester) importBlock(peer string, block *types.Block) error { - f.lock.Lock() - defer f.lock.Unlock() - - // Make sure the parent in known - if _, ok := f.blocks[block.ParentHash()]; !ok { - return errors.New("unknown parent") - } - // Discard any new blocks if the same height already exists - if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() { - return nil - } - // Otherwise build our current chain - f.hashes = append(f.hashes, block.Hash()) - f.blocks[block.Hash()] = block - return nil +// broadcastBlock is a nop placeholder for the block broadcasting. +func (f *fetcherTester) broadcastBlock(block *types.Block) { } // chainHeight retrieves the current height (block number) of the chain. @@ -122,6 +107,31 @@ func (f *fetcherTester) chainHeight() uint64 { return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() } +// insertChain injects a new blocks into the simulated chain. +func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) { + f.lock.Lock() + defer f.lock.Unlock() + + for i, block := range blocks { + // Make sure the parent in known + if _, ok := f.blocks[block.ParentHash()]; !ok { + return i, errors.New("unknown parent") + } + // Discard any new blocks if the same height already exists + if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() { + return i, nil + } + // Otherwise build our current chain + f.hashes = append(f.hashes, block.Hash()) + f.blocks[block.Hash()] = block + } + return 0, nil +} + +// dropPeer is a nop placeholder for the peer removal. +func (f *fetcherTester) dropPeer(peer string) { +} + // peerFetcher retrieves a fetcher associated with a simulated peer. func (f *fetcherTester) makeFetcher(blocks map[common.Hash]*types.Block) blockRequesterFn { // Copy all the blocks to ensure they are not tampered with @@ -330,9 +340,9 @@ func TestImportDeduplication(t *testing.T) { fetcher := tester.makeFetcher(blocks) counter := uint32(0) - tester.fetcher.importBlock = func(peer string, block *types.Block) error { - atomic.AddUint32(&counter, 1) - return tester.importBlock(peer, block) + tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) { + atomic.AddUint32(&counter, uint32(len(blocks))) + return tester.insertChain(blocks) } // Announce the duplicating block, wait for retrieval, and also propagate directly tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher) @@ -400,18 +410,18 @@ func TestCompetingImports(t *testing.T) { first := int32(1) height := uint64(1) - tester.fetcher.importBlock = func(peer string, block *types.Block) error { + tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) { // Check for any phase reordering - if prev := atomic.LoadUint64(&height); block.NumberU64() < prev { - t.Errorf("phase reversal: have %v, want %v", block.NumberU64(), prev) + if prev := atomic.LoadUint64(&height); blocks[0].NumberU64() < prev { + t.Errorf("phase reversal: have %v, want %v", blocks[0].NumberU64(), prev) } - atomic.StoreUint64(&height, block.NumberU64()) + atomic.StoreUint64(&height, blocks[0].NumberU64()) // Sleep a bit on the first import not to race with the enqueues if atomic.CompareAndSwapInt32(&first, 1, 0) { time.Sleep(50 * time.Millisecond) } - return tester.importBlock(peer, block) + return tester.insertChain(blocks) } // Queue up everything but with a missing link for i := 0; i < len(hashesA)-2; i++ { -- cgit v1.2.3 From b91b581b80ec99dfa07b7206104faa919210fc4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 18 Jun 2015 18:00:19 +0300 Subject: eth, eth/fetcher: propagate after header verify, announce only on insert --- eth/fetcher/fetcher.go | 43 ++++++++++++++++++++++++++++--------------- eth/fetcher/fetcher_test.go | 18 +++++++++++------- 2 files changed, 39 insertions(+), 22 deletions(-) (limited to 'eth/fetcher') 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) } -- cgit v1.2.3 From ecd19919c5ec7118862fc88e2bfac19d4abbff53 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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(-) (limited to 'eth/fetcher') 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 v1.2.3 From 13c25036ea6afc0b13d1d07be6bc298f2ee24216 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= 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(-) (limited to 'eth/fetcher') 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 v1.2.3