Increase websocket timeout to send ping to adapt nginx's websocket proxy

[angwangiot]

当我尝试用nginx代理grpc-web中的websocket时，由于nginx有检测断开机制，导致超过60秒没有推送消息会断开。
nginx官网的解释时这样的：http://nginx.org/en/docs/http/websocket.html

When I try to use nginx to proxy the websocket in grpc-web, since nginx has a detection disconnection mechanism, no push message will be disconnected for more than 60 seconds.
Nginx official website explains this

Changes

根据nginx的官网说明，我在服务器的推送上增加了超时发送Ping的功能。

According to the official website of nginx, I added the function of timeout to send Ping on the push of the server.

Verification

经过修改，已经可以实现nginx代理websocket并且不自动断开了。

After modification, the nginx proxy websocket can be implemented and not automatically disconnected.

Sorry, My English is not good, so I did not use English instructions.

(Translations courtesy of google translate - @johanbrandhorst)

[johanbrandhorst]

Thanks for this, seems reasonable enough, there's a few errors in the current implementation.

[johanbrandhorst]

Instead of keeping a tickerCount, how about we store a timestamp lastMessage time.Time and instead of incrementing the counter and resetting it, we can just use lastMessageTime = time.Now(), and when checking if we need to send a ping we can do t.After(lastMessageTime.Add(timeoutInterval))? timeoutInterval would be time.Duration.

[angwangiot]

Thank you for your advice. I'll change it.

[angwangiot]

您好，我尝试使用lastMessageTime代替tickerCount来处理校验是否超时发送Ping。但是使用ticker的思路导致会频繁调用ticker的超时函数进行额外的计算，而且精度会受ticker间隔大小的影响，所以我尝试了使用timer.Reset()方法来重置定时器。目前看起来工作是正常的，但是不知道这样实现是否有未考虑到的隐患。
Hello, I am trying to use lastMessageTime instead of tickerCount to handle whether the check is timed to send a ping. But the idea of using ticker leads to frequent calls to ticker's timeout function for additional calculations, and the precision is affected by the size of the ticker interval, so I tried to reset the timer using the timer.Reset() method. At the moment it seems that the work is normal, but I don't know if there are hidden dangers in this implementation.
（Translations courtesy of google translate）

[johanbrandhorst]

I think there was a slight misunderstanding, sorry. I added some more comments on where we can make this nicer.

[johanbrandhorst]

I think we want a ticker, not a timer. time.Ticker.

Suggested change

	timer *time.Timer
	ticker *time.Ticker

[johanbrandhorst]

We also need a lastMessageTime time.Time so that in the channel loop we can compare if tick.After(w.lastMessageTime.Add(w.timeOutInterval)).

[johanbrandhorst]

And we need a mutex for lastMessageTime.

[angwangiot]

我想，我应该理解了您之前的建议，也确实这么做了（可以参考下面的websocket_wrapper.go未提交之前的修改内容）。
只是在改完之后有了其他的想法，导致我又进行了其他的改动。改动的原因是这样的：
我的意思是，用ticker那样做确实没问题，但是这样有两个地方不是很满意。第一，ticker会在每秒都触发然后进行一次计算，这样会进行比较多的无用的计算。比如，我设置间隔为30秒，那么，ticker进行了30次计算，只有最后一次的计算才有效，除了最后一次的计算，其他的计算与比较相当于是浪费的。第二，因为ticker本身有1秒的间隔，所以，他的超时发送ping消息的间隔并不是很准确，有可能比设置的时间还长一些。比如，我设置超时间隔为30秒，可能在未发送数据后的第31秒才会发送Ping消息。

I think, I should understand your previous suggestions, and I did. (Refer to the following websocket_wrapper.go for the changes that were not submitted before).
Just after the change, I had other ideas, which led me to make other changes. The reason for the change is this:
I mean, it's okay to do it with ticker, but there are two places that are not very satisfactory. First, the ticker will trigger every second and then perform a calculation, which will result in more useless calculations. For example, if I set the interval to 30 seconds, then the ticker performs 30 calculations, only the last calculation is valid. Except for the last calculation, other calculations and comparisons are equivalent to waste. Second, because the ticker itself has a 1 second interval, the interval between sending ping messages over time is not very accurate, and may be longer than the set time. For example, I set the timeout interval to 30 seconds, and the Ping message may be sent 31 seconds after the data is not sent.

（Translations courtesy of google translate）

package grpcweb

import (
	"bufio"
	"bytes"
	"context"
	"encoding/binary"
	"errors"
	"io"
	"net/http"
	"net/textproto"
	"strings"
	// "sync"
	"time"

	"github.com/gorilla/websocket"
	"golang.org/x/net/http2"
)

type webSocketResponseWriter struct {
	writtenHeaders  bool
	wsConn          *websocket.Conn
	headers         http.Header
	flushedHeaders  http.Header
	timeOutInterval time.Duration
	// lastMessageTime     time.Time
	// lastMessageTimeLock *sync.Mutex
	timer *time.Timer
}

func newWebSocketResponseWriter(wsConn *websocket.Conn) *webSocketResponseWriter {
	return &webSocketResponseWriter{
		writtenHeaders: false,
		headers:        make(http.Header),
		flushedHeaders: make(http.Header),
		wsConn:         wsConn,
	}
}

