rework rtt stats

client.go

@@ -20,6 +20,7 @@ import (
 	"github.com/connet-dev/connet/netc"
 	"github.com/connet-dev/connet/pb"
 	"github.com/connet-dev/connet/pbs"
+	"github.com/connet-dev/connet/quicc"
 	"github.com/connet-dev/connet/statusc"
 	"github.com/klev-dev/kleverr"
 	"github.com/quic-go/quic-go"
@@ -88,7 +89,8 @@ func (c *Client) Run(ctx context.Context) error {
 
 	c.logger.Debug("start quic listener")
 	transport := &quic.Transport{
-		Conn: udpConn,
+		Conn:        udpConn,
+		ConnContext: quicc.RTTContext,
 		// TODO review other options
 	}
 	defer transport.Close()
@@ -170,7 +172,8 @@ func (c *Client) connect(ctx context.Context, transport *quic.Transport, retoken
 		RootCAs:    c.controlCAs,
 		NextProtos: []string{"connet"},
 	}, &quic.Config{
-		KeepAlivePeriod: 25 * time.Second,
+		MaxIdleTimeout:  20 * time.Second,
+		KeepAlivePeriod: 10 * time.Second,
 	})
 	if err != nil {
 		return retConnect(err)

client/direct.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/connet-dev/connet/certc"
 	"github.com/connet-dev/connet/pb"
+	"github.com/connet-dev/connet/quicc"
 	"github.com/klev-dev/kleverr"
 	"github.com/quic-go/quic-go"
 	"golang.org/x/sync/errgroup"
@@ -139,7 +140,9 @@ func (s *DirectServer) runServer(ctx context.Context) error {
 	}
 
 	l, err := s.transport.Listen(tlsConf, &quic.Config{
-		KeepAlivePeriod: 25 * time.Second,
+		MaxIdleTimeout:  20 * time.Second,
+		KeepAlivePeriod: 10 * time.Second,
+		Tracer:          quicc.RTTTracer,
 	})
 	if err != nil {
 		return err

client/peer_direct.go

@@ -16,6 +16,7 @@ import (
 	"github.com/connet-dev/connet/pb"
 	"github.com/connet-dev/connet/pbc"
 	"github.com/connet-dev/connet/pbs"
+	"github.com/connet-dev/connet/quicc"
 	"github.com/klev-dev/kleverr"
 	"github.com/quic-go/quic-go"
 	"golang.org/x/sync/errgroup"
@@ -227,6 +228,9 @@ func (p *directPeerIncoming) keepalive(ctx context.Context, conn quic.Connection
 			if err := p.heartbeat(stream); err != nil {
 				return err
 			}
+			if rttStats := quicc.RTTStats(conn); rttStats != nil {
+				p.parent.logger.Debug("rtt-in", "last", rttStats.LatestRTT(), "smoothed", rttStats.SmoothedRTT())
+			}
 		}
 	})
 
@@ -307,13 +311,15 @@ func (p *directPeerOutgoing) connect(ctx context.Context) (quic.Connection, quic
 		addr := net.UDPAddrFromAddrPort(paddr)
 
 		p.parent.logger.Debug("dialing direct", "addr", addr, "server", p.serverConf.name, "cert", p.serverConf.key)
-		conn, err := p.parent.local.direct.transport.Dial(ctx, addr, &tls.Config{
+		conn, err := p.parent.local.direct.transport.Dial(quicc.RTTContext(ctx), addr, &tls.Config{
 			Certificates: []tls.Certificate{p.parent.local.clientCert},
 			RootCAs:      p.serverConf.cas,
 			ServerName:   p.serverConf.name,
 			NextProtos:   []string{"connet-direct"},
 		}, &quic.Config{
-			KeepAlivePeriod: 25 * time.Second,
+			MaxIdleTimeout:  20 * time.Second,
+			KeepAlivePeriod: 10 * time.Second,
+			Tracer:          quicc.RTTTracer,
 		})
 		if err != nil {
 			errs = append(errs, err)
@@ -352,6 +358,9 @@ func (p *directPeerOutgoing) keepalive(ctx context.Context, conn quic.Connection
 		if err := p.heartbeat(ctx, stream); err != nil {
 			return err
 		}
+		if rttStats := quicc.RTTStats(conn); rttStats != nil {
+			p.parent.logger.Debug("rtt-out", "last", rttStats.LatestRTT(), "smoothed", rttStats.SmoothedRTT())
+		}
 	}
 }
 

client/peer_relay.go

@@ -94,7 +94,8 @@ func (r *relayPeer) connect(ctx context.Context) (quic.Connection, error) {
 		ServerName:   cfg.name,
 		NextProtos:   []string{"connet-relay"},
 	}, &quic.Config{
-		KeepAlivePeriod: 25 * time.Second,
+		MaxIdleTimeout:  20 * time.Second,
+		KeepAlivePeriod: 10 * time.Second,
 	})
 }
 

control/server.go

@@ -122,7 +122,8 @@ func (s *Server) runListener(ctx context.Context) error {
 	defer transport.Close()
 
 	l, err := transport.Listen(s.tlsConf, &quic.Config{
-		KeepAlivePeriod: 25 * time.Second,
+		MaxIdleTimeout:  20 * time.Second,
+		KeepAlivePeriod: 10 * time.Second,
 	})
 	if err != nil {
 		return kleverr.Ret(err)

quicc/rtt.go

@@ -0,0 +1,43 @@
+package quicc
+
+import (
+	"context"
+	"sync/atomic"
+
+	"github.com/quic-go/quic-go"
+	"github.com/quic-go/quic-go/logging"
+)
+
+type rttContextKeyType struct{}
+
+var rttContextKey rttContextKeyType
+
+type rttStats struct {
+	stats atomic.Pointer[logging.RTTStats]
+}
+
+func RTTContext(ctx context.Context) context.Context {
+	return context.WithValue(ctx, rttContextKey, &rttStats{})
+}
+
+func RTTTracer(ctx context.Context, pers logging.Perspective, ci quic.ConnectionID) *logging.ConnectionTracer {
+	v, ok := ctx.Value(rttContextKey).(*rttStats)
+	if !ok {
+		return nil
+	}
+	return &logging.ConnectionTracer{
+		UpdatedMetrics: func(rttStats *logging.RTTStats, cwnd, bytesInFlight logging.ByteCount, packetsInFlight int) {
+			// make a copy of the stats at this point of time
+			stats := *rttStats
+			v.stats.Store(&stats)
+		},
+	}
+}
+
+func RTTStats(conn quic.Connection) *logging.RTTStats {
+	v, ok := conn.Context().Value(rttContextKey).(*rttStats)
+	if !ok {
+		return nil
+	}
+	return v.stats.Load()
+}

relay/clients.go

@@ -167,7 +167,8 @@ func (s *clientsServer) removeSource(fcs *forwardClients, conn *clientConn) {
 
 func (s *clientsServer) run(ctx context.Context, transport *quic.Transport) error {
 	l, err := transport.Listen(s.tlsConf, &quic.Config{
-		KeepAlivePeriod: 25 * time.Second,
+		MaxIdleTimeout:  20 * time.Second,
+		KeepAlivePeriod: 10 * time.Second,
 	})
 	if err != nil {
 		return kleverr.Ret(err)

relay/control.go

@@ -196,7 +196,8 @@ func (s *controlClient) connect(ctx context.Context, transport *quic.Transport)
 	}
 
 	conn, err := transport.Dial(ctx, s.controlAddr, s.controlTLSConf, &quic.Config{
-		KeepAlivePeriod: 25 * time.Second,
+		MaxIdleTimeout:  20 * time.Second,
+		KeepAlivePeriod: 10 * time.Second,
 	})
 	if err != nil {
 		return retConnect(err)