mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-02 03:05:02 +03:00
fix(amneziawg): stop the TUN handoff from parking a peer's receiver
stackTun.WriteNotify runs on whichever goroutine made the stack emit a packet; for echo replies and RSTs that is amneziawg-go's per-peer receiver. It waited without bound for room in the outbound queue, whose reader waits on that same receiver while the device rebinds, so a peer that flooded the stack while nothing drained the queue deadlocked the next reconfiguration of the interface. The handoff now tries once, then waits at most 5 ms and drops the packet, as channel.Endpoint already does when its own queue is full.
This commit is contained in:
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user