diff --git a/eth/backend.go b/eth/backend.go index ed79340f59..da2e338ced 100644 --- a/eth/backend.go +++ b/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 @@ -84,8 +81,9 @@ type Ethereum struct { engine consensus.Engine accountManager *accounts.Manager - bloomRequests chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests - bloomIndexer *core.ChainIndexer // Bloom indexer operating during block imports + 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 @@ -145,17 +143,17 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { log.Info("Initialised chain configuration", "config", chainConfig) eth := &Ethereum{ - config: config, - chainDb: chainDb, - eventMux: ctx.EventMux, - accountManager: ctx.AccountManager, - engine: CreateConsensusEngine(ctx, chainConfig, &config.Ethash, config.Miner.Notify, config.Miner.Noverify, chainDb), - shutdownChan: make(chan bool), - networkID: config.NetworkId, - gasPrice: config.Miner.GasPrice, - etherbase: config.Miner.Etherbase, - bloomRequests: make(chan chan *bloombits.Retrieval), - bloomIndexer: NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms), + config: config, + chainDb: chainDb, + eventMux: ctx.EventMux, + accountManager: ctx.AccountManager, + engine: CreateConsensusEngine(ctx, chainConfig, &config.Ethash, config.Miner.Notify, config.Miner.Noverify, chainDb), + closeBloomHandler: make(chan bool), + networkID: config.NetworkId, + gasPrice: config.Miner.GasPrice, + etherbase: config.Miner.Etherbase, + bloomRequests: make(chan chan *bloombits.Retrieval), + bloomIndexer: NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms), } bcVersion := rawdb.ReadDatabaseVersion(chainDb) @@ -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() - s.blockchain.Stop() - s.engine.Close() + // Stop all the peer-related stuff first. 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 } diff --git a/eth/bloombits.go b/eth/bloombits.go index 9a31997d60..35522b9bfa 100644 --- a/eth/bloombits.go +++ b/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: diff --git a/eth/handler.go b/eth/handler.go index 236e50729c..9d2dce6e90 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -87,14 +87,11 @@ type ProtocolManager struct { whitelist map[uint64]common.Hash // channels for fetcher, syncer, txsyncLoop - 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 + newPeerCh chan *peer + txsyncCh chan *txsync + quitSync chan struct{} + wg sync.WaitGroup + peerWG sync.WaitGroup // Test fields or hooks broadcastTxAnnouncesOnly bool // Testing field, disable transaction propagation @@ -105,17 +102,16 @@ type ProtocolManager struct { func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCheckpoint, mode downloader.SyncMode, networkID uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database, cacheLimit int, whitelist map[uint64]common.Hash) (*ProtocolManager, error) { // Create the protocol manager with the base fields manager := &ProtocolManager{ - networkID: networkID, - forkFilter: forkid.NewFilter(blockchain), - eventMux: mux, - txpool: txpool, - blockchain: blockchain, - peers: newPeerSet(), - whitelist: whitelist, - newPeerCh: make(chan *peer), - noMorePeers: make(chan struct{}), - txsyncCh: make(chan *txsync), - quitSync: make(chan struct{}), + networkID: networkID, + forkFilter: forkid.NewFilter(blockchain), + eventMux: mux, + txpool: txpool, + blockchain: blockchain, + peers: newPeerSet(), + whitelist: whitelist, + newPeerCh: make(chan *peer), + txsyncCh: make(chan *txsync), + quitSync: make(chan struct{}), } if mode == downloader.FullSync { // The database seems empty as the current block is the genesis. Yet the fast @@ -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) - defer pm.wg.Done() + pm.peerWG.Add(1) + 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. - // After this send has completed, no new peers will be accepted. - pm.noMorePeers <- struct{}{} - - // Quit fetcher, txsyncLoop. + // Quit syncer and txsync64. + // After this is done, no new peers will be accepted. 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() - - // Wait for all peer handler goroutines and the loops to come down. - pm.wg.Wait() + pm.peerWG.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 } diff --git a/eth/sync.go b/eth/sync.go index 0709706c91..72f9960e83 100644 --- a/eth/sync.go +++ b/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 -// downloading hashes and blocks as well as handling the announcement handler. +// syncer runs in its own goroutine and coordinates sync-related components. 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 - forceSync := time.NewTicker(forceSyncCycle) - defer forceSync.Stop() + for pm.syncTriggerWait() { + pm.synchronise(pm.peers.BestPeer()) + } +} +// 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: - // Make sure we have peers to select from, then sync - if pm.peers.Len() < minDesiredPeerCount { - break + case <-pm.quitSync: + return false + default: + 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 } } }