diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index b338129e00..1964882ddd 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -109,9 +109,8 @@ 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 - syncStatsState stateSyncStats + syncStatsChainOrigin uint64 // Origin block number where syncing started at + syncStatsChainHeight uint64 // Highest block number known when syncing started syncStatsLock sync.RWMutex // Lock protecting the sync stats fields lightchain LightChain @@ -133,11 +132,6 @@ type Downloader struct { receiptWakeCh chan bool // [eth/63] Channel to signal the receipt 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 @@ -207,29 +201,25 @@ func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, chain BlockC } dl := &Downloader{ - mode: mode, - stateDB: stateDb, - mux: mux, - queue: newQueue(), - peers: newPeerSet(), - rttEstimate: uint64(rttMaxEstimate), - rttConfidence: uint64(1000000), - blockchain: chain, - lightchain: lightchain, - dropPeer: dropPeer, - headerCh: make(chan dataPack, 1), - bodyCh: make(chan dataPack, 1), - receiptCh: make(chan dataPack, 1), - bodyWakeCh: make(chan bool, 1), - receiptWakeCh: make(chan bool, 1), - headerProcCh: make(chan []*types.Header, 1), - quitCh: make(chan struct{}), - stateCh: make(chan dataPack), - stateSyncStart: make(chan *stateSync), - trackStateReq: make(chan *stateReq), + mode: mode, + stateDB: stateDb, + mux: mux, + queue: newQueue(), + peers: newPeerSet(), + rttEstimate: uint64(rttMaxEstimate), + rttConfidence: uint64(1000000), + blockchain: chain, + lightchain: lightchain, + dropPeer: dropPeer, + headerCh: make(chan dataPack, 1), + bodyCh: make(chan dataPack, 1), + receiptCh: make(chan dataPack, 1), + bodyWakeCh: make(chan bool, 1), + receiptWakeCh: make(chan bool, 1), + headerProcCh: make(chan []*types.Header, 1), + quitCh: make(chan struct{}), } go dl.qosTuner() - go dl.stateFetcher() return dl } @@ -258,8 +248,6 @@ func (d *Downloader) Progress() ethereum.SyncProgress { StartingBlock: d.syncStatsChainOrigin, CurrentBlock: current, HighestBlock: d.syncStatsChainHeight, - PulledStates: d.syncStatsState.processed, - KnownStates: d.syncStatsState.processed + d.syncStatsState.pending, } } @@ -1369,39 +1357,7 @@ func (d *Downloader) importBlockResults(results []*fetchResult) error { // processFastSyncContent takes fetch results from the queue and writes them to the // database. It also controls the synchronisation of state nodes of the pivot block. 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 - } - } + return nil } func splitAroundPivot(pivot uint64, results []*fetchResult) (p *fetchResult, before, after []*fetchResult) { @@ -1419,48 +1375,10 @@ func splitAroundPivot(pivot uint64, results []*fetchResult) (p *fetchResult, bef 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.blockchain.InsertReceiptChain(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.blockchain.InsertReceiptChain([]*types.Block{b}, []types.Receipts{result.Receipts}); err != nil { return err @@ -1486,7 +1404,7 @@ func (d *Downloader) DeliverReceipts(id string, receipts [][]*types.Receipt) (er // DeliverNodeData injects a new batch of node state data received from a remote node. func (d *Downloader) DeliverNodeData(id string, data [][]byte) (err error) { - return d.deliver(id, d.stateCh, &statePack{id, data}, stateInMeter, stateDropMeter) + return nil } // deliver injects a new batch of data received from a remote node. diff --git a/eth/downloader/statesync.go b/eth/downloader/statesync.go deleted file mode 100644 index 937828b949..0000000000 --- a/eth/downloader/statesync.go +++ /dev/null @@ -1,478 +0,0 @@ -// 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 . - -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" -) - -// stateReq represents a batch of state fetch requests groupped together into -// a single data retrieval network packet. -type stateReq struct { - items []common.Hash // Hashes of the state items to download - tasks map[common.Hash]*stateTask // Download tasks to track previous attempts - timeout time.Duration // Maximum round trip time for this to complete - timer *time.Timer // Timer to fire when the RTT timeout expires - peer *peerConnection // Peer that we're requesting from - response [][]byte // Response data of the peer (nil for timeouts) - dropped bool // Flag whether the peer dropped off early -} - -// timedOut returns if this request timed out. -func (req *stateReq) timedOut() bool { - return req.response == nil -} - -// stateSyncStats is a collection of progress stats to report during a state trie -// sync to RPC requests as well as to display in user logs. -type stateSyncStats struct { - processed uint64 // Number of state entries processed - duplicate uint64 // Number of state entries downloaded twice - unexpected uint64 // Number of non-requested state entries received - pending uint64 // Number of still pending state entries -} - -// syncState 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.quitCh: - 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 a state synchronisation until it completes or another root -// hash is requested to be switched over to. -func (d *Downloader) runStateSync(s *stateSync) *stateSync { - var ( - active = make(map[string]*stateReq) // Currently in-flight requests - finished []*stateReq // Completed or failed requests - timeout = make(chan *stateReq) // Timed out active requests - ) - defer func() { - // Cancel active request timers on exit. Also set peers to idle so they're - // available for the next sync. - for _, req := range active { - req.timer.Stop() - req.peer.SetNodeDataIdle(len(req.items)) - } - }() - // Run the state sync. - go s.run() - defer s.Cancel() - - // Listen for peer departure events to cancel assigned tasks - peerDrop := make(chan *peerConnection, 1024) - peerSub := s.d.peers.SubscribePeerDrops(peerDrop) - defer peerSub.Unsubscribe() - - for { - // Enable sending of the first buffered element if there is one. - var ( - deliverReq *stateReq - deliverReqCh chan *stateReq - ) - if len(finished) > 0 { - deliverReq = finished[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: - // Shift out the first request, but also set the emptied slot to nil for GC - copy(finished, finished[1:]) - finished[len(finished)-1] = nil - finished = finished[:len(finished)-1] - - // Handle incoming state packs: - case pack := <-d.stateCh: - // Discard any data not requested (or previsouly timed out) - req := active[pack.PeerId()] - if req == nil { - log.Debug("Unrequested node data", "peer", pack.PeerId(), "len", pack.Items()) - continue - } - // Finalize the request and queue up for processing - req.timer.Stop() - req.response = pack.(*statePack).states - - finished = append(finished, req) - delete(active, pack.PeerId()) - - // Handle dropped peer connections: - case p := <-peerDrop: - // Skip if no request is currently pending - req := active[p.id] - if req == nil { - continue - } - // Finalize the request and queue up for processing - req.timer.Stop() - req.dropped = true - - finished = append(finished, req) - delete(active, p.id) - - // Handle timed-out requests: - case req := <-timeout: - // If the peer is already requesting something else, ignore the stale timeout. - // This can happen when the timeout and the delivery happens simultaneously, - // causing both pathways to trigger. - if active[req.peer.id] != req { - continue - } - // Move the timed out data back into the download queue - finished = append(finished, req) - delete(active, req.peer.id) - - // Track outgoing state requests: - case req := <-d.trackStateReq: - // If an active request already exists for this peer, we have a problem. In - // theory the trie node schedule must never assign two requests to the same - // peer. In practive however, a peer might receive a request, disconnect and - // immediately reconnect before the previous times out. In this case the first - // request is never honored, alas we must not silently overwrite it, as that - // causes valid requests to go missing and sync to get stuck. - if old := active[req.peer.id]; old != nil { - log.Warn("Busy peer assigned new state fetch", "peer", old.peer.id) - - // Make sure the previous one doesn't get siletly lost - old.timer.Stop() - old.dropped = true - - finished = append(finished, old) - } - // Start a timer to notify the sync loop if the peer stalled. - req.timer = time.AfterFunc(req.timeout, func() { - select { - case timeout <- req: - case <-s.done: - // Prevent leaking of timer goroutines in the unlikely case where a - // timer is fired just before exiting runStateSync. - } - }) - active[req.peer.id] = req - } - } -} - -// stateSync schedules requests for downloading a particular state trie defined -// by a given state root. -type stateSync struct { - d *Downloader // Downloader instance to access and manage current peerset - - sched *trie.TrieSync // State trie sync scheduler defining the tasks - keccak hash.Hash // Keccak256 hasher to verify deliveries with - tasks map[common.Hash]*stateTask // Set of tasks currently queued for retrieval - - numUncommitted int - bytesUncommitted int - - deliver chan *stateReq // Delivery channel multiplexing peer responses - cancel chan struct{} // Channel to signal a termination request - cancelOnce sync.Once // Ensures cancel only ever gets called once - done chan struct{} // Channel to signal termination completion - err error // Any error hit during sync (set before completion) -} - -// stateTask represents a single trie node download taks, containing a set of -// peers already attempted retrieval from to detect stalled syncs and abort. -type stateTask struct { - attempts map[string]struct{} -} - -// newStateSync creates a new state trie download scheduler. This method does not -// yet start the sync. The user needs to call run to initiate. -func newStateSync(d *Downloader, root common.Hash) *stateSync { - return &stateSync{ - d: d, - sched: state.NewStateSync(root, d.stateDB), - keccak: sha3.NewKeccak256(), - tasks: make(map[common.Hash]*stateTask), - deliver: make(chan *stateReq), - cancel: make(chan struct{}), - done: make(chan struct{}), - } -} - -// run starts the task assignment and response processing loop, blocking until -// it finishes, and finally notifying any goroutines waiting for the loop to -// finish. -func (s *stateSync) run() { - s.err = s.loop() - close(s.done) -} - -// Wait blocks until the sync is done or canceled. -func (s *stateSync) Wait() error { - <-s.done - return s.err -} - -// 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() -} - -// loop is the main event loop of a state trie sync. It it responsible for the -// assignment of new tasks to peers (including sending it to them) as well as -// for the processing of inbound data. Note, that the loop does not directly -// receive data from peers, rather those are buffered up in the downloader and -// pushed here async. The reason is to decouple processing from data receipt -// and timeouts. -func (s *stateSync) loop() error { - // Listen for new peer events to assign tasks to them - newPeer := make(chan *peerConnection, 1024) - peerSub := s.d.peers.SubscribeNewPeers(newPeer) - defer peerSub.Unsubscribe() - - // Keep assigning new tasks until the sync completes or aborts - for s.sched.Pending() > 0 { - if err := s.commit(false); err != nil { - return err - } - s.assignTasks() - // Tasks assigned, wait for something to happen - select { - case <-newPeer: - // New peer arrived, try to assign it download tasks - - case <-s.cancel: - return errCancelStateFetch - - case req := <-s.deliver: - // Response, disconnect or timeout triggered, drop the peer if stalling - log.Trace("Received node data response", "peer", req.peer.id, "count", len(req.response), "dropped", req.dropped, "timeout", !req.dropped && req.timedOut()) - if len(req.items) <= 2 && !req.dropped && req.timedOut() { - // 2 items are the minimum requested, if even that times out, we've no use of - // this peer at the moment. - log.Warn("Stalling state sync, dropping peer", "peer", req.peer.id) - s.d.dropPeer(req.peer.id) - } - // Process all the received blobs and check for stale delivery - stale, err := s.process(req) - if err != nil { - log.Warn("Node data write error", "err", err) - return err - } - // The the delivery contains requested data, mark the node idle (otherwise it's a timed out delivery) - if !stale { - req.peer.SetNodeDataIdle(len(req.response)) - } - } - } - return s.commit(true) -} - -func (s *stateSync) commit(force bool) error { - if !force && s.bytesUncommitted < ethdb.IdealBatchSize { - return nil - } - start := time.Now() - b := s.d.stateDB.NewBatch() - s.sched.Commit(b) - if err := b.Write(); err != nil { - return fmt.Errorf("DB write error: %v", err) - } - s.updateStats(s.numUncommitted, 0, 0, time.Since(start)) - s.numUncommitted = 0 - s.bytesUncommitted = 0 - return nil -} - -// assignTasks attempts to assing new tasks to all idle peers, either from the -// batch currently being retried, or fetching new data from the trie sync itself. -func (s *stateSync) assignTasks() { - // Iterate over all idle peers and try to assign them state fetches - peers, _ := s.d.peers.NodeDataIdlePeers() - for _, p := range peers { - // Assign a batch of fetches proportional to the estimated latency/bandwidth - cap := p.NodeDataCapacity(s.d.requestRTT()) - req := &stateReq{peer: p, timeout: s.d.requestTTL()} - s.fillTasks(cap, req) - - // If the peer was assigned tasks to fetch, send the network request - if len(req.items) > 0 { - 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: - } - } - } -} - -// fillTasks fills the given request object with a maximum of n state download -// tasks to send to the remote peer. -func (s *stateSync) fillTasks(n int, req *stateReq) { - // Refill available tasks from the scheduler. - if len(s.tasks) < n { - new := s.sched.Missing(n - len(s.tasks)) - for _, hash := range new { - s.tasks[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.tasks { - // Stop when we've gathered enough requests - if len(req.items) == n { - break - } - // Skip any requests we've already tried from this peer - if _, ok := t.attempts[req.peer.id]; ok { - continue - } - // Assign the request to this peer - t.attempts[req.peer.id] = struct{}{} - req.items = append(req.items, hash) - req.tasks[hash] = t - delete(s.tasks, hash) - } -} - -// process iterates over a batch of delivered state data, injecting each item -// into a running state sync, re-queuing any items that were requested but not -// delivered. -func (s *stateSync) process(req *stateReq) (bool, error) { - // Collect processing stats and update progress if valid data was received - duplicate, unexpected := 0, 0 - - defer func(start time.Time) { - if duplicate > 0 || unexpected > 0 { - s.updateStats(0, duplicate, unexpected, time.Since(start)) - } - }(time.Now()) - - // Iterate over all the delivered data and inject one-by-one into the trie - progress, stale := false, len(req.response) > 0 - - for _, blob := range req.response { - prog, hash, err := s.processNodeData(blob) - switch err { - case nil: - s.numUncommitted++ - s.bytesUncommitted += len(blob) - progress = progress || prog - case trie.ErrNotRequested: - unexpected++ - case trie.ErrAlreadyProcessed: - duplicate++ - default: - return stale, fmt.Errorf("invalid state node %s: %v", hash.TerminalString(), err) - } - // If the node delivered a requested item, mark the delivery non-stale - if _, ok := req.tasks[hash]; ok { - delete(req.tasks, hash) - stale = false - } - } - // If we're inside the critical section, reset fail counter since we progressed. - if progress && 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 into the retry queue - npeers := s.d.peers.Len() - for hash, task := range req.tasks { - // If the node did deliver something, missing items may be due to a protocol - // limit or a previous timeout + delayed delivery. Both cases should permit - // the node to retry the missing items (to avoid single-peer stalls). - if len(req.response) > 0 || req.timedOut() { - delete(task.attempts, req.peer.id) - } - // If we've requested the node too many times already, it may be a malicious - // sync where nobody has the right data. Abort. - if len(task.attempts) >= npeers { - return stale, fmt.Errorf("state node %s failed with all peers (%d tries, %d peers)", hash.TerminalString(), len(task.attempts), npeers) - } - // Missing item, place into the retry queue. - s.tasks[hash] = task - } - return stale, nil -} - -// processNodeData tries to inject a trie node data blob delivered from a remote -// peer into the state trie, returning whether anything useful was written or any -// error occurred. -func (s *stateSync) processNodeData(blob []byte) (bool, common.Hash, error) { - res := trie.SyncResult{Data: blob} - s.keccak.Reset() - s.keccak.Write(blob) - s.keccak.Sum(res.Hash[:0]) - committed, _, err := s.sched.Process([]trie.SyncResult{res}) - return committed, res.Hash, err -} - -// updateStats bumps the various state sync progress counters and displays a log -// message for the user to see. -func (s *stateSync) updateStats(written, duplicate, unexpected int, duration time.Duration) { - s.d.syncStatsLock.Lock() - defer s.d.syncStatsLock.Unlock() - - s.d.syncStatsState.pending = uint64(s.sched.Pending()) - s.d.syncStatsState.processed += uint64(written) - s.d.syncStatsState.duplicate += uint64(duplicate) - s.d.syncStatsState.unexpected += uint64(unexpected) - - if written > 0 || duplicate > 0 || unexpected > 0 { - log.Info("Imported new state entries", "count", written, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.processed, "pending", s.d.syncStatsState.pending, "retry", len(s.tasks), "duplicate", s.d.syncStatsState.duplicate, "unexpected", s.d.syncStatsState.unexpected) - } -}