func (w *webSocketResponseWriter) EnablePing(timeOutInterval time.Duration) {
	if timeOutInterval < time.Second {
		return
	}
	w.timeOutInterval = timeOutInterval
	// w.lastMessageTimeLock = &sync.Mutex{}
	w.timer = time.NewTimer(w.timeOutInterval)
	dispose := make(chan bool)
	w.wsConn.SetCloseHandler(func(code int, text string) error {
		close(dispose)
		return nil
	})
	go w.ping(dispose)
}

func (w *webSocketResponseWriter) ping(dispose chan bool) {
	if dispose == nil {
		return
	}
	// ticker := time.NewTicker(time.Second)
	// defer ticker.Stop()
	defer w.timer.Stop()
	for {
		select {
		case <-dispose:
			return
		// case t := <-ticker.C:
		// 	var sendPing = false
		// 	w.lastMessageTimeLock.Lock()
		// 	if t.After(w.lastMessageTime.Add(w.timeOutInterval)) {
		// 		sendPing = true
		// 		w.lastMessageTime = time.Now()
		// 	}
		// 	w.lastMessageTimeLock.Unlock()
		// 	if sendPing {
		// 		w.wsConn.WriteMessage(websocket.PingMessage, []byte{})
		// 	}
		case <-w.timer.C:
			w.timer.Reset(w.timeOutInterval)
			w.wsConn.WriteMessage(websocket.PingMessage, []byte{})
		}
	}
}

func (w *webSocketResponseWriter) Write(b []byte) (int, error) {
	if !w.writtenHeaders {
		w.WriteHeader(http.StatusOK)
	}
	if w.timeOutInterval > time.Second && w.timer != nil {
		// w.lastMessageTimeLock.Lock()
		// w.lastMessageTime = time.Now()
		// w.lastMessageTimeLock.Unlock()
		w.timer.Reset(w.timeOutInterval)
	}
	return len(b), w.wsConn.WriteMessage(websocket.BinaryMessage, b)
}

[johanbrandhorst]

I'm confused. The ticker will only tick every timeOutInterval. It will not tick every second. With an interval of 30 seconds, what will happen is that every 30 seconds it will check if there was at least 30 seconds since the last message was sent. Sure, if you send a message 0.01s after the ping message, it will take 59.9s until the next ping is sent. So worst case, you just have to configure your timeout interval to 15s if you want a message at least every 30s.

[angwangiot]

是的，这是我关心的。如果我是用户，我会设置一段时间的超时间隔。结果应该是ping消息发送这么长时间以满足nginx的超时要求，而不是可能是超时间隔的两倍。这需要计算以确定满足nginx要求所需的时间，并且发送Pi​​ng消息的间隔不固定，它是设置超时间隔的1到2倍。

Yes, this is what I care about. If I am a user, I will set a timeout interval for a while. The result should be that the ping message is sent for such a long time to meet the timeout requirement of nginx, rather than possibly twice the timeout interval. This requires calculations to determine the time required to meet the nginx requirements, and the interval at which the Pi ng message is sent is not fixed, it is 1 to 2 times the set timeout interval.
（Translations courtesy of google translate）

[johanbrandhorst]

OK, but that is simply a matter of user documentation. We can even ask for a timeout interval and set the ticker to half of that to guarantee a message between X/2 and X. I think your current implementation won't work, as you are reading from the time.Timer in one goroutine and resetting it in another. The documentation says you should only reset the timer if you have already read from it:

Reset should be invoked only on stopped or expired timers with drained channels. If a program has already received a value from t.C, the timer is known to have expired and the channel drained, so t.Reset can be used directly. If a program has not yet received a value from t.C, however, the timer must be stopped and—if Stop reports that the timer expired before being stopped—the channel explicitly drained:

https://golang.org/pkg/time/#Timer.Reset. I think using a ticker is the best option.

[angwangiot]

好的，我明白了，感谢您耐心的解释。我先将timer改为ticker，确保此功能可以使用。
不过，我发现有个库可以规避原生timer.Reset()的限制，地址在这：https://github.com/desertbit/timer 。或者，可以考虑通过channel的方式在ping()中进行reset来保证读取和reset在同一个goroutine 。如下：
Ok, I understand, thank you for your patient explanation. I first changed the timer to ticker to make sure this feature is available.
However, I found that there is a library that can circumvent the limitations of the native timer.Reset(), where the address is: https://github.com/desertbit/timer. Or, we can consider resetting in ping() by channel to ensure that the read and reset operations are in the same goroutine.（Translations courtesy of google translate）
as follows：

func (w *webSocketResponseWriter) ping(dispose chan bool) {
	if dispose == nil {
		return
	}
	defer w.timer.Stop()
	for {
		select {
		case <-dispose:
			return
		case <-resetTimer:
			if !w.timer.Stop() {
 				<-w.timer.C
 			}
 			w.timer.Reset(w.timeOutInterval)
		case <-w.timer.C:
			w.timer.Reset(w.timeOutInterval)
			w.wsConn.WriteMessage(websocket.PingMessage, []byte{})
		}
	}
}

What do you think?

[johanbrandhorst]

Lets use the third party timer package, I don't like the extra complexity added by this 2nd channel implementation. Please change it and I will take another look.

[johanbrandhorst]

LGTM, how do we best test this? Could you build this and test it in your environment to confirm it works for you before we merge this?

[johanbrandhorst]

Great work, changes look good! I'll see if I can coerce CI to agree, it's a bit flaky sometimes.

[johanbrandhorst]

Thanks for this contribution @angwangiot, great work!

[angwangiot]

It's my pleasure to participate in development and to be able to contribute.