diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 93fb342cf5..bfc7a39a2d 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -239,7 +239,7 @@ func (d *Downloader) UnregisterPeer(id string) error { // 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. func (d *Downloader) Synchronise(id string, head common.Hash, td *big.Int) { - glog.V(logger.Detail).Infof("Attempting synchronisation: %v, head 0x%x, TD %v", id, head[:4], td) + glog.V(logger.Detail).Infof("Attempting synchronisation: %v, head [%x…], TD %v", id, head[:4], td) switch err := d.synchronise(id, head, td); err { case nil: @@ -452,7 +452,7 @@ func (d *Downloader) findAncestor61(p *peer) (uint64, error) { } // If the head fetch already found an ancestor, return if !common.EmptyHash(hash) { - glog.V(logger.Debug).Infof("%v: common ancestor: #%d [%x]", p, number, hash[:4]) + glog.V(logger.Debug).Infof("%v: common ancestor: #%d [%x…]", p, number, hash[:4]) return number, nil } // Ancestor not found, we need to binary search over our chain @@ -491,7 +491,7 @@ func (d *Downloader) findAncestor61(p *peer) (uint64, error) { break } if block.NumberU64() != check { - glog.V(logger.Debug).Infof("%v: non requested hash #%d [%x], instead of #%d", p, block.NumberU64(), block.Hash().Bytes()[:4], check) + glog.V(logger.Debug).Infof("%v: non requested hash #%d [%x…], instead of #%d", p, block.NumberU64(), block.Hash().Bytes()[:4], check) return 0, errBadPeer } start = check @@ -821,7 +821,7 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) { } // If the head fetch already found an ancestor, return if !common.EmptyHash(hash) { - glog.V(logger.Debug).Infof("%v: common ancestor: #%d [%x]", p, number, hash[:4]) + glog.V(logger.Debug).Infof("%v: common ancestor: #%d [%x…]", p, number, hash[:4]) return number, nil } // Ancestor not found, we need to binary search over our chain @@ -860,7 +860,7 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) { break } if block.NumberU64() != check { - glog.V(logger.Debug).Infof("%v: non requested header #%d [%x], instead of #%d", p, block.NumberU64(), block.Hash().Bytes()[:4], check) + glog.V(logger.Debug).Infof("%v: non requested header #%d [%x…], instead of #%d", p, block.NumberU64(), block.Hash().Bytes()[:4], check) return 0, errBadPeer } start = check diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 6c8d652269..3239d33214 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -142,8 +142,9 @@ type Fetcher struct { dropPeer peerDropFn // Drops a peer for misbehaving // Testing hooks - fetchingHook func([]common.Hash) // Method to call upon starting a block fetch - importedHook func(*types.Block) // Method to call upon successful block import + fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch + completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62) + importedHook func(*types.Block) // Method to call upon successful block import (both eth/61 and eth/62) } // New creates a block fetcher to retrieve blocks based on hash announcements. @@ -309,8 +310,8 @@ func (f *Fetcher) FilterBodies(transactions [][]*types.Transaction, uncles [][]* // events. func (f *Fetcher) loop() { // Iterate the block fetching until a quit is requested - fetch := time.NewTimer(0) - complete := time.NewTimer(0) + fetchTimer := time.NewTimer(0) + completeTimer := time.NewTimer(0) for { // Clean up any expired block fetches @@ -356,7 +357,7 @@ func (f *Fetcher) loop() { // If we have a valid block number, check that it's potentially useful if notification.number > 0 { if dist := int64(notification.number) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { - glog.V(logger.Debug).Infof("[eth/62] Peer %s: discarded announcement #%d [%x], distance %d", notification.origin, notification.number, notification.hash[:4], dist) + glog.V(logger.Debug).Infof("[eth/62] Peer %s: discarded announcement #%d [%x…], distance %d", notification.origin, notification.number, notification.hash[:4], dist) discardMeter.Mark(1) break } @@ -371,52 +372,7 @@ func (f *Fetcher) loop() { f.announces[notification.origin] = count f.announced[notification.hash] = append(f.announced[notification.hash], notification) if len(f.announced) == 1 { - f.rescheduleFetch(fetch) - } - - case filter := <-f.headerFilter: - // Headers arrived, extract any explicit fetches, return all else - var task *headerFilterTask - select { - case task = <-filter: - case <-f.quit: - return - } - - explicit, download := []*announce{}, []*types.Header{} - for _, header := range task.headers { - hash := header.Hash() - - // Filter explicitly requested headers from hash announcements - if announce := f.fetching[hash]; announce != nil && f.fetched[hash] == nil && f.completing[hash] == nil && f.queued[hash] == nil { - // Discard if already imported by other means - if f.getBlock(hash) == nil { - announce.header = header - announce.time = task.time - explicit = append(explicit, announce) - } else { - f.forgetHash(hash) - } - } else { - download = append(download, header) - } - } - - select { - case filter <- &headerFilterTask{headers: download, time: task.time}: - case <-f.quit: - return - } - // Schedule the retrieved headers for body completion - for _, announce := range explicit { - hash := announce.header.Hash() - if _, ok := f.completing[hash]; ok { - continue - } - f.fetched[hash] = append(f.fetched[hash], announce) - if len(f.fetched) == 1 { - f.rescheduleComplete(complete) - } + f.rescheduleFetch(fetchTimer) } case op := <-f.inject: @@ -429,7 +385,7 @@ func (f *Fetcher) loop() { f.forgetHash(hash) f.forgetBlock(hash) - case <-fetch.C: + case <-fetchTimer.C: // At least one block's timer ran out, check for needing retrieval request := make(map[string][]common.Hash) @@ -451,7 +407,7 @@ func (f *Fetcher) loop() { if glog.V(logger.Detail) && len(hashes) > 0 { list := "[" for _, hash := range hashes { - list += fmt.Sprintf("%x, ", hash[:4]) + list += fmt.Sprintf("%x…, ", hash[:4]) } list = list[:len(list)-2] + "]" @@ -479,9 +435,9 @@ func (f *Fetcher) loop() { }() } // Schedule the next fetch if blocks are still pending - f.rescheduleFetch(fetch) + f.rescheduleFetch(fetchTimer) - case <-complete.C: + case <-completeTimer.C: // At least one header's timer ran out, retrieve everything request := make(map[string][]common.Hash) @@ -501,17 +457,20 @@ func (f *Fetcher) loop() { if glog.V(logger.Detail) && len(hashes) > 0 { list := "[" for _, hash := range hashes { - list += fmt.Sprintf("%x, ", hash[:4]) + list += fmt.Sprintf("%x…, ", hash[:4]) } list = list[:len(list)-2] + "]" glog.V(logger.Detail).Infof("[eth/62] Peer %s: fetching bodies %s", peer, list) } // Create a closure of the fetch and schedule in on a new thread + if f.completingHook != nil { + f.completingHook(hashes) + } go f.completing[hashes[0]].fetchBodies(hashes) } // Schedule the next fetch if blocks are still pending - f.rescheduleComplete(complete) + f.rescheduleComplete(completeTimer) case filter := <-f.blockFilter: // Blocks arrived, extract any explicit fetches, return all else @@ -551,6 +510,71 @@ func (f *Fetcher) loop() { } } + case filter := <-f.headerFilter: + // Headers arrived from a remote peer. Extract those that were explicitly + // requested by the fetcher, and return everything else so it's delivered + // to other parts of the system. + var task *headerFilterTask + select { + case task = <-filter: + case <-f.quit: + return + } + // Split the batch of headers into unknown ones (to return to the caller), + // known incomplete ones (requiring body retrievals) and completed blocks. + unknown, incomplete, complete := []*types.Header{}, []*announce{}, []*types.Block{} + for _, header := range task.headers { + hash := header.Hash() + + // Filter fetcher-requested headers from other synchronisation algorthims + if announce := f.fetching[hash]; announce != nil && f.fetched[hash] == nil && f.completing[hash] == nil && f.queued[hash] == nil { + // Only keep if not imported by other means + if f.getBlock(hash) == nil { + announce.header = header + announce.time = task.time + + // If the block is empty (header only), short circuit into the final import queue + if header.TxHash == types.DeriveSha(types.Transactions{}) && header.UncleHash == types.CalcUncleHash([]*types.Header{}) { + glog.V(logger.Detail).Infof("[eth/62] Peer %s: block #%d [%x…] empty, skipping body retrieval", announce.origin, header.Number.Uint64(), header.Hash().Bytes()[:4]) + + complete = append(complete, types.NewBlockWithHeader(header)) + f.completing[hash] = announce + continue + } + // Otherwise add to the list of blocks needing completion + incomplete = append(incomplete, announce) + } else { + glog.V(logger.Detail).Infof("[eth/62] Peer %s: block #%d [%x…] already imported, discarding header", announce.origin, header.Number.Uint64(), header.Hash().Bytes()[:4]) + f.forgetHash(hash) + } + } else { + // Fetcher doesn't know about it, add to the return list + unknown = append(unknown, header) + } + } + select { + case filter <- &headerFilterTask{headers: unknown, time: task.time}: + case <-f.quit: + return + } + // Schedule the retrieved headers for body completion + for _, announce := range incomplete { + hash := announce.header.Hash() + if _, ok := f.completing[hash]; ok { + continue + } + f.fetched[hash] = append(f.fetched[hash], announce) + if len(f.fetched) == 1 { + f.rescheduleComplete(completeTimer) + } + } + // Schedule the header-only blocks for import + for _, block := range complete { + if announce := f.completing[block.Hash()]; announce != nil { + f.enqueue(announce.origin, block) + } + } + case filter := <-f.bodyFilter: // Block bodies arrived, extract any explicitly requested blocks, return the rest var task *bodyFilterTask @@ -645,13 +669,13 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { // Ensure the peer isn't DOSing us count := f.queues[peer] + 1 if count > blockLimit { - glog.V(logger.Debug).Infof("Peer %s: discarded block #%d [%x], exceeded allowance (%d)", peer, block.NumberU64(), hash.Bytes()[:4], blockLimit) + glog.V(logger.Debug).Infof("Peer %s: discarded block #%d [%x…], exceeded allowance (%d)", peer, block.NumberU64(), hash.Bytes()[:4], blockLimit) f.forgetHash(hash) return } // Discard any past or too distant blocks if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { - glog.V(logger.Debug).Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist) + glog.V(logger.Debug).Infof("Peer %s: discarded block #%d [%x…], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist) discardMeter.Mark(1) f.forgetHash(hash) return @@ -667,7 +691,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { f.queue.Push(op, -float32(block.NumberU64())) if glog.V(logger.Debug) { - glog.Infof("Peer %s: queued block #%d [%x], total %v", peer, block.NumberU64(), 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()) } } } @@ -679,14 +703,14 @@ 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]) + glog.V(logger.Debug).Infof("Peer %s: importing block #%d [%x…]", peer, block.NumberU64(), hash[:4]) go func() { defer func() { f.done <- hash }() // If the parent's unknown, abort insertion parent := f.getBlock(block.ParentHash()) if parent == nil { - glog.V(logger.Debug).Infof("Peer %s: parent []%x] of block #%d [%x] unknown", block.ParentHash().Bytes()[:4], peer, block.NumberU64(), hash[:4]) + glog.V(logger.Debug).Infof("Peer %s: parent []%x] of block #%d [%x…] unknown", block.ParentHash().Bytes()[:4], peer, block.NumberU64(), hash[:4]) return } // Quickly validate the header and propagate the block if it passes @@ -702,13 +726,13 @@ func (f *Fetcher) insert(peer string, block *types.Block) { default: // Something went very wrong, drop the peer - glog.V(logger.Debug).Infof("Peer %s: block #%d [%x] verification failed: %v", peer, block.NumberU64(), hash[:4], err) + glog.V(logger.Debug).Infof("Peer %s: block #%d [%x…] verification failed: %v", peer, block.NumberU64(), hash[:4], err) f.dropPeer(peer) return } // Run the actual import and log any issues if _, err := f.insertChain(types.Blocks{block}); err != nil { - glog.V(logger.Warn).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err) + 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 diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index dc7bdb7c93..0d5825ef41 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -27,21 +27,39 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/params" ) var ( testdb, _ = ethdb.NewMemDatabase() - genesis = core.GenesisBlockForTesting(testdb, common.Address{}, big.NewInt(0)) + testKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + testAddress = crypto.PubkeyToAddress(testKey.PublicKey) + genesis = core.GenesisBlockForTesting(testdb, testAddress, big.NewInt(1000000000)) unknownBlock = types.NewBlock(&types.Header{GasLimit: params.GenesisGasLimit}, nil, nil, nil) ) // makeChain creates a chain of n blocks starting at and including parent. -// the returned hash chain is ordered head->parent. +// the returned hash chain is ordered head->parent. In addition, every 3rd block +// contains a transaction and every 5th an uncle to allow testing correct block +// reassembly. func makeChain(n int, seed byte, parent *types.Block) ([]common.Hash, map[common.Hash]*types.Block) { - blocks := core.GenerateChain(parent, testdb, n, func(i int, gen *core.BlockGen) { - gen.SetCoinbase(common.Address{seed}) + blocks := core.GenerateChain(parent, testdb, n, func(i int, block *core.BlockGen) { + block.SetCoinbase(common.Address{seed}) + + // If the block number is multiple of 3, send a bonus transaction to the miner + if parent == genesis && i%3 == 0 { + tx, err := types.NewTransaction(block.TxNonce(testAddress), common.Address{seed}, big.NewInt(1000), params.TxGas, nil, nil).SignECDSA(testKey) + if err != nil { + panic(err) + } + block.AddTx(tx) + } + // If the block number is a multiple of 5, add a bonus uncle to the block + if i%5 == 0 { + block.AddUncle(&types.Header{ParentHash: block.PrevBlock(i - 1).Hash(), Number: big.NewInt(int64(i - 1))}) + } }) hashes := make([]common.Hash, n+1) hashes[len(hashes)-1] = parent.Hash() @@ -193,6 +211,40 @@ func (f *fetcherTester) makeBodyFetcher(blocks map[common.Hash]*types.Block, dri } } +// verifyFetchingEvent verifies that one single event arrive on an fetching channel. +func verifyFetchingEvent(t *testing.T, fetching chan []common.Hash, arrive bool) { + if arrive { + select { + case <-fetching: + case <-time.After(time.Second): + t.Fatalf("fetching timeout") + } + } else { + select { + case <-fetching: + t.Fatalf("fetching invoked") + case <-time.After(10 * time.Millisecond): + } + } +} + +// verifyCompletingEvent verifies that one single event arrive on an completing channel. +func verifyCompletingEvent(t *testing.T, completing chan []common.Hash, arrive bool) { + if arrive { + select { + case <-completing: + case <-time.After(time.Second): + t.Fatalf("completing timeout") + } + } else { + select { + case <-completing: + t.Fatalf("completing invoked") + case <-time.After(10 * time.Millisecond): + } + } +} + // verifyImportEvent verifies that one single event arrive on an import channel. func verifyImportEvent(t *testing.T, imported chan *types.Block) { select { @@ -226,6 +278,38 @@ func verifyImportDone(t *testing.T, imported chan *types.Block) { // TODO: Test that the block number matches with the advertised one +// Tests that a fetcher accepts block announcements and initiates retrievals for +// them, successfully importing into the local chain. +func TestSequentialAnnouncements61(t *testing.T) { testSequentialAnnouncements(t, 61) } +func TestSequentialAnnouncements62(t *testing.T) { testSequentialAnnouncements(t, 62) } +func TestSequentialAnnouncements63(t *testing.T) { testSequentialAnnouncements(t, 63) } +func TestSequentialAnnouncements64(t *testing.T) { testSequentialAnnouncements(t, 64) } + +func testSequentialAnnouncements(t *testing.T, protocol int) { + // Create a chain of blocks to import + targetBlocks := 4 * hashLimit + hashes, blocks := makeChain(targetBlocks, 0, genesis) + + tester := newTester() + blockFetcher := tester.makeBlockFetcher(blocks) + headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) + + // Iteratively announce blocks until all are imported + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 2; i >= 0; i-- { + if protocol < 62 { + tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), blockFetcher, nil, nil) + } else { + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher) + } + verifyImportEvent(t, imported) + } + verifyImportDone(t, imported) +} + // Tests that if blocks are announced by multiple peers (or even the same buggy // peer), they will only get downloaded at most once. func TestConcurrentAnnouncements61(t *testing.T) { testConcurrentAnnouncements(t, 61) } @@ -570,6 +654,46 @@ func testDistantAnnouncementDiscarding(t *testing.T, protocol int) { } } +// Tests that if a block is empty (i.e. header only), no body request should be +// made, and instead the header should be assembled into a whole block in itself. +func TestEmptyBlockShortCircuit62(t *testing.T) { testEmptyBlockShortCircuit(t, 62) } +func TestEmptyBlockShortCircuit63(t *testing.T) { testEmptyBlockShortCircuit(t, 63) } +func TestEmptyBlockShortCircuit64(t *testing.T) { testEmptyBlockShortCircuit(t, 64) } + +func testEmptyBlockShortCircuit(t *testing.T, protocol int) { + // Create a chain of blocks to import + hashes, blocks := makeChain(32, 0, genesis) + + tester := newTester() + headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) + + // Add a monitoring hook for all internal events + fetching := make(chan []common.Hash) + tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes } + + completing := make(chan []common.Hash) + tester.fetcher.completingHook = func(hashes []common.Hash) { completing <- hashes } + + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + // Iteratively announce blocks until all are imported + for i := len(hashes) - 2; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher) + + // All announces should fetch the header + verifyFetchingEvent(t, fetching, true) + + // Only blocks with data contents should request bodies + verifyCompletingEvent(t, completing, len(blocks[hashes[i]].Transactions()) > 0 || len(blocks[hashes[i]].Uncles()) > 0) + + // Irrelevant of the construct, import should succeed + verifyImportEvent(t, imported) + } + verifyImportDone(t, imported) +} + // Tests that a peer is unable to use unbounded memory with sending infinite // block announcements to a node, but that even in the face of such an attack, // the fetcher remains operational.