app: refactor e2e TestTUN to reduce scheduler overhead on channels

Author: pymq

application_simnet_test.go

@@ -3,7 +3,6 @@ package awl
 import (
 	"fmt"
 	"os"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -40,15 +39,26 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 		bandwidthMbps int
 	}{
 		{
-			name:          "Fiber_300Mbps_1ms",
+			name:          "Fiber_200Mbps_1ms",
 			latency:       1 * time.Millisecond,
-			bandwidthMbps: 300 * Mbps,
+			bandwidthMbps: 200 * Mbps,
 		},
 		{
-			name:          "LongDistFiber_300Mbps_200ms",
+			name:          "LongDistFiber_200Mbps_200ms",
 			latency:       200 * time.Millisecond,
-			bandwidthMbps: 300 * Mbps,
+			bandwidthMbps: 200 * Mbps,
 		},
+		{
+			name:          "Fiber_100Mbps_1ms",
+			latency:       1 * time.Millisecond,
+			bandwidthMbps: 100 * Mbps,
+		},
+		{
+			name:          "LongDistFiber_100Mbps_200ms",
+			latency:       200 * time.Millisecond,
+			bandwidthMbps: 100 * Mbps,
+		},
+
 		{
 			name:          "Cable_10Mbps_1ms",
 			latency:       1 * time.Millisecond,
@@ -74,6 +84,7 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 			latency:       300 * time.Millisecond,
 			bandwidthMbps: 10 * Mbps,
 		},
+
 		{
 			name:          "Cable_50Mbps_1ms",
 			latency:       1 * time.Millisecond,
@@ -99,6 +110,7 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 			latency:       300 * time.Millisecond,
 			bandwidthMbps: 50 * Mbps,
 		},
+
 		{
 			name:          "DSL_20Mbps_25ms",
 			latency:       25 * time.Millisecond,
@@ -126,7 +138,7 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 			// TODO: try with different packet sizes
 			//  we probably should aim for one packet per UDP datagram
 			const packetSize = 3500 // Typical VPN packet size
-			const testDuration = 10 * time.Second
+			const testDuration = 30 * time.Second
 
 			// Setup link properties for the simulation
 			// We simulate a symmetric link between the two peers
@@ -136,6 +148,8 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 			}
 
 			// Create two peers
+			// uncomment to debug QUIC events
+			// t.Setenv("QLOGDIR", "./test-simnet")
 			ctx := t.Context()
 			extraLibp2pOpts := []libp2p.Option{
 				simlibp2p.QUICSimnet(net, linkSettings),
@@ -156,10 +170,14 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 			peer2.tun.ClearInboundCount()
 
 			// Send packets
-			var packetsSent int64
 			done := make(chan struct{})
 			startTime := time.Now()
 
+			packetsBatch := make([][]byte, TestTUNBatchSize*10)
+			for i := range packetsBatch {
+				packetsBatch[i] = packet
+			}
+
 			go func() {
 				defer close(done)
 				timer := time.NewTimer(testDuration)
@@ -171,22 +189,11 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 						return
 					case <-ctx.Done():
 						return
-					default:
+					case peer1.tun.Outbound <- packetsBatch:
 						// ok
 					}
 
-					// Send with "batches" to minimize runtime overhead for select
-					for range 10 {
-						peer1.tun.Outbound <- packet
-						atomic.AddInt64(&packetsSent, 1)
-					}
-
-					// TODO:
-					// Slight throttle to keep drops lower
-					// const sleepEvery = 200
-					// if newCount != 0 && newCount%sleepEvery == 0 {
-					//	time.Sleep(1 * time.Millisecond)
-					// }
+					// TODO: add ratelimit for TestTUN to remove busyloop
 				}
 			}()
 
@@ -199,7 +206,7 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 
 			// Collect metrics
 			received := peer2.tun.InboundCount()
-			sent := atomic.LoadInt64(&packetsSent)
+			sent := peer1.tun.OutboundCount()
 
 			packetLoss := (float64(1) - float64(received)/float64(sent)) * 100
 
