Remeasure round trip time (#70)

* feat(session): remeasure round trip time

periodically re-measure round trip time

* feat(session): measure rtt during keep-alive

* refactor(session): repeat measure rtt via timer

* fix(session): fix flaky test

* simplify RTT timer by using a time.Ticker

* calculate a smoothed RTT

Co-authored-by: Marten Seemann <[email protected]>

mux.go

@@ -25,6 +25,9 @@ type Config struct {
 	// KeepAliveInterval is how often to perform the keep alive
 	KeepAliveInterval time.Duration
 
+	// MeasureRTTInterval is how often to re-measure the round trip time
+	MeasureRTTInterval time.Duration
+
 	// ConnectionWriteTimeout is meant to be a "safety valve" timeout after
 	// we which will suspect a problem with the underlying connection and
 	// close it. This is only applied to writes, where's there's generally
@@ -69,6 +72,7 @@ func DefaultConfig() *Config {
 		PingBacklog:             32,
 		EnableKeepAlive:         true,
 		KeepAliveInterval:       30 * time.Second,
+		MeasureRTTInterval:      30 * time.Second,
 		ConnectionWriteTimeout:  10 * time.Second,
 		MaxIncomingStreams:      1000,
 		InitialStreamWindowSize: initialStreamWindow,
@@ -88,6 +92,10 @@ func VerifyConfig(config *Config) error {
 	if config.KeepAliveInterval == 0 {
 		return fmt.Errorf("keep-alive interval must be positive")
 	}
+	if config.MeasureRTTInterval == 0 {
+		return fmt.Errorf("measure-rtt interval must be positive")
+	}
+
 	if config.InitialStreamWindowSize < initialStreamWindow {
 		return errors.New("InitialStreamWindowSize must be larger or equal 256 kB")
 	}

session.go

@@ -161,7 +161,7 @@ func newSession(config *Config, conn net.Conn, client bool, readBuf int, newMemo
 	}
 	go s.recv()
 	go s.send()
-	go s.measureRTT()
+	go s.startMeasureRTT()
 	return s
 }
 
@@ -341,7 +341,25 @@ func (s *Session) measureRTT() {
 	if err != nil {
 		return
 	}
-	atomic.StoreInt64(&s.rtt, rtt.Nanoseconds())
+	if !atomic.CompareAndSwapInt64(&s.rtt, 0, rtt.Nanoseconds()) {
+		prev := atomic.LoadInt64(&s.rtt)
+		smoothedRTT := prev/2 + rtt.Nanoseconds()/2
+		atomic.StoreInt64(&s.rtt, smoothedRTT)
+	}
+}
+
+func (s *Session) startMeasureRTT() {
+	s.measureRTT()
+	t := time.NewTicker(s.config.MeasureRTTInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-s.CloseChan():
+			return
+		case <-t.C:
+			s.measureRTT()
+		}
+	}
 }
 
 // 0 if we don't yet have a measurement

session_test.go

@@ -238,6 +238,8 @@ func TestPing(t *testing.T) {
 	defer server.Close()
 
 	clientConn := client.conn.(*pipeConn)
+	time.Sleep(time.Millisecond)
+
 	clientConn.BlockWrites()
 	go func() {
 		time.Sleep(10 * time.Millisecond)