diff --git a/tcp/handler.go b/tcp/handler.go index 632c26b..deb6ece 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -368,14 +368,14 @@ func (h *Handler) Send(b []byte) (int, error) { return 0, err } offset := uint8(5) - var holdNew bool + txLimit := TransmitUnlimited if h.policyEnabled() { // 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 + limit, rtxFrom, doRtx := h.policy.PreTx(h, tfrm) + txLimit = limit 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 @@ -441,10 +441,9 @@ func (h *Handler) Send(b []byte) (int, error) { } else { var ok bool 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 + if txLimit < Size(maxPayload) && !h.nextSegmentIsRetransmit() { + // Policy clamped new data. + maxPayload = int(txLimit) } segment, ok = h.scb.PendingSegment(maxPayload) segment.WND = h.recvWindow() diff --git a/tcp/policy_test.go b/tcp/policy_test.go index e86ccae..283c20a 100644 --- a/tcp/policy_test.go +++ b/tcp/policy_test.go @@ -21,7 +21,7 @@ type recordingPolicy struct { keep bool // PreRx result. Default true (see newRecordingPolicy). rtxFrom Value retransmit bool - holdNew bool + txLimit Size // PreTx new-data limit. Default TransmitUnlimited (see newRecordingPolicy). // writeOpts, when non-empty, is appended as TCP options by PreTx. writeOpts []byte } @@ -34,7 +34,9 @@ type txRecord struct { dport uint16 } -func newRecordingPolicy() *recordingPolicy { return &recordingPolicy{keep: true} } +func newRecordingPolicy() *recordingPolicy { + return &recordingPolicy{keep: true, txLimit: TransmitUnlimited} +} var _ Policy = (*recordingPolicy)(nil) @@ -49,7 +51,7 @@ 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) { +func (p *recordingPolicy) PreTx(h *Handler, outgoingOpts Frame) (Size, Value, bool) { p.preTx++ if len(p.writeOpts) > 0 { // Raise the offset first: Options() is sized from it. @@ -57,7 +59,7 @@ func (p *recordingPolicy) PreTx(h *Handler, outgoingOpts Frame) (Value, bool, bo outgoingOpts.SetOffsetAndFlags(words, 0) copy(outgoingOpts.Options(), p.writeOpts) } - return p.rtxFrom, p.retransmit, p.holdNew + return p.txLimit, p.rtxFrom, p.retransmit } func (p *recordingPolicy) PostTx(h *Handler, outgoing Frame) { @@ -365,9 +367,10 @@ func TestPolicy_PreTxRetransmitOutOfRange(t *testing.T) { } } -// TestPolicy_HoldNew verifies holdNew suppresses new data while leaving control -// segments free to go out. -func TestPolicy_HoldNew(t *testing.T) { +// TestPolicy_TransmitLimit verifies the PreTx new-data limit caps the payload +// sent while leaving control segments free to go out: a zero limit suppresses +// data entirely, a partial limit truncates the segment. +func TestPolicy_TransmitLimit(t *testing.T) { const mtu = ethernet.MaxMTU rng := rand.New(rand.NewSource(10)) client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) @@ -378,8 +381,9 @@ func TestPolicy_HoldNew(t *testing.T) { var buf [mtu]byte establish(t, client, server, buf[:]) - pol.holdNew = true - if _, err := client.Write([]byte("payload")); err != nil { + const payload = "payload" + pol.txLimit = 0 + if _, err := client.Write([]byte(payload)); err != nil { t.Fatal("client write:", err) } clear(buf[:]) @@ -388,18 +392,29 @@ func TestPolicy_HoldNew(t *testing.T) { t.Fatal("client send:", err) } if n > sizeHeaderTCP { - t.Fatalf("holdNew must suppress new data, got %d payload bytes", n-sizeHeaderTCP) + t.Fatalf("a zero limit must suppress new data, got %d payload bytes", n-sizeHeaderTCP) } - // Releasing the hold lets the same data out. - pol.holdNew = false + // A partial limit lets only that many bytes out. + pol.txLimit = 3 clear(buf[:]) n, err = client.Send(buf[:]) if err != nil { - t.Fatal("client send after hold:", err) + t.Fatal("client send under partial limit:", err) } - if n <= sizeHeaderTCP { - t.Fatal("data must flow once holdNew is cleared") + if got := n - sizeHeaderTCP; got != int(pol.txLimit) { + t.Fatalf("got %d payload bytes, want the limit of %d", got, pol.txLimit) + } + + // Releasing the limit lets the rest of the data out. + pol.txLimit = TransmitUnlimited + clear(buf[:]) + n, err = client.Send(buf[:]) + if err != nil { + t.Fatal("client send after limit lifted:", err) + } + if got := n - sizeHeaderTCP; got != len(payload)-3 { + t.Fatalf("got %d payload bytes, want the remaining %d once unlimited", got, len(payload)-3) } } diff --git a/tcp/rto/timer.go b/tcp/rto/timer.go index 141ed4a..9ea4a07 100644 --- a/tcp/rto/timer.go +++ b/tcp/rto/timer.go @@ -184,15 +184,17 @@ func (r *Timer) postRx(incoming tcp.Segment, now int64) { // 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 — 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) { +// connection to retransmit from snd.UNA (go-back-N). It writes no TCP options +// and imposes no transmit limit: retransmission timing needs neither, and +// congestion control belongs to a Policy composing this timer. It implements +// [tcp.Policy]. +func (r *Timer) PreTx(h *tcp.Handler, outgoingOpts tcp.Frame) (newTransmitLimit tcp.Size, rtxFrom tcp.Value, retransmit bool) { return r.preTx(r.nanotime(), h.ControlBlock().SendUNA()) } -func (r *Timer) preTx(now int64, una tcp.Value) (rtxFrom tcp.Value, retransmit, holdNew bool) { +func (r *Timer) preTx(now int64, una tcp.Value) (newTransmitLimit tcp.Size, rtxFrom tcp.Value, retransmit bool) { if !r.running || now < r.deadline || r.sndUNA == r.sndNXT { - return 0, false, false + return tcp.TransmitUnlimited, 0, false } r.expirations++ r.timing = false // §5.4: do not sample a retransmitted segment. @@ -202,7 +204,7 @@ func (r *Timer) preTx(now int64, una tcp.Value) (rtxFrom tcp.Value, retransmit, } r.running = true r.deadline = now + int64(r.CurrentRTO()) - return una, true, false + return tcp.TransmitUnlimited, una, true } // PostTx records an emitted segment: it advances the shadow send sequence, diff --git a/tcp/rto/timer_test.go b/tcp/rto/timer_test.go index e0ceebc..14355a3 100644 --- a/tcp/rto/timer_test.go +++ b/tcp/rto/timer_test.go @@ -106,15 +106,15 @@ func TestRTO_RetransmitOnTimeout(t *testing.T) { const iss = uint32(1000) r.postTx(dataSeg(iss, 100), 0) - if _, rtx, _ := r.preTx(int64(rtoInitial)-1, tcp.Value(iss)); rtx { + if _, _, rtx := r.preTx(int64(rtoInitial)-1, tcp.Value(iss)); rtx { t.Fatal("must not retransmit before the deadline") } - from, rtx, hold := r.preTx(int64(rtoInitial), tcp.Value(iss)) + limit, from, rtx := r.preTx(int64(rtoInitial), tcp.Value(iss)) if !rtx { t.Fatal("RTO must fire at the deadline with data outstanding") } - if hold { - t.Error("the estimator never holds new data back") + if limit != tcp.TransmitUnlimited { + t.Error("the estimator never limits new data") } if from != tcp.Value(iss) { t.Errorf("retransmit from %d, want snd.UNA=%d", from, iss) @@ -293,7 +293,7 @@ func TestRTO_RetransmitsZeroWindowProbe(t *testing.T) { now := int64(rtoInitial) prevRTO := r.CurrentRTO() for attempt := 1; attempt <= 4; attempt++ { - from, rtx, _ := r.preTx(now, tcp.Value(iss)) + _, 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) }