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

eth/downloader: improve state node error handling and retry check

Browse source
This commit is contained in:
Felix Lange 2017-05-30 17:22:08 +02:00 committed by Péter Szilágyi
parent 44b8cb852c
commit 5339c79b54
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
1 changed files with 10 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
eth/downloader/statesync.go
View file

@ -292,7 +292,10 @@ func (s *stateSync) popTasks(n int, req *stateReq) error {
func (s *stateSync) process(req *stateReq) (nproc int, err error) { func (s *stateSync) process(req *stateReq) (nproc int, err error) {
batch := s.d.stateDB.NewBatch() batch := s.d.stateDB.NewBatch()
for _, blob := range req.response { for _, blob := range req.response {
if hash, ok := s.processNodeData(blob, batch); ok { hash, err := s.processNodeData(blob, batch)
if err != nil && err != trie.ErrNotRequested {
return 0, fmt.Errorf("invalid state node %s: %v", err)
} else if err == nil {
nproc++ nproc++
delete(req.tasks, hash) delete(req.tasks, hash)
} }
@ -305,16 +308,18 @@ func (s *stateSync) process(req *stateReq) (nproc int, err error) {
atomic.StoreUint32(&s.d.fsPivotFails, 1) // Don't ever reset to 0, as that will unlock the pivot block 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.
npeers := s.d.peers.Len()
for hash, task := range req.tasks { for hash, task := range req.tasks {
if len(req.response) > 0 || req.timedOut() { if len(req.response) > 0 || req.timedOut() {
// Ensure that the item will be retried because it may have been excluded due // Ensure that the item will be retried because it may have been excluded due
// to a protocol limit. // to a protocol limit.
delete(task.triedPeers, req.peer.id) delete(task.triedPeers, req.peer.id)
} }
if npeers := s.d.peers.Len(); len(task.triedPeers) >= npeers { // Check retry limits.
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 nproc, fmt.Errorf("state node %s failed with all peers (%d tries, %d peers)", hash.TerminalString(), len(task.triedPeers), npeers)
} }
if task.numTries > maxStateNodeRetries { if task.numTries > maxStateNodeRetries && task.numTries >= npeers {
return nproc, fmt.Errorf("download of state node %s failed %d times", hash.TerminalString(), task.numTries) return nproc, fmt.Errorf("download of state node %s failed %d times", hash.TerminalString(), task.numTries)
} }
s.tasksAvailable[hash] = task s.tasksAvailable[hash] = task
@ -322,13 +327,13 @@ func (s *stateSync) process(req *stateReq) (nproc int, err error) {
return nproc, nil return nproc, nil
} }
func (s *stateSync) processNodeData(blob []byte, batch ethdb.Batch) (common.Hash, bool) { func (s *stateSync) processNodeData(blob []byte, batch ethdb.Batch) (common.Hash, error) {
res := trie.SyncResult{Data: blob} res := trie.SyncResult{Data: blob}
s.keccak.Reset() s.keccak.Reset()
s.keccak.Write(blob) s.keccak.Write(blob)
s.keccak.Sum(res.Hash[:0]) s.keccak.Sum(res.Hash[:0])
_, _, err := s.sched.Process([]trie.SyncResult{res}, batch) _, _, err := s.sched.Process([]trie.SyncResult{res}, batch)
return res.Hash, err == nil return res.Hash, err
} }
func (s *stateSync) updateStats(processed int, duration time.Duration) { func (s *stateSync) updateStats(processed int, duration time.Duration) {