From 4c6040c3342568ef8b1ece52aa461ef1fd0c3109 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Tue, 30 May 2017 15:03:19 +0200 Subject: [PATCH] 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 | 242 +++++++++++++++++++++-------------- eth/downloader/queue.go | 26 ++-- eth/downloader/statesync.go | 17 +-- 3 files changed, 159 insertions(+), 126 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 75610312e2..fc06cbd62e 100644 --- a/eth/downloader/downloader.go +++ b/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, - 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) }, - ) + 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) }, + } + 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(), - ) - for len(results) != 0 { - // Check for any termination requests - 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 - ) - 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]) - } - default: - index, err = d.insertBlocks(blocks) - } - if 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:] + if err := d.importBlockResults(results); err != nil { + return err } } } -func (d *Downloader) commitPivotBlock(block *types.Block) error { +func (d *Downloader) importBlockResults(results []*fetchResult) error { + for len(results) != 0 { + // Check for any termination requests. This makes clean shutdown faster. + select { + case <-d.quitCh: + return errCancelContentProcessing + 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 downloaded chain", "items", len(results), + "firstnum", first.Number, "firsthash", first.Hash(), + "lastnum", last.Number, "lasthash", last.Hash(), + ) + blocks := make([]*types.Block, items) + for i, result := range results[:items] { + blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles) + } + 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 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 + } + 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(block.Root()).wait(); err != nil { + if err := d.syncState(b.Root()).Wait(); err != nil { return err } - log.Debug("Committing block as new head", "number", block.Number(), "hash", block.Hash()) - return d.commitHeadBlock(block.Hash()) + 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 diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 63dba10194..9e59522658 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -377,26 +377,16 @@ 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. - for j := 0; j < fsHeaderForceVerify; j++ { - if i+j+1 >= len(q.resultCache) || q.resultCache[i+j+1] == nil { - return i - } + // 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) diff --git a/eth/downloader/statesync.go b/eth/downloader/statesync.go index 58e3a33bf4..6423235463 100644 --- a/eth/downloader/statesync.go +++ b/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() {