mirror
/
go-ethereum
mirror of https://github.com/ethereum/go-ethereum
2
0
Fork 1
Code Issues Releases Wiki Activity

miner: smart mining

Browse Source 
Work is now handled and carried over multiple sessions. Previously one
session only was assumed, potentially resulting in invalid (outdated)
work

* Larger work / result queue
* Full validation option
pull/1469/head
Jeffrey Wilcke 9 years ago
parent cecc9cdd2f
commit e870e61bc9
3 changed files with 149 additions and 105 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 24
      miner/agent.go
  2. 96
      miner/remote_agent.go
  3. 134
      miner/worker.go

24
miner/agent.go
Unescape View File

@ -20,7 +20,6 @@ import (
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/pow"
@ -29,10 +28,10 @@ import (
type CpuAgent struct {
mu sync.Mutex
workCh chan *types.Block
workCh chan *Work
quit chan struct{}
quitCurrentOp chan struct{}
returnCh chan<- *types.Block
returnCh chan<- *Result
index int
pow pow.PoW
@ -47,9 +46,9 @@ func NewCpuAgent(index int, pow pow.PoW) *CpuAgent {
return miner
}
func (self *CpuAgent) Work() chan<- *types.Block { return self.workCh }
func (self *CpuAgent) Pow() pow.PoW { return self.pow }
func (self *CpuAgent) SetReturnCh(ch chan<- *types.Block) { self.returnCh = ch }
func (self *CpuAgent) Work() chan<- *Work { return self.workCh }
func (self *CpuAgent) Pow() pow.PoW { return self.pow }
func (self *CpuAgent) SetReturnCh(ch chan<- *Result) { self.returnCh = ch }
func (self *CpuAgent) Stop() {
self.mu.Lock()
@ -65,7 +64,7 @@ func (self *CpuAgent) Start() {
self.quit = make(chan struct{})
// creating current op ch makes sure we're not closing a nil ch
// later on
self.workCh = make(chan *types.Block, 1)
self.workCh = make(chan *Work, 1)
go self.update()
}
@ -74,13 +73,13 @@ func (self *CpuAgent) update() {
out:
for {
select {
case block := <-self.workCh:
case work := <-self.workCh:
self.mu.Lock()
if self.quitCurrentOp != nil {
close(self.quitCurrentOp)
}
self.quitCurrentOp = make(chan struct{})
go self.mine(block, self.quitCurrentOp)
go self.mine(work, self.quitCurrentOp)
self.mu.Unlock()
case <-self.quit:
self.mu.Lock()
@ -106,13 +105,14 @@ done:
}
}
func (self *CpuAgent) mine(block *types.Block, stop <-chan struct{}) {
func (self *CpuAgent) mine(work *Work, stop <-chan struct{}) {
glog.V(logger.Debug).Infof("(re)started agent[%d]. mining...\n", self.index)
// Mine
nonce, mixDigest := self.pow.Search(block, stop)
nonce, mixDigest := self.pow.Search(work.Block, stop)
if nonce != 0 {
self.returnCh <- block.WithMiningResult(nonce, common.BytesToHash(mixDigest))
block := work.Block.WithMiningResult(nonce, common.BytesToHash(mixDigest))
self.returnCh <- &Result{work, block}
} else {
self.returnCh <- nil
}

96
miner/remote_agent.go
Unescape View File

@ -18,39 +18,44 @@ package miner
import (
"math/big"
"sync"
"time"
"github.com/ethereum/ethash"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
)
type RemoteAgent struct {
work *types.Block
currentWork *types.Block
mu sync.Mutex
quit chan struct{}
workCh chan *types.Block
returnCh chan<- *types.Block
workCh chan *Work
returnCh chan<- *Result
currentWork *Work
work map[common.Hash]*Work
}
func NewRemoteAgent() *RemoteAgent {
agent := &RemoteAgent{}
agent := &RemoteAgent{work: make(map[common.Hash]*Work)}
return agent
}
func (a *RemoteAgent) Work() chan<- *types.Block {
func (a *RemoteAgent) Work() chan<- *Work {
return a.workCh
}
func (a *RemoteAgent) SetReturnCh(returnCh chan<- *types.Block) {
func (a *RemoteAgent) SetReturnCh(returnCh chan<- *Result) {
a.returnCh = returnCh
}
func (a *RemoteAgent) Start() {
a.quit = make(chan struct{})
a.workCh = make(chan *types.Block, 1)
go a.run()
a.workCh = make(chan *Work, 1)
go a.maintainLoop()
}
func (a *RemoteAgent) Stop() {
@ -60,47 +65,72 @@ func (a *RemoteAgent) Stop() {
func (a *RemoteAgent) GetHashRate() int64 { return 0 }
func (a *RemoteAgent) run() {
out:
for {
select {
case <-a.quit:
break out
case work := <-a.workCh:
a.work = work
}
}
}
func (a *RemoteAgent) GetWork() [3]string {
a.mu.Lock()
defer a.mu.Unlock()
var res [3]string
if a.work != nil {
a.currentWork = a.work
if a.currentWork != nil {
block := a.currentWork.Block
res[0] = a.work.HashNoNonce().Hex()
seedHash, _ := ethash.GetSeedHash(a.currentWork.NumberU64())
res[0] = block.HashNoNonce().Hex()
seedHash, _ := ethash.GetSeedHash(block.NumberU64())
res[1] = common.BytesToHash(seedHash).Hex()
// Calculate the "target" to be returned to the external miner
n := big.NewInt(1)
n.Lsh(n, 255)
n.Div(n, a.work.Difficulty())
n.Div(n, block.Difficulty())
n.Lsh(n, 1)
res[2] = common.BytesToHash(n.Bytes()).Hex()
a.work[block.HashNoNonce()] = a.currentWork
}
return res
}
func (a *RemoteAgent) SubmitWork(nonce uint64, mixDigest, seedHash common.Hash) bool {
// Return true or false, but does not indicate if the PoW was correct
// Returns true or false, but does not indicate if the PoW was correct
func (a *RemoteAgent) SubmitWork(nonce uint64, mixDigest, hash common.Hash) bool {
a.mu.Lock()
defer a.mu.Unlock()
// Make sure the work submitted is present
if a.work[hash] != nil {
block := a.work[hash].Block.WithMiningResult(nonce, mixDigest)
a.returnCh <- &Result{a.work[hash], block}
delete(a.work, hash)
// Make sure the external miner was working on the right hash
if a.currentWork != nil && a.work != nil {
a.returnCh <- a.currentWork.WithMiningResult(nonce, mixDigest)
//a.returnCh <- Work{a.currentWork.Number().Uint64(), nonce, mixDigest.Bytes(), seedHash.Bytes()}
return true
} else {
glog.V(logger.Info).Infof("Work was submitted for %x but no pending work found\n", hash)
}
return false
}
func (a *RemoteAgent) maintainLoop() {
ticker := time.Tick(5 * time.Second)
out:
for {
select {
case <-a.quit:
break out
case work := <-a.workCh:
a.mu.Lock()
a.currentWork = work
a.mu.Unlock()
case <-ticker:
// cleanup
a.mu.Lock()
for hash, work := range a.work {
if time.Since(work.createdAt) > 7*(12*time.Second) {
delete(a.work, hash)
}
}
a.mu.Unlock()
}
}
}

134
miner/worker.go
Unescape View File

@ -38,25 +38,20 @@ import (
var jsonlogger = logger.NewJsonLogger()
// Work holds the current work
type Work struct {
Number uint64
Nonce uint64
MixDigest []byte
SeedHash []byte
}
const (
resultQueueSize = 10
miningLogAtDepth = 5
)
// Agent can register themself with the worker
type Agent interface {
Work() chan<- *types.Block
SetReturnCh(chan<- *types.Block)
Work() chan<- *Work
SetReturnCh(chan<- *Result)
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)
@ -64,7 +59,7 @@ type uint64RingBuffer struct {
// environment is the workers current environment and holds
// all of the current state information
type environment struct {
type Work 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)
@ -78,11 +73,18 @@ type environment struct {
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
Block *types.Block // the new block
header *types.Header
txs []*types.Transaction
receipts []*types.Receipt
createdAt time.Time
}
type Result struct {
Work *Work
Block *types.Block
}
// worker is the main object which takes care of applying messages to the new state
@ -90,7 +92,7 @@ type worker struct {
mu sync.Mutex
agents []Agent
recv chan *types.Block
recv chan *Result
mux *event.TypeMux
quit chan struct{}
pow pow.PoW
@ -105,7 +107,7 @@ type worker struct {
extra []byte
currentMu sync.Mutex
current *environment
current *Work
uncleMu sync.Mutex
possibleUncles map[common.Hash]*types.Block
@ -116,6 +118,8 @@ type worker struct {
// atomic status counters
mining int32
atWork int32
fullValidation bool
}
func newWorker(coinbase common.Address, eth core.Backend) *worker {
@ -123,7 +127,7 @@ func newWorker(coinbase common.Address, eth core.Backend) *worker {
eth: eth,
mux: eth.EventMux(),
extraDb: eth.ExtraDb(),
recv: make(chan *types.Block),
recv: make(chan *Result, resultQueueSize),
gasPrice: new(big.Int),
chain: eth.ChainManager(),
proc: eth.BlockProcessor(),
@ -131,6 +135,7 @@ func newWorker(coinbase common.Address, eth core.Backend) *worker {
coinbase: coinbase,
txQueue: make(map[common.Hash]*types.Transaction),
quit: make(chan struct{}),
fullValidation: false,
}
go worker.update()
go worker.wait()
@ -163,7 +168,7 @@ func (self *worker) pendingBlock() *types.Block {
self.current.receipts,
)
}
return self.current.block
return self.current.Block
}
func (self *worker) start() {
@ -250,34 +255,53 @@ func newLocalMinedBlock(blockNumber uint64, prevMinedBlocks *uint64RingBuffer) (
func (self *worker) wait() {
for {
for block := range self.recv {
for result := range self.recv {
atomic.AddInt32(&self.atWork, -1)
if block == nil {
if result == nil {
continue
}
block := result.Block
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
}
if self.fullValidation {
if _, err := self.chain.InsertChain(types.Blocks{block}); err != nil {
glog.V(logger.Error).Infoln("mining err", err)
continue
}
go self.mux.Post(core.NewMinedBlockEvent{block})
} else {
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)
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)
}
// 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})
self.mux.Post(logs)
}
}(block, self.current.state.Logs())
}
// check staleness and display confirmation
@ -289,19 +313,8 @@ func (self *worker) wait() {
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})
self.mux.Post(logs)
}
}(block, self.current.state.Logs())
self.commitNewWork()
}
}
@ -320,7 +333,7 @@ func (self *worker) push() {
atomic.AddInt32(&self.atWork, 1)
if agent.Work() != nil {
agent.Work() <- self.current.block
agent.Work() <- self.current
}
}
}
@ -329,13 +342,14 @@ func (self *worker) push() {
// 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{
current := &Work{
state: state,
ancestors: set.New(),
family: set.New(),
uncles: set.New(),
header: header,
coinbase: state.GetOrNewStateObject(self.coinbase),
createdAt: time.Now(),
}
// when 08 is processed ancestors contain 07 (quick block)
@ -391,10 +405,10 @@ func (self *worker) isBlockLocallyMined(deepBlockNum uint64) bool {
return block != nil && block.Coinbase() == self.coinbase
}
func (self *worker) logLocalMinedBlocks(previous *environment) {
func (self *worker) logLocalMinedBlocks(previous *Work) {
if previous != nil && self.current.localMinedBlocks != nil {
nextBlockNum := self.current.block.NumberU64()
for checkBlockNum := previous.block.NumberU64(); checkBlockNum < nextBlockNum; checkBlockNum++ {
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)
@ -480,12 +494,12 @@ func (self *worker) commitNewWork() {
}
// 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())))
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))
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)
}
@ -507,7 +521,7 @@ func (self *worker) commitUncle(uncle *types.Header) error {
return nil
}
func (env *environment) commitTransactions(transactions types.Transactions, gasPrice *big.Int, proc *core.BlockProcessor) {
func (env *Work) 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()
@ -565,7 +579,7 @@ func (env *environment) commitTransactions(transactions types.Transactions, gasP
}
}
func (env *environment) commitTransaction(tx *types.Transaction, proc *core.BlockProcessor) error {
func (env *Work) 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 {

Write Preview
Loading…
Cancel
Save