consensus, miner: stale block supporting

consensus/consensus.go

@@ -110,4 +110,7 @@ type PoW interface {
 
 	// Hashrate returns the current mining hashrate of a PoW consensus engine.
 	Hashrate() float64
+
+	// StaleBlocks returns a channel that returns all stale but acceptable blocks.
+	StaleBlocks() <-chan *types.Block
 }

consensus/ethash/ethash.go

@@ -394,6 +394,8 @@ const (
 	ModeFullFake
 )
 
+const staleChannelSize = 10 // The size of stale block channel.
+
 // Config are the configuration parameters of the ethash.
 type Config struct {
 	CacheDir       string
@@ -444,12 +446,13 @@ type Ethash struct {
 	hashrate metrics.Meter // Meter tracking the average hashrate
 
 	// Remote sealer related fields
-	workCh       chan *types.Block // Notification channel to push new work to remote sealer
-	resultCh     chan *types.Block // Channel used by mining threads to return result
-	fetchWorkCh  chan *sealWork    // Channel used for remote sealer to fetch mining work
-	submitWorkCh chan *mineResult  // Channel used for remote sealer to submit their mining result
-	fetchRateCh  chan chan uint64  // Channel used to gather submitted hash rate for local or remote sealer.
-	submitRateCh chan *hashrate    // Channel used for remote sealer to submit their mining hashrate
+	workCh        chan *types.Block // Notification channel to push new work to remote sealer
+	resultCh      chan *types.Block // Channel used by mining threads to return result
+	fetchWorkCh   chan *sealWork    // Channel used for remote sealer to fetch mining work
+	submitWorkCh  chan *mineResult  // Channel used for remote sealer to submit their mining result
+	fetchRateCh   chan chan uint64  // Channel used to gather submitted hash rate for local or remote sealer.
+	submitRateCh  chan *hashrate    // Channel used for remote sealer to submit their mining hashrate
+	staleResultCh chan *types.Block // Channel used for passing back stale but acceptable solutions.
 
 	// The fields below are hooks for testing
 	shared    *Ethash       // Shared PoW verifier to avoid cache regeneration
@@ -459,6 +462,9 @@ type Ethash struct {
 	lock      sync.Mutex      // Ensures thread safety for the in-memory caches and mining fields
 	closeOnce sync.Once       // Ensures exit channel will not be closed twice.
 	exitCh    chan chan error // Notification channel to exiting backend threads
+
+	// Test relative attributes
+	disableRemoteVerify bool
 }
 
 // New creates a full sized ethash PoW scheme and starts a background thread for
@@ -476,18 +482,19 @@ func New(config Config, notify []string) *Ethash {
 		log.Info("Disk storage enabled for ethash DAGs", "dir", config.DatasetDir, "count", config.DatasetsOnDisk)
 	}
 	ethash := &Ethash{
-		config:       config,
-		caches:       newlru("cache", config.CachesInMem, newCache),
-		datasets:     newlru("dataset", config.DatasetsInMem, newDataset),
-		update:       make(chan struct{}),
-		hashrate:     metrics.NewMeter(),
-		workCh:       make(chan *types.Block),
-		resultCh:     make(chan *types.Block),
-		fetchWorkCh:  make(chan *sealWork),
-		submitWorkCh: make(chan *mineResult),
-		fetchRateCh:  make(chan chan uint64),
-		submitRateCh: make(chan *hashrate),
-		exitCh:       make(chan chan error),
+		config:        config,
+		caches:        newlru("cache", config.CachesInMem, newCache),
+		datasets:      newlru("dataset", config.DatasetsInMem, newDataset),
+		update:        make(chan struct{}),
+		hashrate:      metrics.NewMeter(),
+		workCh:        make(chan *types.Block),
+		resultCh:      make(chan *types.Block),
+		fetchWorkCh:   make(chan *sealWork),
+		submitWorkCh:  make(chan *mineResult),
+		fetchRateCh:   make(chan chan uint64),
+		submitRateCh:  make(chan *hashrate),
+		staleResultCh: make(chan *types.Block, staleChannelSize),
+		exitCh:        make(chan chan error),
 	}
 	go ethash.remote(notify)
 	return ethash
@@ -497,18 +504,19 @@ func New(config Config, notify []string) *Ethash {
 // purposes.
 func NewTester(notify []string) *Ethash {
 	ethash := &Ethash{
-		config:       Config{PowMode: ModeTest},
-		caches:       newlru("cache", 1, newCache),
-		datasets:     newlru("dataset", 1, newDataset),
-		update:       make(chan struct{}),
-		hashrate:     metrics.NewMeter(),
-		workCh:       make(chan *types.Block),
-		resultCh:     make(chan *types.Block),
-		fetchWorkCh:  make(chan *sealWork),
-		submitWorkCh: make(chan *mineResult),
-		fetchRateCh:  make(chan chan uint64),
-		submitRateCh: make(chan *hashrate),
-		exitCh:       make(chan chan error),
+		config:        Config{PowMode: ModeTest},
+		caches:        newlru("cache", 1, newCache),
+		datasets:      newlru("dataset", 1, newDataset),
+		update:        make(chan struct{}),
+		hashrate:      metrics.NewMeter(),
+		workCh:        make(chan *types.Block),
+		resultCh:      make(chan *types.Block),
+		fetchWorkCh:   make(chan *sealWork),
+		submitWorkCh:  make(chan *mineResult),
+		fetchRateCh:   make(chan chan uint64),
+		submitRateCh:  make(chan *hashrate),
+		staleResultCh: make(chan *types.Block, staleChannelSize),
+		exitCh:        make(chan chan error),
 	}
 	go ethash.remote(notify)
 	return ethash

consensus/ethash/sealer.go

@@ -35,6 +35,11 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
+const (
+	// staleThreshold is the maximum distance of the acceptable stale but valid ethash solution.
+	staleThreshold = 7
+)
+
 var (
 	errNoMiningWork      = errors.New("no mining work available yet")
 	errInvalidSealResult = errors.New("invalid or stale proof-of-work solution")
@@ -226,11 +231,15 @@ func (ethash *Ethash) remote(notify []string) {
 	// submitWork verifies the submitted pow solution, returning
 	// whether the solution was accepted or not (not can be both a bad pow as well as
 	// any other error, like no pending work or stale mining result).
-	submitWork := func(nonce types.BlockNonce, mixDigest common.Hash, hash common.Hash) bool {
+	submitWork := func(nonce types.BlockNonce, mixDigest common.Hash, sealhash common.Hash) bool {
 		// Make sure the work submitted is present
-		block := works[hash]
+		block := works[sealhash]
 		if block == nil {
-			log.Info("Work submitted but none pending", "hash", hash)
+			log.Info("Work submitted but none pending", "sealhash", sealhash, "current block number", currentBlock.NumberU64())
+			return false
+		}
+		if currentBlock == nil {
+			log.Error("have pending packages but no pending block, please file an issue", "sealhash", sealhash)
 			return false
 		}
 		// Verify the correctness of submitted result.
@@ -239,26 +248,46 @@ func (ethash *Ethash) remote(notify []string) {
 		header.MixDigest = mixDigest
 
 		start := time.Now()
-		if err := ethash.verifySeal(nil, header, true); err != nil {
-			log.Warn("Invalid proof-of-work submitted", "hash", hash, "elapsed", time.Since(start), "err", err)
-			return false
+		if !ethash.disableRemoteVerify {
+			if err := ethash.verifySeal(nil, header, true); err != nil {
+				log.Warn("Invalid proof-of-work submitted", "sealhash", sealhash, "elapsed", time.Since(start), "err", err)
+				return false
+			}
 		}
 		// Make sure the result channel is created.
 		if ethash.resultCh == nil {
 			log.Warn("Ethash result channel is empty, submitted mining result is rejected")
 			return false
 		}
-		log.Trace("Verified correct proof-of-work", "hash", hash, "elapsed", time.Since(start))
+		log.Trace("Verified correct proof-of-work", "sealhash", sealhash, "elapsed", time.Since(start))
 
 		// Solutions seems to be valid, return to the miner and notify acceptance.
-		select {
-		case ethash.resultCh <- block.WithSeal(header):
-			delete(works, hash)
+		submitBlock := block.WithSeal(header)
+
+		// If submitted block is the mining block of latest round, try to push back through result channel.
+		if submitBlock.ParentHash() == currentBlock.ParentHash() {
+			select {
+			case ethash.resultCh <- submitBlock:
+				log.Info("Work submitted is fresh", "number", submitBlock.NumberU64(), "sealhash", sealhash, "hash", submitBlock.Hash())
+				return true
+			default:
+				// Worker has already received a result for latest mining round, submit it as
+				// a stale block.
+				ethash.staleResultCh <- submitBlock
+				log.Info("Work submitted is stale but acceptable", "number", submitBlock.NumberU64(), "sealhash", sealhash, "hash", submitBlock.Hash())
+				return true
+			}
+		} else if submitBlock.NumberU64()+staleThreshold > currentBlock.NumberU64() {
+			// The submitted block is stale but acceptable, it can be converted to a uncle finally.
+			ethash.staleResultCh <- submitBlock
+			log.Info("Work submitted is stale but acceptable", "number", submitBlock.NumberU64(), "sealhash", sealhash, "hash", submitBlock.Hash())
 			return true
-		default:
-			log.Info("Work submitted is stale", "hash", hash)
+		} else {
+			// The submitted block is too old to accept, drop it.
+			log.Info("The submitted block is too old", "number", submitBlock.NumberU64(), "sealhash", sealhash, "hash", submitBlock)
 			return false
 		}
+		return false
 	}
 
 	ticker := time.NewTicker(5 * time.Second)
@@ -267,10 +296,6 @@ func (ethash *Ethash) remote(notify []string) {
 	for {
 		select {
 		case block := <-ethash.workCh:
-			if currentBlock != nil && block.ParentHash() != currentBlock.ParentHash() {
-				// Start new round mining, throw out all previous work.
-				works = make(map[common.Hash]*types.Block)
-			}
 			// Update current work with new received block.
 			// Note same work can be past twice, happens when changing CPU threads.
 			makeWork(block)
@@ -315,6 +340,14 @@ func (ethash *Ethash) remote(notify []string) {
 					delete(rates, id)
 				}
 			}
+			// Clear stale pending blocks
+			if currentBlock != nil {
+				for hash, block := range works {
+					if block.NumberU64()+staleThreshold <= currentBlock.NumberU64() {
+						delete(works, hash)
+					}
+				}
+			}
 
 		case errc := <-ethash.exitCh:
 			// Exit remote loop if ethash is closed and return relevant error.
@@ -324,3 +357,8 @@ func (ethash *Ethash) remote(notify []string) {
 		}
 	}
 }
+
+// StaleBlocks returns a channel that returns all stale but acceptable blocks.
+func (ethash *Ethash) StaleBlocks() <-chan *types.Block {
+	return ethash.staleResultCh
+}

consensus/ethash/sealer_test.go

@@ -66,7 +66,7 @@ func TestRemoteNotify(t *testing.T) {
 	}
 }
 
-// Tests that pushing work packages fast to the miner doesn't cause any daa race
+// Tests that pushing work packages fast to the miner doesn't cause any data race
 // issues in the notifications.
 func TestRemoteMultiNotify(t *testing.T) {
 	// Start a simple webserver to capture notifications
@@ -113,3 +113,89 @@ func TestRemoteMultiNotify(t *testing.T) {
 		}
 	}
 }
+
+// Tests whether stale solutions are correctly processed.
+func TestStaleSubmission(t *testing.T) {
+	ethash := NewTester(nil)
+	defer ethash.Close()
+	ethash.disableRemoteVerify = true
+	api := &API{ethash}
+
+	fakeNonce, fakeDigest := types.BlockNonce{0x01, 0x02, 0x03}, common.HexToHash("deadbeef")
+
+	testcases := []struct {
+		headers     []*types.Header
+		resCh       chan *types.Block
+		submitIndex int
+		submitRes   bool
+	}{
+		// Case1: submit solution for the latest mining package
+		{
+			[]*types.Header{
+				{ParentHash: common.BytesToHash([]byte{0xa}), Number: big.NewInt(1), Difficulty: big.NewInt(100)},
+			},
+			ethash.resultCh,
+			0,
+			true,
+		},
+		// Case2: submit solution for the previous package but have same parent.
+		{
+			[]*types.Header{
+				{ParentHash: common.BytesToHash([]byte{0xb}), Number: big.NewInt(2), Difficulty: big.NewInt(100)},
+				{ParentHash: common.BytesToHash([]byte{0xb}), Number: big.NewInt(2), Difficulty: big.NewInt(101)},
+			},
+			ethash.resultCh,
+			0,
+			true,
+		},
+		// Case3: submit stale but acceptable solution
+		{
+			[]*types.Header{
+				{ParentHash: common.BytesToHash([]byte{0xc}), Number: big.NewInt(3), Difficulty: big.NewInt(100)},
+				{ParentHash: common.BytesToHash([]byte{0xd}), Number: big.NewInt(9), Difficulty: big.NewInt(100)},
+			},
+			ethash.staleResultCh,
+			0,
+			true,
+		},
+		// Case4: submit very old solution
+		{
+			[]*types.Header{
+				{ParentHash: common.BytesToHash([]byte{0xe}), Number: big.NewInt(10), Difficulty: big.NewInt(100)},
+				{ParentHash: common.BytesToHash([]byte{0xf}), Number: big.NewInt(17), Difficulty: big.NewInt(100)},
+			},
+			nil,
+			0,
+			false,
+		},
+	}
+
+	for id, c := range testcases {
+		for _, h := range c.headers {
+			ethash.Seal(nil, types.NewBlockWithHeader(h), nil)
+		}
+		stop := make(chan struct{})
+		go func(stop chan struct{}) {
+			select {
+			case res := <-c.resCh:
+				if res.Header().Nonce != fakeNonce {
+					t.Errorf("case %d block nonce mismatch, want %s, get %s", id+1, fakeNonce, res.Header().Nonce)
+				}
+				if res.Header().MixDigest != fakeDigest {
+					t.Errorf("case %d block digest mismatch, want %s, get %s", id+1, fakeDigest, res.Header().MixDigest)
+				}
+				if res.Header().Difficulty.Uint64() != c.headers[c.submitIndex].Difficulty.Uint64() {
+					t.Errorf("case %d block difficulty mismatch, want %d, get %d", id+1, c.headers[c.submitIndex].Difficulty, res.Header().Difficulty)
+				}
+			case <-time.NewTimer(time.Second).C:
+				t.Errorf("case %d fetch ethash result timeout", id+1)
+			case <-stop:
+				return
+			}
+		}(stop)
+		if res := api.SubmitWork(fakeNonce, ethash.SealHash(c.headers[c.submitIndex]), fakeDigest); res != c.submitRes {
+			t.Errorf("case %d submit result mismatch, want %t, get %t", id+1, c.submitRes, res)
+		}
+		close(stop)
+	}
+}

eth/api_tracer.go

@@ -294,7 +294,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl
 				failed = err
 				break
 			}
-			// Reference the trie twice, once for us, once for the trancer
+			// Reference the trie twice, once for us, once for the tracer
 			database.TrieDB().Reference(root, common.Hash{})
 			if number >= origin {
 				database.TrieDB().Reference(root, common.Hash{})

miner/worker.go

@@ -98,6 +98,7 @@ type task struct {
 	receipts  []*types.Receipt
 	state     *state.StateDB
 	block     *types.Block
+	logs      []*types.Log
 	createdAt time.Time
 }
 
@@ -209,6 +210,10 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend,
 	go worker.resultLoop()
 	go worker.taskLoop()
 
+	// Start the stale result loop if we are running the ethash engine.
+	if pow, ok := worker.engine.(consensus.PoW); ok {
+		go worker.staleResultLoop(pow)
+	}
 	// Submit first work to initialize pending state.
 	worker.startCh <- struct{}{}
 
@@ -480,7 +485,6 @@ func (w *worker) seal(t *task, stop <-chan struct{}) {
 	// A previous sealing action will be canceled by subsequent actions,
 	// however, remote miner may submit a result based on the cancelled task.
 	// So we should only submit the pending state corresponding to the seal result.
-	// TODO(rjl493456442) Replace the seal-wait logic structure
 	w.pendingMu.Lock()
 	w.pendingTasks[w.engine.SealHash(t.block.Header())] = t
 	w.pendingMu.Unlock()
@@ -488,18 +492,34 @@ func (w *worker) seal(t *task, stop <-chan struct{}) {
 	if block, err := w.engine.Seal(w.chain, t.block, stop); block != nil {
 		sealhash := w.engine.SealHash(block.Header())
 		w.pendingMu.RLock()
-		task, exist := w.pendingTasks[sealhash]
+		t, exist := w.pendingTasks[sealhash]
 		w.pendingMu.RUnlock()
 		if !exist {
 			log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", block.Hash())
 			return
 		}
-		// Assemble sealing result
-		task.block = block
+		// Different block could share same sealhash, deep copy here to prevent write-write conflict.
+		var (
+			receipts = make([]*types.Receipt, len(t.receipts))
+			logs     []*types.Log
+		)
+		for i, receipt := range t.receipts {
+			receipts[i] = new(types.Receipt)
+			*receipts[i] = *receipt
+			// Update the block hash in all logs since it is now available and not when the
+			// receipt/log of individual transactions were created.
+			for _, log := range receipt.Logs {
+				log.BlockHash = block.Hash()
+				logs = append(logs, log)
+			}
+		}
+		// The reason we don't deep copy state here is: state commit is single-threaded and
+		// the same state can avoid repeated data writes and trie operations.
+		cpy := &task{receipts: receipts, state: t.state, block: block, logs: logs, createdAt: t.createdAt}
 		log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealhash, "hash", block.Hash(),
-			"elapsed", common.PrettyDuration(time.Since(task.createdAt)))
+			"elapsed", common.PrettyDuration(time.Since(cpy.createdAt)))
 		select {
-		case w.resultCh <- task:
+		case w.resultCh <- cpy:
 		case <-w.exitCh:
 		}
 	} else if err != nil {
@@ -544,6 +564,53 @@ func (w *worker) taskLoop() {
 	}
 }
 
+// staleResultLoop is the loop to fetch all stale but acceptable PoW solutions
+// and submit them to the blockchain.
+// Stale block is a unique product of the pow algorithm. Since the mining pool
+// wll accept solutions for the past 7 block heights as the uncle block or canonical
+// block, so that ethash engine will forward the stale solutions to the worker.
+func (w *worker) staleResultLoop(poW consensus.PoW) {
+	for {
+		select {
+		case block := <-poW.StaleBlocks():
+			sealhash := w.engine.SealHash(block.Header())
+			w.pendingMu.Lock()
+			t, exist := w.pendingTasks[sealhash]
+			w.pendingMu.Unlock()
+			if !exist {
+				log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", block.Hash())
+				continue
+			}
+			// Different block could share same sealhash, deep copy here to prevent write-write conflict.
+			var (
+				receipts = make([]*types.Receipt, len(t.receipts))
+				logs     []*types.Log
+			)
+			for i, receipt := range t.receipts {
+				receipts[i] = new(types.Receipt)
+				*receipts[i] = *receipt
+				// Update the block hash in all logs since it is now available and not when the
+				// receipt/log of individual transactions were created.
+				for _, log := range receipt.Logs {
+					log.BlockHash = block.Hash()
+					logs = append(logs, log)
+				}
+			}
+			// The reason we don't deep copy state here is: state commit is single-threaded and
+			// the same state can avoid repeated data writes and trie operations.
+			cpy := &task{receipts: receipts, state: t.state, block: block, logs: logs, createdAt: t.createdAt}
+			log.Info("Successfully sealed stale block", "number", block.Number(), "sealhash", sealhash, "hash", block.Hash(),
+				"elapsed", common.PrettyDuration(time.Since(cpy.createdAt)))
+			select {
+			case w.resultCh <- cpy:
+			case <-w.exitCh:
+			}
+		case <-w.exitCh:
+			return
+		}
+	}
+}
+
 // resultLoop is a standalone goroutine to handle sealing result submitting
 // and flush relative data to the database.
 func (w *worker) resultLoop() {
@@ -559,17 +626,10 @@ func (w *worker) resultLoop() {
 			if w.chain.HasBlock(block.Hash(), block.NumberU64()) {
 				continue
 			}
-			// 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 result.receipts {
-				for _, l := range r.Logs {
-					l.BlockHash = block.Hash()
-				}
-			}
-			for _, log := range result.state.Logs() {
-				log.BlockHash = block.Hash()
-			}
 			// Commit block and state to database.
+			// Note same state changes could be committed many times because two different valid blocks
+			// sharing same state.
+			// Except for the first commit, the remaining commits will not write more state data to the database.
 			stat, err := w.chain.WriteBlockWithState(block, result.receipts, result.state)
 			if err != nil {
 				log.Error("Failed writing block to chain", "err", err)
@@ -577,18 +637,15 @@ func (w *worker) resultLoop() {
 			}
 			// Broadcast the block and announce chain insertion event
 			w.mux.Post(core.NewMinedBlockEvent{Block: block})
-			var (
-				events []interface{}
-				logs   = result.state.Logs()
-			)
+			var events []interface{}
 			switch stat {
 			case core.CanonStatTy:
-				events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs})
+				events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: result.logs})
 				events = append(events, core.ChainHeadEvent{Block: block})
 			case core.SideStatTy:
 				events = append(events, core.ChainSideEvent{Block: block})
 			}
-			w.chain.PostChainEvents(events, logs)
+			w.chain.PostChainEvents(events, result.logs)
 
 			// Insert the block into the set of pending ones to resultLoop for confirmations
 			w.unconfirmed.Insert(block.NumberU64(), block.Hash())