miner: move agent logic to worker (#17351)

* miner: move agent logic to worker

* miner: polish

* core: persist block before reorg

core/blockchain.go

@@ -899,9 +899,7 @@ func (bc *BlockChain) WriteBlockWithState(block *types.Block, receipts []*types.
 	if err := bc.hc.WriteTd(block.Hash(), block.NumberU64(), externTd); err != nil {
 		return NonStatTy, err
 	}
-	// Write other block data using a batch.
+	rawdb.WriteBlock(bc.db, block)
-	batch := bc.db.NewBatch()
-	rawdb.WriteBlock(batch, block)
 
 	root, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()))
 	if err != nil {
@@ -955,6 +953,9 @@ func (bc *BlockChain) WriteBlockWithState(block *types.Block, receipts []*types.
 			}
 		}
 	}
+
+	// Write other block data using a batch.
+	batch := bc.db.NewBatch()
 	rawdb.WriteReceipts(batch, block.Hash(), block.NumberU64(), receipts)
 
 	// If the total difficulty is higher than our known, add it to the canonical chain

miner/agent.go

@@ -1,116 +0,0 @@
-// Copyright 2015 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-package miner
-
-import (
-	"sync"
-	"sync/atomic"
-
-	"github.com/ethereum/go-ethereum/consensus"
-	"github.com/ethereum/go-ethereum/log"
-)
-
-type CpuAgent struct {
-	mu sync.Mutex
-
-	taskCh        chan *Package
-	returnCh      chan<- *Package
-	stop          chan struct{}
-	quitCurrentOp chan struct{}
-
-	chain  consensus.ChainReader
-	engine consensus.Engine
-
-	started int32 // started indicates whether the agent is currently started
-}
-
-func NewCpuAgent(chain consensus.ChainReader, engine consensus.Engine) *CpuAgent {
-	agent := &CpuAgent{
-		chain:  chain,
-		engine: engine,
-		stop:   make(chan struct{}, 1),
-		taskCh: make(chan *Package, 1),
-	}
-	return agent
-}
-
-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) {
-		return // agent already started
-	}
-	go self.update()
-}
-
-func (self *CpuAgent) Stop() {
-	if !atomic.CompareAndSwapInt32(&self.started, 1, 0) {
-		return // agent already stopped
-	}
-	self.stop <- struct{}{}
-done:
-	// Empty work channel
-	for {
-		select {
-		case <-self.taskCh:
-		default:
-			break done
-		}
-	}
-}
-
-func (self *CpuAgent) update() {
-out:
-	for {
-		select {
-		case p := <-self.taskCh:
-			self.mu.Lock()
-			if self.quitCurrentOp != nil {
-				close(self.quitCurrentOp)
-			}
-			self.quitCurrentOp = make(chan struct{})
-			go self.mine(p, self.quitCurrentOp)
-			self.mu.Unlock()
-		case <-self.stop:
-			self.mu.Lock()
-			if self.quitCurrentOp != nil {
-				close(self.quitCurrentOp)
-				self.quitCurrentOp = nil
-			}
-			self.mu.Unlock()
-			break out
-		}
-	}
-}
-
-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)
-		}
-		self.returnCh <- nil
-	}
-}

miner/miner.go

