Skip to content

Commit 2b08887

Browse files
committed
perfvar checkpoint
1 parent cb0a26c commit 2b08887

15 files changed

Lines changed: 1065 additions & 125 deletions

ip.go

Lines changed: 219 additions & 75 deletions
Large diffs are not rendered by default.

ip_flow_limit_test.go

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,9 @@ func TestLocalUserNatSettingsMemoryScaled(t *testing.T) {
3232
AssertEqual(t, tcpSettings.SequenceBufferSize, 384)
3333
AssertEqual(t, tcpSettings.ReadBufferByteCount, 24576)
3434
AssertEqual(t, tcpSettings.MinWindowSize, uint32(65536))
35-
// 384 KiB scaled, snapped down to a power of 2 multiple of the min window
36-
AssertEqual(t, tcpSettings.MaxWindowSize, uint32(262144))
35+
AssertEqual(t, tcpSettings.InitialWindowSize, uint32(262144))
36+
// 6 MiB scaled, snapped down to a power of 2 multiple of the min window.
37+
AssertEqual(t, tcpSettings.MaxWindowSize, uint32(4194304))
3738
AssertEqual(t, tcpSettings.GlobalLimit, 192)
3839

3940
icmpSettings := DefaultIcmpBufferSettings()
@@ -56,7 +57,8 @@ func TestLocalUserNatSettingsMemoryScaled(t *testing.T) {
5657
tcpSettings = DefaultTcpBufferSettings()
5758
AssertEqual(t, tcpSettings.SequenceBufferSize, 192)
5859
AssertEqual(t, tcpSettings.ReadBufferByteCount, 16384)
59-
AssertEqual(t, tcpSettings.MaxWindowSize, uint32(131072))
60+
AssertEqual(t, tcpSettings.InitialWindowSize, uint32(131072))
61+
AssertEqual(t, tcpSettings.MaxWindowSize, uint32(2097152))
6062
AssertEqual(t, tcpSettings.GlobalLimit, 64)
6163
icmpSettings = DefaultIcmpBufferSettings()
6264
AssertEqual(t, icmpSettings.SequenceBufferSize, 16)
@@ -74,7 +76,8 @@ func TestLocalUserNatSettingsMemoryScaled(t *testing.T) {
7476
tcpSettings = DefaultTcpBufferSettings()
7577
AssertEqual(t, tcpSettings.SequenceBufferSize, 1024)
7678
AssertEqual(t, tcpSettings.ReadBufferByteCount, 65536)
77-
AssertEqual(t, tcpSettings.MaxWindowSize, uint32(1048576))
79+
AssertEqual(t, tcpSettings.InitialWindowSize, uint32(1048576))
80+
AssertEqual(t, tcpSettings.MaxWindowSize, uint32(16777216))
7881
AssertEqual(t, tcpSettings.GlobalLimit, 0)
7982
icmpSettings = DefaultIcmpBufferSettings()
8083
AssertEqual(t, icmpSettings.SequenceBufferSize, 64)
@@ -101,19 +104,22 @@ func TestLocalUserNatSettingsMemoryScaled(t *testing.T) {
101104
AssertEqual(t, budgetedProviderSettings.UdpBufferSettings.GlobalLimit, 768)
102105
SetMemoryBudget(0)
103106

104-
// invariants at every budget tier:
105-
// - the tcp channel depth must cover the max window in mtu packets, so a
106-
// full window burst is never dropped (the nat implements no retransmit
107-
// toward the socket)
108-
// - the max window stays a power of 2 multiple of the min window (the
109-
// window doubling ladder must land exactly on the max)
107+
// Invariants at every budget tier:
108+
// - the TCP channel remains memory-scaled rather than expanding to the
109+
// high-BDP maximum window for every live flow; and
110+
// - the maximum window stays a power-of-two multiple of the minimum window,
111+
// so the window-doubling ladder lands exactly on the maximum.
110112
for _, budget := range []ByteCount{0, mib(8), mib(16), mib(24), mib(32), mib(48), mib(64), mib(128)} {
111113
SetMemoryBudget(budget)
112114
tcpSettings := DefaultTcpBufferSettings()
113-
payloadByteCount := tcpSettings.Mtu - Ipv6HeaderSize - TcpHeaderSizeWithoutExtensions
114-
if int64(tcpSettings.SequenceBufferSize)*int64(payloadByteCount) < int64(tcpSettings.MaxWindowSize) {
115-
t.Errorf("budget %d: tcp depth %d x payload %d does not cover the max window %d",
116-
budget, tcpSettings.SequenceBufferSize, payloadByteCount, tcpSettings.MaxWindowSize)
115+
wantSequenceBufferSize := MemoryScaledCount(defaultTcpFlowBufferSize, 192)
116+
if tcpSettings.SequenceBufferSize != wantSequenceBufferSize {
117+
t.Errorf(
118+
"budget %d: tcp depth=%d, want memory-scaled depth=%d",
119+
budget,
120+
tcpSettings.SequenceBufferSize,
121+
wantSequenceBufferSize,
122+
)
117123
}
118124
if tcpSettings.MaxWindowSize < tcpSettings.MinWindowSize {
119125
t.Errorf("budget %d: max window %d below min window %d",

ip_packet_gopacket_test.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package connect
33
import (
44
"bytes"
55
"encoding/binary"
6+
"math"
67
"net"
78
"testing"
89

@@ -67,7 +68,7 @@ func TestSynAckMatchesGopacketReference(t *testing.T) {
6768
Ack: state.sendSeq,
6869
ACK: true,
6970
SYN: true,
70-
Window: state.encodedWindowSize(),
71+
Window: uint16(min(state.windowSize, uint32(math.MaxUint16))),
7172
Options: opts,
7273
}
7374
tcp.SetNetworkLayerForChecksum(ip)

ip_provider_fanout_test.go

Lines changed: 15 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,9 @@ import (
1111

1212
const providerColdPageTestFlowCount = 192
1313

14-
// TestProviderNatTargetUsesMinimalLosslessTcpQueue verifies an explicit
15-
// provider target removes generic process-profile headroom from every flow
16-
// without shrinking below one full advertised tcp window.
17-
func TestProviderNatTargetUsesMinimalLosslessTcpQueue(t *testing.T) {
14+
// An explicit provider target must not expand each flow's preallocated queue
15+
// to the much larger demand-driven high-BDP window.
16+
func TestProviderNatTargetKeepsBoundedTcpQueue(t *testing.T) {
1817
defer SetMemoryBudget(0)
1918

2019
cases := []struct {
@@ -27,28 +26,28 @@ func TestProviderNatTargetUsesMinimalLosslessTcpQueue(t *testing.T) {
2726
}
2827
for _, c := range cases {
2928
SetMemoryBudget(c.memoryBudgetByteCount)
29+
defaultSettings := DefaultTcpBufferSettings()
3030
settings := DefaultProviderLocalUserNatSettingsWithMemoryTarget(4 * 1024 * 1024)
3131
tcpSettings := settings.TcpBufferSettings
32-
payloadByteCount := tcpSettings.Mtu - Ipv6HeaderSize - TcpHeaderSizeWithoutExtensions
33-
want := int(
34-
(tcpSettings.MaxWindowSize + uint32(payloadByteCount) - 1) /
35-
uint32(payloadByteCount),
36-
)
37-
if tcpSettings.SequenceBufferSize != want {
32+
if tcpSettings.SequenceBufferSize != defaultSettings.SequenceBufferSize {
3833
t.Errorf(
39-
"budget=%d tcp queue=%d, want minimal max-window depth %d",
34+
"budget=%d tcp queue=%d, want bounded default depth=%d",
4035
c.memoryBudgetByteCount,
4136
tcpSettings.SequenceBufferSize,
42-
want,
37+
defaultSettings.SequenceBufferSize,
4338
)
4439
}
45-
if tcpSettings.SequenceBufferSize*payloadByteCount < int(tcpSettings.MaxWindowSize) {
40+
payloadByteCount := tcpSettings.Mtu - Ipv6HeaderSize - TcpHeaderSizeWithoutExtensions
41+
windowPacketCount := int(
42+
(tcpSettings.MaxWindowSize + uint32(payloadByteCount) - 1) /
43+
uint32(payloadByteCount),
44+
)
45+
if windowPacketCount <= tcpSettings.SequenceBufferSize {
4646
t.Errorf(
47-
"budget=%d tcp queue %d x payload %d does not cover window %d",
47+
"budget=%d high-BDP window depth=%d did not exceed bounded queue=%d",
4848
c.memoryBudgetByteCount,
49+
windowPacketCount,
4950
tcpSettings.SequenceBufferSize,
50-
payloadByteCount,
51-
tcpSettings.MaxWindowSize,
5251
)
5352
}
5453
}

ip_tcp_options_test.go

Lines changed: 200 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,200 @@
1+
// These tests and benchmarks isolate provider-side TCP option negotiation from
2+
// carrier performance, including packet ownership and asymmetric path limits.
3+
package connect
4+
5+
import (
6+
"encoding/binary"
7+
"net"
8+
"testing"
9+
10+
"github.com/urnetwork/connect/protocol"
11+
)
12+
13+
// Builds a return packetizer with the same IPv4 and timestamp overhead as an
14+
// established provider flow.
15+
func newTcpOptionPacketizationState(peerMss uint32) *ConnectionState {
16+
return &ConnectionState{
17+
ipVersion: 4,
18+
sourceIp: net.IPv4(10, 0, 0, 1).To4(),
19+
sourcePort: 40000,
20+
destinationIp: net.IPv4(203, 0, 113, 7).To4(),
21+
destinationPort: 443,
22+
receiveSeq: 1000,
23+
sendSeq: 2000,
24+
windowSize: 1024 * 1024,
25+
windowScale: 5,
26+
enableTimestamp: true,
27+
timestampRecent: 3000,
28+
peerMss: peerMss,
29+
timestampValueForTest: func() uint32 {
30+
return 4000
31+
},
32+
}
33+
}
34+
35+
// A smaller source MTU produces a smaller advertised MSS. The provider must
36+
// subtract its timestamp option and keep every return packet inside that MTU.
37+
func TestTcpSequenceHonorsPeerMssInReturnPacketization(t *testing.T) {
38+
settings := DefaultTcpBufferSettingsWithBufferSize(8)
39+
sequence := newTcpSequenceWithTransferKey(
40+
t.Context(),
41+
func(
42+
source TransferPath,
43+
transferKey TransferKey,
44+
provideMode protocol.ProvideMode,
45+
recoveryMode receiveRecoveryMode,
46+
ipPath *IpPath,
47+
packet []byte,
48+
) {
49+
},
50+
TransferPath{},
51+
TransferKey{},
52+
protocol.ProvideMode_Public,
53+
4,
54+
net.IPv4(10, 0, 0, 1).To4(),
55+
40000,
56+
net.IPv4(203, 0, 113, 7).To4(),
57+
443,
58+
0,
59+
settings,
60+
)
61+
defer sequence.Close()
62+
sequence.timestampValueForTest = func() uint32 { return 4000 }
63+
64+
const receiverMtu = 640
65+
const peerMss = receiverMtu - Ipv4HeaderSizeWithoutExtensions - TcpHeaderSizeWithoutExtensions
66+
synOptions := []byte{
67+
2, 4, 0, 0,
68+
1, 1, 8, 10,
69+
0, 0, 0, 1,
70+
0, 0, 0, 0,
71+
}
72+
binary.BigEndian.PutUint16(synOptions[2:4], peerMss)
73+
sequence.mutex.Lock()
74+
sequence.initializeSynWithLock(&parsedTcp{
75+
seq: 1000,
76+
windowSize: 65535,
77+
options: synOptions,
78+
})
79+
sequence.mutex.Unlock()
80+
if sequence.peerMss != peerMss {
81+
t.Fatalf("peer MSS=%d, want %d", sequence.peerMss, peerMss)
82+
}
83+
84+
payload := make([]byte, 8*1024)
85+
packets, err := sequence.DataPackets(payload, len(payload), settings.Mtu)
86+
if err != nil {
87+
t.Fatalf("packetize provider return: %v", err)
88+
}
89+
payloadByteCount := 0
90+
for packetIndex, packet := range packets {
91+
if receiverMtu < len(packet) {
92+
t.Errorf(
93+
"return packet %d is %d bytes, exceeds receiver MTU %d",
94+
packetIndex,
95+
len(packet),
96+
receiverMtu,
97+
)
98+
}
99+
_, _, _, transport, ok := parseIpv4(packet)
100+
if !ok {
101+
t.Fatalf("return packet %d is not IPv4", packetIndex)
102+
}
103+
tcp := &parsedTcp{}
104+
if !parseTcpPacket(sequence.destinationIp, sequence.sourceIp, transport, tcp) {
105+
t.Fatalf("return packet %d is not TCP", packetIndex)
106+
}
107+
payloadByteCount += len(tcp.payload)
108+
MessagePoolReturn(packet)
109+
}
110+
if payloadByteCount != len(payload) {
111+
t.Fatalf("packetized payload=%d bytes, want %d", payloadByteCount, len(payload))
112+
}
113+
}
114+
115+
// A peer MSS equal to the local fixed-header limit must not reduce the normal
116+
// packet size beyond the timestamp bytes the sender already has to include.
117+
func TestTcpSequencePeerMssPreservesEqualMtuPacketization(t *testing.T) {
118+
const localMtu = 1440
119+
const peerMss = localMtu - Ipv4HeaderSizeWithoutExtensions - TcpHeaderSizeWithoutExtensions
120+
state := newTcpOptionPacketizationState(peerMss)
121+
packetPayloadByteCount := peerMss - tcpTimestampOptionByteCount
122+
payload := make([]byte, packetPayloadByteCount)
123+
packets, err := state.DataPackets(payload, len(payload), localMtu)
124+
if err != nil {
125+
t.Fatalf("packetize equal-MTU return: %v", err)
126+
}
127+
if len(packets) != 1 || len(packets[0]) != localMtu {
128+
t.Fatalf(
129+
"equal-MTU packets=(%d, %d bytes), want (1, %d bytes)",
130+
len(packets),
131+
len(packets[0]),
132+
localMtu,
133+
)
134+
}
135+
MessagePoolReturn(packets[0])
136+
}
137+
138+
// Measures provider packetization and counts only packets a receiver with the
139+
// advertised MTU can accept.
140+
func benchmarkTcpReturnPacketization(b *testing.B, peerMss uint32, receiverMtu int) {
141+
const providerMtu = 1440
142+
const payloadByteCount = 64 * 1024
143+
state := newTcpOptionPacketizationState(peerMss)
144+
payload := make([]byte, payloadByteCount)
145+
acceptedPayloadByteCount := 0
146+
packetCount := 0
147+
droppedPacketCount := 0
148+
149+
b.ResetTimer()
150+
for range b.N {
151+
packets, err := state.DataPackets(payload, len(payload), providerMtu)
152+
if err != nil {
153+
b.Fatal(err)
154+
}
155+
for _, packet := range packets {
156+
packetCount += 1
157+
if len(packet) <= receiverMtu {
158+
_, _, _, transport, ok := parseIpv4(packet)
159+
if !ok {
160+
b.Fatal("generated packet is not IPv4")
161+
}
162+
tcp := &parsedTcp{}
163+
if !parseTcpPacket(state.destinationIp, state.sourceIp, transport, tcp) {
164+
b.Fatal("generated packet is not TCP")
165+
}
166+
acceptedPayloadByteCount += len(tcp.payload)
167+
} else {
168+
droppedPacketCount += 1
169+
}
170+
MessagePoolReturn(packet)
171+
}
172+
}
173+
b.StopTimer()
174+
if b.N == 0 {
175+
return
176+
}
177+
b.SetBytes(int64(acceptedPayloadByteCount / b.N))
178+
b.ReportMetric(float64(packetCount)/float64(b.N), "packets/op")
179+
b.ReportMetric(float64(droppedPacketCount)/float64(b.N), "oversized-drops/op")
180+
}
181+
182+
// Historical behavior ignores a smaller peer MSS.
183+
func BenchmarkTcpReturnPacketizationAsymmetricLocalMtu(b *testing.B) {
184+
benchmarkTcpReturnPacketization(b, 0, 640)
185+
}
186+
187+
// The candidate behavior honors a smaller peer MSS.
188+
func BenchmarkTcpReturnPacketizationAsymmetricPeerMss(b *testing.B) {
189+
benchmarkTcpReturnPacketization(b, 600, 640)
190+
}
191+
192+
// Historical behavior on the normal equal-MTU path.
193+
func BenchmarkTcpReturnPacketizationEqualLocalMtu(b *testing.B) {
194+
benchmarkTcpReturnPacketization(b, 0, 1440)
195+
}
196+
197+
// The candidate behavior on the normal equal-MTU path.
198+
func BenchmarkTcpReturnPacketizationEqualPeerMss(b *testing.B) {
199+
benchmarkTcpReturnPacketization(b, 1400, 1440)
200+
}

ip_tcp_reorder_test.go

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ type tcpReorderTestHarness struct {
2424
nextSeq uint32
2525
synAckReceived chan struct{}
2626
ackNumbers chan uint32
27+
acks chan parsedTcp
2728
reorderDecisions chan tcpReorderDisposition
2829
runDone chan struct{}
2930
closeOnce sync.Once
@@ -53,6 +54,25 @@ func newTcpReorderTestHarnessWithSetup(
5354
sequenceBufferSize int,
5455
reorderByteCount int,
5556
setupSequence func(*TcpSequence),
57+
) *tcpReorderTestHarness {
58+
return newTcpReorderTestHarnessWithSynOptions(
59+
t,
60+
initialSynSeq,
61+
sequenceBufferSize,
62+
reorderByteCount,
63+
setupSequence,
64+
nil,
65+
)
66+
}
67+
68+
// Establishes the same harness with explicit source SYN options.
69+
func newTcpReorderTestHarnessWithSynOptions(
70+
t *testing.T,
71+
initialSynSeq uint32,
72+
sequenceBufferSize int,
73+
reorderByteCount int,
74+
setupSequence func(*TcpSequence),
75+
synOptions []byte,
5676
) *tcpReorderTestHarness {
5777
t.Helper()
5878

@@ -82,6 +102,7 @@ func newTcpReorderTestHarnessWithSetup(
82102
nextSeq: initialSynSeq + 1,
83103
synAckReceived: make(chan struct{}, 1),
84104
ackNumbers: make(chan uint32, 64),
105+
acks: make(chan parsedTcp, 64),
85106
reorderDecisions: make(chan tcpReorderDisposition, 64),
86107
runDone: make(chan struct{}),
87108
}
@@ -111,6 +132,10 @@ func newTcpReorderTestHarnessWithSetup(
111132
return
112133
}
113134
if tcp.ack {
135+
select {
136+
case harness.acks <- *tcp:
137+
default:
138+
}
114139
select {
115140
case harness.ackNumbers <- tcp.ackNumber:
116141
default:
@@ -144,6 +169,7 @@ func newTcpReorderTestHarnessWithSetup(
144169
synItem := harness.newSendItem(initialSynSeq, nil, false)
145170
synItem.tcp.syn = true
146171
synItem.tcp.ack = false
172+
synItem.tcp.options = synOptions
147173
harness.sendItem(synItem)
148174
select {
149175
case <-harness.synAckReceived:

0 commit comments

Comments
 (0)