verified concurrency fixes in beacon sync and general downloader stability

core/blockchain.go

@@ -1156,28 +1156,29 @@ func (bc *BlockChain) SnapSyncComplete(hash common.Hash) error {
 	if !bc.chainmu.TryLock() {
 		return errChainStopped
 	}
-	defer bc.chainmu.Unlock()
-
 	// Reset the trie database with the fresh snap synced state.
 	root := block.Root()
 	if bc.triedb.Scheme() == rawdb.PathScheme {
 		if err := bc.triedb.Enable(root); err != nil {
+			bc.chainmu.Unlock()
 			return err
 		}
 	}
 	if !bc.HasState(root) {
+		bc.chainmu.Unlock()
 		return fmt.Errorf("non existent state [%x..]", root[:4])
 	}
+	// If all checks out, manually set the head block.
+	bc.currentBlock.Store(block.Header())
+	headBlockGauge.Update(int64(block.NumberU64()))
+	bc.chainmu.Unlock() // Unlock specificly before Rebuild to avoid deadlock
+
 	// Destroy any existing state snapshot and regenerate it in the background,
 	// also resuming the normal maintenance of any previously paused snapshot.
 	if bc.snaps != nil {
 		bc.snaps.Rebuild(root)
 	}
 
-	// If all checks out, manually set the head block.
-	bc.currentBlock.Store(block.Header())
-	headBlockGauge.Update(int64(block.NumberU64()))
-
 	log.Info("Committed new head block", "number", block.Number(), "hash", hash)
 	return nil
 }

eth/downloader/beaconsync.go

@@ -49,56 +49,44 @@ func newBeaconBackfiller(dl *Downloader, success func()) backfiller {
 	}
 }
 
+// suspend cancels any background downloader threads and returns the last header
+// that has been successfully backfilled (potentially in a previous run), or the
+// genesis.
 // suspend cancels any background downloader threads and returns the last header
 // that has been successfully backfilled (potentially in a previous run), or the
 // genesis.
 func (b *beaconBackfiller) suspend() *types.Header {
-	// If no filling is running, don't waste cycles
 	b.lock.Lock()
-	filling := b.filling
+	defer b.lock.Unlock()
-	filled := b.filled
-	started := b.started
-	b.lock.Unlock()
 
-	if !filling {
+	if !b.filling {
-		log.Debug("Backfiller was inactive")
+		return b.filled
-		return filled // Return the filled header on the previous sync completion
 	}
-	// A previous filling should be running, though it may happen that it hasn't
+	// Stop the downloader. It is safe to call Cancel concurrently with synchronise
-	// yet started (being done on a new goroutine). Many concurrent beacon head
+	// (which might be running in resume's goroutine), as Cancel is thread-safe.
-	// announcements can lead to sync start/stop thrashing. In that case we need
-	// to wait for initialization before we can safely cancel it. It is safe to
-	// read this channel multiple times, it gets closed on startup.
-	<-started
-
-	// Now that we're sure the downloader successfully started up, we can cancel
-	// it safely without running the risk of data races.
 	b.downloader.Cancel()
-	log.Debug("Backfiller has been suspended")
+	b.filling = false
 
-	// Sync cycle was just terminated, retrieve and return the last filled header.
+	// Wait for the sync routine to finish to ensure no zombie processes
-	// Can't use `filled` as that contains a stale value from before cancellation.
+	<-b.started
+
+	log.Debug("Backfiller suspended")
 	return b.downloader.blockchain.CurrentSnapBlock()
 }
 
 // resume starts the downloader threads for backfilling state and chain data.
 func (b *beaconBackfiller) resume() {
 	b.lock.Lock()
+	defer b.lock.Unlock()
+
 	if b.filling {
-		// If a previous filling cycle is still running, just ignore this start
-		// request. // TODO(karalabe): We should make this channel driven
-		b.lock.Unlock()
-		log.Debug("Backfiller is running")
 		return
 	}
 	b.filling = true
 	b.filled = nil
 	b.started = make(chan struct{})
-	b.lock.Unlock()
 
-	// Start the backfilling on its own thread since the downloader does not have
+	go func(started chan struct{}) {
-	// its own lifecycle runloop.
-	go func() {
 		// Set the backfiller to non-filling when download completes
 		defer func() {
 			b.lock.Lock()
@@ -108,7 +96,7 @@ func (b *beaconBackfiller) resume() {
 		}()
 		// If the downloader fails, report an error as in beacon chain mode there
 		// should be no errors as long as the chain we're syncing to is valid.
-		if err := b.downloader.synchronise(b.started); err != nil {
+		if err := b.downloader.synchronise(started); err != nil {
 			log.Error("Beacon backfilling failed", "err", err)
 			return
 		}
@@ -118,7 +106,8 @@ func (b *beaconBackfiller) resume() {
 			b.success()
 		}
 		log.Debug("Backfilling completed")
-	}()
+	}(b.started)
+
 	log.Debug("Backfilling started")
 }