@@ -219,6 +226,9 @@ func TestSimulatedTunnelPerformance(t *testing.T) {
 				// TODO: calculate p50/p95/p99 latency, jitter
 			})
 		})
+
+		// cool down a bit
+		time.Sleep(time.Second)
 	}
 
 	table.Render()

application_test.go

@@ -10,7 +10,6 @@ import (
 	"runtime"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -583,7 +582,7 @@ func TestUpdatePeerSettingsIPAddr(t *testing.T) {
 		// Send packets from peer1 to peer2
 		packet := testPacketWithDest(packetSize, newIP)
 		for i := 0; i < packetsCount; i++ {
-			peer1.tun.Outbound <- packet
+			peer1.tun.Outbound <- [][]byte{packet}
 		}
 
 		// Wait for packet processing
@@ -681,14 +680,15 @@ func TestTunnelPackets(t *testing.T) {
 	sendPackets := func(peer, peerWithInbound TestPeer) {
 		defer wg.Done()
 		packet := testPacket(packetSize)
+		packetsBatch := make([][]byte, TestTUNBatchSize)
+		for i := range packetsBatch {
+			packetsBatch[i] = packet
+		}
 
-		for i := 0; i < packetsCount; i++ {
-			peer.tun.Outbound <- packet
-			// to don't have packets loss
-			inbound := peerWithInbound.tun.InboundCount()
-			if (int64(i) - inbound) >= 50 {
-				time.Sleep(50 * time.Millisecond)
-			}
+		for i := 0; i < packetsCount/TestTUNBatchSize; i++ {
+			peer.tun.Outbound <- packetsBatch
+			// to avoid packet loss
+			time.Sleep(100 * time.Millisecond)
 		}
 	}
 
@@ -697,7 +697,7 @@ func TestTunnelPackets(t *testing.T) {
 	go sendPackets(peer2, peer1)
 	wg.Wait()
 
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * time.Second)
 	received1 := peer1.tun.InboundCount()
 	received2 := peer2.tun.InboundCount()
 	ts.EqualValues(packetsCount, received1)
@@ -717,21 +717,24 @@ func BenchmarkTunnelPackets(b *testing.B) {
 			b.ResetTimer()
 
 			b.SetBytes(int64(packetSize))
-			var packetsSent int64
 			packet := testPacket(packetSize)
 			peer2.tun.ReferenceInboundPacketLen = len(packet)
 			peer2.tun.ClearInboundCount()
+			packetsBatch := make([][]byte, TestTUNBatchSize*10)
+			for i := range packetsBatch {
+				packetsBatch[i] = packet
+			}
+
 			for i := 0; i < b.N; i++ {
-				peer1.tun.Outbound <- packet
-				atomic.AddInt64(&packetsSent, 1)
+				peer1.tun.Outbound <- packetsBatch
 				// to have packet_loss at reasonable level (but more than 0)
 				const sleepEvery = 100
 				if i != 0 && i%sleepEvery == 0 {
 					time.Sleep(1 * time.Millisecond)
 				}
 			}
 			received := peer2.tun.InboundCount()
-			sent := atomic.LoadInt64(&packetsSent)
+			sent := peer1.tun.OutboundCount()
 			packetLoss := (float64(1) - float64(received)/float64(sent)) * 100
 			bandwidth := float64(received) * float64(packetSize) / 1024 / 1024
 			b.ReportMetric(bandwidth, "MB/s")

test_suite_test.go

@@ -35,6 +35,8 @@ import (
 	"github.com/anywherelan/awl/vpn"
 )
 
+const TestTUNBatchSize = 100
+
 func init() {
 	// TODO: move to config
 	useAwldns = false
@@ -116,6 +118,8 @@ func (ts *TestSuite) newTestPeer(disableLogging bool, listenAddrs []multiaddr.Mu
 	app.Conf.HttpListenOnAdminHost = false
 	app.Conf.SetListenAddresses(listenAddrs)
 	app.Conf.P2pNode.BootstrapPeers = ts.bootstrapAddrsStr
+	app.Conf.P2pNode.ParallelSendingStreamsCount = 1
+	app.Conf.P2pNode.UseDedicatedConnForEachStream = false
 	if ts.isSimnet {
 		app.Conf.SOCKS5 = config.SOCKS5Config{
 			ListenerEnabled: false,
@@ -256,28 +260,36 @@ func (ts *TestSuite) sendAndAcceptFriendRequest(peer1, peer2 TestPeer) {
 }
 
 type TestTUN struct {
-	Outbound                  chan []byte
+	Outbound                  chan [][]byte
+	outboundBuf               [][]byte
 	ReferenceInboundPacketLen int
 
-	inboundCount int64
-	closed       chan struct{}
-	events       chan tun.Event
-	tun          testTun
+	inboundCount  int64
+	outboundCount int64
+	isClosed      atomic.Bool
+	events        chan tun.Event
+	tun           *testTun
 }
 
 func NewTestTUN() *TestTUN {
 	c := &TestTUN{
-		Outbound: make(chan []byte),
-		closed:   make(chan struct{}),
+		Outbound: make(chan [][]byte),
 		events:   make(chan tun.Event, 1),
+		tun:      &testTun{},
 	}
-	c.tun.t = c
 	c.events <- tun.EventUp
+
+	c.tun.t = c
+
 	return c
 }
 
 func (c *TestTUN) TUN() tun.Device {
-	return &c.tun
+	return c.tun
+}
+
+func (c *TestTUN) OutboundCount() int64 {
+	return atomic.LoadInt64(&c.outboundCount)
 }
 
 func (c *TestTUN) InboundCount() int64 {
@@ -295,31 +307,37 @@ type testTun struct {
 func (t *testTun) File() *os.File { return nil }
 
 func (t *testTun) Read(bufs [][]byte, sizes []int, offset int) (n int, err error) {
-	for i, buf := range bufs {
-		select {
-		case <-t.t.closed:
-			return n, os.ErrClosed
-		case msg := <-t.t.Outbound:
-			copyN := copy(buf[offset:], msg)
-			sizes[i] = copyN
-			n++
-		}
+	if t.t.isClosed.Load() {
+		return 0, os.ErrClosed
+	}
+	if len(t.t.outboundBuf) == 0 {
+		t.t.outboundBuf = <-t.t.Outbound
 	}
 
+	for i := range min(len(bufs), len(t.t.outboundBuf)) {
+		outboundMsg := t.t.outboundBuf[i]
+		copyN := copy(bufs[i][offset:], outboundMsg)
+		sizes[i] = copyN
+		n++
+	}
+
+	t.t.outboundBuf = t.t.outboundBuf[n:]
+	atomic.AddInt64(&t.t.outboundCount, int64(n))
+
 	return n, nil
 }
 
 func (t *testTun) Write(bufs [][]byte, offset int) (n int, err error) {
+	if t.t.isClosed.Load() {
+		return 0, os.ErrClosed
+	}
+
 	for _, buf := range bufs {
 		msg := buf[offset:]
 		if len(msg) != t.t.ReferenceInboundPacketLen {
 			return n, errors.New("packets length mismatch")
 		}
-		select {
-		case <-t.t.closed:
-			return n, os.ErrClosed
-		default:
-		}
+
 		atomic.AddInt64(&t.t.inboundCount, 1)
 		n++
 	}
@@ -328,15 +346,15 @@ func (t *testTun) Write(bufs [][]byte, offset int) (n int, err error) {
 }
 
 func (t *testTun) BatchSize() int {
-	return 1
+	return TestTUNBatchSize
 }
 
 func (t *testTun) Flush() error             { return nil }
 func (t *testTun) MTU() (int, error)        { return vpn.InterfaceMTU, nil }
 func (t *testTun) Name() (string, error)    { return "testTun", nil }
 func (t *testTun) Events() <-chan tun.Event { return t.t.events }
 func (t *testTun) Close() error {
-	close(t.t.closed)
+	t.t.isClosed.Store(true)
 	close(t.t.events)
 	return nil
 }