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

Merge pull request #1587 from obscuren/miner-race

Browse Source 
miner: fixed worker race condition
pull/1583/head
Jeffrey Wilcke 9 years ago
parent 56219a5e7a d8590d9790
commit 1f4ac6b05a
1 changed files with 44 additions and 43 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. 87
      miner/worker.go

87
miner/worker.go
Unescape View File

@ -263,8 +263,9 @@ func (self *worker) wait() {
continue continue
} }
block := result.Block block := result.Block
work := result.Work
self.current.state.Sync() work.state.Sync()
if self.fullValidation { if self.fullValidation {
if _, err := self.chain.InsertChain(types.Blocks{block}); err != nil { if _, err := self.chain.InsertChain(types.Blocks{block}); err != nil {
glog.V(logger.Error).Infoln("mining err", err) glog.V(logger.Error).Infoln("mining err", err)
@ -292,7 +293,7 @@ func (self *worker) wait() {
// This puts transactions in a extra db for rpc // This puts transactions in a extra db for rpc
core.PutTransactions(self.extraDb, block, block.Transactions()) core.PutTransactions(self.extraDb, block, block.Transactions())
// store the receipts // store the receipts
core.PutReceipts(self.extraDb, self.current.receipts) core.PutReceipts(self.extraDb, work.receipts)
} }
// broadcast before waiting for validation // broadcast before waiting for validation
@ -303,7 +304,7 @@ func (self *worker) wait() {
self.mux.Post(core.ChainHeadEvent{block}) self.mux.Post(core.ChainHeadEvent{block})
self.mux.Post(logs) self.mux.Post(logs)
} }
}(block, self.current.state.Logs()) }(block, work.state.Logs())
} }
// check staleness and display confirmation // check staleness and display confirmation
@ -313,7 +314,7 @@ func (self *worker) wait() {
stale = "stale " stale = "stale "
} else { } else {
confirm = "Wait 5 blocks for confirmation" confirm = "Wait 5 blocks for confirmation"
self.current.localMinedBlocks = newLocalMinedBlock(block.Number().Uint64(), self.current.localMinedBlocks) work.localMinedBlocks = newLocalMinedBlock(block.Number().Uint64(), work.localMinedBlocks)
} }
glog.V(logger.Info).Infof("🔨 Mined %sblock (#%v / %x). %s", stale, block.Number(), block.Hash().Bytes()[:4], confirm) glog.V(logger.Info).Infof("🔨 Mined %sblock (#%v / %x). %s", stale, block.Number(), block.Hash().Bytes()[:4], confirm)
@ -322,9 +323,9 @@ func (self *worker) wait() {
} }
} }
func (self *worker) push() { func (self *worker) push(work *Work) {
if atomic.LoadInt32(&self.mining) == 1 { if atomic.LoadInt32(&self.mining) == 1 {
if core.Canary(self.current.state) { if core.Canary(work.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("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") glog.Infoln("You turn back and abort mining")
return return
@ -335,7 +336,7 @@ func (self *worker) push() {
atomic.AddInt32(&self.atWork, 1) atomic.AddInt32(&self.atWork, 1)
if agent.Work() != nil { if agent.Work() != nil {
agent.Work() <- self.current agent.Work() <- work
} }
} }
} }
@ -344,7 +345,7 @@ func (self *worker) push() {
// makeCurrent creates a new environment for the current cycle. // makeCurrent creates a new environment for the current cycle.
func (self *worker) makeCurrent(parent *types.Block, header *types.Header) { func (self *worker) makeCurrent(parent *types.Block, header *types.Header) {
state := state.New(parent.Root(), self.eth.StateDb()) state := state.New(parent.Root(), self.eth.StateDb())
current := &Work{ work := &Work{
state: state, state: state,
ancestors: set.New(), ancestors: set.New(),
family: set.New(), family: set.New(),
@ -357,23 +358,23 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) {
// when 08 is processed ancestors contain 07 (quick block) // when 08 is processed ancestors contain 07 (quick block)
for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) { for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) {
for _, uncle := range ancestor.Uncles() { for _, uncle := range ancestor.Uncles() {
current.family.Add(uncle.Hash()) work.family.Add(uncle.Hash())
} }
current.family.Add(ancestor.Hash()) work.family.Add(ancestor.Hash())
current.ancestors.Add(ancestor.Hash()) work.ancestors.Add(ancestor.Hash())
} }
accounts, _ := self.eth.AccountManager().Accounts() accounts, _ := self.eth.AccountManager().Accounts()
// Keep track of transactions which return errors so they can be removed // Keep track of transactions which return errors so they can be removed
current.remove = set.New() work.remove = set.New()
current.tcount = 0 work.tcount = 0
current.ignoredTransactors = set.New() work.ignoredTransactors = set.New()
current.lowGasTransactors = set.New() work.lowGasTransactors = set.New()
current.ownedAccounts = accountAddressesSet(accounts) work.ownedAccounts = accountAddressesSet(accounts)
if self.current != nil { if self.current != nil {
current.localMinedBlocks = self.current.localMinedBlocks work.localMinedBlocks = self.current.localMinedBlocks
} }
self.current = current self.current = work
} }
func (w *worker) setGasPrice(p *big.Int) { func (w *worker) setGasPrice(p *big.Int) {
@ -387,13 +388,13 @@ func (w *worker) setGasPrice(p *big.Int) {
w.mux.Post(core.GasPriceChanged{w.gasPrice}) w.mux.Post(core.GasPriceChanged{w.gasPrice})
} }
func (self *worker) isBlockLocallyMined(deepBlockNum uint64) bool { func (self *worker) isBlockLocallyMined(current *Work, deepBlockNum uint64) bool {
//Did this instance mine a block at {deepBlockNum} ? //Did this instance mine a block at {deepBlockNum} ?
var isLocal = false var isLocal = false
for idx, blockNum := range self.current.localMinedBlocks.ints { for idx, blockNum := range current.localMinedBlocks.ints {
if deepBlockNum == blockNum { if deepBlockNum == blockNum {
isLocal = true isLocal = true
self.current.localMinedBlocks.ints[idx] = 0 //prevent showing duplicate logs current.localMinedBlocks.ints[idx] = 0 //prevent showing duplicate logs
break break
} }
} }
@ -407,12 +408,12 @@ func (self *worker) isBlockLocallyMined(deepBlockNum uint64) bool {
return block != nil && block.Coinbase() == self.coinbase return block != nil && block.Coinbase() == self.coinbase
} }
func (self *worker) logLocalMinedBlocks(previous *Work) { func (self *worker) logLocalMinedBlocks(current, previous *Work) {
if previous != nil && self.current.localMinedBlocks != nil { if previous != nil && current.localMinedBlocks != nil {
nextBlockNum := self.current.Block.NumberU64() nextBlockNum := current.Block.NumberU64()
for checkBlockNum := previous.Block.NumberU64(); checkBlockNum < nextBlockNum; checkBlockNum++ { for checkBlockNum := previous.Block.NumberU64(); checkBlockNum < nextBlockNum; checkBlockNum++ {
inspectBlockNum := checkBlockNum - miningLogAtDepth inspectBlockNum := checkBlockNum - miningLogAtDepth
if self.isBlockLocallyMined(inspectBlockNum) { if self.isBlockLocallyMined(current, inspectBlockNum) {
glog.V(logger.Info).Infof("🔨 🔗 Mined %d blocks back: block #%v", miningLogAtDepth, inspectBlockNum) glog.V(logger.Info).Infof("🔨 🔗 Mined %d blocks back: block #%v", miningLogAtDepth, inspectBlockNum)
} }
} }
@ -454,14 +455,14 @@ func (self *worker) commitNewWork() {
previous := self.current previous := self.current
self.makeCurrent(parent, header) self.makeCurrent(parent, header)
current := self.current work := self.current
// commit transactions for this run. // commit transactions for this run.
transactions := self.eth.TxPool().GetTransactions() transactions := self.eth.TxPool().GetTransactions()
sort.Sort(types.TxByNonce{transactions}) sort.Sort(types.TxByNonce{transactions})
current.coinbase.SetGasLimit(header.GasLimit) work.coinbase.SetGasLimit(header.GasLimit)
current.commitTransactions(transactions, self.gasPrice, self.proc) work.commitTransactions(transactions, self.gasPrice, self.proc)
self.eth.TxPool().RemoveTransactions(current.lowGasTxs) self.eth.TxPool().RemoveTransactions(work.lowGasTxs)
// compute uncles for the new block. // compute uncles for the new block.
var ( var (
@ -472,7 +473,7 @@ func (self *worker) commitNewWork() {
if len(uncles) == 2 { if len(uncles) == 2 {
break break
} }
if err := self.commitUncle(uncle.Header()); err != nil { if err := self.commitUncle(work, uncle.Header()); err != nil {
if glog.V(logger.Ridiculousness) { 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).Infof("Bad uncle found and will be removed (%x)\n", hash[:4])
glog.V(logger.Detail).Infoln(uncle) glog.V(logger.Detail).Infoln(uncle)
@ -489,36 +490,36 @@ func (self *worker) commitNewWork() {
if atomic.LoadInt32(&self.mining) == 1 { if atomic.LoadInt32(&self.mining) == 1 {
// commit state root after all state transitions. // commit state root after all state transitions.
core.AccumulateRewards(self.current.state, header, uncles) core.AccumulateRewards(work.state, header, uncles)
current.state.SyncObjects() work.state.SyncObjects()
header.Root = current.state.Root() header.Root = work.state.Root()
} }
// create the new block whose nonce will be mined. // create the new block whose nonce will be mined.
current.Block = types.NewBlock(header, current.txs, uncles, current.receipts) work.Block = types.NewBlock(header, work.txs, uncles, work.receipts)
self.current.Block.Td = new(big.Int).Set(core.CalcTD(self.current.Block, self.chain.GetBlock(self.current.Block.ParentHash()))) work.Block.Td = new(big.Int).Set(core.CalcTD(work.Block, self.chain.GetBlock(work.Block.ParentHash())))
// We only care about logging if we're actually mining. // We only care about logging if we're actually mining.
if atomic.LoadInt32(&self.mining) == 1 { 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", work.Block.Number(), work.tcount, len(uncles), time.Since(tstart))
self.logLocalMinedBlocks(previous) self.logLocalMinedBlocks(work, previous)
} }
self.push() self.push(work)
} }
func (self *worker) commitUncle(uncle *types.Header) error { func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
hash := uncle.Hash() hash := uncle.Hash()
if self.current.uncles.Has(hash) { if work.uncles.Has(hash) {
return core.UncleError("Uncle not unique") return core.UncleError("Uncle not unique")
} }
if !self.current.ancestors.Has(uncle.ParentHash) { if !work.ancestors.Has(uncle.ParentHash) {
return core.UncleError(fmt.Sprintf("Uncle's parent unknown (%x)", uncle.ParentHash[0:4])) return core.UncleError(fmt.Sprintf("Uncle's parent unknown (%x)", uncle.ParentHash[0:4]))
} }
if self.current.family.Has(hash) { if work.family.Has(hash) {
return core.UncleError(fmt.Sprintf("Uncle already in family (%x)", hash)) return core.UncleError(fmt.Sprintf("Uncle already in family (%x)", hash))
} }
self.current.uncles.Add(uncle.Hash()) work.uncles.Add(uncle.Hash())
return nil return nil
} }

Write Preview
Loading…
Cancel
Save