path: root/miner/worker.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
    extraDb common.Database

    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(),
        extraDb:        eth.ExtraDb(),
        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
            }

            parent := self.chain.GetBlock(block.ParentHash())
            if parent == nil {
                glog.V(logger.Error).Infoln("Invalid block found during mining")
                continue
            }
            if err := core.ValidateHeader(self.eth.BlockProcessor().Pow, block.Header(), parent, true); err != nil && err != core.BlockFutureErr {
                glog.V(logger.Error).Infoln("Invalid header on mined block:", err)
                continue
            }

            stat, err := self.chain.WriteBlock(block, false)
            if err != nil {
                glog.V(logger.Error).Infoln("error writing block to chain", err)
                continue
            }
            // check if canon block and write transactions
            if stat == core.CanonStatTy {
                // This puts transactions in a extra db for rpc
                core.PutTransactions(self.extraDb, block, block.Transactions())
                // store the receipts
                core.PutReceipts(self.extraDb, self.current.receipts)
            }

            // 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 func(block *types.Block, logs state.Logs) {
                self.mux.Post(core.NewMinedBlockEvent{block})
                self.mux.Post(core.ChainEvent{block, block.Hash(), logs})
                if stat == core.CanonStatTy {
                    self.mux.Post(core.ChainHeadEvent{block})
                }
            }(block, self.current.state.Logs())

            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 <= int64(parent.Time()) {
        tstamp = int64(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(int64(tstamp), int64(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
}