eth/fetcher: detect empty blocks and short circuit body fetch

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

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
+	fetchingHook   func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch
-	importedHook func(*types.Block)  // Method to call upon successful block import
+	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)
+	fetchTimer := time.NewTimer(0)
-	complete := 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)
+				f.rescheduleFetch(fetchTimer)
-			}
-
-		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)
-				}
 			}
 
 		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

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) {
+	blocks := core.GenerateChain(parent, testdb, n, func(i int, block *core.BlockGen) {
-		gen.SetCoinbase(common.Address{seed})
+		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.