From a89b28f9b4ddf129cb55d4e30274b842658ce3bd Mon Sep 17 00:00:00 2001 From: zsfelfoldi Date: Sat, 26 Mar 2016 14:52:41 +0100 Subject: [PATCH] eth: quit peer handlers when shutting down, wait for event loop handlers to exit --- common/shutdown.go | 79 ++++++++++++++++++++++++++++++++++++ core/tx_pool.go | 4 ++ eth/filters/filter_system.go | 4 ++ eth/gasprice.go | 4 ++ eth/handler.go | 44 ++++++++++++++++++-- eth/peer.go | 14 +++++++ event/event.go | 22 +++++++++- miner/miner.go | 4 ++ miner/worker.go | 4 ++ 9 files changed, 175 insertions(+), 4 deletions(-) create mode 100644 common/shutdown.go diff --git a/common/shutdown.go b/common/shutdown.go new file mode 100644 index 0000000000..8eedaeddf6 --- /dev/null +++ b/common/shutdown.go @@ -0,0 +1,79 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package common + +import ( + "sync" +) + +// ShutdownManager implements a sync mechanism that waits for multiple jobs +// to finish and also prevents starting new jobs after the shutdown process +// has started. +type ShutdownManager struct { + wg sync.WaitGroup + lock sync.RWMutex + stop bool + stopChn chan struct{} +} + +// NewShutdownManager creates a new ShutdownManager instance +func NewShutdownManager() *ShutdownManager { + return &ShutdownManager{stopChn: make(chan struct{})} +} + +// Enter should be called before beginning a new job. If Enter returns false, +// the shutdown process has already been started. In this case, Exit should not +// be called. +func (self *ShutdownManager) Enter() bool { + self.lock.RLock() + defer self.lock.RUnlock() + + if self.stop { + return false + } + self.wg.Add(1) + return true +} + +// Exit should be called after finishing a job +func (self *ShutdownManager) Exit() { + self.wg.Done() +} + +// Shutdown initiates the shutdown process. After calling it, any subsequent +// calls to Enter will return false. Shutdown returns after all active jobs +// have finished and called Exit. +func (self *ShutdownManager) Shutdown() { + self.lock.Lock() + self.stop = true + self.lock.Unlock() + close(self.stopChn) + self.wg.Wait() +} + +// Stopped returns true if the shutdown process has already been started +func (self *ShutdownManager) Stopped() bool { + self.lock.RLock() + defer self.lock.RUnlock() + + return self.stop +} + +// StopChannel returns a channel which is closed when the shutdown starts +func (self *ShutdownManager) StopChannel() chan struct{} { + return self.stopChn +} \ No newline at end of file diff --git a/core/tx_pool.go b/core/tx_pool.go index f4e964bf79..c5e7ba55e3 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -98,6 +98,10 @@ func (pool *TxPool) eventLoop() { // Track chain events. When a chain events occurs (new chain canon block) // we need to know the new state. The new state will help us determine // the nonces in the managed state + if !pool.events.LoopStarted() { + return + } + defer pool.events.LoopStopped() for ev := range pool.events.Chan() { switch ev := ev.Data.(type) { case ChainHeadEvent: diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index b61a493b62..f851693a1d 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -131,6 +131,10 @@ func (fs *FilterSystem) Get(id int) *Filter { // filterLoop waits for specific events from ethereum and fires their handlers // when the filter matches the requirements. func (fs *FilterSystem) filterLoop() { + if !fs.sub.LoopStarted() { + return + } + defer fs.sub.LoopStopped() for event := range fs.sub.Chan() { switch ev := event.Data.(type) { case core.ChainEvent: diff --git a/eth/gasprice.go b/eth/gasprice.go index e0de89e621..20368d6cd0 100644 --- a/eth/gasprice.go +++ b/eth/gasprice.go @@ -103,6 +103,10 @@ func (self *GasPriceOracle) processPastBlocks(chain *core.BlockChain) { func (self *GasPriceOracle) listenLoop() { events := self.eth.EventMux().Subscribe(core.ChainEvent{}, core.ChainSplitEvent{}) defer events.Unsubscribe() + if !events.LoopStarted() { + return + } + defer events.LoopStopped() for event := range events.Chan() { switch event := event.Data.(type) { diff --git a/eth/handler.go b/eth/handler.go index 2c5cae479f..5591647d78 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -82,6 +82,7 @@ type ProtocolManager struct { // and processing wg sync.WaitGroup quit bool + sm *common.ShutdownManager } // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable @@ -104,6 +105,7 @@ func NewProtocolManager(fastSync bool, networkId int, mux *event.TypeMux, txpool newPeerCh: make(chan *peer, 1), txsyncCh: make(chan *txsync), quitSync: make(chan struct{}), + sm: common.NewShutdownManager(), } // Initiate a sub-protocol for every implemented version we can handle manager.SubProtocols = make([]p2p.Protocol, 0, len(ProtocolVersions)) @@ -119,9 +121,17 @@ func NewProtocolManager(fastSync bool, networkId int, mux *event.TypeMux, txpool Version: version, Length: ProtocolLengths[i], Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { + if !manager.sm.Enter() { + return nil + } + defer manager.sm.Exit() peer := manager.newPeer(int(version), p, rw) - manager.newPeerCh <- peer - return manager.handle(peer) + select { + case manager.newPeerCh <- peer: + return manager.handle(peer) + case <-manager.sm.StopChannel(): + return nil + } }, NodeInfo: func() interface{} { return manager.NodeInfo() @@ -196,8 +206,25 @@ func (pm *ProtocolManager) Stop() { pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop close(pm.quitSync) // quits syncer, fetcher, txsyncLoop + stopped := make(chan struct{}) + go func() { + for { + // drop all peers + for _, id := range pm.peers.AllPeerIDs() { + pm.removePeer(id) + } + select { + case <-stopped: + return + case <-time.After(time.Millisecond * 100): + } + } + }() + // Wait for any process action + pm.sm.Shutdown() pm.wg.Wait() + close(stopped) glog.V(logger.Info).Infoln("Ethereum protocol handler stopped") } @@ -239,7 +266,7 @@ func (pm *ProtocolManager) handle(p *peer) error { pm.syncTransactions(p) // main loop. handle incoming messages. - for { + for !pm.sm.Stopped() { if err := pm.handleMsg(p); err != nil { glog.V(logger.Debug).Infof("%v: message handling failed: %v", p, err) return err @@ -256,6 +283,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err != nil { return err } + if pm.sm.Stopped() { + return nil + } if msg.Size > ProtocolMaxMsgSize { return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) } @@ -734,6 +764,10 @@ func (pm *ProtocolManager) BroadcastTx(hash common.Hash, tx *types.Transaction) // Mined broadcast loop func (self *ProtocolManager) minedBroadcastLoop() { // automatically stops if unsubscribe + if !self.minedBlockSub.LoopStarted() { + return + } + defer self.minedBlockSub.LoopStopped() for obj := range self.minedBlockSub.Chan() { switch ev := obj.Data.(type) { case core.NewMinedBlockEvent: @@ -745,6 +779,10 @@ func (self *ProtocolManager) minedBroadcastLoop() { func (self *ProtocolManager) txBroadcastLoop() { // automatically stops if unsubscribe + if !self.txSub.LoopStarted() { + return + } + defer self.txSub.LoopStopped() for obj := range self.txSub.Chan() { event := obj.Data.(core.TxPreEvent) self.BroadcastTx(event.Tx.Hash(), event.Tx) diff --git a/eth/peer.go b/eth/peer.go index 15ba22ff53..4d054a9132 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -388,6 +388,20 @@ func (ps *peerSet) Unregister(id string) error { return nil } +// AllPeerIDs returns a list of all registered peer IDs +func (ps *peerSet) AllPeerIDs() []string { + ps.lock.RLock() + defer ps.lock.RUnlock() + + res := make([]string, len(ps.peers)) + idx := 0 + for id, _ := range ps.peers { + res[idx] = id + idx++ + } + return res +} + // Peer retrieves the registered peer with the given id. func (ps *peerSet) Peer(id string) *peer { ps.lock.RLock() diff --git a/event/event.go b/event/event.go index 57dd52baa1..44c068298e 100644 --- a/event/event.go +++ b/event/event.go @@ -23,6 +23,8 @@ import ( "reflect" "sync" "time" + + "github.com/ethereum/go-ethereum/common" ) // Event is a time-tagged notification pushed to subscribers. @@ -42,6 +44,8 @@ type Subscription interface { // The event channel is closed. // Unsubscribe can be called more than once. Unsubscribe() + LoopStarted() bool + LoopStopped() } // A TypeMux dispatches events to registered receivers. Receivers can be @@ -53,6 +57,7 @@ type TypeMux struct { mutex sync.RWMutex subm map[reflect.Type][]*muxsub stopped bool + sm *common.ShutdownManager } // ErrMuxClosed is returned when Posting on a closed TypeMux. @@ -68,6 +73,9 @@ func (mux *TypeMux) Subscribe(types ...interface{}) Subscription { if mux.stopped { close(sub.postC) } else { + if mux.sm == nil { + mux.sm = common.NewShutdownManager() + } if mux.subm == nil { mux.subm = make(map[reflect.Type][]*muxsub) } @@ -119,7 +127,11 @@ func (mux *TypeMux) Stop() { } mux.subm = nil mux.stopped = true + sm := mux.sm mux.mutex.Unlock() + if sm != nil { + sm.Shutdown() + } } func (mux *TypeMux) del(s *muxsub) { @@ -187,6 +199,14 @@ func (s *muxsub) Unsubscribe() { s.closewait() } +func (s *muxsub) LoopStarted() bool { + return s.mux.sm.Enter() +} + +func (s *muxsub) LoopStopped() { + s.mux.sm.Exit() +} + func (s *muxsub) closewait() { s.closeMu.Lock() defer s.closeMu.Unlock() @@ -215,4 +235,4 @@ func (s *muxsub) deliver(event *Event) { case s.postC <- event: case <-s.closing: } -} +} \ No newline at end of file diff --git a/miner/miner.go b/miner/miner.go index e52cefaab2..54561b97f5 100644 --- a/miner/miner.go +++ b/miner/miner.go @@ -64,6 +64,10 @@ func New(eth core.Backend, mux *event.TypeMux, pow pow.PoW) *Miner { // and halt your mining operation for as long as the DOS continues. func (self *Miner) update() { events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{}) + if !events.LoopStarted() { + return + } + defer events.LoopStopped() out: for ev := range events.Chan() { switch ev.Data.(type) { diff --git a/miner/worker.go b/miner/worker.go index b3ddf9707a..fe941b8551 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -216,6 +216,9 @@ func (self *worker) update() { eventSub := self.mux.Subscribe(core.ChainHeadEvent{}, core.ChainSideEvent{}, core.TxPreEvent{}) defer eventSub.Unsubscribe() + if !eventSub.LoopStarted() { + return + } eventCh := eventSub.Chan() for { select { @@ -223,6 +226,7 @@ func (self *worker) update() { if !ok { // Event subscription closed, set the channel to nil to stop spinning eventCh = nil + eventSub.LoopStopped() continue } // A real event arrived, process interesting content