From 41c7e3c445c63c0d712b92a0cfdf5647f58aaa4f Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Wed, 22 Jul 2026 13:09:12 -0300 Subject: [PATCH] add loss.go (#168) --- .gitignore | 2 + tcp/conn.go | 15 +++ tcp/control.go | 10 ++ tcp/handler.go | 70 +++++++++++-- tcp/loss.go | 83 +++++++++++++++ tcp/loss_test.go | 262 +++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 436 insertions(+), 6 deletions(-) create mode 100644 tcp/loss.go create mode 100644 tcp/loss_test.go diff --git a/.gitignore b/.gitignore index 7da9257..cec7615 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,8 @@ # Ignore files with no extensions trick: # First ignore all. * +# Unignore directories since they usually have no extensions +!*/ # Unignore all with extensions. !*.* !LICENSE diff --git a/tcp/conn.go b/tcp/conn.go index 5196982..547563d 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -76,6 +76,16 @@ type ConnConfig struct { // Logger sets the [Conn] logger. // Lower level logging available at [Handler.SetLoggers] via [Conn.InternalHandler]. Logger *slog.Logger + // LossRecovery is the optional packet-loss recovery algorithm (RTO, + // congestion control, ...) for the connection. If set, Nanotime must also be + // set (else Configure returns an error). Leaving it nil disables loss + // recovery. See [LossRecovery]. + LossRecovery LossRecovery + // Nanotime is the monotonic time source in nanoseconds (the func() int64 + // convention used across lneto) that drives LossRecovery. It is required when + // LossRecovery is set and unused otherwise. The tcp package reads it only to + // stamp the loss-recovery hooks; it holds no clock itself. + Nanotime func() int64 } // Configure should be called on any newly created connection before usage. See [ConnConfig]. @@ -83,6 +93,10 @@ func (conn *Conn) Configure(config ConnConfig) (err error) { if config.RWBackoff == nil { return lneto.ErrMissingHALConfig } + if config.LossRecovery != nil && config.Nanotime == nil { + // The tcp package holds no clock: a loss-recovery algorithm cannot run without it. + return lneto.ErrInvalidConfig + } conn.mu.Lock() defer conn.mu.Unlock() err = conn.h.SetBuffers(config.TxBuf, config.RxBuf, config.TxPacketQueueSize) @@ -91,6 +105,7 @@ func (conn *Conn) Configure(config ConnConfig) (err error) { } conn._backoff = config.RWBackoff conn.logger.log = config.Logger + conn.h.SetLossRecovery(config.LossRecovery, config.Nanotime) return nil } diff --git a/tcp/control.go b/tcp/control.go index 0effb55..16960a7 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -257,6 +257,16 @@ func (tcb *ControlBlock) HasPendingRetransmit() bool { return tcb._state.TxDataOpen() && tcb.dupack >= retransmitAfterDupacks && tcb.nRetransmit <= tcb.dupack-retransmitAfterDupacks } +// RetransmitAll rewinds snd.NXT back to snd.UNA so the next PendingSegment and +// Send calls retransmit all unacknowledged data from the oldest sequence number +// (go-back-N). It must be paired with ringTx.RetransmitFromUNA to rewind the +// transmit buffer. Implements RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT). +func (tcb *ControlBlock) RetransmitAll() { + tcb.snd.NXT = tcb.snd.UNA + tcb.dupack = 0 + tcb.nRetransmit = 0 +} + // PendingSegment calculates a suitable next segment to send from a payload length. // It does not modify the ControlBlock state or pending segment queue. func (tcb *ControlBlock) PendingSegment(payloadLen int) (_ Segment, ok bool) { diff --git a/tcp/handler.go b/tcp/handler.go index 369a1ef..9ff7a29 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -31,7 +31,15 @@ type Handler struct { optcodec OptionCodec // reasm tracks out-of-order segments staged in bufRx's free region. Always // enabled once buffers are set (see [Handler.SetBuffers]). - reasm reassembly + reasm reassembly + // loss is the optional packet-loss recovery algorithm (RTO, congestion + // control, ...) driven from the rx/tx hooks. nil disables loss recovery, in + // which case the connection behaves as if no timing existed. nanotime is the + // monotonic time source (nanoseconds) passed to those hooks; it is non-nil + // whenever loss is non-nil (enforced by [Conn.Configure]). See [LossRecovery]. + loss LossRecovery + nanotime func() int64 + closing bool shutdownRx bool // nRetransmit stores the number of times the oldest packet was retransmit. @@ -71,6 +79,28 @@ func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error { return h.bufTx.ResetOrReuse(txbuf, packets, 0) } +// SetLossRecovery installs the packet-loss recovery algorithm and the monotonic +// time source (nanoseconds, the func() int64 convention used across lneto) that +// drives it. The tcp package keeps no clock of its own; nanotime is read only to +// stamp the rx/tx hooks (see [LossRecovery]). Passing loss == nil disables loss +// recovery. It should be set before the connection is opened. +func (h *Handler) SetLossRecovery(loss LossRecovery, nanotime func() int64) { + h.loss = loss + h.nanotime = nanotime +} + +func (h *Handler) lossEnabled() bool { return h.loss != nil } + +// NextDeadline returns the monotonic-nanosecond instant at which the connection +// must next be serviced by a transmit attempt (e.g. an RTO expiry), or 0 when +// there is no deadline or no loss recovery is configured. See [LossRecovery]. +func (h *Handler) NextDeadline() int64 { + if h.loss == nil { + return 0 + } + return h.loss.NextDeadline() +} + // LocalPort returns the local port of the connection. Returns 0 if the connection is closed and uninitialized. func (h *Handler) LocalPort() uint16 { return h.localPort @@ -129,15 +159,22 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { *h = Handler{ connid: h.connid + 1, scb: h.scb, - bufTx: h.bufTx, - bufRx: h.bufRx, localPort: localPort, remotePort: remotePort, - validator: h.validator, - logger: h.logger, - reasm: h.reasm, closing: false, shutdownRx: false, + // Persist configuration across reopen: + validator: h.validator, + loss: h.loss, + nanotime: h.nanotime, + logger: h.logger, + // persist memory across repoen: + bufTx: h.bufTx, + bufRx: h.bufRx, + reasm: h.reasm, + } + if h.lossEnabled() { + h.loss.Reset() } h.reasm.clear() // preserve metadata capacity across reopen, drop held segments. h.bufTx.ResetOrReuse(nil, 0, iss) @@ -175,6 +212,12 @@ func (h *Handler) Recv(incomingPacket []byte) error { return nil } + // Notify loss recovery of the received segment (RTT sampling, timer + // management) and let it drop the segment before processing if it asks to. + if h.lossEnabled() && !h.loss.PreRx(segIncoming, h.nanotime()).Keep { + return nil + } + // Out-of-order reassembly: buffer in-window data that arrived ahead of the // next expected sequence number before the ControlBlock (sequential-only) // would reject it. Buffered segments live in bufRx's free region. @@ -337,6 +380,18 @@ func (h *Handler) Send(b []byte) (int, error) { if h.IsTxOver() { return 0, net.ErrClosed } + var now int64 + if h.lossEnabled() { + now = h.nanotime() + if h.loss.PreTx(now).RetransmitAll { + // Go-back-N retransmission directed by loss recovery: rewind the + // send sequence and transmit buffer so unacknowledged data is resent + // from snd.UNA. Done before the early short-circuit below so an + // expired RTO retransmits even with no new data queued. + h.scb.RetransmitAll() + h.bufTx.RetransmitFromUNA() + } + } awaitingSyn := h.AwaitingSynSend() requeueControl := h.requeueControl buffered := h.bufTx.BufferedUnsent() @@ -419,6 +474,9 @@ func (h *Handler) Send(b []byte) (int, error) { } else if prevState != h.scb.State() && h.logenabled(slog.LevelInfo) { h.info("tcp.Handler:tx-statechange", slog.Uint64("port", uint64(h.localPort)), slog.String("oldState", prevState.String()), slog.String("newState", h.scb.State().String()), slog.String("txflags", segment.Flags.String())) } + if h.lossEnabled() { + h.loss.PostTx(segment, now) + } h.requeueControl = false tfrm.SetSourcePort(h.localPort) tfrm.SetDestinationPort(h.remotePort) diff --git a/tcp/loss.go b/tcp/loss.go new file mode 100644 index 0000000..98cfe5c --- /dev/null +++ b/tcp/loss.go @@ -0,0 +1,83 @@ +package tcp + +// LossRecovery abstracts TCP packet-loss recovery: RTO, congestion control and +// any similar algorithm that observes segment traffic and steers the +// connection's transmit behaviour. As far as the tcp package is concerned these +// are all the same thing — packet-loss recovery algorithms — so they share one +// interface (see discussion #157). +// +// The tcp package stays free of any time source: the current monotonic time in +// nanoseconds (the func() int64 convention used across lneto) is passed in at +// each hook boundary. It originates from [ConnConfig.Nanotime] and satisfies the +// "WHEN was this segment rx/tx'd" requirement without a clock living inside the +// state machine, which also keeps implementations deterministic for testing +// (see issue #140). +// +// The interface is intentionally free of errors: an implementation handles or +// reports its own errors rather than propagating them into lneto internals. +// +// Introspection (smoothed RTT, current window, ...) is deliberately left off the +// interface; expose it on the concrete implementation the caller constructs and +// hands to [ConnConfig]. +type LossRecovery interface { + // Reset returns the implementation to its initial, pre-connection state. It + // is invoked whenever the connection is (re)opened or aborted so a single + // LossRecovery value can be reused across the lifetime of connection reuse + // (see discussion #115). + Reset() + + // NextDeadline returns the monotonic-nanosecond instant at which the + // connection must next be serviced by a transmit attempt — typically the RTO + // expiry. A return of 0 means there is no pending deadline. It replaces a + // poll/atomic-flag scheme with a deadline the caller's event loop can + // schedule against. + NextDeadline() int64 + + // PreRx is called for every segment received on the TCP port before the + // state machine processes it, with the monotonic time the segment arrived. It + // returns whether the segment should be kept (processed) or dropped. + PreRx(incoming Segment, now int64) RxDirective + + // PreTx is called on entering the transmit path (Encapsulate), before a + // segment is built, with the current monotonic time. Its directive tells the + // connection whether to retransmit unacknowledged data, rewind the send + // pointer, or hold back new data. + PreTx(now int64) TxDirective + + // PostTx is called on leaving the transmit path with the segment that was + // actually emitted and the monotonic time it was sent. This is where segment + // timing (for RTT sampling and the retransmission timer) is recorded. + PostTx(outgoing Segment, now int64) +} + +// TxDirective is returned by [LossRecovery.PreTx] to steer the transmit path. +// The zero value directs the connection to proceed normally (send new data if +// available, no retransmission). +type TxDirective struct { + // RewindNXT is the number of sequence-space octets to rewind snd.NXT by + // before transmitting, for partial (e.g. selective) retransmission. Zero + // means no rewind. It is independent of Retransmit, which rewinds fully to + // snd.UNA. + // RewindNXT uint32 + + // RetransmitAll requests go-back-N retransmission: the connection rewinds + // snd.NXT to snd.UNA and resends unacknowledged data from the oldest + // sequence number. + RetransmitAll bool + // HoldNew pauses transmission of new data (for example when the congestion + // window is exhausted). Retransmissions already directed by this same + // directive still proceed. + // HoldNew bool +} + +// RxDirective is returned by [LossRecovery.PreRx]. +// +// NOTE: its shape is the minimum viable contract — it mirrors the original +// PreRx "keep" boolean from discussion #157 — and is the one element of the +// interface not yet fully settled there. It is a struct (rather than a bare +// bool) so fields can be added without breaking implementations. +type RxDirective struct { + // Keep reports whether the received segment should be handed to the state + // machine. A false value drops the segment before it is processed. + Keep bool +} diff --git a/tcp/loss_test.go b/tcp/loss_test.go new file mode 100644 index 0000000..18dd13a --- /dev/null +++ b/tcp/loss_test.go @@ -0,0 +1,262 @@ +package tcp + +import ( + "math/rand" + "testing" + + "github.com/soypat/lneto/ethernet" +) + +// recordingLoss is a test LossRecovery that records every hook invocation and +// lets the test steer the directives returned to the Handler. It is the +// interface counterpart driven by the Handler under test. +type recordingLoss struct { + resets int + preRx []hookCall + preTx []int64 + postTx []hookCall + deadline int64 // value NextDeadline reports back. + + // Directives handed back to the Handler. + keep bool // PreRx result. Default true (see newRecordingLoss). + tx TxDirective // PreTx result. +} + +type hookCall struct { + seg Segment + now int64 +} + +func newRecordingLoss() *recordingLoss { return &recordingLoss{keep: true} } + +var _ LossRecovery = (*recordingLoss)(nil) + +func (l *recordingLoss) Reset() { l.resets++ } +func (l *recordingLoss) NextDeadline() int64 { return l.deadline } + +func (l *recordingLoss) PreRx(incoming Segment, now int64) RxDirective { + l.preRx = append(l.preRx, hookCall{seg: incoming, now: now}) + return RxDirective{Keep: l.keep} +} + +func (l *recordingLoss) PreTx(now int64) TxDirective { + l.preTx = append(l.preTx, now) + return l.tx +} + +func (l *recordingLoss) PostTx(outgoing Segment, now int64) { + l.postTx = append(l.postTx, hookCall{seg: outgoing, now: now}) +} + +// TestLossRecovery_DisabledByDefault verifies the Handler runs normally with no +// loss recovery installed: NextDeadline reports no deadline and the transmit/ +// receive paths never touch a nil LossRecovery. +func TestLossRecovery_DisabledByDefault(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(1)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + setupClientServer(t, rng, client, server) + + if d := client.NextDeadline(); d != 0 { + t.Fatalf("NextDeadline with no loss recovery = %d, want 0", d) + } + var buf [mtu]byte + establish(t, client, server, buf[:]) // must not panic on nil loss recovery. +} + +// TestLossRecovery_HooksInvoked verifies the Handler drives the full hook +// contract across a handshake: Reset on open, PreTx+PostTx on every transmit, +// PreRx on every receive, each stamped with the configured monotonic clock. +func TestLossRecovery_HooksInvoked(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(2)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + loss := newRecordingLoss() + const clockNow = 1_000_000 + client.SetLossRecovery(loss, func() int64 { return clockNow }) + + setupClientServer(t, rng, client, server) // OpenActive → reset → Reset(). + if loss.resets == 0 { + t.Fatal("Reset not called on open") + } + + var buf [mtu]byte + establish(t, client, server, buf[:]) + + // Client emitted SYN and the final ACK: both paths must have hit PreTx/PostTx. + if len(loss.preTx) == 0 { + t.Fatal("PreTx never called on transmit") + } + if len(loss.postTx) == 0 { + t.Fatal("PostTx never called on transmit") + } + if len(loss.preTx) != len(loss.postTx) { + t.Fatalf("PreTx calls=%d, PostTx calls=%d, want equal", len(loss.preTx), len(loss.postTx)) + } + // Client received the SYN-ACK: PreRx must have seen it. + if len(loss.preRx) == 0 { + t.Fatal("PreRx never called on receive") + } + + // The Handler holds no clock: every hook must be stamped from the supplied + // nanotime source. + for i, c := range loss.postTx { + if c.now != clockNow { + t.Fatalf("PostTx[%d].now = %d, want clock %d", i, c.now, clockNow) + } + } + for i, now := range loss.preTx { + if now != clockNow { + t.Fatalf("PreTx[%d].now = %d, want clock %d", i, now, clockNow) + } + } + for i, c := range loss.preRx { + if c.now != clockNow { + t.Fatalf("PreRx[%d].now = %d, want clock %d", i, c.now, clockNow) + } + } + + // PostTx receives the segment actually emitted: the first is the SYN. + if !loss.postTx[0].seg.Flags.HasAny(FlagSYN) { + t.Fatalf("first PostTx segment flags=%s, want SYN", loss.postTx[0].seg.Flags) + } +} + +// TestLossRecovery_NextDeadlineDelegates verifies NextDeadline is forwarded to +// the installed LossRecovery unchanged. +func TestLossRecovery_NextDeadlineDelegates(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(3)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + loss := newRecordingLoss() + loss.deadline = 4242 + client.SetLossRecovery(loss, func() int64 { return 1 }) + setupClientServer(t, rng, client, server) + + if d := client.NextDeadline(); d != 4242 { + t.Fatalf("NextDeadline = %d, want delegated 4242", d) + } +} + +// TestLossRecovery_PreRxDropsSegment verifies a PreRx directive of Keep=false +// drops the segment before the state machine sees it: the payload is not +// buffered and connection state is untouched. +func TestLossRecovery_PreRxDropsSegment(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(4)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + loss := newRecordingLoss() + server.SetLossRecovery(loss, func() int64 { return 1 }) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) // keep=true so handshake completes. + + // Now start dropping everything the server receives. + loss.keep = false + preRxBefore := len(loss.preRx) + + data := []byte("dropme") + if _, err := client.Write(data); err != nil { + t.Fatal("client write:", err) + } + clear(buf[:]) + n, err := client.Send(buf[:]) + if err != nil { + t.Fatal("client send:", err) + } + + if err := server.Recv(buf[:n]); err != nil { + t.Fatalf("dropped segment must return nil, got %v", err) + } + if len(loss.preRx) != preRxBefore+1 { + t.Fatalf("PreRx calls=%d, want %d (segment must reach PreRx)", len(loss.preRx), preRxBefore+1) + } + if server.BufferedInput() != 0 { + t.Fatalf("dropped segment must not be buffered, got %d bytes", server.BufferedInput()) + } + if server.State() != StateEstablished { + t.Fatalf("dropped segment must not change state, got %s", server.State()) + } +} + +// TestLossRecovery_PreTxRetransmitAll verifies a PreTx directive of +// RetransmitAll drives go-back-N: the Handler rewinds and re-emits already-sent, +// unacknowledged data from snd.UNA on the next transmit. +func TestLossRecovery_PreTxRetransmitAll(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(5)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + loss := newRecordingLoss() + client.SetLossRecovery(loss, func() int64 { return 1 }) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + // Emit one data segment; server never ACKs, so it stays unacknowledged. + data := []byte("payload") + if _, err := client.Write(data); err != nil { + t.Fatal("client write:", err) + } + clear(buf[:]) + n, err := client.Send(buf[:]) + if err != nil { + t.Fatal("client send data:", err) + } + if n <= sizeHeaderTCP { + t.Fatal("expected data segment") + } + firstSeg := mustSegment(t, buf[:n], n-sizeHeaderTCP) + + // Direct go-back-N on the next transmit. + loss.tx = TxDirective{RetransmitAll: true} + clear(buf[:]) + n, err = client.Send(buf[:]) + if err != nil { + t.Fatal("client send retransmit:", err) + } + if n <= sizeHeaderTCP { + t.Fatal("expected retransmitted data segment") + } + rtSeg := mustSegment(t, buf[:n], n-sizeHeaderTCP) + + if rtSeg.SEQ != firstSeg.SEQ { + t.Fatalf("retransmit SEQ=%d, want original UNA SEQ=%d (go-back-N)", rtSeg.SEQ, firstSeg.SEQ) + } + if rtSeg.DATALEN != firstSeg.DATALEN { + t.Fatalf("retransmit DATALEN=%d, want %d", rtSeg.DATALEN, firstSeg.DATALEN) + } +} + +// TestLossRecovery_ResetOnReopen verifies Reset fires on every (re)open and on +// Abort, so a single LossRecovery value can be reused across connection reuse. +func TestLossRecovery_ResetOnReopen(t *testing.T) { + const mtu = ethernet.MaxMTU + client := newHandler(t, mtu, 3) + loss := newRecordingLoss() + client.SetLossRecovery(loss, func() int64 { return 1 }) + + if err := client.OpenActive(1234, 5678, 0); err != nil { + t.Fatal("open 1:", err) + } + afterOpen := loss.resets + if afterOpen == 0 { + t.Fatal("Reset not called on first open") + } + + client.Abort() + if loss.resets <= afterOpen { + t.Fatalf("Reset not called on Abort: resets=%d, want >%d", loss.resets, afterOpen) + } + afterAbort := loss.resets + + if err := client.OpenActive(1234, 5678, 0); err != nil { + t.Fatal("open 2:", err) + } + if loss.resets <= afterAbort { + t.Fatalf("Reset not called on reopen: resets=%d, want >%d", loss.resets, afterAbort) + } +}