diff --git a/x/xnet/retransmit_test.go b/x/xnet/retransmit_test.go index 0169443..d9270da 100644 --- a/x/xnet/retransmit_test.go +++ b/x/xnet/retransmit_test.go @@ -12,11 +12,14 @@ import ( "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/tcp" + "github.com/soypat/lneto/tcp/rto" ) // TestTCPRetransmitsLostSegment drops exactly one data segment and requires the -// bytes to arrive anyway, which is what [TCPPoolConfig.NanoTime] already promises -// in its own documentation. Without a LossRecovery installed the loss is terminal. +// bytes to arrive anyway. It covers [TCPPoolConfig.NewPolicy] reaching the +// pooled and dialed connections alike: without a [tcp.Policy] installed the +// loss is terminal, which is what this test asserts against. func TestTCPRetransmitsLostSegment(t *testing.T) { const ( MTU = ethernet.MaxMTU @@ -55,6 +58,13 @@ func TestTCPRetransmitsLostSegment(t *testing.T) { EstablishedTimeout: 30 * time.Second, ClosingTimeout: 30 * time.Second, NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, + NewPolicy: func() tcp.Policy { + timer := new(rto.Timer) + if err := timer.Configure(func() int64 { return time.Now().UnixNano() }); err != nil { + t.Fatal(err) + } + return timer + }, } svGo := sv.StackBlocking(backoffYield).StackGo(StackGoConfig{ListenerPoolConfig: pool}) clGo := client.StackBlocking(backoffYield).StackGo(StackGoConfig{ diff --git a/x/xnet/stack-go.go b/x/xnet/stack-go.go index f576d7c..9e82dcf 100644 --- a/x/xnet/stack-go.go +++ b/x/xnet/stack-go.go @@ -176,17 +176,19 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype if isDial { var conn tcp.Conn // DIAL TCP: active connection a.k.a TCP Client branch. - err = conn.Configure(tcp.ConnConfig{ + conncfg := tcp.ConnConfig{ // TODO(pato): Eventually add UDP configuration. we use TCP for now for simplicity's sake. TxBuf: make([]byte, s.plcfg.TxBufSize), RxBuf: make([]byte, s.plcfg.RxBufSize), TxPacketQueueSize: s.plcfg.QueueSize, RWBackoff: s.plcfg.NewBackoff(), - // A dialed connection needs a retransmission timer as much as a - // pooled one. See [NewTCPPool]. - LossRecovery: new(tcp.RTO), - Nanotime: s.blk.nanotime, - }) + } + if s.plcfg.NewPolicy != nil { + // A dialed connection needs loss recovery as much as a pooled + // one. See [TCPPoolConfig.NewPolicy]. + conncfg.Policy = s.plcfg.NewPolicy() + } + err = conn.Configure(conncfg) if err != nil { return nil, err } diff --git a/x/xnet/tcppool.go b/x/xnet/tcppool.go index 5fed680..be92bf5 100644 --- a/x/xnet/tcppool.go +++ b/x/xnet/tcppool.go @@ -42,8 +42,9 @@ type TCPPoolConfig struct { ConnLogger *slog.Logger // NanoTime returns the current monotonic time in nanoseconds. - // Used for pool timeout tracking and passed to each [tcp.Conn] for - // retransmission timing (RFC 6298). If nil, defaults to time.Now().UnixNano(). + // Used for pool timeout tracking. If nil, defaults to time.Now().UnixNano(). + // Retransmission timing is not driven by this clock: a [tcp.Policy] carries + // its own. See NewPolicy. NanoTime func() int64 // EstablishedTimeout sets the timeout for a TCP connection since it is acquired until it is established. // If the connection does not establish in this time it will be closed by the pool. @@ -56,6 +57,9 @@ type TCPPoolConfig struct { // NewBackoff returns the backoff to use for every newly configured TCP connection. Must be non-nil. // This should always return a static(non-method) function unless you know what you are doing. NewBackoff func() lneto.BackoffStrategy + // NewPolicy if non-nil creates a [tcp.Policy] for each [tcp.Conn] used by the configured Listener. + // NewPolicy should not return reused policies unless the algorithm is stateless. See [tcp.Policy] for more information. + NewPolicy func() tcp.Policy } func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { @@ -87,11 +91,11 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { TxPacketQueueSize: cfg.QueueSize, Logger: cfg.ConnLogger, RWBackoff: cfg.NewBackoff(), - // Retransmission timing, as NanoTime's contract promises. One RTO per - // connection: it shadows that connection's send sequence space and so - // cannot be shared. - LossRecovery: new(tcp.RTO), - Nanotime: pool.now, + } + if cfg.NewPolicy != nil { + // One Policy per connection: it shadows that connection's send + // sequence space and so cannot be shared. + conncfg.Policy = cfg.NewPolicy() } err := pool.conns[i].Configure(conncfg) if err != nil {