Go-ethereum 源碼解析之 miner/worker.go （上）
Source Code
// 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/>.
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/core/vm"
"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 *params.ChainConfig
engine consensus.Engine
eth Backend
chain *core.BlockChain
gasFloor uint64
gasCeil uint64
// 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.
possibleUncles 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.
// 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 *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, recommit time.Duration, gasFloor, gasCeil uint64) *worker {
worker := &worker{
config: config,
engine: engine,
eth: eth,
mux: mux,
chain: eth.BlockChain(),
gasFloor: gasFloor,
gasCeil: gasCeil,
possibleUncles: 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.
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.config.Clique == nil || w.config.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:
if _, exist := w.possibleUncles[ev.Block.Hash()]; exist {
continue
}
// Add side block to possible uncle block set.
w.possibleUncles[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.possibleUncles[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 {
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)
w.commitTransactions(txset, coinbase, nil)
w.updateSnapshot()
} else {
// If we're mining, but nothing is being processed, wake on new transactions
if w.config.Clique != nil && w.config.Clique.Period == 0 {
w.commitNewWork(nil, false, 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 {
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.config.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.possibleUncles[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.config, w.chain, &coinbase, w.current.gasPool, w.current.state, w.current.header, tx, &w.current.header.GasUsed, vm.Config{})
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.config.IsEIP155(w.current.header.Number) {
log.Trace("Ignoring reply protected transaction", "hash", tx.Hash(), "eip155", w.config.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().Cmp(new(big.Int).SetInt64(timestamp)) >= 0 {
timestamp = parent.Time().Int64() + 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.gasFloor, w.gasCeil),
Extra: w.extra,
Time: big.NewInt(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.config.DAOForkBlock; daoBlock != nil {
// Check whether the block is among the fork extra-override range
limit := new(big.Int).Add(daoBlock, params.DAOForkExtraRange)
if header.Number.Cmp(daoBlock) >= 0 && header.Number.Cmp(limit) < 0 {
// Depending whether we support or oppose the fork, override differently
if w.config.DAOForkSupport {
header.Extra = common.CopyBytes(params.DAOForkBlockExtra)
} else if bytes.Equal(header.Extra, params.DAOForkBlockExtra) {
header.Extra = []byte{} // If miner opposes, don't let it use the reserved extra-data
}
}
}
// 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.config.DAOForkSupport && w.config.DAOForkBlock != nil && w.config.DAOForkBlock.Cmp(header.Number) == 0 {
misc.ApplyDAOHardFork(env.state)
}
// Accumulate the uncles for the current block
for hash, uncle := range w.possibleUncles {
if uncle.NumberU64()+staleThreshold <= header.Number.Uint64() {
delete(w.possibleUncles, hash)
}
}
uncles := make([]*types.Header, 0, 2)
for hash, uncle := range w.possibleUncles {
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())
}
}
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.Finalize(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
}
Appendix A. 協(xié)程批注
本附錄中用于描述本文件中啟動(dòng)了哪些協(xié)程，各自又是通過哪些通道進(jìn)行消息交互的。同時(shí)，對(duì)于會(huì)被使用到的外部協(xié)程也進(jìn)行了簡單的描述。
1. 命名協(xié)程
- 在對(duì)象 miner.worker 的構(gòu)造函數(shù) newWorker() 中啟動(dòng)新的獨(dú)立協(xié)程運(yùn)行方法 worker.mainLoop()，不妨將此協(xié)程稱作命名協(xié)程 worker.mainLoop()。
- 在對(duì)象 miner.worker 的構(gòu)造函數(shù) newWorker() 中啟動(dòng)新的獨(dú)立協(xié)程運(yùn)行方法 worker.newWorkLoop(recommit)，不妨將此協(xié)程稱作命名協(xié)程 worker.newWorkLoop()。
- 在對(duì)象 miner.worker 的構(gòu)造函數(shù) newWorker() 中啟動(dòng)新的獨(dú)立協(xié)程運(yùn)行方法 worker.resultLoop()，不妨將此協(xié)程稱作命名協(xié)程 worker.resultLoop()。
- 在對(duì)象 miner.worker 的構(gòu)造函數(shù) newWorker() 中啟動(dòng)新的獨(dú)立協(xié)程運(yùn)行方法 worker.taskLoop()，不妨將此協(xié)程稱作命名協(xié)程 worker.taskLoop()。
消息在上述四個(gè)命名協(xié)程中的流轉(zhuǎn)方向：
- 命名協(xié)程 worker.newWorkLoop() 基于接收到的消息向命名協(xié)程 worker.mainLoop() 提交事件 miner.newWorkReq，事件的提交最終是在命名協(xié)程 worker.newWorkLoop() 的內(nèi)置函數(shù) commit() 中完成。
- 命名協(xié)程 worker.mainLoop() 基于接收到的消息向命名協(xié)程 worker.taskLoop() 提交任務(wù) miner.task，任務(wù)的提交最終是在命名協(xié)程 worker.mainLoop() 調(diào)用的方法 worker.commitNewWork() 和方法 worker.commit() 中完成。
- 命名協(xié)程 worker.taskLoop() 基于接收到的消息向命名協(xié)程 worker.resultLoop() 提交已簽名區(qū)塊 types.Block，已簽名區(qū)塊的提交最終是在共識(shí)引擎的簽名方法 clique.Seal() 的匿名協(xié)程中完成。
各命名協(xié)程接收消息和發(fā)送消息的具體描述：
- 命名協(xié)程 worker.newWorkLoop() 從通道 worker.startCh 接收驅(qū)動(dòng) worker 的開始事件 struct{}，從通道 worker.chainHeadCh 接收事件 core.ChainHeadEvent，從通道 timer.C 接收事件 time.Time，從通道 worker.resubmitIntervalCh 接收事件 time.Duration，從通道 worker.resubmitAdjustCh 接收事件 intervalAdjust。命名協(xié)程 worker.newWorkLoop() 向通道 worker.newWorkCh 發(fā)送事件 miner.newWorkReq。
- 命名協(xié)程 worker.mainLoop() 從通道 worker.newWorkCh 接收事件 miner.newWorkReq，從通道 worker.chainSideCh 接收事件 core.ChainSideEvent，從通道 worker.txsCh 接收事件 core.NewTxsEvent。命名協(xié)程 worker.mainLoop() 向通道 worker.taskCh 發(fā)送事件 miner.task。
- 命名協(xié)程 worker.taskLoop() 從通道 worker.taskCh 接收事件 miner.task。命名協(xié)程 worker.taskLoop() 通過共識(shí)引擎的簽名方法 clique.Seal() 最終向通道 worker.resultCh 發(fā)送消息 types.Block。
- 命名協(xié)程 worker.resultLoop() 從通道 worker.resultCh 接收事件 types.Block。命名協(xié)程 worker.resultLoop() 通過方法 TypeMux.Post() 將最終的簽名區(qū)塊廣播給網(wǎng)絡(luò)中其它節(jié)點(diǎn)，通過 BlockChain.PostChainEvents() 將簽名區(qū)塊及其對(duì)應(yīng)的事件向本地節(jié)點(diǎn)的事件訂閱者通過 JSON-RPC 的方式發(fā)送事件。
2. 匿名協(xié)程
- 在共識(shí)引擎的簽名方法 Cilque.Seal() 中啟動(dòng)了匿名協(xié)程，用于將已簽名區(qū)塊發(fā)送給通道 worker.resultCh。
- 在方法 commitTransactions() 中啟動(dòng)一個(gè)獨(dú)立的匿名協(xié)程，將所有得到正常處理的交易產(chǎn)生的日志集合通過方法 TypeMux.Post() 發(fā)送給訂閱者。方法 commitTransactions() 由命名協(xié)程 worker.mainLoop() 調(diào)用。
Appendix B. 日志信息
這些日志記錄了關(guān)鍵的流程，同時(shí)記錄了可能的出錯(cuò)原因。
1. 需要重點(diǎn)關(guān)注的日志
- log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealhash, "hash", hash, "elapsed", common.PrettyDuration(time.Since(task.createdAt)))
- log.Trace("Not enough gas for further transactions", "have", w.current.gasPool, "want", params.TxGas)
- log.Trace("Gas limit exceeded for current block", "sender", from)
- log.Info("Mining too far in the future", "wait", common.PrettyDuration(wait))
- log.Error("Refusing to mine without etherbase")
- log.Error("Failed to prepare header for mining", "err", err)
- log.Debug("Committing new uncle to block", "hash", hash)
- 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)))
2. 其它日志
- log.Warn("Sanitizing miner recommit interval", "provided", recommit, "updated", minRecommitInterval)
- log.Warn("Sanitizing miner recommit interval", "provided", interval, "updated", minRecommitInterval)
- log.Info("Miner recommit interval update", "from", minRecommit, "to", interval)
- log.Trace("Increase miner recommit interval", "from", before, "to", recommit)
- log.Trace("Decrease miner recommit interval", "from", before, "to", recommit)
- log.Warn("Block sealing failed", "err", err)
- log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", hash)
- log.Error("Failed writing block to chain", "err", err)
- log.Trace("Ignoring reply protected transaction", "hash", tx.Hash(), "eip155", w.config.EIP155Block)
- log.Trace("Skipping transaction with low nonce", "sender", from, "nonce", tx.Nonce())
- log.Trace("Skipping account with hight nonce", "sender", from, "nonce", tx.Nonce())
- log.Debug("Transaction failed, account skipped", "hash", tx.Hash(), "err", err)
- log.Error("Failed to create mining context", "err", err)
- log.Trace("Possible uncle rejected", "hash", hash, "reason", err)
- log.Error("Failed to fetch pending transactions", "err", err)
- log.Info("Worker has exited")
Appendix C. 總體批注
1. const
定義了挖礦流程相關(guān)的一些常數(shù)。特別需要注意兩個(gè)以共識(shí)協(xié)議相關(guān)的常量 miningLogAtDepth 和 staleThreshold，目前都是 7。可以簡單地這樣理解，對(duì)于給定區(qū)塊，該區(qū)塊在經(jīng)過 miningLogAtDepth 個(gè)區(qū)塊之后被整個(gè)鏈確認(rèn)。
2. type environment struct
定義了數(shù)據(jù)結(jié)構(gòu) environment，用于描述當(dāng)前挖礦所需的環(huán)境。
3. type task struct
定義了數(shù)據(jù)結(jié)構(gòu) task，用于描述發(fā)送給共識(shí)引擎進(jìn)行簽名的待簽名區(qū)塊，以及從共識(shí)引擎接收的已簽名區(qū)塊。
4. const
定義了中斷相關(guān)的一些枚舉值，用于描述中斷信號(hào)。
5. type newWorkReq struct
定義了數(shù)據(jù)結(jié)構(gòu) newWorkReq，用于描述如何開始一個(gè)新任務(wù)。任務(wù)所需的具體信息包含在當(dāng)前環(huán)境 environment 中。
6. type intervalAdjust struct
定義了數(shù)據(jù)結(jié)構(gòu) intervalAdjust，描述重新提交間隔調(diào)整所需的參數(shù)。同時(shí)，需要注意，另外一些參數(shù)是通過 timer.Time 定時(shí)器提供的。
7. type worker struct
定義了數(shù)據(jù)結(jié)構(gòu) worker。對(duì)象 worker 是挖礦的主要實(shí)現(xiàn)，啟動(dòng)了多個(gè)協(xié)程來執(zhí)行獨(dú)立的邏輯流程：
構(gòu)建挖礦的當(dāng)前環(huán)境
接收交易
接收已簽名區(qū)塊
提交交易
構(gòu)建當(dāng)前正在挖的區(qū)塊及任務(wù)
組裝區(qū)塊頭
使用共識(shí)引擎設(shè)定區(qū)塊頭中的共識(shí)相關(guān)字段、對(duì)整個(gè)區(qū)塊進(jìn)行最終的簽名
將經(jīng)共識(shí)引擎已簽名的區(qū)塊進(jìn)行廣播
構(gòu)造函數(shù) newWorker() 用于根據(jù)給定參數(shù)構(gòu)建 worker。
方法 setEtherbase() 設(shè)置用于初始化區(qū)塊 coinbase 字段的 etherbase。
方法 setExtra() 設(shè)置用于初始化區(qū)塊額外字段的內(nèi)容。
方法 setRecommitInterval() 更新礦工簽名工作重新提交的間隔。
方法 pending() 返回待處理的狀態(tài)和相應(yīng)的區(qū)塊。
方法 pendingBlock() 返回待處理的區(qū)塊。
方法 start() 采用原子操作將 running 字段置為 1，并觸發(fā)新工作的提交。
方法 stop() 采用原子操作將 running 字段置為 0。
方法 isRunning() 返回 worker 是否正在運(yùn)行的指示符。
方法 close() 終止由 worker 維護(hù)的所有后臺(tái)線程。注意 worker 不支持被關(guān)閉多次，這是由 Go 語言不允許多次關(guān)閉同一個(gè)通道決定的。
方法 newWorkLoop() 是一個(gè)獨(dú)立的協(xié)程，基于接收到的事件提交新的挖礦工作。不妨將此協(xié)程稱作命名協(xié)程 worker.newWorkLoop()。
方法 mainLoop() 是一個(gè)獨(dú)立的協(xié)程，用于根據(jù)接收到的事件重新生成簽名任務(wù)。不妨將此協(xié)程稱作命名協(xié)程 worker.mainLoop()。
方法 taskLoop() 是一個(gè)獨(dú)立的協(xié)程，用于從生成器中獲取待簽名任務(wù)，并將它們提交給共識(shí)引擎。不妨將此協(xié)程稱作命名協(xié)程 worker.taskLoop()。
方法 resultLoop() 是一個(gè)獨(dú)立的協(xié)程，用于處理簽名區(qū)塊的提交和廣播，以及更新相關(guān)數(shù)據(jù)到數(shù)據(jù)庫。不妨將此協(xié)程稱作命名協(xié)程 worker.resultLoop()。
方法 makeCurrent() 為當(dāng)前周期創(chuàng)建新的環(huán)境 environment。
方法 commitUncle() 將給定的區(qū)塊添加至叔區(qū)塊集合中，如果添加失敗則返回錯(cuò)誤。
方法 updateSnapshot() 更新待處理區(qū)塊和狀態(tài)的快照。注意，此函數(shù)確保當(dāng)前變量是線程安全的。
方法 commitTransactions() 提交交易列表 txs，并附上交易的發(fā)起者地址。根據(jù)整個(gè)交易列表 txs 是否都被有效提交，返回 true 或 false。
方法 commitNewWork() 基于父區(qū)塊生成幾個(gè)新的簽名任務(wù)。
方法 commit() 運(yùn)行任何交易的后續(xù)狀態(tài)修改，組裝最終區(qū)塊，并在共識(shí)引擎運(yùn)行時(shí)提交新工作。
Reference
Contributor
- Windstamp, https://github.com/windstamp