@@ -21,14 +21,12 @@ import (
 	"fmt"
 	"sync/atomic"
 
-	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus"
 	"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/eth/downloader"
-	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
@@ -36,10 +34,8 @@ import (
 
 // Backend wraps all methods required for mining.
 type Backend interface {
-	AccountManager() *accounts.Manager
 	BlockChain() *core.BlockChain
 	TxPool() *core.TxPool
-	ChainDb() ethdb.Database
 }
 
 // Miner creates blocks and searches for proof-of-work values.
@@ -49,6 +45,7 @@ type Miner struct {
 	coinbase common.Address
 	eth      Backend
 	engine   consensus.Engine
+	exitCh   chan struct{}
 
 	canStart    int32 // can start indicates whether we can start the mining operation
 	shouldStart int32 // should start indicates whether we should start after sync
@@ -59,10 +56,10 @@ func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine con
 		eth:      eth,
 		mux:      mux,
 		engine:   engine,
+		exitCh:   make(chan struct{}),
 		worker:   newWorker(config, engine, eth, mux),
 		canStart: 1,
 	}
-	miner.Register(NewCpuAgent(eth.BlockChain(), engine))
 	go miner.update()
 
 	return miner
@@ -74,28 +71,35 @@ func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine con
 // and halt your mining operation for as long as the DOS continues.
 func (self *Miner) update() {
 	events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
-out:
+	defer events.Unsubscribe()
-	for ev := range events.Chan() {
-		switch ev.Data.(type) {
-		case downloader.StartEvent:
-			atomic.StoreInt32(&self.canStart, 0)
-			if self.Mining() {
-				self.Stop()
-				atomic.StoreInt32(&self.shouldStart, 1)
-				log.Info("Mining aborted due to sync")
-			}
-		case downloader.DoneEvent, downloader.FailedEvent:
-			shouldStart := atomic.LoadInt32(&self.shouldStart) == 1
 
-			atomic.StoreInt32(&self.canStart, 1)
+	for {
-			atomic.StoreInt32(&self.shouldStart, 0)
+		select {
-			if shouldStart {
+		case ev := <-events.Chan():
-				self.Start(self.coinbase)
+			if ev == nil {
+				return
 			}
-			// unsubscribe. we're only interested in this event once
+			switch ev.Data.(type) {
-			events.Unsubscribe()
+			case downloader.StartEvent:
-			// stop immediately and ignore all further pending events
+				atomic.StoreInt32(&self.canStart, 0)
-			break out
+				if self.Mining() {
+					self.Stop()
+					atomic.StoreInt32(&self.shouldStart, 1)
+					log.Info("Mining aborted due to sync")
+				}
+			case downloader.DoneEvent, downloader.FailedEvent:
+				shouldStart := atomic.LoadInt32(&self.shouldStart) == 1
+
+				atomic.StoreInt32(&self.canStart, 1)
+				atomic.StoreInt32(&self.shouldStart, 0)
+				if shouldStart {
+					self.Start(self.coinbase)
+				}
+				// stop immediately and ignore all further pending events
+				return
+			}
+		case <-self.exitCh:
+			return
 		}
 	}
 }
@@ -109,7 +113,6 @@ func (self *Miner) Start(coinbase common.Address) {
 		return
 	}
 	self.worker.start()
-	self.worker.commitNewWork()
 }
 
 func (self *Miner) Stop() {
@@ -117,12 +120,9 @@ func (self *Miner) Stop() {
 	atomic.StoreInt32(&self.shouldStart, 0)
 }
 
-func (self *Miner) Register(agent Agent) {
+func (self *Miner) Close() {
-	self.worker.register(agent)
+	self.worker.close()
-}
+	close(self.exitCh)
-
-func (self *Miner) Unregister(agent Agent) {
-	self.worker.unregister(agent)
 }
 
 func (self *Miner) Mining() bool {

miner/worker_test.go

@@ -0,0 +1,212 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package miner
+
+import (
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus"
+	"github.com/ethereum/go-ethereum/consensus/clique"
+	"github.com/ethereum/go-ethereum/consensus/ethash"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/core/vm"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/params"
+)
+
+var (
+	// Test chain configurations
+	testTxPoolConfig  core.TxPoolConfig
+	ethashChainConfig *params.ChainConfig
+	cliqueChainConfig *params.ChainConfig
+
+	// Test accounts
+	testBankKey, _  = crypto.GenerateKey()
+	testBankAddress = crypto.PubkeyToAddress(testBankKey.PublicKey)
+	testBankFunds   = big.NewInt(1000000000000000000)
+
+	acc1Key, _ = crypto.GenerateKey()
+	acc1Addr   = crypto.PubkeyToAddress(acc1Key.PublicKey)
+
+	// Test transactions
+	pendingTxs []*types.Transaction
+	newTxs     []*types.Transaction
+)
+
+func init() {
+	testTxPoolConfig = core.DefaultTxPoolConfig
+	testTxPoolConfig.Journal = ""
+	ethashChainConfig = params.TestChainConfig
+	cliqueChainConfig = params.TestChainConfig
+	cliqueChainConfig.Clique = &params.CliqueConfig{
+		Period: 1,
+		Epoch:  30000,
+	}
+	tx1, _ := types.SignTx(types.NewTransaction(0, acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+	pendingTxs = append(pendingTxs, tx1)
+	tx2, _ := types.SignTx(types.NewTransaction(1, acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+	newTxs = append(newTxs, tx2)
+}
+
+// testWorkerBackend implements worker.Backend interfaces and wraps all information needed during the testing.
+type testWorkerBackend struct {
+	db         ethdb.Database
+	txPool     *core.TxPool
+	chain      *core.BlockChain
+	testTxFeed event.Feed
+}
+
+func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) *testWorkerBackend {
+	var (
+		db    = ethdb.NewMemDatabase()
+		gspec = core.Genesis{
+			Config: chainConfig,
+			Alloc:  core.GenesisAlloc{testBankAddress: {Balance: testBankFunds}},
+		}
+	)
+
+	switch engine.(type) {
+	case *clique.Clique:
+		gspec.ExtraData = make([]byte, 32+common.AddressLength+65)
+		copy(gspec.ExtraData[32:], testBankAddress[:])
+	case *ethash.Ethash:
+	default:
+		t.Fatal("unexpect consensus engine type")
+	}
+	gspec.MustCommit(db)
+
+	chain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{})
+	txpool := core.NewTxPool(testTxPoolConfig, chainConfig, chain)
+
+	return &testWorkerBackend{
+		db:     db,
+		chain:  chain,
+		txPool: txpool,
+	}
+}
+
+func (b *testWorkerBackend) BlockChain() *core.BlockChain { return b.chain }
+func (b *testWorkerBackend) TxPool() *core.TxPool         { return b.txPool }
+func (b *testWorkerBackend) PostChainEvents(events []interface{}) {
+	b.chain.PostChainEvents(events, nil)
+}
+
+func newTestWorker(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) (*worker, *testWorkerBackend) {
+	backend := newTestWorkerBackend(t, chainConfig, engine)
+	backend.txPool.AddLocals(pendingTxs)
+	w := newWorker(chainConfig, engine, backend, new(event.TypeMux))
+	w.setEtherbase(testBankAddress)
+	return w, backend
+}
+
+func TestPendingStateAndBlockEthash(t *testing.T) {
+	testPendingStateAndBlock(t, ethashChainConfig, ethash.NewFaker())
+}
+func TestPendingStateAndBlockClique(t *testing.T) {
+	testPendingStateAndBlock(t, cliqueChainConfig, clique.New(cliqueChainConfig.Clique, ethdb.NewMemDatabase()))
+}
+
+func testPendingStateAndBlock(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) {
+	defer engine.Close()
+
+	w, b := newTestWorker(t, chainConfig, engine)
+	defer w.close()
+
+	// Ensure snapshot has been updated.
+	time.Sleep(100 * time.Millisecond)
+	block, state := w.pending()
+	if block.NumberU64() != 1 {
+		t.Errorf("block number mismatch, has %d, want %d", block.NumberU64(), 1)
+	}
+	if balance := state.GetBalance(acc1Addr); balance.Cmp(big.NewInt(1000)) != 0 {
+		t.Errorf("account balance mismatch, has %d, want %d", balance, 1000)
+	}
+	b.txPool.AddLocals(newTxs)
+	// Ensure the new tx events has been processed
+	time.Sleep(100 * time.Millisecond)
+	block, state = w.pending()
+	if balance := state.GetBalance(acc1Addr); balance.Cmp(big.NewInt(2000)) != 0 {
+		t.Errorf("account balance mismatch, has %d, want %d", balance, 2000)
+	}
+}
+
+func TestEmptyWorkEthash(t *testing.T) {
+	testEmptyWork(t, ethashChainConfig, ethash.NewFaker())
+}
+func TestEmptyWorkClique(t *testing.T) {
+	testEmptyWork(t, cliqueChainConfig, clique.New(cliqueChainConfig.Clique, ethdb.NewMemDatabase()))
+}
+
+func testEmptyWork(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) {
+	defer engine.Close()
+
+	w, _ := newTestWorker(t, chainConfig, engine)
+	defer w.close()
+
+	var (
+		taskCh    = make(chan struct{}, 2)
+		taskIndex int
+	)
+
+	checkEqual := func(t *testing.T, task *task, index int) {
+		receiptLen, balance := 0, big.NewInt(0)
+		if index == 1 {
+			receiptLen, balance = 1, big.NewInt(1000)
+		}
+		if len(task.receipts) != receiptLen {
+			t.Errorf("receipt number mismatch has %d, want %d", len(task.receipts), receiptLen)
+		}
+		if task.state.GetBalance(acc1Addr).Cmp(balance) != 0 {
+			t.Errorf("account balance mismatch has %d, want %d", task.state.GetBalance(acc1Addr), balance)
+		}
+	}
+
+	w.newTaskHook = func(task *task) {
+		if task.block.NumberU64() == 1 {
+			checkEqual(t, task, taskIndex)
+			taskIndex += 1
+			taskCh <- struct{}{}
+		}
+	}
+	w.fullTaskInterval = func() {
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	// Ensure worker has finished initialization
+	for {
+		b := w.pendingBlock()
+		if b != nil && b.NumberU64() == 1 {
+			break
+		}
+	}
+
+	w.start()
+	for i := 0; i < 2; i += 1 {
+		to := time.NewTimer(time.Second)
+		select {
+		case <-taskCh:
+		case <-to.C:
+			t.Error("new task timeout")
+		}
+	}
+}