eth: improve shutdown synchronization

Most goroutines started by eth.Ethereum didn't have any shutdown sync at
all, which lead to weird error messages when quitting the client.

This change improves the clean shutdown path by stopping all internal
components in dependency order and waiting for them to actually be
stopped before shutdown is considered done. In particular, we now stop
everything related to peers before stopping 'resident' parts such as
core.BlockChain.

eth/backend.go

@@ -67,9 +67,6 @@ type LesServer interface {
 type Ethereum struct {
 	config *Config
 
-	// Channel for shutting down the service
-	shutdownChan chan bool
-
 	// Handlers
 	txPool          *core.TxPool
 	blockchain      *core.BlockChain
@@ -86,6 +83,7 @@ type Ethereum struct {
 
 	bloomRequests     chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests
 	bloomIndexer      *core.ChainIndexer             // Bloom indexer operating during block imports
+	closeBloomHandler chan bool
 
 	APIBackend *EthAPIBackend
 
@@ -150,7 +148,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 		eventMux:          ctx.EventMux,
 		accountManager:    ctx.AccountManager,
 		engine:            CreateConsensusEngine(ctx, chainConfig, &config.Ethash, config.Miner.Notify, config.Miner.Noverify, chainDb),
-		shutdownChan:   make(chan bool),
+		closeBloomHandler: make(chan bool),
 		networkID:         config.NetworkId,
 		gasPrice:          config.Miner.GasPrice,
 		etherbase:         config.Miner.Etherbase,
@@ -557,18 +555,20 @@ func (s *Ethereum) Start(srvr *p2p.Server) error {
 // Stop implements node.Service, terminating all internal goroutines used by the
 // Ethereum protocol.
 func (s *Ethereum) Stop() error {
-	s.bloomIndexer.Close()
+	// Stop all the peer-related stuff first.
-	s.blockchain.Stop()
-	s.engine.Close()
 	s.protocolManager.Stop()
 	if s.lesServer != nil {
 		s.lesServer.Stop()
 	}
+
+	// Then stop everything else.
+	s.bloomIndexer.Close()
+	close(s.closeBloomHandler)
 	s.txPool.Stop()
 	s.miner.Stop()
-	s.eventMux.Stop()
+	s.blockchain.Stop()
-
+	s.engine.Close()
 	s.chainDb.Close()
-	close(s.shutdownChan)
+	s.eventMux.Stop()
 	return nil
 }

eth/bloombits.go

@@ -54,7 +54,7 @@ func (eth *Ethereum) startBloomHandlers(sectionSize uint64) {
 		go func() {
 			for {
 				select {
-				case <-eth.shutdownChan:
+				case <-eth.closeBloomHandler:
 					return
 
 				case request := <-eth.bloomRequests:

eth/handler.go

@@ -90,11 +90,8 @@ type ProtocolManager struct {
 	newPeerCh chan *peer
 	txsyncCh  chan *txsync
 	quitSync  chan struct{}
-	noMorePeers chan struct{}
-
-	// wait group is used for graceful shutdowns during downloading
-	// and processing
 	wg        sync.WaitGroup
+	peerWG    sync.WaitGroup
 
 	// Test fields or hooks
 	broadcastTxAnnouncesOnly bool // Testing field, disable transaction propagation
@@ -113,7 +110,6 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh
 		peers:      newPeerSet(),
 		whitelist:  whitelist,
 		newPeerCh:  make(chan *peer),
-		noMorePeers: make(chan struct{}),
 		txsyncCh:   make(chan *txsync),
 		quitSync:   make(chan struct{}),
 	}
@@ -216,8 +212,8 @@ func (pm *ProtocolManager) makeProtocol(version uint) p2p.Protocol {
 			peer := pm.newPeer(int(version), p, rw, pm.txpool.Get)
 			select {
 			case pm.newPeerCh <- peer:
-				pm.wg.Add(1)
+				pm.peerWG.Add(1)
-				defer pm.wg.Done()
+				defer pm.peerWG.Done()
 				return pm.handle(peer)
 			case <-pm.quitSync:
 				return p2p.DiscQuitting
@@ -260,40 +256,38 @@ func (pm *ProtocolManager) Start(maxPeers int) {
 	pm.maxPeers = maxPeers
 
 	// broadcast transactions
+	pm.wg.Add(1)
 	pm.txsCh = make(chan core.NewTxsEvent, txChanSize)
 	pm.txsSub = pm.txpool.SubscribeNewTxsEvent(pm.txsCh)
 	go pm.txBroadcastLoop()
 
 	// broadcast mined blocks
+	pm.wg.Add(1)
 	pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
 	go pm.minedBroadcastLoop()
 
 	// start sync handlers
+	pm.wg.Add(2)
 	go pm.syncer()
 	go pm.txsyncLoop64() // TODO(karalabe): Legacy initial tx echange, drop with eth/64.
 }
 
 func (pm *ProtocolManager) Stop() {
-	log.Info("Stopping Ethereum protocol")
-
 	pm.txsSub.Unsubscribe()        // quits txBroadcastLoop
 	pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
 
-	// Quit the sync loop.
+	// Quit syncer and txsync64.
-	// After this send has completed, no new peers will be accepted.
+	// After this is done, no new peers will be accepted.
-	pm.noMorePeers <- struct{}{}
-
-	// Quit fetcher, txsyncLoop.
 	close(pm.quitSync)
+	pm.downloader.Cancel()
+	pm.wg.Wait()
 
 	// 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
 	// will exit when they try to register.
 	pm.peers.Close()
-
+	pm.peerWG.Wait()
-	// Wait for all peer handler goroutines and the loops to come down.
-	pm.wg.Wait()
 
 	log.Info("Ethereum protocol stopped")
 }
@@ -883,9 +877,10 @@ func (pm *ProtocolManager) BroadcastTransactions(txs types.Transactions, propaga
 	}
 }
 
-// Mined broadcast loop
+// minedBroadcastLoop sends mined blocks to connected peers.
 func (pm *ProtocolManager) minedBroadcastLoop() {
-	// automatically stops if unsubscribe
+	defer pm.wg.Done()
+
 	for obj := range pm.minedBlockSub.Chan() {
 		if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
 			pm.BroadcastBlock(ev.Block, true)  // First propagate block to peers
@@ -894,7 +889,10 @@ func (pm *ProtocolManager) minedBroadcastLoop() {
 	}
 }
 
+// txBroadcastLoop announces new transactions to connected peers.
 func (pm *ProtocolManager) txBroadcastLoop() {
+	defer pm.wg.Done()
+
 	for {
 		select {
 		case event := <-pm.txsCh:
@@ -906,7 +904,6 @@ func (pm *ProtocolManager) txBroadcastLoop() {
 			pm.BroadcastTransactions(event.Txs, true)  // First propagate transactions to peers
 			pm.BroadcastTransactions(event.Txs, false) // Only then announce to the rest
 
-		// Err() channel will be closed when unsubscribing.
 		case <-pm.txsSub.Err():
 			return
 		}

eth/sync.go

@@ -32,7 +32,7 @@ const (
 	forceSyncCycle      = 10 * time.Second // Time interval to force syncs, even if few peers are available
 	minDesiredPeerCount = 5                // Amount of peers desired to start syncing
 
-	// This is the target size for the packs of transactions sent by txsyncLoop.
+	// This is the target size for the packs of transactions sent by txsyncLoop64.
 	// A pack can get larger than this if a single transactions exceeds this size.
 	txsyncPackSize = 100 * 1024
 )
@@ -81,12 +81,15 @@ func (pm *ProtocolManager) syncTransactions(p *peer) {
 // transactions. In order to minimise egress bandwidth usage, we send
 // the transactions in small packs to one peer at a time.
 func (pm *ProtocolManager) txsyncLoop64() {
+	defer pm.wg.Done()
+
 	var (
 		pending = make(map[enode.ID]*txsync)
 		sending = false               // whether a send is active
 		pack    = new(txsync)         // the pack that is being sent
 		done    = make(chan error, 1) // result of the send
 	)
+
 	// send starts a sending a pack of transactions from the sync.
 	send := func(s *txsync) {
 		if s.p.version >= eth65 {
@@ -149,35 +152,41 @@ func (pm *ProtocolManager) txsyncLoop64() {
 	}
 }
 
-// syncer is responsible for periodically synchronising with the network, both
+// syncer runs in its own goroutine and coordinates sync-related components.
-// downloading hashes and blocks as well as handling the announcement handler.
 func (pm *ProtocolManager) syncer() {
-	// Start and ensure cleanup of sync mechanisms
+	defer pm.wg.Done()
+
 	pm.blockFetcher.Start()
 	pm.txFetcher.Start()
 	defer pm.blockFetcher.Stop()
 	defer pm.txFetcher.Stop()
 	defer pm.downloader.Terminate()
 
-	// Wait for different events to fire synchronisation operations
+	for pm.syncTriggerWait() {
-	forceSync := time.NewTicker(forceSyncCycle)
+		pm.synchronise(pm.peers.BestPeer())
-	defer forceSync.Stop()
+	}
+}
 
+// syncTriggerWait waits for sync start conditions to be met.
+func (pm *ProtocolManager) syncTriggerWait() bool {
+	force := time.NewTimer(forceSyncCycle)
+	defer force.Stop()
 	for {
 		select {
-		case <-pm.newPeerCh:
+		case <-pm.quitSync:
-			// Make sure we have peers to select from, then sync
+			return false
-			if pm.peers.Len() < minDesiredPeerCount {
+		default:
-				break
+			if pm.peers.Len() >= minDesiredPeerCount {
+				return true
+			}
+			select {
+			case <-pm.newPeerCh:
+				// Go round and check the peer count again.
+			case <-force.C:
+				return true
+			case <-pm.quitSync:
+				return false
 			}
-			go pm.synchronise(pm.peers.BestPeer())
-
-		case <-forceSync.C:
-			// Force a sync even if not enough peers are present
-			go pm.synchronise(pm.peers.BestPeer())
-
-		case <-pm.noMorePeers:
-			return
 		}
 	}
 }