// Copyright 2015 The go-ethereum Authors // This file is part of the go-ethereum library. // // The go-ethereum library is free software: you can redistribute it and/or modify // it under the terms of the GNU Lesser General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // The go-ethereum library is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Lesser General Public License for more details. // // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // // NOTE: this piece of code is modified by Ted Yin. // The modification is also licensed under the same LGPL. package miner import ( "bytes" "errors" "math/big" "sync" "sync/atomic" "time" "github.com/ava-labs/coreth/consensus" "github.com/ava-labs/coreth/consensus/misc" "github.com/ava-labs/coreth/core" "github.com/ava-labs/coreth/core/state" "github.com/ava-labs/coreth/core/types" "github.com/ava-labs/coreth/params" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/event" "github.com/ava-labs/go-ethereum/log" mapset "github.com/deckarep/golang-set" ) const ( // resultQueueSize is the size of channel listening to sealing result. resultQueueSize = 10 // txChanSize is the size of channel listening to NewTxsEvent. // The number is referenced from the size of tx pool. txChanSize = 4096 // chainHeadChanSize is the size of channel listening to ChainHeadEvent. chainHeadChanSize = 10 // chainSideChanSize is the size of channel listening to ChainSideEvent. chainSideChanSize = 10 // resubmitAdjustChanSize is the size of resubmitting interval adjustment channel. resubmitAdjustChanSize = 10 // miningLogAtDepth is the number of confirmations before logging successful mining. miningLogAtDepth = 7 // minRecommitInterval is the minimal time interval to recreate the mining block with // any newly arrived transactions. minRecommitInterval = 1 * time.Second // maxRecommitInterval is the maximum time interval to recreate the mining block with // any newly arrived transactions. maxRecommitInterval = 15 * time.Second // intervalAdjustRatio is the impact a single interval adjustment has on sealing work // resubmitting interval. intervalAdjustRatio = 0.1 // intervalAdjustBias is applied during the new resubmit interval calculation in favor of // increasing upper limit or decreasing lower limit so that the limit can be reachable. intervalAdjustBias = 200 * 1000.0 * 1000.0 // staleThreshold is the maximum depth of the acceptable stale block. staleThreshold = 7 ) // environment is the worker's current environment and holds all of the current state information. type environment struct { signer types.Signer state *state.StateDB // apply state changes here ancestors mapset.Set // ancestor set (used for checking uncle parent validity) family mapset.Set // family set (used for checking uncle invalidity) uncles mapset.Set // uncle set tcount int // tx count in cycle gasPool *core.GasPool // available gas used to pack transactions header *types.Header txs []*types.Transaction receipts []*types.Receipt } // task contains all information for consensus engine sealing and result submitting. type task struct { receipts []*types.Receipt state *state.StateDB block *types.Block createdAt time.Time } const ( commitInterruptNone int32 = iota commitInterruptNewHead commitInterruptResubmit ) // newWorkReq represents a request for new sealing work submitting with relative interrupt notifier. type newWorkReq struct { interrupt *int32 noempty bool timestamp int64 } // intervalAdjust represents a resubmitting interval adjustment. type intervalAdjust struct { ratio float64 inc bool } type MinerCallbacks struct { OnSealFinish func(*types.Block) error OnSealDrop func(*types.Block) OnHeaderNew func(*types.Header) } // worker is the main object which takes care of submitting new work to consensus engine // and gathering the sealing result. type worker struct { config *Config chainConfig *params.ChainConfig engine consensus.Engine eth Backend chain *core.BlockChain // Subscriptions mux *event.TypeMux txsCh chan core.NewTxsEvent txsSub event.Subscription chainHeadCh chan core.ChainHeadEvent chainHeadSub event.Subscription chainSideCh chan core.ChainSideEvent chainSideSub event.Subscription // Channels newWorkCh chan *newWorkReq taskCh chan *task resultCh chan *types.Block startCh chan struct{} exitCh chan struct{} resubmitIntervalCh chan time.Duration resubmitAdjustCh chan *intervalAdjust current *environment // An environment for current running cycle. localUncles map[common.Hash]*types.Block // A set of side blocks generated locally as the possible uncle blocks. remoteUncles map[common.Hash]*types.Block // A set of side blocks as the possible uncle blocks. unconfirmed *unconfirmedBlocks // A set of locally mined blocks pending canonicalness confirmations. mu sync.RWMutex // The lock used to protect the coinbase and extra fields coinbase common.Address extra []byte pendingMu sync.RWMutex pendingTasks map[common.Hash]*task snapshotMu sync.RWMutex // The lock used to protect the block snapshot and state snapshot snapshotBlock *types.Block snapshotState *state.StateDB // atomic status counters running int32 // The indicator whether the consensus engine is running or not. newTxs int32 // New arrival transaction count since last sealing work submitting. // External functions isLocalBlock func(block *types.Block) bool // Function used to determine whether the specified block is mined by local miner. // Test hooks newTaskHook func(*task) // Method to call upon receiving a new sealing task. skipSealHook func(*task) bool // Method to decide whether skipping the sealing. fullTaskHook func() // Method to call before pushing the full sealing task. resubmitHook func(time.Duration, time.Duration) // Method to call upon updating resubmitting interval. manualMining bool manualUncle bool disableUncle bool minerCallbacks *MinerCallbacks } func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, isLocalBlock func(*types.Block) bool, mcb *MinerCallbacks) *worker { worker := &worker{ config: config, chainConfig: chainConfig, engine: engine, eth: eth, mux: mux, chain: eth.BlockChain(), isLocalBlock: isLocalBlock, localUncles: make(map[common.Hash]*types.Block), remoteUncles: make(map[common.Hash]*types.Block), unconfirmed: newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth), pendingTasks: make(map[common.Hash]*task), txsCh: make(chan core.NewTxsEvent, txChanSize), chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize), chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize), newWorkCh: make(chan *newWorkReq), taskCh: make(chan *task), resultCh: make(chan *types.Block, resultQueueSize), exitCh: make(chan struct{}), startCh: make(chan struct{}, 1), resubmitIntervalCh: make(chan time.Duration), resubmitAdjustCh: make(chan *intervalAdjust, resubmitAdjustChanSize), manualMining: config.ManualMining, manualUncle: config.ManualUncle, disableUncle: config.DisableUncle, minerCallbacks: mcb, } if worker.disableUncle { worker.manualUncle = true } // Subscribe NewTxsEvent for tx pool worker.txsSub = eth.TxPool().SubscribeNewTxsEvent(worker.txsCh) // Subscribe events for blockchain worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh) worker.chainSideSub = eth.BlockChain().SubscribeChainSideEvent(worker.chainSideCh) // Sanitize recommit interval if the user-specified one is too short. recommit := worker.config.Recommit if recommit < minRecommitInterval { log.Warn("Sanitizing miner recommit interval", "provided", recommit, "updated", minRecommitInterval) recommit = minRecommitInterval } go worker.mainLoop() go worker.newWorkLoop(recommit) go worker.resultLoop() go worker.taskLoop() // Submit first work to initialize pending state. worker.startCh <- struct{}{} return worker } // setEtherbase sets the etherbase used to initialize the block coinbase field. func (w *worker) setEtherbase(addr common.Address) { w.mu.Lock() defer w.mu.Unlock() w.coinbase = addr } // setExtra sets the content used to initialize the block extra field. func (w *worker) setExtra(extra []byte) { w.mu.Lock() defer w.mu.Unlock() w.extra = extra } // setRecommitInterval updates the interval for miner sealing work recommitting. func (w *worker) setRecommitInterval(interval time.Duration) { w.resubmitIntervalCh <- interval } // pending returns the pending state and corresponding block. func (w *worker) pending() (*types.Block, *state.StateDB) { // return a snapshot to avoid contention on currentMu mutex w.snapshotMu.RLock() defer w.snapshotMu.RUnlock() if w.snapshotState == nil { return nil, nil } return w.snapshotBlock, w.snapshotState.Copy() } // pendingBlock returns pending block. func (w *worker) pendingBlock() *types.Block { // return a snapshot to avoid contention on currentMu mutex w.snapshotMu.RLock() defer w.snapshotMu.RUnlock() return w.snapshotBlock } // start sets the running status as 1 and triggers new work submitting. func (w *worker) start() { atomic.StoreInt32(&w.running, 1) w.startCh <- struct{}{} } // stop sets the running status as 0. func (w *worker) stop() { atomic.StoreInt32(&w.running, 0) } // isRunning returns an indicator whether worker is running or not. func (w *worker) isRunning() bool { return atomic.LoadInt32(&w.running) == 1 } // close terminates all background threads maintained by the worker. // Note the worker does not support being closed multiple times. func (w *worker) close() { close(w.exitCh) } func (w *worker) genBlock() { interrupt := new(int32) *interrupt = commitInterruptNone w.newWorkCh <- &newWorkReq{ interrupt: interrupt, noempty: false, timestamp: time.Now().Unix(), } } // newWorkLoop is a standalone goroutine to submit new mining work upon received events. func (w *worker) newWorkLoop(recommit time.Duration) { var ( interrupt *int32 minRecommit = recommit // minimal resubmit interval specified by user. timestamp int64 // timestamp for each round of mining. ) timer := time.NewTimer(0) <-timer.C // discard the initial tick // commit aborts in-flight transaction execution with given signal and resubmits a new one. commit := func(noempty bool, s int32) { if interrupt != nil { atomic.StoreInt32(interrupt, s) } interrupt = new(int32) w.newWorkCh <- &newWorkReq{interrupt: interrupt, noempty: noempty, timestamp: timestamp} timer.Reset(recommit) atomic.StoreInt32(&w.newTxs, 0) } // recalcRecommit recalculates the resubmitting interval upon feedback. recalcRecommit := func(target float64, inc bool) { var ( prev = float64(recommit.Nanoseconds()) next float64 ) if inc { next = prev*(1-intervalAdjustRatio) + intervalAdjustRatio*(target+intervalAdjustBias) // Recap if interval is larger than the maximum time interval if next > float64(maxRecommitInterval.Nanoseconds()) { next = float64(maxRecommitInterval.Nanoseconds()) } } else { next = prev*(1-intervalAdjustRatio) + intervalAdjustRatio*(target-intervalAdjustBias) // Recap if interval is less than the user specified minimum if next < float64(minRecommit.Nanoseconds()) { next = float64(minRecommit.Nanoseconds()) } } recommit = time.Duration(int64(next)) } // clearPending cleans the stale pending tasks. clearPending := func(number uint64) { w.pendingMu.Lock() for h, t := range w.pendingTasks { if t.block.NumberU64()+staleThreshold <= number { delete(w.pendingTasks, h) } } w.pendingMu.Unlock() } for { select { case <-w.startCh: clearPending(w.chain.CurrentBlock().NumberU64()) timestamp = time.Now().Unix() if !w.manualMining { log.Trace("commit ch") commit(false, commitInterruptNewHead) } case head := <-w.chainHeadCh: clearPending(head.Block.NumberU64()) timestamp = time.Now().Unix() if !w.manualMining { log.Trace("commit update") commit(false, commitInterruptNewHead) } case <-timer.C: // If mining is running resubmit a new work cycle periodically to pull in // higher priced transactions. Disable this overhead for pending blocks. if w.isRunning() && (w.chainConfig.Clique == nil || w.chainConfig.Clique.Period > 0) { // Short circuit if no new transaction arrives. if atomic.LoadInt32(&w.newTxs) == 0 { timer.Reset(recommit) continue } log.Trace("commit resubmit") commit(true, commitInterruptResubmit) } case interval := <-w.resubmitIntervalCh: // Adjust resubmit interval explicitly by user. if interval < minRecommitInterval { log.Warn("Sanitizing miner recommit interval", "provided", interval, "updated", minRecommitInterval) interval = minRecommitInterval } log.Info("Miner recommit interval update", "from", minRecommit, "to", interval) minRecommit, recommit = interval, interval if w.resubmitHook != nil { w.resubmitHook(minRecommit, recommit) } case adjust := <-w.resubmitAdjustCh: // Adjust resubmit interval by feedback. if adjust.inc { before := recommit recalcRecommit(float64(recommit.Nanoseconds())/adjust.ratio, true) log.Trace("Increase miner recommit interval", "from", before, "to", recommit) } else { before := recommit recalcRecommit(float64(minRecommit.Nanoseconds()), false) log.Trace("Decrease miner recommit interval", "from", before, "to", recommit) } if w.resubmitHook != nil { w.resubmitHook(minRecommit, recommit) } case <-w.exitCh: return } } } // mainLoop is a standalone goroutine to regenerate the sealing task based on the received event. func (w *worker) mainLoop() { defer w.txsSub.Unsubscribe() defer w.chainHeadSub.Unsubscribe() defer w.chainSideSub.Unsubscribe() for { select { case req := <-w.newWorkCh: w.commitNewWork(req.interrupt, req.noempty, req.timestamp) case ev := <-w.chainSideCh: // Short circuit for duplicate side blocks if _, exist := w.localUncles[ev.Block.Hash()]; exist { continue } if _, exist := w.remoteUncles[ev.Block.Hash()]; exist { continue } // Add side block to possible uncle block set depending on the author. if w.isLocalBlock != nil && w.isLocalBlock(ev.Block) { w.localUncles[ev.Block.Hash()] = ev.Block } else { w.remoteUncles[ev.Block.Hash()] = ev.Block } // If our mining block contains less than 2 uncle blocks, // add the new uncle block if valid and regenerate a mining block. if w.isRunning() && w.current != nil && w.current.uncles.Cardinality() < 2 && !w.manualUncle { start := time.Now() if err := w.commitUncle(w.current, ev.Block.Header()); err == nil { var uncles []*types.Header w.current.uncles.Each(func(item interface{}) bool { hash, ok := item.(common.Hash) if !ok { return false } uncle, exist := w.localUncles[hash] if !exist { uncle, exist = w.remoteUncles[hash] } if !exist { return false } uncles = append(uncles, uncle.Header()) return false }) if !w.manualUncle { w.commit(uncles, nil, true, start) } } } case ev := <-w.txsCh: // Apply transactions to the pending state if we're not mining. // // Note all transactions received may not be continuous with transactions // already included in the current mining block. These transactions will // be automatically eliminated. if !w.isRunning() && w.current != nil { // If block is already full, abort if gp := w.current.gasPool; gp != nil && gp.Gas() < params.TxGas { continue } w.mu.RLock() coinbase := w.coinbase w.mu.RUnlock() txs := make(map[common.Address]types.Transactions) for _, tx := range ev.Txs { acc, _ := types.Sender(w.current.signer, tx) txs[acc] = append(txs[acc], tx) } txset := types.NewTransactionsByPriceAndNonce(w.current.signer, txs) tcount := w.current.tcount w.commitTransactions(txset, coinbase, nil) // Only update the snapshot if any new transactons were added // to the pending block if tcount != w.current.tcount { w.updateSnapshot() } } else { // If clique is running in dev mode(period is 0), disable // advance sealing here. if w.chainConfig.Clique != nil && w.chainConfig.Clique.Period == 0 { w.commitNewWork(nil, true, time.Now().Unix()) } } atomic.AddInt32(&w.newTxs, int32(len(ev.Txs))) // System stopped case <-w.exitCh: return case <-w.txsSub.Err(): return case <-w.chainHeadSub.Err(): return case <-w.chainSideSub.Err(): return } } } // taskLoop is a standalone goroutine to fetch sealing task from the generator and // push them to consensus engine. func (w *worker) taskLoop() { var ( stopCh chan struct{} prev common.Hash ) // interrupt aborts the in-flight sealing task. interrupt := func() { if stopCh != nil { close(stopCh) stopCh = nil } } for { select { case task := <-w.taskCh: if w.newTaskHook != nil { w.newTaskHook(task) } // Reject duplicate sealing work due to resubmitting. sealHash := w.engine.SealHash(task.block.Header()) if sealHash == prev { log.Warn("Reject duplicate sealing work due to resubmitting.") if w.minerCallbacks.OnSealDrop != nil { w.minerCallbacks.OnSealDrop(task.block) } continue } // Interrupt previous sealing operation interrupt() stopCh, prev = make(chan struct{}), sealHash if w.skipSealHook != nil && w.skipSealHook(task) { if w.minerCallbacks.OnSealDrop != nil { w.minerCallbacks.OnSealDrop(task.block) } continue } w.pendingMu.Lock() w.pendingTasks[w.engine.SealHash(task.block.Header())] = task w.pendingMu.Unlock() if err := w.engine.Seal(w.chain, task.block, w.resultCh, stopCh); err != nil { log.Warn("Block sealing failed", "err", err) } case <-w.exitCh: interrupt() return } } } // resultLoop is a standalone goroutine to handle sealing result submitting // and flush relative data to the database. func (w *worker) resultLoop() { for { select { case block := <-w.resultCh: // Short circuit when receiving empty result. if block == nil { continue } // Short circuit when receiving duplicate result caused by resubmitting. if w.chain.HasBlock(block.Hash(), block.NumberU64()) { continue } var ( sealhash = w.engine.SealHash(block.Header()) hash = block.Hash() ) w.pendingMu.RLock() task, exist := w.pendingTasks[sealhash] w.pendingMu.RUnlock() if !exist { log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", hash) continue } // Different block could share same sealhash, deep copy here to prevent write-write conflict. var ( receipts = make([]*types.Receipt, len(task.receipts)) logs []*types.Log ) for i, receipt := range task.receipts { // add block location fields receipt.BlockHash = hash receipt.BlockNumber = block.Number() receipt.TransactionIndex = uint(i) receipts[i] = new(types.Receipt) *receipts[i] = *receipt // Update the block hash in all logs since it is now available and not when the // receipt/log of individual transactions were created. for _, log := range receipt.Logs { log.BlockHash = hash } logs = append(logs, receipt.Logs...) } // Commit block and state to database. //fmt.Printf("parent1: %s\n", w.chain.CurrentBlock().Hash().String()) stat, err := w.chain.WriteBlockWithState(block, receipts, task.state) stat = core.CanonStatTy //fmt.Printf("parent2: %s\n", w.chain.CurrentBlock().Hash().String()) if err != nil { log.Error("Failed writing block to chain", "err", err) continue } log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealhash, "hash", hash, "elapsed", common.PrettyDuration(time.Since(task.createdAt))) if w.minerCallbacks.OnSealFinish != nil { w.minerCallbacks.OnSealFinish(block) } // Broadcast the block and announce chain insertion event w.mux.Post(core.NewMinedBlockEvent{Block: block}) var events []interface{} switch stat { case core.CanonStatTy: events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs}) events = append(events, core.ChainHeadEvent{Block: block}) case core.SideStatTy: events = append(events, core.ChainSideEvent{Block: block}) } w.chain.PostChainEvents(events, logs) // Insert the block into the set of pending ones to resultLoop for confirmations w.unconfirmed.Insert(block.NumberU64(), block.Hash()) case <-w.exitCh: return } } } // makeCurrent creates a new environment for the current cycle. func (w *worker) makeCurrent(parent *types.Block, header *types.Header) error { state, err := w.chain.StateAt(parent.Root()) if err != nil { return err } env := &environment{ signer: types.NewEIP155Signer(w.chainConfig.ChainID), state: state, ancestors: mapset.NewSet(), family: mapset.NewSet(), uncles: mapset.NewSet(), header: header, } // when 08 is processed ancestors contain 07 (quick block) for _, ancestor := range w.chain.GetBlocksFromHash(parent.Hash(), 7) { for _, uncle := range ancestor.Uncles() { env.family.Add(uncle.Hash()) } env.family.Add(ancestor.Hash()) env.ancestors.Add(ancestor.Hash()) } // Keep track of transactions which return errors so they can be removed env.tcount = 0 w.current = env return nil } // commitUncle adds the given block to uncle block set, returns error if failed to add. func (w *worker) commitUncle(env *environment, uncle *types.Header) error { if w.disableUncle { return nil } hash := uncle.Hash() if env.uncles.Contains(hash) { return errors.New("uncle not unique") } if env.header.ParentHash == uncle.ParentHash { return errors.New("uncle is sibling") } if !env.ancestors.Contains(uncle.ParentHash) { return errors.New("uncle's parent unknown") } if env.family.Contains(hash) { return errors.New("uncle already included") } env.uncles.Add(uncle.Hash()) return nil } // updateSnapshot updates pending snapshot block and state. // Note this function assumes the current variable is thread safe. func (w *worker) updateSnapshot() { w.snapshotMu.Lock() defer w.snapshotMu.Unlock() var uncles []*types.Header w.current.uncles.Each(func(item interface{}) bool { hash, ok := item.(common.Hash) if !ok { return false } uncle, exist := w.localUncles[hash] if !exist { uncle, exist = w.remoteUncles[hash] } if !exist { return false } uncles = append(uncles, uncle.Header()) return false }) w.snapshotBlock = types.NewBlock( w.current.header, w.current.txs, uncles, w.current.receipts, ) w.snapshotState = w.current.state.Copy() } func (w *worker) commitTransaction(tx *types.Transaction, coinbase common.Address) ([]*types.Log, error) { snap := w.current.state.Snapshot() receipt, _, err := core.ApplyTransaction(w.chainConfig, w.chain, &coinbase, w.current.gasPool, w.current.state, w.current.header, tx, &w.current.header.GasUsed, *w.chain.GetVMConfig()) if err != nil { w.current.state.RevertToSnapshot(snap) return nil, err } w.current.txs = append(w.current.txs, tx) w.current.receipts = append(w.current.receipts, receipt) return receipt.Logs, nil } func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coinbase common.Address, interrupt *int32) bool { // Short circuit if current is nil if w.current == nil { return true } if w.current.gasPool == nil { w.current.gasPool = new(core.GasPool).AddGas(w.current.header.GasLimit) } var coalescedLogs []*types.Log for { // In the following three cases, we will interrupt the execution of the transaction. // (1) new head block event arrival, the interrupt signal is 1 // (2) worker start or restart, the interrupt signal is 1 // (3) worker recreate the mining block with any newly arrived transactions, the interrupt signal is 2. // For the first two cases, the semi-finished work will be discarded. // For the third case, the semi-finished work will be submitted to the consensus engine. if interrupt != nil && atomic.LoadInt32(interrupt) != commitInterruptNone { // Notify resubmit loop to increase resubmitting interval due to too frequent commits. if atomic.LoadInt32(interrupt) == commitInterruptResubmit { ratio := float64(w.current.header.GasLimit-w.current.gasPool.Gas()) / float64(w.current.header.GasLimit) if ratio < 0.1 { ratio = 0.1 } w.resubmitAdjustCh <- &intervalAdjust{ ratio: ratio, inc: true, } } return atomic.LoadInt32(interrupt) == commitInterruptNewHead } // If we don't have enough gas for any further transactions then we're done if w.current.gasPool.Gas() < params.TxGas { log.Trace("Not enough gas for further transactions", "have", w.current.gasPool, "want", params.TxGas) break } // Retrieve the next transaction and abort if all done tx := txs.Peek() if tx == nil { break } // Error may be ignored here. The error has already been checked // during transaction acceptance is the transaction pool. // // We use the eip155 signer regardless of the current hf. from, _ := types.Sender(w.current.signer, tx) // Check whether the tx is replay protected. If we're not in the EIP155 hf // phase, start ignoring the sender until we do. if tx.Protected() && !w.chainConfig.IsEIP155(w.current.header.Number) { log.Trace("Ignoring reply protected transaction", "hash", tx.Hash(), "eip155", w.chainConfig.EIP155Block) txs.Pop() continue } // Start executing the transaction w.current.state.Prepare(tx.Hash(), common.Hash{}, w.current.tcount) logs, err := w.commitTransaction(tx, coinbase) switch err { case core.ErrGasLimitReached: // Pop the current out-of-gas transaction without shifting in the next from the account log.Trace("Gas limit exceeded for current block", "sender", from) txs.Pop() case core.ErrNonceTooLow: // New head notification data race between the transaction pool and miner, shift log.Trace("Skipping transaction with low nonce", "sender", from, "nonce", tx.Nonce()) txs.Shift() case core.ErrNonceTooHigh: // Reorg notification data race between the transaction pool and miner, skip account = log.Trace("Skipping account with hight nonce", "sender", from, "nonce", tx.Nonce()) txs.Pop() case nil: // Everything ok, collect the logs and shift in the next transaction from the same account coalescedLogs = append(coalescedLogs, logs...) w.current.tcount++ txs.Shift() default: // Strange error, discard the transaction and get the next in line (note, the // nonce-too-high clause will prevent us from executing in vain). log.Debug("Transaction failed, account skipped", "hash", tx.Hash(), "err", err) txs.Shift() } } if !w.isRunning() && len(coalescedLogs) > 0 { // We don't push the pendingLogsEvent while we are mining. The reason is that // when we are mining, the worker will regenerate a mining block every 3 seconds. // In order to avoid pushing the repeated pendingLog, we disable the pending log pushing. // make a copy, the state caches the logs and these logs get "upgraded" from pending to mined // logs by filling in the block hash when the block was mined by the local miner. This can // cause a race condition if a log was "upgraded" before the PendingLogsEvent is processed. cpy := make([]*types.Log, len(coalescedLogs)) for i, l := range coalescedLogs { cpy[i] = new(types.Log) *cpy[i] = *l } go w.mux.Post(core.PendingLogsEvent{Logs: cpy}) } // Notify resubmit loop to decrease resubmitting interval if current interval is larger // than the user-specified one. if interrupt != nil { w.resubmitAdjustCh <- &intervalAdjust{inc: false} } return false } // commitNewWork generates several new sealing tasks based on the parent block. func (w *worker) commitNewWork(interrupt *int32, noempty bool, timestamp int64) { w.mu.RLock() defer w.mu.RUnlock() tstart := time.Now() parent := w.chain.CurrentBlock() if parent.Time() >= uint64(timestamp) { //timestamp = int64(parent.Time() + 1) timestamp = int64(parent.Time()) } // this will ensure we're not going off too far in the future if now := time.Now().Unix(); timestamp > now+1 { wait := time.Duration(timestamp-now) * time.Second log.Info("Mining too far in the future", "wait", common.PrettyDuration(wait)) time.Sleep(wait) } num := parent.Number() header := &types.Header{ ParentHash: parent.Hash(), Number: num.Add(num, common.Big1), GasLimit: core.CalcGasLimit(parent, w.config.GasFloor, w.config.GasCeil), Extra: w.extra, Time: uint64(timestamp), } // Only set the coinbase if our consensus engine is running (avoid spurious block rewards) if w.isRunning() { if w.coinbase == (common.Address{}) { log.Error("Refusing to mine without etherbase") return } header.Coinbase = w.coinbase } 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 := w.chainConfig.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 w.chainConfig.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 } } } if w.minerCallbacks.OnHeaderNew != nil { w.minerCallbacks.OnHeaderNew(header) } // Could potentially happen if starting to mine in an odd state. 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 env := w.current if w.chainConfig.DAOForkSupport && w.chainConfig.DAOForkBlock != nil && w.chainConfig.DAOForkBlock.Cmp(header.Number) == 0 { misc.ApplyDAOHardFork(env.state) } // Accumulate the uncles for the current block uncles := make([]*types.Header, 0, 2) commitUncles := func(blocks map[common.Hash]*types.Block) { // Clean up stale uncle blocks first for hash, uncle := range blocks { if uncle.NumberU64()+staleThreshold <= header.Number.Uint64() { delete(blocks, hash) } } if w.disableUncle { return } for hash, uncle := range blocks { if len(uncles) == 2 { break } if err := w.commitUncle(env, uncle.Header()); err != nil { log.Trace("Possible uncle rejected", "hash", hash, "reason", err) } else { log.Debug("Committing new uncle to block", "hash", hash) uncles = append(uncles, uncle.Header()) } } } // Prefer to locally generated uncle commitUncles(w.localUncles) commitUncles(w.remoteUncles) if !noempty && !w.manualMining { // Create an empty block based on temporary copied state for sealing in advance without waiting block // execution finished. w.commit(uncles, nil, false, tstart) } // Fill the block with all available pending transactions. pending, err := w.eth.TxPool().Pending() if err != nil { log.Error("Failed to fetch pending transactions", "err", err) return } // Short circuit if there is no available pending transactions if len(pending) == 0 && !w.manualMining { w.updateSnapshot() return } // Split the pending transactions into locals and remotes localTxs, remoteTxs := make(map[common.Address]types.Transactions), pending for _, account := range w.eth.TxPool().Locals() { if txs := remoteTxs[account]; len(txs) > 0 { delete(remoteTxs, account) localTxs[account] = txs } } if len(localTxs) > 0 { txs := types.NewTransactionsByPriceAndNonce(w.current.signer, localTxs) if w.commitTransactions(txs, w.coinbase, interrupt) { return } } if len(remoteTxs) > 0 { txs := types.NewTransactionsByPriceAndNonce(w.current.signer, remoteTxs) if w.commitTransactions(txs, w.coinbase, interrupt) { return } } w.commit(uncles, w.fullTaskHook, true, tstart) } // commit runs any post-transaction state modifications, assembles the final block // and commits new work if consensus engine is running. func (w *worker) commit(uncles []*types.Header, interval func(), update bool, start time.Time) error { // Deep copy receipts here to avoid interaction between different tasks. receipts := make([]*types.Receipt, len(w.current.receipts)) for i, l := range w.current.receipts { receipts[i] = new(types.Receipt) *receipts[i] = *l } s := w.current.state.Copy() block, err := w.engine.FinalizeAndAssemble(w.chain, w.current.header, s, w.current.txs, uncles, w.current.receipts) if err != nil { return err } if w.isRunning() { if interval != nil { interval() } select { case w.taskCh <- &task{receipts: receipts, state: s, block: block, createdAt: time.Now()}: w.unconfirmed.Shift(block.NumberU64() - 1) feesWei := new(big.Int) for i, tx := range block.Transactions() { feesWei.Add(feesWei, new(big.Int).Mul(new(big.Int).SetUint64(receipts[i].GasUsed), tx.GasPrice())) } feesEth := new(big.Float).Quo(new(big.Float).SetInt(feesWei), new(big.Float).SetInt(big.NewInt(params.Ether))) log.Info("Commit new mining work", "number", block.Number(), "sealhash", w.engine.SealHash(block.Header()), "uncles", len(uncles), "txs", w.current.tcount, "gas", block.GasUsed(), "fees", feesEth, "elapsed", common.PrettyDuration(time.Since(start))) case <-w.exitCh: log.Info("Worker has exited") } } if update { w.updateSnapshot() } return nil }