feat: optimizations

Author: minhd-vu

cmd/p2p/sensor/sensor.go

@@ -58,6 +58,7 @@ type (
 		ShouldBroadcastTxHashes      bool
 		ShouldBroadcastBlocks        bool
 		ShouldBroadcastBlockHashes   bool
+		BroadcastWorkers             int
 		ShouldRunPprof               bool
 		PprofPort                    uint
 		ShouldRunPrometheus          bool
@@ -218,6 +219,7 @@ var SensorCmd = &cobra.Command{
 			ShouldBroadcastTxHashes:    inputSensorParams.ShouldBroadcastTxHashes,
 			ShouldBroadcastBlocks:      inputSensorParams.ShouldBroadcastBlocks,
 			ShouldBroadcastBlockHashes: inputSensorParams.ShouldBroadcastBlockHashes,
+			BroadcastWorkers:           inputSensorParams.BroadcastWorkers,
 		})
 
 		opts := p2p.EthProtocolOptions{
@@ -491,6 +493,7 @@ will result in less chance of missing data but can significantly increase memory
 	f.BoolVar(&inputSensorParams.ShouldBroadcastTxHashes, "broadcast-tx-hashes", false, "broadcast transaction hashes to peers")
 	f.BoolVar(&inputSensorParams.ShouldBroadcastBlocks, "broadcast-blocks", false, "broadcast full blocks to peers")
 	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.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")

doc/polycli_p2p_sensor.md

@@ -126,6 +126,7 @@ polycli p2p sensor amoy-nodes.json \
       --broadcast-blocks              broadcast full blocks to peers
       --broadcast-tx-hashes           broadcast transaction hashes to peers
       --broadcast-txs                 broadcast full transactions to peers
+      --broadcast-workers int         number of concurrent broadcast workers (default 4)
       --database string               which database to persist data to, options are:
                                         - datastore (GCP Datastore)
                                         - json (output to stdout)

p2p/conns.go

@@ -1,6 +1,7 @@
 package p2p
 
 import (
+	"bytes"
 	"math/big"
 	"sync"
 	"sync/atomic"
@@ -35,6 +36,7 @@ type ConnsOptions struct {
 	ShouldBroadcastTxHashes    bool
 	ShouldBroadcastBlocks      bool
 	ShouldBroadcastBlockHashes bool
+	BroadcastWorkers           int
 }
 
 // Conns manages a collection of active peer connections for transaction broadcasting.
@@ -94,7 +96,15 @@ func NewConns(opts ConnsOptions) *Conns {
 		shouldBroadcastBlockHashes: opts.ShouldBroadcastBlockHashes,
 		txBroadcastCh:              make(chan types.Transactions, 100000),
 	}
-	go c.txBroadcastLoop()
+
+	workers := opts.BroadcastWorkers
+	if workers <= 0 {
+		workers = 4
+	}
+	for i := 0; i < workers; i++ {
+		go c.txBroadcastLoop()
+	}
+
 	return c
 }
 
@@ -212,44 +222,48 @@ func (c *Conns) BroadcastTxs(txs types.Transactions) int {
 
 // broadcastTxs sends RPC-submitted transactions to all peers via TransactionsMsg.
 // Used by txBroadcastLoop to process transactions from EnqueueTxBroadcast.
-// Has a timeout to prevent slow peers from blocking the broadcast loop.
-func (c *Conns) broadcastTxs(txs types.Transactions, hashes []common.Hash, peers []*conn) int {
+func (c *Conns) broadcastTxs(txs types.Transactions, hashes []common.Hash, peers []*conn) {
 	rawList, err := rlp.EncodeToRawList([]*types.Transaction(txs))
 	if err != nil {
 		log.Debug().Err(err).Msg("Failed to encode transactions")
-		return 0
+		return
 	}
 	packet := &eth.TransactionsPacket{RawList: rawList}
 
-	var count atomic.Int32
-	var wg sync.WaitGroup
+	// Pre-encode the entire message once to avoid re-encoding for each peer.
+	encodedMsg, err := rlp.EncodeToBytes(packet)
+	if err != nil {
+		log.Debug().Err(err).Msg("Failed to encode message")
+		return
+	}
+	msgSize := uint32(len(encodedMsg))
+
+	// Cache loop-invariant values
+	msgName := packet.Name()
+	txCount := float64(len(txs))
+	rebroadcasting := c.shouldBroadcastTx || c.shouldBroadcastTxHashes
 
 	for _, peer := range peers {
-		wg.Go(func() {
-			peer.countMsgSent(packet.Name(), float64(len(txs)))
-			if err := ethp2p.Send(peer.rw, eth.TransactionsMsg, packet); err != nil {
+		go func(peer *conn) {
+			peer.countMsgSent(msgName, txCount)
+
+			// Use WriteMsg directly with pre-encoded bytes instead of Send()
+			msg := ethp2p.Msg{
+				Code:    eth.TransactionsMsg,
+				Size:    msgSize,
+				Payload: bytes.NewReader(encodedMsg),
+			}
+			if err := peer.rw.WriteMsg(msg); err != nil {
 				peer.logger.Debug().Err(err).Msg("Failed to send transactions")
 				return
 			}
-			peer.addKnownTxHashes(hashes)
-			count.Add(1)
-		})
-	}
-
-	// Wait with timeout
-	done := make(chan struct{})
-	go func() {
-		wg.Wait()
-		close(done)
-	}()
 
-	select {
-	case <-done:
-	case <-time.After(5 * time.Second):
-		log.Warn().Int("peers", len(peers)).Msg("Broadcast timed out")
+			// Only track known hashes if rebroadcasting is enabled
+			if rebroadcasting {
+				peer.addKnownTxHashes(hashes)
+			}
+		}(peer)
 	}
-
-	return int(count.Load())
 }
 
 // txBroadcastLoop is the worker that drains the broadcast channel and sends
@@ -297,7 +311,7 @@ func (c *Conns) pullTxBatch() (types.Transactions, []common.Hash) {
 	}
 
 	// Drain more until max size or timeout
-	timer := time.NewTimer(time.Second)
+	timer := time.NewTimer(100 * time.Millisecond)
 	defer timer.Stop()
 
 	for batchSize < maxTxPacketSize {