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

eth/downloader, trie: various polishes around duplicate items

Browse source 
This commit explicitly tracks duplicate and unexpected state
delieveries done against a trie Sync structure, also adding
there to import info logs.

The commit moves the db batch used to commit trie changes one
level deeper so its flushed after every node insertion. This
is needed to avoid a lot of duplicate retrievals caused by
inconsistencies between Sync internals and database. A better
approach is to track not-yet-written states in trie.Sync and
flush on commit, but I'm focuing on correctness first now.

The commit fixes a regression around pivot block fail count.
The counter previously was reset to 1 if and only if a sync
cycle progressed (inserted at least 1 entry to the database).
The current code reset it already if a node was delivered,
which is not stong enough, because unless it ends up written
to disk, an attacker can just loop and attack ad infinitum.

The commit also fixes a regression around state deliveries
and timeouts. The old downloader tracked if a delivery is
stale (none of the deliveries were requestedt), in which
case it didn't mark the node idle and did not send further
requests, since it signals a past timeout. The current code
did mark it idle even on stale deliveries, which eventually
caused two requests to be in flight at the same time, making
the deliveries always stale and mass duplicating retrievals
between multiple peers.
This commit is contained in:
Péter Szilágyi 2017-06-16 11:33:58 +03:00
parent 829a042087
commit e3717b48c3
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
3 changed files with 99 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
eth/downloader/downloader.go
View file

@ -233,8 +233,8 @@ func (d *Downloader) Progress() ethereum.SyncProgress {
StartingBlock: d.syncStatsChainOrigin,
CurrentBlock: current,
HighestBlock: d.syncStatsChainHeight,
PulledStates: d.syncStatsState.done,
KnownStates: d.syncStatsState.done + d.syncStatsState.pending,
PulledStates: d.syncStatsState.processed,
KnownStates: d.syncStatsState.processed + d.syncStatsState.pending,
}
}

122
eth/downloader/statesync.go
View file

@ -26,7 +26,6 @@ import (
"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"
)
@ -45,8 +44,10 @@ func (req *stateReq) timedOut() bool {
}
type stateSyncStats struct {
done uint64 // number of entries pulled
pending uint64 // number of pending entries
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
}
// syncPivotState starts downloading state with the given root hash.
@ -216,20 +217,22 @@ func (s *stateSync) loop() error {
case <-s.cancel:
return errCancelStateFetch
case req := <-s.deliver:
// response or timeout
log.Trace("Received node data response", "peer", req.peer.id, "count", len(req.items), "timeout", req.timedOut())
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
}
// Response or timeout triggered, drop the peer if stalling
log.Trace("Received node data response", "peer", req.peer.id, "count", len(req.response), "timeout", req.timedOut())
if len(req.items) == 1 && req.timedOut() {
log.Warn("Node data timeout, dropping peer", "peer", req.peer.id)
s.d.dropPeer(req.peer.id)
}
s.updateStats(n, time.Since(procStart))
// 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 nil
@ -237,6 +240,7 @@ func (s *stateSync) loop() error {
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)
@ -286,54 +290,102 @@ func (s *stateSync) popTasks(n int, req *stateReq) error {
return nil
}
func (s *stateSync) process(req *stateReq) (nproc int, err error) {
batch := s.d.stateDB.NewBatch()
// 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
processed, duplicate, unexpected := 0, 0, 0
defer func(start time.Time) {
if processed+duplicate+unexpected > 0 {
s.updateStats(processed, 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 {
hash, err := s.processNodeData(blob, batch)
if err != nil && err != trie.ErrNotRequested {
return 0, fmt.Errorf("invalid state node %s: %v", hash.TerminalString(), err)
} else if err == nil {
nproc++
prog, hash, err := s.processNodeData(blob)
switch err {
case nil:
processed++
case trie.ErrNotRequested:
unexpected++
case trie.ErrAlreadyProcessed:
duplicate++
default:
return stale, fmt.Errorf("invalid state node %s: %v", hash.TerminalString(), err)
}
if prog {
progress = true
}
// 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 err := batch.Write(); err != nil {
return 0, err
}
if nproc > 0 && atomic.LoadUint32(&s.d.fsPivotFails) > 1 {
// If some data managed to hit the database, sync progressed so reset any pivot fail counter
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.
// 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() {
// Ensure that the item will be retried because it may have been excluded due
// to a protocol limit.
delete(task.triedPeers, req.peer.id)
}
// Check retry limit.
// 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.triedPeers) >= npeers {
return nproc, fmt.Errorf("state node %s failed with all peers (%d tries, %d peers)", hash.TerminalString(), len(task.triedPeers), npeers)
return stale, fmt.Errorf("state node %s failed with all peers (%d tries, %d peers)", hash.TerminalString(), len(task.triedPeers), npeers)
}
// Missing item, place into the retry queue.
s.tasksAvailable[hash] = task
}
return nproc, nil
return stale, nil
}
func (s *stateSync) processNodeData(blob []byte, batch ethdb.Batch) (common.Hash, error) {
// processNodeData tries to inject a trie node data blob delivered from a remote
// peer into te state trie, returning whether anything useful was written or any
// error occured.
func (s *stateSync) processNodeData(blob []byte) (bool, common.Hash, error) {
// Convert the delivered data blob into a trie sync result
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
// Process the trie node and write any finalized data
batch := s.d.stateDB.NewBatch()
committed, _, err := s.sched.Process([]trie.SyncResult{res}, batch)
if committed {
if werr := batch.Write(); werr != nil {
return false, common.Hash{}, werr
}
}
return committed, res.Hash, err
}
func (s *stateSync) updateStats(processed int, duration time.Duration) {
// updateStats bumps the various state sync progress counters and displays a log
// message for the user to see.
func (s *stateSync) updateStats(processed, 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.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))
s.d.syncStatsState.processed += uint64(processed)
s.d.syncStatsState.duplicate += uint64(duplicate)
s.d.syncStatsState.unexpected += uint64(unexpected)
log.Info("Imported new state entries", "count", processed, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.processed, "pending", s.d.syncStatsState.pending, "retry", len(s.tasksAvailable), "duplicate", s.d.syncStatsState.duplicate, "unexpected", s.d.syncStatsState.unexpected)
}

7
trie/sync.go
View file

@ -28,6 +28,10 @@ import (
// node it did not request.
var ErrNotRequested = errors.New("not requested")
// ErrAlreadyProcessed is returned by the trie sync when it's requested to process a
// node it already processed previously.
var ErrAlreadyProcessed = errors.New("already processed")
// request represents a scheduled or already in-flight state retrieval request.
type request struct {
hash common.Hash // Hash of the node data content to retrieve
@ -153,6 +157,9 @@ func (s *TrieSync) Process(results []SyncResult, dbw DatabaseWriter) (bool, int,
if request == nil {
return committed, i, ErrNotRequested
}
if request.data != nil {
return committed, i, ErrAlreadyProcessed
}
// If the item is a raw entry request, commit directly
if request.raw {
request.data = item.Data