aboutsummaryrefslogtreecommitdiffstats
path: root/miner
diff options
context:
space:
mode:
authorgary rong <garyrong0905@gmail.com>2018-08-06 17:55:44 +0800
committerPéter Szilágyi <peterke@gmail.com>2018-08-06 17:55:44 +0800
commit941018b570aef77f1a47197a88e7a0b533b718f7 (patch)
tree1155be1b294b115d49dc4e04865caa338802ccf9 /miner
parenta72ba5a55be3abb1be85126838ee0b4a9c475be4 (diff)
downloadgo-tangerine-941018b570aef77f1a47197a88e7a0b533b718f7.tar
go-tangerine-941018b570aef77f1a47197a88e7a0b533b718f7.tar.gz
go-tangerine-941018b570aef77f1a47197a88e7a0b533b718f7.tar.bz2
go-tangerine-941018b570aef77f1a47197a88e7a0b533b718f7.tar.lz
go-tangerine-941018b570aef77f1a47197a88e7a0b533b718f7.tar.xz
go-tangerine-941018b570aef77f1a47197a88e7a0b533b718f7.tar.zst
go-tangerine-941018b570aef77f1a47197a88e7a0b533b718f7.zip
miner: seperate state, receipts for different mining work (#17323)
Diffstat (limited to 'miner')
-rw-r--r--miner/agent.go29
-rw-r--r--miner/worker.go102
2 files changed, 67 insertions, 64 deletions
diff --git a/miner/agent.go b/miner/agent.go
index 95d835bd7..e922ea153 100644
--- a/miner/agent.go
+++ b/miner/agent.go
@@ -27,10 +27,10 @@ import (
type CpuAgent struct {
mu sync.Mutex
- workCh chan *Work
+ taskCh chan *Package
+ returnCh chan<- *Package
stop chan struct{}
quitCurrentOp chan struct{}
- returnCh chan<- *Result
chain consensus.ChainReader
engine consensus.Engine
@@ -43,13 +43,17 @@ func NewCpuAgent(chain consensus.ChainReader, engine consensus.Engine) *CpuAgent
chain: chain,
engine: engine,
stop: make(chan struct{}, 1),
- workCh: make(chan *Work, 1),
+ taskCh: make(chan *Package, 1),
}
return agent
}
-func (self *CpuAgent) Work() chan<- *Work { return self.workCh }
-func (self *CpuAgent) SetReturnCh(ch chan<- *Result) { self.returnCh = ch }
+func (self *CpuAgent) AssignTask(p *Package) {
+ if atomic.LoadInt32(&self.started) == 1 {
+ self.taskCh <- p
+ }
+}
+func (self *CpuAgent) DeliverTo(ch chan<- *Package) { self.returnCh = ch }
func (self *CpuAgent) Start() {
if !atomic.CompareAndSwapInt32(&self.started, 0, 1) {
@@ -67,7 +71,7 @@ done:
// Empty work channel
for {
select {
- case <-self.workCh:
+ case <-self.taskCh:
default:
break done
}
@@ -78,13 +82,13 @@ func (self *CpuAgent) update() {
out:
for {
select {
- case work := <-self.workCh:
+ case p := <-self.taskCh:
self.mu.Lock()
if self.quitCurrentOp != nil {
close(self.quitCurrentOp)
}
self.quitCurrentOp = make(chan struct{})
- go self.mine(work, self.quitCurrentOp)
+ go self.mine(p, self.quitCurrentOp)
self.mu.Unlock()
case <-self.stop:
self.mu.Lock()
@@ -98,10 +102,11 @@ out:
}
}
-func (self *CpuAgent) mine(work *Work, stop <-chan struct{}) {
- if result, err := self.engine.Seal(self.chain, work.Block, stop); result != nil {
- log.Info("Successfully sealed new block", "number", result.Number(), "hash", result.Hash())
- self.returnCh <- &Result{work, result}
+func (self *CpuAgent) mine(p *Package, stop <-chan struct{}) {
+ var err error
+ if p.Block, err = self.engine.Seal(self.chain, p.Block, stop); p.Block != nil {
+ log.Info("Successfully sealed new block", "number", p.Block.Number(), "hash", p.Block.Hash())
+ self.returnCh <- p
} else {
if err != nil {
log.Warn("Block sealing failed", "err", err)
diff --git a/miner/worker.go b/miner/worker.go
index f1194fa18..ae695f019 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -51,17 +51,16 @@ const (
chainSideChanSize = 10
)
-// Agent can register themself with the worker
+// Agent can register themselves with the worker
type Agent interface {
- Work() chan<- *Work
- SetReturnCh(chan<- *Result)
+ AssignTask(*Package)
+ DeliverTo(chan<- *Package)
Start()
Stop()
}
-// Work is the workers current environment and holds
-// all of the current state information
-type Work struct {
+// Env is the workers current environment and holds all of the current state information.
+type Env struct {
config *params.ChainConfig
signer types.Signer
@@ -72,8 +71,6 @@ type Work struct {
tcount int // tx count in cycle
gasPool *core.GasPool // available gas used to pack transactions
- Block *types.Block // the new block
-
header *types.Header
txs []*types.Transaction
receipts []*types.Receipt
@@ -81,9 +78,11 @@ type Work struct {
createdAt time.Time
}
-type Result struct {
- Work *Work
- Block *types.Block
+// Package contains all information for consensus engine sealing and result submitting.
+type Package struct {
+ Receipts []*types.Receipt
+ State *state.StateDB
+ Block *types.Block
}
// worker is the main object which takes care of applying messages to the new state
@@ -103,7 +102,7 @@ type worker struct {
chainSideSub event.Subscription
agents map[Agent]struct{}
- recv chan *Result
+ recv chan *Package
eth Backend
chain *core.BlockChain
@@ -114,7 +113,7 @@ type worker struct {
extra []byte
currentMu sync.Mutex
- current *Work
+ current *Env
snapshotMu sync.RWMutex
snapshotBlock *types.Block
@@ -126,7 +125,6 @@ type worker struct {
unconfirmed *unconfirmedBlocks // set of locally mined blocks pending canonicalness confirmations
// atomic status counters
- atWork int32 // The number of in-flight consensus engine work.
running int32 // The indicator whether the consensus engine is running or not.
}
@@ -140,7 +138,7 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend,
chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize),
chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize),
chainDb: eth.ChainDb(),
- recv: make(chan *Result, resultQueueSize),
+ recv: make(chan *Package, resultQueueSize),
chain: eth.BlockChain(),
proc: eth.BlockChain().Validator(),
possibleUncles: make(map[common.Hash]*types.Block),
@@ -203,7 +201,6 @@ func (self *worker) stop() {
for agent := range self.agents {
agent.Stop()
}
- atomic.StoreInt32(&self.atWork, 0)
}
func (self *worker) isRunning() bool {
@@ -214,7 +211,7 @@ func (self *worker) register(agent Agent) {
self.mu.Lock()
defer self.mu.Unlock()
self.agents[agent] = struct{}{}
- agent.SetReturnCh(self.recv)
+ agent.DeliverTo(self.recv)
if self.isRunning() {
agent.Start()
}
@@ -284,26 +281,24 @@ func (self *worker) update() {
func (self *worker) wait() {
for {
for result := range self.recv {
- atomic.AddInt32(&self.atWork, -1)
if result == nil {
continue
}
block := result.Block
- work := result.Work
// Update the block hash in all logs since it is now available and not when the
// receipt/log of individual transactions were created.
- for _, r := range work.receipts {
+ for _, r := range result.Receipts {
for _, l := range r.Logs {
l.BlockHash = block.Hash()
}
}
- for _, log := range work.state.Logs() {
+ for _, log := range result.State.Logs() {
log.BlockHash = block.Hash()
}
self.currentMu.Lock()
- stat, err := self.chain.WriteBlockWithState(block, work.receipts, work.state)
+ stat, err := self.chain.WriteBlockWithState(block, result.Receipts, result.State)
self.currentMu.Unlock()
if err != nil {
log.Error("Failed writing block to chain", "err", err)
@@ -313,7 +308,7 @@ func (self *worker) wait() {
self.mux.Post(core.NewMinedBlockEvent{Block: block})
var (
events []interface{}
- logs = work.state.Logs()
+ logs = result.State.Logs()
)
events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs})
if stat == core.CanonStatTy {
@@ -328,12 +323,9 @@ func (self *worker) wait() {
}
// push sends a new work task to currently live miner agents.
-func (self *worker) push(work *Work) {
+func (self *worker) push(p *Package) {
for agent := range self.agents {
- atomic.AddInt32(&self.atWork, 1)
- if ch := agent.Work(); ch != nil {
- ch <- work
- }
+ agent.AssignTask(p)
}
}
@@ -343,7 +335,7 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
if err != nil {
return err
}
- work := &Work{
+ env := &Env{
config: self.config,
signer: types.NewEIP155Signer(self.config.ChainID),
state: state,
@@ -357,15 +349,15 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
// when 08 is processed ancestors contain 07 (quick block)
for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) {
for _, uncle := range ancestor.Uncles() {
- work.family.Add(uncle.Hash())
+ env.family.Add(uncle.Hash())
}
- work.family.Add(ancestor.Hash())
- work.ancestors.Add(ancestor.Hash())
+ env.family.Add(ancestor.Hash())
+ env.ancestors.Add(ancestor.Hash())
}
// Keep track of transactions which return errors so they can be removed
- work.tcount = 0
- self.current = work
+ env.tcount = 0
+ self.current = env
return nil
}
@@ -431,9 +423,9 @@ func (self *worker) commitNewWork() {
return
}
// Create the current work task and check any fork transitions needed
- work := self.current
+ env := self.current
if self.config.DAOForkSupport && self.config.DAOForkBlock != nil && self.config.DAOForkBlock.Cmp(header.Number) == 0 {
- misc.ApplyDAOHardFork(work.state)
+ misc.ApplyDAOHardFork(env.state)
}
// compute uncles for the new block.
@@ -445,7 +437,7 @@ func (self *worker) commitNewWork() {
if len(uncles) == 2 {
break
}
- if err := self.commitUncle(work, uncle.Header()); err != nil {
+ if err := self.commitUncle(env, uncle.Header()); err != nil {
log.Trace("Bad uncle found and will be removed", "hash", hash)
log.Trace(fmt.Sprint(uncle))
@@ -459,17 +451,23 @@ func (self *worker) commitNewWork() {
delete(self.possibleUncles, hash)
}
+ var (
+ emptyBlock *types.Block
+ fullBlock *types.Block
+ )
+
// Create an empty block based on temporary copied state for sealing in advance without waiting block
// execution finished.
- if work.Block, err = self.engine.Finalize(self.chain, header, work.state.Copy(), nil, uncles, nil); err != nil {
+ emptyState := env.state.Copy()
+ if emptyBlock, err = self.engine.Finalize(self.chain, header, emptyState, nil, uncles, nil); err != nil {
log.Error("Failed to finalize block for temporary sealing", "err", err)
} else {
// Push empty work in advance without applying pending transaction.
// The reason is transactions execution can cost a lot and sealer need to
// take advantage of this part time.
if self.isRunning() {
- log.Info("Commit new empty mining work", "number", work.Block.Number(), "uncles", len(uncles))
- self.push(work)
+ log.Info("Commit new empty mining work", "number", emptyBlock.Number(), "uncles", len(uncles))
+ self.push(&Package{nil, emptyState, emptyBlock})
}
}
@@ -480,34 +478,34 @@ func (self *worker) commitNewWork() {
return
}
txs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending)
- work.commitTransactions(self.mux, txs, self.chain, self.coinbase)
+ env.commitTransactions(self.mux, txs, self.chain, self.coinbase)
// Create the full block to seal with the consensus engine
- if work.Block, err = self.engine.Finalize(self.chain, header, work.state, work.txs, uncles, work.receipts); err != nil {
+ if fullBlock, err = self.engine.Finalize(self.chain, header, env.state, env.txs, uncles, env.receipts); err != nil {
log.Error("Failed to finalize block for sealing", "err", err)
return
}
// We only care about logging if we're actually mining.
if self.isRunning() {
- log.Info("Commit new full mining work", "number", work.Block.Number(), "txs", work.tcount, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
- self.unconfirmed.Shift(work.Block.NumberU64() - 1)
- self.push(work)
+ log.Info("Commit new full mining work", "number", fullBlock.Number(), "txs", env.tcount, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
+ self.unconfirmed.Shift(fullBlock.NumberU64() - 1)
+ self.push(&Package{env.receipts, env.state, fullBlock})
}
self.updateSnapshot()
}
-func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
+func (self *worker) commitUncle(env *Env, uncle *types.Header) error {
hash := uncle.Hash()
- if work.uncles.Contains(hash) {
+ if env.uncles.Contains(hash) {
return fmt.Errorf("uncle not unique")
}
- if !work.ancestors.Contains(uncle.ParentHash) {
+ if !env.ancestors.Contains(uncle.ParentHash) {
return fmt.Errorf("uncle's parent unknown (%x)", uncle.ParentHash[0:4])
}
- if work.family.Contains(hash) {
+ if env.family.Contains(hash) {
return fmt.Errorf("uncle already in family (%x)", hash)
}
- work.uncles.Add(uncle.Hash())
+ env.uncles.Add(uncle.Hash())
return nil
}
@@ -533,7 +531,7 @@ func (self *worker) updateSnapshot() {
self.snapshotState = self.current.state.Copy()
}
-func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) {
+func (env *Env) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) {
if env.gasPool == nil {
env.gasPool = new(core.GasPool).AddGas(env.header.GasLimit)
}
@@ -618,7 +616,7 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
}
}
-func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, coinbase common.Address, gp *core.GasPool) (error, []*types.Log) {
+func (env *Env) commitTransaction(tx *types.Transaction, bc *core.BlockChain, coinbase common.Address, gp *core.GasPool) (error, []*types.Log) {
snap := env.state.Snapshot()
receipt, _, err := core.ApplyTransaction(env.config, bc, &coinbase, gp, env.state, env.header, tx, &env.header.GasUsed, vm.Config{})