Skip to content

Commit 1682809

Browse files
pymqpymq
committed
app: add socks5 proxy benchmark with simulated network with different bandwidth/latency
1 parent 6d19268 commit 1682809

2 files changed

Lines changed: 287 additions & 36 deletions

File tree

‎application_simnet_test.go‎

Lines changed: 232 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,19 @@
11
package awl
22

33
import (
4+
"bytes"
45
"fmt"
6+
"io"
7+
"net"
58
"os"
69
"testing"
710
"time"
811

9-
"github.com/libp2p/go-libp2p"
10-
simlibp2p "github.com/libp2p/go-libp2p/x/simlibp2p"
11-
"github.com/marcopolo/simnet"
12-
"github.com/multiformats/go-multiaddr"
1312
"github.com/olekukonko/tablewriter"
13+
"golang.org/x/net/proxy"
14+
15+
"github.com/anywherelan/awl/entity"
16+
"github.com/anywherelan/awl/vpn"
1417
)
1518

1619
/*
@@ -123,47 +126,23 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
123126
},
124127
}
125128

129+
// TODO: try with different packet sizes
130+
// we probably should aim for one packet per UDP datagram
131+
const packetSize = vpn.InterfaceMTU
132+
const testDuration = 30 * time.Second
133+
126134
table := tablewriter.NewWriter(os.Stdout)
127135
table.SetHeader([]string{"Scenario", "Latency", "Bandwidth Limit", "Actual Throughput", "Utilization", "Packet Loss"})
128136

129137
for _, sc := range scenarios {
130138
t.Run(sc.name, func(t *testing.T) {
131139
ts := NewSimnetTestSuite(t)
132140

133-
net := &simnet.Simnet{}
134-
net.LatencyFunc = simnet.StaticLatency(sc.latency)
135-
net.Start()
136-
defer net.Close()
137-
138-
// TODO: try with different packet sizes
139-
// we probably should aim for one packet per UDP datagram
140-
const packetSize = 3500 // Typical VPN packet size
141-
const testDuration = 30 * time.Second
142-
143-
// Setup link properties for the simulation
144-
// We simulate a symmetric link between the two peers
145-
linkSettings := simnet.NodeBiDiLinkSettings{
146-
Downlink: simnet.LinkSettings{BitsPerSecond: sc.bandwidthMbps},
147-
Uplink: simnet.LinkSettings{BitsPerSecond: sc.bandwidthMbps},
148-
}
149-
150-
// Create two peers
141+
// Create two peers connected over simulated network
151142
// uncomment to debug QUIC events
152143
// t.Setenv("QLOGDIR", "./test-simnet")
153144
ctx := t.Context()
154-
extraLibp2pOpts := []libp2p.Option{
155-
simlibp2p.QUICSimnet(net, linkSettings),
156-
}
157-
158-
listenAddrs1 := []multiaddr.Multiaddr{
159-
multiaddr.StringCast("/ip4/1.2.3.1/udp/1234/quic-v1"),
160-
}
161-
peer1 := ts.newTestPeer(true, listenAddrs1, extraLibp2pOpts)
162-
listenAddrs2 := []multiaddr.Multiaddr{
163-
multiaddr.StringCast("/ip4/1.2.3.2/udp/1234/quic-v1"),
164-
}
165-
peer2 := ts.newTestPeer(true, listenAddrs2, extraLibp2pOpts)
166-
ts.makeFriendsSimnet(peer1, peer2)
145+
peer1, peer2 := ts.NewSimnetPeerPair(sc.latency, sc.bandwidthMbps, nil, nil)
167146

168147
packet := testPacket(packetSize)
169148
peer2.tun.ReferenceInboundPacketLen = packetSize
@@ -233,3 +212,221 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
233212

234213
table.Render()
235214
}
215+
216+
/*
217+
TestSimulatedSOCKS5ProxyPerformance benchmarks SOCKS5 proxy performance under simulated network conditions.
218+
219+
It uses:
220+
1. simlibp2p for simulated QUIC transport between peers
221+
2. simnet for configurable network latency and bandwidth
222+
3. Real SOCKS5 client/server for actual proxy connections
223+
4. Real HTTP client/server for end-to-end measurements
224+
225+
The data flow is:
226+
HTTP Client → SOCKS5 Listener (real TCP) → p2p Stream (simnet QUIC) → SOCKS5 Server → HTTP Server (real TCP)
227+
*/
228+
func TestSimulatedSOCKS5ProxyPerformance(t *testing.T) {
229+
// TODO: we have test for socks5 client receiving. Add test for socks5 client sending
230+
231+
const Mbps = 1_000_000
232+
233+
if os.Getenv("CI") != "" {
234+
t.Skip("skip in CI because it's a benchmark")
235+
}
236+
237+
scenarios := []struct {
238+
name string
239+
latency time.Duration
240+
bandwidthMbps int
241+
}{
242+
{
243+
name: "WARM-UP",
244+
latency: 10 * time.Millisecond,
245+
bandwidthMbps: 50 * Mbps,
246+
},
247+
{
248+
name: "Fiber_100Mbps_1ms",
249+
latency: 1 * time.Millisecond,
250+
bandwidthMbps: 100 * Mbps,
251+
},
252+
{
253+
name: "LongDistFiber_100Mbps_200ms",
254+
latency: 200 * time.Millisecond,
255+
bandwidthMbps: 100 * Mbps,
256+
},
257+
258+
{
259+
name: "Cable_50Mbps_10ms",
260+
latency: 10 * time.Millisecond,
261+
bandwidthMbps: 50 * Mbps,
262+
},
263+
{
264+
name: "LongDistCable_50Mbps_300ms",
265+
latency: 300 * time.Millisecond,
266+
bandwidthMbps: 50 * Mbps,
267+
},
268+
269+
{
270+
name: "Cable_10Mbps_100ms",
271+
latency: 100 * time.Millisecond,
272+
bandwidthMbps: 10 * Mbps,
273+
},
274+
{
275+
name: "LongDistCable_10Mbps_300ms",
276+
latency: 300 * time.Millisecond,
277+
bandwidthMbps: 10 * Mbps,
278+
},
279+
}
280+
281+
const testDuration = 20 * time.Second
282+
const testLastDuration = 5 * time.Second
283+
284+
table := tablewriter.NewWriter(os.Stdout)
285+
table.SetHeader([]string{"Scenario", "Latency", "Bandwidth Limit", "Throughput\navg", "Throughput\nlast 5 sec", "Utilization", "TTFB"})
286+
287+
for _, sc := range scenarios {
288+
t.Run(sc.name, func(t *testing.T) {
289+
ts := NewSimnetTestSuite(t)
290+
291+
ctx := t.Context()
292+
293+
// Create two peers connected over simulated network
294+
// peer1: SOCKS5 client side (listener enabled)
295+
// peer2: SOCKS5 server side (proxying enabled)
296+
peer1, peer2 := ts.NewSimnetPeerPair(sc.latency, sc.bandwidthMbps,
297+
&SOCKS5PeerConfig{ListenerEnabled: true, ProxyingEnabled: false},
298+
&SOCKS5PeerConfig{ListenerEnabled: false, ProxyingEnabled: true},
299+
)
300+
301+
// Configure peer2 to allow peer1 to use as exit node
302+
peer1Config, err := peer2.api.KnownPeerConfig(peer1.PeerID())
303+
ts.NoError(err)
304+
305+
err = peer2.api.UpdatePeerSettings(entity.UpdatePeerSettingsRequest{
306+
PeerID: peer1.PeerID(),
307+
Alias: peer1Config.Alias,
308+
DomainName: peer1Config.DomainName,
309+
IPAddr: peer1Config.IPAddr,
310+
AllowUsingAsExitNode: true,
311+
})
312+
ts.NoError(err)
313+
314+
// Wait for status exchange to propagate AllowedUsingAsExitNode to peer1
315+
ts.Eventually(func() bool {
316+
peer2Config, err := peer1.api.KnownPeerConfig(peer2.PeerID())
317+
ts.NoError(err)
318+
return peer2Config.AllowedUsingAsExitNode
319+
}, 2*time.Second, 100*time.Millisecond)
320+
321+
// Set peer2 as proxy for peer1
322+
peer1.app.SOCKS5.SetProxyPeerID(peer2.PeerID())
323+
peer2.app.SOCKS5.SetProxyingLocalhostEnabled(true)
324+
325+
// Setup raw TCP server that sends unlimited data
326+
tcpAddr := startUnlimitedTCPServer(t)
327+
328+
dialer, err := proxy.SOCKS5("tcp", peer1.app.Conf.SOCKS5.ListenAddress, nil, nil)
329+
ts.NoError(err)
330+
331+
// Measure TTFB and throughput
332+
connectStart := time.Now()
333+
334+
// Connect through SOCKS5 proxy to TCP server
335+
conn, err := dialer.Dial("tcp", tcpAddr)
336+
ts.NoError(err)
337+
defer conn.Close()
338+
339+
// Read first byte to measure TTFB
340+
firstByte := make([]byte, 1)
341+
_, err = io.ReadFull(conn, firstByte)
342+
ts.NoError(err)
343+
ttfb := time.Since(connectStart)
344+
345+
// Measure throughput for testDuration
346+
const bufSize = 1 << 20
347+
buf := make([]byte, bufSize)
348+
349+
totalBytes := int64(1)
350+
startTime := time.Now()
351+
352+
startTimeLastSeconds := time.Time{}
353+
bytesLastSeconds := int64(0)
354+
355+
for time.Since(startTime) < testDuration || ctx.Err() != nil {
356+
if startTimeLastSeconds.IsZero() && time.Since(startTime) > testDuration-testLastDuration {
357+
startTimeLastSeconds = time.Now()
358+
}
359+
360+
n, err := conn.Read(buf)
361+
totalBytes += int64(n)
362+
if !startTimeLastSeconds.IsZero() {
363+
bytesLastSeconds += int64(n)
364+
}
365+
if err != nil {
366+
t.Errorf("Read error after %d bytes: %v", totalBytes, err)
367+
break
368+
}
369+
}
370+
371+
duration := time.Since(startTime)
372+
durationLast := time.Since(startTimeLastSeconds)
373+
374+
// Calculate metrics
375+
throughputMbps := (float64(totalBytes) * 8) / duration.Seconds() / float64(Mbps)
376+
throughputMbpsLastSeconds := (float64(bytesLastSeconds) * 8) / durationLast.Seconds() / float64(Mbps)
377+
expectedMbps := float64(sc.bandwidthMbps) / float64(Mbps)
378+
utilization := (throughputMbps / expectedMbps) * 100
379+
380+
table.Append([]string{
381+
sc.name,
382+
sc.latency.String(),
383+
fmt.Sprintf("%d Mbps", sc.bandwidthMbps/Mbps),
384+
fmt.Sprintf("%.2f Mbps", throughputMbps),
385+
fmt.Sprintf("%.2f Mbps", throughputMbpsLastSeconds),
386+
fmt.Sprintf("%.2f %%", utilization),
387+
ttfb.Round(100 * time.Microsecond).String(),
388+
})
389+
})
390+
391+
// Cool down between tests
392+
time.Sleep(time.Second)
393+
}
394+
395+
table.Render()
396+
}
397+
398+
// startUnlimitedTCPServer starts a TCP server that sends unlimited data to any client.
399+
// Returns the server address. Server is automatically closed when test ends.
400+
func startUnlimitedTCPServer(t *testing.T) string {
401+
listener, err := net.Listen("tcp", "127.0.0.1:0")
402+
if err != nil {
403+
t.Fatalf("Failed to start TCP server: %v", err)
404+
}
405+
406+
t.Cleanup(func() {
407+
listener.Close()
408+
})
409+
410+
const chunkSize = 1 << 20 // 1 MB
411+
chunk := bytes.Repeat([]byte("X"), chunkSize)
412+
413+
go func() {
414+
for {
415+
conn, err := listener.Accept()
416+
if err != nil {
417+
return // Listener closed
418+
}
419+
go func(c net.Conn) {
420+
defer c.Close()
421+
for {
422+
_, err2 := c.Write(chunk)
423+
if err2 != nil {
424+
return
425+
}
426+
}
427+
}(conn)
428+
}
429+
}()
430+
431+
return listener.Addr().String()
432+
}

