diff --git a/tcp/conn.go b/tcp/conn.go index 547563d..f1e63e0 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -80,7 +80,7 @@ type ConnConfig struct { // 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 + 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 @@ -105,7 +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) + conn.h.SetPolicy(config.LossRecovery) return nil } diff --git a/tcp/control.go b/tcp/control.go index 16960a7..859e971 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -257,14 +257,20 @@ func (tcb *ControlBlock) HasPendingRetransmit() bool { return tcb._state.TxDataOpen() && tcb.dupack >= retransmitAfterDupacks && tcb.nRetransmit <= tcb.dupack-retransmitAfterDupacks } +func (tcb *ControlBlock) RetransmitFrom(newNxt Value) bool { + panic("not yet implemented") + tcb.snd.NXT = newNxt + tcb.dupack = 0 + tcb.nRetransmit = 0 + return true +} + // 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 + tcb.RetransmitFrom(tcb.snd.UNA) } // PendingSegment calculates a suitable next segment to send from a payload length. diff --git a/tcp/handler.go b/tcp/handler.go index 9ff7a29..6621035 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -31,13 +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 - // 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 + reasm reassembly + policy Policy nanotime func() int64 closing bool @@ -79,27 +74,10 @@ 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() +func (h *Handler) SetPolicy(policy Policy) { + h.policy = policy } +func (h *Handler) policyEnabled() bool { return h.policy != nil } // LocalPort returns the local port of the connection. Returns 0 if the connection is closed and uninitialized. func (h *Handler) LocalPort() uint16 { @@ -165,7 +143,7 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { shutdownRx: false, // Persist configuration across reopen: validator: h.validator, - loss: h.loss, + policy: h.policy, nanotime: h.nanotime, logger: h.logger, // persist memory across repoen: @@ -173,8 +151,8 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { bufRx: h.bufRx, reasm: h.reasm, } - if h.lossEnabled() { - h.loss.Reset() + if h.policyEnabled() { + h.policy.Reset() } h.reasm.clear() // preserve metadata capacity across reopen, drop held segments. h.bufTx.ResetOrReuse(nil, 0, iss) @@ -212,9 +190,7 @@ 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 { + if h.policyEnabled() && !h.policy.PreRx(h, tfrm) { return nil } @@ -380,16 +356,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 { + if h.policyEnabled() { + tfrm, err := NewFrame(b) + if err != nil { + return 0, err + } + 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.RetransmitAll() - h.bufTx.RetransmitFromUNA() + h.scb.RetransmitFrom(rtxFrom) } } awaitingSyn := h.AwaitingSynSend() @@ -474,8 +452,8 @@ 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) + if h.policyEnabled() { + h.policy.PostTx(h, tfrm) } h.requeueControl = false tfrm.SetSourcePort(h.localPort) diff --git a/tcp/loss.go b/tcp/loss.go deleted file mode 100644 index 98cfe5c..0000000 --- a/tcp/loss.go +++ /dev/null @@ -1,83 +0,0 @@ -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/policy.go b/tcp/policy.go new file mode 100644 index 0000000..814815c --- /dev/null +++ b/tcp/policy.go @@ -0,0 +1,15 @@ +package tcp + +type Policy interface { + 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. + 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. + PostRx(h *Handler, prevState State, accepted Frame) + // PostTx called on leaving the transmit path. + PostTx(h *Handler, outgoing Frame) +} diff --git a/tcp/rto.go b/tcp/rto/timer.go similarity index 59% rename from tcp/rto.go rename to tcp/rto/timer.go index 180d518..3a65c70 100644 --- a/tcp/rto.go +++ b/tcp/rto/timer.go @@ -1,6 +1,10 @@ -package tcp +package rto -import "time" +import ( + "time" + + "github.com/soypat/lneto/tcp" +) // RFC 6298 retransmission-timeout (RTO) parameters. The algorithm keeps a // single retransmission timer per connection (RFC 6298 §5): the timer is @@ -30,61 +34,67 @@ const ( backoffMax = 12 ) -// RTO implements the RFC 6298 round-trip-time estimator and the single -// retransmission timer as a [LossRecovery]. Construct it with new(RTO) and hand -// it to [ConnConfig.LossRecovery]; the connection calls [RTO.Reset] on open, so +// 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. // -// RTO is a pure, reactive state machine: it observes the segments a connection -// sends and receives (via the LossRecovery hooks) and the monotonic time handed +// 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). // -// RTO tracks its own shadow of the send sequence space purely from the segments -// it observes: [RTO.PostTx] advances the highest sequence sent and [RTO.PreRx] +// 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] // advances the highest sequence acknowledged. This is what lets it manage the // timer (RFC 6298 §5.2/§5.3) without reaching into the tcp state machine, and it // is also how retransmissions are distinguished for Karn's algorithm — a segment // whose sequence space is not beyond the shadow snd.NXT is a retransmission and // is never RTT-sampled. -type RTO struct { +type Timer struct { srtt time.Duration // smoothed round-trip time (SRTT). rttvar time.Duration // round-trip-time variation (RTTVAR). rto time.Duration // current retransmission timeout. haveRTT bool // false until the first RTT sample is taken. // Shadow of the send sequence space, derived from observed segments. - haveSeq bool // false until the first data segment is observed. - sndUNA Value // highest acknowledged sequence number seen on the wire. - sndNXT Value // one past the highest sequence number sent. + haveSeq bool // false until the first data segment is observed. + sndUNA tcp.Value // highest acknowledged sequence number seen on the wire. + sndNXT tcp.Value // one past the highest sequence number sent. // RTT sampling state (Karn's algorithm, RFC 6298 §3): at most one segment is // timed at a time and retransmitted segments are never sampled. timing bool - timedSeq Value // ACK at or beyond this value completes the sample. - timedAt int64 // send time (monotonic ns) of the timed segment. + timedSeq tcp.Value // ACK at or beyond this value completes the sample. + timedAt int64 // send time (monotonic ns) of the timed segment. // Retransmission timer state. running bool deadline int64 // time (monotonic ns) at which the timer expires. backoff uint8 // consecutive timeouts, for exponential backoff. + + // 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. + expirations uint32 } -var _ LossRecovery = (*RTO)(nil) +var _ tcp.Policy = (*Timer)(nil) // Reset returns the estimator to its pre-connection state with the initial RTO. -// It implements [LossRecovery] and is called when the connection opens or aborts +// It implements [tcp.Policy] and is called when the connection opens or aborts // so the estimator can be reused across connection reuse. -func (r *RTO) Reset() { *r = RTO{rto: rtoInitial} } +func (r *Timer) Reset() { *r = Timer{rto: rtoInitial} } // SmoothedRTT returns the current smoothed round-trip time (SRTT), or zero // before the first RTT measurement. It is concrete-type introspection and is -// intentionally not part of [LossRecovery]. -func (r *RTO) SmoothedRTT() time.Duration { return r.srtt } +// intentionally not part of [tcp.Policy]. +func (r *Timer) SmoothedRTT() time.Duration { return r.srtt } // CurrentRTO returns the timeout currently in effect, clamped to [rtoMin, rtoMax]. -func (r *RTO) CurrentRTO() time.Duration { +func (r *Timer) CurrentRTO() time.Duration { rto := r.rto if rto < rtoMin { rto = rtoMin @@ -95,23 +105,41 @@ func (r *RTO) CurrentRTO() time.Duration { } // Running reports whether the retransmission timer is currently armed. -func (r *RTO) Running() bool { return r.running } +func (r *Timer) Running() bool { return r.running } + +// Expirations returns how many times the retransmission timer has expired since +// [Timer.Reset]. A policy that shares this timer rather than driving it watches +// this for a change to learn that a timeout happened, since it never sees the +// timer's own directive. It is concrete-type introspection and is intentionally +// not part of [tcp.Policy]. +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 [LossRecovery]. -func (r *RTO) NextDeadline() int64 { +// expires, or 0 when it is not armed. It implements [tcp.Policy]. +func (r *Timer) NextDeadline() int64 { if !r.running { return 0 } return r.deadline } -// PreRx samples the RTT and manages the retransmission timer from a received -// segment (RFC 6298 §5.2/§5.3). It implements [LossRecovery] and always keeps -// the segment (the estimator never drops traffic). -func (r *RTO) PreRx(incoming Segment, now int64) RxDirective { - if !r.haveSeq || !incoming.Flags.HasAny(FlagACK) { - return RxDirective{Keep: true} +// 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} +} + +// 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) { + return } ack := incoming.ACK if r.timing && !ack.LessThan(r.timedSeq) { @@ -132,18 +160,23 @@ func (r *RTO) PreRx(incoming Segment, now int64) RxDirective { r.running = true r.deadline = now + int64(r.CurrentRTO()) } - return RxDirective{Keep: true} } +// 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 [LossRecovery]. -func (r *RTO) PreTx(now int64) TxDirective { +// implements [tcp.Policy]. +func (r *Timer) PreTx(intent tcp.TxIntent) tcp.TxDirective { + now := intent.Now if !r.running || now < r.deadline || r.sndUNA == r.sndNXT { - return TxDirective{} + return tcp.TxDirective{} } + r.expirations++ r.timing = false // §5.4: do not sample a retransmitted segment. if r.backoff < backoffMax { r.backoff++ @@ -151,27 +184,27 @@ func (r *RTO) PreTx(now int64) TxDirective { } r.running = true r.deadline = now + int64(r.CurrentRTO()) - return TxDirective{RetransmitAll: true} + return tcp.TxDirective{Retransmit: true, RetransmitFrom: intent.UNA} } // PostTx records an emitted segment: it advances the shadow send sequence, // begins timing newly transmitted data (RFC 6298 §3) and arms the timer (§5.1). // 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 [LossRecovery]. -func (r *RTO) PostTx(outgoing Segment, now int64) { +// ignored. It implements [tcp.Policy]. +func (r *Timer) PostTx(outgoing tcp.Segment, now int64) { if outgoing.DATALEN == 0 { return // only data segments are timed / arm the RTO. } segStart := outgoing.SEQ - segEnd := segStart + Value(outgoing.LEN()) + segEnd := segStart + tcp.Value(outgoing.LEN()) if !r.haveSeq { r.haveSeq = true r.sndUNA = segStart r.sndNXT = segStart } if !r.sndNXT.LessThan(segEnd) { - // Segment does not extend the send sequence: it is a retransmission. + // tcp.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 @@ -189,9 +222,20 @@ func (r *RTO) PostTx(outgoing Segment, now int64) { } } +// ObserveRTT folds a round-trip measurement taken by other means into the +// estimator, for a policy that composes this timer and can measure the round trip +// more accurately than acknowledgement timing allows. The RFC 7323 timestamp echo +// is the case this exists for. +// +// Unlike the timer's own sampling this does not apply Karn's algorithm, because a +// sample derived from an echoed timestamp is unambiguous even when the segment +// carrying it was a retransmission (RFC 7323 §4.1). Non-positive samples are +// ignored. +func (r *Timer) ObserveRTT(rtt time.Duration) { r.updateRTT(rtt) } + // updateRTT folds a round-trip measurement into SRTT/RTTVAR/RTO using the // integer-shift form of RFC 6298 §2.2/§2.3. -func (r *RTO) updateRTT(sample time.Duration) { +func (r *Timer) updateRTT(sample time.Duration) { if sample <= 0 { return } diff --git a/tcp/rto/timer_test.go b/tcp/rto/timer_test.go new file mode 100644 index 0000000..e360c88 --- /dev/null +++ b/tcp/rto/timer_test.go @@ -0,0 +1,286 @@ +package rto + +import ( + "testing" + "time" + + "github.com/soypat/lneto/tcp" +) + +const rtoMs = int64(time.Millisecond) + +// dataSeg builds a data segment of datalen octets starting at seq. +func dataSeg(seq uint32, datalen int) tcp.Segment { + return tcp.Segment{SEQ: tcp.Value(seq), DATALEN: tcp.Size(datalen), Flags: tcp.FlagPSH | tcp.FlagACK} +} + +// ackSeg builds a bare ACK acknowledging up to ack. +func ackSeg(ack uint32) tcp.Segment { + return tcp.Segment{ACK: tcp.Value(ack), Flags: tcp.FlagACK} +} + +func newRTO() *Timer { + var r Timer + r.Reset() + 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} +} + +// 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_Reset(t *testing.T) { + var r Timer + r.Reset() + if r.rto != rtoInitial { + t.Errorf("initial rto=%v, want %v", r.rto, rtoInitial) + } + if r.CurrentRTO() != rtoInitial { + t.Errorf("CurrentRTO=%v, want %v", r.CurrentRTO(), rtoInitial) + } + if r.haveRTT { + t.Error("haveRTT should be false before first sample") + } + if r.Running() || r.NextDeadline() != 0 { + t.Error("timer must be disarmed after Reset") + } +} + +// TestRTO_ArmOnSendSampleOnAck sends data, verifies the timer arms, then acks it +// and verifies an RTT sample is taken and the timer stops once all data is acked. +func TestRTO_ArmOnSendSampleOnAck(t *testing.T) { + r := newRTO() + const iss = uint32(1000) + + r.PostTx(dataSeg(iss, 100), 0) + if !r.Running() { + t.Fatal("timer must arm after sending data") + } + if r.NextDeadline() != int64(rtoInitial) { + t.Errorf("deadline=%d, want %d", r.NextDeadline(), int64(rtoInitial)) + } + + // ACK arrives one RTT (40ms) later covering all sent data. + if !r.PreRx(rxAt(ackSeg(iss+100), 40*rtoMs)).Keep { + t.Error("PreRx must keep the segment") + } + r.PostRx(acceptedAt(ackSeg(iss+100), 40*rtoMs)) + if r.Running() { + t.Error("timer must stop once all data is acknowledged") + } + if r.SmoothedRTT() != 40*time.Millisecond { + t.Errorf("srtt=%v, want 40ms", r.SmoothedRTT()) + } +} + +// TestRTO_RetransmitOnTimeout verifies PreTx directs a go-back-N retransmit once +// the deadline passes with data outstanding, and backs the RTO off. +func TestRTO_RetransmitOnTimeout(t *testing.T) { + r := newRTO() + const iss = uint32(1000) + r.PostTx(dataSeg(iss, 100), 0) + + if r.PreTx(txAt(int64(rtoInitial) - 1)).Retransmit { + 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 { + 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 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)) + if r.timing { + t.Error("retransmitted segment must not be RTT-sampled (Karn)") + } +} + +// TestRTO_KarnNoSampleOnRetransmittedAck verifies that after a retransmission the +// ACK does not produce an RTT sample (Karn's algorithm). +func TestRTO_KarnNoSampleOnRetransmittedAck(t *testing.T) { + r := newRTO() + const iss = uint32(1000) + r.PostTx(dataSeg(iss, 100), 0) + // Timeout and retransmit. + r.PreTx(txAt(int64(rtoInitial))) + 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)) + if r.haveRTT { + t.Error("no RTT sample should exist after a retransmission (Karn)") + } +} + +// TestRTO_TimerRestartsWhilePartiallyAcked verifies the timer restarts (not +// stops) when an ACK advances UNA but data remains in flight (RFC 6298 §5.3). +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.PostRx(acceptedAt(ackSeg(iss+100), 40*rtoMs)) // acks first 100 only. + if !r.Running() { + t.Fatal("timer must remain armed while data is still in flight") + } + if r.NextDeadline() != 40*rtoMs+int64(r.CurrentRTO()) { + t.Errorf("deadline=%d, want %d", r.NextDeadline(), 40*rtoMs+int64(r.CurrentRTO())) + } +} + +// TestRTO_NoArmWithoutData verifies control-only segments neither arm the timer +// 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. + if r.Running() || r.timing { + t.Error("pure control segment must not arm the timer or start a sample") + } +} + +// TestRTO_BackoffCollapsesOnValidSample verifies a valid RTT measurement +// collapses the exponential backoff counter (RFC 6298 §5.7). +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). + 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)) + if r.backoff != 0 { + t.Errorf("backoff=%d, want 0 after a valid RTT sample", r.backoff) + } +} + +// TestRTO_Clamped verifies CurrentRTO is clamped to [rtoMin, rtoMax]. +func TestRTO_Clamped(t *testing.T) { + var r Timer + r.Reset() + r.rto = time.Nanosecond + if got := r.CurrentRTO(); got != rtoMin { + t.Errorf("CurrentRTO=%v, want floor %v", got, rtoMin) + } + r.rto = time.Hour + if got := r.CurrentRTO(); got != rtoMax { + t.Errorf("CurrentRTO=%v, want ceiling %v", got, rtoMax) + } +} + +// 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.updateRTT(100 * time.Millisecond) + if r.srtt != 100*time.Millisecond { + t.Errorf("srtt=%v, want 100ms", r.srtt) + } + if r.rttvar != 50*time.Millisecond { + t.Errorf("rttvar=%v, want 50ms", r.rttvar) + } + // RTO = SRTT + K*RTTVAR = 100 + 4*50 = 300ms. + if r.rto != 300*time.Millisecond { + t.Errorf("rto=%v, want 300ms", r.rto) + } +} + +// 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") + } + if !lr.PreRx(rxAt(ackSeg(1100), 10*rtoMs)).Keep { + t.Error("PreRx must keep") + } + lr.PostRx(acceptedAt(ackSeg(1100), 10*rtoMs)) + if lr.NextDeadline() != 0 { + t.Error("expected disarmed timer after full ack") + } +} + +// 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) { + r := newRTO() + const iss = uint32(1000) + 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 { + 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) + } + if r.SmoothedRTT() != 0 { + t.Errorf("took an RTT sample of %v from a refused segment", r.SmoothedRTT()) + } + if !r.Running() { + t.Error("timer disarmed by a refused acknowledgement") + } +} + +// TestRTO_RetransmitsZeroWindowProbe verifies the timer takes over the periodic +// probing of a closed send window. A zero-window probe is a single octet the peer +// cannot accept, so it goes unacknowledged; the timer must keep resending it, with +// exponential backoff, which is the persist-timer behaviour of RFC 9293 §3.8.6.1. +// The tcp package relies on this and refuses to probe without a policy installed. +func TestRTO_RetransmitsZeroWindowProbe(t *testing.T) { + r := newRTO() + const iss = uint32(5000) + probe := dataSeg(iss, 1) // The one-octet probe. + 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 { + 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 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) + now += int64(prevRTO) + } +} diff --git a/tcp/rto_test.go b/tcp/rto_test.go deleted file mode 100644 index 44686a2..0000000 --- a/tcp/rto_test.go +++ /dev/null @@ -1,206 +0,0 @@ -package tcp - -import ( - "testing" - "time" -) - -const rtoMs = int64(time.Millisecond) - -// rtoDataSeg builds a data segment of datalen octets starting at seq. -func rtoDataSeg(seq uint32, datalen int) Segment { - return Segment{SEQ: Value(seq), DATALEN: Size(datalen), Flags: FlagPSH | FlagACK} -} - -// rtoAckSeg builds a bare ACK acknowledging up to ack. -func rtoAckSeg(ack uint32) Segment { - return Segment{ACK: Value(ack), Flags: FlagACK} -} - -func newRTO() *RTO { - var r RTO - r.Reset() - return &r -} - -func TestRTO_Reset(t *testing.T) { - var r RTO - r.Reset() - if r.rto != rtoInitial { - t.Errorf("initial rto=%v, want %v", r.rto, rtoInitial) - } - if r.CurrentRTO() != rtoInitial { - t.Errorf("CurrentRTO=%v, want %v", r.CurrentRTO(), rtoInitial) - } - if r.haveRTT { - t.Error("haveRTT should be false before first sample") - } - if r.Running() || r.NextDeadline() != 0 { - t.Error("timer must be disarmed after Reset") - } -} - -// TestRTO_ArmOnSendSampleOnAck sends data, verifies the timer arms, then acks it -// and verifies an RTT sample is taken and the timer stops once all data is acked. -func TestRTO_ArmOnSendSampleOnAck(t *testing.T) { - r := newRTO() - const iss = uint32(1000) - - r.PostTx(rtoDataSeg(iss, 100), 0) - if !r.Running() { - t.Fatal("timer must arm after sending data") - } - if r.NextDeadline() != int64(rtoInitial) { - t.Errorf("deadline=%d, want %d", r.NextDeadline(), int64(rtoInitial)) - } - - // ACK arrives one RTT (40ms) later covering all sent data. - dir := r.PreRx(rtoAckSeg(iss+100), 40*rtoMs) - if !dir.Keep { - t.Error("PreRx must keep the segment") - } - if r.Running() { - t.Error("timer must stop once all data is acknowledged") - } - if r.SmoothedRTT() != 40*time.Millisecond { - t.Errorf("srtt=%v, want 40ms", r.SmoothedRTT()) - } -} - -// TestRTO_RetransmitOnTimeout verifies PreTx directs a go-back-N retransmit once -// the deadline passes with data outstanding, and backs the RTO off. -func TestRTO_RetransmitOnTimeout(t *testing.T) { - r := newRTO() - const iss = uint32(1000) - r.PostTx(rtoDataSeg(iss, 100), 0) - - if r.PreTx(int64(rtoInitial) - 1).RetransmitAll { - t.Fatal("must not retransmit before the deadline") - } - dir := r.PreTx(int64(rtoInitial)) - if !dir.RetransmitAll { - t.Fatal("RTO must fire at the deadline with data outstanding") - } - 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(rtoDataSeg(iss, 100), int64(rtoInitial)) - if r.timing { - t.Error("retransmitted segment must not be RTT-sampled (Karn)") - } -} - -// TestRTO_KarnNoSampleOnRetransmittedAck verifies that after a retransmission the -// ACK does not produce an RTT sample (Karn's algorithm). -func TestRTO_KarnNoSampleOnRetransmittedAck(t *testing.T) { - r := newRTO() - const iss = uint32(1000) - r.PostTx(rtoDataSeg(iss, 100), 0) - // Timeout and retransmit. - r.PreTx(int64(rtoInitial)) - r.PostTx(rtoDataSeg(iss, 100), int64(rtoInitial)) - // ACK now arrives; no sample should be taken since timing was discarded. - r.PreRx(rtoAckSeg(iss+100), int64(rtoInitial)+10*rtoMs) - if r.haveRTT { - t.Error("no RTT sample should exist after a retransmission (Karn)") - } -} - -// TestRTO_TimerRestartsWhilePartiallyAcked verifies the timer restarts (not -// stops) when an ACK advances UNA but data remains in flight (RFC 6298 §5.3). -func TestRTO_TimerRestartsWhilePartiallyAcked(t *testing.T) { - r := newRTO() - const iss = uint32(1000) - r.PostTx(rtoDataSeg(iss, 100), 0) - r.PostTx(rtoDataSeg(iss+100, 100), 0) // 200 octets outstanding, iss..iss+200. - - dir := r.PreRx(rtoAckSeg(iss+100), 40*rtoMs) // acks first 100 only. - if !r.Running() { - t.Fatal("timer must remain armed while data is still in flight") - } - if r.NextDeadline() != 40*rtoMs+int64(r.CurrentRTO()) { - t.Errorf("deadline=%d, want %d", r.NextDeadline(), 40*rtoMs+int64(r.CurrentRTO())) - } - if !dir.Keep { - t.Error("PreRx must keep the segment") - } -} - -// TestRTO_NoArmWithoutData verifies control-only segments neither arm the timer -// nor start an RTT sample. -func TestRTO_NoArmWithoutData(t *testing.T) { - r := newRTO() - r.PostTx(Segment{SEQ: 1000, Flags: 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") - } -} - -// TestRTO_BackoffCollapsesOnValidSample verifies a valid RTT measurement -// collapses the exponential backoff counter (RFC 6298 §5.7). -func TestRTO_BackoffCollapsesOnValidSample(t *testing.T) { - r := newRTO() - const iss = uint32(1000) - r.PostTx(rtoDataSeg(iss, 100), 0) - r.PreTx(int64(rtoInitial)) // one timeout: backoff=1. - r.PostTx(rtoDataSeg(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(rtoDataSeg(iss+100, 100), int64(rtoInitial)+rtoMs) - r.PreRx(rtoAckSeg(iss+200), int64(rtoInitial)+30*rtoMs) - if r.backoff != 0 { - t.Errorf("backoff=%d, want 0 after a valid RTT sample", r.backoff) - } -} - -// TestRTO_Clamped verifies CurrentRTO is clamped to [rtoMin, rtoMax]. -func TestRTO_Clamped(t *testing.T) { - var r RTO - r.Reset() - r.rto = time.Nanosecond - if got := r.CurrentRTO(); got != rtoMin { - t.Errorf("CurrentRTO=%v, want floor %v", got, rtoMin) - } - r.rto = time.Hour - if got := r.CurrentRTO(); got != rtoMax { - t.Errorf("CurrentRTO=%v, want ceiling %v", got, rtoMax) - } -} - -// TestRTO_UpdateRTTFirstSample verifies the first-measurement initialization of -// SRTT/RTTVAR (RFC 6298 §2.2). -func TestRTO_UpdateRTTFirstSample(t *testing.T) { - var r RTO - r.Reset() - r.updateRTT(100 * time.Millisecond) - if r.srtt != 100*time.Millisecond { - t.Errorf("srtt=%v, want 100ms", r.srtt) - } - if r.rttvar != 50*time.Millisecond { - t.Errorf("rttvar=%v, want 50ms", r.rttvar) - } - // RTO = SRTT + K*RTTVAR = 100 + 4*50 = 300ms. - if r.rto != 300*time.Millisecond { - t.Errorf("rto=%v, want 300ms", r.rto) - } -} - -// TestRTO_ImplementsLossRecovery exercises RTO through the [LossRecovery] -// interface: sending data arms a deadline and a full ACK disarms it. -func TestRTO_ImplementsLossRecovery(t *testing.T) { - var lr LossRecovery = newRTO() - lr.Reset() - lr.PostTx(rtoDataSeg(1000, 100), 0) - if lr.NextDeadline() == 0 { - t.Error("expected an armed deadline after sending data") - } - if !lr.PreRx(rtoAckSeg(1100), 10*rtoMs).Keep { - t.Error("PreRx must keep") - } - if lr.NextDeadline() != 0 { - t.Error("expected disarmed timer after full ack") - } -}