forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-28 23:56:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/downloader: separate state sync from queue

Browse source 
Scheduling of state node downloads hogged the downloader queue lock when
new requests were scheduled. This caused timeouts for other requests.
With this change, state sync is fully independent of all other downloads
and doesn't involve the queue at all.

State sync is started and checked on in processContent. This is slightly
awkward because processContent doesn't have a select loop. Instead, the
queue is closed by an auxiliary goroutine when state sync fails. We
tried several alternatives to this but settled on the current approach
because it's the least amount of change overall.

Handling of the pivot block has changed slightly: the queue previously
prevented import of pivot block receipts before the state of the pivot
block was available. In this commit, the receipt will be imported before
the state. This causes an annoyance where the pivot block is committed
as fast block head even when state downloads fail. Stay tuned for more
updates in this area ;)
This commit is contained in:
Felix Lange 2017-05-29 17:36:29 +02:00 committed by Péter Szilágyi
parent 431cf2a1e4
commit 1b19b21cb2
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
4 changed files with 419 additions and 318 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

169
eth/downloader/downloader.go
View file

@ -34,7 +34,6 @@ import (
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/trie"
"github.com/rcrowley/go-metrics"
)
@ -99,8 +98,9 @@ type Downloader struct {
mode SyncMode // Synchronisation mode defining the strategy used (per sync cycle)
mux *event.TypeMux // Event multiplexer to announce sync operation events
queue *queue // Scheduler for selecting the hashes to download
peers *peerSet // Set of active peers from which download can proceed
queue *queue // Scheduler for selecting the hashes to download
peers *peerSet // Set of active peers from which download can proceed
stateDB ethdb.Database
fsPivotLock *types.Header // Pivot header on critical section entry (cannot change between retries)
fsPivotFails uint32 // Number of subsequent fast sync failures in the critical section
@ -109,9 +109,9 @@ type Downloader struct {
rttConfidence uint64 // Confidence in the estimated RTT (unit: millionths to allow atomic ops)
// Statistics
syncStatsChainOrigin uint64 // Origin block number where syncing started at
syncStatsChainHeight uint64 // Highest block number known when syncing started
syncStatsStateDone uint64 // Number of state trie entries already pulled
syncStatsChainOrigin uint64 // Origin block number where syncing started at
syncStatsChainHeight uint64 // Highest block number known when syncing started
syncStatsState stateSyncStats
syncStatsLock sync.RWMutex // Lock protecting the sync stats fields
// Callbacks
@ -136,16 +136,18 @@ type Downloader struct {
notified int32
// Channels
newPeerCh chan *peer
headerCh chan dataPack // [eth/62] Channel receiving inbound block headers
bodyCh chan dataPack // [eth/62] Channel receiving inbound block bodies
receiptCh chan dataPack // [eth/63] Channel receiving inbound receipts
stateCh chan dataPack // [eth/63] Channel receiving inbound node state data
bodyWakeCh chan bool // [eth/62] Channel to signal the block body fetcher of new tasks
receiptWakeCh chan bool // [eth/63] Channel to signal the receipt fetcher of new tasks
stateWakeCh chan bool // [eth/63] Channel to signal the state fetcher of new tasks
headerProcCh chan []*types.Header // [eth/62] Channel to feed the header processor new tasks
// for stateFetcher
stateSyncStart chan *stateSync
trackStateReq chan *stateReq
stateCh chan dataPack // [eth/63] Channel receiving inbound node state data
// Cancellation and termination
cancelPeer string // Identifier of the peer currently being used as the master (cancel on drop)
cancelCh chan struct{} // Channel to cancel mid-flight syncs
@ -170,8 +172,9 @@ func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, hasHeader he
dl := &Downloader{
mode: mode,
mux: mux,
queue: newQueue(stateDb),
queue: newQueue(),
peers: newPeerSet(),
stateDB: stateDb,
rttEstimate: uint64(rttMaxEstimate),
rttConfidence: uint64(1000000),
hasHeader: hasHeader,
@ -188,18 +191,20 @@ func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, hasHeader he
insertReceipts: insertReceipts,
rollback: rollback,
dropPeer: dropPeer,
newPeerCh: make(chan *peer, 1),
headerCh: make(chan dataPack, 1),
bodyCh: make(chan dataPack, 1),
receiptCh: make(chan dataPack, 1),
stateCh: make(chan dataPack, 1),
bodyWakeCh: make(chan bool, 1),
receiptWakeCh: make(chan bool, 1),
stateWakeCh: make(chan bool, 1),
headerProcCh: make(chan []*types.Header, 1),
quitCh: make(chan struct{}),
// for stateFetcher
stateSyncStart: make(chan *stateSync),
trackStateReq: make(chan *stateReq),
stateCh: make(chan dataPack),
}
go dl.qosTuner()
go dl.stateFetcher()
return dl
}
@ -211,9 +216,6 @@ func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, hasHeader he
// of processed and the total number of known states are also returned. Otherwise
// these are zero.
func (d *Downloader) Progress() ethereum.SyncProgress {
// Fetch the pending state count outside of the lock to prevent unforeseen deadlocks
pendingStates := uint64(d.queue.PendingNodeData())
// Lock the current stats and return the progress
d.syncStatsLock.RLock()
defer d.syncStatsLock.RUnlock()
@ -231,8 +233,8 @@ func (d *Downloader) Progress() ethereum.SyncProgress {
StartingBlock: d.syncStatsChainOrigin,
CurrentBlock: current,
HighestBlock: d.syncStatsChainHeight,
PulledStates: d.syncStatsStateDone,
KnownStates: d.syncStatsStateDone + pendingStates,
PulledStates: d.syncStatsState.done,
KnownStates: d.syncStatsState.done + d.syncStatsState.pending,
}
}
@ -324,13 +326,13 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode
d.queue.Reset()
d.peers.Reset()
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh, d.stateWakeCh} {
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
select {
case <-ch:
default:
}
}
for _, ch := range []chan dataPack{d.headerCh, d.bodyCh, d.receiptCh, d.stateCh} {
for _, ch := range []chan dataPack{d.headerCh, d.bodyCh, d.receiptCh} {
for empty := false; !empty; {
select {
case <-ch:
@ -426,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 {
@ -439,12 +441,13 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
if d.syncInitHook != nil {
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.processHeaders(origin+1, td) }, // 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.fetchNodeData() }, // Node state data is retrieved during fast sync
func() error { return d.processHeaders(origin+1, td) }, // Headers are always retrieved
func() error { return d.processContent(latest) },
)
}
@ -452,17 +455,16 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
// separate goroutines, returning the first error that appears.
func (d *Downloader) spawnSync(origin uint64, fetchers ...func() error) error {
var wg sync.WaitGroup
errc := make(chan error, len(fetchers)+1)
wg.Add(len(fetchers) + 1)
go func() { defer wg.Done(); errc <- d.processContent() }()
errc := make(chan error, len(fetchers))
wg.Add(len(fetchers))
for _, fn := range fetchers {
fn := fn
go func() { defer wg.Done(); errc <- fn() }()
}
// Wait for the first error, then terminate the others.
var err error
for i := 0; i < len(fetchers)+1; i++ {
if i == len(fetchers) {
for i := 0; i < len(fetchers); i++ {
if i == len(fetchers)-1 {
// Close the queue when all fetchers have exited.
// This will cause the block processor to end when
// it has processed the queue.
@ -552,7 +554,6 @@ func (d *Downloader) fetchHeight(p *peer) (*types.Header, error) {
return nil, errTimeout
case <-d.bodyCh:
case <-d.stateCh:
case <-d.receiptCh:
// Out of bounds delivery, ignore
}
@ -649,7 +650,6 @@ func (d *Downloader) findAncestor(p *peer, height uint64) (uint64, error) {
return 0, errTimeout
case <-d.bodyCh:
case <-d.stateCh:
case <-d.receiptCh:
// Out of bounds delivery, ignore
}
@ -714,7 +714,6 @@ func (d *Downloader) findAncestor(p *peer, height uint64) (uint64, error) {
return 0, errTimeout
case <-d.bodyCh:
case <-d.stateCh:
case <-d.receiptCh:
// Out of bounds delivery, ignore
}
@ -827,7 +826,7 @@ func (d *Downloader) fetchHeaders(p *peer, from uint64) error {
d.dropPeer(p.id)
// Finish the sync gracefully instead of dumping the gathered data though
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh, d.stateWakeCh} {
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
select {
case ch <- false:
case <-d.cancelCh:
@ -927,68 +926,6 @@ func (d *Downloader) fetchReceipts(from uint64) error {
return err
}
// fetchNodeData iteratively downloads the scheduled state trie nodes, taking any
// available peers, reserving a chunk of nodes for each, waiting for delivery and
// also periodically checking for timeouts.
func (d *Downloader) fetchNodeData() error {
log.Debug("Downloading node state data")
var (
deliver = func(packet dataPack) (int, error) {
start := time.Now()
return d.queue.DeliverNodeData(packet.PeerId(), packet.(*statePack).states, func(delivered int, progressed bool, err error) {
// If the peer returned old-requested data, forgive
if err == trie.ErrNotRequested {
log.Debug("Forgiving reply to stale state request", "peer", packet.PeerId())
return
}
if err != nil {
// If the node data processing failed, the root hash is very wrong, abort
log.Error("State processing failed", "peer", packet.PeerId(), "err", err)
d.Cancel()
return
}
// Processing succeeded, notify state fetcher of continuation
pending := d.queue.PendingNodeData()
if pending > 0 {
select {
case d.stateWakeCh <- true:
default:
}
}
d.syncStatsLock.Lock()
d.syncStatsStateDone += uint64(delivered)
syncStatsStateDone := d.syncStatsStateDone // Thread safe copy for the log below
d.syncStatsLock.Unlock()
// If real database progress was made, reset any fast-sync pivot failure
if progressed && atomic.LoadUint32(&d.fsPivotFails) > 1 {
log.Debug("Fast-sync progressed, resetting fail counter", "previous", atomic.LoadUint32(&d.fsPivotFails))
atomic.StoreUint32(&d.fsPivotFails, 1) // Don't ever reset to 0, as that will unlock the pivot block
}
// Log a message to the user and return
if delivered > 0 {
log.Info("Imported new state entries", "count", delivered, "elapsed", common.PrettyDuration(time.Since(start)), "processed", syncStatsStateDone, "pending", pending)
}
})
}
expire = func() map[string]int { return d.queue.ExpireNodeData(d.requestTTL()) }
throttle = func() bool { return false }
reserve = func(p *peer, count int) (*fetchRequest, bool, error) {
return d.queue.ReserveNodeData(p, count), false, nil
}
fetch = func(p *peer, req *fetchRequest) error { return p.FetchNodeData(req) }
capacity = func(p *peer) int { return p.NodeDataCapacity(d.requestRTT()) }
setIdle = func(p *peer, accepted int) { p.SetNodeDataIdle(accepted) }
)
err := d.fetchParts(errCancelStateFetch, d.stateCh, deliver, d.stateWakeCh, expire,
d.queue.PendingNodeData, d.queue.InFlightNodeData, throttle, reserve, nil, fetch,
d.queue.CancelNodeData, capacity, d.peers.NodeDataIdlePeers, setIdle, "states")
log.Debug("Node state data download terminated", "err", err)
return err
}
// fetchParts iteratively downloads scheduled block parts, taking any available
// peers, reserving a chunk of fetch requests for each, waiting for delivery and
// also periodically checking for timeouts.
@ -1229,7 +1166,7 @@ func (d *Downloader) processHeaders(origin uint64, td *big.Int) error {
// Terminate header processing if we synced up
if len(headers) == 0 {
// Notify everyone that headers are fully processed
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh, d.stateWakeCh} {
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
select {
case ch <- false:
case <-d.cancelCh:
@ -1341,7 +1278,7 @@ func (d *Downloader) processHeaders(origin uint64, td *big.Int) error {
origin += uint64(limit)
}
// Signal the content downloaders of the availablility of new tasks
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh, d.stateWakeCh} {
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
select {
case ch <- true:
default:
@ -1353,12 +1290,28 @@ 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() error {
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()
for {
results := d.queue.WaitResults()
if len(results) == 0 {
return nil // queue empty
if stateSync != nil {
return stateSync.Cancel()
}
return nil
}
if d.chainInsertHook != nil {
d.chainInsertHook(results)
@ -1376,6 +1329,11 @@ func (d *Downloader) processContent() error {
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)
@ -1402,8 +1360,9 @@ func (d *Downloader) processContent() error {
case len(receipts) > 0:
index, err = d.insertReceipts(blocks, receipts)
if err == nil && blocks[len(blocks)-1].NumberU64() == pivot {
log.Debug("Committing block as new head", "number", blocks[len(blocks)-1].Number(), "hash", blocks[len(blocks)-1].Hash())
index, err = len(blocks)-1, d.commitHeadBlock(blocks[len(blocks)-1].Hash())
index = len(blocks) - 1
stateSync.Cancel()
err = d.commitPivotBlock(blocks[len(blocks)-1])
}
default:
index, err = d.insertBlocks(blocks)
@ -1418,6 +1377,16 @@ func (d *Downloader) processContent() error {
}
}
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 {
return err
}
log.Debug("Committing block as new head", "number", block.Number(), "hash", block.Hash())
return d.commitHeadBlock(block.Hash())
}
// DeliverHeaders injects a new batch of block headers received from a remote
// node into the download schedule.
func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) (err error) {

25
eth/downloader/peer.go
View file

@ -30,6 +30,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
)
@ -195,7 +196,7 @@ func (p *peer) FetchReceipts(request *fetchRequest) error {
}
// FetchNodeData sends a node state data retrieval request to the remote peer.
func (p *peer) FetchNodeData(request *fetchRequest) error {
func (p *peer) FetchNodeData(hashes []common.Hash) error {
// Sanity check the protocol version
if p.version < 63 {
panic(fmt.Sprintf("node data fetch [eth/63+] requested on eth/%d", p.version))
@ -205,14 +206,7 @@ func (p *peer) FetchNodeData(request *fetchRequest) error {
return errAlreadyFetching
}
p.stateStarted = time.Now()
// Convert the hash set to a retrievable slice
hashes := make([]common.Hash, 0, len(request.Hashes))
for hash := range request.Hashes {
hashes = append(hashes, hash)
}
go p.getNodeData(hashes)
return nil
}
@ -343,8 +337,9 @@ func (p *peer) Lacks(hash common.Hash) bool {
// peerSet represents the collection of active peer participating in the chain
// download procedure.
type peerSet struct {
peers map[string]*peer
lock sync.RWMutex
peers map[string]*peer
newPeerFeed event.Feed
lock sync.RWMutex
}
// newPeerSet creates a new peer set top track the active download sources.
@ -354,6 +349,10 @@ func newPeerSet() *peerSet {
}
}
func (ps *peerSet) SubscribeNewPeers(ch chan<- *peer) event.Subscription {
return ps.newPeerFeed.Subscribe(ch)
}
// Reset iterates over the current peer set, and resets each of the known peers
// to prepare for a next batch of block retrieval.
func (ps *peerSet) Reset() {
@ -377,9 +376,8 @@ func (ps *peerSet) Register(p *peer) error {
// Register the new peer with some meaningful defaults
ps.lock.Lock()
defer ps.lock.Unlock()
if _, ok := ps.peers[p.id]; ok {
ps.lock.Unlock()
return errAlreadyRegistered
}
if len(ps.peers) > 0 {
@ -399,6 +397,9 @@ func (ps *peerSet) Register(p *peer) error {
p.stateThroughput /= float64(len(ps.peers))
}
ps.peers[p.id] = p
ps.lock.Unlock()
ps.newPeerFeed.Send(p)
return nil
}

212
eth/downloader/queue.go
View file

@ -26,12 +26,8 @@ import (
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/trie"
"github.com/rcrowley/go-metrics"
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
)
@ -94,15 +90,6 @@ type queue struct {
receiptPendPool map[string]*fetchRequest // [eth/63] Currently pending receipt retrieval operations
receiptDonePool map[common.Hash]struct{} // [eth/63] Set of the completed receipt fetches
stateTaskIndex int // [eth/63] Counter indexing the added hashes to ensure prioritised retrieval order
stateTaskPool map[common.Hash]int // [eth/63] Pending node data retrieval tasks, mapping to their priority
stateTaskQueue *prque.Prque // [eth/63] Priority queue of the hashes to fetch the node data for
statePendPool map[string]*fetchRequest // [eth/63] Currently pending node data retrieval operations
stateDatabase ethdb.Database // [eth/63] Trie database to populate during state reassembly
stateScheduler *state.StateSync // [eth/63] State trie synchronisation scheduler and integrator
stateWriters int // [eth/63] Number of running state DB writer goroutines
resultCache []*fetchResult // Downloaded but not yet delivered fetch results
resultOffset uint64 // Offset of the first cached fetch result in the block chain
@ -112,7 +99,7 @@ type queue struct {
}
// newQueue creates a new download queue for scheduling block retrieval.
func newQueue(stateDb ethdb.Database) *queue {
func newQueue() *queue {
lock := new(sync.Mutex)
return &queue{
headerPendPool: make(map[string]*fetchRequest),
@ -125,10 +112,6 @@ func newQueue(stateDb ethdb.Database) *queue {
receiptTaskQueue: prque.New(),
receiptPendPool: make(map[string]*fetchRequest),
receiptDonePool: make(map[common.Hash]struct{}),
stateTaskPool: make(map[common.Hash]int),
stateTaskQueue: prque.New(),
statePendPool: make(map[string]*fetchRequest),
stateDatabase: stateDb,
resultCache: make([]*fetchResult, blockCacheLimit),
active: sync.NewCond(lock),
lock: lock,
@ -158,12 +141,6 @@ func (q *queue) Reset() {
q.receiptPendPool = make(map[string]*fetchRequest)
q.receiptDonePool = make(map[common.Hash]struct{})
q.stateTaskIndex = 0
q.stateTaskPool = make(map[common.Hash]int)
q.stateTaskQueue.Reset()
q.statePendPool = make(map[string]*fetchRequest)
q.stateScheduler = nil
q.resultCache = make([]*fetchResult, blockCacheLimit)
q.resultOffset = 0
}
@ -201,28 +178,6 @@ func (q *queue) PendingReceipts() int {
return q.receiptTaskQueue.Size()
}
// PendingNodeData retrieves the number of node data entries pending for retrieval.
func (q *queue) PendingNodeData() int {
q.lock.Lock()
defer q.lock.Unlock()
return q.pendingNodeDataLocked()
}
// pendingNodeDataLocked retrieves the number of node data entries pending for retrieval.
// The caller must hold q.lock.
func (q *queue) pendingNodeDataLocked() int {
var n int
if q.stateScheduler != nil {
n = q.stateScheduler.Pending()
}
// Ensure that PendingNodeData doesn't return 0 until all state is written.
if q.stateWriters > 0 {
n++
}
return n
}
// InFlightHeaders retrieves whether there are header fetch requests currently
// in flight.
func (q *queue) InFlightHeaders() bool {
@ -250,28 +205,15 @@ func (q *queue) InFlightReceipts() bool {
return len(q.receiptPendPool) > 0
}
// InFlightNodeData retrieves whether there are node data entry fetch requests
// currently in flight.
func (q *queue) InFlightNodeData() bool {
q.lock.Lock()
defer q.lock.Unlock()
return len(q.statePendPool)+q.stateWriters > 0
}
// Idle returns if the queue is fully idle or has some data still inside. This
// method is used by the tester to detect termination events.
// Idle returns if the queue is fully idle or has some data still inside.
func (q *queue) Idle() bool {
q.lock.Lock()
defer q.lock.Unlock()
queued := q.blockTaskQueue.Size() + q.receiptTaskQueue.Size() + q.stateTaskQueue.Size()
pending := len(q.blockPendPool) + len(q.receiptPendPool) + len(q.statePendPool)
queued := q.blockTaskQueue.Size() + q.receiptTaskQueue.Size()
pending := len(q.blockPendPool) + len(q.receiptPendPool)
cached := len(q.blockDonePool) + len(q.receiptDonePool)
if q.stateScheduler != nil {
queued += q.stateScheduler.Pending()
}
return (queued + pending + cached) == 0
}
@ -389,19 +331,6 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
q.receiptTaskPool[hash] = header
q.receiptTaskQueue.Push(header, -float32(header.Number.Uint64()))
}
if q.mode == FastSync && header.Number.Uint64() == q.fastSyncPivot {
// Pivoting point of the fast sync, switch the state retrieval to this
log.Debug("Switching state downloads to new block", "number", header.Number, "hash", hash)
q.stateTaskIndex = 0
q.stateTaskPool = make(map[common.Hash]int)
q.stateTaskQueue.Reset()
for _, req := range q.statePendPool {
req.Hashes = make(map[common.Hash]int) // Make sure executing requests fail, but don't disappear
}
q.stateScheduler = state.NewStateSync(header.Root, q.stateDatabase)
}
inserts = append(inserts, header)
q.headerHead = hash
from++
@ -452,25 +381,19 @@ func (q *queue) countProcessableItems() int {
if q.mode == FastSync {
bnum := result.Header.Number.Uint64()
if bnum == q.fastSyncPivot {
// If the state of the pivot block is not
// available yet, we cannot proceed and return 0.
//
// 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 i > 0 || len(q.stateTaskPool) > 0 || q.pendingNodeDataLocked() > 0 {
return i
}
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 fast sync pivot, stop as well
// 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.process.
// This simplifies handling the switchover in d.processContent.
if bnum == q.fastSyncPivot+1 && i > 0 {
return i
}
@ -519,25 +442,6 @@ func (q *queue) ReserveHeaders(p *peer, count int) *fetchRequest {
return request
}
// ReserveNodeData reserves a set of node data hashes for the given peer, skipping
// any previously failed download.
func (q *queue) ReserveNodeData(p *peer, count int) *fetchRequest {
// Create a task generator to fetch status-fetch tasks if all schedules ones are done
generator := func(max int) {
if q.stateScheduler != nil {
for _, hash := range q.stateScheduler.Missing(max) {
q.stateTaskPool[hash] = q.stateTaskIndex
q.stateTaskQueue.Push(hash, -float32(q.stateTaskIndex))
q.stateTaskIndex++
}
}
}
q.lock.Lock()
defer q.lock.Unlock()
return q.reserveHashes(p, count, q.stateTaskQueue, generator, q.statePendPool, maxInFlightStates)
}
// reserveHashes reserves a set of hashes for the given peer, skipping previously
// failed ones.
//
@ -722,12 +626,6 @@ func (q *queue) CancelReceipts(request *fetchRequest) {
q.cancel(request, q.receiptTaskQueue, q.receiptPendPool)
}
// CancelNodeData aborts a node state data fetch request, returning all pending
// hashes to the task queue.
func (q *queue) CancelNodeData(request *fetchRequest) {
q.cancel(request, q.stateTaskQueue, q.statePendPool)
}
// Cancel aborts a fetch request, returning all pending hashes to the task queue.
func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool map[string]*fetchRequest) {
q.lock.Lock()
@ -764,12 +662,6 @@ func (q *queue) Revoke(peerId string) {
}
delete(q.receiptPendPool, peerId)
}
if request, ok := q.statePendPool[peerId]; ok {
for hash, index := range request.Hashes {
q.stateTaskQueue.Push(hash, float32(index))
}
delete(q.statePendPool, peerId)
}
}
// ExpireHeaders checks for in flight requests that exceeded a timeout allowance,
@ -799,15 +691,6 @@ func (q *queue) ExpireReceipts(timeout time.Duration) map[string]int {
return q.expire(timeout, q.receiptPendPool, q.receiptTaskQueue, receiptTimeoutMeter)
}
// ExpireNodeData checks for in flight node data requests that exceeded a timeout
// allowance, canceling them and returning the responsible peers for penalisation.
func (q *queue) ExpireNodeData(timeout time.Duration) map[string]int {
q.lock.Lock()
defer q.lock.Unlock()
return q.expire(timeout, q.statePendPool, q.stateTaskQueue, stateTimeoutMeter)
}
// expire is the generic check that move expired tasks from a pending pool back
// into a task pool, returning all entities caught with expired tasks.
//
@ -1044,84 +927,6 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
}
}
// DeliverNodeData injects a node state data retrieval response into the queue.
// The method returns the number of node state accepted from the delivery.
func (q *queue) DeliverNodeData(id string, data [][]byte, callback func(int, bool, error)) (int, error) {
q.lock.Lock()
defer q.lock.Unlock()
// Short circuit if the data was never requested
request := q.statePendPool[id]
if request == nil {
return 0, errNoFetchesPending
}
stateReqTimer.UpdateSince(request.Time)
delete(q.statePendPool, id)
// If no data was retrieved, mark their hashes as unavailable for the origin peer
if len(data) == 0 {
for hash := range request.Hashes {
request.Peer.MarkLacking(hash)
}
}
// Iterate over the downloaded data and verify each of them
errs := make([]error, 0)
process := []trie.SyncResult{}
for _, blob := range data {
// Skip any state trie entries that were not requested
hash := common.BytesToHash(crypto.Keccak256(blob))
if _, ok := request.Hashes[hash]; !ok {
errs = append(errs, fmt.Errorf("non-requested state data %x", hash))
continue
}
// Inject the next state trie item into the processing queue
process = append(process, trie.SyncResult{Hash: hash, Data: blob})
delete(request.Hashes, hash)
delete(q.stateTaskPool, hash)
}
// Return all failed or missing fetches to the queue
for hash, index := range request.Hashes {
q.stateTaskQueue.Push(hash, float32(index))
}
if q.stateScheduler == nil {
return 0, errNoFetchesPending
}
// Run valid nodes through the trie download scheduler. It writes completed nodes to a
// batch, which is committed asynchronously. This may lead to over-fetches because the
// scheduler treats everything as written after Process has returned, but it's
// unlikely to be an issue in practice.
batch := q.stateDatabase.NewBatch()
progressed, nproc, procerr := q.stateScheduler.Process(process, batch)
q.stateWriters += 1
go func() {
if procerr == nil {
nproc = len(process)
procerr = batch.Write()
}
// Return processing errors through the callback so the sync gets canceled. The
// number of writers is decremented prior to the call so PendingNodeData will
// return zero when the callback runs.
q.lock.Lock()
q.stateWriters -= 1
q.lock.Unlock()
callback(nproc, progressed, procerr)
// Wake up WaitResults after the state has been written because it might be
// waiting for completion of the pivot block's state download.
q.active.Signal()
}()
// If none of the data items were good, it's a stale delivery
switch {
case len(errs) == 0:
return len(process), nil
case len(errs) == len(request.Hashes):
return len(process), errStaleDelivery
default:
return len(process), fmt.Errorf("multiple failures: %v", errs)
}
}
// Prepare configures the result cache to allow accepting and caching inbound
// fetch results.
func (q *queue) Prepare(offset uint64, mode SyncMode, pivot uint64, head *types.Header) {
@ -1134,9 +939,4 @@ func (q *queue) Prepare(offset uint64, mode SyncMode, pivot uint64, head *types.
}
q.fastSyncPivot = pivot
q.mode = mode
// If long running fast sync, also start up a head stateretrieval immediately
if mode == FastSync && pivot > 0 {
q.stateScheduler = state.NewStateSync(head.Root, q.stateDatabase)
}
}

331
eth/downloader/statesync.go Normal file
View file

@ -0,0 +1,331 @@
// Copyright 2017 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package downloader
import (
"fmt"
"hash"
"sync"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/crypto/sha3"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/trie"
)
type stateReq struct {
items []common.Hash
tasks map[common.Hash]*stateTask
timeout time.Duration
timer *time.Timer // fires when timeout has elapsed
peer *peer // the peer that we're requesting from
response [][]byte // the response. this is nil for timed-out requests.
}
type stateSyncStats struct {
done uint64 // number of entries pulled
pending uint64 // number of pending entries
}
// syncPivotState starts downloading state with the given root hash.
func (d *Downloader) syncState(root common.Hash) *stateSync {
s := newStateSync(d, root)
select {
case d.stateSyncStart <- s:
case <-d.cancelCh:
s.err = errCancelStateFetch
close(s.done)
}
return s
}
// stateFetcher manages the active state sync and accepts requests
// on its behalf.
func (d *Downloader) stateFetcher() {
for {
select {
case s := <-d.stateSyncStart:
for next := s; next != nil; {
next = d.runStateSync(next)
}
case <-d.stateCh:
// Ignore state responses while no sync is running.
case <-d.quitCh:
return
}
}
}
// runStateSync runs s until it completes or another stateSync is requested.
func (d *Downloader) runStateSync(s *stateSync) *stateSync {
var (
activeReqs = make(map[string]*stateReq)
finishedReqs []*stateReq
timeout = make(chan *stateReq)
cancelTimeout = make(chan struct{})
)
// Cancel active request timers on exit.
// The cancelTimeout channel prevents leaking of timer goroutines
// in the unlikely case where a timer is fired just before canceling it.
defer func() {
for _, req := range activeReqs {
req.timer.Stop()
}
close(cancelTimeout)
}()
// Run the state sync.
go s.run()
defer s.Cancel()
for {
// Enable sending of the first buffered element if there is one.
var deliverReq *stateReq
var deliverReqCh chan *stateReq
if len(finishedReqs) > 0 {
deliverReq = finishedReqs[0]
deliverReqCh = s.deliver
}
select {
// The stateSync lifecycle:
case next := <-d.stateSyncStart:
return next
case <-s.done:
return nil
// Send the next finished request to the current sync:
case deliverReqCh <- deliverReq:
finishedReqs = finishedReqs[1:]
// Handle incoming state packs:
case pack := <-d.stateCh:
req := activeReqs[pack.PeerId()]
if req == nil {
log.Debug("Unrequested node data", "peer", pack.PeerId(), "len", pack.Items())
continue
}
delete(activeReqs, pack.PeerId())
req.timer.Stop()
req.response = pack.(*statePack).states
finishedReqs = append(finishedReqs, req)
// Handle timed-out requests:
case req := <-timeout:
if activeReqs[req.peer.id] != req {
continue // ignore old timeout
}
finishedReqs = append(finishedReqs, req)
delete(activeReqs, req.peer.id)
// Track outgoing state requests:
case req := <-d.trackStateReq:
activeReqs[req.peer.id] = req
req.timer = time.AfterFunc(req.timeout, func() {
select {
case timeout <- req:
case <-cancelTimeout:
}
})
}
}
}
// stateSync schedules requests for downloading a particular state root.
type stateSync struct {
d *Downloader
sched *state.StateSync
keccak hash.Hash
tasksAvailable map[common.Hash]*stateTask
deliver chan *stateReq
cancel chan struct{}
cancelOnce sync.Once
done chan struct{}
err error // set after done is closed
}
type stateTask struct {
triedPeers map[string]struct{}
}
func newStateSync(d *Downloader, root common.Hash) *stateSync {
return &stateSync{
d: d,
sched: state.NewStateSync(root, d.stateDB),
keccak: sha3.NewKeccak256(),
tasksAvailable: make(map[common.Hash]*stateTask),
deliver: make(chan *stateReq),
cancel: make(chan struct{}),
done: make(chan struct{}, 1),
}
}
// 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
}
}
func (s *stateSync) Cancel() error {
s.cancelOnce.Do(func() { close(s.cancel) })
return s.wait()
}
func (s *stateSync) run() {
s.err = s.loop()
close(s.done)
}
func (s *stateSync) loop() error {
newPeer := make(chan *peer, 200)
peerSub := s.d.peers.SubscribeNewPeers(newPeer)
defer peerSub.Unsubscribe()
for s.sched.Pending() > 0 {
if err := s.assignTasks(); err != nil {
return err
}
select {
case <-newPeer:
// assign new tasks
case <-s.cancel:
return errCancelStateFetch
case req := <-s.deliver:
// response or timeout
req.peer.SetNodeDataIdle(len(req.response))
procStart := time.Now()
n, err := s.process(req)
if err != nil {
log.Warn("Node data write error", "err", err)
return err
}
if len(req.items) == 1 && req.response == nil {
log.Warn("Node data timeout, dropping peer", "peer", req.peer.id)
s.d.dropPeer(req.peer.id)
}
s.updateStats(n, time.Since(procStart))
}
}
return nil
}
func (s *stateSync) sendReq(req *stateReq) {
req.peer.log.Trace("Requesting new batch of data", "type", "state", "count", len(req.items))
select {
case s.d.trackStateReq <- req:
req.peer.FetchNodeData(req.items)
case <-s.cancel:
}
}
func (s *stateSync) assignTasks() error {
peers, _ := s.d.peers.NodeDataIdlePeers()
for _, p := range peers {
cap := p.NodeDataCapacity(s.d.requestRTT())
req := &stateReq{peer: p, timeout: s.d.requestTTL()}
if err := s.popTasks(cap, req); err != nil {
return err
}
if len(req.items) > 0 {
s.sendReq(req)
}
}
return nil
}
func (s *stateSync) popTasks(n int, req *stateReq) error {
// Refill available tasks from the scheduler.
if len(s.tasksAvailable) < n {
new := s.sched.Missing(n - len(s.tasksAvailable))
for _, hash := range new {
s.tasksAvailable[hash] = &stateTask{make(map[string]struct{})}
}
}
// Find tasks that haven't been tried with the request's peer.
req.items = make([]common.Hash, 0, n)
req.tasks = make(map[common.Hash]*stateTask, n)
for hash, t := range s.tasksAvailable {
if len(req.items) == n {
break
}
if _, ok := t.triedPeers[req.peer.id]; ok {
continue
}
t.triedPeers[req.peer.id] = struct{}{}
req.items = append(req.items, hash)
req.tasks[hash] = t
delete(s.tasksAvailable, hash)
}
return nil
}
func (s *stateSync) process(req *stateReq) (nproc int, err error) {
batch := s.d.stateDB.NewBatch()
for _, blob := range req.response {
if hash, ok := s.processNodeData(blob, batch); ok {
nproc++
delete(req.tasks, hash)
}
}
if err := batch.Write(); err != nil {
return 0, err
}
if nproc > 0 && atomic.LoadUint32(&s.d.fsPivotFails) > 1 {
log.Trace("Fast-sync progressed, resetting fail counter", "previous", atomic.LoadUint32(&s.d.fsPivotFails))
atomic.StoreUint32(&s.d.fsPivotFails, 1) // Don't ever reset to 0, as that will unlock the pivot block
}
// Put unfulfilled tasks back.
for hash, task := range req.tasks {
if npeers := s.d.peers.Len(); len(task.triedPeers) >= npeers {
return nproc, fmt.Errorf("state node %s failed with all peers (%d tries, %d peers)", hash.TerminalString(), len(task.triedPeers), npeers)
}
s.tasksAvailable[hash] = task
}
return nproc, nil
}
func (s *stateSync) processNodeData(blob []byte, batch ethdb.Batch) (common.Hash, bool) {
res := trie.SyncResult{Data: blob}
s.keccak.Reset()
s.keccak.Write(blob)
s.keccak.Sum(res.Hash[:0])
_, _, err := s.sched.Process([]trie.SyncResult{res}, batch)
return res.Hash, err == nil
}
func (s *stateSync) updateStats(processed int, duration time.Duration) {
s.d.syncStatsLock.Lock()
defer s.d.syncStatsLock.Unlock()
s.d.syncStatsState.pending = uint64(s.sched.Pending())
s.d.syncStatsState.done += uint64(processed)
log.Info("Imported new state entries", "count", processed, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.done, "pending", s.d.syncStatsState.pending, "retry", len(s.tasksAvailable))
}