diff --git a/internal/amneziawgnet/netstack.go b/internal/amneziawgnet/netstack.go index b05467f53..24f1937fc 100644 --- a/internal/amneziawgnet/netstack.go +++ b/internal/amneziawgnet/netstack.go @@ -8,6 +8,7 @@ import ( "os" "sync" "syscall" + "time" awgtun "github.com/amnezia-vpn/amneziawg-go/v3/tun" @@ -27,6 +28,10 @@ import ( // 1024 starved simultaneous TCP slow-starts; channel.Endpoint drops silently when full. const tunQueueDepth = 8192 +// tunQueueMaxWait bounds how long WriteNotify parks the goroutine that made the stack +// emit a packet; past it the packet is dropped, as channel.Endpoint drops when full. +const tunQueueMaxWait = 5 * time.Millisecond + // maxReceiveBuffer caps per-connection receive autotuning; gVisor's own 4 MB // cap held a single client upload near 310 Mbit/s at 50 ms RTT. const maxReceiveBuffer = 8 << 20 @@ -191,10 +196,21 @@ func (t *stackTun) WriteNotify() { view := pkt.ToView() pkt.DecRef() + select { + case t.incomingPacket <- view: + return + default: + } + // Echo replies and RSTs are emitted on amneziawg-go's per-peer receiver, which the + // reader of this queue itself waits on during a rebind: never park here for good. + timer := time.NewTimer(tunQueueMaxWait) + defer timer.Stop() // Select against done so racing dispatch abandons packet on close // without blocking Close or panicking on closed channel. select { case t.incomingPacket <- view: + case <-timer.C: + view.Release() case <-t.done: view.Release() } diff --git a/internal/amneziawgnet/netstack_test.go b/internal/amneziawgnet/netstack_test.go index f50ae42fc..bcda75d41 100644 --- a/internal/amneziawgnet/netstack_test.go +++ b/internal/amneziawgnet/netstack_test.go @@ -1,9 +1,14 @@ package amneziawgnet import ( + "net/netip" "testing" + "time" "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/checksum" + "gvisor.dev/gvisor/pkg/tcpip/header" "gvisor.dev/gvisor/pkg/tcpip/link/channel" ) @@ -128,3 +133,44 @@ func TestStackTunReadReturnsViewsToPool(t *testing.T) { t.Fatalf("Read allocates %v times per packet, want <=1: drained views are not being released", allocs) } } + +// TestStackTunWriteSurvivesFullOutboundQueue: Write runs on amneziawg-go's per-peer +// receiver, so a reply the stack cannot queue must be dropped rather than wait for Read. +func TestStackTunWriteSurvivesFullOutboundQueue(t *testing.T) { + tun, _, err := createNetTUNWithStack([]netip.Addr{netip.MustParseAddr("10.9.0.1")}, 1420) + if err != nil { + t.Fatalf("createNetTUNWithStack: %v", err) + } + defer tun.Close() + + ping := make([]byte, header.IPv4MinimumSize+header.ICMPv4MinimumSize) + ip := header.IPv4(ping) + ip.Encode(&header.IPv4Fields{ + TotalLength: uint16(len(ping)), + TTL: 64, + Protocol: uint8(header.ICMPv4ProtocolNumber), + SrcAddr: tcpip.AddrFrom4([4]byte{10, 9, 0, 2}), + DstAddr: tcpip.AddrFrom4([4]byte{10, 9, 0, 1}), + }) + ip.SetChecksum(^ip.CalculateChecksum()) + icmp := header.ICMPv4(ping[header.IPv4MinimumSize:]) + icmp.SetType(header.ICMPv4Echo) + icmp.SetChecksum(^checksum.Checksum(icmp, 0)) + + // Nobody calls Read, so the echo replies fill incomingPacket and overflow it. + done := make(chan struct{}) + go func() { + defer close(done) + for range tunQueueDepth + 8 { + _, _ = tun.Write([][]byte{ping}, 0) + } + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("Write is still waiting for room in the outbound queue: the receiver would deadlock a rebind") + } + if got := len(tun.(*stackTun).incomingPacket); got != tunQueueDepth { + t.Fatalf("incomingPacket holds %d replies, want %d: the queue never filled", got, tunQueueDepth) + } +}