diff --git a/miner/agent.go b/miner/agent.go index e3cebbd2e2..d095245c1c 100644 --- a/miner/agent.go +++ b/miner/agent.go @@ -49,70 +49,70 @@ func NewCpuAgent(chain consensus.ChainReader, engine consensus.Engine) *CpuAgent return miner } -func (self *CpuAgent) Work() chan<- *Work { return self.workCh } -func (self *CpuAgent) SetReturnCh(ch chan<- *Result) { self.returnCh = ch } +func (a *CpuAgent) Work() chan<- *Work { return a.workCh } +func (a *CpuAgent) SetReturnCh(ch chan<- *Result) { a.returnCh = ch } -func (self *CpuAgent) Stop() { - if !atomic.CompareAndSwapInt32(&self.isMining, 1, 0) { +func (a *CpuAgent) Stop() { + if !atomic.CompareAndSwapInt32(&a.isMining, 1, 0) { return // agent already stopped } - self.stop <- struct{}{} + a.stop <- struct{}{} done: // Empty work channel for { select { - case <-self.workCh: + case <-a.workCh: default: break done } } } -func (self *CpuAgent) Start() { - if !atomic.CompareAndSwapInt32(&self.isMining, 0, 1) { +func (a *CpuAgent) Start() { + if !atomic.CompareAndSwapInt32(&a.isMining, 0, 1) { return // agent already started } - go self.update() + go a.update() } -func (self *CpuAgent) update() { +func (a *CpuAgent) update() { out: for { select { - case work := <-self.workCh: - self.mu.Lock() - if self.quitCurrentOp != nil { - close(self.quitCurrentOp) + case work := <-a.workCh: + a.mu.Lock() + if a.quitCurrentOp != nil { + close(a.quitCurrentOp) } - self.quitCurrentOp = make(chan struct{}) - go self.mine(work, self.quitCurrentOp) - self.mu.Unlock() - case <-self.stop: - self.mu.Lock() - if self.quitCurrentOp != nil { - close(self.quitCurrentOp) - self.quitCurrentOp = nil + a.quitCurrentOp = make(chan struct{}) + go a.mine(work, a.quitCurrentOp) + a.mu.Unlock() + case <-a.stop: + a.mu.Lock() + if a.quitCurrentOp != nil { + close(a.quitCurrentOp) + a.quitCurrentOp = nil } - self.mu.Unlock() + a.mu.Unlock() break out } } } -func (self *CpuAgent) mine(work *Work, stop <-chan struct{}) { - if result, err := self.engine.Seal(self.chain, work.Block, stop); result != nil { +func (a *CpuAgent) mine(work *Work, stop <-chan struct{}) { + if result, err := a.engine.Seal(a.chain, work.Block, stop); result != nil { log.Info("Successfully sealed new block", "number", result.Number(), "hash", result.Hash()) - self.returnCh <- &Result{work, result} + a.returnCh <- &Result{work, result} } else { if err != nil { log.Warn("Block sealing failed", "err", err) } - self.returnCh <- nil + a.returnCh <- nil } } -func (self *CpuAgent) GetHashRate() int64 { - if pow, ok := self.engine.(consensus.PoW); ok { +func (a *CpuAgent) GetHashRate() int64 { + if pow, ok := a.engine.(consensus.PoW); ok { return int64(pow.Hashrate()) } return 0 diff --git a/miner/miner.go b/miner/miner.go index d9256e9787..167847b69a 100644 --- a/miner/miner.go +++ b/miner/miner.go @@ -75,25 +75,25 @@ func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine con // It's entered once and as soon as `Done` or `Failed` has been broadcasted the events are unregistered and // the loop is exited. This to prevent a major security vuln where external parties can DOS you with blocks // and halt your mining operation for as long as the DOS continues. -func (self *Miner) update() { - events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{}) +func (m *Miner) update() { + events := m.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{}) out: for ev := range events.Chan() { switch ev.Data.(type) { case downloader.StartEvent: - atomic.StoreInt32(&self.canStart, 0) - if self.Mining() { - self.Stop() - atomic.StoreInt32(&self.shouldStart, 1) + atomic.StoreInt32(&m.canStart, 0) + if m.Mining() { + m.Stop() + atomic.StoreInt32(&m.shouldStart, 1) log.Info("Mining aborted due to sync") } case downloader.DoneEvent, downloader.FailedEvent: - shouldStart := atomic.LoadInt32(&self.shouldStart) == 1 + shouldStart := atomic.LoadInt32(&m.shouldStart) == 1 - atomic.StoreInt32(&self.canStart, 1) - atomic.StoreInt32(&self.shouldStart, 0) + atomic.StoreInt32(&m.canStart, 1) + atomic.StoreInt32(&m.shouldStart, 0) if shouldStart { - self.Start(self.coinbase) + m.Start(m.coinbase) } // unsubscribe. we're only interested in this event once events.Unsubscribe() @@ -103,50 +103,50 @@ out: } } -func (self *Miner) Start(coinbase common.Address) { - atomic.StoreInt32(&self.shouldStart, 1) - self.SetEtherbase(coinbase) +func (m *Miner) Start(coinbase common.Address) { + atomic.StoreInt32(&m.shouldStart, 1) + m.SetEtherbase(coinbase) - if atomic.LoadInt32(&self.canStart) == 0 { + if atomic.LoadInt32(&m.canStart) == 0 { log.Info("Network syncing, will start miner afterwards") return } - atomic.StoreInt32(&self.mining, 1) + atomic.StoreInt32(&m.mining, 1) log.Info("Starting mining operation") - self.worker.start() - self.worker.commitNewWork() + m.worker.start() + m.worker.commitNewWork() } -func (self *Miner) Stop() { - self.worker.stop() - atomic.StoreInt32(&self.mining, 0) - atomic.StoreInt32(&self.shouldStart, 0) +func (m *Miner) Stop() { + m.worker.stop() + atomic.StoreInt32(&m.mining, 0) + atomic.StoreInt32(&m.shouldStart, 0) } -func (self *Miner) Register(agent Agent) { - if self.Mining() { +func (m *Miner) Register(agent Agent) { + if m.Mining() { agent.Start() } - self.worker.register(agent) + m.worker.register(agent) } -func (self *Miner) Unregister(agent Agent) { - self.worker.unregister(agent) +func (m *Miner) Unregister(agent Agent) { + m.worker.unregister(agent) } -func (self *Miner) Mining() bool { - return atomic.LoadInt32(&self.mining) > 0 +func (m *Miner) Mining() bool { + return atomic.LoadInt32(&m.mining) > 0 } -func (self *Miner) HashRate() (tot int64) { - if pow, ok := self.engine.(consensus.PoW); ok { +func (m *Miner) HashRate() (tot int64) { + if pow, ok := m.engine.(consensus.PoW); ok { tot += int64(pow.Hashrate()) } // do we care this might race? is it worth we're rewriting some // aspects of the worker/locking up agents so we can get an accurate // hashrate? - for agent := range self.worker.agents { + for agent := range m.worker.agents { if _, ok := agent.(*CpuAgent); !ok { tot += agent.GetHashRate() } @@ -154,17 +154,17 @@ func (self *Miner) HashRate() (tot int64) { return } -func (self *Miner) SetExtra(extra []byte) error { +func (m *Miner) SetExtra(extra []byte) error { if uint64(len(extra)) > params.MaximumExtraDataSize { return fmt.Errorf("Extra exceeds max length. %d > %v", len(extra), params.MaximumExtraDataSize) } - self.worker.setExtra(extra) + m.worker.setExtra(extra) return nil } // Pending returns the currently pending block and associated state. -func (self *Miner) Pending() (*types.Block, *state.StateDB) { - return self.worker.pending() +func (m *Miner) Pending() (*types.Block, *state.StateDB) { + return m.worker.pending() } // PendingBlock returns the currently pending block. @@ -172,11 +172,11 @@ func (self *Miner) Pending() (*types.Block, *state.StateDB) { // Note, to access both the pending block and the pending state // simultaneously, please use Pending(), as the pending state can // change between multiple method calls -func (self *Miner) PendingBlock() *types.Block { - return self.worker.pendingBlock() +func (m *Miner) PendingBlock() *types.Block { + return m.worker.pendingBlock() } -func (self *Miner) SetEtherbase(addr common.Address) { - self.coinbase = addr - self.worker.setEtherbase(addr) +func (m *Miner) SetEtherbase(addr common.Address) { + m.coinbase = addr + m.worker.setEtherbase(addr) } diff --git a/miner/worker.go b/miner/worker.go index 48b0b27652..15c029c724 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -162,137 +162,137 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com return worker } -func (self *worker) setEtherbase(addr common.Address) { - self.mu.Lock() - defer self.mu.Unlock() - self.coinbase = addr +func (w *worker) setEtherbase(addr common.Address) { + w.mu.Lock() + defer w.mu.Unlock() + w.coinbase = addr } -func (self *worker) setExtra(extra []byte) { - self.mu.Lock() - defer self.mu.Unlock() - self.extra = extra +func (w *worker) setExtra(extra []byte) { + w.mu.Lock() + defer w.mu.Unlock() + w.extra = extra } -func (self *worker) pending() (*types.Block, *state.StateDB) { - if atomic.LoadInt32(&self.mining) == 0 { +func (w *worker) pending() (*types.Block, *state.StateDB) { + if atomic.LoadInt32(&w.mining) == 0 { // return a snapshot to avoid contention on currentMu mutex - self.snapshotMu.RLock() - defer self.snapshotMu.RUnlock() - return self.snapshotBlock, self.snapshotState.Copy() + w.snapshotMu.RLock() + defer w.snapshotMu.RUnlock() + return w.snapshotBlock, w.snapshotState.Copy() } - self.currentMu.Lock() - defer self.currentMu.Unlock() - return self.current.Block, self.current.state.Copy() + w.currentMu.Lock() + defer w.currentMu.Unlock() + return w.current.Block, w.current.state.Copy() } -func (self *worker) pendingBlock() *types.Block { - if atomic.LoadInt32(&self.mining) == 0 { +func (w *worker) pendingBlock() *types.Block { + if atomic.LoadInt32(&w.mining) == 0 { // return a snapshot to avoid contention on currentMu mutex - self.snapshotMu.RLock() - defer self.snapshotMu.RUnlock() - return self.snapshotBlock + w.snapshotMu.RLock() + defer w.snapshotMu.RUnlock() + return w.snapshotBlock } - self.currentMu.Lock() - defer self.currentMu.Unlock() - return self.current.Block + w.currentMu.Lock() + defer w.currentMu.Unlock() + return w.current.Block } -func (self *worker) start() { - self.mu.Lock() - defer self.mu.Unlock() +func (w *worker) start() { + w.mu.Lock() + defer w.mu.Unlock() - atomic.StoreInt32(&self.mining, 1) + atomic.StoreInt32(&w.mining, 1) // spin up agents - for agent := range self.agents { + for agent := range w.agents { agent.Start() } } -func (self *worker) stop() { - self.wg.Wait() +func (w *worker) stop() { + w.wg.Wait() - self.mu.Lock() - defer self.mu.Unlock() - if atomic.LoadInt32(&self.mining) == 1 { - for agent := range self.agents { + w.mu.Lock() + defer w.mu.Unlock() + if atomic.LoadInt32(&w.mining) == 1 { + for agent := range w.agents { agent.Stop() } } - atomic.StoreInt32(&self.mining, 0) - atomic.StoreInt32(&self.atWork, 0) + atomic.StoreInt32(&w.mining, 0) + atomic.StoreInt32(&w.atWork, 0) } -func (self *worker) register(agent Agent) { - self.mu.Lock() - defer self.mu.Unlock() - self.agents[agent] = struct{}{} - agent.SetReturnCh(self.recv) +func (w *worker) register(agent Agent) { + w.mu.Lock() + defer w.mu.Unlock() + w.agents[agent] = struct{}{} + agent.SetReturnCh(w.recv) } -func (self *worker) unregister(agent Agent) { - self.mu.Lock() - defer self.mu.Unlock() - delete(self.agents, agent) +func (w *worker) unregister(agent Agent) { + w.mu.Lock() + defer w.mu.Unlock() + delete(w.agents, agent) agent.Stop() } -func (self *worker) update() { - defer self.txSub.Unsubscribe() - defer self.chainHeadSub.Unsubscribe() - defer self.chainSideSub.Unsubscribe() +func (w *worker) update() { + defer w.txSub.Unsubscribe() + defer w.chainHeadSub.Unsubscribe() + defer w.chainSideSub.Unsubscribe() for { // A real event arrived, process interesting content select { // Handle ChainHeadEvent - case <-self.chainHeadCh: - self.commitNewWork() + case <-w.chainHeadCh: + w.commitNewWork() // Handle ChainSideEvent - case ev := <-self.chainSideCh: - self.uncleMu.Lock() - self.possibleUncles[ev.Block.Hash()] = ev.Block - self.uncleMu.Unlock() + case ev := <-w.chainSideCh: + w.uncleMu.Lock() + w.possibleUncles[ev.Block.Hash()] = ev.Block + w.uncleMu.Unlock() // Handle TxPreEvent - case ev := <-self.txCh: + case ev := <-w.txCh: // Apply transaction to the pending state if we're not mining - if atomic.LoadInt32(&self.mining) == 0 { - self.currentMu.Lock() - acc, _ := types.Sender(self.current.signer, ev.Tx) + if atomic.LoadInt32(&w.mining) == 0 { + w.currentMu.Lock() + acc, _ := types.Sender(w.current.signer, ev.Tx) txs := map[common.Address]types.Transactions{acc: {ev.Tx}} - txset := types.NewTransactionsByPriceAndNonce(self.current.signer, txs) + txset := types.NewTransactionsByPriceAndNonce(w.current.signer, txs) - self.current.commitTransactions(self.mux, txset, self.chain, self.coinbase) - self.updateSnapshot() - self.currentMu.Unlock() + w.current.commitTransactions(w.mux, txset, w.chain, w.coinbase) + w.updateSnapshot() + w.currentMu.Unlock() } else { // If we're mining, but nothing is being processed, wake on new transactions - if self.config.Clique != nil && self.config.Clique.Period == 0 { - self.commitNewWork() + if w.config.Clique != nil && w.config.Clique.Period == 0 { + w.commitNewWork() } } // System stopped - case <-self.txSub.Err(): + case <-w.txSub.Err(): return - case <-self.chainHeadSub.Err(): + case <-w.chainHeadSub.Err(): return - case <-self.chainSideSub.Err(): + case <-w.chainSideSub.Err(): return } } } -func (self *worker) wait() { +func (w *worker) wait() { for { mustCommitNewWork := true - for result := range self.recv { - atomic.AddInt32(&self.atWork, -1) + for result := range w.recv { + atomic.AddInt32(&w.atWork, -1) if result == nil { continue @@ -310,7 +310,7 @@ func (self *worker) wait() { for _, log := range work.state.Logs() { log.BlockHash = block.Hash() } - stat, err := self.chain.WriteBlockWithState(block, work.receipts, work.state) + stat, err := w.chain.WriteBlockWithState(block, work.receipts, work.state) if err != nil { log.Error("Failed writing block to chain", "err", err) continue @@ -321,7 +321,7 @@ func (self *worker) wait() { mustCommitNewWork = false } // Broadcast the block and announce chain insertion event - self.mux.Post(core.NewMinedBlockEvent{Block: block}) + w.mux.Post(core.NewMinedBlockEvent{Block: block}) var ( events []interface{} logs = work.state.Logs() @@ -330,25 +330,25 @@ func (self *worker) wait() { if stat == core.CanonStatTy { events = append(events, core.ChainHeadEvent{Block: block}) } - self.chain.PostChainEvents(events, logs) + w.chain.PostChainEvents(events, logs) // Insert the block into the set of pending ones to wait for confirmations - self.unconfirmed.Insert(block.NumberU64(), block.Hash()) + w.unconfirmed.Insert(block.NumberU64(), block.Hash()) if mustCommitNewWork { - self.commitNewWork() + w.commitNewWork() } } } } // push sends a new work task to currently live miner agents. -func (self *worker) push(work *Work) { - if atomic.LoadInt32(&self.mining) != 1 { +func (w *worker) push(work *Work) { + if atomic.LoadInt32(&w.mining) != 1 { return } - for agent := range self.agents { - atomic.AddInt32(&self.atWork, 1) + for agent := range w.agents { + atomic.AddInt32(&w.atWork, 1) if ch := agent.Work(); ch != nil { ch <- work } @@ -356,14 +356,14 @@ func (self *worker) push(work *Work) { } // makeCurrent creates a new environment for the current cycle. -func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error { - state, err := self.chain.StateAt(parent.Root()) +func (w *worker) makeCurrent(parent *types.Block, header *types.Header) error { + state, err := w.chain.StateAt(parent.Root()) if err != nil { return err } work := &Work{ - config: self.config, - signer: types.NewEIP155Signer(self.config.ChainId), + config: w.config, + signer: types.NewEIP155Signer(w.config.ChainId), state: state, ancestors: set.New(), family: set.New(), @@ -373,7 +373,7 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error } // when 08 is processed ancestors contain 07 (quick block) - for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) { + for _, ancestor := range w.chain.GetBlocksFromHash(parent.Hash(), 7) { for _, uncle := range ancestor.Uncles() { work.family.Add(uncle.Hash()) } @@ -383,20 +383,20 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error // Keep track of transactions which return errors so they can be removed work.tcount = 0 - self.current = work + w.current = work return nil } -func (self *worker) commitNewWork() { - self.mu.Lock() - defer self.mu.Unlock() - self.uncleMu.Lock() - defer self.uncleMu.Unlock() - self.currentMu.Lock() - defer self.currentMu.Unlock() +func (w *worker) commitNewWork() { + w.mu.Lock() + defer w.mu.Unlock() + w.uncleMu.Lock() + defer w.uncleMu.Unlock() + w.currentMu.Lock() + defer w.currentMu.Unlock() tstart := time.Now() - parent := self.chain.CurrentBlock() + parent := w.chain.CurrentBlock() tstamp := tstart.Unix() if parent.Time().Cmp(new(big.Int).SetInt64(tstamp)) >= 0 { @@ -414,24 +414,24 @@ func (self *worker) commitNewWork() { ParentHash: parent.Hash(), Number: num.Add(num, common.Big1), GasLimit: core.CalcGasLimit(parent), - Extra: self.extra, + Extra: w.extra, Time: big.NewInt(tstamp), } // Only set the coinbase if we are mining (avoid spurious block rewards) - if atomic.LoadInt32(&self.mining) == 1 { - header.Coinbase = self.coinbase + if atomic.LoadInt32(&w.mining) == 1 { + header.Coinbase = w.coinbase } - if err := self.engine.Prepare(self.chain, header); err != nil { + if err := w.engine.Prepare(w.chain, header); err != nil { log.Error("Failed to prepare header for mining", "err", err) return } // If we are care about TheDAO hard-fork check whether to override the extra-data or not - if daoBlock := self.config.DAOForkBlock; daoBlock != nil { + if daoBlock := w.config.DAOForkBlock; daoBlock != nil { // Check whether the block is among the fork extra-override range limit := new(big.Int).Add(daoBlock, params.DAOForkExtraRange) if header.Number.Cmp(daoBlock) >= 0 && header.Number.Cmp(limit) < 0 { // Depending whether we support or oppose the fork, override differently - if self.config.DAOForkSupport { + if w.config.DAOForkSupport { header.Extra = common.CopyBytes(params.DAOForkBlockExtra) } else if bytes.Equal(header.Extra, params.DAOForkBlockExtra) { header.Extra = []byte{} // If miner opposes, don't let it use the reserved extra-data @@ -439,34 +439,34 @@ func (self *worker) commitNewWork() { } } // Could potentially happen if starting to mine in an odd state. - err := self.makeCurrent(parent, header) + err := w.makeCurrent(parent, header) if err != nil { log.Error("Failed to create mining context", "err", err) return } // Create the current work task and check any fork transitions needed - work := self.current - if self.config.DAOForkSupport && self.config.DAOForkBlock != nil && self.config.DAOForkBlock.Cmp(header.Number) == 0 { + work := w.current + if w.config.DAOForkSupport && w.config.DAOForkBlock != nil && w.config.DAOForkBlock.Cmp(header.Number) == 0 { misc.ApplyDAOHardFork(work.state) } - pending, err := self.eth.TxPool().Pending() + pending, err := w.eth.TxPool().Pending() if err != nil { log.Error("Failed to fetch pending transactions", "err", err) return } - txs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending) - work.commitTransactions(self.mux, txs, self.chain, self.coinbase) + txs := types.NewTransactionsByPriceAndNonce(w.current.signer, pending) + work.commitTransactions(w.mux, txs, w.chain, w.coinbase) // compute uncles for the new block. var ( uncles []*types.Header badUncles []common.Hash ) - for hash, uncle := range self.possibleUncles { + for hash, uncle := range w.possibleUncles { if len(uncles) == 2 { break } - if err := self.commitUncle(work, uncle.Header()); err != nil { + if err := w.commitUncle(work, uncle.Header()); err != nil { log.Trace("Bad uncle found and will be removed", "hash", hash) log.Trace(fmt.Sprint(uncle)) @@ -477,23 +477,23 @@ func (self *worker) commitNewWork() { } } for _, hash := range badUncles { - delete(self.possibleUncles, hash) + delete(w.possibleUncles, hash) } // Create the new block to seal with the consensus engine - if work.Block, err = self.engine.Finalize(self.chain, header, work.state, work.txs, uncles, work.receipts); err != nil { + if work.Block, err = w.engine.Finalize(w.chain, header, work.state, work.txs, uncles, work.receipts); err != nil { log.Error("Failed to finalize block for sealing", "err", err) return } // We only care about logging if we're actually mining. - if atomic.LoadInt32(&self.mining) == 1 { + if atomic.LoadInt32(&w.mining) == 1 { log.Info("Commit new 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) + w.unconfirmed.Shift(work.Block.NumberU64() - 1) } - self.push(work) - self.updateSnapshot() + w.push(work) + w.updateSnapshot() } -func (self *worker) commitUncle(work *Work, uncle *types.Header) error { +func (w *worker) commitUncle(work *Work, uncle *types.Header) error { hash := uncle.Hash() if work.uncles.Has(hash) { return fmt.Errorf("uncle not unique") @@ -508,17 +508,17 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error { return nil } -func (self *worker) updateSnapshot() { - self.snapshotMu.Lock() - defer self.snapshotMu.Unlock() +func (w *worker) updateSnapshot() { + w.snapshotMu.Lock() + defer w.snapshotMu.Unlock() - self.snapshotBlock = types.NewBlock( - self.current.header, - self.current.txs, + w.snapshotBlock = types.NewBlock( + w.current.header, + w.current.txs, nil, - self.current.receipts, + w.current.receipts, ) - self.snapshotState = self.current.state.Copy() + w.snapshotState = w.current.state.Copy() } func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) { @@ -616,4 +616,4 @@ func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, c env.receipts = append(env.receipts, receipt) return nil, receipt.Logs -} +} \ No newline at end of file