feat: add broadcast to all cap

Author: yiweichi

cmd/geth/main.go

@@ -179,6 +179,7 @@ var (
 		utils.GossipTxBroadcastDisabledFlag,
 		utils.GossipTxReceivingDisabledFlag,
 		utils.GossipBroadcastToAllEnabledFlag,
+		utils.GossipBroadcastToAllCapFlag,
 		utils.DASyncEnabledFlag,
 		utils.DABlockNativeAPIEndpointFlag,
 		utils.DABlobScanAPIEndpointFlag,

cmd/geth/usage.go

@@ -252,6 +252,7 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.GossipTxBroadcastDisabledFlag,
 			utils.GossipTxReceivingDisabledFlag,
 			utils.GossipBroadcastToAllEnabledFlag,
+			utils.GossipBroadcastToAllCapFlag,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -906,6 +906,10 @@ var (
 		Name:  "gossip.enablebroadcasttoall",
 		Usage: "Enable gossip broadcast blocks and transactions to all peers",
 	}
+	GossipBroadcastToAllCapFlag = cli.IntFlag{
+		Name:  "gossip.broadcasttoallcap",
+		Usage: "Maximum number of peers for broadcasting blocks and transactions (effective only when gossip.enablebroadcasttoall is enabled)",
+	}
 
 	// DA syncing settings
 	DASyncEnabledFlag = cli.BoolFlag{
@@ -1823,6 +1827,12 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 		cfg.GossipBroadcastToAllEnabled = ctx.GlobalBool(GossipBroadcastToAllEnabledFlag.Name)
 		log.Info("Gossip broadcast to all enabled", "enabled", cfg.GossipBroadcastToAllEnabled)
 	}
+	// Only configure the gossip broadcast-to-all flag if --gossip.enablebroadcasttoall is set to true.
+	if ctx.GlobalIsSet(GossipBroadcastToAllCapFlag.Name) && cfg.GossipBroadcastToAllEnabled {
+		cfg.GossipBroadcastToAllCap = ctx.GlobalInt(GossipBroadcastToAllCapFlag.Name)
+		log.Info("Maximum number of peers for broadcasting blocks and transactions is set", "cap", cfg.GossipBroadcastToAllCap)
+	}
+	
 
 	// Cap the cache allowance and tune the garbage collector
 	mem, err := gopsutil.VirtualMemory()

eth/backend.go

@@ -286,6 +286,7 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ether
 		DisableTxBroadcast:   config.GossipTxBroadcastDisabled,
 		DisableTxReceiving:   config.GossipTxReceivingDisabled,
 		EnableBroadcastToAll: config.GossipBroadcastToAllEnabled,
+		BroadcastToAllCap:    config.GossipBroadcastToAllCap,
 	}); err != nil {
 		return nil, err
 	}

eth/ethconfig/config.go

@@ -234,6 +234,7 @@ type Config struct {
 	GossipTxBroadcastDisabled   bool
 	GossipTxReceivingDisabled   bool
 	GossipBroadcastToAllEnabled bool
+	GossipBroadcastToAllCap     int
 }
 
 // CreateConsensusEngine creates a consensus engine for the given chain configuration.

eth/handler.go

@@ -94,9 +94,11 @@ type handlerConfig struct {
 	Whitelist         map[uint64]common.Hash    // Hard coded whitelist for sync challenged
 	ShadowForkPeerIDs []string                  // List of peer ids that take part in the shadow-fork
 
+	// Gossip configs
 	DisableTxBroadcast   bool
 	DisableTxReceiving   bool
 	EnableBroadcastToAll bool
+	BroadcastToAllCap    int
 }
 
 type handler struct {
@@ -139,6 +141,7 @@ type handler struct {
 	disableTxBroadcast   bool
 	disableTxReceiving   bool
 	enableBroadcastToAll bool
+	broadcastToAllCap    int
 }
 
 // newHandler returns a handler for all Ethereum chain management protocol.
@@ -162,6 +165,11 @@ func newHandler(config *handlerConfig) (*handler, error) {
 		disableTxReceiving:   config.DisableTxReceiving,
 		enableBroadcastToAll: config.EnableBroadcastToAll,
 	}
+	h.broadcastToAllCap = config.BroadcastToAllCap
+	if config.BroadcastToAllCap == 0 && config.EnableBroadcastToAll {
+		// Set default broadcast cap to 30 if not specified
+		h.broadcastToAllCap = 30
+	}
 	if config.Sync == downloader.FullSync {
 		// The database seems empty as the current block is the genesis. Yet the fast
 		// block is ahead, so fast sync was enabled for this node at a certain point.
@@ -483,7 +491,7 @@ func (h *handler) BroadcastBlock(block *types.Block, propagate bool) {
 		numDirect := int(math.Sqrt(float64(len(peers))))
 		// If enableBroadcastToAll is true, broadcast blocks directly to all peers (capped at 100).
 		if h.enableBroadcastToAll {
-			numDirect = min(100, len(peers))
+			numDirect = min(h.broadcastToAllCap, len(peers))
 		}
 		transfer := peers[:numDirect]
 		for _, peer := range transfer {
@@ -528,7 +536,7 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		numDirect := int(math.Sqrt(float64(len(peers))))
 		// If enableBroadcastToAll is true, broadcast transactions directly to all peers (capped at 100).
 		if h.enableBroadcastToAll {
-			numDirect = min(100, len(peers))
+			numDirect = min(h.broadcastToAllCap, len(peers))
 		}
 		for _, peer := range peers[:numDirect] {
 			txset[peer] = append(txset[peer], tx.Hash())