aboutsummaryrefslogtreecommitdiffstats
path: root/miner/worker.go
diff options
context:
space:
mode:
authorFelix Lange <fjl@twurst.com>2016-03-29 09:08:16 +0800
committerFelix Lange <fjl@twurst.com>2016-05-09 19:03:08 +0800
commit56ed6152a11592d20220daf6322e94a009e6236d (patch)
tree0a0d5985832e32fdd1d9c3dc1deff89a85811099 /miner/worker.go
parentf821b0188a27bca08cada87c5b746ef9455a2e96 (diff)
downloaddexon-56ed6152a11592d20220daf6322e94a009e6236d.tar
dexon-56ed6152a11592d20220daf6322e94a009e6236d.tar.gz
dexon-56ed6152a11592d20220daf6322e94a009e6236d.tar.bz2
dexon-56ed6152a11592d20220daf6322e94a009e6236d.tar.lz
dexon-56ed6152a11592d20220daf6322e94a009e6236d.tar.xz
dexon-56ed6152a11592d20220daf6322e94a009e6236d.tar.zst
dexon-56ed6152a11592d20220daf6322e94a009e6236d.zip
core, eth, miner: improve shutdown synchronisation
Shutting down geth prints hundreds of annoying error messages in some cases. The errors appear because the Stop method of eth.ProtocolManager, miner.Miner and core.TxPool is asynchronous. Left over peer sessions generate events which are processed after Stop even though the database has already been closed. The fix is to make Stop synchronous using sync.WaitGroup. For eth.ProtocolManager, in order to make use of WaitGroup safe, we need a way to stop new peer sessions from being added while waiting on the WaitGroup. The eth protocol Run function now selects on a signaling channel and adds to the WaitGroup only if ProtocolManager is not shutting down. For miner.worker and core.TxPool the number of goroutines is static, WaitGroup can be used in the usual way without additional synchronisation.
Diffstat (limited to 'miner/worker.go')
-rw-r--r--miner/worker.go62
1 files changed, 28 insertions, 34 deletions
diff --git a/miner/worker.go b/miner/worker.go
index 21588e310..3d1928bf6 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -94,10 +94,13 @@ type worker struct {
mu sync.Mutex
+ // update loop
+ mux *event.TypeMux
+ events event.Subscription
+ wg sync.WaitGroup
+
agents map[Agent]struct{}
recv chan *Result
- mux *event.TypeMux
- quit chan struct{}
pow pow.PoW
eth core.Backend
@@ -138,13 +141,14 @@ func newWorker(config *core.ChainConfig, coinbase common.Address, eth core.Backe
possibleUncles: make(map[common.Hash]*types.Block),
coinbase: coinbase,
txQueue: make(map[common.Hash]*types.Transaction),
- quit: make(chan struct{}),
agents: make(map[Agent]struct{}),
fullValidation: false,
}
+ worker.events = worker.mux.Subscribe(core.ChainHeadEvent{}, core.ChainSideEvent{}, core.TxPreEvent{})
+ worker.wg.Add(1)
go worker.update()
- go worker.wait()
+ go worker.wait()
worker.commitNewWork()
return worker
@@ -184,9 +188,12 @@ func (self *worker) start() {
}
func (self *worker) stop() {
+ // Quit update.
+ self.events.Unsubscribe()
+ self.wg.Wait()
+
self.mu.Lock()
defer self.mu.Unlock()
-
if atomic.LoadInt32(&self.mining) == 1 {
// Stop all agents.
for agent := range self.agents {
@@ -217,36 +224,23 @@ func (self *worker) unregister(agent Agent) {
}
func (self *worker) update() {
- eventSub := self.mux.Subscribe(core.ChainHeadEvent{}, core.ChainSideEvent{}, core.TxPreEvent{})
- defer eventSub.Unsubscribe()
-
- eventCh := eventSub.Chan()
- for {
- select {
- case event, ok := <-eventCh:
- if !ok {
- // Event subscription closed, set the channel to nil to stop spinning
- eventCh = nil
- continue
- }
- // A real event arrived, process interesting content
- switch ev := event.Data.(type) {
- case core.ChainHeadEvent:
- self.commitNewWork()
- case core.ChainSideEvent:
- self.uncleMu.Lock()
- self.possibleUncles[ev.Block.Hash()] = ev.Block
- self.uncleMu.Unlock()
- case core.TxPreEvent:
- // Apply transaction to the pending state if we're not mining
- if atomic.LoadInt32(&self.mining) == 0 {
- self.currentMu.Lock()
- self.current.commitTransactions(self.mux, types.Transactions{ev.Tx}, self.gasPrice, self.chain)
- self.currentMu.Unlock()
- }
+ defer self.wg.Done()
+ for event := range self.events.Chan() {
+ // A real event arrived, process interesting content
+ switch ev := event.Data.(type) {
+ case core.ChainHeadEvent:
+ self.commitNewWork()
+ case core.ChainSideEvent:
+ self.uncleMu.Lock()
+ self.possibleUncles[ev.Block.Hash()] = ev.Block
+ self.uncleMu.Unlock()
+ case core.TxPreEvent:
+ // Apply transaction to the pending state if we're not mining
+ if atomic.LoadInt32(&self.mining) == 0 {
+ self.currentMu.Lock()
+ self.current.commitTransactions(self.mux, types.Transactions{ev.Tx}, self.gasPrice, self.chain)
+ self.currentMu.Unlock()
}
- case <-self.quit:
- return
}
}
}