From ca9b09f72b2946d110fb89bf8709fa740e96527c Mon Sep 17 00:00:00 2001 From: devopsbo3 <69951731+devopsbo3@users.noreply.github.com> Date: Fri, 10 Nov 2023 12:27:53 -0600 Subject: [PATCH] Revert "rpc: attempt to fix ping/pong logic race (#27733)" This reverts commit 7588f79bd8ac27bd9a103c16f78be97bffa3ca3d. --- rpc/websocket.go | 43 +++++++++++++++++-------------------------- 1 file changed, 17 insertions(+), 26 deletions(-) diff --git a/rpc/websocket.go b/rpc/websocket.go index 86cf50594c..b1213fdfa6 100644 --- a/rpc/websocket.go +++ b/rpc/websocket.go @@ -278,21 +278,24 @@ type websocketCodec struct { conn *websocket.Conn info PeerInfo - wg sync.WaitGroup - pingReset chan struct{} - pongReceived chan struct{} + wg sync.WaitGroup + pingReset chan struct{} } func newWebsocketCodec(conn *websocket.Conn, host string, req http.Header) ServerCodec { conn.SetReadLimit(wsMessageSizeLimit) + conn.SetPongHandler(func(appData string) error { + conn.SetReadDeadline(time.Time{}) + return nil + }) + encode := func(v interface{}, isErrorResponse bool) error { return conn.WriteJSON(v) } wc := &websocketCodec{ - jsonCodec: NewFuncCodec(conn, encode, conn.ReadJSON).(*jsonCodec), - conn: conn, - pingReset: make(chan struct{}, 1), - pongReceived: make(chan struct{}), + jsonCodec: NewFuncCodec(conn, encode, conn.ReadJSON).(*jsonCodec), + conn: conn, + pingReset: make(chan struct{}, 1), info: PeerInfo{ Transport: "ws", RemoteAddr: conn.RemoteAddr().String(), @@ -303,13 +306,6 @@ func newWebsocketCodec(conn *websocket.Conn, host string, req http.Header) Serve wc.info.HTTP.Origin = req.Get("Origin") wc.info.HTTP.UserAgent = req.Get("User-Agent") // Start pinger. - conn.SetPongHandler(func(appData string) error { - select { - case wc.pongReceived <- struct{}{}: - case <-wc.closed(): - } - return nil - }) wc.wg.Add(1) go wc.pingLoop() return wc @@ -338,31 +334,26 @@ func (wc *websocketCodec) writeJSON(ctx context.Context, v interface{}, isError // pingLoop sends periodic ping frames when the connection is idle. func (wc *websocketCodec) pingLoop() { - var pingTimer = time.NewTimer(wsPingInterval) + var timer = time.NewTimer(wsPingInterval) defer wc.wg.Done() - defer pingTimer.Stop() + defer timer.Stop() for { select { case <-wc.closed(): return - case <-wc.pingReset: - if !pingTimer.Stop() { - <-pingTimer.C + if !timer.Stop() { + <-timer.C } - pingTimer.Reset(wsPingInterval) - - case <-pingTimer.C: + timer.Reset(wsPingInterval) + case <-timer.C: wc.jsonCodec.encMu.Lock() wc.conn.SetWriteDeadline(time.Now().Add(wsPingWriteTimeout)) wc.conn.WriteMessage(websocket.PingMessage, nil) wc.conn.SetReadDeadline(time.Now().Add(wsPongTimeout)) wc.jsonCodec.encMu.Unlock() - pingTimer.Reset(wsPingInterval) - - case <-wc.pongReceived: - wc.conn.SetReadDeadline(time.Time{}) + timer.Reset(wsPingInterval) } } }