d8fe64acaa
This fixes an issue with the lru cache not being available when calling WriteBlock. WriteBlock previously always assumed to be called from the InsertChain where the lru cache was always created prior to calling WriteBlock. When being called from the worker this could lead in to a nil pointer exception being thrown and causing database corruption.
555 lines
16 KiB
Go
555 lines
16 KiB
Go
package miner
|
|
|
|
import (
|
|
"fmt"
|
|
"math/big"
|
|
"sort"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/accounts"
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"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/logger"
|
|
"github.com/ethereum/go-ethereum/logger/glog"
|
|
"github.com/ethereum/go-ethereum/pow"
|
|
"gopkg.in/fatih/set.v0"
|
|
)
|
|
|
|
var jsonlogger = logger.NewJsonLogger()
|
|
|
|
// Work holds the current work
|
|
type Work struct {
|
|
Number uint64
|
|
Nonce uint64
|
|
MixDigest []byte
|
|
SeedHash []byte
|
|
}
|
|
|
|
// Agent can register themself with the worker
|
|
type Agent interface {
|
|
Work() chan<- *types.Block
|
|
SetReturnCh(chan<- *types.Block)
|
|
Stop()
|
|
Start()
|
|
GetHashRate() int64
|
|
}
|
|
|
|
const miningLogAtDepth = 5
|
|
|
|
type uint64RingBuffer struct {
|
|
ints []uint64 //array of all integers in buffer
|
|
next int //where is the next insertion? assert 0 <= next < len(ints)
|
|
}
|
|
|
|
// environment is the workers current environment and holds
|
|
// all of the current state information
|
|
type environment struct {
|
|
state *state.StateDB // apply state changes here
|
|
coinbase *state.StateObject // the miner's account
|
|
ancestors *set.Set // ancestor set (used for checking uncle parent validity)
|
|
family *set.Set // family set (used for checking uncle invalidity)
|
|
uncles *set.Set // uncle set
|
|
remove *set.Set // tx which will be removed
|
|
tcount int // tx count in cycle
|
|
ignoredTransactors *set.Set
|
|
lowGasTransactors *set.Set
|
|
ownedAccounts *set.Set
|
|
lowGasTxs types.Transactions
|
|
localMinedBlocks *uint64RingBuffer // the most recent block numbers that were mined locally (used to check block inclusion)
|
|
|
|
block *types.Block // the new block
|
|
|
|
header *types.Header
|
|
txs []*types.Transaction
|
|
receipts []*types.Receipt
|
|
}
|
|
|
|
// worker is the main object which takes care of applying messages to the new state
|
|
type worker struct {
|
|
mu sync.Mutex
|
|
|
|
agents []Agent
|
|
recv chan *types.Block
|
|
mux *event.TypeMux
|
|
quit chan struct{}
|
|
pow pow.PoW
|
|
|
|
eth core.Backend
|
|
chain *core.ChainManager
|
|
proc *core.BlockProcessor
|
|
|
|
coinbase common.Address
|
|
gasPrice *big.Int
|
|
extra []byte
|
|
|
|
currentMu sync.Mutex
|
|
current *environment
|
|
|
|
uncleMu sync.Mutex
|
|
possibleUncles map[common.Hash]*types.Block
|
|
|
|
txQueueMu sync.Mutex
|
|
txQueue map[common.Hash]*types.Transaction
|
|
|
|
// atomic status counters
|
|
mining int32
|
|
atWork int32
|
|
}
|
|
|
|
func newWorker(coinbase common.Address, eth core.Backend) *worker {
|
|
worker := &worker{
|
|
eth: eth,
|
|
mux: eth.EventMux(),
|
|
recv: make(chan *types.Block),
|
|
gasPrice: new(big.Int),
|
|
chain: eth.ChainManager(),
|
|
proc: eth.BlockProcessor(),
|
|
possibleUncles: make(map[common.Hash]*types.Block),
|
|
coinbase: coinbase,
|
|
txQueue: make(map[common.Hash]*types.Transaction),
|
|
quit: make(chan struct{}),
|
|
}
|
|
go worker.update()
|
|
go worker.wait()
|
|
|
|
worker.commitNewWork()
|
|
|
|
return worker
|
|
}
|
|
|
|
func (self *worker) pendingState() *state.StateDB {
|
|
self.currentMu.Lock()
|
|
defer self.currentMu.Unlock()
|
|
return self.current.state
|
|
}
|
|
|
|
func (self *worker) pendingBlock() *types.Block {
|
|
self.currentMu.Lock()
|
|
defer self.currentMu.Unlock()
|
|
if atomic.LoadInt32(&self.mining) == 0 {
|
|
return types.NewBlock(
|
|
self.current.header,
|
|
self.current.txs,
|
|
nil,
|
|
self.current.receipts,
|
|
)
|
|
}
|
|
return self.current.block
|
|
}
|
|
|
|
func (self *worker) start() {
|
|
self.mu.Lock()
|
|
defer self.mu.Unlock()
|
|
|
|
atomic.StoreInt32(&self.mining, 1)
|
|
|
|
// spin up agents
|
|
for _, agent := range self.agents {
|
|
agent.Start()
|
|
}
|
|
}
|
|
|
|
func (self *worker) stop() {
|
|
self.mu.Lock()
|
|
defer self.mu.Unlock()
|
|
|
|
if atomic.LoadInt32(&self.mining) == 1 {
|
|
var keep []Agent
|
|
// stop all agents
|
|
for _, agent := range self.agents {
|
|
agent.Stop()
|
|
// keep all that's not a cpu agent
|
|
if _, ok := agent.(*CpuAgent); !ok {
|
|
keep = append(keep, agent)
|
|
}
|
|
}
|
|
self.agents = keep
|
|
}
|
|
|
|
atomic.StoreInt32(&self.mining, 0)
|
|
atomic.StoreInt32(&self.atWork, 0)
|
|
}
|
|
|
|
func (self *worker) register(agent Agent) {
|
|
self.mu.Lock()
|
|
defer self.mu.Unlock()
|
|
self.agents = append(self.agents, agent)
|
|
agent.SetReturnCh(self.recv)
|
|
}
|
|
|
|
func (self *worker) update() {
|
|
events := self.mux.Subscribe(core.ChainHeadEvent{}, core.ChainSideEvent{}, core.TxPreEvent{})
|
|
|
|
out:
|
|
for {
|
|
select {
|
|
case event := <-events.Chan():
|
|
switch ev := event.(type) {
|
|
case core.ChainHeadEvent:
|
|
self.commitNewWork()
|
|
case core.ChainSideEvent:
|
|
self.uncleMu.Lock()
|
|
self.possibleUncles[ev.Block.Hash()] = ev.Block
|
|
self.uncleMu.Unlock()
|
|
case core.TxPreEvent:
|
|
// Apply transaction to the pending state if we're not mining
|
|
if atomic.LoadInt32(&self.mining) == 0 {
|
|
self.mu.Lock()
|
|
self.current.commitTransactions(types.Transactions{ev.Tx}, self.gasPrice, self.proc)
|
|
self.mu.Unlock()
|
|
}
|
|
}
|
|
case <-self.quit:
|
|
break out
|
|
}
|
|
}
|
|
|
|
events.Unsubscribe()
|
|
}
|
|
|
|
func newLocalMinedBlock(blockNumber uint64, prevMinedBlocks *uint64RingBuffer) (minedBlocks *uint64RingBuffer) {
|
|
if prevMinedBlocks == nil {
|
|
minedBlocks = &uint64RingBuffer{next: 0, ints: make([]uint64, miningLogAtDepth+1)}
|
|
} else {
|
|
minedBlocks = prevMinedBlocks
|
|
}
|
|
|
|
minedBlocks.ints[minedBlocks.next] = blockNumber
|
|
minedBlocks.next = (minedBlocks.next + 1) % len(minedBlocks.ints)
|
|
return minedBlocks
|
|
}
|
|
|
|
func (self *worker) wait() {
|
|
for {
|
|
for block := range self.recv {
|
|
atomic.AddInt32(&self.atWork, -1)
|
|
|
|
if block == nil {
|
|
continue
|
|
}
|
|
|
|
_, err := self.chain.WriteBlock(block, false)
|
|
if err != nil {
|
|
glog.V(logger.Error).Infoln("error writing block to chain", err)
|
|
continue
|
|
}
|
|
|
|
// check staleness and display confirmation
|
|
var stale, confirm string
|
|
canonBlock := self.chain.GetBlockByNumber(block.NumberU64())
|
|
if canonBlock != nil && canonBlock.Hash() != block.Hash() {
|
|
stale = "stale "
|
|
} else {
|
|
confirm = "Wait 5 blocks for confirmation"
|
|
self.current.localMinedBlocks = newLocalMinedBlock(block.Number().Uint64(), self.current.localMinedBlocks)
|
|
}
|
|
|
|
glog.V(logger.Info).Infof("🔨 Mined %sblock (#%v / %x). %s", stale, block.Number(), block.Hash().Bytes()[:4], confirm)
|
|
|
|
// broadcast before waiting for validation
|
|
go self.mux.Post(core.NewMinedBlockEvent{block})
|
|
|
|
self.commitNewWork()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (self *worker) push() {
|
|
if atomic.LoadInt32(&self.mining) == 1 {
|
|
if core.Canary(self.current.state) {
|
|
glog.Infoln("Toxicity levels rising to deadly levels. Your canary has died. You can go back or continue down the mineshaft --more--")
|
|
glog.Infoln("You turn back and abort mining")
|
|
return
|
|
}
|
|
|
|
// push new work to agents
|
|
for _, agent := range self.agents {
|
|
atomic.AddInt32(&self.atWork, 1)
|
|
|
|
if agent.Work() != nil {
|
|
agent.Work() <- self.current.block
|
|
} else {
|
|
common.Report(fmt.Sprintf("%v %T\n", agent, agent))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// makeCurrent creates a new environment for the current cycle.
|
|
func (self *worker) makeCurrent(parent *types.Block, header *types.Header) {
|
|
state := state.New(parent.Root(), self.eth.StateDb())
|
|
current := &environment{
|
|
state: state,
|
|
ancestors: set.New(),
|
|
family: set.New(),
|
|
uncles: set.New(),
|
|
header: header,
|
|
coinbase: state.GetOrNewStateObject(self.coinbase),
|
|
}
|
|
|
|
// when 08 is processed ancestors contain 07 (quick block)
|
|
for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) {
|
|
for _, uncle := range ancestor.Uncles() {
|
|
current.family.Add(uncle.Hash())
|
|
}
|
|
current.family.Add(ancestor.Hash())
|
|
current.ancestors.Add(ancestor.Hash())
|
|
}
|
|
accounts, _ := self.eth.AccountManager().Accounts()
|
|
|
|
// Keep track of transactions which return errors so they can be removed
|
|
current.remove = set.New()
|
|
current.tcount = 0
|
|
current.ignoredTransactors = set.New()
|
|
current.lowGasTransactors = set.New()
|
|
current.ownedAccounts = accountAddressesSet(accounts)
|
|
if self.current != nil {
|
|
current.localMinedBlocks = self.current.localMinedBlocks
|
|
}
|
|
self.current = current
|
|
}
|
|
|
|
func (w *worker) setGasPrice(p *big.Int) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
|
|
// calculate the minimal gas price the miner accepts when sorting out transactions.
|
|
const pct = int64(90)
|
|
w.gasPrice = gasprice(p, pct)
|
|
|
|
w.mux.Post(core.GasPriceChanged{w.gasPrice})
|
|
}
|
|
|
|
func (self *worker) isBlockLocallyMined(deepBlockNum uint64) bool {
|
|
//Did this instance mine a block at {deepBlockNum} ?
|
|
var isLocal = false
|
|
for idx, blockNum := range self.current.localMinedBlocks.ints {
|
|
if deepBlockNum == blockNum {
|
|
isLocal = true
|
|
self.current.localMinedBlocks.ints[idx] = 0 //prevent showing duplicate logs
|
|
break
|
|
}
|
|
}
|
|
//Short-circuit on false, because the previous and following tests must both be true
|
|
if !isLocal {
|
|
return false
|
|
}
|
|
|
|
//Does the block at {deepBlockNum} send earnings to my coinbase?
|
|
var block = self.chain.GetBlockByNumber(deepBlockNum)
|
|
return block != nil && block.Coinbase() == self.coinbase
|
|
}
|
|
|
|
func (self *worker) logLocalMinedBlocks(previous *environment) {
|
|
if previous != nil && self.current.localMinedBlocks != nil {
|
|
nextBlockNum := self.current.block.NumberU64()
|
|
for checkBlockNum := previous.block.NumberU64(); checkBlockNum < nextBlockNum; checkBlockNum++ {
|
|
inspectBlockNum := checkBlockNum - miningLogAtDepth
|
|
if self.isBlockLocallyMined(inspectBlockNum) {
|
|
glog.V(logger.Info).Infof("🔨 🔗 Mined %d blocks back: block #%v", miningLogAtDepth, inspectBlockNum)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (self *worker) commitNewWork() {
|
|
self.mu.Lock()
|
|
defer self.mu.Unlock()
|
|
self.uncleMu.Lock()
|
|
defer self.uncleMu.Unlock()
|
|
self.currentMu.Lock()
|
|
defer self.currentMu.Unlock()
|
|
|
|
tstart := time.Now()
|
|
parent := self.chain.CurrentBlock()
|
|
tstamp := tstart.Unix()
|
|
if tstamp <= parent.Time() {
|
|
tstamp = parent.Time() + 1
|
|
}
|
|
// this will ensure we're not going off too far in the future
|
|
if now := time.Now().Unix(); tstamp > now+4 {
|
|
wait := time.Duration(tstamp-now) * time.Second
|
|
glog.V(logger.Info).Infoln("We are too far in the future. Waiting for", wait)
|
|
time.Sleep(wait)
|
|
}
|
|
|
|
num := parent.Number()
|
|
header := &types.Header{
|
|
ParentHash: parent.Hash(),
|
|
Number: num.Add(num, common.Big1),
|
|
Difficulty: core.CalcDifficulty(tstamp, parent.Time(), parent.Difficulty()),
|
|
GasLimit: core.CalcGasLimit(parent),
|
|
GasUsed: new(big.Int),
|
|
Coinbase: self.coinbase,
|
|
Extra: self.extra,
|
|
Time: uint64(tstamp),
|
|
}
|
|
|
|
previous := self.current
|
|
self.makeCurrent(parent, header)
|
|
current := self.current
|
|
|
|
// commit transactions for this run.
|
|
transactions := self.eth.TxPool().GetTransactions()
|
|
sort.Sort(types.TxByNonce{transactions})
|
|
current.coinbase.SetGasLimit(header.GasLimit)
|
|
current.commitTransactions(transactions, self.gasPrice, self.proc)
|
|
self.eth.TxPool().RemoveTransactions(current.lowGasTxs)
|
|
|
|
// compute uncles for the new block.
|
|
var (
|
|
uncles []*types.Header
|
|
badUncles []common.Hash
|
|
)
|
|
for hash, uncle := range self.possibleUncles {
|
|
if len(uncles) == 2 {
|
|
break
|
|
}
|
|
if err := self.commitUncle(uncle.Header()); err != nil {
|
|
if glog.V(logger.Ridiculousness) {
|
|
glog.V(logger.Detail).Infof("Bad uncle found and will be removed (%x)\n", hash[:4])
|
|
glog.V(logger.Detail).Infoln(uncle)
|
|
}
|
|
badUncles = append(badUncles, hash)
|
|
} else {
|
|
glog.V(logger.Debug).Infof("commiting %x as uncle\n", hash[:4])
|
|
uncles = append(uncles, uncle.Header())
|
|
}
|
|
}
|
|
for _, hash := range badUncles {
|
|
delete(self.possibleUncles, hash)
|
|
}
|
|
|
|
if atomic.LoadInt32(&self.mining) == 1 {
|
|
// commit state root after all state transitions.
|
|
core.AccumulateRewards(self.current.state, header, uncles)
|
|
current.state.Update()
|
|
self.current.state.Sync()
|
|
header.Root = current.state.Root()
|
|
}
|
|
|
|
// create the new block whose nonce will be mined.
|
|
current.block = types.NewBlock(header, current.txs, uncles, current.receipts)
|
|
self.current.block.Td = new(big.Int).Set(core.CalcTD(self.current.block, self.chain.GetBlock(self.current.block.ParentHash())))
|
|
|
|
// We only care about logging if we're actually mining.
|
|
if atomic.LoadInt32(&self.mining) == 1 {
|
|
glog.V(logger.Info).Infof("commit new work on block %v with %d txs & %d uncles. Took %v\n", current.block.Number(), current.tcount, len(uncles), time.Since(tstart))
|
|
self.logLocalMinedBlocks(previous)
|
|
}
|
|
|
|
self.push()
|
|
}
|
|
|
|
func (self *worker) commitUncle(uncle *types.Header) error {
|
|
hash := uncle.Hash()
|
|
if self.current.uncles.Has(hash) {
|
|
return core.UncleError("Uncle not unique")
|
|
}
|
|
if !self.current.ancestors.Has(uncle.ParentHash) {
|
|
return core.UncleError(fmt.Sprintf("Uncle's parent unknown (%x)", uncle.ParentHash[0:4]))
|
|
}
|
|
if self.current.family.Has(hash) {
|
|
return core.UncleError(fmt.Sprintf("Uncle already in family (%x)", hash))
|
|
}
|
|
self.current.uncles.Add(uncle.Hash())
|
|
return nil
|
|
}
|
|
|
|
func (env *environment) commitTransactions(transactions types.Transactions, gasPrice *big.Int, proc *core.BlockProcessor) {
|
|
for _, tx := range transactions {
|
|
// We can skip err. It has already been validated in the tx pool
|
|
from, _ := tx.From()
|
|
|
|
// Check if it falls within margin. Txs from owned accounts are always processed.
|
|
if tx.GasPrice().Cmp(gasPrice) < 0 && !env.ownedAccounts.Has(from) {
|
|
// ignore the transaction and transactor. We ignore the transactor
|
|
// because nonce will fail after ignoring this transaction so there's
|
|
// no point
|
|
env.lowGasTransactors.Add(from)
|
|
|
|
glog.V(logger.Info).Infof("transaction(%x) below gas price (tx=%v ask=%v). All sequential txs from this address(%x) will be ignored\n", tx.Hash().Bytes()[:4], common.CurrencyToString(tx.GasPrice()), common.CurrencyToString(gasPrice), from[:4])
|
|
}
|
|
|
|
// Continue with the next transaction if the transaction sender is included in
|
|
// the low gas tx set. This will also remove the tx and all sequential transaction
|
|
// from this transactor
|
|
if env.lowGasTransactors.Has(from) {
|
|
// add tx to the low gas set. This will be removed at the end of the run
|
|
// owned accounts are ignored
|
|
if !env.ownedAccounts.Has(from) {
|
|
env.lowGasTxs = append(env.lowGasTxs, tx)
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Move on to the next transaction when the transactor is in ignored transactions set
|
|
// This may occur when a transaction hits the gas limit. When a gas limit is hit and
|
|
// the transaction is processed (that could potentially be included in the block) it
|
|
// will throw a nonce error because the previous transaction hasn't been processed.
|
|
// Therefor we need to ignore any transaction after the ignored one.
|
|
if env.ignoredTransactors.Has(from) {
|
|
continue
|
|
}
|
|
|
|
env.state.StartRecord(tx.Hash(), common.Hash{}, 0)
|
|
|
|
err := env.commitTransaction(tx, proc)
|
|
switch {
|
|
case core.IsNonceErr(err) || core.IsInvalidTxErr(err):
|
|
env.remove.Add(tx.Hash())
|
|
|
|
if glog.V(logger.Detail) {
|
|
glog.Infof("TX (%x) failed, will be removed: %v\n", tx.Hash().Bytes()[:4], err)
|
|
}
|
|
case state.IsGasLimitErr(err):
|
|
// ignore the transactor so no nonce errors will be thrown for this account
|
|
// next time the worker is run, they'll be picked up again.
|
|
env.ignoredTransactors.Add(from)
|
|
|
|
glog.V(logger.Detail).Infof("Gas limit reached for (%x) in this block. Continue to try smaller txs\n", from[:4])
|
|
default:
|
|
env.tcount++
|
|
}
|
|
}
|
|
}
|
|
|
|
func (env *environment) commitTransaction(tx *types.Transaction, proc *core.BlockProcessor) error {
|
|
snap := env.state.Copy()
|
|
receipt, _, err := proc.ApplyTransaction(env.coinbase, env.state, env.header, tx, env.header.GasUsed, true)
|
|
if err != nil && (core.IsNonceErr(err) || state.IsGasLimitErr(err) || core.IsInvalidTxErr(err)) {
|
|
env.state.Set(snap)
|
|
return err
|
|
}
|
|
env.txs = append(env.txs, tx)
|
|
env.receipts = append(env.receipts, receipt)
|
|
return nil
|
|
}
|
|
|
|
// TODO: remove or use
|
|
func (self *worker) HashRate() int64 {
|
|
return 0
|
|
}
|
|
|
|
// gasprice calculates a reduced gas price based on the pct
|
|
// XXX Use big.Rat?
|
|
func gasprice(price *big.Int, pct int64) *big.Int {
|
|
p := new(big.Int).Set(price)
|
|
p.Div(p, big.NewInt(100))
|
|
p.Mul(p, big.NewInt(pct))
|
|
return p
|
|
}
|
|
|
|
func accountAddressesSet(accounts []accounts.Account) *set.Set {
|
|
accountSet := set.New()
|
|
for _, account := range accounts {
|
|
accountSet.Add(account.Address)
|
|
}
|
|
return accountSet
|
|
}
|