diff --git a/consensus/consensus.go b/consensus/consensus.go index 27799f13c0..6e8b9cd16d 100644 --- a/consensus/consensus.go +++ b/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 } diff --git a/consensus/ethash/ethash.go b/consensus/ethash/ethash.go index d98c3371c5..0590ebc27e 100644 --- a/consensus/ethash/ethash.go +++ b/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 diff --git a/consensus/ethash/sealer.go b/consensus/ethash/sealer.go index a458c60f66..bfeee566db 100644 --- a/consensus/ethash/sealer.go +++ b/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 +} diff --git a/consensus/ethash/sealer_test.go b/consensus/ethash/sealer_test.go index d1b66f9cf3..4624da52eb 100644 --- a/consensus/ethash/sealer_test.go +++ b/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) + } +} diff --git a/eth/api_tracer.go b/eth/api_tracer.go index 704a6cdbaa..0a8b9a9942 100644 --- a/eth/api_tracer.go +++ b/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{}) diff --git a/miner/worker.go b/miner/worker.go index 1a881799d5..fb8dd07f11 100644 --- a/miner/worker.go +++ b/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())