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 } if _, err := self.chain.InsertChain(types.Blocks{block}); err == nil { for _, uncle := range block.Uncles() { delete(self.possibleUncles, uncle.Hash()) } self.mux.Post(core.NewMinedBlockEvent{block}) 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) jsonlogger.LogJson(&logger.EthMinerNewBlock{ BlockHash: block.Hash().Hex(), BlockNumber: block.Number(), ChainHeadHash: block.ParentHash().Hex(), BlockPrevHash: block.ParentHash().Hex(), }) } else { self.commitNewWork() } } } } func (self *worker) push() { if atomic.LoadInt32(&self.mining) == 1 { // 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 } 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) } // commit state root after all state transitions. core.AccumulateRewards(self.current.state, header, uncles) current.state.Update() header.Root = current.state.Root() // create the new block whose nonce will be mined. current.block = types.NewBlock(header, current.txs, uncles, current.receipts) // 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 }