diff options
Diffstat (limited to 'miner')
| -rw-r--r-- | miner/miner.go | 51 | ||||
| -rw-r--r-- | miner/unconfirmed.go | 136 | ||||
| -rw-r--r-- | miner/worker.go | 1002 |
3 files changed, 1189 insertions, 0 deletions
diff --git a/miner/miner.go b/miner/miner.go new file mode 100644 index 0000000..b342bb8 --- /dev/null +++ b/miner/miner.go @@ -0,0 +1,51 @@ +// Copyright 2014 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/>. + +// Package miner implements Ethereum block creation and mining. +package miner + +import ( + + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/consensus" + "github.com/ethereum/go-ethereum/params" + eminer "github.com/ethereum/go-ethereum/miner" +) + +// Backend wraps all methods required for mining. +type Backend interface { + BlockChain() *core.BlockChain + TxPool() *core.TxPool +} + +// Config is the configuration parameters of mining. +type Config = eminer.Config + +type Worker = worker + +func NewWorker(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, isLocalBlock func(*types.Block) bool) *Worker { + return newWorker(config, chainConfig, engine, eth, mux, isLocalBlock) +} + +func (self *Worker) Start() { + self.start() +} + +func (self *Worker) Stop() { + self.stop() +} diff --git a/miner/unconfirmed.go b/miner/unconfirmed.go new file mode 100644 index 0000000..3a176e8 --- /dev/null +++ b/miner/unconfirmed.go @@ -0,0 +1,136 @@ +// Copyright 2016 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/>. + +package miner + +import ( + "container/ring" + "sync" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/log" +) + +// chainRetriever is used by the unconfirmed block set to verify whether a previously +// mined block is part of the canonical chain or not. +type chainRetriever interface { + // GetHeaderByNumber retrieves the canonical header associated with a block number. + GetHeaderByNumber(number uint64) *types.Header + + // GetBlockByNumber retrieves the canonical block associated with a block number. + GetBlockByNumber(number uint64) *types.Block +} + +// unconfirmedBlock is a small collection of metadata about a locally mined block +// that is placed into a unconfirmed set for canonical chain inclusion tracking. +type unconfirmedBlock struct { + index uint64 + hash common.Hash +} + +// unconfirmedBlocks implements a data structure to maintain locally mined blocks +// have not yet reached enough maturity to guarantee chain inclusion. It is +// used by the miner to provide logs to the user when a previously mined block +// has a high enough guarantee to not be reorged out of the canonical chain. +type unconfirmedBlocks struct { + chain chainRetriever // Blockchain to verify canonical status through + depth uint // Depth after which to discard previous blocks + blocks *ring.Ring // Block infos to allow canonical chain cross checks + lock sync.RWMutex // Protects the fields from concurrent access +} + +// newUnconfirmedBlocks returns new data structure to track currently unconfirmed blocks. +func newUnconfirmedBlocks(chain chainRetriever, depth uint) *unconfirmedBlocks { + return &unconfirmedBlocks{ + chain: chain, + depth: depth, + } +} + +// Insert adds a new block to the set of unconfirmed ones. +func (set *unconfirmedBlocks) Insert(index uint64, hash common.Hash) { + // If a new block was mined locally, shift out any old enough blocks + set.Shift(index) + + // Create the new item as its own ring + item := ring.New(1) + item.Value = &unconfirmedBlock{ + index: index, + hash: hash, + } + // Set as the initial ring or append to the end + set.lock.Lock() + defer set.lock.Unlock() + + if set.blocks == nil { + set.blocks = item + } else { + set.blocks.Move(-1).Link(item) + } + // Display a log for the user to notify of a new mined block unconfirmed + log.Info("🔨 mined potential block", "number", index, "hash", hash) +} + +// Shift drops all unconfirmed blocks from the set which exceed the unconfirmed sets depth +// allowance, checking them against the canonical chain for inclusion or staleness +// report. +func (set *unconfirmedBlocks) Shift(height uint64) { + set.lock.Lock() + defer set.lock.Unlock() + + for set.blocks != nil { + // Retrieve the next unconfirmed block and abort if too fresh + next := set.blocks.Value.(*unconfirmedBlock) + if next.index+uint64(set.depth) > height { + break + } + // Block seems to exceed depth allowance, check for canonical status + header := set.chain.GetHeaderByNumber(next.index) + switch { + case header == nil: + log.Warn("Failed to retrieve header of mined block", "number", next.index, "hash", next.hash) + case header.Hash() == next.hash: + log.Info("🔗 block reached canonical chain", "number", next.index, "hash", next.hash) + default: + // Block is not canonical, check whether we have an uncle or a lost block + included := false + for number := next.index; !included && number < next.index+uint64(set.depth) && number <= height; number++ { + if block := set.chain.GetBlockByNumber(number); block != nil { + for _, uncle := range block.Uncles() { + if uncle.Hash() == next.hash { + included = true + break + } + } + } + } + if included { + log.Info("⑂ block became an uncle", "number", next.index, "hash", next.hash) + } else { + log.Info("😱 block lost", "number", next.index, "hash", next.hash) + } + } + // Drop the block out of the ring + if set.blocks.Value == set.blocks.Next().Value { + set.blocks = nil + } else { + set.blocks = set.blocks.Move(-1) + set.blocks.Unlink(1) + set.blocks = set.blocks.Move(1) + } + } +} diff --git a/miner/worker.go b/miner/worker.go new file mode 100644 index 0000000..edfd3f9 --- /dev/null +++ b/miner/worker.go @@ -0,0 +1,1002 @@ +// 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 adopted from +// github.com/ethereum/go-ethereum/miner/worker.go, +// modified by Ted Yin. +// The modification is also licensed under the same LGPL. + +package miner + +import ( + "bytes" + "errors" + "math/big" + "sync" + "sync/atomic" + "time" + + mapset "github.com/deckarep/golang-set" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus" + "github.com/ethereum/go-ethereum/consensus/misc" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/params" +) + +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 +} + +// 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. +} + +func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, isLocalBlock func(*types.Block) bool) *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), + } + // 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) +} + +// 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() + commit(false, commitInterruptNewHead) + + case head := <-w.chainHeadCh: + clearPending(head.Block.NumberU64()) + timestamp = time.Now().Unix() + 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 + } + 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 { + 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 + }) + 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 { + continue + } + // Interrupt previous sealing operation + interrupt() + stopCh, prev = make(chan struct{}), sealHash + + if w.skipSealHook != nil && w.skipSealHook(task) { + 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. + stat, err := w.chain.WriteBlockWithState(block, receipts, task.state) + 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))) + + // 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 { + 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) + } + // 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 + } + } + } + // 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) + } + } + 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 { + // 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.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 +} |