// 工作者是負責將消息應用到新狀態的主要對象

type worker struct {config *ConfigchainConfig *params.ChainConfigengine consensus.Engineeth Backendchain *core.BlockChain// FeedspendingLogsFeed event.Feed// Subscriptionsmux *event.TypeMuxtxsCh chan core.NewTxsEvent // 用來接受txPool里面的交易的通道txsSub event.Subscription // 用來接受txPool里面的交易的訂閱器chainHeadCh chan core.ChainHeadEvent // 用來接受區塊頭的通道chainHeadSub event.SubscriptionchainSideCh chan core.ChainSideEvent chainSideSub event.Subscription// ChannelsnewWorkCh chan *newWorkReqtaskCh chan *taskresultCh chan *types.BlockstartCh chan struct{}exitCh chan struct{}resubmitIntervalCh chan time.DurationresubmitAdjustCh chan *intervalAdjustcurrent *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 fieldscoinbase common.Address extra []bytependingMu sync.RWMutexpendingTasks map[common.Hash]*tasksnapshotMu sync.RWMutex // The lock used to protect the block snapshot and state snapshotsnapshotBlock *types.Block // 快照 BlocksnapshotState *state.StateDB // 快照 StateDB// atomic status countersrunning int32 // The indicator whether the consensus engine is running or not.newTxs int32 // New arrival transaction count since last sealing work submitting.// noempty is the flag used to control whether the feature of pre-seal empty// block is enabled. The default value is false(pre-seal is enabled by default).// But in some special scenario the consensus engine will seal blocks instantaneously,// in this case this feature will add all empty blocks into canonical chain// non-stop and no real transaction will be included.noempty uint32// External functionsisLocalBlock func(block *types.Block) bool // Function used to determine whether the specified block is mined by local miner.// Test hooksnewTaskHook 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, init 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 poolworker.txsSub = eth.TxPool().SubscribeNewTxsEvent(worker.txsCh)// Subscribe events for blockchainworker.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.Recommitif 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.if init {worker.startCh <- struct{}{}}return worker }

// newWorkLoop 是一個獨立goroutine，用于在接收到的事件上提交新的挖掘工作。

func (w *worker) newWorkLoop(recommit time.Duration) {var (interrupt *int32minRecommit = recommit // minimal resubmit interval specified by user.timestamp int64 // timestamp for each round of mining.)timer := time.NewTimer(0)defer timer.Stop()<-timer.C // discard the initial tick// commit使用給定信號中止正在執行的事務，并重新提交一個新的事務。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)}//清除掛起的陳舊任務。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://由用戶顯式調整重新提交的時間間隔。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, intervalif w.resubmitHook != nil {w.resubmitHook(minRecommit, recommit)}case adjust := <-w.resubmitAdjustCh://通過反饋調整重新提交的時間間隔。if adjust.inc {before := recommittarget := float64(recommit.Nanoseconds()) / adjust.ratiorecommit = recalcRecommit(minRecommit, recommit, target, true)log.Trace("Increase miner recommit interval", "from", before, "to", recommit)} else {before := recommitrecommit = recalcRecommit(minRecommit, recommit, 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是一個獨立的goroutine，根據接收到的事件重新生成密封任務。

func (w *worker) mainLoop() {defer w.txsSub.Unsubscribe()defer w.chainHeadSub.Unsubscribe()defer w.chainSideSub.Unsubscribe()for {select {//從newWorkLoop接受一個新任務case req := <-w.newWorkCh://提交新的任務w.commitNewWork(req.interrupt, req.noempty, req.timestamp)case ev := <-w.chainSideCh:// Short circuit for duplicate side blocksif _, 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.Headerw.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://如果不進行挖掘，則將事務應用到掛起狀態。//注:所有收到的交易可能不是連續的已經包含在當前挖掘塊中。這些交易將自動消除。if !w.isRunning() && w.current != nil {// If block is already full, abortif gp := w.current.gasPool; gp != nil && gp.Gas() < params.TxGas {continue}w.mu.RLock()coinbase := w.coinbasew.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.tcountw.commitTransactions(txset, coinbase, nil)// Only update the snapshot if any new transactons were added// to the pending blockif tcount != w.current.tcount {w.updateSnapshot()}} else {//如果不進行挖掘，則將事務應用到掛起狀態。特殊情況，如果共識引擎為0周期派系(dev模式)，//在這里提交挖礦，因為所有的空提交將被拒絕,預先密封(空提交)是禁用的。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)))......}} }

如果不進行挖掘，則將事務應用到掛起狀態。taskLoop是一個獨立的goroutine，用于從生成器獲取密封任務并將其推送到共識引擎。

// 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)}//因重新提交而拒絕重復密封工作。sealHash := w.engine.SealHash(task.block.Header())if sealHash == prev {continue}// Interrupt previous sealing operationinterrupt()stopCh, prev = make(chan struct{}), sealHashif w.skipSealHook != nil && w.skipSealHook(task) {continue}w.pendingMu.Lock()w.pendingTasks[sealHash] = taskw.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是一個獨立的goroutine處理密封結果提交并將相關數據刷新到數據庫。

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 fieldsreceipt.BlockHash = hashreceipt.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._, err := w.chain.WriteBlockWithState(block, receipts, logs, task.state, true)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)))// 廣播塊并宣布鏈插入事件w.mux.Post(core.NewMinedBlockEvent{Block: block})// 將該塊插入到resultLoop中等待的塊中以進行確認w.unconfirmed.Insert(block.NumberU64(), block.Hash())case <-w.exitCh:return}} }

為當前周期創建一個新環境。

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 removedenv.tcount = 0w.current = envreturn nil }

更新掛起的快照塊和狀態。注意:此函數假設當前變量是線程安全的。

func (w *worker) updateSnapshot() {w.snapshotMu.Lock()defer w.snapshotMu.Unlock()var uncles []*types.Headerw.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,new(trie.Trie),)w.snapshotState = w.current.state.Copy() } func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coinbase common.Address, interrupt *int32) bool {// Short circuit if current is nilif w.current == nil {return true}if w.current.gasPool == nil {w.current.gasPool = new(core.GasPool).AddGas(w.current.header.GasLimit)}var coalescedLogs []*types.Logfor {//在以下三種情況下，我們將中斷事務的執行。//(1)新頭塊事件到達時，中斷信號為1//(2)worker啟動或重啟時，中斷信號為1// (3) worker用任何新到達的事務重新創建挖掘塊，中斷信號為2。//對于前兩種情況，半成品將被丟棄。//對于第三種情況，半成品工作將提交給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 doneif 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 donetx := 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 transactionw.current.state.Prepare(tx.Hash(), common.Hash{}, w.current.tcount)logs, err := w.commitTransaction(tx, coinbase)......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}w.pendingLogsFeed.Send(cpy)}//如果當前間隔較大，通知重新提交循環以減少重新提交的時間間隔而不是用戶指定的。if interrupt != nil {w.resubmitAdjustCh <- &intervalAdjust{inc: false}}return false } func (w *worker) commitTransaction(tx *types.Transaction, coinbase common.Address) ([]*types.Log, error) {snap := w.current.state.Snapshot()//嘗試將一個事務應用到給定的狀態數據庫并為其環境使用輸入參數。它會返回收據對于事務，使用了gas，如果事務失敗則出現錯誤，表示該塊無效。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) 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 futureif now := time.Now().Unix(); timestamp > now+1 {wait := time.Duration(timestamp-now) * time.Secondlog.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),}// 只有在我們的共識引擎運行時才設置coinbase(避免虛假的塊獎勵)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 daoBlock := w.chainConfig.DAOForkBlock; daoBlock != nil {// Check whether the block is among the fork extra-override rangelimit := 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 differentlyif 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 neededenv := w.currentif w.chainConfig.DAOForkSupport && w.chainConfig.DAOForkBlock != nil && w.chainConfig.DAOForkBlock.Cmp(header.Number) == 0 {misc.ApplyDAOHardFork(env.state)}// Accumulate the uncles for the current blockuncles := make([]*types.Header, 0, 2)commitUncles := func(blocks map[common.Hash]*types.Block) {// Clean up stale uncle blocks firstfor 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 unclecommitUncles(w.localUncles)commitUncles(w.remoteUncles)//根據臨時復制的狀態創建一個空塊提前封口，無需等待封口執行完成。if !noempty && atomic.LoadUint32(&w.noempty) == 0 {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.// But if we disable empty precommit already, ignore it. Since// empty block is necessary to keep the liveness of the network.if len(pending) == 0 && atomic.LoadUint32(&w.noempty) == 0 {w.updateSnapshot()return}// Split the pending transactions into locals and remoteslocalTxs, remoteTxs := make(map[common.Address]types.Transactions), pendingfor _, 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) }

計算ETH的總消費費用。區塊交易和收據必須有相同的順序。

func totalFees(block *types.Block, receipts []*types.Receipt) *big.Float {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()))}return new(big.Float).Quo(new(big.Float).SetInt(feesWei), new(big.Float).SetInt(big.NewInt(params.Ether))) }

總結

以上是生活随笔為你收集整理的以太坊源码系列之miner解析(2)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。