miner: commit state which is relative with sealing result

consensus/clique/clique.go

@@ -135,14 +135,14 @@ var (
 // backing account.
 type SignerFn func(accounts.Account, []byte) ([]byte, error)
 
-// sigHash returns the hash which is used as input for the proof-of-authority
+// SigHash returns the hash which is used as input for the proof-of-authority
 // signing. It is the hash of the entire header apart from the 65 byte signature
 // contained at the end of the extra data.
 //
 // Note, the method requires the extra data to be at least 65 bytes, otherwise it
 // panics. This is done to avoid accidentally using both forms (signature present
 // or not), which could be abused to produce different hashes for the same header.
-func sigHash(header *types.Header) (hash common.Hash) {
+func SigHash(header *types.Header) (hash common.Hash) {
 	hasher := sha3.NewKeccak256()
 
 	rlp.Encode(hasher, []interface{}{
@@ -180,7 +180,7 @@ func ecrecover(header *types.Header, sigcache *lru.ARCCache) (common.Address, er
 	signature := header.Extra[len(header.Extra)-extraSeal:]
 
 	// Recover the public key and the Ethereum address
-	pubkey, err := crypto.Ecrecover(sigHash(header).Bytes(), signature)
+	pubkey, err := crypto.Ecrecover(SigHash(header).Bytes(), signature)
 	if err != nil {
 		return common.Address{}, err
 	}
@@ -643,7 +643,7 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch
 	case <-time.After(delay):
 	}
 	// Sign all the things!
-	sighash, err := signFn(accounts.Account{Address: signer}, sigHash(header).Bytes())
+	sighash, err := signFn(accounts.Account{Address: signer}, SigHash(header).Bytes())
 	if err != nil {
 		return nil, err
 	}

consensus/clique/snapshot_test.go

@@ -56,7 +56,7 @@ func (ap *testerAccountPool) sign(header *types.Header, signer string) {
 		ap.accounts[signer], _ = crypto.GenerateKey()
 	}
 	// Sign the header and embed the signature in extra data
-	sig, _ := crypto.Sign(sigHash(header).Bytes(), ap.accounts[signer])
+	sig, _ := crypto.Sign(SigHash(header).Bytes(), ap.accounts[signer])
 	copy(header.Extra[len(header.Extra)-65:], sig)
 }
 

miner/worker.go

@@ -27,6 +27,7 @@ import (
 	mapset "github.com/deckarep/golang-set"
 	"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/misc"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/state"
@@ -150,6 +151,9 @@ type worker struct {
 	coinbase common.Address
 	extra    []byte
 
+	pendingMu    sync.RWMutex
+	pendingTasks map[common.Hash]*task
+
 	snapshotMu    sync.RWMutex // The lock used to protect the block snapshot and state snapshot
 	snapshotBlock *types.Block
 	snapshotState *state.StateDB
@@ -174,6 +178,7 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend,
 		chain:              eth.BlockChain(),
 		possibleUncles:     make(map[common.Hash]*types.Block),
 		unconfirmed:        newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth),
+		pendingTasks:       make(map[common.Hash]*task),
 		txsCh:              make(chan core.NewTxsEvent, txChanSize),
 		chainHeadCh:        make(chan core.ChainHeadEvent, chainHeadChanSize),
 		chainSideCh:        make(chan core.ChainSideEvent, chainSideChanSize),
@@ -317,13 +322,25 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
 		}
 		recommit = time.Duration(int64(next))
 	}
+	// clearPending cleans the stale pending tasks.
+	clearPending := func(number uint64) {
+		w.pendingMu.Lock()
+		for h, t := range w.pendingTasks {
+			if t.block.NumberU64() <= number {
+				delete(w.pendingTasks, h)
+			}
+		}
+		w.pendingMu.Unlock()
+	}
 
 	for {
 		select {
 		case <-w.startCh:
+			clearPending(w.chain.CurrentBlock().NumberU64())
 			commit(false, commitInterruptNewHead)
 
-		case <-w.chainHeadCh:
+		case head := <-w.chainHeadCh:
+			clearPending(head.Block.NumberU64())
 			commit(false, commitInterruptNewHead)
 
 		case <-timer.C:
@@ -454,29 +471,45 @@ func (w *worker) mainLoop() {
 
 // seal pushes a sealing task to consensus engine and submits the result.
 func (w *worker) seal(t *task, stop <-chan struct{}) {
-	var (
-		err error
-		res *task
-	)
-
 	if w.skipSealHook != nil && w.skipSealHook(t) {
 		return
 	}
+	// makeId creates an id for pending task based on consensus engine type.
+	makeId := func(block *types.Block) common.Hash {
+		hash := t.block.HashNoNonce()
+		if _, ok := w.engine.(*clique.Clique); ok {
+			hash = clique.SigHash(t.block.Header())
+		}
+		return hash
+	}
+	// The reason for caching task first is:
+	// 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[makeId(t.block)] = t
+	w.pendingMu.Unlock()
 
-	if t.block, err = w.engine.Seal(w.chain, t.block, stop); t.block != nil {
-		log.Info("Successfully sealed new block", "number", t.block.Number(), "hash", t.block.Hash(),
-			"elapsed", common.PrettyDuration(time.Since(t.createdAt)))
-		res = t
-	} else {
-		if err != nil {
-			log.Warn("Block sealing failed", "err", err)
-		}
-		res = nil
+	if block, err := w.engine.Seal(w.chain, t.block, stop); block != nil {
+		w.pendingMu.RLock()
+		task, exist := w.pendingTasks[makeId(block)]
+		w.pendingMu.RUnlock()
+		if !exist {
+			log.Error("Block found but no relative pending task", "number", block.Number(), "hash", block.Hash())
+			return
 		}
+		// Assemble sealing result
+		task.block = block
+		log.Info("Successfully sealed new block", "number", block.Number(), "hash", block.Hash(),
+			"elapsed", common.PrettyDuration(time.Since(task.createdAt)))
 		select {
-	case w.resultCh <- res:
+		case w.resultCh <- task:
 		case <-w.exitCh:
 		}
+	} else if err != nil {
+		log.Warn("Block sealing failed", "err", err)
+	}
 }
 
 // taskLoop is a standalone goroutine to fetch sealing task from the generator and