miner: expose consensus engine running status to miner

miner/agent.go

@@ -18,10 +18,10 @@ package miner
 
 import (
 	"sync"
-	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/consensus"
 	"github.com/ethereum/go-ethereum/log"
+	"sync/atomic"
 )
 
 type CpuAgent struct {
@@ -35,27 +35,39 @@ type CpuAgent struct {
 	chain  consensus.ChainReader
 	engine consensus.Engine
 
-	isMining int32 // isMining indicates whether the agent is currently mining
+	started int32 // started indicates whether the agent is currently started
 }
 
 func NewCpuAgent(chain consensus.ChainReader, engine consensus.Engine) *CpuAgent {
-	miner := &CpuAgent{
+	agent := &CpuAgent{
 		chain:  chain,
 		engine: engine,
 		stop:   make(chan struct{}, 1),
 		workCh: make(chan *Work, 1),
 	}
-	return miner
+	return agent
 }
 
 func (self *CpuAgent) Work() chan<- *Work            { return self.workCh }
 func (self *CpuAgent) SetReturnCh(ch chan<- *Result) { self.returnCh = ch }
 
+func (self *CpuAgent) Start() {
+	if !atomic.CompareAndSwapInt32(&self.started, 0, 1) {
+		return // agent already started
+	}
+	go self.update()
+}
+
 func (self *CpuAgent) Stop() {
-	if !atomic.CompareAndSwapInt32(&self.isMining, 1, 0) {
+	if !atomic.CompareAndSwapInt32(&self.started, 1, 0) {
 		return // agent already stopped
 	}
-	self.stop <- struct{}{}
+	// Close the pending routines.
+	select {
+	case self.stop <- struct{}{}:
+	default:
+	}
+
 done:
 	// Empty work channel
 	for {
@@ -67,13 +79,6 @@ done:
 	}
 }
 
-func (self *CpuAgent) Start() {
-	if !atomic.CompareAndSwapInt32(&self.isMining, 0, 1) {
-		return // agent already started
-	}
-	go self.update()
-}
-
 func (self *CpuAgent) update() {
 out:
 	for {
@@ -103,7 +108,7 @@ func (self *CpuAgent) mine(work *Work, stop <-chan struct{}) {
 		log.Info("Successfully sealed new block", "number", result.Number(), "hash", result.Hash())
 		self.returnCh <- &Result{work, result}
 	} else {
-		if err != nil {
+		if err != nil && err != consensus.ErrEngineNotStart {
 			log.Warn("Block sealing failed", "err", err)
 		}
 		self.returnCh <- nil

miner/miner.go

@@ -45,11 +45,8 @@ type Backend interface {
 // Miner creates blocks and searches for proof-of-work values.
 type Miner struct {
 	mux      *event.TypeMux
-
 	worker   *worker
-
 	coinbase common.Address
-	mining   int32
 	eth      Backend
 	engine   consensus.Engine
 
@@ -111,23 +108,19 @@ func (self *Miner) Start(coinbase common.Address) {
 		log.Info("Network syncing, will start miner afterwards")
 		return
 	}
-	atomic.StoreInt32(&self.mining, 1)
-
 	log.Info("Starting mining operation")
+	self.engine.Start()
 	self.worker.start()
 	self.worker.commitNewWork()
 }
 
 func (self *Miner) Stop() {
+	self.engine.Stop()
 	self.worker.stop()
-	atomic.StoreInt32(&self.mining, 0)
 	atomic.StoreInt32(&self.shouldStart, 0)
 }
 
 func (self *Miner) Register(agent Agent) {
-	if self.Mining() {
-		agent.Start()
-	}
 	self.worker.register(agent)
 }
 
@@ -136,7 +129,7 @@ func (self *Miner) Unregister(agent Agent) {
 }
 
 func (self *Miner) Mining() bool {
-	return atomic.LoadInt32(&self.mining) > 0
+	return self.engine.IsRunning()
 }
 
 func (self *Miner) HashRate() (tot int64) {

miner/worker.go

@@ -55,8 +55,8 @@ const (
 type Agent interface {
 	Work() chan<- *Work
 	SetReturnCh(chan<- *Result)
-	Stop()
 	Start()
+	Stop()
 }
 
 // Work is the workers current environment and holds
@@ -231,6 +231,7 @@ func (self *worker) register(agent Agent) {
 	defer self.mu.Unlock()
 	self.agents[agent] = struct{}{}
 	agent.SetReturnCh(self.recv)
+	agent.Start()
 }
 
 func (self *worker) unregister(agent Agent) {
@@ -342,9 +343,6 @@ func (self *worker) wait() {
 
 // push sends a new work task to currently live miner agents.
 func (self *worker) push(work *Work) {
-	if atomic.LoadInt32(&self.mining) != 1 {
-		return
-	}
 	for agent := range self.agents {
 		atomic.AddInt32(&self.atWork, 1)
 		if ch := agent.Work(); ch != nil {
@@ -415,8 +413,8 @@ func (self *worker) commitNewWork() {
 		Extra:      self.extra,
 		Time:       big.NewInt(tstamp),
 	}
-	// Only set the coinbase if we are mining (avoid spurious block rewards)
-	if atomic.LoadInt32(&self.mining) == 1 {
+	// Only set the coinbase if our consensus engine is running (avoid spurious block rewards)
+	if self.engine.IsRunning() {
 		header.Coinbase = self.coinbase
 	}
 	if err := self.engine.Prepare(self.chain, header); err != nil {
@@ -479,7 +477,7 @@ func (self *worker) commitNewWork() {
 		// Push empty work in advance without applying pending transaction.
 		// The reason is transactions execution can cost a lot and sealer need to
 		// take advantage of this part time.
-		if atomic.LoadInt32(&self.mining) == 1 {
+		if self.engine.IsRunning() {
 			log.Info("Commit new empty mining work", "number", work.Block.Number(), "uncles", len(uncles))
 		}
 		self.push(work)
@@ -500,7 +498,7 @@ func (self *worker) commitNewWork() {
 		return
 	}
 	// We only care about logging if we're actually mining.
-	if atomic.LoadInt32(&self.mining) == 1 {
+	if self.engine.IsRunning() {
 		log.Info("Commit new full mining work", "number", work.Block.Number(), "txs", work.tcount, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
 		self.unconfirmed.Shift(work.Block.NumberU64() - 1)
 	}