eth/downloader: move pivot block splitting to processContent

This change also ensures that pivot block receipts aren't imported
before the pivot block itself.

eth/downloader/downloader.go

@@ -428,7 +428,7 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
 			pivot = d.fsPivotLock.Number.Uint64()
 		}
 		// If the point is below the origin, move origin back to ensure state download
-		if pivot <= origin {
+		if pivot < origin {
 			if pivot > 0 {
 				origin = pivot - 1
 			} else {
@@ -442,18 +442,28 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
 		d.syncInitHook(origin, height)
 	}
 
-	return d.spawnSync(origin+1,
+	fetchers := []func() error{
 		func() error { return d.fetchHeaders(p, origin+1) }, // Headers are always retrieved
 		func() error { return d.fetchBodies(origin + 1) },   // Bodies are retrieved during normal and fast sync
 		func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during fast sync
-		func() error { return d.processHeaders(origin+1, td) }, // Headers are always retrieved
-		func() error { return d.processContent(latest) },
-	)
+		func() error { return d.processHeaders(origin+1, td) },
+	}
+	if d.mode == FastSync {
+		fetchers = append(fetchers, func() error { return d.processFastSyncContent(latest) })
+	} else {
+		fetchers = append(fetchers, d.processFullSyncContent)
+	}
+	err = d.spawnSync(fetchers)
+	if err != nil && d.mode == FastSync && d.fsPivotLock != nil {
+		// If sync failed in the critical section, bump the fail counter.
+		atomic.AddUint32(&d.fsPivotFails, 1)
+	}
+	return err
 }
 
 // spawnSync runs d.process and all given fetcher functions to completion in
 // separate goroutines, returning the first error that appears.
-func (d *Downloader) spawnSync(origin uint64, fetchers ...func() error) error {
+func (d *Downloader) spawnSync(fetchers []func() error) error {
 	var wg sync.WaitGroup
 	errc := make(chan error, len(fetchers))
 	wg.Add(len(fetchers))
@@ -477,11 +487,6 @@ func (d *Downloader) spawnSync(origin uint64, fetchers ...func() error) error {
 	d.queue.Close()
 	d.Cancel()
 	wg.Wait()
-
-	// If sync failed in the critical section, bump the fail counter
-	if err != nil && d.mode == FastSync && d.fsPivotLock != nil {
-		atomic.AddUint32(&d.fsPivotFails, 1)
-	}
 	return err
 }
 
@@ -1288,103 +1293,150 @@ func (d *Downloader) processHeaders(origin uint64, td *big.Int) error {
 	}
 }
 
-// processContent takes fetch results from the queue and tries to import them
-// into the chain. The type of import operation will depend on the result contents.
-func (d *Downloader) processContent(head *types.Header) error {
-	var stateSync *stateSync
-	if d.mode == FastSync {
-		// Start syncing state of the reported head block.
-		// This should get us most of the state of the pivot block.
-		stateSync = d.syncState(head.Root)
-		defer stateSync.Cancel()
-		go func() {
-			if err := stateSync.wait(); err != nil {
-				d.queue.Close() // wake up WaitResults
-			}
-		}()
-	}
-
-	pivot := d.queue.FastSyncPivot()
+// processBlocks takes fetch results from the queue and tries to import them
+// into the chain.
+func (d *Downloader) processFullSyncContent() error {
 	for {
 		results := d.queue.WaitResults()
 		if len(results) == 0 {
-			if stateSync != nil {
-				return stateSync.Cancel()
-			}
 			return nil
 		}
 		if d.chainInsertHook != nil {
 			d.chainInsertHook(results)
 		}
-		// Actually import the blocks
-		first, last := results[0].Header, results[len(results)-1].Header
-		log.Debug("Inserting downloaded chain", "items", len(results),
-			"firstnum", first.Number, "firsthash", first.Hash(),
-			"lastnum", last.Number, "lasthash", last.Hash(),
-		)
+		if err := d.importBlockResults(results); err != nil {
+			return err
+		}
+	}
+}
+
+func (d *Downloader) importBlockResults(results []*fetchResult) error {
 	for len(results) != 0 {
-			// Check for any termination requests
+		// Check for any termination requests. This makes clean shutdown faster.
 		select {
 		case <-d.quitCh:
 			return errCancelContentProcessing
 		default:
 		}
-			if stateSync != nil {
-				if _, err := stateSync.checkDone(); err != nil {
-					return err
-				}
-			}
 		// Retrieve the a batch of results to import
-			var (
-				blocks   = make([]*types.Block, 0, maxResultsProcess)
-				receipts = make([]types.Receipts, 0, maxResultsProcess)
-			)
 		items := int(math.Min(float64(len(results)), float64(maxResultsProcess)))
-			for _, result := range results[:items] {
-				switch {
-				case d.mode == FullSync:
-					blocks = append(blocks, types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles))
-				case d.mode == FastSync:
-					blocks = append(blocks, types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles))
-					if result.Header.Number.Uint64() <= pivot {
-						receipts = append(receipts, result.Receipts)
-					}
-				}
-			}
-			// Try to process the results, aborting if there's an error
-			var (
-				err   error
-				index int
+		first, last := results[0].Header, results[items-1].Header
+		log.Debug("Inserting downloaded chain", "items", len(results),
+			"firstnum", first.Number, "firsthash", first.Hash(),
+			"lastnum", last.Number, "lasthash", last.Hash(),
 		)
-			switch {
-			case len(receipts) > 0:
-				index, err = d.insertReceipts(blocks, receipts)
-				if err == nil && blocks[len(blocks)-1].NumberU64() == pivot {
-					index = len(blocks) - 1
-					stateSync.Cancel()
-					err = d.commitPivotBlock(blocks[len(blocks)-1])
+		blocks := make([]*types.Block, items)
+		for i, result := range results[:items] {
+			blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles)
 		}
-			default:
-				index, err = d.insertBlocks(blocks)
-			}
-			if err != nil {
+		if index, err := d.insertBlocks(blocks); err != nil {
 			log.Debug("Downloaded item processing failed", "number", results[index].Header.Number, "hash", results[index].Header.Hash(), "err", err)
 			return errInvalidChain
 		}
 		// Shift the results to the next batch
 		results = results[items:]
 	}
+	return nil
+}
+
+func (d *Downloader) processFastSyncContent(latest *types.Header) error {
+	// Start syncing state of the reported head block.
+	// This should get us most of the state of the pivot block.
+	stateSync := d.syncState(latest.Root)
+	defer stateSync.Cancel()
+	go func() {
+		if err := stateSync.Wait(); err != nil {
+			d.queue.Close() // wake up WaitResults
+		}
+	}()
+
+	pivot := d.queue.FastSyncPivot()
+	for {
+		results := d.queue.WaitResults()
+		if len(results) == 0 {
+			return stateSync.Cancel()
+		}
+		if d.chainInsertHook != nil {
+			d.chainInsertHook(results)
+		}
+		P, beforeP, afterP := splitAroundPivot(pivot, results)
+		if err := d.commitFastSyncData(beforeP, stateSync); err != nil {
+			return err
+		}
+		if P != nil {
+			stateSync.Cancel()
+			if err := d.commitPivotBlock(P); err != nil {
+				return err
+			}
+		}
+		if err := d.importBlockResults(afterP); err != nil {
+			return err
+		}
 	}
 }
 
-func (d *Downloader) commitPivotBlock(block *types.Block) error {
-	// Sync the pivot block state. This should complete reasonably quickly because
-	// we've already synced up to the reported head block state earlier.
-	if err := d.syncState(block.Root()).wait(); err != nil {
+func splitAroundPivot(pivot uint64, results []*fetchResult) (p *fetchResult, before, after []*fetchResult) {
+	for _, result := range results {
+		num := result.Header.Number.Uint64()
+		switch {
+		case num < pivot:
+			before = append(before, result)
+		case num == pivot:
+			p = result
+		default:
+			after = append(after, result)
+		}
+	}
+	return p, before, after
+}
+
+func (d *Downloader) commitFastSyncData(results []*fetchResult, stateSync *stateSync) error {
+	for len(results) != 0 {
+		// Check for any termination requests.
+		select {
+		case <-d.quitCh:
+			return errCancelContentProcessing
+		case <-stateSync.done:
+			if err := stateSync.Wait(); err != nil {
 				return err
 			}
-	log.Debug("Committing block as new head", "number", block.Number(), "hash", block.Hash())
-	return d.commitHeadBlock(block.Hash())
+		default:
+		}
+		// Retrieve the a batch of results to import
+		items := int(math.Min(float64(len(results)), float64(maxResultsProcess)))
+		first, last := results[0].Header, results[items-1].Header
+		log.Debug("Inserting fast-sync blocks", "items", len(results),
+			"firstnum", first.Number, "firsthash", first.Hash(),
+			"lastnumn", last.Number, "lasthash", last.Hash(),
+		)
+		blocks := make([]*types.Block, items)
+		receipts := make([]types.Receipts, items)
+		for i, result := range results[:items] {
+			blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles)
+			receipts[i] = result.Receipts
+		}
+		if index, err := d.insertReceipts(blocks, receipts); err != nil {
+			log.Debug("Downloaded item processing failed", "number", results[index].Header.Number, "hash", results[index].Header.Hash(), "err", err)
+			return errInvalidChain
+		}
+		// Shift the results to the next batch
+		results = results[items:]
+	}
+	return nil
+}
+
+func (d *Downloader) commitPivotBlock(result *fetchResult) error {
+	b := types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles)
+	// Sync the pivot block state. This should complete reasonably quickly because
+	// we've already synced up to the reported head block state earlier.
+	if err := d.syncState(b.Root()).Wait(); err != nil {
+		return err
+	}
+	log.Debug("Committing fast sync pivot as new head", "number", b.Number(), "hash", b.Hash())
+	if _, err := d.insertReceipts([]*types.Block{b}, []types.Receipts{result.Receipts}); err != nil {
+		return err
+	}
+	return d.commitHeadBlock(b.Hash())
 }
 
 // DeliverHeaders injects a new batch of block headers received from a remote

eth/downloader/queue.go

@@ -377,27 +377,17 @@ func (q *queue) countProcessableItems() int {
 		if result == nil || result.Pending > 0 {
 			return i
 		}
-		// Special handling for the fast-sync pivot block:
-		if q.mode == FastSync {
-			bnum := result.Header.Number.Uint64()
-			if bnum == q.fastSyncPivot {
 		// Stop before processing the pivot block to ensure that
 		// resultCache has space for fsHeaderForceVerify items. Not
 		// doing this could leave us unable to download the required
 		// amount of headers.
+		if q.mode == FastSync && result.Header.Number.Uint64() == q.fastSyncPivot {
 			for j := 0; j < fsHeaderForceVerify; j++ {
 				if i+j+1 >= len(q.resultCache) || q.resultCache[i+j+1] == nil {
 					return i
 				}
 			}
 		}
-			// If we're just the after fast sync pivot, stop as well
-			// because the following batch needs different insertion.
-			// This simplifies handling the switchover in d.processContent.
-			if bnum == q.fastSyncPivot+1 && i > 0 {
-				return i
-			}
-		}
 	}
 	return len(q.resultCache)
 }

eth/downloader/statesync.go

@@ -182,25 +182,16 @@ func newStateSync(d *Downloader, root common.Hash) *stateSync {
 	}
 }
 
-// wait blocks until the sync is done or canceled.
-func (s *stateSync) wait() error {
+// Wait blocks until the sync is done or canceled.
+func (s *stateSync) Wait() error {
 	<-s.done
 	return s.err
 }
 
-// wait blocks until the sync is done or canceled.
-func (s *stateSync) checkDone() (bool, error) {
-	select {
-	case <-s.done:
-		return true, s.err
-	default:
-		return false, nil
-	}
-}
-
+// Cancel cancels the sync and waits until it has shut down.
 func (s *stateSync) Cancel() error {
 	s.cancelOnce.Do(func() { close(s.cancel) })
-	return s.wait()
+	return s.Wait()
 }
 
 func (s *stateSync) run() {