diff --git a/consensus/clique/clique.go b/consensus/clique/clique.go index 3b89e71560..6a8d8ce619 100644 --- a/consensus/clique/clique.go +++ b/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. diff --git a/consensus/ethash/api.go b/consensus/ethash/api.go index 6a1e015506..eae6c169a8 100644 --- a/consensus/ethash/api.go +++ b/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 { diff --git a/consensus/ethash/consensus.go b/consensus/ethash/consensus.go index 2cb72676e9..bee4efe9e6 100644 --- a/consensus/ethash/consensus.go +++ b/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(ðash.running, 1) + if atomic.CompareAndSwapInt32(ðash.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(ðash.running, 0) + if atomic.CompareAndSwapInt32(ðash.running, 1, 0) { + log.Info("Stop ethash consensus engine") + } } // IsRunning implements consensus.Engine, returning an indication if the ethash engine is currently mining. diff --git a/consensus/ethash/ethash.go b/consensus/ethash/ethash.go index 7e9e661f78..caf272ab07 100644 --- a/consensus/ethash/ethash.go +++ b/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 { diff --git a/miner/agent.go b/miner/agent.go index beaf30556a..8d37921ef5 100644 --- a/miner/agent.go +++ b/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 diff --git a/miner/miner.go b/miner/miner.go index e7add025de..949ce0f2b2 100644 --- a/miner/miner.go +++ b/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 { diff --git a/miner/worker.go b/miner/worker.go index ba39cb83dd..1cd666f17c 100644 --- a/miner/worker.go +++ b/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()