4cb7b0d fix(dgram): account paced wire bytes · vev

M internal/adapters/dgram/network_sim_test.go +2 -2


@@ -306,7 +306,7 @@ func TestTransportFloodClassification(t *testing.T) {
306	306		}
307	307		})
308	308
309		-	t.Run("bounded byte queue advances and releases on fake clock", func(t *testing.T) {
	309	+	t.Run("bandwidth_queue", func(t *testing.T) {
310	310		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
311	311		const (
312	312		mtu = 128

@@ -333,7 +333,7 @@ func TestTransportFloodClassification(t *testing.T) {
333	333		reverse := newSimulatedLink(clk, packetPolicy{
334	334		Delay: oneWayDelay,
335	335		MaxQueueBytes: queueLimit,
336		-	BytesPerSecond: 10 * bandwidth,
	336	+	BytesPerSecond: bandwidth,
337	337		})
338	338		aPC, bPC := newFloodPair(nil)
339	339		aPC.link = forward


@@ -306,7 +306,7 @@ func TestTransportFloodClassification(t *testing.T) {
306		}	306		}
307		})	307		})
308			308
309	-	t.Run("bounded byte queue advances and releases on fake clock", func(t *testing.T) {	309	+	t.Run("bandwidth_queue", func(t *testing.T) {
310		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))	310		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
311		const (	311		const (
312		mtu = 128	312		mtu = 128

@@ -333,7 +333,7 @@ func TestTransportFloodClassification(t *testing.T) {
333		reverse := newSimulatedLink(clk, packetPolicy{	333		reverse := newSimulatedLink(clk, packetPolicy{
334		Delay: oneWayDelay,	334		Delay: oneWayDelay,
335		MaxQueueBytes: queueLimit,	335		MaxQueueBytes: queueLimit,
336	-	BytesPerSecond: 10 * bandwidth,	336	+	BytesPerSecond: bandwidth,
337		})	337		})
338		aPC, bPC := newFloodPair(nil)	338		aPC, bPC := newFloodPair(nil)
339		aPC.link = forward	339		aPC.link = forward

M internal/adapters/dgram/transport.go +20 -7


@@ -395,7 +395,7 @@ func (t *Transport) send(f ports.Frame, async bool) error {
395	395		if len(t.pending) == 0 {
396	396		t.health.pendingStarted(now)
397	397		}
398		-	t.pending[seq] = &pending{frame: f, enqueued: now, wireBytes: dataRecordHeaderSize + len(f.Payload)}
	398	+	t.pending[seq] = &pending{frame: f, enqueued: now}
399	399		}
400	400		var done chan error
401	401		if !async {

@@ -1035,20 +1035,33 @@ func (t Transport) writeDataJob(pacer bytePacer, job dataSendJob) error {
1035	1035		if beforeDataPace != nil {
1036	1036		beforeDataPace()
1037	1037		}
	1038	+	packets := make([][]byte, 0, len(frags))
	1039	+	wireBytes := 0
	1040	+	for _, frag := range frags {
	1041	+	raw, err := pdgram.MarshalFragment(frag)
	1042	+	if err != nil {
	1043	+	return err
	1044	+	}
	1045	+	pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
	1046	+	packets = append(packets, pkt)
	1047	+	wireBytes += len(pkt)
	1048	+	}
	1049	+	if job.reliable {
	1050	+	t.mu.Lock()
	1051	+	if pending := t.pending[job.seq]; pending != nil {
	1052	+	pending.wireBytes = wireBytes
	1053	+	}
	1054	+	t.mu.Unlock()
	1055	+	}
1038	1056		limits := func() (int, int) {
1039	1057		t.mu.Lock()
1040	1058		defer t.mu.Unlock()
1041	1059		return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
1042	1060		}
1043		-	for i, frag := range frags {
	1061	+	for i, pkt := range packets {
1044	1062		if job.reliable && !t.pendingExists(job.seq) {
1045	1063		return nil
1046	1064		}
1047		-	raw, err := pdgram.MarshalFragment(frag)
1048		-	if err != nil {
1049		-	return err
1050		-	}
1051		-	pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
1052	1065		if err := pacer.wait(t.done, len(pkt), limits); err != nil {
1053	1066		return err
1054	1067		}

 		if len(t.pending) == 0 {
 			t.health.pendingStarted(now)
 		}
-		t.pending[seq] = &pending{frame: f, enqueued: now, wireBytes: dataRecordHeaderSize + len(f.Payload)}
+		t.pending[seq] = &pending{frame: f, enqueued: now}
 	}
 	var done chan error
 	if !async {
 	if beforeDataPace != nil {
 		beforeDataPace()
 	}
+	packets := make([][]byte, 0, len(frags))
+	wireBytes := 0
+	for _, frag := range frags {
+		raw, err := pdgram.MarshalFragment(frag)
+		if err != nil {
+			return err
+		}
+		pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
+		packets = append(packets, pkt)
+		wireBytes += len(pkt)
+	}
+	if job.reliable {
+		t.mu.Lock()
+		if pending := t.pending[job.seq]; pending != nil {
+			pending.wireBytes = wireBytes
+		}
+		t.mu.Unlock()
+	}
 	limits := func() (int, int) {
 		t.mu.Lock()
 		defer t.mu.Unlock()
 		return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
 	}
-	for i, frag := range frags {
+	for i, pkt := range packets {
 		if job.reliable && !t.pendingExists(job.seq) {
 			return nil
 		}
-		raw, err := pdgram.MarshalFragment(frag)
-		if err != nil {
-			return err
-		}
-		pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
 		if err := pacer.wait(t.done, len(pkt), limits); err != nil {
 			return err
 		}

M internal/adapters/dgram/transport_test.go +42 -0


@@ -329,6 +329,48 @@ func BenchmarkCongestionControllerACK(b *testing.B) {
329	329		}
330	330		}
331	331
	332	+	func TestPendingWireBytesMatchSealedDatagrams(t *testing.T) {
	333	+	const mtu = 128
	334	+	aPC, bPC := newPair()
	335	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	336	+	MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
	337	+	})
	338	+	if err != nil {
	339	+	t.Fatal(err)
	340	+	}
	341	+	defer func() { _ = a.Close() }()
	342	+
	343	+	maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead() - 16
	344	+	frame := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, maxFragmentData-dataRecordHeaderSize+1)}
	345	+	if err := a.Send(frame); err != nil {
	346	+	t.Fatal(err)
	347	+	}
	348	+
	349	+	want := 0
	350	+	for len(bPC.in) > 0 {
	351	+	want += len((<-bPC.in).b)
	352	+	}
	353	+	a.mu.Lock()
	354	+	got := a.pending[1].wireBytes
	355	+	initialCwnd := a.congestion.cwndBytes
	356	+	a.mu.Unlock()
	357	+	if got != want {
	358	+	t.Fatalf("pending wire bytes=%d, want sealed datagram bytes=%d", got, want)
	359	+	}
	360	+
	361	+	ack := make([]byte, 9)
	362	+	ack[0] = recAck
	363	+	binary.BigEndian.PutUint64(ack[1:], 1)
	364	+	a.handleRecord(ack)
	365	+	a.mu.Lock()
	366	+	gotCwnd := a.congestion.cwndBytes
	367	+	a.mu.Unlock()
	368	+	wantCwnd := initialCwnd + max(1, mtu*want/initialCwnd)
	369	+	if gotCwnd != wantCwnd {
	370	+	t.Fatalf("cwnd after ACK=%d, want %d from %d acknowledged sealed bytes", gotCwnd, wantCwnd, want)
	371	+	}
	372	+	}
	373	+
332	374		func TestSendAsyncTransfersFragmentedRecordToBytePacedSender(t *testing.T) {
333	375		const mtu = 128
334	376		aPC, bPC := newPair()


@@ -329,6 +329,48 @@ func BenchmarkCongestionControllerACK(b *testing.B) {
329	}	329		}
330	}	330		}
331		331
		332	+	func TestPendingWireBytesMatchSealedDatagrams(t *testing.T) {
		333	+	const mtu = 128
		334	+	aPC, bPC := newPair()
		335	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
		336	+	MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
		337	+	})
		338	+	if err != nil {
		339	+	t.Fatal(err)
		340	+	}
		341	+	defer func() { _ = a.Close() }()
		342	+
		343	+	maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead() - 16
		344	+	frame := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, maxFragmentData-dataRecordHeaderSize+1)}
		345	+	if err := a.Send(frame); err != nil {
		346	+	t.Fatal(err)
		347	+	}
		348	+
		349	+	want := 0
		350	+	for len(bPC.in) > 0 {
		351	+	want += len((<-bPC.in).b)
		352	+	}
		353	+	a.mu.Lock()
		354	+	got := a.pending[1].wireBytes
		355	+	initialCwnd := a.congestion.cwndBytes
		356	+	a.mu.Unlock()
		357	+	if got != want {
		358	+	t.Fatalf("pending wire bytes=%d, want sealed datagram bytes=%d", got, want)
		359	+	}
		360	+
		361	+	ack := make([]byte, 9)
		362	+	ack[0] = recAck
		363	+	binary.BigEndian.PutUint64(ack[1:], 1)
		364	+	a.handleRecord(ack)
		365	+	a.mu.Lock()
		366	+	gotCwnd := a.congestion.cwndBytes
		367	+	a.mu.Unlock()
		368	+	wantCwnd := initialCwnd + max(1, mtu*want/initialCwnd)
		369	+	if gotCwnd != wantCwnd {
		370	+	t.Fatalf("cwnd after ACK=%d, want %d from %d acknowledged sealed bytes", gotCwnd, wantCwnd, want)
		371	+	}
		372	+	}
		373	+
332	func TestSendAsyncTransfersFragmentedRecordToBytePacedSender(t *testing.T) {	374		func TestSendAsyncTransfersFragmentedRecordToBytePacedSender(t *testing.T) {
333	const mtu = 128	375		const mtu = 128
334	aPC, bPC := newPair()	376		aPC, bPC := newPair()