consensus, miner: fix some logics

consensus/clique/clique.go

@@ -680,14 +680,16 @@ func CalcDifficulty(snap *Snapshot, signer common.Address) *big.Int {
 
 // Start implements consensus.Engine, starting the clique consensus engine.
 func (c *Clique) Start() {
-	log.Info("Start clique consensus engine")
-	atomic.StoreInt32(&c.running, 1)
+	if atomic.CompareAndSwapInt32(&c.running, 0, 1) {
+		log.Info("Start clique consensus engine")
+	}
 }
 
 // Stop implements consensus.Engine, stopping the clique consensus engine.
 func (c *Clique) Stop() {
-	log.Info("Stop clique consensus engine")
-	atomic.StoreInt32(&c.running, 0)
+	if atomic.CompareAndSwapInt32(&c.running, 1, 0) {
+		log.Info("Stop clique consensus engine")
+	}
 }
 
 // IsRunning implements consensus.Engine, returning an indication if the clique engine is currently mining.

consensus/ethash/api.go

@@ -24,11 +24,14 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
-var errEthashStopped = errors.New("ethash stopped")
+var (
+	errEthashStopped   = errors.New("ethash stopped")
+	errAPINotSupported = errors.New("the current ethash running mode does not support this API")
+)
 
 // API exposes ethash related methods for the RPC interface.
 type API struct {
-	ethash *Ethash
+	ethash *Ethash // Make sure the mode of ethash is normal.
 }
 
 // GetWork returns a work package for external miner.
@@ -38,6 +41,10 @@ type API struct {
 //   result[1] - 32 bytes hex encoded seed hash used for DAG
 //   result[2] - 32 bytes hex encoded boundary condition ("target"), 2^256/difficulty
 func (api *API) GetWork() ([3]string, error) {
+	if api.ethash.config.PowMode != ModeNormal && api.ethash.config.PowMode != ModeTest {
+		return [3]string{}, errAPINotSupported
+	}
+
 	var (
 		workCh = make(chan [3]string, 1)
 		errCh  = make(chan error, 1)
@@ -66,6 +73,10 @@ func (api *API) GetWork() ([3]string, error) {
 // It returns an indication if the work was accepted.
 // Note either an invalid solution, a stale work a non-existent work will return false.
 func (api *API) SubmitWork(nonce types.BlockNonce, hash, digest common.Hash) bool {
+	if api.ethash.config.PowMode != ModeNormal && api.ethash.config.PowMode != ModeTest {
+		return false
+	}
+
 	var errCh = make(chan error, 1)
 
 	select {
@@ -90,6 +101,10 @@ func (api *API) SubmitWork(nonce types.BlockNonce, hash, digest common.Hash) boo
 // It accepts the miner hash rate and an identifier which must be unique
 // between nodes.
 func (api *API) SubmitHashRate(rate hexutil.Uint64, id common.Hash) bool {
+	if api.ethash.config.PowMode != ModeNormal && api.ethash.config.PowMode != ModeTest {
+		return false
+	}
+
 	var doneCh = make(chan struct{}, 1)
 
 	select {

consensus/ethash/consensus.go

@@ -557,14 +557,16 @@ func accumulateRewards(config *params.ChainConfig, state *state.StateDB, header
 
 // Start implements consensus.Engine, starting the ethash engine.
 func (ethash *Ethash) Start() {
-	log.Info("Start ethash consensus engine")
-	atomic.StoreInt32(&ethash.running, 1)
+	if atomic.CompareAndSwapInt32(&ethash.running, 0, 1) {
+		log.Info("Start ethash consensus engine")
+	}
 }
 
 // Stop implements consensus.Engine, stopping the ethash engine.
 func (ethash *Ethash) Stop() {
-	log.Info("Stop ethash consensus engine")
-	atomic.StoreInt32(&ethash.running, 0)
+	if atomic.CompareAndSwapInt32(&ethash.running, 1, 0) {
+		log.Info("Stop ethash consensus engine")
+	}
 }
 
 // IsRunning implements consensus.Engine, returning an indication if the ethash engine is currently mining.

consensus/ethash/ethash.go

@@ -508,7 +508,6 @@ func NewFaker() *Ethash {
 		config: Config{
 			PowMode: ModeFake,
 		},
-		exitCh: make(chan chan error),
 	}
 }
 
@@ -521,7 +520,6 @@ func NewFakeFailer(fail uint64) *Ethash {
 			PowMode: ModeFake,
 		},
 		fakeFail: fail,
-		exitCh:   make(chan chan error),
 	}
 }
 
@@ -534,7 +532,6 @@ func NewFakeDelayer(delay time.Duration) *Ethash {
 			PowMode: ModeFake,
 		},
 		fakeDelay: delay,
-		exitCh:    make(chan chan error),
 	}
 }
 
@@ -545,30 +542,27 @@ func NewFullFaker() *Ethash {
 		config: Config{
 			PowMode: ModeFullFake,
 		},
-		exitCh: make(chan chan error),
 	}
 }
 
 // NewShared creates a full sized ethash PoW shared between all requesters running
 // in the same process.
 func NewShared() *Ethash {
-	return &Ethash{
-		shared: sharedEthash,
-		exitCh: make(chan chan error),
-	}
+	return &Ethash{shared: sharedEthash}
 }
 
 // Close closes the exit channel to notify all backend threads exiting.
 func (ethash *Ethash) Close() error {
 	var err error
 	ethash.closeOnce.Do(func() {
-		var errCh = make(chan error)
-		select {
-		case ethash.exitCh <- errCh:
-			err = <-errCh
-			close(ethash.exitCh)
-		default:
+		// Short circuit if the exit channel is not allocated.
+		if ethash.exitCh == nil {
+			return
 		}
+		errCh := make(chan error)
+		ethash.exitCh <- errCh
+		err = <-errCh
+		close(ethash.exitCh)
 	})
 	return err
 }
@@ -648,6 +642,10 @@ func (ethash *Ethash) SetThreads(threads int) {
 // Note the returned hashrate includes local hashrate, but also includes the total
 // hashrate of all remote miner.
 func (ethash *Ethash) Hashrate() float64 {
+	// Short circuit if we are run the ethash in normal/test mode.
+	if ethash.config.PowMode != ModeNormal && ethash.config.PowMode != ModeTest {
+		return ethash.hashrate.Rate1()
+	}
 	var resCh = make(chan uint64, 1)
 
 	select {

miner/agent.go

@@ -63,10 +63,7 @@ func (self *CpuAgent) Stop() {
 		return // agent already stopped
 	}
 	// Close the pending routines.
-	select {
-	case self.stop <- struct{}{}:
-	default:
-	}
+	close(self.stop)
 
 done:
 	// Empty work channel

miner/miner.go

@@ -109,21 +109,11 @@ func (self *Miner) Start(coinbase common.Address) {
 		log.Info("Network syncing, will start miner afterwards")
 		return
 	}
-	if !self.engine.IsRunning() {
-		self.engine.Start()
-	}
-	if !self.worker.isRunning() {
-		self.worker.start()
-	}
+	self.worker.start()
 }
 
 func (self *Miner) Stop() {
-	if self.engine.IsRunning() {
-		self.engine.Stop()
-	}
-	if self.worker.isRunning() {
-		self.worker.stop()
-	}
+	self.worker.stop()
 	atomic.StoreInt32(&self.shouldStart, 0)
 }
 
@@ -139,11 +129,11 @@ func (self *Miner) Mining() bool {
 	return self.engine.IsRunning()
 }
 
-func (self *Miner) HashRate() (tot uint64) {
+func (self *Miner) HashRate() uint64 {
 	if pow, ok := self.engine.(consensus.PoW); ok {
-		tot += uint64(pow.Hashrate())
+		return uint64(pow.Hashrate())
 	}
-	return
+	return 0
 }
 
 func (self *Miner) SetExtra(extra []byte) error {

miner/worker.go

@@ -101,7 +101,6 @@ type worker struct {
 	chainHeadSub event.Subscription
 	chainSideCh  chan core.ChainSideEvent
 	chainSideSub event.Subscription
-	wg           sync.WaitGroup
 
 	agents map[Agent]struct{}
 	recv   chan *Result
@@ -127,8 +126,7 @@ type worker struct {
 	unconfirmed *unconfirmedBlocks // set of locally mined blocks pending canonicalness confirmations
 
 	// atomic status counters
-	running int32
-	atWork  int32
+	atWork int32 // The number of in-flight consensus engine work.
 }
 
 func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase common.Address, eth Backend, mux *event.TypeMux) *worker {
@@ -175,60 +173,38 @@ func (self *worker) setExtra(extra []byte) {
 }
 
 func (self *worker) pending() (*types.Block, *state.StateDB) {
-	if atomic.LoadInt32(&self.running) == 0 {
-		// return a snapshot to avoid contention on currentMu mutex
-		self.snapshotMu.RLock()
-		defer self.snapshotMu.RUnlock()
-		return self.snapshotBlock, self.snapshotState.Copy()
-	}
-
-	self.currentMu.Lock()
-	defer self.currentMu.Unlock()
-	return self.current.Block, self.current.state.Copy()
+	// return a snapshot to avoid contention on currentMu mutex
+	self.snapshotMu.RLock()
+	defer self.snapshotMu.RUnlock()
+	return self.snapshotBlock, self.snapshotState.Copy()
 }
 
 func (self *worker) pendingBlock() *types.Block {
-	if atomic.LoadInt32(&self.running) == 0 {
-		// return a snapshot to avoid contention on currentMu mutex
-		self.snapshotMu.RLock()
-		defer self.snapshotMu.RUnlock()
-		return self.snapshotBlock
-	}
-
-	self.currentMu.Lock()
-	defer self.currentMu.Unlock()
-	return self.current.Block
+	// return a snapshot to avoid contention on currentMu mutex
+	self.snapshotMu.RLock()
+	defer self.snapshotMu.RUnlock()
+	return self.snapshotBlock
 }
 
 func (self *worker) start() {
 	self.mu.Lock()
 	defer self.mu.Unlock()
 
-	atomic.StoreInt32(&self.running, 1)
-
-	// spin up agents
+	self.engine.Start()
 	for agent := range self.agents {
 		agent.Start()
 	}
 }
 
 func (self *worker) stop() {
-	self.wg.Wait()
-
 	self.mu.Lock()
 	defer self.mu.Unlock()
-	if atomic.LoadInt32(&self.running) == 1 {
-		for agent := range self.agents {
-			agent.Stop()
-		}
-	}
-	atomic.StoreInt32(&self.running, 0)
-	atomic.StoreInt32(&self.atWork, 0)
-}
 
-// isRunning returns an indicator whether worker is currently running or not.
-func (self *worker) isRunning() bool {
-	return atomic.LoadInt32(&self.running) > 0
+	self.engine.Stop()
+	for agent := range self.agents {
+		agent.Stop()
+	}
+	atomic.StoreInt32(&self.atWork, 0)
 }
 
 func (self *worker) register(agent Agent) {
@@ -276,7 +252,7 @@ func (self *worker) update() {
 			// Note all transactions received may not be continuous with transactions
 			// already included in the current mining block. These transactions will
 			// be automatically eliminated.
-			if atomic.LoadInt32(&self.running) == 0 {
+			if !self.engine.IsRunning() {
 				self.currentMu.Lock()
 				txs := make(map[common.Address]types.Transactions)
 				for _, tx := range ev.Txs {
@@ -364,6 +340,11 @@ func (self *worker) wait() {
 
 // push sends a new work task to currently live miner agents.
 func (self *worker) push(work *Work) {
+	// Never send task to consensus engine if the etherbase is not specified.
+	if self.engine.IsRunning() && work.header.Coinbase == (common.Address{}) {
+		log.Info("Please explicitly specifies the etherbase")
+		return
+	}
 	for agent := range self.agents {
 		atomic.AddInt32(&self.atWork, 1)
 		if ch := agent.Work(); ch != nil {
@@ -547,10 +528,19 @@ func (self *worker) updateSnapshot() {
 	self.snapshotMu.Lock()
 	defer self.snapshotMu.Unlock()
 
+	var uncles []*types.Header
+	self.current.uncles.Each(func(item interface{}) bool {
+		if header, ok := item.(*types.Header); ok {
+			uncles = append(uncles, header)
+			return true
+		}
+		return false
+	})
+
 	self.snapshotBlock = types.NewBlock(
 		self.current.header,
 		self.current.txs,
-		nil,
+		uncles,
 		self.current.receipts,
 	)
 	self.snapshotState = self.current.state.Copy()