diff --git a/consensus/clique/clique.go b/consensus/clique/clique.go index 3730c91f62..5472909846 100644 --- a/consensus/clique/clique.go +++ b/consensus/clique/clique.go @@ -590,17 +590,17 @@ func (c *Clique) Authorize(signer common.Address, signFn SignerFn) { // Seal implements consensus.Engine, attempting to create a sealed block using // the local signing credentials. -func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-chan struct{}) (*types.Block, error) { +func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { header := block.Header() // Sealing the genesis block is not supported number := header.Number.Uint64() if number == 0 { - return nil, errUnknownBlock + return errUnknownBlock } // For 0-period chains, refuse to seal empty blocks (no reward but would spin sealing) if c.config.Period == 0 && len(block.Transactions()) == 0 { - return nil, errWaitTransactions + return errWaitTransactions } // Don't hold the signer fields for the entire sealing procedure c.lock.RLock() @@ -610,10 +610,10 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch // Bail out if we're unauthorized to sign a block snap, err := c.snapshot(chain, number-1, header.ParentHash, nil) if err != nil { - return nil, err + return err } if _, authorized := snap.Signers[signer]; !authorized { - return nil, errUnauthorized + return errUnauthorized } // If we're amongst the recent signers, wait for the next block for seen, recent := range snap.Recents { @@ -621,8 +621,7 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch // Signer is among recents, only wait if the current block doesn't shift it out if limit := uint64(len(snap.Signers)/2 + 1); number < limit || seen > number-limit { log.Info("Signed recently, must wait for others") - <-stop - return nil, nil + return nil } } } @@ -635,21 +634,29 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch log.Trace("Out-of-turn signing requested", "wiggle", common.PrettyDuration(wiggle)) } - log.Trace("Waiting for slot to sign and propagate", "delay", common.PrettyDuration(delay)) - - select { - case <-stop: - return nil, nil - case <-time.After(delay): - } // Sign all the things! sighash, err := signFn(accounts.Account{Address: signer}, sigHash(header).Bytes()) if err != nil { - return nil, err + return err } copy(header.Extra[len(header.Extra)-extraSeal:], sighash) + // Wait until sealing is terminated or delay timeout. + log.Trace("Waiting for slot to sign and propagate", "delay", common.PrettyDuration(delay)) + go func() { + select { + case <-stop: + return + case <-time.After(delay): + } - return block.WithSeal(header), nil + select { + case results <- block.WithSeal(header): + default: + log.Warn("Sealing result is not read by miner", "sealhash", c.SealHash(header)) + } + }() + + return nil } // CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty diff --git a/consensus/consensus.go b/consensus/consensus.go index 6e8b9cd16d..3fa0ae2bb1 100644 --- a/consensus/consensus.go +++ b/consensus/consensus.go @@ -86,9 +86,9 @@ type Engine interface { Finalize(chain ChainReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) - // Seal generates a new block for the given input block with the local miner's - // seal place on top. - Seal(chain ChainReader, block *types.Block, stop <-chan struct{}) (*types.Block, error) + // Seal generates a new sealing request for the given input block and pushes + // the result in the given channel. + Seal(chain ChainReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error // SealHash returns the hash of a block prior to it being sealed. SealHash(header *types.Header) common.Hash @@ -110,7 +110,4 @@ 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 0590ebc27e..5cfc4196e7 100644 --- a/consensus/ethash/ethash.go +++ b/consensus/ethash/ethash.go @@ -394,8 +394,6 @@ const ( ModeFullFake ) -const staleChannelSize = 10 // The size of stale block channel. - // Config are the configuration parameters of the ethash. type Config struct { CacheDir string @@ -407,6 +405,12 @@ type Config struct { PowMode Mode } +// sealTask wraps a seal block with relative result channel for remote sealer thread. +type sealTask struct { + block *types.Block + results chan<- *types.Block +} + // mineResult wraps the pow solution parameters for the specified block. type mineResult struct { nonce types.BlockNonce @@ -446,13 +450,11 @@ 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 - staleResultCh chan *types.Block // Channel used for passing back stale but acceptable solutions. + workCh chan *sealTask // Notification channel to push new work and relative result channel to remote sealer + 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 // The fields below are hooks for testing shared *Ethash // Shared PoW verifier to avoid cache regeneration @@ -482,19 +484,17 @@ 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), - staleResultCh: make(chan *types.Block, staleChannelSize), - 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 *sealTask), + fetchWorkCh: make(chan *sealWork), + submitWorkCh: make(chan *mineResult), + fetchRateCh: make(chan chan uint64), + submitRateCh: make(chan *hashrate), + exitCh: make(chan chan error), } go ethash.remote(notify) return ethash @@ -504,19 +504,17 @@ 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), - staleResultCh: make(chan *types.Block, staleChannelSize), - 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 *sealTask), + fetchWorkCh: make(chan *sealWork), + submitWorkCh: make(chan *mineResult), + fetchRateCh: make(chan chan uint64), + submitRateCh: make(chan *hashrate), + exitCh: make(chan chan error), } go ethash.remote(notify) return ethash diff --git a/consensus/ethash/ethash_test.go b/consensus/ethash/ethash_test.go index b190d63d6b..21c3df9091 100644 --- a/consensus/ethash/ethash_test.go +++ b/consensus/ethash/ethash_test.go @@ -37,14 +37,20 @@ func TestTestMode(t *testing.T) { ethash := NewTester(nil) defer ethash.Close() - block, err := ethash.Seal(nil, types.NewBlockWithHeader(header), nil) + results := make(chan *types.Block) + err := ethash.Seal(nil, types.NewBlockWithHeader(header), results, nil) if err != nil { t.Fatalf("failed to seal block: %v", err) } - header.Nonce = types.EncodeNonce(block.Nonce()) - header.MixDigest = block.MixDigest() - if err := ethash.VerifySeal(nil, header); err != nil { - t.Fatalf("unexpected verification error: %v", err) + select { + case block := <-results: + header.Nonce = types.EncodeNonce(block.Nonce()) + header.MixDigest = block.MixDigest() + if err := ethash.VerifySeal(nil, header); err != nil { + t.Fatalf("unexpected verification error: %v", err) + } + case <-time.NewTimer(time.Second).C: + t.Error("sealing result timeout") } } @@ -97,7 +103,8 @@ func TestRemoteSealer(t *testing.T) { sealhash := ethash.SealHash(header) // Push new work. - ethash.Seal(nil, block, nil) + results := make(chan *types.Block) + ethash.Seal(nil, block, results, nil) var ( work [3]string @@ -114,20 +121,11 @@ func TestRemoteSealer(t *testing.T) { header = &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(1000)} block = types.NewBlockWithHeader(header) sealhash = ethash.SealHash(header) - ethash.Seal(nil, block, nil) + ethash.Seal(nil, block, results, nil) if work, err = api.GetWork(); err != nil || work[0] != sealhash.Hex() { t.Error("expect to return the latest pushed work") } - // Push block with higher block number. - newHead := &types.Header{Number: big.NewInt(2), Difficulty: big.NewInt(100)} - newBlock := types.NewBlockWithHeader(newHead) - newSealhash := ethash.SealHash(newHead) - ethash.Seal(nil, newBlock, nil) - - if res := api.SubmitWork(types.BlockNonce{}, newSealhash, common.Hash{}); res { - t.Error("expect to return false when submit a stale solution") - } } func TestHashRate(t *testing.T) { diff --git a/consensus/ethash/sealer.go b/consensus/ethash/sealer.go index bfeee566db..607109afca 100644 --- a/consensus/ethash/sealer.go +++ b/consensus/ethash/sealer.go @@ -36,7 +36,7 @@ import ( ) const ( - // staleThreshold is the maximum distance of the acceptable stale but valid ethash solution. + // staleThreshold is the maximum depth of the acceptable stale but valid ethash solution. staleThreshold = 7 ) @@ -47,16 +47,21 @@ var ( // Seal implements consensus.Engine, attempting to find a nonce that satisfies // the block's difficulty requirements. -func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, stop <-chan struct{}) (*types.Block, error) { +func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { // If we're running a fake PoW, simply return a 0 nonce immediately if ethash.config.PowMode == ModeFake || ethash.config.PowMode == ModeFullFake { header := block.Header() header.Nonce, header.MixDigest = types.BlockNonce{}, common.Hash{} - return block.WithSeal(header), nil + select { + case results <- block.WithSeal(header): + default: + log.Warn("Sealing result is not read by miner", "sealhash", ethash.SealHash(block.Header())) + } + return nil } // If we're running a shared PoW, delegate sealing to it if ethash.shared != nil { - return ethash.shared.Seal(chain, block, stop) + return ethash.shared.Seal(chain, block, results, stop) } // Create a runner and the multiple search threads it directs abort := make(chan struct{}) @@ -67,7 +72,7 @@ func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, stop seed, err := crand.Int(crand.Reader, big.NewInt(math.MaxInt64)) if err != nil { ethash.lock.Unlock() - return nil, err + return err } ethash.rand = rand.New(rand.NewSource(seed.Int64())) } @@ -80,34 +85,44 @@ func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, stop } // Push new work to remote sealer if ethash.workCh != nil { - ethash.workCh <- block + ethash.workCh <- &sealTask{block: block, results: results} } - var pend sync.WaitGroup + var ( + pend sync.WaitGroup + localResults = make(chan *types.Block) + ) for i := 0; i < threads; i++ { pend.Add(1) go func(id int, nonce uint64) { defer pend.Done() - ethash.mine(block, id, nonce, abort, ethash.resultCh) + ethash.mine(block, id, nonce, abort, localResults) }(i, uint64(ethash.rand.Int63())) } // Wait until sealing is terminated or a nonce is found - var result *types.Block - select { - case <-stop: - // Outside abort, stop all miner threads - close(abort) - case result = <-ethash.resultCh: - // One of the threads found a block, abort all others - close(abort) - case <-ethash.update: - // Thread count was changed on user request, restart - close(abort) + go func() { + var result *types.Block + select { + case <-stop: + // Outside abort, stop all miner threads + close(abort) + case result = <-localResults: + // One of the threads found a block, abort all others + select { + case results <- result: + default: + log.Warn("Sealing result is not read by miner", "sealhash", ethash.SealHash(block.Header())) + } + close(abort) + case <-ethash.update: + // Thread count was changed on user request, restart + close(abort) + pend.Wait() + ethash.Seal(chain, block, results, stop) + } + // Wait for all miners to terminate and return the block pend.Wait() - return ethash.Seal(chain, block, stop) - } - // Wait for all miners to terminate and return the block - pend.Wait() - return result, nil + }() + return nil } // mine is the actual proof-of-work miner that searches for a nonce starting from @@ -175,6 +190,7 @@ func (ethash *Ethash) remote(notify []string) { works = make(map[common.Hash]*types.Block) rates = make(map[common.Hash]hashrate) + results chan<- *types.Block currentBlock *types.Block currentWork [3]string @@ -232,14 +248,14 @@ func (ethash *Ethash) remote(notify []string) { // 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, sealhash common.Hash) bool { + if currentBlock == nil { + log.Error("Pending work without block", "sealhash", sealhash) + return false + } // Make sure the work submitted is present block := works[sealhash] if block == nil { - 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) + log.Warn("Work submitted but none pending", "sealhash", sealhash, "curnumber", currentBlock.NumberU64()) return false } // Verify the correctness of submitted result. @@ -254,39 +270,29 @@ func (ethash *Ethash) remote(notify []string) { return false } } - // Make sure the result channel is created. - if ethash.resultCh == nil { + // Make sure the result channel is assigned. + if results == nil { log.Warn("Ethash result channel is empty, submitted mining result is rejected") return false } 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. - submitBlock := block.WithSeal(header) + solution := 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() { + // The submitted solution is within the scope of acceptance. + if solution.NumberU64()+staleThreshold > currentBlock.NumberU64() { select { - case ethash.resultCh <- submitBlock: - log.Info("Work submitted is fresh", "number", submitBlock.NumberU64(), "sealhash", sealhash, "hash", submitBlock.Hash()) + case results <- solution: + log.Debug("Work submitted is acceptable", "number", solution.NumberU64(), "sealhash", sealhash, "hash", solution.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 + log.Warn("Sealing result is not read by miner", "sealhash", sealhash) + return false } - } 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 - } 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 } + // The submitted block is too old to accept, drop it. + log.Debug("Work submitted is too old", "number", solution.NumberU64(), "sealhash", sealhash, "hash", solution.Hash()) return false } @@ -295,10 +301,12 @@ func (ethash *Ethash) remote(notify []string) { for { select { - case block := <-ethash.workCh: + case work := <-ethash.workCh: // Update current work with new received block. // Note same work can be past twice, happens when changing CPU threads. - makeWork(block) + results = work.results + + makeWork(work.block) // Notify and requested URLs of the new work availability notifyWork() @@ -357,8 +365,3 @@ 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 4624da52eb..866f9b5f8d 100644 --- a/consensus/ethash/sealer_test.go +++ b/consensus/ethash/sealer_test.go @@ -48,7 +48,7 @@ func TestRemoteNotify(t *testing.T) { header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, nil) + ethash.Seal(nil, block, nil, nil) select { case work := <-sink: if want := ethash.SealHash(header).Hex(); work[0] != want { @@ -103,7 +103,7 @@ func TestRemoteMultiNotify(t *testing.T) { header := &types.Header{Number: big.NewInt(int64(i)), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, nil) + ethash.Seal(nil, block, nil, nil) } for i := 0; i < cap(sink); i++ { select { @@ -125,77 +125,76 @@ func TestStaleSubmission(t *testing.T) { 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)}, + {ParentHash: common.BytesToHash([]byte{0xa}), Number: big.NewInt(1), Difficulty: big.NewInt(100000000)}, }, - 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)}, + {ParentHash: common.BytesToHash([]byte{0xb}), Number: big.NewInt(2), Difficulty: big.NewInt(100000000)}, + {ParentHash: common.BytesToHash([]byte{0xb}), Number: big.NewInt(2), Difficulty: big.NewInt(100000001)}, }, - 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)}, + {ParentHash: common.BytesToHash([]byte{0xc}), Number: big.NewInt(3), Difficulty: big.NewInt(100000000)}, + {ParentHash: common.BytesToHash([]byte{0xd}), Number: big.NewInt(9), Difficulty: big.NewInt(100000000)}, }, - 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)}, + {ParentHash: common.BytesToHash([]byte{0xe}), Number: big.NewInt(10), Difficulty: big.NewInt(100000000)}, + {ParentHash: common.BytesToHash([]byte{0xf}), Number: big.NewInt(17), Difficulty: big.NewInt(100000000)}, }, - nil, 0, false, }, } + results := make(chan *types.Block, 16) for id, c := range testcases { for _, h := range c.headers { - ethash.Seal(nil, types.NewBlockWithHeader(h), nil) + ethash.Seal(nil, types.NewBlockWithHeader(h), results, 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) + if !c.submitRes { + continue + } + select { + case res := <-results: + 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) + } + if res.Header().Number.Uint64() != c.headers[c.submitIndex].Number.Uint64() { + t.Errorf("case %d block number mismatch, want %d, get %d", id+1, c.headers[c.submitIndex].Number.Uint64(), res.Header().Number.Uint64()) + } + if res.Header().ParentHash != c.headers[c.submitIndex].ParentHash { + t.Errorf("case %d block parent hash mismatch, want %d, get %d", id+1, c.headers[c.submitIndex].ParentHash.Hex(), res.Header().ParentHash.Hex()) + } + case <-time.NewTimer(time.Second).C: + t.Errorf("case %d fetch ethash result timeout", id+1) + } } } diff --git a/miner/worker.go b/miner/worker.go index fb8dd07f11..d32680e4c9 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -73,7 +73,7 @@ const ( // increasing upper limit or decreasing lower limit so that the limit can be reachable. intervalAdjustBias = 200 * 1000.0 * 1000.0 - // staleThreshold is the maximum distance of the acceptable stale block. + // staleThreshold is the maximum depth of the acceptable stale block. staleThreshold = 7 ) @@ -98,7 +98,6 @@ type task struct { receipts []*types.Receipt state *state.StateDB block *types.Block - logs []*types.Log createdAt time.Time } @@ -140,7 +139,7 @@ type worker struct { // Channels newWorkCh chan *newWorkReq taskCh chan *task - resultCh chan *task + resultCh chan *types.Block startCh chan struct{} exitCh chan struct{} resubmitIntervalCh chan time.Duration @@ -187,7 +186,7 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend, chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize), newWorkCh: make(chan *newWorkReq), taskCh: make(chan *task), - resultCh: make(chan *task, resultQueueSize), + resultCh: make(chan *types.Block, resultQueueSize), exitCh: make(chan struct{}), startCh: make(chan struct{}, 1), resubmitIntervalCh: make(chan time.Duration), @@ -210,10 +209,6 @@ 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{}{} @@ -274,18 +269,10 @@ func (w *worker) isRunning() bool { 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. // Note the worker does not support being closed multiple times. func (w *worker) close() { close(w.exitCh) - // Clean up buffered channels - for empty := false; !empty; { - select { - case <-w.resultCh: - default: - empty = true - } - } } // newWorkLoop is a standalone goroutine to submit new mining work upon received events. @@ -476,57 +463,6 @@ 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{}) { - if w.skipSealHook != nil && w.skipSealHook(t) { - return - } - // 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. - w.pendingMu.Lock() - w.pendingTasks[w.engine.SealHash(t.block.Header())] = t - w.pendingMu.Unlock() - - if block, err := w.engine.Seal(w.chain, t.block, stop); block != nil { - sealhash := w.engine.SealHash(block.Header()) - w.pendingMu.RLock() - 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 - } - // 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(cpy.createdAt))) - select { - case w.resultCh <- cpy: - 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 // push them to consensus engine. func (w *worker) taskLoop() { @@ -553,59 +489,22 @@ func (w *worker) taskLoop() { if sealHash == prev { continue } + // Interrupt previous sealing operation interrupt() - stopCh = make(chan struct{}) - prev = sealHash - go w.seal(task, stopCh) - case <-w.exitCh: - interrupt() - return - } - } -} + stopCh, prev = make(chan struct{}), sealHash -// 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()) + if w.skipSealHook != nil && w.skipSealHook(task) { 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: + w.pendingMu.Lock() + w.pendingTasks[w.engine.SealHash(task.block.Header())] = task + w.pendingMu.Unlock() + + if err := w.engine.Seal(w.chain, task.block, w.resultCh, stopCh); err != nil { + log.Warn("Block sealing failed", "err", err) } case <-w.exitCh: + interrupt() return } } @@ -616,21 +515,40 @@ func (w *worker) staleResultLoop(poW consensus.PoW) { func (w *worker) resultLoop() { for { select { - case result := <-w.resultCh: + case block := <-w.resultCh: // Short circuit when receiving empty result. - if result == nil { + if block == nil { continue } // Short circuit when receiving duplicate result caused by resubmitting. - block := result.block if w.chain.HasBlock(block.Hash(), block.NumberU64()) { continue } + sealhash := w.engine.SealHash(block.Header()) + w.pendingMu.RLock() + task, 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()) + continue + } + // Different block could share same sealhash, deep copy here to prevent write-write conflict. + var ( + receipts = make([]*types.Receipt, len(task.receipts)) + logs []*types.Log + ) + for i, receipt := range task.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) + } + } // 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) + stat, err := w.chain.WriteBlockWithState(block, receipts, task.state) if err != nil { log.Error("Failed writing block to chain", "err", err) continue @@ -640,12 +558,12 @@ func (w *worker) resultLoop() { var events []interface{} switch stat { case core.CanonStatTy: - events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: result.logs}) + events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs}) events = append(events, core.ChainHeadEvent{Block: block}) case core.SideStatTy: events = append(events, core.ChainSideEvent{Block: block}) } - w.chain.PostChainEvents(events, result.logs) + w.chain.PostChainEvents(events, logs) // Insert the block into the set of pending ones to resultLoop for confirmations w.unconfirmed.Insert(block.NumberU64(), block.Hash())