forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-17 18:30:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/downloader: flush state data before exit #16280 (#1042)

Browse source
This commit is contained in:
Daniel Liu 2025-05-26 15:15:32 +08:00 committed by GitHub
parent c7349e77d0
commit e581aa5240
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 13 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
eth/downloader/statesync.go
View file

@ -320,16 +320,22 @@ func (s *stateSync) Cancel() error {
// receive data from peers, rather those are buffered up in the downloader and // 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 // pushed here async. The reason is to decouple processing from data receipt
// and timeouts. // and timeouts.
func (s *stateSync) loop() error { func (s *stateSync) loop() (err error) {
close(s.started) close(s.started)
// Listen for new peer events to assign tasks to them // Listen for new peer events to assign tasks to them
newPeer := make(chan *peerConnection, 1024) newPeer := make(chan *peerConnection, 1024)
peerSub := s.d.peers.SubscribeNewPeers(newPeer) peerSub := s.d.peers.SubscribeNewPeers(newPeer)
defer peerSub.Unsubscribe() defer peerSub.Unsubscribe()
defer func() {
cerr := s.commit(true)
if err == nil {
err = cerr
}
}()
// Keep assigning new tasks until the sync completes or aborts // Keep assigning new tasks until the sync completes or aborts
for s.sched.Pending() > 0 { for s.sched.Pending() > 0 {
if err := s.commit(false); err != nil { if err = s.commit(false); err != nil {
return err return err
} }
s.assignTasks() s.assignTasks()
@ -342,7 +348,7 @@ func (s *stateSync) loop() error {
return errCancelStateFetch return errCancelStateFetch
case <-s.d.cancelCh: case <-s.d.cancelCh:
return errCancelStateFetch return errCanceled
case req := <-s.deliver: case req := <-s.deliver:
// Response, disconnect or timeout triggered, drop the peer if stalling // Response, disconnect or timeout triggered, drop the peer if stalling
@ -376,10 +382,9 @@ func (s *stateSync) loop() error {
log.Warn("Node data write error", "err", err) log.Warn("Node data write error", "err", err)
return err return err
} }
req.peer.SetNodeDataIdle(delivered)
} }
} }
return s.commit(true) return nil
} }
func (s *stateSync) commit(force bool) error { func (s *stateSync) commit(force bool) error {
@ -388,7 +393,9 @@ func (s *stateSync) commit(force bool) error {
} }
start := time.Now() start := time.Now()
b := s.d.stateDB.NewBatch() b := s.d.stateDB.NewBatch()
s.sched.Commit(b) if err := s.sched.Commit(b); err != nil {
return err
}
if err := b.Write(); err != nil { if err := b.Write(); err != nil {
return fmt.Errorf("write DB error: %v", err) return fmt.Errorf("write DB error: %v", err)
} }