consensus/ethash: refactor remote sealer (#20335)

The original idea behind this change was to remove a use of the
deprecated CancelRequest method. Simply removing it would've been an
option, but I couldn't resist and did a bit of a refactoring instead.

All remote sealing code was contained in a single giant function. Remote
sealing is now extracted into its own object, remoteSealer.

Author: fjl

consensus/ethash/algorithm_test.go

@@ -729,7 +729,7 @@ func TestConcurrentDiskCacheGeneration(t *testing.T) {
 
 		go func(idx int) {
 			defer pend.Done()
-			ethash := New(Config{cachedir, 0, 1, "", 0, 0, ModeNormal}, nil, false)
+			ethash := New(Config{cachedir, 0, 1, "", 0, 0, ModeNormal, nil}, nil, false)
 			defer ethash.Close()
 			if err := ethash.VerifySeal(nil, block.Header()); err != nil {
 				t.Errorf("proc %d: block verification failed: %v", idx, err)

consensus/ethash/api.go

@@ -28,7 +28,7 @@ var errEthashStopped = errors.New("ethash stopped")
 
 // API exposes ethash related methods for the RPC interface.
 type API struct {
-	ethash *Ethash // Make sure the mode of ethash is normal.
+	ethash *Ethash
 }
 
 // GetWork returns a work package for external miner.
@@ -39,21 +39,19 @@ type API struct {
 //   result[2] - 32 bytes hex encoded boundary condition ("target"), 2^256/difficulty
 //   result[3] - hex encoded block number
 func (api *API) GetWork() ([4]string, error) {
-	if api.ethash.config.PowMode != ModeNormal && api.ethash.config.PowMode != ModeTest {
+	if api.ethash.remote == nil {
 		return [4]string{}, errors.New("not supported")
 	}
 
 	var (
 		workCh = make(chan [4]string, 1)
 		errc   = make(chan error, 1)
 	)
-
 	select {
-	case api.ethash.fetchWorkCh <- &sealWork{errc: errc, res: workCh}:
-	case <-api.ethash.exitCh:
+	case api.ethash.remote.fetchWorkCh <- &sealWork{errc: errc, res: workCh}:
+	case <-api.ethash.remote.exitCh:
 		return [4]string{}, errEthashStopped
 	}
-
 	select {
 	case work := <-workCh:
 		return work, nil
@@ -66,23 +64,21 @@ func (api *API) GetWork() ([4]string, error) {
 // It returns an indication if the work was accepted.
 // Note either an invalid solution, a stale work a non-existent work will return false.
 func (api *API) SubmitWork(nonce types.BlockNonce, hash, digest common.Hash) bool {
-	if api.ethash.config.PowMode != ModeNormal && api.ethash.config.PowMode != ModeTest {
+	if api.ethash.remote == nil {
 		return false
 	}
 
 	var errc = make(chan error, 1)
-
 	select {
-	case api.ethash.submitWorkCh <- &mineResult{
+	case api.ethash.remote.submitWorkCh <- &mineResult{
 		nonce:     nonce,
 		mixDigest: digest,
 		hash:      hash,
 		errc:      errc,
 	}:
-	case <-api.ethash.exitCh:
+	case <-api.ethash.remote.exitCh:
 		return false
 	}
-
 	err := <-errc
 	return err == nil
 }
@@ -94,21 +90,19 @@ func (api *API) SubmitWork(nonce types.BlockNonce, hash, digest common.Hash) boo
 // It accepts the miner hash rate and an identifier which must be unique
 // between nodes.
 func (api *API) SubmitHashRate(rate hexutil.Uint64, id common.Hash) bool {
-	if api.ethash.config.PowMode != ModeNormal && api.ethash.config.PowMode != ModeTest {
+	if api.ethash.remote == nil {
 		return false
 	}
 
 	var done = make(chan struct{}, 1)
-
 	select {
-	case api.ethash.submitRateCh <- &hashrate{done: done, rate: uint64(rate), id: id}:
-	case <-api.ethash.exitCh:
+	case api.ethash.remote.submitRateCh <- &hashrate{done: done, rate: uint64(rate), id: id}:
+	case <-api.ethash.remote.exitCh:
 		return false
 	}
 
 	// Block until hash rate submitted successfully.
 	<-done
-
 	return true
 }
 

consensus/ethash/ethash.go

@@ -34,9 +34,7 @@ import (
 	"unsafe"
 
 	mmap "github.com/edsrzf/mmap-go"
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus"
-	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/rpc"
@@ -50,7 +48,7 @@ var (
 	two256 = new(big.Int).Exp(big.NewInt(2), big.NewInt(256), big.NewInt(0))
 
 	// sharedEthash is a full instance that can be shared between multiple users.
-	sharedEthash = New(Config{"", 3, 0, "", 1, 0, ModeNormal}, nil, false)
+	sharedEthash = New(Config{"", 3, 0, "", 1, 0, ModeNormal, nil}, nil, false)
 
 	// algorithmRevision is the data structure version used for file naming.
 	algorithmRevision = 23
@@ -403,36 +401,8 @@ type Config struct {
 	DatasetsInMem  int
 	DatasetsOnDisk int
 	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
-	mixDigest common.Hash
-	hash      common.Hash
-
-	errc chan error
-}
-
-// hashrate wraps the hash rate submitted by the remote sealer.
-type hashrate struct {
-	id   common.Hash
-	ping time.Time
-	rate uint64
 
-	done chan struct{}
-}
-
-// sealWork wraps a seal work package for remote sealer.
-type sealWork struct {
-	errc chan error
-	res  chan [4]string
+	Log log.Logger `toml:"-"`
 }
 
 // Ethash is a consensus engine based on proof-of-work implementing the ethash
@@ -448,72 +418,56 @@ type Ethash struct {
 	threads  int           // Number of threads to mine on if mining
 	update   chan struct{} // Notification channel to update mining parameters
 	hashrate metrics.Meter // Meter tracking the average hashrate
-
-	// Remote sealer related fields
-	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
+	remote   *remoteSealer
 
 	// The fields below are hooks for testing
 	shared    *Ethash       // Shared PoW verifier to avoid cache regeneration
 	fakeFail  uint64        // Block number which fails PoW check even in fake mode
 	fakeDelay time.Duration // Time delay to sleep for before returning from verify
 
-	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
+	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.
 }
 
 // New creates a full sized ethash PoW scheme and starts a background thread for
 // remote mining, also optionally notifying a batch of remote services of new work
 // packages.
 func New(config Config, notify []string, noverify bool) *Ethash {
+	if config.Log == nil {
+		config.Log = log.Root()
+	}
 	if config.CachesInMem <= 0 {
-		log.Warn("One ethash cache must always be in memory", "requested", config.CachesInMem)
+		config.Log.Warn("One ethash cache must always be in memory", "requested", config.CachesInMem)
 		config.CachesInMem = 1
 	}
 	if config.CacheDir != "" && config.CachesOnDisk > 0 {
-		log.Info("Disk storage enabled for ethash caches", "dir", config.CacheDir, "count", config.CachesOnDisk)
+		config.Log.Info("Disk storage enabled for ethash caches", "dir", config.CacheDir, "count", config.CachesOnDisk)
 	}
 	if config.DatasetDir != "" && config.DatasetsOnDisk > 0 {
-		log.Info("Disk storage enabled for ethash DAGs", "dir", config.DatasetDir, "count", config.DatasetsOnDisk)
+		config.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.NewMeterForced(),
-		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, noverify)
+		config:   config,
+		caches:   newlru("cache", config.CachesInMem, newCache),
+		datasets: newlru("dataset", config.DatasetsInMem, newDataset),
+		update:   make(chan struct{}),
+		hashrate: metrics.NewMeterForced(),
+	}
+	ethash.remote = startRemoteSealer(ethash, notify, noverify)
 	return ethash
 }
 
 // NewTester creates a small sized ethash PoW scheme useful only for testing
 // purposes.
 func NewTester(notify []string, noverify bool) *Ethash {
 	ethash := &Ethash{
-		config:       Config{PowMode: ModeTest},
-		caches:       newlru("cache", 1, newCache),
-		datasets:     newlru("dataset", 1, newDataset),
-		update:       make(chan struct{}),
-		hashrate:     metrics.NewMeterForced(),
-		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, noverify)
+		config:   Config{PowMode: ModeTest, Log: log.Root()},
+		caches:   newlru("cache", 1, newCache),
+		datasets: newlru("dataset", 1, newDataset),
+		update:   make(chan struct{}),
+		hashrate: metrics.NewMeterForced(),
+	}
+	ethash.remote = startRemoteSealer(ethash, notify, noverify)
 	return ethash
 }
 
@@ -524,6 +478,7 @@ func NewFaker() *Ethash {
 	return &Ethash{
 		config: Config{
 			PowMode: ModeFake,
+			Log:     log.Root(),
 		},
 	}
 }
@@ -535,6 +490,7 @@ func NewFakeFailer(fail uint64) *Ethash {
 	return &Ethash{
 		config: Config{
 			PowMode: ModeFake,
+			Log:     log.Root(),
 		},
 		fakeFail: fail,
 	}
@@ -547,6 +503,7 @@ func NewFakeDelayer(delay time.Duration) *Ethash {
 	return &Ethash{
 		config: Config{
 			PowMode: ModeFake,
+			Log:     log.Root(),
 		},
 		fakeDelay: delay,
 	}
@@ -558,6 +515,7 @@ func NewFullFaker() *Ethash {
 	return &Ethash{
 		config: Config{
 			PowMode: ModeFullFake,
+			Log:     log.Root(),
 		},
 	}
 }
@@ -573,13 +531,11 @@ func (ethash *Ethash) Close() error {
 	var err error
 	ethash.closeOnce.Do(func() {
 		// Short circuit if the exit channel is not allocated.
-		if ethash.exitCh == nil {
+		if ethash.remote == nil {
 			return
 		}
-		errc := make(chan error)
-		ethash.exitCh <- errc
-		err = <-errc
-		close(ethash.exitCh)
+		close(ethash.remote.requestExit)
+		<-ethash.remote.exitCh
 	})
 	return err
 }
@@ -680,8 +636,8 @@ func (ethash *Ethash) Hashrate() float64 {
 	var res = make(chan uint64, 1)
 
 	select {
-	case ethash.fetchRateCh <- res:
-	case <-ethash.exitCh:
+	case ethash.remote.fetchRateCh <- res:
+	case <-ethash.remote.exitCh:
 		// Return local hashrate only if ethash is stopped.
 		return ethash.hashrate.Rate1()
 	}