From 6b42535909afdcaedf291090a5a8cfd21739ff77 Mon Sep 17 00:00:00 2001
From: Bjorn Neergaard <bneergaard@mirantis.com>
Date: Thu, 13 Apr 2023 10:49:30 -0600
Subject: [PATCH] man/state/raft/trans/peer: use AfterFunc for context watchdog

Address
https://github.com/moby/swarmkit/pull/3122#discussion_r1161904910 by
taking the recommendation to reduce resource churn.

Signed-off-by: Bjorn Neergaard <bneergaard@mirantis.com>
---
 manager/state/raft/transport/peer.go | 30 +++++++---------------------
 1 file changed, 7 insertions(+), 23 deletions(-)

diff --git a/manager/state/raft/transport/peer.go b/manager/state/raft/transport/peer.go
index 23eb6748b6..071f6dc76f 100644
--- a/manager/state/raft/transport/peer.go
+++ b/manager/state/raft/transport/peer.go
@@ -229,18 +229,10 @@ func (p *peer) sendProcessMessage(ctx context.Context, m raftpb.Message) error {
 	//
 	// By doing it as this watchdog-type structure, we can time out individual
 	// operations by canceling the context on our own terms.
-	bump := make(chan struct{})
-	go func() {
-		for {
-			select {
-			case <-bump:
-			case <-time.After(p.tr.config.SendTimeout):
-				cancel()
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
+	t := time.AfterFunc(p.tr.config.SendTimeout, cancel)
+	defer t.Stop()
+
+	bump := func() { t.Reset(p.tr.config.SendTimeout) }
 
 	var err error
 	var stream api.Raft_StreamRaftMessageClient
@@ -260,22 +252,14 @@ func (p *peer) sendProcessMessage(ctx context.Context, m raftpb.Message) error {
 		// Stream
 		for _, msg := range msgs {
 			err = stream.Send(&msg)
-
-			// If the send succeeds, bump the watchdog timer.
-			//
-			// We cannot just do a naked send to the bump channel. If we try to
-			// send, for example, and the timer has elapsed, then the context
-			// will have been canceled, the watchdog loop will have exited, and
-			// there would be no receiver. We'd block here forever.
-			select {
-			case bump <- struct{}{}:
-			case <-ctx.Done():
-			}
 			if err != nil {
 				log.G(ctx).WithError(err).Error("error streaming message to peer")
 				stream.CloseAndRecv()
 				break
 			}
+
+			// If the send succeeds, bump the watchdog timer.
+			bump()
 		}
 
 		// Finished sending all the messages.