feat: add configurable tx broadcast queue size

Author: minhd-vu

cmd/p2p/sensor/sensor.go

@@ -59,6 +59,7 @@ type (
 		ShouldBroadcastBlockHashes   bool
 		BroadcastWorkers             int
 		TxBatchTimeout               time.Duration
+		TxBroadcastQueueSize         int
 		ShouldRunPprof               bool
 		PprofPort                    uint
 		ShouldRunPrometheus          bool
@@ -221,6 +222,7 @@ var SensorCmd = &cobra.Command{
 			ShouldBroadcastBlockHashes: inputSensorParams.ShouldBroadcastBlockHashes,
 			BroadcastWorkers:           inputSensorParams.BroadcastWorkers,
 			TxBatchTimeout:             inputSensorParams.TxBatchTimeout,
+			TxBroadcastQueueSize:       inputSensorParams.TxBroadcastQueueSize,
 		})
 
 		opts := p2p.EthProtocolOptions{
@@ -497,6 +499,7 @@ will result in less chance of missing data but can significantly increase memory
 	f.BoolVar(&inputSensorParams.ShouldBroadcastBlockHashes, "broadcast-block-hashes", false, "broadcast block hashes to peers")
 	f.IntVar(&inputSensorParams.BroadcastWorkers, "broadcast-workers", 4, "number of concurrent broadcast workers")
 	f.DurationVar(&inputSensorParams.TxBatchTimeout, "tx-batch-timeout", 500*time.Millisecond, "timeout for batching transactions before broadcast")
+	f.IntVar(&inputSensorParams.TxBroadcastQueueSize, "tx-broadcast-queue-size", 100000, "capacity of transaction broadcast queue")
 	f.BoolVar(&inputSensorParams.ShouldRunPprof, "pprof", false, "run pprof server")
 	f.UintVar(&inputSensorParams.PprofPort, "pprof-port", 6060, "port pprof runs on")
 	f.BoolVar(&inputSensorParams.ShouldRunPrometheus, "prom", true, "run Prometheus server")

p2p/conns.go

@@ -38,6 +38,7 @@ type ConnsOptions struct {
 	ShouldBroadcastBlockHashes bool
 	BroadcastWorkers           int
 	TxBatchTimeout             time.Duration
+	TxBroadcastQueueSize       int
 }
 
 // Conns manages a collection of active peer connections for transaction broadcasting.
@@ -91,6 +92,11 @@ func NewConns(opts ConnsOptions) *Conns {
 		txBatchTimeout = 500 * time.Millisecond
 	}
 
+	txBroadcastQueueSize := opts.TxBroadcastQueueSize
+	if txBroadcastQueueSize <= 0 {
+		txBroadcastQueueSize = 100000
+	}
+
 	c := &Conns{
 		conns:                      make(map[string]*conn),
 		blocks:                     ds.NewLRU[common.Hash, BlockCache](opts.BlocksCache),
@@ -103,7 +109,7 @@ func NewConns(opts ConnsOptions) *Conns {
 		shouldBroadcastTxHashes:    opts.ShouldBroadcastTxHashes,
 		shouldBroadcastBlocks:      opts.ShouldBroadcastBlocks,
 		shouldBroadcastBlockHashes: opts.ShouldBroadcastBlockHashes,
-		txBroadcastCh:              make(chan types.Transactions, 100000),
+		txBroadcastCh:              make(chan types.Transactions, txBroadcastQueueSize),
 		txBatchTimeout:             txBatchTimeout,
 	}