performance_test.go

package mipstack

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/netip"
	"testing"
	"time"
)

func benchmarkTCPControllerConnection(b *testing.B, algorithm CongestionControl) net.Conn {
	b.Helper()
	clientAddress := netip.MustParseAddr("192.0.2.201")
	serverAddress := netip.MustParseAddr("192.0.2.202")
	client, err := New(Config{
		LocalAddresses:    []netip.Prefix{netip.PrefixFrom(clientAddress, 32)},
		CongestionControl: algorithm,
	})
	if err != nil {
		b.Fatal(err)
	}
	server, err := New(Config{
		LocalAddresses:    []netip.Prefix{netip.PrefixFrom(serverAddress, 32)},
		CongestionControl: algorithm,
	})
	if err != nil {
		b.Fatal(err)
	}
	if err = client.Start(); err != nil {
		b.Fatal(err)
	}
	if err = server.Start(); err != nil {
		b.Fatal(err)
	}
	// Benchmarks use the same packet pump as tests and own its lifetime here.
	bridge := &stackBridge{client: client, peer: server, done: make(chan struct{}, 2)}
	go bridge.run(client, server, true)
	go bridge.run(server, client, false)
	listener, err := server.ListenTCP(context.Background(), "tcp4", netip.AddrPortFrom(serverAddress, 0))
	if err != nil {
		b.Fatal(err)
	}
	accepted := make(chan net.Conn, 1)
	go func() {
		connection, acceptErr := listener.Accept()
		if acceptErr != nil {
			accepted <- nil
			return
		}
		accepted <- connection
		_, _ = io.Copy(connection, connection)
		_ = connection.Close()
	}()
	connection, err := client.DialTCP(context.Background(), "tcp4", netip.AddrPort{}, netip.AddrPortFrom(serverAddress, listener.Addr().(*net.TCPAddr).AddrPort().Port()))
	if err != nil {
		b.Fatal(err)
	}
	serverConnection := <-accepted
	if serverConnection == nil {
		b.Fatal("benchmark accept failed")
	}
	b.Cleanup(func() {
		_ = connection.Close()
		_ = serverConnection.Close()
		_ = listener.Close()
		_ = client.Close()
		_ = server.Close()
		<-bridge.done
		<-bridge.done
	})
	return connection
}

func BenchmarkTCPControllerThroughput(b *testing.B) {
	const size = 4 * 1024 * 1024
	for _, algorithm := range []CongestionControl{CongestionControlReno, CongestionControlCUBIC, CongestionControlBBR} {
		b.Run(string(algorithm), func(b *testing.B) {
			connection := benchmarkTCPControllerConnection(b, algorithm)
			payload := bytes.Repeat([]byte{0x5a}, size)
			received := make([]byte, size)
			b.SetBytes(2 * size)
			b.ResetTimer()
			for iteration := 0; iteration < b.N; iteration++ {
				writeDone := make(chan error, 1)
				go func() {
					_, writeErr := connection.Write(payload)
					writeDone <- writeErr
				}()
				if _, err := io.ReadFull(connection, received); err != nil {
					b.Fatal(err)
				}
				if err := <-writeDone; err != nil {
					b.Fatal(err)
				}
				if !bytes.Equal(received, payload) {
					b.Fatal("echo payload mismatch")
				}
			}
		})
	}
}

func BenchmarkTCPControllerLatency(b *testing.B) {
	for _, algorithm := range []CongestionControl{CongestionControlReno, CongestionControlCUBIC, CongestionControlBBR} {
		b.Run(string(algorithm), func(b *testing.B) {
			connection := benchmarkTCPControllerConnection(b, algorithm)
			_ = connection.SetDeadline(time.Now().Add(time.Minute))
			request := []byte{0x5a}
			response := make([]byte, 1)
			b.ResetTimer()
			for iteration := 0; iteration < b.N; iteration++ {
				if _, err := connection.Write(request); err != nil {
					b.Fatal(err)
				}
				if _, err := io.ReadFull(connection, response); err != nil {
					b.Fatal(err)
				}
			}
		})
	}
}

