forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

miner: polish

Browse source
This commit is contained in:
rjl493456442 2018-08-13 23:01:21 +08:00
parent dc9c87e21b
commit 2848c14385
3 changed files with 41 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
miner/miner.go
View file

@ -41,7 +41,7 @@ type Backend interface {
// Miner creates blocks and searches for proof-of-work values. // Miner creates blocks and searches for proof-of-work values.
type Miner struct { type Miner struct {
mux *event.TypeMux mux *event.TypeMux
worker *Worker worker *worker
coinbase common.Address coinbase common.Address
eth Backend eth Backend
engine consensus.Engine engine consensus.Engine

70
miner/worker.go
View file

@ -166,17 +166,17 @@ func (env *Env) commitTransactions(mux *event.TypeMux, txs *types.TransactionsBy
} }
} }
// Task contains all information for consensus engine sealing and result submitting. // task contains all information for consensus engine sealing and result submitting.
type Task struct { type task struct {
receipts []*types.Receipt receipts []*types.Receipt
state *state.StateDB state *state.StateDB
block *types.Block block *types.Block
createdAt time.Time createdAt time.Time
} }
// Worker is the main object which takes care of submitting new work to consensus engine // worker is the main object which takes care of submitting new work to consensus engine
// and gathering the sealing result. // and gathering the sealing result.
type Worker struct { type worker struct {
config *params.ChainConfig config *params.ChainConfig
engine consensus.Engine engine consensus.Engine
eth Backend eth Backend
@ -193,8 +193,8 @@ type Worker struct {
// Channels // Channels
newWork chan struct{} newWork chan struct{}
taskCh chan *Task taskCh chan *task
resultCh chan *Task resultCh chan *task
exitCh chan struct{} exitCh chan struct{}
current *Env // An environment for current running cycle. current *Env // An environment for current running cycle.
@ -213,12 +213,12 @@ type Worker struct {
running int32 // The indicator whether the consensus engine is running or not. running int32 // The indicator whether the consensus engine is running or not.
// Test hooks // Test hooks
newTaskHook func(*Task) // Method to call upon receiving a new sealing task newTaskHook func(*task) // Method to call upon receiving a new sealing task
fullTaskInterval func() // Method to call before pushing the full sealing task fullTaskInterval func() // Method to call before pushing the full sealing task
} }
func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux) *Worker { func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux) *worker {
worker := &Worker{ worker := &worker{
config: config, config: config,
engine: engine, engine: engine,
eth: eth, eth: eth,
@ -230,8 +230,8 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend,
chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize), chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize),
chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize), chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize),
newWork: make(chan struct{}, 1), newWork: make(chan struct{}, 1),
taskCh: make(chan *Task), taskCh: make(chan *task),
resultCh: make(chan *Task, resultQueueSize), resultCh: make(chan *task, resultQueueSize),
exitCh: make(chan struct{}), exitCh: make(chan struct{}),
} }
// Subscribe NewTxsEvent for tx pool // Subscribe NewTxsEvent for tx pool
@ -250,21 +250,21 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend,
} }
// setEtherbase sets the etherbase used to initialize the block coinbase field. // setEtherbase sets the etherbase used to initialize the block coinbase field.
func (w *Worker) setEtherbase(addr common.Address) { func (w *worker) setEtherbase(addr common.Address) {
w.mu.Lock() w.mu.Lock()
defer w.mu.Unlock() defer w.mu.Unlock()
w.coinbase = addr w.coinbase = addr
} }
// setExtra sets the content used to initialize the block extra field. // setExtra sets the content used to initialize the block extra field.
func (w *Worker) setExtra(extra []byte) { func (w *worker) setExtra(extra []byte) {
w.mu.Lock() w.mu.Lock()
defer w.mu.Unlock() defer w.mu.Unlock()
w.extra = extra w.extra = extra
} }
// pending returns the pending state and corresponding block. // pending returns the pending state and corresponding block.
func (w *Worker) pending() (*types.Block, *state.StateDB) { func (w *worker) pending() (*types.Block, *state.StateDB) {
// return a snapshot to avoid contention on currentMu mutex // return a snapshot to avoid contention on currentMu mutex
w.snapshotMu.RLock() w.snapshotMu.RLock()
defer w.snapshotMu.RUnlock() defer w.snapshotMu.RUnlock()
@ -275,7 +275,7 @@ func (w *Worker) pending() (*types.Block, *state.StateDB) {
} }
// pendingBlock returns pending block. // pendingBlock returns pending block.
func (w *Worker) pendingBlock() *types.Block { func (w *worker) pendingBlock() *types.Block {
// return a snapshot to avoid contention on currentMu mutex // return a snapshot to avoid contention on currentMu mutex
w.snapshotMu.RLock() w.snapshotMu.RLock()
defer w.snapshotMu.RUnlock() defer w.snapshotMu.RUnlock()
@ -283,24 +283,24 @@ func (w *Worker) pendingBlock() *types.Block {
} }
// start sets the running status as 1 and triggers new work submitting. // start sets the running status as 1 and triggers new work submitting.
func (w *Worker) start() { func (w *worker) start() {
atomic.StoreInt32(&w.running, 1) atomic.StoreInt32(&w.running, 1)
w.newWork <- struct{}{} w.newWork <- struct{}{}
} }
// stop sets the running status as 0. // stop sets the running status as 0.
func (w *Worker) stop() { func (w *worker) stop() {
atomic.StoreInt32(&w.running, 0) atomic.StoreInt32(&w.running, 0)
} }
// isRunning returns an indicator whether worker is running or not. // isRunning returns an indicator whether worker is running or not.
func (w *Worker) isRunning() bool { func (w *worker) isRunning() bool {
return atomic.LoadInt32(&w.running) == 1 return atomic.LoadInt32(&w.running) == 1
} }
// close terminates all background threads maintained by the worker and cleans up buffered channels. // close terminates all background threads maintained by the worker and cleans up buffered channels.
// Note the worker does not support being closed multiple times. // Note the worker does not support being closed multiple times.
func (w *Worker) close() { func (w *worker) close() {
close(w.exitCh) close(w.exitCh)
// Clean up buffered channels // Clean up buffered channels
for empty := false; !empty; { for empty := false; !empty; {
@ -313,7 +313,7 @@ func (w *Worker) close() {
} }
// mainLoop is a standalone goroutine to regenerate the sealing task based on the received event. // mainLoop is a standalone goroutine to regenerate the sealing task based on the received event.
func (w *Worker) mainLoop() { func (w *worker) mainLoop() {
defer w.txsSub.Unsubscribe() defer w.txsSub.Unsubscribe()
defer w.chainHeadSub.Unsubscribe() defer w.chainHeadSub.Unsubscribe()
defer w.chainSideSub.Unsubscribe() defer w.chainSideSub.Unsubscribe()
@ -339,13 +339,17 @@ func (w *Worker) mainLoop() {
// already included in the current mining block. These transactions will // already included in the current mining block. These transactions will
// be automatically eliminated. // be automatically eliminated.
if !w.isRunning() && w.current != nil { if !w.isRunning() && w.current != nil {
w.mu.Lock()
coinbase := w.coinbase
w.mu.Unlock()
txs := make(map[common.Address]types.Transactions) txs := make(map[common.Address]types.Transactions)
for _, tx := range ev.Txs { for _, tx := range ev.Txs {
acc, _ := types.Sender(w.current.signer, tx) acc, _ := types.Sender(w.current.signer, tx)
txs[acc] = append(txs[acc], tx) txs[acc] = append(txs[acc], tx)
} }
txset := types.NewTransactionsByPriceAndNonce(w.current.signer, txs) txset := types.NewTransactionsByPriceAndNonce(w.current.signer, txs)
w.current.commitTransactions(w.mux, txset, w.chain, w.coinbase) w.current.commitTransactions(w.mux, txset, w.chain, coinbase)
w.updateSnapshot() w.updateSnapshot()
} else { } else {
// If we're mining, but nothing is being processed, wake on new transactions // If we're mining, but nothing is being processed, wake on new transactions
@ -368,10 +372,10 @@ func (w *Worker) mainLoop() {
} }
// seal pushes a sealing task to consensus engine and submits the result. // seal pushes a sealing task to consensus engine and submits the result.
func (w *Worker) seal(t *Task, stop <-chan struct{}) { func (w *worker) seal(t *task, stop <-chan struct{}) {
var ( var (
err error err error
res *Task res *task
) )
if t.block, err = w.engine.Seal(w.chain, t.block, stop); t.block != nil { if t.block, err = w.engine.Seal(w.chain, t.block, stop); t.block != nil {
@ -392,7 +396,7 @@ func (w *Worker) seal(t *Task, stop <-chan struct{}) {
// taskLoop is a standalone goroutine to fetch sealing task from the generator and // taskLoop is a standalone goroutine to fetch sealing task from the generator and
// push them to consensus engine. // push them to consensus engine.
func (w *Worker) taskLoop() { func (w *worker) taskLoop() {
var stopCh chan struct{} var stopCh chan struct{}
// interrupt aborts the in-flight sealing task. // interrupt aborts the in-flight sealing task.
@ -420,15 +424,15 @@ func (w *Worker) taskLoop() {
// resultLoop is a standalone goroutine to handle sealing result submitting // resultLoop is a standalone goroutine to handle sealing result submitting
// and flush relative data to the database. // and flush relative data to the database.
func (w *Worker) resultLoop() { func (w *worker) resultLoop() {
for { for {
select { select {
case result := <-w.resultCh: case result := <-w.resultCh:
if result == nil { if result == nil {
continue continue
} }
block := result.block block := result.block
// Update the block hash in all logs since it is now available and not when the // Update the block hash in all logs since it is now available and not when the
// receipt/log of individual transactions were created. // receipt/log of individual transactions were created.
for _, r := range result.receipts { for _, r := range result.receipts {
@ -469,8 +473,8 @@ func (w *Worker) resultLoop() {
} }
} }
// makeCurrent creates an new environment for the current cycle. // makeCurrent creates a new environment for the current cycle.
func (w *Worker) makeCurrent(parent *types.Block, header *types.Header) error { func (w *worker) makeCurrent(parent *types.Block, header *types.Header) error {
state, err := w.chain.StateAt(parent.Root()) state, err := w.chain.StateAt(parent.Root())
if err != nil { if err != nil {
return err return err
@ -501,7 +505,7 @@ func (w *Worker) makeCurrent(parent *types.Block, header *types.Header) error {
} }
// commitUncle adds the given block to uncle block set, returns error if failed to add. // commitUncle adds the given block to uncle block set, returns error if failed to add.
func (w *Worker) commitUncle(env *Env, uncle *types.Header) error { func (w *worker) commitUncle(env *Env, uncle *types.Header) error {
hash := uncle.Hash() hash := uncle.Hash()
if env.uncles.Contains(hash) { if env.uncles.Contains(hash) {
return fmt.Errorf("uncle not unique") return fmt.Errorf("uncle not unique")
@ -518,7 +522,7 @@ func (w *Worker) commitUncle(env *Env, uncle *types.Header) error {
// updateSnapshot updates pending snapshot block and state. // updateSnapshot updates pending snapshot block and state.
// Note this function assumes the current variable is thread safe. // Note this function assumes the current variable is thread safe.
func (w *Worker) updateSnapshot() { func (w *worker) updateSnapshot() {
w.snapshotMu.Lock() w.snapshotMu.Lock()
defer w.snapshotMu.Unlock() defer w.snapshotMu.Unlock()
@ -547,7 +551,7 @@ func (w *Worker) updateSnapshot() {
} }
// commitNewWork generates several new sealing tasks based on the parent block. // commitNewWork generates several new sealing tasks based on the parent block.
func (w *Worker) commitNewWork() { func (w *worker) commitNewWork() {
w.mu.RLock() w.mu.RLock()
defer w.mu.RUnlock() defer w.mu.RUnlock()
@ -649,7 +653,7 @@ func (w *Worker) commitNewWork() {
// take advantage of this part time. // take advantage of this part time.
if w.isRunning() { if w.isRunning() {
select { select {
case w.taskCh <- &Task{receipts: nil, state: emptyState, block: emptyBlock, createdAt: time.Now()}: case w.taskCh <- &task{receipts: nil, state: emptyState, block: emptyBlock, createdAt: time.Now()}:
log.Info("Commit new empty mining work", "number", emptyBlock.Number(), "uncles", len(uncles)) log.Info("Commit new empty mining work", "number", emptyBlock.Number(), "uncles", len(uncles))
case <-w.exitCh: case <-w.exitCh:
log.Info("Worker has exited") log.Info("Worker has exited")
@ -691,7 +695,7 @@ func (w *Worker) commitNewWork() {
} }
select { select {
case w.taskCh <- &Task{receipts: cpy, state: fullState, block: fullBlock, createdAt: time.Now()}: case w.taskCh <- &task{receipts: cpy, state: fullState, block: fullBlock, createdAt: time.Now()}:
w.unconfirmed.Shift(fullBlock.NumberU64() - 1) w.unconfirmed.Shift(fullBlock.NumberU64() - 1)
log.Info("Commit new full mining work", "number", fullBlock.Number(), "txs", env.tcount, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart))) log.Info("Commit new full mining work", "number", fullBlock.Number(), "txs", env.tcount, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
case <-w.exitCh: case <-w.exitCh:

6
miner/worker_test.go
View file

@ -111,7 +111,7 @@ func (b *testWorkerBackend) PostChainEvents(events []interface{}) {
b.chain.PostChainEvents(events, nil) b.chain.PostChainEvents(events, nil)
} }
func newTestWorker(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) (*Worker, *testWorkerBackend) { func newTestWorker(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) (*worker, *testWorkerBackend) {
backend := newTestWorkerBackend(t, chainConfig, engine) backend := newTestWorkerBackend(t, chainConfig, engine)
backend.txPool.AddLocals(pendingTxs) backend.txPool.AddLocals(pendingTxs)
w := newWorker(chainConfig, engine, backend, new(event.TypeMux)) w := newWorker(chainConfig, engine, backend, new(event.TypeMux))
@ -168,7 +168,7 @@ func testEmptyWork(t *testing.T, chainConfig *params.ChainConfig, engine consens
taskIndex int taskIndex int
) )
checkEqual := func(t *testing.T, task *Task, index int) { checkEqual := func(t *testing.T, task *task, index int) {
receiptLen, balance := 0, big.NewInt(0) receiptLen, balance := 0, big.NewInt(0)
if index == 1 { if index == 1 {
receiptLen, balance = 1, big.NewInt(1000) receiptLen, balance = 1, big.NewInt(1000)
@ -181,7 +181,7 @@ func testEmptyWork(t *testing.T, chainConfig *params.ChainConfig, engine consens
} }
} }
w.newTaskHook = func(task *Task) { w.newTaskHook = func(task *task) {
if task.block.NumberU64() == 1 { if task.block.NumberU64() == 1 {
checkEqual(t, task, taskIndex) checkEqual(t, task, taskIndex)
taskIndex += 1 taskIndex += 1