cmd/geth, consensus/ethash: add support for --miner.notify.full flag (#22558)

The PR implements the --miner.notify.full flag that enables full pending block
notifications. When this flag is used, the block notifications sent to mining
endpoints contain the complete block header JSON instead of a work package
array.

Co-authored-by: AlexSSD7 <[email protected]>
Co-authored-by: Martin Holst Swende <[email protected]>

Author: 3 people

cmd/geth/main.go

@@ -152,6 +152,7 @@ var (
 		utils.GpoMaxGasPriceFlag,
 		utils.EWASMInterpreterFlag,
 		utils.EVMInterpreterFlag,
+		utils.MinerNotifyFullFlag,
 		configFileFlag,
 	}
 

cmd/geth/usage.go

@@ -180,6 +180,7 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.MiningEnabledFlag,
 			utils.MinerThreadsFlag,
 			utils.MinerNotifyFlag,
+			utils.MinerNotifyFullFlag,
 			utils.MinerGasPriceFlag,
 			utils.MinerGasTargetFlag,
 			utils.MinerGasLimitFlag,

cmd/utils/flags.go

@@ -427,6 +427,10 @@ var (
 		Name:  "miner.notify",
 		Usage: "Comma separated HTTP URL list to notify of new work packages",
 	}
+	MinerNotifyFullFlag = cli.BoolFlag{
+		Name:  "miner.notify.full",
+		Usage: "Notify with pending block headers instead of work packages",
+	}
 	MinerGasTargetFlag = cli.Uint64Flag{
 		Name:  "miner.gastarget",
 		Usage: "Target gas floor for mined blocks",
@@ -1359,6 +1363,7 @@ func setMiner(ctx *cli.Context, cfg *miner.Config) {
 	if ctx.GlobalIsSet(MinerNotifyFlag.Name) {
 		cfg.Notify = strings.Split(ctx.GlobalString(MinerNotifyFlag.Name), ",")
 	}
+	cfg.NotifyFull = ctx.GlobalBool(MinerNotifyFullFlag.Name)
 	if ctx.GlobalIsSet(MinerExtraDataFlag.Name) {
 		cfg.ExtraData = []byte(ctx.GlobalString(MinerExtraDataFlag.Name))
 	}

consensus/ethash/algorithm_test.go

@@ -726,10 +726,14 @@ func TestConcurrentDiskCacheGeneration(t *testing.T) {
 
 	for i := 0; i < 3; i++ {
 		pend.Add(1)
-
 		go func(idx int) {
 			defer pend.Done()
-			ethash := New(Config{cachedir, 0, 1, false, "", 0, 0, false, ModeNormal, nil}, nil, false)
+
+			config := Config{
+				CacheDir:     cachedir,
+				CachesOnDisk: 1,
+			}
+			ethash := New(config, nil, false)
 			defer ethash.Close()
 			if err := ethash.verifySeal(nil, block.Header(), false); err != nil {
 				t.Errorf("proc %d: block verification failed: %v", idx, err)

consensus/ethash/ethash.go

@@ -48,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, false, "", 1, 0, false, ModeNormal, nil}, nil, false)
+	sharedEthash *Ethash
 
 	// algorithmRevision is the data structure version used for file naming.
 	algorithmRevision = 23
@@ -57,6 +57,15 @@ var (
 	dumpMagic = []uint32{0xbaddcafe, 0xfee1dead}
 )
 
+func init() {
+	sharedConfig := Config{
+		PowMode:       ModeNormal,
+		CachesInMem:   3,
+		DatasetsInMem: 1,
+	}
+	sharedEthash = New(sharedConfig, nil, false)
+}
+
 // isLittleEndian returns whether the local system is running in little or big
 // endian byte order.
 func isLittleEndian() bool {
@@ -411,6 +420,10 @@ type Config struct {
 	DatasetsLockMmap bool
 	PowMode          Mode
 
+	// When set, notifications sent by the remote sealer will
+	// be block header JSON objects instead of work package arrays.
+	NotifyFull bool
+
 	Log log.Logger `toml:"-"`
 }
 
@@ -462,22 +475,17 @@ func New(config Config, notify []string, noverify bool) *Ethash {
 		update:   make(chan struct{}),
 		hashrate: metrics.NewMeterForced(),
 	}
+	if config.PowMode == ModeShared {
+		ethash.shared = sharedEthash
+	}
 	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, 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
+	return New(Config{PowMode: ModeTest}, notify, noverify)
 }
 
 // NewFaker creates a ethash consensus engine with a fake PoW scheme that accepts

consensus/ethash/ethash_test.go

@@ -62,7 +62,14 @@ func TestCacheFileEvict(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer os.RemoveAll(tmpdir)
-	e := New(Config{CachesInMem: 3, CachesOnDisk: 10, CacheDir: tmpdir, PowMode: ModeTest}, nil, false)
+
+	config := Config{
+		CachesInMem:  3,
+		CachesOnDisk: 10,
+		CacheDir:     tmpdir,
+		PowMode:      ModeTest,
+	}
+	e := New(config, nil, false)
 	defer e.Close()
 
 	workers := 8

consensus/ethash/sealer.go

@@ -358,7 +358,16 @@ func (s *remoteSealer) makeWork(block *types.Block) {
 // new work to be processed.
 func (s *remoteSealer) notifyWork() {
 	work := s.currentWork
-	blob, _ := json.Marshal(work)
+
+	// Encode the JSON payload of the notification. When NotifyFull is set,
+	// this is the complete block header, otherwise it is a JSON array.
+	var blob []byte
+	if s.ethash.config.NotifyFull {
+		blob, _ = json.Marshal(s.currentBlock.Header())
+	} else {
+		blob, _ = json.Marshal(work)
+	}
+
 	s.reqWG.Add(len(s.notifyURLs))
 	for _, url := range s.notifyURLs {
 		go s.sendNotification(s.notifyCtx, url, blob, work)

consensus/ethash/sealer_test.go

@@ -22,6 +22,7 @@ import (
 	"math/big"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"testing"
 	"time"
 
@@ -74,6 +75,50 @@ func TestRemoteNotify(t *testing.T) {
 	}
 }
 
+// Tests whether remote HTTP servers are correctly notified of new work. (Full pending block body / --miner.notify.full)
+func TestRemoteNotifyFull(t *testing.T) {
+	// Start a simple web server to capture notifications.
+	sink := make(chan map[string]interface{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		blob, err := ioutil.ReadAll(req.Body)
+		if err != nil {
+			t.Errorf("failed to read miner notification: %v", err)
+		}
+		var work map[string]interface{}
+		if err := json.Unmarshal(blob, &work); err != nil {
+			t.Errorf("failed to unmarshal miner notification: %v", err)
+		}
+		sink <- work
+	}))
+	defer server.Close()
+
+	// Create the custom ethash engine.
+	config := Config{
+		PowMode:    ModeTest,
+		NotifyFull: true,
+		Log:        testlog.Logger(t, log.LvlWarn),
+	}
+	ethash := New(config, []string{server.URL}, false)
+	defer ethash.Close()
+
+	// Stream a work task and ensure the notification bubbles out.
+	header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)}
+	block := types.NewBlockWithHeader(header)
+
+	ethash.Seal(nil, block, nil, nil)
+	select {
+	case work := <-sink:
+		if want := "0x" + strconv.FormatUint(header.Number.Uint64(), 16); work["number"] != want {
+			t.Errorf("pending block number mismatch: have %v, want %v", work["number"], want)
+		}
+		if want := "0x" + header.Difficulty.Text(16); work["difficulty"] != want {
+			t.Errorf("pending block difficulty mismatch: have %s, want %s", work["difficulty"], want)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatalf("notification timed out")
+	}
+}
+
 // Tests that pushing work packages fast to the miner doesn't cause any data race
 // issues in the notifications.
 func TestRemoteMultiNotify(t *testing.T) {
@@ -119,6 +164,55 @@ func TestRemoteMultiNotify(t *testing.T) {
 	}
 }
 
+// Tests that pushing work packages fast to the miner doesn't cause any data race
+// issues in the notifications. Full pending block body / --miner.notify.full)
+func TestRemoteMultiNotifyFull(t *testing.T) {
+	// Start a simple web server to capture notifications.
+	sink := make(chan map[string]interface{}, 64)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		blob, err := ioutil.ReadAll(req.Body)
+		if err != nil {
+			t.Errorf("failed to read miner notification: %v", err)
+		}
+		var work map[string]interface{}
+		if err := json.Unmarshal(blob, &work); err != nil {
+			t.Errorf("failed to unmarshal miner notification: %v", err)
+		}
+		sink <- work
+	}))
+	defer server.Close()
+
+	// Create the custom ethash engine.
+	config := Config{
+		PowMode:    ModeTest,
+		NotifyFull: true,
+		Log:        testlog.Logger(t, log.LvlWarn),
+	}
+	ethash := New(config, []string{server.URL}, false)
+	defer ethash.Close()
+
+	// Provide a results reader.
+	// Otherwise the unread results will be logged asynchronously
+	// and this can happen after the test is finished, causing a panic.
+	results := make(chan *types.Block, cap(sink))
+
+	// Stream a lot of work task and ensure all the notifications bubble out.
+	for i := 0; i < cap(sink); i++ {
+		header := &types.Header{Number: big.NewInt(int64(i)), Difficulty: big.NewInt(100)}
+		block := types.NewBlockWithHeader(header)
+		ethash.Seal(nil, block, results, nil)
+	}
+
+	for i := 0; i < cap(sink); i++ {
+		select {
+		case <-sink:
+			<-results
+		case <-time.After(10 * time.Second):
+			t.Fatalf("notification %d timed out", i)
+		}
+	}
+}
+
 // Tests whether stale solutions are correctly processed.
 func TestStaleSubmission(t *testing.T) {
 	ethash := NewTester(nil, true)

eth/backend.go

@@ -121,6 +121,10 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 	}
 	log.Info("Allocated trie memory caches", "clean", common.StorageSize(config.TrieCleanCache)*1024*1024, "dirty", common.StorageSize(config.TrieDirtyCache)*1024*1024)
 
+	// Transfer mining-related config to the ethash config.
+	ethashConfig := config.Ethash
+	ethashConfig.NotifyFull = config.Miner.NotifyFull
+
 	// Assemble the Ethereum object
 	chainDb, err := stack.OpenDatabaseWithFreezer("chaindata", config.DatabaseCache, config.DatabaseHandles, config.DatabaseFreezer, "eth/db/chaindata/", false)
 	if err != nil {
@@ -140,7 +144,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		chainDb:           chainDb,
 		eventMux:          stack.EventMux(),
 		accountManager:    stack.AccountManager(),
-		engine:            ethconfig.CreateConsensusEngine(stack, chainConfig, &config.Ethash, config.Miner.Notify, config.Miner.Noverify, chainDb),
+		engine:            ethconfig.CreateConsensusEngine(stack, chainConfig, &ethashConfig, config.Miner.Notify, config.Miner.Noverify, chainDb),
 		closeBloomHandler: make(chan struct{}),
 		networkID:         config.NetworkId,
 		gasPrice:          config.Miner.GasPrice,

eth/ethconfig/config.go

@@ -213,25 +213,23 @@ func CreateConsensusEngine(stack *node.Node, chainConfig *params.ChainConfig, co
 	switch config.PowMode {
 	case ethash.ModeFake:
 		log.Warn("Ethash used in fake mode")
-		return ethash.NewFaker()
 	case ethash.ModeTest:
 		log.Warn("Ethash used in test mode")
-		return ethash.NewTester(nil, noverify)
 	case ethash.ModeShared:
 		log.Warn("Ethash used in shared mode")
-		return ethash.NewShared()
-	default:
-		engine := ethash.New(ethash.Config{
-			CacheDir:         stack.ResolvePath(config.CacheDir),
-			CachesInMem:      config.CachesInMem,
-			CachesOnDisk:     config.CachesOnDisk,
-			CachesLockMmap:   config.CachesLockMmap,
-			DatasetDir:       config.DatasetDir,
-			DatasetsInMem:    config.DatasetsInMem,
-			DatasetsOnDisk:   config.DatasetsOnDisk,
-			DatasetsLockMmap: config.DatasetsLockMmap,
-		}, notify, noverify)
-		engine.SetThreads(-1) // Disable CPU mining
-		return engine
 	}
+	engine := ethash.New(ethash.Config{
+		PowMode:          config.PowMode,
+		CacheDir:         stack.ResolvePath(config.CacheDir),
+		CachesInMem:      config.CachesInMem,
+		CachesOnDisk:     config.CachesOnDisk,
+		CachesLockMmap:   config.CachesLockMmap,
+		DatasetDir:       config.DatasetDir,
+		DatasetsInMem:    config.DatasetsInMem,
+		DatasetsOnDisk:   config.DatasetsOnDisk,
+		DatasetsLockMmap: config.DatasetsLockMmap,
+		NotifyFull:       config.NotifyFull,
+	}, notify, noverify)
+	engine.SetThreads(-1) // Disable CPU mining
+	return engine
 }