diff --git a/internal/amneziawgnet/netstack.go b/internal/amneziawgnet/netstack.go index 6dd0f3ace..af1c17d00 100644 --- a/internal/amneziawgnet/netstack.go +++ b/internal/amneziawgnet/netstack.go @@ -132,6 +132,21 @@ func (t *stackTun) Events() <-chan awgtun.Event { return t.events } func (t *stackTun) MTU() (int, error) { return t.mtu, nil } func (t *stackTun) BatchSize() int { return 1 } +// Read blocks for the first packet, then opportunistically drains any more +// that are already buffered (non-blocking), up to len(buf). amneziawg-go's +// caller (RoutineReadFromTUN) sizes buf/sizes to device.BatchSize(), which +// is the UDP bind's own batch size (128 on Linux, see conn.IdealBatchSize) +// since that's larger than BatchSize()'s 1 below -- so real buffer capacity +// for a batch is already there. Without this drain loop, Read always +// returned exactly one packet no matter how many buf could hold, so every +// downstream step (peer lookup, per-peer staging, and ultimately the UDP +// bind's own genuinely batched Send/sendmmsg) processed the download +// direction one packet at a time while the upload direction's equivalent +// (bind.Receive/recvmmsg -> decrypt -> stackTun.Write, which already loops +// over its whole buf) processed up to 128 per cycle. That asymmetry is +// real, not gVisor/amneziawg-go's -- both the receive and send paths on the +// UDP bind support batching identically, only this Read implementation +// didn't use it. func (t *stackTun) Read(buf [][]byte, sizes []int, offset int) (int, error) { view, ok := <-t.incomingPacket if !ok { @@ -142,7 +157,24 @@ func (t *stackTun) Read(buf [][]byte, sizes []int, offset int) (int, error) { return 0, err } sizes[0] = n - return 1, nil + count := 1 + for count < len(buf) { + select { + case view, ok := <-t.incomingPacket: + if !ok { + return count, nil + } + n, err := view.Read(buf[count][offset:]) + if err != nil { + return count, nil + } + sizes[count] = n + count++ + default: + return count, nil + } + } + return count, nil } func (t *stackTun) Write(buf [][]byte, offset int) (int, error) { diff --git a/internal/amneziawgnet/netstack_test.go b/internal/amneziawgnet/netstack_test.go new file mode 100644 index 000000000..52fda2d18 --- /dev/null +++ b/internal/amneziawgnet/netstack_test.go @@ -0,0 +1,88 @@ +package amneziawgnet + +import ( + "testing" + + "gvisor.dev/gvisor/pkg/buffer" +) + +// TestStackTunReadDrainsBufferedBatch is a regression test for a real +// throughput bug: Read used to always return exactly one packet per call +// no matter how many were already queued, forcing amneziawg-go's TUN +// reader to pay a full peer-lookup+staging+syscall cycle per packet on the +// download path while the upload path (via the UDP bind's own +// recvmmsg/sendmmsg batching) amortized that cost across up to 128 +// packets. Confirmed live: this alone took real download throughput from +// 30-40 Mbit/s to 130-250 Mbit/s on a real test connection (see commit +// 6436fd9c's message and internal/amneziawgnet/netstack.go's own comment +// on tunQueueDepth for the full story) -- this test locks in the second, +// finer-grained fix on top of that: Read must actually drain what's +// already buffered instead of returning after the first packet. +func TestStackTunReadDrainsBufferedBatch(t *testing.T) { + t.Parallel() + + tun := &stackTun{incomingPacket: make(chan *buffer.View, tunQueueDepth)} + packets := [][]byte{{1, 2, 3}, {4, 5}, {6, 7, 8, 9}} + for _, p := range packets { + tun.incomingPacket <- buffer.NewViewWithData(p) + } + + buf := make([][]byte, 8) + sizes := make([]int, 8) + for i := range buf { + buf[i] = make([]byte, 64) + } + + n, err := tun.Read(buf, sizes, 0) + if err != nil { + t.Fatalf("Read: %v", err) + } + if n != len(packets) { + t.Fatalf("Read returned %d packets, want %d (all buffered packets in one call)", n, len(packets)) + } + for i, want := range packets { + got := buf[i][:sizes[i]] + if string(got) != string(want) { + t.Errorf("packet %d = %v, want %v", i, got, want) + } + } +} + +// TestStackTunReadStopsAtBufCapacity confirms Read never returns more +// packets than the caller's buf can hold, and that whatever didn't fit is +// still there (in order) for the next call -- draining must respect the +// caller's batch size, not just gulp everything queued. +func TestStackTunReadStopsAtBufCapacity(t *testing.T) { + t.Parallel() + + tun := &stackTun{incomingPacket: make(chan *buffer.View, tunQueueDepth)} + packets := [][]byte{{1}, {2}, {3}} + for _, p := range packets { + tun.incomingPacket <- buffer.NewViewWithData(p) + } + + buf := make([][]byte, 2) + sizes := make([]int, 2) + for i := range buf { + buf[i] = make([]byte, 64) + } + + n, err := tun.Read(buf, sizes, 0) + if err != nil { + t.Fatalf("first Read: %v", err) + } + if n != 2 { + t.Fatalf("first Read returned %d, want 2 (buf capacity)", n) + } + + n, err = tun.Read(buf, sizes, 0) + if err != nil { + t.Fatalf("second Read: %v", err) + } + if n != 1 { + t.Fatalf("second Read returned %d, want 1 (the leftover packet)", n) + } + if got := buf[0][:sizes[0]]; string(got) != "\x03" { + t.Errorf("leftover packet = %v, want [3]", got) + } +}