tcp.Policy: add newTransmitLimit output

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)
+		limit, rtxFrom, doRtx := h.policy.PreTx(h, tfrm)
-		holdNew = hold
+		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() {
+		if txLimit < Size(maxPayload) && !h.nextSegmentIsRetransmit() {
-			// Policy is holding new data back (congestion window exhausted).
+			// Policy clamped new data.
-			// A retransmission it directed in this same call still proceeds.
+			maxPayload = int(txLimit)
-			maxPayload = 0
 		}
 		segment, ok = h.scb.PendingSegment(maxPayload)
 		segment.WND = h.recvWindow()

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
+// TestPolicy_TransmitLimit verifies the PreTx new-data limit caps the payload
-// segments free to go out.
+// sent while leaving control segments free to go out: a zero limit suppresses
-func TestPolicy_HoldNew(t *testing.T) {
+// 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
+	const payload = "payload"
-	if _, err := client.Write([]byte("payload")); err != nil {
+	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.
+	// A partial limit lets only that many bytes out.
-	pol.holdNew = false
+	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 {
+	if got := n - sizeHeaderTCP; got != int(pol.txLimit) {
-		t.Fatal("data must flow once holdNew is cleared")
+		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)
 	}
 }
 

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:
+// 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].
+// and imposes no transmit limit: retransmission timing needs neither, and
-func (r *Timer) PreTx(h *tcp.Handler, outgoingOpts tcp.Frame) (rtxFrom tcp.Value, retransmit, holdNew bool) {
+// 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,

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 {
+	if limit != tcp.TransmitUnlimited {
-		t.Error("the estimator never holds new data back")
+		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)
 		}