forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

consensus, miner: refactor seal signature

Browse source
This commit is contained in:
rjl493456442 2018-08-27 20:02:27 +08:00
parent 8fee9f1d8a
commit 0bfd286b11
7 changed files with 209 additions and 289 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
consensus/clique/clique.go
View file

@ -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

9
consensus/consensus.go
View file

@ -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
}

20
consensus/ethash/ethash.go
View file

@ -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
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
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
@ -487,13 +489,11 @@ func New(config Config, notify []string) *Ethash {
datasets: newlru("dataset", config.DatasetsInMem, newDataset),
update: make(chan struct{}),
hashrate: metrics.NewMeter(),
workCh: make(chan *types.Block),
resultCh: make(chan *types.Block),
workCh: make(chan *sealTask),
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)
@ -509,13 +509,11 @@ func NewTester(notify []string) *Ethash {
datasets: newlru("dataset", 1, newDataset),
update: make(chan struct{}),
hashrate: metrics.NewMeter(),
workCh: make(chan *types.Block),
resultCh: make(chan *types.Block),
workCh: make(chan *sealTask),
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)

22
consensus/ethash/ethash_test.go
View file

@ -37,15 +37,21 @@ 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)
}
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")
}
}
// This test checks that cache lru logic doesn't crash under load.
@ -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) {

91
consensus/ethash/sealer.go
View file

@ -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
go func() {
var result *types.Block
select {
case <-stop:
// Outside abort, stop all miner threads
close(abort)
case result = <-ethash.resultCh:
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()
return ethash.Seal(chain, block, stop)
ethash.Seal(chain, block, results, 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
}
} 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)
log.Warn("Sealing result is not read by miner", "sealhash", sealhash)
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
}

49
consensus/ethash/sealer_test.go
View file

@ -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,59 +125,59 @@ 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)
}
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)
}
if !c.submitRes {
continue
}
stop := make(chan struct{})
go func(stop chan struct{}) {
select {
case res := <-c.resCh:
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)
}
@ -187,15 +187,14 @@ func TestStaleSubmission(t *testing.T) {
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)
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)
}
}

166
miner/worker.go
View file

@ -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())