‎test_suite_test.go‎

Lines changed: 55 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ import (
2323
"github.com/libp2p/go-libp2p/p2p/host/eventbus"
2424
"github.com/libp2p/go-libp2p/p2p/host/peerstore/pstoremem"
2525
rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
26+
simlibp2p "github.com/libp2p/go-libp2p/x/simlibp2p"
27+
"github.com/marcopolo/simnet"
2628
"github.com/multiformats/go-multiaddr"
2729
"github.com/stretchr/testify/require"
2830
"go.uber.org/zap/zapcore"
@@ -94,7 +96,17 @@ func (ts *TestSuite) NewTestPeer(disableLogging bool) TestPeer {
9496
return ts.newTestPeer(disableLogging, listenAddrs, nil)
9597
}
9698

99+
// SOCKS5PeerConfig configures SOCKS5 settings for test peers
100+
type SOCKS5PeerConfig struct {
101+
ListenerEnabled bool
102+
ProxyingEnabled bool
103+
}
104+
97105
func (ts *TestSuite) newTestPeer(disableLogging bool, listenAddrs []multiaddr.Multiaddr, extraLibp2pOpts []libp2p.Option) TestPeer {
106+
return ts.newTestPeerWithSOCKS5(disableLogging, listenAddrs, extraLibp2pOpts, nil)
107+
}
108+
109+
func (ts *TestSuite) newTestPeerWithSOCKS5(disableLogging bool, listenAddrs []multiaddr.Multiaddr, extraLibp2pOpts []libp2p.Option, socks5Conf *SOCKS5PeerConfig) TestPeer {
98110
tempDir := ts.t.TempDir()
99111
ts.t.Setenv(config.AppDataDirEnvKey, tempDir)
100112
tempConf := config.NewConfig(eventbus.NewBus())
@@ -120,7 +132,16 @@ func (ts *TestSuite) newTestPeer(disableLogging bool, listenAddrs []multiaddr.Mu
120132
app.Conf.P2pNode.BootstrapPeers = ts.bootstrapAddrsStr
121133
app.Conf.P2pNode.ParallelSendingStreamsCount = 1
122134
app.Conf.P2pNode.UseDedicatedConnForEachStream = false
123-
if ts.isSimnet {
135+
136+
// Configure SOCKS5 based on provided config or defaults
137+
if socks5Conf != nil {
138+
app.Conf.SOCKS5 = config.SOCKS5Config{
139+
ListenerEnabled: socks5Conf.ListenerEnabled,
140+
ProxyingEnabled: socks5Conf.ProxyingEnabled,
141+
ListenAddress: pickFreeAddr(ts.t),
142+
UsingPeerID: "",
143+
}
144+
} else if ts.isSimnet {
124145
app.Conf.SOCKS5 = config.SOCKS5Config{
125146
ListenerEnabled: false,
126147
ProxyingEnabled: false,
@@ -235,6 +256,39 @@ func (ts *TestSuite) makeFriendsSimnet(peer1, peer2 TestPeer) {
235256
ts.sendAndAcceptFriendRequest(peer1, peer2)
236257
}
237258

259+
// NewSimnetPeerPair creates two peers connected over a simulated network with configurable
260+
// latency and bandwidth. Both peers are made friends automatically.
261+
func (ts *TestSuite) NewSimnetPeerPair(latency time.Duration, bandwidthBps int, socks5Conf1, socks5Conf2 *SOCKS5PeerConfig) (TestPeer, TestPeer) {
262+
simNet := &simnet.Simnet{}
263+
simNet.LatencyFunc = simnet.StaticLatency(latency)
264+
simNet.Start()
265+
266+
ts.t.Cleanup(simNet.Close)
267+
268+
linkSettings := simnet.NodeBiDiLinkSettings{
269+
Downlink: simnet.LinkSettings{BitsPerSecond: bandwidthBps},
270+
Uplink: simnet.LinkSettings{BitsPerSecond: bandwidthBps},
271+
}
272+
273+
extraLibp2pOpts := []libp2p.Option{
274+
simlibp2p.QUICSimnet(simNet, linkSettings),
275+
}
276+
277+
listenAddrs1 := []multiaddr.Multiaddr{
278+
multiaddr.StringCast("/ip4/1.2.3.1/udp/1234/quic-v1"),
279+
}
280+
peer1 := ts.newTestPeerWithSOCKS5(true, listenAddrs1, extraLibp2pOpts, socks5Conf1)
281+
282+
listenAddrs2 := []multiaddr.Multiaddr{
283+
multiaddr.StringCast("/ip4/1.2.3.2/udp/1234/quic-v1"),
284+
}
285+
peer2 := ts.newTestPeerWithSOCKS5(true, listenAddrs2, extraLibp2pOpts, socks5Conf2)
286+
287+
ts.makeFriendsSimnet(peer1, peer2)
288+
289+
return peer1, peer2
290+
}
291+
238292
func (ts *TestSuite) sendAndAcceptFriendRequest(peer1, peer2 TestPeer) {
239293
err := peer1.api.SendFriendRequest(peer2.PeerID(), "peer_2")
240294
ts.NoError(err)

0 commit comments

Comments
 (0)