eth, les: wait for the downloads to be complete before stopping

This commit is contained in:
Igor Mandrigin 2018-02-26 22:06:38 +01:00
parent 46a5532ac5
commit c7f4965e97
No known key found for this signature in database
GPG key ID: 6343C2A9279A887E
5 changed files with 14 additions and 2 deletions

View file

@ -141,6 +141,8 @@ type Downloader struct {
quitCh chan struct{} // Quit channel to signal termination
quitLock sync.RWMutex // Lock to prevent double closes
downloads sync.WaitGroup // Keeps track of the currently active downloads
// Testing hooks
syncInitHook func(uint64, uint64) // Method to call upon initiating a new sync run
bodyFetchHook func([]*types.Header) // Method to call upon starting a block body fetch
@ -398,7 +400,9 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode
// specified peer and head hash.
func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.Int) (err error) {
d.mux.Post(StartEvent{})
d.downloads.Add(1)
defer func() {
d.downloads.Done()
// reset on error
if err != nil {
d.mux.Post(FailedEvent{err})
@ -528,6 +532,10 @@ func (d *Downloader) Terminate() {
// Cancel any pending download requests
d.Cancel()
// Wait, so external dependencies aren't destroyed
// until the download processing is done.
d.downloads.Wait()
}
// fetchHeight retrieves the head header of the remote peer to aid in estimating

View file

@ -230,6 +230,9 @@ func (pm *ProtocolManager) Stop() {
// Quit fetcher, txsyncLoop.
close(pm.quitSync)
// Stop downloader and make sure that all the running downloads are complete.
pm.downloader.Terminate()
// Disconnect existing sessions.
// This also closes the gate for any new registrations on the peer set.
// sessions which are already established but not added to pm.peers yet

View file

@ -135,7 +135,6 @@ func (pm *ProtocolManager) syncer() {
// Start and ensure cleanup of sync mechanisms
pm.fetcher.Start()
defer pm.fetcher.Stop()
defer pm.downloader.Terminate()
// Wait for different events to fire synchronisation operations
forceSync := time.NewTicker(forceSyncCycle)

View file

@ -241,6 +241,9 @@ func (pm *ProtocolManager) Stop() {
close(pm.quitSync) // quits syncer, fetcher
// Stop downloader and make sure that all the running downloads are complete.
pm.downloader.Terminate()
// Disconnect existing sessions.
// This also closes the gate for any new registrations on the peer set.
// sessions which are already established but not added to pm.peers yet

View file

@ -36,7 +36,6 @@ func (pm *ProtocolManager) syncer() {
// Start and ensure cleanup of sync mechanisms
//pm.fetcher.Start()
//defer pm.fetcher.Stop()
defer pm.downloader.Terminate()
// Wait for different events to fire synchronisation operations
//forceSync := time.Tick(forceSyncCycle)