chore: clean up

Author: minhd-vu

p2p/conns.go

@@ -69,11 +69,10 @@ type Conns struct {
 	shouldBroadcastBlockHashes bool
 
 	// txBroadcastQueue is an unbounded queue of transactions for async broadcast
-	txBroadcastQueue []types.Transactions
-	txBroadcastMu    sync.Mutex
-	txBroadcastCond  *sync.Cond
-	// closeCh signals the broadcast worker to stop
-	closeCh chan struct{}
+	txBroadcastQueue  []types.Transactions
+	txBroadcastMu     sync.Mutex
+	txBroadcastCond   *sync.Cond
+	txBroadcastClosed bool
 }
 
 // NewConns creates a new connection manager with a blocks cache.
@@ -96,7 +95,6 @@ func NewConns(opts ConnsOptions) *Conns {
 		shouldBroadcastTxHashes:    opts.ShouldBroadcastTxHashes,
 		shouldBroadcastBlocks:      opts.ShouldBroadcastBlocks,
 		shouldBroadcastBlockHashes: opts.ShouldBroadcastBlockHashes,
-		closeCh:                    make(chan struct{}),
 	}
 	c.txBroadcastCond = sync.NewCond(&c.txBroadcastMu)
 	go c.txBroadcastLoop()
@@ -272,13 +270,10 @@ func (c *Conns) broadcastTxs(txs types.Transactions, hashes []common.Hash, peers
 // While broadcasting, more transactions accumulate in the queue for the next batch.
 func (c *Conns) txBroadcastLoop() {
 	for {
-		batch, done := c.pullTxBatch()
-		if done {
+		batch := c.pullTxBatch()
+		if batch == nil {
 			return
 		}
-		if len(batch) == 0 {
-			continue
-		}
 
 		peers := c.snapshotPeers()
 		if len(peers) == 0 {
@@ -291,25 +286,22 @@ func (c *Conns) txBroadcastLoop() {
 		}
 
 		c.broadcastTxs(batch, hashes, peers)
-		log.Info().Int("txs", len(batch)).Int("peers", len(peers)).Msg("Broadcasted batch")
 	}
 }
 
 // pullTxBatch pulls transactions from the queue up to maxTxPacketSize.
-// Returns the batch and whether the worker should exit.
-func (c *Conns) pullTxBatch() (types.Transactions, bool) {
+// Returns nil when the worker should exit.
+func (c *Conns) pullTxBatch() types.Transactions {
 	c.txBroadcastMu.Lock()
 	defer c.txBroadcastMu.Unlock()
 
 	// Wait for transactions or close signal
-	for len(c.txBroadcastQueue) == 0 {
-		select {
-		case <-c.closeCh:
-			return nil, true
-		default:
-		}
+	for len(c.txBroadcastQueue) == 0 && !c.txBroadcastClosed {
 		c.txBroadcastCond.Wait()
 	}
+	if c.txBroadcastClosed {
+		return nil
+	}
 
 	var batch types.Transactions
 	var batchSize uint64
@@ -325,13 +317,8 @@ func (c *Conns) pullTxBatch() (types.Transactions, bool) {
 
 			// If adding this tx would exceed max size and we have something, stop
 			if len(batch) > 0 && batchSize+txSize > maxTxPacketSize {
-				// Put remaining txs back at front of queue
-				if len(txs) > 0 {
-					c.txBroadcastQueue[0] = txs
-				} else {
-					c.txBroadcastQueue = c.txBroadcastQueue[1:]
-				}
-				return batch, false
+				c.txBroadcastQueue[0] = txs
+				return batch
 			}
 
 			batch = append(batch, tx)
@@ -343,7 +330,7 @@ func (c *Conns) pullTxBatch() (types.Transactions, bool) {
 		c.txBroadcastQueue = c.txBroadcastQueue[1:]
 	}
 
-	return batch, false
+	return batch
 }
 
 // EnqueueTxBroadcast adds transactions to the broadcast queue for async sending.
@@ -357,7 +344,9 @@ func (c *Conns) EnqueueTxBroadcast(txs types.Transactions) {
 
 // Close stops the broadcast worker.
 func (c *Conns) Close() {
-	close(c.closeCh)
+	c.txBroadcastMu.Lock()
+	c.txBroadcastClosed = true
+	c.txBroadcastMu.Unlock()
 	c.txBroadcastCond.Broadcast()
 }