func benchmarkTCPControllerConnections(b *testing.B, algorithm CongestionControl, count int) []net.Conn {
	b.Helper()
	clientAddress := netip.MustParseAddr("192.0.2.211")
	serverAddress := netip.MustParseAddr("192.0.2.212")
	client, err := New(Config{LocalAddresses: []netip.Prefix{netip.PrefixFrom(clientAddress, 32)}, CongestionControl: algorithm})
	if err != nil {
		b.Fatal(err)
	}
	server, err := New(Config{LocalAddresses: []netip.Prefix{netip.PrefixFrom(serverAddress, 32)}, CongestionControl: algorithm})
	if err != nil {
		b.Fatal(err)
	}
	if err = client.Start(); err != nil {
		b.Fatal(err)
	}
	if err = server.Start(); err != nil {
		b.Fatal(err)
	}
	bridge := &stackBridge{client: client, peer: server, done: make(chan struct{}, 2)}
	go bridge.run(client, server, true)
	go bridge.run(server, client, false)
	listener, err := server.ListenTCP(context.Background(), "tcp4", netip.AddrPortFrom(serverAddress, 0))
	if err != nil {
		b.Fatal(err)
	}
	accepted := make(chan net.Conn, count)
	acceptErrors := make(chan error, 1)
	go func() {
		for index := 0; index < count; index++ {
			connection, acceptErr := listener.Accept()
			if acceptErr != nil {
				acceptErrors <- acceptErr
				return
			}
			accepted <- connection
			go func(connection net.Conn) { _, _ = io.Copy(connection, connection) }(connection)
		}
		acceptErrors <- nil
	}()
	endpoint := netip.AddrPortFrom(serverAddress, listener.Addr().(*net.TCPAddr).AddrPort().Port())
	connections := make([]net.Conn, count)
	dialErrors := make(chan error, count)
	dialLimit := make(chan struct{}, 32)
	for index := range connections {
		go func(index int) {
			dialLimit <- struct{}{}
			connection, dialErr := client.DialTCP(context.Background(), "tcp4", netip.AddrPort{}, endpoint)
			<-dialLimit
			connections[index] = connection
			dialErrors <- dialErr
		}(index)
	}
	for range connections {
		if err = <-dialErrors; err != nil {
			b.Fatal(err)
		}
	}
	serverConnections := make([]net.Conn, 0, count)
	for len(serverConnections) < count {
		select {
		case connection := <-accepted:
			serverConnections = append(serverConnections, connection)
		case err = <-acceptErrors:
			if err != nil {
				b.Fatal(err)
			}
		}
	}
	b.Cleanup(func() {
		for _, connection := range connections {
			_ = connection.Close()
		}
		for _, connection := range serverConnections {
			_ = connection.Close()
		}
		_ = listener.Close()
		_ = client.Close()
		_ = server.Close()
		<-bridge.done
		<-bridge.done
	})
	return connections
}

func BenchmarkTCPControllerConcurrency(b *testing.B) {
	const size = 128 * 1024
	for _, algorithm := range []CongestionControl{CongestionControlReno, CongestionControlCUBIC, CongestionControlBBR} {
		for _, count := range []int{16, 64, 256, 512, 1024, 2048} {
			b.Run(fmt.Sprintf("%s-%d", algorithm, count), func(b *testing.B) {
				connections := benchmarkTCPControllerConnections(b, algorithm, count)
				payload := bytes.Repeat([]byte{0x6b}, size)
				b.SetBytes(int64(2 * size * count))
				b.ResetTimer()
				for iteration := 0; iteration < b.N; iteration++ {
					results := make(chan error, count)
					for _, connection := range connections {
						go func(connection net.Conn) {
							_ = connection.SetDeadline(time.Now().Add(2 * time.Minute))
							writeDone := make(chan error, 1)
							go func() {
								_, writeErr := connection.Write(payload)
								writeDone <- writeErr
							}()
							received := make([]byte, size)
							_, readErr := io.ReadFull(connection, received)
							if writeErr := <-writeDone; readErr == nil {
								readErr = writeErr
							}
							if readErr == nil && !bytes.Equal(received, payload) {
								readErr = errors.New("echo payload mismatch")
							}
							results <- readErr
						}(connection)
					}
					for range connections {
						if err := <-results; err != nil {
							b.Fatal(err)
						}
					}
				}
			})
		}
	}
}