From 52a3926428b654185676af1d8beccae656cfb3a8 Mon Sep 17 00:00:00 2001 From: Patricio Whittingslow Date: Mon, 24 Aug 2026 16:47:06 -0300 Subject: [PATCH] implement tcp.Policy and refactor rto to use it --- tcp/conn.go | 20 +- tcp/control.go | 20 +- tcp/handler.go | 90 ++++--- tcp/loss_test.go | 262 -------------------- tcp/policy.go | 11 +- tcp/policy_test.go | 434 +++++++++++++++++++++++++++++++++ tcp/rto/timer.go | 92 ++++--- tcp/rto/timer_test.go | 171 +++++++------ tcp/txqueue.go | 73 +++++- tcp/txqueue_retransmit_test.go | 215 ++++++++++++++++ 10 files changed, 961 insertions(+), 427 deletions(-) delete mode 100644 tcp/loss_test.go create mode 100644 tcp/policy_test.go create mode 100644 tcp/txqueue_retransmit_test.go diff --git a/tcp/conn.go b/tcp/conn.go index f1e63e0..fdb4c7c 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -76,16 +76,10 @@ 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 Policy - // 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 + // Policy is the optional transmit-steering algorithm (RTO, congestion + // control, ...) for the connection. nil disables it. A Policy needing time + // carries its own clock. See [Policy]. + Policy Policy } // Configure should be called on any newly created connection before usage. See [ConnConfig]. @@ -93,10 +87,6 @@ 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) @@ -105,7 +95,7 @@ func (conn *Conn) Configure(config ConnConfig) (err error) { } conn._backoff = config.RWBackoff conn.logger.log = config.Logger - conn.h.SetPolicy(config.LossRecovery) + conn.h.SetPolicy(config.Policy) return nil } diff --git a/tcp/control.go b/tcp/control.go index 859e971..d6b142a 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -95,6 +95,12 @@ func (tcb *ControlBlock) RecvWindow() Size { return tcb.rcv.WND } // ISS returns the initial sequence number of the connection that was defined on a call to Open by user. func (tcb *ControlBlock) ISS() Value { return tcb.snd.ISS } +// SendUNA returns snd.UNA, the oldest sequence number not yet acked by the remote. +func (tcb *ControlBlock) SendUNA() Value { return tcb.snd.UNA } + +// SendNext returns snd.NXT, one past the highest sequence number sent. +func (tcb *ControlBlock) SendNext() Value { return tcb.snd.NXT } + // MaxInFlightData returns the maximum size of a segment that can be sent by taking into account // the send window size and the unacked data. Returns 0 before StateSynRcvd. func (tcb *ControlBlock) MaxInFlightData() Size { @@ -257,8 +263,20 @@ func (tcb *ControlBlock) HasPendingRetransmit() bool { return tcb._state.TxDataOpen() && tcb.dupack >= retransmitAfterDupacks && tcb.nRetransmit <= tcb.dupack-retransmitAfterDupacks } +// RetransmitFrom rewinds snd.NXT back to newNxt so the next PendingSegment and +// Send calls retransmit unacknowledged data from that sequence number onwards. +// It must be paired with ringTx.RetransmitFrom to rewind the transmit buffer to +// the same point. Implements RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT). +// +// It reports false and changes nothing when newNxt falls outside the +// unacknowledged range [snd.UNA, snd.NXT] or the connection cannot send data, so +// a misbehaving [Policy] cannot corrupt the send sequence space. func (tcb *ControlBlock) RetransmitFrom(newNxt Value) bool { - panic("not yet implemented") + if !tcb._state.TxDataOpen() { + return false + } else if newNxt.LessThan(tcb.snd.UNA) || tcb.snd.NXT.LessThan(newNxt) { + return false + } tcb.snd.NXT = newNxt tcb.dupack = 0 tcb.nRetransmit = 0 diff --git a/tcp/handler.go b/tcp/handler.go index 6621035..7cffc89 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -31,9 +31,8 @@ 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 - policy Policy - nanotime func() int64 + reasm reassembly + policy Policy closing bool shutdownRx bool @@ -74,11 +73,17 @@ func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error { return h.bufTx.ResetOrReuse(txbuf, packets, 0) } +// SetPolicy installs the transmit-steering algorithm. nil disables it. +// It should be set before the connection is opened. See [Policy]. func (h *Handler) SetPolicy(policy Policy) { h.policy = policy } func (h *Handler) policyEnabled() bool { return h.policy != nil } +// ControlBlock returns the state machine underlying the Handler, mainly so a +// [Policy] can read the sequence spaces. Not for modification. +func (h *Handler) ControlBlock() *ControlBlock { return &h.scb } + // 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 @@ -144,7 +149,6 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { // Persist configuration across reopen: validator: h.validator, policy: h.policy, - nanotime: h.nanotime, logger: h.logger, // persist memory across repoen: bufTx: h.bufTx, @@ -221,6 +225,9 @@ func (h *Handler) Recv(incomingPacket []byte) error { if prevState != h.scb.State() { h.info("tcp.Handler:rx-statechange", slog.Uint64("port", uint64(h.localPort)), slog.String("old", prevState.String()), slog.String("new", h.scb.State().String()), slog.String("rxflags", segIncoming.Flags.String())) } + if h.policyEnabled() { + h.policy.PostRx(h, prevState, tfrm) + } if segIncoming.DATALEN != 0 && h.shutdownRx && (h.scb.State() == StateFinWait1 || h.scb.State() == StateFinWait2) { // soypat/lneto#50: the application is done in both directions — read side // shut down (CloseRead) and our FIN sent (Close) — so inbound data has no @@ -356,18 +363,28 @@ func (h *Handler) Send(b []byte) (int, error) { if h.IsTxOver() { return 0, net.ErrClosed } + tfrm, err := NewFrame(b) + if err != nil { + return 0, err + } + offset := uint8(5) + var holdNew bool if h.policyEnabled() { - tfrm, err := NewFrame(b) - if err != nil { - return 0, err + // Hand the Policy a defined frame: zeroed header at the minimum offset. + // It may append options and raise the offset, which is read back below. + tfrm.ClearHeader() + tfrm.SetOffsetAndFlags(offset, 0) + rtxFrom, doRtx, hold := h.policy.PreTx(h, tfrm) + holdNew = hold + if doRtx && h.scb.RetransmitFrom(rtxFrom) { + // Retransmission directed by the Policy: rewind the transmit buffer + // to match the send sequence so unacknowledged data is resent. Done + // before the early short-circuit below so an expired RTO + // retransmits even with no new data queued. + h.bufTx.RetransmitFrom(rtxFrom) } - rtxFrom, doRtx, _ := h.policy.PreTx(h, tfrm) - if doRtx { - // 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.RetransmitFrom(rtxFrom) + if o, _ := tfrm.OffsetAndFlags(); o > offset && int(o)*4 < len(b) { + offset = o } } awaitingSyn := h.AwaitingSynSend() @@ -383,29 +400,27 @@ func (h *Handler) Send(b []byte) (int, error) { // Early nop short circuit. return 0, nil } - tfrm, err := NewFrame(b) - if err != nil { - return 0, err - } if buffered == 0 && h.closing && (h.scb.State() != StateCloseWait || !h.scb.HasPending()) { // If Close called and no more data to be sent, terminate connection. // In CLOSE-WAIT: wait until the pending ACK is sent first, since scb.Close() // overwrites pending with [FIN|ACK] (unlike ESTABLISHED which merges via bitmask). h.closing = false - err = h.scb.Close() + err := h.scb.Close() if err != nil { h.logerr("tcp.Handler.Close", slog.String("err", errstr(err)), slog.String("state", h.State().String())) h.Abort() return 0, io.EOF } } - offset := uint8(5) - mss := uint16(len(b) - sizeHeaderTCP) + // optHead is where the Handler's own options begin: after the fixed header + // and after any options the Policy already wrote, so neither clobbers the other. + optHead := int(offset) * 4 + mss := uint16(len(b) - optHead) var segment Segment if awaitingSyn || requeueControl && h.scb.State() == StateSynSent { // Handling init syn segment. segment = ClientSynSegment(h.bufTx.iss, Size(h.bufRx.Size())) - h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss) + h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss) offset++ if requeueControl { h.info("tcp.Handler:requeue-syn", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort))) @@ -417,7 +432,7 @@ func (h *Handler) Send(b []byte) (int, error) { WND: Size(h.bufRx.Free()), Flags: synack, } - h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss) + h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss) offset++ h.info("tcp.Handler:requeue-synack", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort))) } else if requeueControl { @@ -425,17 +440,22 @@ func (h *Handler) Send(b []byte) (int, error) { return 0, nil } else { var ok bool - maxPayload := len(b) - sizeHeaderTCP + maxPayload := len(b) - optHead + if holdNew && !h.nextSegmentIsRetransmit() { + // Policy is holding new data back (congestion window exhausted). + // A retransmission it directed in this same call still proceeds. + maxPayload = 0 + } segment, ok = h.scb.PendingSegment(maxPayload) segment.WND = h.recvWindow() if !ok { // No pending control segment or data to send. Yield. return 0, nil } else if segment.Flags == synack { - h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss) + h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss) offset++ } else if segment.DATALEN > 0 { - n, err := h.bufTx.MakePacket(b[sizeHeaderTCP:sizeHeaderTCP+segment.DATALEN], segment.SEQ) + n, err := h.bufTx.MakePacket(b[optHead:optHead+int(segment.DATALEN)], segment.SEQ) if err != nil { return 0, err } @@ -452,15 +472,19 @@ 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.policyEnabled() { - h.policy.PostTx(h, tfrm) - } h.requeueControl = false tfrm.SetSourcePort(h.localPort) tfrm.SetDestinationPort(h.remotePort) tfrm.SetSegment(segment, offset) tfrm.SetUrgentPtr(0) datalen := int(offset)*4 + int(segment.DATALEN) + if h.policyEnabled() { + // Frame trimmed to what is actually emitted so the Policy's Payload() + // is the segment data and nothing more. + if sent, err := NewFrame(b[:datalen]); err == nil { + h.policy.PostTx(h, sent) + } + } closedSuccess := prevState == StateTimeWait && segment.Flags.HasAny(FlagACK) if closedSuccess { h.reset(0, 0, 0) @@ -472,6 +496,14 @@ func (h *Handler) Send(b []byte) (int, error) { return datalen, nil } +// nextSegmentIsRetransmit reports whether the next data segment would resend +// already-transmitted bytes rather than open new sequence space. Used to let a +// retransmission through while a [Policy] holds new data back. +func (h *Handler) nextSegmentIsRetransmit() bool { + endSeq, hasSent := h.bufTx.sentEndSeq() + return hasSent && h.scb.snd.NXT.LessThan(endSeq) +} + // Write implements [io.Writer] by copying b to a internal buffer to be sent over the network on the next // [Handler.Send] call that can send data to remote peer. Use [Handler.Free] to know the maximum length the argument slice can be before erroring. func (h *Handler) Write(b []byte) (int, error) { diff --git a/tcp/loss_test.go b/tcp/loss_test.go deleted file mode 100644 index 18dd13a..0000000 --- a/tcp/loss_test.go +++ /dev/null @@ -1,262 +0,0 @@ -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) - } -} diff --git a/tcp/policy.go b/tcp/policy.go index 814815c..200438c 100644 --- a/tcp/policy.go +++ b/tcp/policy.go @@ -1,15 +1,24 @@ package tcp +// Policy observes segment traffic and steers transmit behaviour: RTO, +// congestion control and the like (discussion #157). The tcp package holds no +// clock, so a Policy needing time carries its own (issue #140). +// Introspection stays off the interface; put it on the concrete type. type Policy interface { + // Reset returns the Policy to its pre-connection state. Called on every + // (re)open and Abort. Must preserve configuration such as a clock. Reset() // PreTx is called before writing to a frame. // The outgoing frame options can be set by the Policy and will be respected if Frame offset >5. + // rtxFrom is ignored unless within [snd.UNA, snd.NXT]. Nothing is committed + // until PostTx: a transmit attempt may emit no segment at all. PreTx(h *Handler, outgoingOpts Frame) (rtxFrom Value, retransmit, holdNew bool) // PreRx is called by [Handler] on every incoming segment. // PreRx can choose to drop segment if it returns keep=false. PreRx(h *Handler, incoming Frame) (keep bool) // PostRx is called by [Handler] after accepting an incoming segment. + // TODO: congestion control will also want the pre-Recv snd.UNA here. PostRx(h *Handler, prevState State, accepted Frame) - // PostTx called on leaving the transmit path. + // PostTx called on leaving the transmit path with the fully written frame. PostTx(h *Handler, outgoing Frame) } diff --git a/tcp/policy_test.go b/tcp/policy_test.go new file mode 100644 index 0000000..e86ccae --- /dev/null +++ b/tcp/policy_test.go @@ -0,0 +1,434 @@ +package tcp + +import ( + "math/rand" + "testing" + + "github.com/soypat/lneto/ethernet" +) + +// recordingPolicy records every hook invocation and lets the test steer what is +// returned to the Handler. It is the [Policy] counterpart driven by the Handler +// under test. +type recordingPolicy struct { + resets int + preRx []Segment + preTx int + postRx []Segment + postTx []txRecord + + // Values handed back to the Handler. + keep bool // PreRx result. Default true (see newRecordingPolicy). + rtxFrom Value + retransmit bool + holdNew bool + // writeOpts, when non-empty, is appended as TCP options by PreTx. + writeOpts []byte +} + +// txRecord is what PostTx observed on the emitted frame. +type txRecord struct { + seg Segment + offset uint8 + sport uint16 + dport uint16 +} + +func newRecordingPolicy() *recordingPolicy { return &recordingPolicy{keep: true} } + +var _ Policy = (*recordingPolicy)(nil) + +func (p *recordingPolicy) Reset() { p.resets++ } + +func (p *recordingPolicy) PreRx(h *Handler, incoming Frame) bool { + p.preRx = append(p.preRx, incoming.Segment(len(incoming.Payload()))) + return p.keep +} + +func (p *recordingPolicy) PostRx(h *Handler, prevState State, accepted Frame) { + p.postRx = append(p.postRx, accepted.Segment(len(accepted.Payload()))) +} + +func (p *recordingPolicy) PreTx(h *Handler, outgoingOpts Frame) (Value, bool, bool) { + p.preTx++ + if len(p.writeOpts) > 0 { + // Raise the offset first: Options() is sized from it. + words := uint8(5 + (len(p.writeOpts)+3)/4) + outgoingOpts.SetOffsetAndFlags(words, 0) + copy(outgoingOpts.Options(), p.writeOpts) + } + return p.rtxFrom, p.retransmit, p.holdNew +} + +func (p *recordingPolicy) PostTx(h *Handler, outgoing Frame) { + offset, _ := outgoing.OffsetAndFlags() + p.postTx = append(p.postTx, txRecord{ + seg: outgoing.Segment(len(outgoing.Payload())), + offset: offset, + sport: outgoing.SourcePort(), + dport: outgoing.DestinationPort(), + }) +} + +// TestPolicy_DisabledByDefault verifies the Handler runs normally with no Policy +// installed: the transmit and receive paths never touch a nil Policy. +func TestPolicy_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) + + var buf [mtu]byte + establish(t, client, server, buf[:]) // must not panic on nil Policy. +} + +// TestPolicy_HooksInvoked verifies the Handler drives the full hook contract +// across a handshake: Reset on open, PreTx+PostTx on transmit, PreRx+PostRx on +// receive. +func TestPolicy_HooksInvoked(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(2)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + client.SetPolicy(pol) + + setupClientServer(t, rng, client, server) // OpenActive → reset → Reset(). + if pol.resets == 0 { + t.Fatal("Reset not called on open") + } + + var buf [mtu]byte + establish(t, client, server, buf[:]) + + if pol.preTx == 0 { + t.Fatal("PreTx never called on transmit") + } + if len(pol.postTx) == 0 { + t.Fatal("PostTx never called on transmit") + } + if pol.preTx < len(pol.postTx) { + t.Fatalf("PreTx calls=%d < PostTx calls=%d: PostTx must never fire without PreTx", pol.preTx, len(pol.postTx)) + } + // Client received the SYN-ACK and accepted it. + if len(pol.preRx) == 0 { + t.Fatal("PreRx never called on receive") + } + if len(pol.postRx) == 0 { + t.Fatal("PostRx never called on accepted receive") + } + // PostTx receives the segment actually emitted: the first is the SYN. + if !pol.postTx[0].seg.Flags.HasAny(FlagSYN) { + t.Fatalf("first PostTx segment flags=%s, want SYN", pol.postTx[0].seg.Flags) + } +} + +// TestPolicy_PostTxSeesWrittenFrame verifies PostTx observes the fully populated +// frame — ports, sequence numbers and payload length as emitted — and not the +// frame as it stood before the segment was written into it. +func TestPolicy_PostTxSeesWrittenFrame(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(6)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + client.SetPolicy(pol) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + 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:", err) + } + last := pol.postTx[len(pol.postTx)-1] + wantSeg := mustSegment(t, buf[:n], n-int(last.offset)*4) + if last.seg != wantSeg { + t.Fatalf("PostTx segment=%+v, want emitted %+v", last.seg, wantSeg) + } + if int(last.seg.DATALEN) != len(data) { + t.Fatalf("PostTx DATALEN=%d, want %d", last.seg.DATALEN, len(data)) + } + if last.sport != client.LocalPort() || last.dport != client.RemotePort() { + t.Fatalf("PostTx ports=%d→%d, want %d→%d", last.sport, last.dport, client.LocalPort(), client.RemotePort()) + } +} + +// TestPolicy_NoPostTxWithoutSegment verifies a transmit attempt that emits +// nothing still runs PreTx but never PostTx, so a Policy cannot mistake a +// no-op Send for a segment on the wire. +func TestPolicy_NoPostTxWithoutSegment(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(7)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + client.SetPolicy(pol) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + preTxBefore, postTxBefore := pol.preTx, len(pol.postTx) + n, err := client.Send(buf[:]) // Nothing queued: no segment. + if err != nil { + t.Fatal("client send:", err) + } + if n != 0 { + t.Fatalf("expected no segment, got %d bytes", n) + } + if pol.preTx != preTxBefore+1 { + t.Fatalf("PreTx calls=%d, want %d: PreTx must run on every attempt", pol.preTx, preTxBefore+1) + } + if len(pol.postTx) != postTxBefore { + t.Fatalf("PostTx calls=%d, want %d: no segment was emitted", len(pol.postTx), postTxBefore) + } +} + +// TestPolicy_PreTxOptions verifies options written by PreTx survive to the wire: +// the data offset accounts for them and the payload starts after them. +func TestPolicy_PreTxOptions(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(8)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + client.SetPolicy(pol) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + // One 4-byte option word: NOP,NOP,NOP,EOL. + opts := []byte{1, 1, 1, 0} + pol.writeOpts = opts + + 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:", err) + } + frm, err := NewFrame(buf[:n]) + if err != nil { + t.Fatal("frame:", err) + } + offset, _ := frm.OffsetAndFlags() + if offset != 6 { + t.Fatalf("data offset=%d, want 6 (header + one option word)", offset) + } + if got := frm.Options(); string(got) != string(opts) { + t.Fatalf("options=%v, want %v", got, opts) + } + if got := frm.Payload(); string(got) != string(data) { + t.Fatalf("payload=%q, want %q: options must not overlap data", got, data) + } + if n != int(offset)*4+len(data) { + t.Fatalf("frame length=%d, want %d", n, int(offset)*4+len(data)) + } +} + +// TestPolicy_PreRxDropsSegment verifies keep=false drops the segment before the +// state machine sees it: the payload is not buffered, connection state is +// untouched and PostRx never fires. +func TestPolicy_PreRxDropsSegment(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(4)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + server.SetPolicy(pol) + 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. + pol.keep = false + preRxBefore, postRxBefore := len(pol.preRx), len(pol.postRx) + + 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(pol.preRx) != preRxBefore+1 { + t.Fatalf("PreRx calls=%d, want %d (segment must reach PreRx)", len(pol.preRx), preRxBefore+1) + } + if len(pol.postRx) != postRxBefore { + t.Fatalf("PostRx calls=%d, want %d: a dropped segment was never accepted", len(pol.postRx), postRxBefore) + } + 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()) + } +} + +// TestPolicy_PreTxRetransmit verifies a PreTx retransmit directive drives +// go-back-N: the Handler rewinds the send sequence and the transmit buffer +// together and re-emits already-sent, unacknowledged data from snd.UNA. +func TestPolicy_PreTxRetransmit(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(5)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + client.SetPolicy(pol) + 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) + firstData := append([]byte(nil), buf[sizeHeaderTCP:n]...) + + // Direct go-back-N on the next transmit. + pol.rtxFrom, pol.retransmit = client.ControlBlock().SendUNA(), 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) + } + if got := buf[sizeHeaderTCP:n]; string(got) != string(firstData) { + t.Fatalf("retransmit payload=%q, want %q", got, firstData) + } +} + +// TestPolicy_PreTxRetransmitOutOfRange verifies an out-of-range rtxFrom is +// refused, leaving the send sequence and transmit buffer untouched. +func TestPolicy_PreTxRetransmitOutOfRange(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(9)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + client.SetPolicy(pol) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + if _, err := client.Write([]byte("payload")); err != nil { + t.Fatal("client write:", err) + } + clear(buf[:]) + if _, err := client.Send(buf[:]); err != nil { + t.Fatal("client send data:", err) + } + nxtBefore := client.ControlBlock().SendNext() + + // Well beyond snd.NXT: must be refused. + pol.rtxFrom, pol.retransmit = nxtBefore+1000, true + clear(buf[:]) + if _, err := client.Send(buf[:]); err != nil { + t.Fatal("client send:", err) + } + if got := client.ControlBlock().SendNext(); got != nxtBefore { + t.Fatalf("snd.NXT=%d, want unchanged %d: out-of-range rtxFrom must be refused", got, nxtBefore) + } +} + +// TestPolicy_HoldNew verifies holdNew suppresses new data while leaving control +// segments free to go out. +func TestPolicy_HoldNew(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(10)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + + pol := newRecordingPolicy() + client.SetPolicy(pol) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + pol.holdNew = true + if _, err := client.Write([]byte("payload")); err != nil { + t.Fatal("client write:", err) + } + clear(buf[:]) + n, err := client.Send(buf[:]) + if err != nil { + t.Fatal("client send:", err) + } + if n > sizeHeaderTCP { + t.Fatalf("holdNew must suppress new data, got %d payload bytes", n-sizeHeaderTCP) + } + + // Releasing the hold lets the same data out. + pol.holdNew = false + clear(buf[:]) + n, err = client.Send(buf[:]) + if err != nil { + t.Fatal("client send after hold:", err) + } + if n <= sizeHeaderTCP { + t.Fatal("data must flow once holdNew is cleared") + } +} + +// TestPolicy_ResetOnReopen verifies Reset fires on every (re)open and on Abort, +// so a single Policy value can be reused across connection reuse. +func TestPolicy_ResetOnReopen(t *testing.T) { + const mtu = ethernet.MaxMTU + client := newHandler(t, mtu, 3) + pol := newRecordingPolicy() + client.SetPolicy(pol) + + if err := client.OpenActive(1234, 5678, 0); err != nil { + t.Fatal("open 1:", err) + } + afterOpen := pol.resets + if afterOpen == 0 { + t.Fatal("Reset not called on first open") + } + + client.Abort() + if pol.resets <= afterOpen { + t.Fatalf("Reset not called on Abort: resets=%d, want >%d", pol.resets, afterOpen) + } + afterAbort := pol.resets + + if err := client.OpenActive(1234, 5678, 0); err != nil { + t.Fatal("open 2:", err) + } + if pol.resets <= afterAbort { + t.Fatalf("Reset not called on reopen: resets=%d, want >%d", pol.resets, afterAbort) + } +} diff --git a/tcp/rto/timer.go b/tcp/rto/timer.go index 3a65c70..141ed4a 100644 --- a/tcp/rto/timer.go +++ b/tcp/rto/timer.go @@ -3,6 +3,7 @@ package rto import ( "time" + "github.com/soypat/lneto" "github.com/soypat/lneto/tcp" ) @@ -35,15 +36,14 @@ const ( ) // Timer implements the RFC 6298 round-trip-time estimator and the single -// retransmission timer as a [tcp.Policy]. Construct it with new(Timer) and hand -// it to [tcp.ConnConfig.Policy]; the connection calls [Timer.Reset] on open, so -// the zero value is ready to use. +// retransmission timer as a [tcp.Policy]. Construct it with [NewTimer] and hand +// it to [tcp.ConnConfig.Policy]. // // Timer is a pure, reactive state machine: it observes the segments a connection -// sends and receives (via the tcp.Policy hooks) and the monotonic time handed -// in at each hook, and from those alone derives RTT estimates and retransmission -// decisions. It holds no clock and allocates nothing, which keeps it -// deterministic for unit testing (see issue #140). +// sends and receives (via the tcp.Policy hooks) and from those alone derives RTT +// estimates and retransmission decisions. The tcp package holds no clock, so the +// Timer carries its own; injecting it keeps the estimator deterministic for unit +// testing (see issue #140). // // Timer tracks its own shadow of the send sequence space purely from the segments // it observes: [Timer.PostTx] advances the highest sequence sent and [Timer.PreRx] @@ -53,6 +53,9 @@ const ( // whose sequence space is not beyond the shadow snd.NXT is a retransmission and // is never RTT-sampled. type Timer struct { + // nanotime is the monotonic time source in nanoseconds. Preserved by Reset. + nanotime func() int64 + srtt time.Duration // smoothed round-trip time (SRTT). rttvar time.Duration // round-trip-time variation (RTTVAR). rto time.Duration // current retransmission timeout. @@ -76,17 +79,28 @@ type Timer struct { // expirations counts timeouts since Reset. It exists so a policy sharing this // timer can notice a timeout it did not itself drive: a congestion controller - // must collapse its window on one, and when the timer is a peer in a - // [tcp.Composite] the controller never sees the timer's directive. + // must collapse its window on one, and a policy that composes the timer as a + // peer never sees the timer's own directive. expirations uint32 } var _ tcp.Policy = (*Timer)(nil) -// Reset returns the estimator to its pre-connection state with the initial RTO. -// It implements [tcp.Policy] and is called when the connection opens or aborts -// so the estimator can be reused across connection reuse. -func (r *Timer) Reset() { *r = Timer{rto: rtoInitial} } +// Configure prepares the Timer for use with nanotime, the monotonic time source +// in nanoseconds (the func() int64 convention used across lneto). It must be +// called before the connection is opened. +func (r *Timer) Configure(nanotime func() int64) error { + if nanotime == nil { + return lneto.ErrMissingHALConfig // The estimator cannot run without a clock. + } + *r = Timer{rto: rtoInitial, nanotime: nanotime} + return nil +} + +// Reset returns the estimator to its pre-connection state with the initial RTO, +// preserving the configured clock. It implements [tcp.Policy] and is called when +// the connection opens or aborts so the estimator survives connection reuse. +func (r *Timer) Reset() { *r = Timer{rto: rtoInitial, nanotime: r.nanotime} } // SmoothedRTT returns the current smoothed round-trip time (SRTT), or zero // before the first RTT measurement. It is concrete-type introspection and is @@ -115,7 +129,9 @@ func (r *Timer) Running() bool { return r.running } func (r *Timer) Expirations() uint32 { return r.expirations } // NextDeadline returns the monotonic-nanosecond instant at which the timer -// expires, or 0 when it is not armed. It implements [tcp.Policy]. +// expires, or 0 when it is not armed. It is concrete-type introspection, not +// part of [tcp.Policy]: an event loop that wants to schedule against the RTO +// holds the Timer it configured and reads this. func (r *Timer) NextDeadline() int64 { if !r.running { return 0 @@ -126,19 +142,22 @@ func (r *Timer) NextDeadline() int64 { // PreRx keeps every segment: the estimator never drops traffic and records // nothing before the connection has decided whether the segment counts. It // implements [tcp.Policy]. -func (r *Timer) PreRx(rx tcp.RxMeta) tcp.RxDirective { - return tcp.RxDirective{Keep: true} +func (r *Timer) PreRx(h *tcp.Handler, incoming tcp.Frame) bool { + return true } // PostRx samples the RTT and manages the retransmission timer from a segment the // connection accepted (RFC 6298 §5.2/§5.3). It implements [tcp.Policy]. // -// A refused segment is ignored. Acting on one would let an acknowledgement the -// state machine rejected, for data never sent, collapse the backoff and take a -// bogus RTT sample. -func (r *Timer) PostRx(event tcp.RxEvent) { - incoming, now := event.Segment, event.Now - if !event.Accepted || !r.haveSeq || !incoming.Flags.HasAny(tcp.FlagACK) { +// Only accepted segments reach here. Acting on a refused one would let an +// acknowledgement the state machine rejected, for data never sent, collapse the +// backoff and take a bogus RTT sample. +func (r *Timer) PostRx(h *tcp.Handler, prevState tcp.State, accepted tcp.Frame) { + r.postRx(accepted.Segment(len(accepted.Payload())), r.nanotime()) +} + +func (r *Timer) postRx(incoming tcp.Segment, now int64) { + if !r.haveSeq || !incoming.Flags.HasAny(tcp.FlagACK) { return } ack := incoming.ACK @@ -162,19 +181,18 @@ func (r *Timer) PostRx(event tcp.RxEvent) { } } -// WriteOptions adds no TCP options: retransmission timing needs none of its -// own. It implements [tcp.Policy]. -func (r *Timer) WriteOptions(plan tcp.TxPlan, opts []byte) uint8 { return 0 } - // PreTx reports whether the retransmission timer has expired and, if so, applies // the RFC 6298 §5.4–§5.6 timeout response — discard the outstanding RTT sample -// (Karn), back the RTO off exponentially and restart the timer — returning a -// directive that asks the connection to retransmit from snd.UNA (go-back-N). It -// implements [tcp.Policy]. -func (r *Timer) PreTx(intent tcp.TxIntent) tcp.TxDirective { - now := intent.Now +// (Karn), back the RTO off exponentially and restart the timer — and asks the +// connection to retransmit from snd.UNA (go-back-N). It writes no TCP options: +// retransmission timing needs none of its own. It implements [tcp.Policy]. +func (r *Timer) PreTx(h *tcp.Handler, outgoingOpts tcp.Frame) (rtxFrom tcp.Value, retransmit, holdNew bool) { + return r.preTx(r.nanotime(), h.ControlBlock().SendUNA()) +} + +func (r *Timer) preTx(now int64, una tcp.Value) (rtxFrom tcp.Value, retransmit, holdNew bool) { if !r.running || now < r.deadline || r.sndUNA == r.sndNXT { - return tcp.TxDirective{} + return 0, false, false } r.expirations++ r.timing = false // §5.4: do not sample a retransmitted segment. @@ -184,7 +202,7 @@ func (r *Timer) PreTx(intent tcp.TxIntent) tcp.TxDirective { } r.running = true r.deadline = now + int64(r.CurrentRTO()) - return tcp.TxDirective{Retransmit: true, RetransmitFrom: intent.UNA} + return una, true, false } // PostTx records an emitted segment: it advances the shadow send sequence, @@ -192,7 +210,11 @@ func (r *Timer) PreTx(intent tcp.TxIntent) tcp.TxDirective { // Segments that do not extend the send sequence are retransmissions and are // never RTT-sampled (Karn's algorithm). Control-only segments (no data) are // ignored. It implements [tcp.Policy]. -func (r *Timer) PostTx(outgoing tcp.Segment, now int64) { +func (r *Timer) PostTx(h *tcp.Handler, outgoing tcp.Frame) { + r.postTx(outgoing.Segment(len(outgoing.Payload())), r.nanotime()) +} + +func (r *Timer) postTx(outgoing tcp.Segment, now int64) { if outgoing.DATALEN == 0 { return // only data segments are timed / arm the RTO. } @@ -204,7 +226,7 @@ func (r *Timer) PostTx(outgoing tcp.Segment, now int64) { r.sndNXT = segStart } if !r.sndNXT.LessThan(segEnd) { - // tcp.Segment does not extend the send sequence: it is a retransmission. + // Segment does not extend the send sequence: it is a retransmission. // Discard any outstanding RTT sample per Karn's algorithm. The timer was // already (re)armed by PreTx on the timeout that triggered this resend. r.timing = false diff --git a/tcp/rto/timer_test.go b/tcp/rto/timer_test.go index e360c88..e0ceebc 100644 --- a/tcp/rto/timer_test.go +++ b/tcp/rto/timer_test.go @@ -21,25 +21,42 @@ func ackSeg(ack uint32) tcp.Segment { func newRTO() *Timer { var r Timer - r.Reset() + if err := r.Configure(func() int64 { return 0 }); err != nil { + panic(err) + } return &r } -// rxAt builds the minimal tcp.RxMeta for driving PreRx directly. -func rxAt(seg tcp.Segment, now int64) tcp.RxMeta { return tcp.RxMeta{Segment: seg, Now: now} } - -// acceptedAt builds the event for a segment the connection accepted, which is what -// drives the estimator. Timing state is only allowed to move for those. -func acceptedAt(seg tcp.Segment, now int64) tcp.RxEvent { - return tcp.RxEvent{Segment: seg, Now: now, Accepted: true} +// frameOf renders a segment as the wire frame the [tcp.Policy] hooks receive. +func frameOf(t *testing.T, s tcp.Segment) tcp.Frame { + t.Helper() + frm, err := tcp.NewFrame(make([]byte, 20+int(s.DATALEN))) + if err != nil { + t.Fatal(err) + } + frm.SetSegment(s, 5) + return frm } -// txAt builds the minimal tcp.TxIntent for driving Timer.PreTx directly: the timer -// tracks the send sequence itself via PostTx and only reads the clock. -func txAt(now int64) tcp.TxIntent { return tcp.TxIntent{Now: now} } +func TestRTO_Configure(t *testing.T) { + var r Timer + if err := r.Configure(nil); err == nil { + t.Error("Configure must reject a nil clock") + } + if err := r.Configure(func() int64 { return 0 }); err != nil { + t.Fatal(err) + } + if r.nanotime == nil { + t.Fatal("clock not stored") + } + r.Reset() + if r.nanotime == nil { + t.Error("Reset must preserve the configured clock") + } +} func TestRTO_Reset(t *testing.T) { - var r Timer + r := newRTO() r.Reset() if r.rto != rtoInitial { t.Errorf("initial rto=%v, want %v", r.rto, rtoInitial) @@ -61,7 +78,7 @@ func TestRTO_ArmOnSendSampleOnAck(t *testing.T) { r := newRTO() const iss = uint32(1000) - r.PostTx(dataSeg(iss, 100), 0) + r.postTx(dataSeg(iss, 100), 0) if !r.Running() { t.Fatal("timer must arm after sending data") } @@ -70,10 +87,10 @@ func TestRTO_ArmOnSendSampleOnAck(t *testing.T) { } // ACK arrives one RTT (40ms) later covering all sent data. - if !r.PreRx(rxAt(ackSeg(iss+100), 40*rtoMs)).Keep { + if !r.PreRx(nil, frameOf(t, ackSeg(iss+100))) { t.Error("PreRx must keep the segment") } - r.PostRx(acceptedAt(ackSeg(iss+100), 40*rtoMs)) + r.postRx(ackSeg(iss+100), 40*rtoMs) if r.Running() { t.Error("timer must stop once all data is acknowledged") } @@ -87,23 +104,26 @@ func TestRTO_ArmOnSendSampleOnAck(t *testing.T) { func TestRTO_RetransmitOnTimeout(t *testing.T) { r := newRTO() const iss = uint32(1000) - r.PostTx(dataSeg(iss, 100), 0) + r.postTx(dataSeg(iss, 100), 0) - if r.PreTx(txAt(int64(rtoInitial) - 1)).Retransmit { + if _, rtx, _ := r.preTx(int64(rtoInitial)-1, tcp.Value(iss)); rtx { t.Fatal("must not retransmit before the deadline") } - dir := r.PreTx(tcp.TxIntent{Now: int64(rtoInitial), UNA: tcp.Value(iss), NXT: tcp.Value(iss + 100)}) - if !dir.Retransmit { + from, rtx, hold := r.preTx(int64(rtoInitial), tcp.Value(iss)) + if !rtx { t.Fatal("RTO must fire at the deadline with data outstanding") } - if dir.RetransmitFrom != tcp.Value(iss) { - t.Errorf("retransmit from %d, want snd.UNA=%d", dir.RetransmitFrom, iss) + if hold { + t.Error("the estimator never holds new data back") + } + if from != tcp.Value(iss) { + t.Errorf("retransmit from %d, want snd.UNA=%d", from, iss) } if r.CurrentRTO() != 2*rtoInitial { t.Errorf("rto=%v after one backoff, want %v", r.CurrentRTO(), 2*rtoInitial) } - // The connection resends from snd.UNA; PostTx sees a retransmission. - r.PostTx(dataSeg(iss, 100), int64(rtoInitial)) + // The connection resends from snd.UNA; postTx sees a retransmission. + r.postTx(dataSeg(iss, 100), int64(rtoInitial)) if r.timing { t.Error("retransmitted segment must not be RTT-sampled (Karn)") } @@ -114,12 +134,12 @@ func TestRTO_RetransmitOnTimeout(t *testing.T) { func TestRTO_KarnNoSampleOnRetransmittedAck(t *testing.T) { r := newRTO() const iss = uint32(1000) - r.PostTx(dataSeg(iss, 100), 0) + r.postTx(dataSeg(iss, 100), 0) // Timeout and retransmit. - r.PreTx(txAt(int64(rtoInitial))) - r.PostTx(dataSeg(iss, 100), int64(rtoInitial)) + r.preTx(int64(rtoInitial), tcp.Value(iss)) + r.postTx(dataSeg(iss, 100), int64(rtoInitial)) // ACK now arrives; no sample should be taken since timing was discarded. - r.PostRx(acceptedAt(ackSeg(iss+100), int64(rtoInitial)+10*rtoMs)) + r.postRx(ackSeg(iss+100), int64(rtoInitial)+10*rtoMs) if r.haveRTT { t.Error("no RTT sample should exist after a retransmission (Karn)") } @@ -130,10 +150,10 @@ func TestRTO_KarnNoSampleOnRetransmittedAck(t *testing.T) { func TestRTO_TimerRestartsWhilePartiallyAcked(t *testing.T) { r := newRTO() const iss = uint32(1000) - r.PostTx(dataSeg(iss, 100), 0) - r.PostTx(dataSeg(iss+100, 100), 0) // 200 octets outstanding, iss..iss+200. + r.postTx(dataSeg(iss, 100), 0) + r.postTx(dataSeg(iss+100, 100), 0) // 200 octets outstanding, iss..iss+200. - r.PostRx(acceptedAt(ackSeg(iss+100), 40*rtoMs)) // acks first 100 only. + r.postRx(ackSeg(iss+100), 40*rtoMs) // acks first 100 only. if !r.Running() { t.Fatal("timer must remain armed while data is still in flight") } @@ -146,7 +166,7 @@ func TestRTO_TimerRestartsWhilePartiallyAcked(t *testing.T) { // nor start an RTT sample. func TestRTO_NoArmWithoutData(t *testing.T) { r := newRTO() - r.PostTx(tcp.Segment{SEQ: 1000, Flags: tcp.FlagACK}, 0) // pure ACK, DATALEN==0. + r.postTx(tcp.Segment{SEQ: 1000, Flags: tcp.FlagACK}, 0) // pure ACK, DATALEN==0. if r.Running() || r.timing { t.Error("pure control segment must not arm the timer or start a sample") } @@ -157,15 +177,15 @@ func TestRTO_NoArmWithoutData(t *testing.T) { func TestRTO_BackoffCollapsesOnValidSample(t *testing.T) { r := newRTO() const iss = uint32(1000) - r.PostTx(dataSeg(iss, 100), 0) - r.PreTx(txAt(int64(rtoInitial))) // one timeout: backoff=1. - r.PostTx(dataSeg(iss, 100), int64(rtoInitial)) // retransmit (no sample). + r.postTx(dataSeg(iss, 100), 0) + r.preTx(int64(rtoInitial), tcp.Value(iss)) // one timeout: backoff=1. + r.postTx(dataSeg(iss, 100), int64(rtoInitial)) // retransmit (no sample). if r.backoff != 1 { t.Fatalf("backoff=%d, want 1 after a timeout", r.backoff) } // New data sent and freshly sampled, then acked. - r.PostTx(dataSeg(iss+100, 100), int64(rtoInitial)+rtoMs) - r.PostRx(acceptedAt(ackSeg(iss+200), int64(rtoInitial)+30*rtoMs)) + r.postTx(dataSeg(iss+100, 100), int64(rtoInitial)+rtoMs) + r.postRx(ackSeg(iss+200), int64(rtoInitial)+30*rtoMs) if r.backoff != 0 { t.Errorf("backoff=%d, want 0 after a valid RTT sample", r.backoff) } @@ -173,8 +193,7 @@ func TestRTO_BackoffCollapsesOnValidSample(t *testing.T) { // TestRTO_Clamped verifies CurrentRTO is clamped to [rtoMin, rtoMax]. func TestRTO_Clamped(t *testing.T) { - var r Timer - r.Reset() + r := newRTO() r.rto = time.Nanosecond if got := r.CurrentRTO(); got != rtoMin { t.Errorf("CurrentRTO=%v, want floor %v", got, rtoMin) @@ -188,8 +207,7 @@ func TestRTO_Clamped(t *testing.T) { // TestRTO_UpdateRTTFirstSample verifies the first-measurement initialization of // SRTT/RTTVAR (RFC 6298 §2.2). func TestRTO_UpdateRTTFirstSample(t *testing.T) { - var r Timer - r.Reset() + r := newRTO() r.updateRTT(100 * time.Millisecond) if r.srtt != 100*time.Millisecond { t.Errorf("srtt=%v, want 100ms", r.srtt) @@ -203,53 +221,61 @@ func TestRTO_UpdateRTTFirstSample(t *testing.T) { } } -// TestRTO_ImplementsPolicy exercises Timer through the [tcp.Policy] -// interface: sending data arms a deadline and a full ACK disarms it. -func TestRTO_ImplementsPolicy(t *testing.T) { - var lr tcp.Policy = newRTO() - lr.Reset() - lr.PostTx(dataSeg(1000, 100), 0) - if lr.NextDeadline() == 0 { - t.Error("expected an armed deadline after sending data") +// TestRTO_PolicyHooksDeriveFromFrame exercises Timer through the [tcp.Policy] +// hooks, verifying it reads the segment out of the frame it is handed: sending +// data arms a deadline and a full ACK disarms it and yields the RTT sample. +func TestRTO_PolicyHooksDeriveFromFrame(t *testing.T) { + var clock int64 + var r Timer + if err := r.Configure(func() int64 { return clock }); err != nil { + t.Fatal(err) } - if !lr.PreRx(rxAt(ackSeg(1100), 10*rtoMs)).Keep { + var pol tcp.Policy = &r + pol.Reset() + + pol.PostTx(nil, frameOf(t, dataSeg(1000, 100))) + if r.NextDeadline() == 0 { + t.Fatal("expected an armed deadline after sending data") + } + clock = 10 * rtoMs + if !pol.PreRx(nil, frameOf(t, ackSeg(1100))) { t.Error("PreRx must keep") } - lr.PostRx(acceptedAt(ackSeg(1100), 10*rtoMs)) - if lr.NextDeadline() != 0 { + pol.PostRx(nil, tcp.StateEstablished, frameOf(t, ackSeg(1100))) + if r.NextDeadline() != 0 { t.Error("expected disarmed timer after full ack") } + if r.SmoothedRTT() != 10*time.Millisecond { + t.Errorf("srtt=%v, want 10ms sampled through the hooks", r.SmoothedRTT()) + } } -// TestRTO_IgnoresRejectedSegment verifies the estimator does not act on a segment -// the connection refused. PreRx runs before the state machine has judged the -// segment, so an acknowledgement for data never sent would otherwise collapse the -// backoff and take a bogus round-trip sample. -func TestRTO_IgnoresRejectedSegment(t *testing.T) { +// TestRTO_PreRxNeverDrops verifies the estimator keeps every segment and records +// nothing at PreRx time. Dropping is not its business, and the connection has not +// yet judged the segment: an acknowledgement for data never sent would otherwise +// collapse the backoff and take a bogus round-trip sample. Only accepted segments +// reach PostRx, which the Handler guarantees. +func TestRTO_PreRxNeverDrops(t *testing.T) { r := newRTO() const iss = uint32(1000) - r.PostTx(dataSeg(iss, 100), 0) + r.postTx(dataSeg(iss, 100), 0) armed := r.NextDeadline() if armed == 0 { t.Fatal("timer must be armed after sending data") } - // An acknowledgement far beyond anything sent, refused by the connection. - bogus := ackSeg(iss + 100000) - if !r.PreRx(rxAt(bogus, 40*rtoMs)).Keep { + // An acknowledgement far beyond anything sent, which the connection refuses. + if !r.PreRx(nil, frameOf(t, ackSeg(iss+100000))) { t.Error("PreRx must keep: dropping is not the estimator's business") } - r.PostRx(tcp.RxEvent{Segment: bogus, Now: 40 * rtoMs, Accepted: false}) - if r.NextDeadline() != armed { - t.Errorf("deadline moved to %d on a refused segment, want it left at %d", - r.NextDeadline(), armed) + t.Errorf("deadline moved to %d at PreRx, want it left at %d", r.NextDeadline(), armed) } if r.SmoothedRTT() != 0 { - t.Errorf("took an RTT sample of %v from a refused segment", r.SmoothedRTT()) + t.Errorf("took an RTT sample of %v at PreRx", r.SmoothedRTT()) } if !r.Running() { - t.Error("timer disarmed by a refused acknowledgement") + t.Error("timer disarmed at PreRx") } } @@ -262,25 +288,24 @@ func TestRTO_RetransmitsZeroWindowProbe(t *testing.T) { r := newRTO() const iss = uint32(5000) probe := dataSeg(iss, 1) // The one-octet probe. - r.PostTx(probe, 0) + r.postTx(probe, 0) now := int64(rtoInitial) prevRTO := r.CurrentRTO() for attempt := 1; attempt <= 4; attempt++ { - dir := r.PreTx(tcp.TxIntent{Now: now, UNA: tcp.Value(iss), NXT: tcp.Value(iss + 1)}) - if !dir.Retransmit { + from, rtx, _ := r.preTx(now, tcp.Value(iss)) + if !rtx { t.Fatalf("attempt %d: timer did not fire; the probe would never be resent", attempt) } - if dir.RetransmitFrom != tcp.Value(iss) { - t.Errorf("attempt %d: retransmit from %d, want the probe octet at %d", - attempt, dir.RetransmitFrom, iss) + if from != tcp.Value(iss) { + t.Errorf("attempt %d: retransmit from %d, want the probe octet at %d", attempt, from, iss) } if got := r.CurrentRTO(); got <= prevRTO { t.Errorf("attempt %d: rto %v did not back off past %v", attempt, got, prevRTO) } prevRTO = r.CurrentRTO() // The peer still cannot accept the octet, so it stays unacknowledged. - r.PostTx(probe, now) + r.postTx(probe, now) now += int64(prevRTO) } } diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 386b911..390a1ba 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -227,18 +227,39 @@ func (rtx *ringTx) RetransmitFromUNA() { if oldest == nil { return // Nothing in the retransmission queue. } - unaSeq := oldest.seq - if rtx.sentend != 0 { - // Merge sent region [sentoff, sentend) back into unsent. - rtx.unsentoff = rtx.sentoff - if rtx.unsentend == 0 { - rtx.unsentend = rtx.sentend - } - rtx.sentoff = 0 - rtx.sentend = 0 + rtx.RetransmitFrom(oldest.seq) +} + +// RetransmitFrom rewinds the transmit queue so sent-but-unacked data at and +// after seq becomes unsent again; the next MakePacket calls re-send it. seq is +// snapped down to the start of the packet containing it — the retransmission +// queue tracks whole packets, so sub-packet rewind is not representable. It is +// a no-op when seq is not covered by any queued packet (nothing to resend). +// +// Callers must pair this with [ControlBlock.RetransmitFrom] using the same seq +// so the send sequence space and the transmit buffer rewind together. +func (rtx *ringTx) RetransmitFrom(seq Value) { + pkt := rtx.slist.packetContaining(seq) + if pkt == nil { + return // seq not in the retransmission queue. } - // Clear packet metadata; sequence tracking restarts from UNA. - rtx.slist.Reset(cap(rtx.slist.pkts), unaSeq) + rewindOff, rewindSeq := pkt.off, pkt.seq + // The write position is unsentend, except when the unsent region is empty + // (unsentend==0) in which case data ends where the sent region ends. Capture + // it before reopening the unsent region over the rewound packets. + writeEnd := rtx.unsentend + if writeEnd == 0 { + writeEnd = rtx.sentend + } + if rewindOff == rtx.sentoff { + rtx.sentoff = 0 // Whole queue rewound: sent region becomes empty. + rtx.sentend = 0 + } else { + rtx.sentend = rewindOff + } + rtx.unsentoff = rewindOff + rtx.unsentend = writeEnd + rtx.slist.truncateFrom(rewindSeq) } func (rtx *ringTx) consolidateBufs() { @@ -331,6 +352,36 @@ func (sl *sentlist) Free() int { return cap(sl.pkts) - len(sl.pkts) } +// packetContaining returns the queued packet whose sequence range covers seq, or +// nil when no packet does. It is the floor lookup a retransmission rewind needs: +// seq lands inside a packet and the whole packet is resent. +func (sl *sentlist) packetContaining(seq Value) *ringidx { + for i := range sl.pkts { + pkt := &sl.pkts[i] + if pkt.seq.LessThanEq(seq) && seq.LessThan(pkt.endSeq()) { + return pkt + } + } + return nil +} + +// truncateFrom drops the packet starting at seq and every packet sent after it, +// so their data can be re-queued as unsent. seq must be a packet start sequence +// (see [sentlist.packetContaining]). When no packet survives, the auxiliary +// sequence counter is rewound to seq so [sentlist.EndSeq] keeps reporting where +// the next packet begins. +func (sl *sentlist) truncateFrom(seq Value) { + for i := range sl.pkts { + if sl.pkts[i].seq == seq { + sl.pkts = sl.pkts[:i] + if i == 0 { + sl.ssn = seq + } + return + } + } +} + func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value) *ringidx { free := sl.Free() if free == 0 { diff --git a/tcp/txqueue_retransmit_test.go b/tcp/txqueue_retransmit_test.go new file mode 100644 index 0000000..e778c21 --- /dev/null +++ b/tcp/txqueue_retransmit_test.go @@ -0,0 +1,215 @@ +package tcp + +import ( + "bytes" + "testing" +) + +// newRetransmitQueue builds a queue holding npkt sent packets of pktlen octets +// each, starting at iss, plus any leftover unsent data. It returns the queue and +// the full byte stream that was written. +func newRetransmitQueue(t *testing.T, bufsize, maxPkts, npkt, pktlen, unsent int, iss Value) (*ringTx, []byte) { + t.Helper() + var rtx ringTx + if err := rtx.Reset(make([]byte, bufsize), maxPkts, iss); err != nil { + t.Fatal(err) + } + stream := make([]byte, npkt*pktlen+unsent) + for i := range stream { + stream[i] = byte(i + 1) // Non-zero so a stale ring shows up as a mismatch. + } + if n, err := rtx.Write(stream); err != nil || n != len(stream) { + t.Fatalf("write n=%d err=%v", n, err) + } + seq := iss + scratch := make([]byte, pktlen) + for i := 0; i < npkt; i++ { + n, err := rtx.MakePacket(scratch, seq) + if err != nil { + t.Fatalf("packet %d: %v", i, err) + } + if n != pktlen { + t.Fatalf("packet %d: n=%d, want %d", i, n, pktlen) + } + seq += Value(n) + } + testQueueSanity(t, &rtx) + return &rtx, stream +} + +// mustRemake asserts the queue re-emits datalen octets at seq matching want. +func mustRemake(t *testing.T, rtx *ringTx, seq Value, want []byte) { + t.Helper() + got := make([]byte, len(want)) + n, err := rtx.MakePacket(got, seq) + if err != nil { + t.Fatalf("MakePacket at seq %d: %v", seq, err) + } + if n != len(want) { + t.Fatalf("MakePacket at seq %d: n=%d, want %d", seq, n, len(want)) + } + if !bytes.Equal(got, want) { + t.Fatalf("MakePacket at seq %d: got %v, want %v", seq, got, want) + } +} + +// TestRingTx_RetransmitFromBoundary rewinds to the start of the second of three +// sent packets: the first stays sent, the rest become unsent and re-emit their +// original bytes. +func TestRingTx_RetransmitFromBoundary(t *testing.T) { + const iss, pktlen = Value(100), 4 + rtx, stream := newRetransmitQueue(t, 64, 4, 3, pktlen, 0, iss) + + sentBefore := rtx.BufferedSent() + rtx.RetransmitFrom(iss + pktlen) // Start of packet 2. + testQueueSanity(t, rtx) + + if got := rtx.BufferedSent(); got != pktlen { + t.Fatalf("sent=%d, want %d (only packet 1 remains sent)", got, pktlen) + } + if got := rtx.BufferedUnsent(); got != sentBefore-pktlen { + t.Fatalf("unsent=%d, want %d", got, sentBefore-pktlen) + } + mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen]) + testQueueSanity(t, rtx) + mustRemake(t, rtx, iss+2*pktlen, stream[2*pktlen:3*pktlen]) + testQueueSanity(t, rtx) +} + +// TestRingTx_RetransmitFromMidPacket verifies a sequence inside a packet is +// snapped down to that packet's start: the queue tracks whole packets. +func TestRingTx_RetransmitFromMidPacket(t *testing.T) { + const iss, pktlen = Value(100), 4 + rtx, stream := newRetransmitQueue(t, 64, 4, 3, pktlen, 0, iss) + + rtx.RetransmitFrom(iss + pktlen + 2) // Two octets into packet 2. + testQueueSanity(t, rtx) + + if got := rtx.BufferedSent(); got != pktlen { + t.Fatalf("sent=%d, want %d: rewind must floor to the packet start", got, pktlen) + } + mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen]) +} + +// TestRingTx_RetransmitFromOldest rewinds the whole queue, which must match +// RetransmitFromUNA. +func TestRingTx_RetransmitFromOldest(t *testing.T) { + const iss, pktlen, npkt = Value(100), 4, 3 + rtx, stream := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss) + rtx.RetransmitFrom(iss) + testQueueSanity(t, rtx) + + viaUNA, _ := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss) + viaUNA.RetransmitFromUNA() + testQueueSanity(t, viaUNA) + + if rtx.BufferedSent() != 0 { + t.Fatalf("sent=%d, want 0 after a full rewind", rtx.BufferedSent()) + } + if rtx.BufferedUnsent() != npkt*pktlen { + t.Fatalf("unsent=%d, want %d", rtx.BufferedUnsent(), npkt*pktlen) + } + if rtx.BufferedSent() != viaUNA.BufferedSent() || rtx.BufferedUnsent() != viaUNA.BufferedUnsent() { + t.Fatal("RetransmitFrom(oldest) must match RetransmitFromUNA") + } + mustRemake(t, rtx, iss, stream[:pktlen]) +} + +// TestRingTx_RetransmitFromUnknownSeq verifies a sequence covered by no queued +// packet leaves the queue untouched. +func TestRingTx_RetransmitFromUnknownSeq(t *testing.T) { + const iss, pktlen, npkt = Value(100), 4, 3 + rtx, _ := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss) + sent, unsent := rtx.BufferedSent(), rtx.BufferedUnsent() + + rtx.RetransmitFrom(iss - 1) // Before the queue. + rtx.RetransmitFrom(iss + npkt*pktlen) // One past the last octet sent. + rtx.RetransmitFrom(iss + 1000) // Far beyond. + testQueueSanity(t, rtx) + + if rtx.BufferedSent() != sent || rtx.BufferedUnsent() != unsent { + t.Fatalf("queue moved: sent %d→%d, unsent %d→%d", sent, rtx.BufferedSent(), unsent, rtx.BufferedUnsent()) + } +} + +// TestRingTx_RetransmitWithUnsentTail verifies a rewind reopens the unsent region +// over the rewound packets without losing the unsent tail behind them. +func TestRingTx_RetransmitWithUnsentTail(t *testing.T) { + const iss, pktlen, npkt, tail = Value(100), 4, 2, 5 + rtx, stream := newRetransmitQueue(t, 64, 4, npkt, pktlen, tail, iss) + if got := rtx.BufferedUnsent(); got != tail { + t.Fatalf("unsent tail=%d, want %d", got, tail) + } + + rtx.RetransmitFrom(iss + pktlen) // Rewind the second packet only. + testQueueSanity(t, rtx) + + if got := rtx.BufferedUnsent(); got != pktlen+tail { + t.Fatalf("unsent=%d, want %d (rewound packet plus the tail)", got, pktlen+tail) + } + // The rewound packet re-emits first, then the tail follows in order. + mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen]) + testQueueSanity(t, rtx) + mustRemake(t, rtx, iss+2*pktlen, stream[2*pktlen:]) +} + +// TestRingTx_RetransmitAfterDrainedUnsent pins the write-position recovery: when +// every octet written has been packetized the unsent region is empty, so the +// rewind must reconstruct where data ends from the sent region. +func TestRingTx_RetransmitAfterDrainedUnsent(t *testing.T) { + const iss, pktlen, npkt = Value(100), 4, 3 + rtx, stream := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss) + if got := rtx.BufferedUnsent(); got != 0 { + t.Fatalf("unsent=%d, want 0: all written data was packetized", got) + } + + rtx.RetransmitFrom(iss + pktlen) + testQueueSanity(t, rtx) + + if got := rtx.BufferedUnsent(); got != 2*pktlen { + t.Fatalf("unsent=%d, want %d: rewind lost the end of the data", got, 2*pktlen) + } + mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen]) + testQueueSanity(t, rtx) + mustRemake(t, rtx, iss+2*pktlen, stream[2*pktlen:3*pktlen]) +} + +// TestRingTx_RetransmitWrapped exercises a rewind on a queue whose regions wrap +// the end of the ring buffer. +func TestRingTx_RetransmitWrapped(t *testing.T) { + const bufsize, pktlen = 16, 4 + const iss = Value(100) + var rtx ringTx + if err := rtx.Reset(make([]byte, bufsize), 4, iss); err != nil { + t.Fatal(err) + } + // Push the queue most of the way around the ring, acking as we go. + seq := iss + scratch := make([]byte, pktlen) + for round := 0; round < 3; round++ { + chunk := make([]byte, pktlen) + for i := range chunk { + chunk[i] = byte(round*pktlen + i + 1) + } + if _, err := rtx.Write(chunk); err != nil { + t.Fatal(err) + } + if _, err := rtx.MakePacket(scratch, seq); err != nil { + t.Fatal(err) + } + seq += Value(pktlen) + if round < 2 { + if err := rtx.RecvACK(seq); err != nil { + t.Fatal(err) + } + } + testQueueSanity(t, &rtx) + } + // Two packets outstanding, straddling the wrap. Rewind the newest. + rewindSeq := seq - Value(pktlen) + want := append([]byte(nil), scratch...) + rtx.RetransmitFrom(rewindSeq) + testQueueSanity(t, &rtx) + mustRemake(t, &rtx, rewindSeq, want) + testQueueSanity(t, &rtx) +}