stop timer before reset

internal/blocksync/pool.go

@@ -459,6 +459,7 @@ func (pool *BlockPool) sendError(err error, peerID types.NodeID) {
 }
 
 // for debugging purposes
+//
 //nolint:unused
 func (pool *BlockPool) debug() string {
 	pool.mtx.Lock()
@@ -519,6 +520,7 @@ func (peer *bpPeer) resetTimeout() {
 	if peer.timeout == nil {
 		peer.timeout = time.AfterFunc(peerTimeout, peer.onTimeout)
 	} else {
+		peer.timeout.Stop()
 		peer.timeout.Reset(peerTimeout)
 	}
 }

internal/consensus/ticker.go

@@ -114,6 +114,7 @@ func (t *timeoutTicker) timeoutRoutine(ctx context.Context) {
 			// update timeoutInfo and reset timer
 			// NOTE time.Timer allows duration to be non-positive
 			ti = newti
+			t.timer.Stop()
 			t.timer.Reset(ti.Duration)
 			t.logger.Debug("Internal state machine timeout scheduled", "duration", ti.Duration, "height", ti.Height, "round", ti.Round, "step", ti.Step)
 		case <-t.timer.C:

internal/evidence/reactor.go

@@ -309,6 +309,7 @@ func (r *Reactor) broadcastEvidenceLoop(ctx context.Context, peerID types.NodeID
 
 		case <-next.NextWaitChan():
 			next = next.Next()
+			timer.Stop()
 
 		case <-ctx.Done():
 			return

internal/p2p/conn/connection.go

@@ -676,11 +676,13 @@ func newChannel(conn *MConnection, desc ChannelDescriptor) *channel {
 // Goroutine-safe
 // Times out (and returns false) after defaultSendTimeout
 func (ch *channel) sendBytes(bytes []byte) bool {
+	timer := time.NewTimer(defaultSendTimeout)
+	defer timer.Stop()
 	select {
 	case ch.sendQueue <- bytes:
 		atomic.AddInt32(&ch.sendQueueSize, 1)
 		return true
-	case <-time.After(defaultSendTimeout):
+	case <-timer.C:
 		return false
 	}
 }

internal/p2p/pex/reactor.go

@@ -57,7 +57,7 @@ type NoPeersAvailableError struct {
 }
 
 func (e *NoPeersAvailableError) Error() string {
-    return fmt.Sprintf("no available peers to send a PEX request to (retrying)")
+	return fmt.Sprintf("no available peers to send a PEX request to (retrying)")
 }
 
 // TODO: We should decide whether we want channel descriptors to be housed
@@ -114,7 +114,7 @@ type Reactor struct {
 	channel *p2p.Channel
 
 	// Used to signal a restart the node on the application level
-	restartCh chan struct{}
+	restartCh                     chan struct{}
 	restartNoAvailablePeersWindow time.Duration
 }
 
@@ -127,14 +127,14 @@ func NewReactor(
 	selfRemediationConfig *config.SelfRemediationConfig,
 ) *Reactor {
 	r := &Reactor{
-		logger:               logger,
-		peerManager:          peerManager,
-		peerEvents:           peerEvents,
-		availablePeers:       make(map[types.NodeID]struct{}),
-		lastNoAvailablePeers: time.Time{},
-		requestsSent:         make(map[types.NodeID]struct{}),
-		lastReceivedRequests: make(map[types.NodeID]time.Time),
-		restartCh:            restartCh,
+		logger:                        logger,
+		peerManager:                   peerManager,
+		peerEvents:                    peerEvents,
+		availablePeers:                make(map[types.NodeID]struct{}),
+		lastNoAvailablePeers:          time.Time{},
+		requestsSent:                  make(map[types.NodeID]struct{}),
+		lastReceivedRequests:          make(map[types.NodeID]time.Time),
+		restartCh:                     restartCh,
 		restartNoAvailablePeersWindow: time.Duration(selfRemediationConfig.P2pNoPeersRestarWindowSeconds) * time.Second,
 	}
 
@@ -235,6 +235,8 @@ func (r *Reactor) processPexCh(ctx context.Context, pexCh *p2p.Channel) {
 				// We got a useful result; update the poll timer.
 				nextPeerRequest = dur
 			}
+
+			timer.Stop()
 		}
 	}
 }

internal/p2p/rqueue.go

@@ -54,6 +54,7 @@ func (q *simpleQueue) run(ctx context.Context) {
 	pq := make(priorityQueue, 0, q.maxSize)
 	heap.Init(&pq)
 	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
 	// must have a buffer of exactly one because both sides of
 	// this channel are used in this loop, and simply signals adds
 	// to the heap