diff --git a/consensus/clique/clique.go b/consensus/clique/clique.go index 6a8d8ce619..5963900c97 100644 --- a/consensus/clique/clique.go +++ b/consensus/clique/clique.go @@ -23,7 +23,6 @@ import ( "math/big" "math/rand" "sync" - "sync/atomic" "time" "github.com/ethereum/go-ethereum/accounts" @@ -195,9 +194,8 @@ func ecrecover(header *types.Header, sigcache *lru.ARCCache) (common.Address, er // Clique is the proof-of-authority consensus engine proposed to support the // Ethereum testnet following the Ropsten attacks. type Clique struct { - config *params.CliqueConfig // Consensus engine configuration parameters - db ethdb.Database // Database to store and retrieve snapshot checkpoints - running int32 // Indicator whether clique engine is running or not. + config *params.CliqueConfig // Consensus engine configuration parameters + db ethdb.Database // Database to store and retrieve snapshot checkpoints recents *lru.ARCCache // Snapshots for recent block to speed up reorgs signatures *lru.ARCCache // Signatures of recent blocks to speed up mining @@ -603,10 +601,6 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch if c.config.Period == 0 && len(block.Transactions()) == 0 { return nil, errWaitTransactions } - // Make sure clique engine is started. - if !c.IsRunning() { - return nil, consensus.ErrEngineNotStart - } // Don't hold the signer fields for the entire sealing procedure c.lock.RLock() signer, signFn := c.signer, c.signFn @@ -678,25 +672,6 @@ func CalcDifficulty(snap *Snapshot, signer common.Address) *big.Int { return new(big.Int).Set(diffNoTurn) } -// Start implements consensus.Engine, starting the clique consensus engine. -func (c *Clique) Start() { - 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() { - 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. -func (c *Clique) IsRunning() bool { - return atomic.LoadInt32(&c.running) > 0 -} - // Close implements consensus.Engine, returning internal error and close the clique. func (c *Clique) Close() error { return nil diff --git a/consensus/consensus.go b/consensus/consensus.go index 3562388d51..ae0fefb490 100644 --- a/consensus/consensus.go +++ b/consensus/consensus.go @@ -97,15 +97,6 @@ type Engine interface { // APIs returns the RPC APIs this consensus engine provides. APIs(chain ChainReader) []rpc.API - // Start starts the consensus engine. - Start() - - // Stop stops the consensus engine. - Stop() - - // IsRunning returns an indication whether consensus engine is running or not. - IsRunning() bool - // Close closes the consensus engine. Close() error } diff --git a/consensus/errors.go b/consensus/errors.go index c7d7073e19..a005c5f63d 100644 --- a/consensus/errors.go +++ b/consensus/errors.go @@ -34,7 +34,4 @@ var ( // ErrInvalidNumber is returned if a block's number doesn't equal it's parent's // plus one. ErrInvalidNumber = errors.New("invalid block number") - - // ErrEngineNotStart is returned if the consensus engine is not started. - ErrEngineNotStart = errors.New("consensus engine is not started") ) diff --git a/consensus/ethash/api.go b/consensus/ethash/api.go index eae6c169a8..b950fbd92c 100644 --- a/consensus/ethash/api.go +++ b/consensus/ethash/api.go @@ -51,12 +51,6 @@ func (api *API) GetWork() ([3]string, error) { err error ) - // Trigger ethash to start in remote mining mode(local/cpu mining is disabled) - // if ethash is not running. - if !api.ethash.IsRunning() { - api.ethash.StartMining(new(int)) - } - select { case api.ethash.fetchWorkCh <- &sealWork{errCh: errCh, resCh: workCh}: case <-api.ethash.exitCh: diff --git a/consensus/ethash/consensus.go b/consensus/ethash/consensus.go index bee4efe9e6..eb0f73d98b 100644 --- a/consensus/ethash/consensus.go +++ b/consensus/ethash/consensus.go @@ -22,7 +22,6 @@ import ( "fmt" "math/big" "runtime" - "sync/atomic" "time" mapset "github.com/deckarep/golang-set" @@ -32,7 +31,6 @@ import ( "github.com/ethereum/go-ethereum/consensus/misc" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/params" ) @@ -554,22 +552,3 @@ func accumulateRewards(config *params.ChainConfig, state *state.StateDB, header } state.AddBalance(header.Coinbase, reward) } - -// Start implements consensus.Engine, starting the ethash engine. -func (ethash *Ethash) Start() { - 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() { - 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. -func (ethash *Ethash) IsRunning() bool { - return atomic.LoadInt32(ðash.running) > 0 -} diff --git a/consensus/ethash/ethash.go b/consensus/ethash/ethash.go index caf272ab07..3fff266410 100644 --- a/consensus/ethash/ethash.go +++ b/consensus/ethash/ethash.go @@ -428,7 +428,6 @@ type Ethash struct { threads int // Number of threads to mine on if mining update chan struct{} // Notification channel to update mining parameters hashrate metrics.Meter // Meter tracking the average hashrate - running int32 // Indicator whether ethash engine is running or not. // Remote sealer related fields workCh chan *types.Block // Notification channel to push new work to remote sealer @@ -487,7 +486,6 @@ func NewTester() *Ethash { datasets: newlru("dataset", 1, newDataset), update: make(chan struct{}), hashrate: metrics.NewMeter(), - running: 1, // enable local mining by default workCh: make(chan *types.Block), resultCh: make(chan *types.Block), fetchWorkCh: make(chan *sealWork), @@ -680,20 +678,6 @@ func (ethash *Ethash) APIs(chain consensus.ChainReader) []rpc.API { } } -// StartMining starts the ethash engine with the given number of threads. -// If threads is nil the number of workers started is equal to the number of logical CPUs -// that are usable by this process. If threads is 0, than local/cpu mining will be disabled. -// If mining is already running, this method adjust the number of threads allowed to use. -func (ethash *Ethash) StartMining(threads *int) { - if threads == nil { - threads = new(int) - } else if *threads == 0 { - *threads = -1 // Disable local/cpu mining. - } - ethash.Start() - ethash.SetThreads(*threads) -} - // SeedHash is the seed to use for generating a verification cache and the mining // dataset. func SeedHash(block uint64) []byte { diff --git a/consensus/ethash/sealer.go b/consensus/ethash/sealer.go index 4c164b779b..539e783a6c 100644 --- a/consensus/ethash/sealer.go +++ b/consensus/ethash/sealer.go @@ -50,10 +50,6 @@ func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, stop if ethash.shared != nil { return ethash.shared.Seal(chain, block, stop) } - // Make sure ethash engine is started. - if !ethash.IsRunning() { - return nil, consensus.ErrEngineNotStart - } // Create a runner and the multiple search threads it directs abort := make(chan struct{}) diff --git a/core/tx_pool.go b/core/tx_pool.go index 0fb5ca52b2..7007f85ddf 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -465,9 +465,6 @@ func (pool *TxPool) SetGasPrice(price *big.Int) { pool.mu.Lock() defer pool.mu.Unlock() - if pool.gasPrice == price { - return - } pool.gasPrice = price for _, tx := range pool.priced.Cap(price, pool.locals) { pool.removeTx(tx.Hash(), false) diff --git a/eth/api.go b/eth/api.go index a6b4319872..c1fbcb6d40 100644 --- a/eth/api.go +++ b/eth/api.go @@ -112,14 +112,16 @@ func (api *PrivateMinerAPI) Start(threads *int) error { log.Info("Updated mining threads", "threads", *threads) th.SetThreads(*threads) } - // Propagate the initial price point to the transaction pool - api.e.lock.RLock() - price := api.e.gasPrice - api.e.lock.RUnlock() - api.e.txPool.SetGasPrice(price) - // Start the miner and return - return api.e.StartMining(true) + if !api.e.IsMining() { + // Propagate the initial price point to the transaction pool + api.e.lock.RLock() + price := api.e.gasPrice + api.e.lock.RUnlock() + api.e.txPool.SetGasPrice(price) + return api.e.StartMining(true) + } + return nil } // Stop the miner diff --git a/eth/backend.go b/eth/backend.go index a41adebcfd..32946a0ab3 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -167,12 +167,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { return nil, err } - // Specify etherbase explicitly. - var etherbase common.Address - if addr, err := eth.Etherbase(); err == nil { - etherbase = addr - } - eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine, etherbase) + eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine) eth.miner.SetExtra(makeExtraData(config.ExtraData)) eth.APIBackend = &EthAPIBackend{eth, nil} @@ -344,7 +339,7 @@ func (s *Ethereum) StartMining(local bool) error { log.Error("Cannot start mining without etherbase", "err", err) return fmt.Errorf("etherbase missing: %v", err) } - if clique, ok := s.engine.(*clique.Clique); ok && !clique.IsRunning() { + if clique, ok := s.engine.(*clique.Clique); ok { wallet, err := s.accountManager.Find(accounts.Account{Address: eb}) if wallet == nil || err != nil { log.Error("Etherbase account unavailable locally", "err", err) diff --git a/miner/agent.go b/miner/agent.go index 8d37921ef5..95d835bd78 100644 --- a/miner/agent.go +++ b/miner/agent.go @@ -62,9 +62,7 @@ func (self *CpuAgent) Stop() { if !atomic.CompareAndSwapInt32(&self.started, 1, 0) { return // agent already stopped } - // Close the pending routines. - close(self.stop) - + self.stop <- struct{}{} done: // Empty work channel for { @@ -105,7 +103,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 && err != consensus.ErrEngineNotStart { + if err != nil { log.Warn("Block sealing failed", "err", err) } self.returnCh <- nil diff --git a/miner/miner.go b/miner/miner.go index 949ce0f2b2..4c5717c8ad 100644 --- a/miner/miner.go +++ b/miner/miner.go @@ -54,13 +54,12 @@ type Miner struct { shouldStart int32 // should start indicates whether we should start after sync } -func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine consensus.Engine, coinbase common.Address) *Miner { +func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine consensus.Engine) *Miner { miner := &Miner{ eth: eth, mux: mux, engine: engine, - coinbase: coinbase, - worker: newWorker(config, engine, coinbase, eth, mux), + worker: newWorker(config, engine, eth, mux), canStart: 1, } miner.Register(NewCpuAgent(eth.BlockChain(), engine)) @@ -110,6 +109,7 @@ func (self *Miner) Start(coinbase common.Address) { return } self.worker.start() + self.worker.commitNewWork() } func (self *Miner) Stop() { @@ -126,7 +126,7 @@ func (self *Miner) Unregister(agent Agent) { } func (self *Miner) Mining() bool { - return self.engine.IsRunning() + return self.worker.isRunning() } func (self *Miner) HashRate() uint64 { diff --git a/miner/worker.go b/miner/worker.go index 1cd666f17c..f1194fa18a 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -126,10 +126,11 @@ type worker struct { unconfirmed *unconfirmedBlocks // set of locally mined blocks pending canonicalness confirmations // atomic status counters - atWork int32 // The number of in-flight consensus engine work. + atWork int32 // The number of in-flight consensus engine work. + running int32 // The indicator whether the consensus engine is running or not. } -func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase common.Address, eth Backend, mux *event.TypeMux) *worker { +func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux) *worker { worker := &worker{ config: config, engine: engine, @@ -143,7 +144,6 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com chain: eth.BlockChain(), proc: eth.BlockChain().Validator(), possibleUncles: make(map[common.Hash]*types.Block), - coinbase: coinbase, agents: make(map[Agent]struct{}), unconfirmed: newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth), } @@ -189,8 +189,7 @@ func (self *worker) pendingBlock() *types.Block { func (self *worker) start() { self.mu.Lock() defer self.mu.Unlock() - - self.engine.Start() + atomic.StoreInt32(&self.running, 1) for agent := range self.agents { agent.Start() } @@ -200,19 +199,25 @@ func (self *worker) stop() { self.mu.Lock() defer self.mu.Unlock() - self.engine.Stop() + atomic.StoreInt32(&self.running, 0) for agent := range self.agents { agent.Stop() } atomic.StoreInt32(&self.atWork, 0) } +func (self *worker) isRunning() bool { + return atomic.LoadInt32(&self.running) == 1 +} + func (self *worker) register(agent Agent) { self.mu.Lock() defer self.mu.Unlock() self.agents[agent] = struct{}{} agent.SetReturnCh(self.recv) - agent.Start() + if self.isRunning() { + agent.Start() + } } func (self *worker) unregister(agent Agent) { @@ -227,11 +232,6 @@ func (self *worker) update() { defer self.chainHeadSub.Unsubscribe() defer self.chainSideSub.Unsubscribe() - ticker := time.NewTicker(500 * time.Millisecond) - defer ticker.Stop() - - var started bool // Indication whether consensus engine is started - for { // A real event arrived, process interesting content select { @@ -252,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 !self.engine.IsRunning() { + if !self.isRunning() { self.currentMu.Lock() txs := make(map[common.Address]types.Transactions) for _, tx := range ev.Txs { @@ -270,17 +270,6 @@ func (self *worker) update() { } } - // Commit new work when consensus engine is started. - case <-ticker.C: - if self.engine.IsRunning() { - if !started { - self.commitNewWork() - started = true - } - } else { - started = false - } - // System stopped case <-self.txsSub.Err(): return @@ -340,11 +329,6 @@ 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 { @@ -416,7 +400,11 @@ func (self *worker) commitNewWork() { Time: big.NewInt(tstamp), } // Only set the coinbase if our consensus engine is running (avoid spurious block rewards) - if self.engine.IsRunning() { + if self.isRunning() { + if self.coinbase == (common.Address{}) { + log.Error("Refusing to mine without etherbase") + return + } header.Coinbase = self.coinbase } if err := self.engine.Prepare(self.chain, header); err != nil { @@ -479,10 +467,10 @@ 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 self.engine.IsRunning() { + if self.isRunning() { log.Info("Commit new empty mining work", "number", work.Block.Number(), "uncles", len(uncles)) + self.push(work) } - self.push(work) } // Fill the block with all available pending transactions. @@ -500,12 +488,11 @@ func (self *worker) commitNewWork() { return } // We only care about logging if we're actually mining. - if self.engine.IsRunning() { + if self.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) + self.push(work) } - // Push full work to sealer, which will replace the empty work sent before automatically. - self.push(work) self.updateSnapshot() }