mirror of
https://github.com/tailscale/tailscale.git
synced 2026-11-09 05:33:27 +09:00
all: use ~128KiB packet buf for batched i/o
This commit bumps the wireguard-go dependency to incorporate changes to the packet memory model and the tun.Device.Read and conn.ReceiveFunc I/O interfaces. It updates their implementations accordingly. These changes improve throughput in all measured benchmarks and reduce peak RSS in six of eight cases. The two regressions will be addressed in a follow-up commit that reduces peak RSS below the baseline measured at1e69418. That work is kept separate to simplify review. The following throughput and peak RSS benchmarks were performed with iperf3 between two Intel i5-12400 nodes running Ubuntu 24.04 (Linux 6.8). The UDP benchmarks did not use UDP GSO on the sender, so they were roughly equivalent to single packet I/O through wireguard-go. TCP/1 signifies one TCP stream; TCP/128 signifies 128 parallel TCP streams. Throughput (Mb/s) Test1e69418After Change TCP/1 10,371 11,354 +9.5% TCP/128 7,886 8,404 +6.6% UDP/1 2,111 2,853 +35.1% UDP/128 1,747 2,235 +28.0% Peak memory (VmHWM, kB) Test Side1e69418After Change TCP/1 TX 98,240 52,596 -46.5% RX 287,748 73,384 -74.5% TCP/128 TX 101,196 52,812 -47.8% RX 290,420 63,620 -78.1% UDP/1 TX 58,864 160,840 +173.2% RX 137,516 49,900 -63.7% UDP/128 TX 66,148 116,096 +75.5% RX 154,384 56,556 -63.4% Updates tailscale/corp#46716 Updates tailscale/corp#22467 Updates tailscale/corp#36989 Updates tailscale/corp#37878 Signed-off-by: Jordan Whited <jordan@tailscale.com>
This commit is contained in:
Jordan Whited authored and Jordan Whited committed
1 parent
b62350fbec
commit
8fc6dca15e
32 files changed
+620
-611
No files matched your search
@@ -792,7 +792,7 @@ tailscale.com/cmd/k8s-operator dependencies: (generated by github.com/tailscale/
|
||||
tailscale.com/logtail/filch from tailscale.com/log/sockstatlog+
|
||||
tailscale.com/metrics from tailscale.com/tsweb+
|
||||
tailscale.com/net/bakedroots from tailscale.com/net/tlsdial+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock
|
||||
tailscale.com/net/dns from tailscale.com/ipn/ipnlocal+
|
||||
tailscale.com/net/dns/publicdns from tailscale.com/net/dns+
|
||||
tailscale.com/net/dns/resolvconffile from tailscale.com/cmd/k8s-operator+
|
||||
|
||||
@@ -84,7 +84,7 @@ tailscale.com/cmd/tailscaled dependencies: (generated by github.com/tailscale/de
|
||||
tailscale.com/logtail from tailscale.com/cmd/tailscaled+
|
||||
tailscale.com/logtail/filch from tailscale.com/log/sockstatlog+
|
||||
tailscale.com/net/bakedroots from tailscale.com/net/tlsdial
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock
|
||||
tailscale.com/net/dns from tailscale.com/cmd/tailscaled+
|
||||
tailscale.com/net/dns/publicdns from tailscale.com/net/dns+
|
||||
tailscale.com/net/dns/resolvconffile from tailscale.com/net/dns+
|
||||
|
||||
@@ -101,7 +101,7 @@ tailscale.com/cmd/tailscaled dependencies: (generated by github.com/tailscale/de
|
||||
tailscale.com/logtail/filch from tailscale.com/log/sockstatlog+
|
||||
tailscale.com/net/ace from tailscale.com/cmd/tailscale/cli
|
||||
tailscale.com/net/bakedroots from tailscale.com/net/tlsdial
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock
|
||||
tailscale.com/net/dns from tailscale.com/cmd/tailscaled+
|
||||
tailscale.com/net/dns/publicdns from tailscale.com/net/dns+
|
||||
tailscale.com/net/dns/resolvconffile from tailscale.com/net/dns+
|
||||
|
||||
@@ -181,7 +181,7 @@ tailscale.com/cmd/tsidp dependencies: (generated by github.com/tailscale/depawar
|
||||
tailscale.com/logtail/filch from tailscale.com/log/sockstatlog+
|
||||
tailscale.com/metrics from tailscale.com/tsweb+
|
||||
tailscale.com/net/bakedroots from tailscale.com/net/tlsdial+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock
|
||||
tailscale.com/net/dns from tailscale.com/ipn/ipnlocal+
|
||||
tailscale.com/net/dns/publicdns from tailscale.com/net/dns+
|
||||
tailscale.com/net/dns/resolvconffile from tailscale.com/net/dns+
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
wgtun "github.com/tailscale/wireguard-go/tun"
|
||||
"github.com/tailscale/wireguard-go/tun/tuntest"
|
||||
"go4.org/netipx"
|
||||
"tailscale.com/net/packet"
|
||||
@@ -414,17 +415,16 @@ func TestHandlePacketFromWireGuard(t *testing.T) {
|
||||
t.Errorf("unexpected packet dst: want %v, got %v", want, got)
|
||||
}
|
||||
if tt.expectedInjectedPkt != nil {
|
||||
var buf [tstun.MaxPacketSize]byte
|
||||
bufs := [][]byte{buf[:]}
|
||||
sizes := []int{0}
|
||||
n, err := tun.Read(bufs, sizes, 0)
|
||||
slab := make([]byte, (2*wgtun.ReadPacketSpacing)+(2*(1<<16-1)))
|
||||
packets := make([]wgtun.ReadPacket, 1)
|
||||
n, err := tun.Read(slab, packets)
|
||||
if err != nil {
|
||||
t.Errorf("error reading injected packet: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Errorf("expected to read 1 packet, got %d", n)
|
||||
}
|
||||
if want, got := tt.expectedInjectedPkt, buf[:sizes[0]]; !bytes.Equal(want, got) {
|
||||
if want, got := tt.expectedInjectedPkt, slab[packets[0].Offset:packets[0].Offset+packets[0].Size]; !bytes.Equal(want, got) {
|
||||
t.Errorf("unexpected contents of injected packet: want %+x, got %+x", want, got)
|
||||
|
||||
}
|
||||
|
||||
+12
-17
@@ -103,6 +103,7 @@ const (
|
||||
|
||||
// handleTAPFrame handles receiving a raw TAP ethernet frame and reports whether
|
||||
// it's been handled (that is, whether it should NOT be passed to wireguard).
|
||||
// handleTAPFrame returns [consumePacket] (true) if len(ethBuf) < [ethernetFrameSize].
|
||||
func (t *tapDevice) handleTAPFrame(ethBuf []byte) bool {
|
||||
|
||||
if len(ethBuf) < ethernetFrameSize {
|
||||
@@ -419,28 +420,22 @@ func (t *tapDevice) Name() (string, error) {
|
||||
return t.name, nil
|
||||
}
|
||||
|
||||
// Read reads an IP packet from the TAP device. It strips the ethernet frame header.
|
||||
func (t *tapDevice) Read(buffs [][]byte, sizes []int, offset int) (int, error) {
|
||||
n, err := t.ReadEthernet(buffs, sizes, offset)
|
||||
if err != nil || n == 0 {
|
||||
return n, err
|
||||
}
|
||||
// Strip the ethernet frame header.
|
||||
copy(buffs[0][offset:], buffs[0][offset+ethernetFrameSize:offset+sizes[0]])
|
||||
sizes[0] -= ethernetFrameSize
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
// ReadEthernet reads a raw ethernet frame from the TAP device.
|
||||
func (t *tapDevice) ReadEthernet(buffs [][]byte, sizes []int, offset int) (int, error) {
|
||||
n, err := t.file.Read(buffs[0][offset:])
|
||||
// Read implements [tun.Device.Read]. Read swallows frames that should not be
|
||||
// passed to wireguard-go, as evaluated by [tapDevice.handleTAPFrame]. Read
|
||||
// excludes the Ethernet header for returned IP packets described by packets[:n].
|
||||
func (t *tapDevice) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
|
||||
buf := slab[tun.ReadPacketSpacing : len(slab)-tun.ReadPacketSpacing]
|
||||
n, err := t.file.Read(buf)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if t.handleTAPFrame(buffs[0][offset : offset+n]) {
|
||||
if t.handleTAPFrame(buf[:n]) {
|
||||
return 0, nil
|
||||
}
|
||||
sizes[0] = n
|
||||
packets[0] = tun.ReadPacket{
|
||||
Offset: tun.ReadPacketSpacing + ethernetFrameSize,
|
||||
Size: n - ethernetFrameSize,
|
||||
}
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -34,14 +34,14 @@ type fakeDevice struct {
|
||||
name string
|
||||
}
|
||||
|
||||
func (f *fakeDevice) File() *os.File { return nil }
|
||||
func (f *fakeDevice) Read(bufs [][]byte, sizes []int, offset int) (n int, err error) { return 0, nil }
|
||||
func (f *fakeDevice) Write(bufs [][]byte, offset int) (int, error) { return 0, nil }
|
||||
func (f *fakeDevice) MTU() (int, error) { return 0, nil }
|
||||
func (f *fakeDevice) Name() (string, error) { return f.name, nil }
|
||||
func (f *fakeDevice) Events() <-chan tun.Event { return nil }
|
||||
func (f *fakeDevice) Close() error { return nil }
|
||||
func (f *fakeDevice) BatchSize() int { return 0 }
|
||||
func (f *fakeDevice) File() *os.File { return nil }
|
||||
func (f *fakeDevice) Read(slab []byte, packets []tun.ReadPacket) (n int, err error) { return 0, nil }
|
||||
func (f *fakeDevice) Write(bufs [][]byte, offset int) (int, error) { return 0, nil }
|
||||
func (f *fakeDevice) MTU() (int, error) { return 0, nil }
|
||||
func (f *fakeDevice) Name() (string, error) { return f.name, nil }
|
||||
func (f *fakeDevice) Events() <-chan tun.Event { return nil }
|
||||
func (f *fakeDevice) Close() error { return nil }
|
||||
func (f *fakeDevice) BatchSize() int { return 0 }
|
||||
|
||||
func Test_poller(t *testing.T) {
|
||||
getTXQDropsMetric().Set(0) // reset for test count > 1
|
||||
|
||||
@@ -173,4 +173,4 @@
|
||||
});
|
||||
};
|
||||
}
|
||||
# nix-direnv cache busting line: sha256-w7VuEoXsnL7wmHbKlwd6qgUqyRj09JAqHw073Zgnidc=
|
||||
# nix-direnv cache busting line: sha256-edmocquubPBqaakW8/TfQTLZS48HR7SOcWjYiLeJGLY=
|
||||
+2
-2
@@ -4,7 +4,7 @@
|
||||
"sri": "sha256-+cTlDLusJthxTayVx6yGoqZHRN4ylvs/9Qpy4P2dlDI="
|
||||
},
|
||||
"vendor": {
|
||||
"goModSum": "sha256-aYPi3BUkgTUZ3ESpK33W2CtHruPSgb7gMJlAP/tUL0I=",
|
||||
"sri": "sha256-w7VuEoXsnL7wmHbKlwd6qgUqyRj09JAqHw073Zgnidc="
|
||||
"goModSum": "sha256-PGvkYIOJdO2bTiy4Rsp+82idPQuvqqPW5XSFutoEaqQ=",
|
||||
"sri": "sha256-edmocquubPBqaakW8/TfQTLZS48HR7SOcWjYiLeJGLY="
|
||||
}
|
||||
}
|
||||
@@ -126,7 +126,7 @@ require (
|
||||
github.com/tailscale/ts-gokrazy v0.0.0-20260630224145-b83088f2e52e
|
||||
github.com/tailscale/web-client-prebuilt v0.0.0-20250124233751-d4cd19a26976
|
||||
github.com/tailscale/wf v0.0.0-20240214030419-6fbb0a674ee6
|
||||
github.com/tailscale/wireguard-go v0.0.0-20260904023712-e855235c55a2
|
||||
github.com/tailscale/wireguard-go v0.0.0-20260904160635-24b5b6917431
|
||||
github.com/tailscale/xnet v0.0.0-20240729143630-8497ac4dab2e
|
||||
github.com/tc-hib/winres v0.2.1
|
||||
github.com/tcnksm/go-httpstat v0.2.0
|
||||
|
||||
@@ -1217,8 +1217,8 @@ github.com/tailscale/web-client-prebuilt v0.0.0-20250124233751-d4cd19a26976 h1:U
|
||||
github.com/tailscale/web-client-prebuilt v0.0.0-20250124233751-d4cd19a26976/go.mod h1:agQPE6y6ldqCOui2gkIh7ZMztTkIQKH049tv8siLuNQ=
|
||||
github.com/tailscale/wf v0.0.0-20240214030419-6fbb0a674ee6 h1:l10Gi6w9jxvinoiq15g8OToDdASBni4CyJOdHY1Hr8M=
|
||||
github.com/tailscale/wf v0.0.0-20240214030419-6fbb0a674ee6/go.mod h1:ZXRML051h7o4OcI0d3AaILDIad/Xw0IkXaHM17dic1Y=
|
||||
github.com/tailscale/wireguard-go v0.0.0-20260904023712-e855235c55a2 h1:nRCnARSGqRvSAK9S+vbH87dMykfv2vWwx6kb8rw9Ayk=
|
||||
github.com/tailscale/wireguard-go v0.0.0-20260904023712-e855235c55a2/go.mod h1:rUelGmuK4UnSJYM5gl5Mknp6YbwwcL8+VAPMhNYe+jg=
|
||||
github.com/tailscale/wireguard-go v0.0.0-20260904160635-24b5b6917431 h1:opfY1yfV/mJzmbxJZw/cXLkmNwp4qjRZ51tC6RG3VPI=
|
||||
github.com/tailscale/wireguard-go v0.0.0-20260904160635-24b5b6917431/go.mod h1:rUelGmuK4UnSJYM5gl5Mknp6YbwwcL8+VAPMhNYe+jg=
|
||||
github.com/tailscale/xnet v0.0.0-20240729143630-8497ac4dab2e h1:zOGKqN5D5hHhiYUp091JqK7DPCqSARyUfduhGUY8Bek=
|
||||
github.com/tailscale/xnet v0.0.0-20240729143630-8497ac4dab2e/go.mod h1:orPd6JZXXRyuDusYilywte7k094d7dycXXU5YnWsrwg=
|
||||
github.com/tc-hib/winres v0.2.1 h1:YDE0FiP0VmtRaDn7+aaChp1KiF4owBiJa5l964l5ujA=
|
||||
|
||||
+33
-23
@@ -9,33 +9,41 @@ import (
|
||||
|
||||
"golang.org/x/net/ipv4"
|
||||
"golang.org/x/net/ipv6"
|
||||
"tailscale.com/envknob"
|
||||
"tailscale.com/net/packet"
|
||||
"tailscale.com/types/nettype"
|
||||
)
|
||||
|
||||
// BatchSizeFromEnv returns ideal, unless the TS_DEBUG_WG_BATCH_SIZE
|
||||
// environment variable is set to a positive integer, in which case it returns
|
||||
// that value clamped to [1, ideal].
|
||||
//
|
||||
// Batch size determines how much packet memory wireguard-go pins per reader
|
||||
// goroutine (batch size × 64 KiB message buffers per reader, ~32 MiB total at
|
||||
// the Linux default of 128), so memory-budget tests set the env var to 1 to
|
||||
// approximate the configuration used on memory-constrained (mobile)
|
||||
// platforms, where batch size is always 1.
|
||||
func BatchSizeFromEnv(ideal int) int {
|
||||
if v, ok := envknob.LookupInt("TS_DEBUG_WG_BATCH_SIZE"); ok && v > 0 {
|
||||
return min(v, ideal)
|
||||
}
|
||||
return ideal
|
||||
}
|
||||
|
||||
var (
|
||||
// This acts as a compile-time check for our usage of ipv6.Message in
|
||||
// [Conn] for both IPv6 and IPv4 operations.
|
||||
_ ipv6.Message = ipv4.Message{}
|
||||
)
|
||||
|
||||
// ReceivedPacket describes a packet read by [Conn.ReadBatch].
|
||||
type ReceivedPacket struct {
|
||||
// Offset is the starting byte offset into the slab supplied to [Conn.ReadBatch].
|
||||
Offset int
|
||||
// Size is the size of the packet.
|
||||
Size int
|
||||
// Source is the source address that sent the packet.
|
||||
Source netip.AddrPort
|
||||
}
|
||||
|
||||
const (
|
||||
// ReadSlabMultiple is the minimum slab length accepted by [Conn.ReadBatch],
|
||||
// and is the suggested multiple when passing a larger value.
|
||||
ReadSlabMultiple = 1<<16 - 1
|
||||
// MinimumReadBatchSize is the minimum number of packets descriptors accepted
|
||||
// by [Conn.ReadBatch].
|
||||
MinimumReadBatchSize = udpGROCountMax
|
||||
// MaximumWriteBatchSize is the maximum number of buffs accepted by
|
||||
// [Conn.WriteBatchTo].
|
||||
MaximumWriteBatchSize = 128
|
||||
// udpGROCountMax is the maximum number of datagrams the kernel will coalesce
|
||||
// together to present in a single recvmmsg() slot.
|
||||
udpGROCountMax = 64
|
||||
)
|
||||
|
||||
// Conn is a [nettype.PacketConn] that provides batched i/o using
|
||||
// platform-specific optimizations, e.g. {recv,send}mmsg & UDP GSO/GRO.
|
||||
//
|
||||
@@ -54,19 +62,21 @@ type Conn interface {
|
||||
//
|
||||
// TODO: consider detaching Conn from [nettype.PacketConn]
|
||||
ReadFromUDPAddrPort([]byte) (int, netip.AddrPort, error)
|
||||
// ReadBatch reads messages from [Conn] into msgs. It returns the number of
|
||||
// messages the caller should evaluate for nonzero len, as a zero len
|
||||
// message may fall on either side of a nonzero.
|
||||
// ReadBatch reads datagrams from [Conn] into slab, and describes them in
|
||||
// packets. It returns the number of populated packet descriptors. A single
|
||||
// GRO-coalesced datagram may produce multiple descriptors.
|
||||
//
|
||||
// Each [ipv6.Message.OOB] must be sized to at least MinControlMessageSize().
|
||||
ReadBatch(msgs []ipv6.Message, flags int) (n int, err error)
|
||||
// packets must have a length >= [MinimumReadBatchSize]. slab must have a
|
||||
// length >= [ReadSlabMultiple]. ReadBatch reads only as many datagrams as
|
||||
// both arguments can accommodate.
|
||||
ReadBatch(slab []byte, packets []ReceivedPacket) (n int, err error)
|
||||
// WriteBatchTo writes buffs to addr.
|
||||
//
|
||||
// If geneve.VNI.IsSet(), then geneve is encoded into the space preceding
|
||||
// offset, and offset must equal [packet.GeneveFixedHeaderLength]. If
|
||||
// !geneve.VNI.IsSet() then the space preceding offset is ignored.
|
||||
//
|
||||
// len(buffs) must be <= batchSize supplied in TryUpgradeToConn().
|
||||
// len(buffs) must be <= [MaximumWriteBatchSize].
|
||||
//
|
||||
// WriteBatchTo may return a [neterror.ErrUDPGSODisabled] error if UDP GSO
|
||||
// was disabled as a result of a send error.
|
||||
|
||||
@@ -11,14 +11,6 @@ import (
|
||||
)
|
||||
|
||||
// TryUpgradeToConn is no-op on all platforms except linux.
|
||||
func TryUpgradeToConn(pconn nettype.PacketConn, _ string, _ int, _ string, _ *controlknobs.Knobs) nettype.PacketConn {
|
||||
func TryUpgradeToConn(pconn nettype.PacketConn, _ string, _ string, _ *controlknobs.Knobs) nettype.PacketConn {
|
||||
return pconn
|
||||
}
|
||||
|
||||
var controlMessageSize = 0
|
||||
|
||||
func MinControlMessageSize() int {
|
||||
return controlMessageSize
|
||||
}
|
||||
|
||||
const IdealBatchSize = 1
|
||||
+126
-99
@@ -7,6 +7,7 @@ import (
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/netip"
|
||||
"runtime"
|
||||
@@ -58,7 +59,7 @@ type linuxBatchingConn struct {
|
||||
xpc xnetBatchReaderWriter
|
||||
rxOffload bool // supports UDP GRO or similar
|
||||
txOffload atomic.Bool // supports UDP GSO or similar
|
||||
sendBatchPool sync.Pool
|
||||
msgsPool sync.Pool
|
||||
rxqOverflowsMetric *clientmetric.Metric
|
||||
// neverGSOEqualTail, when non-nil and true, enables a sentinel-tail
|
||||
// workaround in the UDP GSO TX path. It points at a
|
||||
@@ -231,24 +232,31 @@ func (c *linuxBatchingConn) coalesceMessages(addr *net.UDPAddr, geneve packet.Ge
|
||||
return base + 1
|
||||
}
|
||||
|
||||
type sendBatch struct {
|
||||
msgs []ipv6.Message
|
||||
ua *net.UDPAddr
|
||||
// msgsBatch contains recyclable (via [linuxBatchingConn.msgsPool] elements
|
||||
// used across [linuxBatchingConn.ReadBatch] & [linuxBatchingConn.WriteBatchTo].
|
||||
type msgsBatch struct {
|
||||
msgs []ipv6.Message
|
||||
writeBatchToUDPAddr *net.UDPAddr
|
||||
}
|
||||
|
||||
func (c *linuxBatchingConn) getSendBatch() *sendBatch {
|
||||
batch := c.sendBatchPool.Get().(*sendBatch)
|
||||
func (c *linuxBatchingConn) getMsgsBatch() *msgsBatch {
|
||||
batch := c.msgsPool.Get().(*msgsBatch)
|
||||
return batch
|
||||
}
|
||||
|
||||
func (c *linuxBatchingConn) putSendBatch(batch *sendBatch) {
|
||||
for i := range batch.msgs {
|
||||
// Non coalesced write paths access only batch.msgs[i].Buffers[0],
|
||||
// putMsgsBatch resets each message in batch.msgs[:usedMsgs] for reuse, then
|
||||
// returns batch to its [sync.Pool].
|
||||
func (c *linuxBatchingConn) putMsgsBatch(batch *msgsBatch, usedMsgs int) {
|
||||
// Clear references to previously used packet buffers, otherwise they
|
||||
// may never be GC'd.
|
||||
for i := range batch.msgs[:usedMsgs] {
|
||||
clear((batch.msgs)[i].Buffers)
|
||||
// Non-coalesced write paths access only batch.msgs[i].Buffers[0],
|
||||
// but we append more during [linuxBatchingConn.coalesceMessages].
|
||||
// Leave index zero accessible:
|
||||
batch.msgs[i] = ipv6.Message{Buffers: batch.msgs[i].Buffers[:1], OOB: batch.msgs[i].OOB}
|
||||
batch.msgs[i] = ipv6.Message{Buffers: batch.msgs[i].Buffers[:1], OOB: batch.msgs[i].OOB[:0]}
|
||||
}
|
||||
c.sendBatchPool.Put(batch)
|
||||
c.msgsPool.Put(batch)
|
||||
}
|
||||
|
||||
// appendSentinelTailBatchSizeThreshold represents the minimum batch size
|
||||
@@ -273,18 +281,19 @@ func (c *linuxBatchingConn) putSendBatch(batch *sendBatch) {
|
||||
const appendSentinelTailBatchSizeThreshold = 8
|
||||
|
||||
func (c *linuxBatchingConn) WriteBatchTo(buffs [][]byte, addr netip.AddrPort, geneve packet.GeneveHeader, offset int) error {
|
||||
batch := c.getSendBatch()
|
||||
defer c.putSendBatch(batch)
|
||||
batch := c.getMsgsBatch()
|
||||
usedMsgs := len(buffs)
|
||||
defer c.putMsgsBatch(batch, usedMsgs)
|
||||
if addr.Addr().Is6() {
|
||||
as16 := addr.Addr().As16()
|
||||
copy(batch.ua.IP, as16[:])
|
||||
batch.ua.IP = batch.ua.IP[:16]
|
||||
copy(batch.writeBatchToUDPAddr.IP, as16[:])
|
||||
batch.writeBatchToUDPAddr.IP = batch.writeBatchToUDPAddr.IP[:16]
|
||||
} else {
|
||||
as4 := addr.Addr().As4()
|
||||
copy(batch.ua.IP, as4[:])
|
||||
batch.ua.IP = batch.ua.IP[:4]
|
||||
copy(batch.writeBatchToUDPAddr.IP, as4[:])
|
||||
batch.writeBatchToUDPAddr.IP = batch.writeBatchToUDPAddr.IP[:4]
|
||||
}
|
||||
batch.ua.Port = int(addr.Port())
|
||||
batch.writeBatchToUDPAddr.Port = int(addr.Port())
|
||||
// Load the control knob once per write so a single call sees a consistent
|
||||
// value even if the knob flips concurrently.
|
||||
neverGSOEqualTail := c.neverGSOEqualTail != nil && c.neverGSOEqualTail.Load()
|
||||
@@ -294,7 +303,7 @@ func (c *linuxBatchingConn) WriteBatchTo(buffs [][]byte, addr netip.AddrPort, ge
|
||||
)
|
||||
retry:
|
||||
if c.txOffload.Load() && (!neverGSOEqualTail || len(buffs) >= appendSentinelTailBatchSizeThreshold) {
|
||||
n = c.coalesceMessages(batch.ua, geneve, buffs, batch.msgs, offset, neverGSOEqualTail)
|
||||
n = c.coalesceMessages(batch.writeBatchToUDPAddr, geneve, buffs, batch.msgs, offset, neverGSOEqualTail)
|
||||
} else {
|
||||
mutableOffset := offset // don't mutate offset across retries
|
||||
vniIsSet := geneve.VNI.IsSet()
|
||||
@@ -310,7 +319,7 @@ retry:
|
||||
// coalesceMessages during a first pass, and landed here as part of
|
||||
// goto retry.
|
||||
batch.msgs[i].Buffers = batch.msgs[i].Buffers[:1]
|
||||
batch.msgs[i].Addr = batch.ua
|
||||
batch.msgs[i].Addr = batch.writeBatchToUDPAddr
|
||||
batch.msgs[i].OOB = batch.msgs[i].OOB[:0]
|
||||
}
|
||||
n = len(buffs)
|
||||
@@ -347,55 +356,6 @@ func (c *linuxBatchingConn) writeBatch(msgs []ipv6.Message) error {
|
||||
}
|
||||
}
|
||||
|
||||
// splitCoalescedMessages splits coalesced messages from the tail of dst
|
||||
// beginning at index 'firstMsgAt' into the head of the same slice. It reports
|
||||
// the number of elements to evaluate in msgs for nonzero len (msgs[i].N). An
|
||||
// error is returned if a socket control message cannot be parsed or a split
|
||||
// operation would overflow msgs.
|
||||
func (c *linuxBatchingConn) splitCoalescedMessages(msgs []ipv6.Message, firstMsgAt int) (n int, err error) {
|
||||
for i := firstMsgAt; i < len(msgs); i++ {
|
||||
msg := &msgs[i]
|
||||
if msg.N == 0 {
|
||||
return n, err
|
||||
}
|
||||
var (
|
||||
gsoSize int
|
||||
start int
|
||||
end = msg.N
|
||||
numToSplit = 1
|
||||
)
|
||||
gsoSize, err = getGSOSizeFromControl(msg.OOB[:msg.NN])
|
||||
if err != nil {
|
||||
return n, err
|
||||
}
|
||||
if gsoSize > 0 {
|
||||
numToSplit = (msg.N + gsoSize - 1) / gsoSize
|
||||
end = gsoSize
|
||||
}
|
||||
for j := 0; j < numToSplit; j++ {
|
||||
if n > i {
|
||||
return n, errors.New("splitting coalesced packet resulted in overflow")
|
||||
}
|
||||
copied := copy(msgs[n].Buffers[0], msg.Buffers[0][start:end])
|
||||
msgs[n].N = copied
|
||||
msgs[n].Addr = msg.Addr
|
||||
start = end
|
||||
end += gsoSize
|
||||
if end > msg.N {
|
||||
end = msg.N
|
||||
}
|
||||
n++
|
||||
}
|
||||
if i != n-1 {
|
||||
// It is legal for bytes to move within msg.Buffers[0] as a result
|
||||
// of splitting, so we only zero the source msg len when it is not
|
||||
// the destination of the last split operation above.
|
||||
msg.N = 0
|
||||
}
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// getDataFromControl returns the data portion of the first control msg with
|
||||
// matching cmsgLevel, matching cmsgType, and min data len of minDataLen, in
|
||||
// control. If no matching cmsg is found or the len(control) < unix.SizeofCmsghdr,
|
||||
@@ -439,11 +399,11 @@ func getRXQOverflowsFromControl(control []byte) (uint32, error) {
|
||||
|
||||
// handleRXQOverflowCounter handles any rx queue overflow counter contained in
|
||||
// the tail of msgs.
|
||||
func (c *linuxBatchingConn) handleRXQOverflowCounter(msgs []ipv6.Message, n int, rxErr error) {
|
||||
if n == 0 || rxErr != nil || c.rxqOverflowsMetric == nil {
|
||||
func (c *linuxBatchingConn) handleRXQOverflowCounter(msgs []ipv6.Message, rxErr error) {
|
||||
if len(msgs) == 0 || rxErr != nil || c.rxqOverflowsMetric == nil {
|
||||
return
|
||||
}
|
||||
tailMsg := msgs[n-1] // we only care about the latest value as it's a cumulative counter
|
||||
tailMsg := msgs[len(msgs)-1] // we only care about the latest value as it's a cumulative counter
|
||||
if tailMsg.NN == 0 {
|
||||
return
|
||||
}
|
||||
@@ -462,22 +422,98 @@ func (c *linuxBatchingConn) handleRXQOverflowCounter(msgs []ipv6.Message, n int,
|
||||
c.rxqOverflows = rxqOverflows
|
||||
}
|
||||
|
||||
func (c *linuxBatchingConn) ReadBatch(msgs []ipv6.Message, flags int) (n int, err error) {
|
||||
const (
|
||||
// groEnabledSlotSize is the size to supply per recvmmsg() slot when GRO is
|
||||
// enabled. If we supply less than this, coalesced datagrams may be silently
|
||||
// truncated by the kernel.
|
||||
groEnabledSlotSize = 1<<16 - 1
|
||||
// groDisabledSlotSize is the size to supply per recvmmsg() slot when GRO is
|
||||
// disabled. It's ~twice the size of max Ethernet MTU, but smaller than
|
||||
// [ReadSlabMultiple]. Without GRO we might as well read more datagrams per
|
||||
// syscall, and it's reasonable to expect datagrams to be no larger than this
|
||||
// size.
|
||||
//
|
||||
// TODO: Reconsider divisor. Older kernels without UDP GRO support may
|
||||
// benefit from more slots per recvmmsg syscall. Run benchmarks to inform.
|
||||
groDisabledSlotSize = groEnabledSlotSize / 4
|
||||
)
|
||||
|
||||
func (c *linuxBatchingConn) ReadBatch(slab []byte, packets []ReceivedPacket) (n int, err error) {
|
||||
if len(slab) < ReadSlabMultiple {
|
||||
return 0, fmt.Errorf("len(slab): %d < ReadSlabMultiple(%d)", len(slab), ReadSlabMultiple)
|
||||
}
|
||||
if len(packets) < MinimumReadBatchSize {
|
||||
return 0, fmt.Errorf("len(packets): %d < MinimumReadBatchSize(%d)", len(packets), MinimumReadBatchSize)
|
||||
}
|
||||
c.readOpMu.Lock()
|
||||
defer c.readOpMu.Unlock()
|
||||
if !c.rxOffload || len(msgs) < 2 {
|
||||
n, err = c.xpc.ReadBatch(msgs, flags)
|
||||
c.handleRXQOverflowCounter(msgs, n, err)
|
||||
return n, err
|
||||
batch := c.getMsgsBatch()
|
||||
recvmmsgSlotSize := groEnabledSlotSize
|
||||
maxMsgsPerRecvmmsgSlot := udpGROCountMax
|
||||
if !c.rxOffload {
|
||||
recvmmsgSlotSize = groDisabledSlotSize
|
||||
maxMsgsPerRecvmmsgSlot = 1
|
||||
}
|
||||
// Read into the tail of msgs, split into the head.
|
||||
readAt := len(msgs) - 2
|
||||
n, err = c.xpc.ReadBatch(msgs[readAt:], 0)
|
||||
if err != nil || n == 0 {
|
||||
maxMsgCount := min(
|
||||
len(slab)/recvmmsgSlotSize, // how many slots fit in slab
|
||||
len(packets)/maxMsgsPerRecvmmsgSlot, // how many datagrams fit in packets
|
||||
len(batch.msgs), // how many [ipv6.Message] we have
|
||||
)
|
||||
defer c.putMsgsBatch(batch, maxMsgCount)
|
||||
rem := slab
|
||||
for i := range maxMsgCount {
|
||||
end := recvmmsgSlotSize
|
||||
if len(rem) < recvmmsgSlotSize {
|
||||
end = len(rem)
|
||||
}
|
||||
batch.msgs[i].Buffers[0] = rem[:end]
|
||||
batch.msgs[i].OOB = batch.msgs[i].OOB[:cap(batch.msgs[i].OOB)]
|
||||
rem = rem[end:]
|
||||
}
|
||||
n, err = c.xpc.ReadBatch(batch.msgs[:maxMsgCount], 0)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
c.handleRXQOverflowCounter(msgs[readAt:], n, err)
|
||||
return c.splitCoalescedMessages(msgs, readAt)
|
||||
c.handleRXQOverflowCounter(batch.msgs[:n], err)
|
||||
return fillReceivedPackets(batch.msgs[:n], recvmmsgSlotSize, packets, c.rxOffload)
|
||||
}
|
||||
|
||||
func fillReceivedPackets(msgs []ipv6.Message, slabOffset int, packets []ReceivedPacket, rxOffload bool) (n int, err error) {
|
||||
for i, msg := range msgs {
|
||||
var (
|
||||
gsoSize int
|
||||
start int
|
||||
numToSplit = 1
|
||||
)
|
||||
if rxOffload {
|
||||
gsoSize, err = getGSOSizeFromControl(msg.OOB[:msg.NN])
|
||||
if err != nil {
|
||||
return n, err
|
||||
}
|
||||
if gsoSize > 0 {
|
||||
numToSplit = (msg.N + gsoSize - 1) / gsoSize
|
||||
}
|
||||
}
|
||||
addrPort := msg.Addr.(*net.UDPAddr).AddrPort()
|
||||
regionOffset := i * slabOffset // region may contain multiple coalesced packets
|
||||
for j := 0; j < numToSplit; j++ {
|
||||
if n >= len(packets) {
|
||||
return n, fmt.Errorf("%w: filling received packet metadata resulted in overflow", io.ErrShortBuffer)
|
||||
}
|
||||
end := msg.N
|
||||
if gsoSize > 0 && start+gsoSize < end {
|
||||
end = start + gsoSize
|
||||
}
|
||||
packets[n] = ReceivedPacket{
|
||||
Offset: regionOffset + start,
|
||||
Size: end - start,
|
||||
Source: addrPort,
|
||||
}
|
||||
n++
|
||||
start = end
|
||||
}
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (c *linuxBatchingConn) LocalAddr() net.Addr {
|
||||
@@ -603,12 +639,11 @@ func getRXQOverflowsMetric(name string) *clientmetric.Metric {
|
||||
}
|
||||
|
||||
// TryUpgradeToConn probes the capabilities of the OS and pconn, and upgrades
|
||||
// pconn to a [Conn] if appropriate. A batch size of [IdealBatchSize] is
|
||||
// suggested for the best performance. If len(rxqOverflowsMetricName) is
|
||||
// pconn to a [Conn] if appropriate. If len(rxqOverflowsMetricName) is
|
||||
// nonzero, then read ops will propagate the SO_RXQ_OVFL control message counter
|
||||
// to a clientmetric with the supplied name. If knobs is non-nil, UDP GSO
|
||||
// and/or UDP GRO may be disabled via control-plane node attributes.
|
||||
func TryUpgradeToConn(pconn nettype.PacketConn, network string, batchSize int, rxqOverflowsMetricName string, knobs *controlknobs.Knobs) nettype.PacketConn {
|
||||
func TryUpgradeToConn(pconn nettype.PacketConn, network string, rxqOverflowsMetricName string, knobs *controlknobs.Knobs) nettype.PacketConn {
|
||||
if runtime.GOOS != "linux" {
|
||||
// Exclude Android.
|
||||
return pconn
|
||||
@@ -632,20 +667,20 @@ func TryUpgradeToConn(pconn nettype.PacketConn, network string, batchSize int, r
|
||||
}
|
||||
b := &linuxBatchingConn{
|
||||
pc: uc,
|
||||
sendBatchPool: sync.Pool{
|
||||
msgsPool: sync.Pool{
|
||||
New: func() any {
|
||||
ua := &net.UDPAddr{
|
||||
IP: make([]byte, 16),
|
||||
}
|
||||
msgs := make([]ipv6.Message, batchSize)
|
||||
msgs := make([]ipv6.Message, MaximumWriteBatchSize)
|
||||
for i := range msgs {
|
||||
msgs[i].Buffers = make([][]byte, 1)
|
||||
msgs[i].Addr = ua
|
||||
msgs[i].OOB = make([]byte, controlMessageSize)
|
||||
}
|
||||
return &sendBatch{
|
||||
ua: ua,
|
||||
msgs: msgs,
|
||||
return &msgsBatch{
|
||||
writeBatchToUDPAddr: ua,
|
||||
msgs: msgs,
|
||||
}
|
||||
},
|
||||
},
|
||||
@@ -682,11 +717,3 @@ func init() {
|
||||
unix.CmsgSpace(2) + // UDP_GRO or UDP_SEGMENT gsoSize (uint16)
|
||||
unix.CmsgSpace(4) // SO_RXQ_OVFL counter (uint32)
|
||||
}
|
||||
|
||||
// MinControlMessageSize returns the minimum control message size required to
|
||||
// support read batching via [Conn.ReadBatch].
|
||||
func MinControlMessageSize() int {
|
||||
return controlMessageSize
|
||||
}
|
||||
|
||||
const IdealBatchSize = 128
|
||||
@@ -24,16 +24,18 @@ import (
|
||||
"tailscale.com/net/packet"
|
||||
)
|
||||
|
||||
func Test_linuxBatchingConn_splitCoalescedMessages(t *testing.T) {
|
||||
c := &linuxBatchingConn{}
|
||||
func Test_fillReceivedPackets(t *testing.T) {
|
||||
const maxDatagramSize = 1<<16 - 1
|
||||
|
||||
source := netip.MustParseAddrPort("192.0.2.1:1234")
|
||||
newMsg := func(n int, gso uint16) ipv6.Message {
|
||||
msg := ipv6.Message{
|
||||
Buffers: [][]byte{make([]byte, 1024)},
|
||||
Buffers: [][]byte{make([]byte, maxDatagramSize)},
|
||||
N: n,
|
||||
OOB: gsoControl(gso),
|
||||
Addr: net.UDPAddrFromAddrPort(source),
|
||||
}
|
||||
if gso > 0 {
|
||||
msg.OOB = gsoControl(gso)
|
||||
msg.NN = len(msg.OOB)
|
||||
}
|
||||
return msg
|
||||
@@ -42,103 +44,133 @@ func Test_linuxBatchingConn_splitCoalescedMessages(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
msgs []ipv6.Message
|
||||
firstMsgAt int
|
||||
wantNumEval int
|
||||
wantMsgLens []int
|
||||
wantPackets [][2]int // {offset, size}
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
name: "second-last-split-last-empty",
|
||||
name: "first_split",
|
||||
msgs: []ipv6.Message{
|
||||
newMsg(0, 0),
|
||||
newMsg(0, 0),
|
||||
newMsg(3, 1),
|
||||
newMsg(0, 0),
|
||||
},
|
||||
firstMsgAt: 2,
|
||||
wantNumEval: 3,
|
||||
wantMsgLens: []int{1, 1, 1, 0},
|
||||
wantErr: false,
|
||||
wantPackets: [][2]int{
|
||||
{0, 1},
|
||||
{1, 1},
|
||||
{2, 1},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "second-last-no-split-last-empty",
|
||||
name: "first_no_split",
|
||||
msgs: []ipv6.Message{
|
||||
newMsg(0, 0),
|
||||
newMsg(0, 0),
|
||||
newMsg(1, 0),
|
||||
newMsg(0, 0),
|
||||
},
|
||||
firstMsgAt: 2,
|
||||
wantNumEval: 1,
|
||||
wantMsgLens: []int{1, 0, 0, 0},
|
||||
wantErr: false,
|
||||
wantPackets: [][2]int{
|
||||
{0, 1},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "second-last-no-split-last-no-split",
|
||||
name: "first_no_split_last_no_split",
|
||||
msgs: []ipv6.Message{
|
||||
newMsg(0, 0),
|
||||
newMsg(0, 0),
|
||||
newMsg(1, 0),
|
||||
newMsg(1, 0),
|
||||
},
|
||||
firstMsgAt: 2,
|
||||
wantNumEval: 2,
|
||||
wantMsgLens: []int{1, 1, 0, 0},
|
||||
wantErr: false,
|
||||
wantPackets: [][2]int{
|
||||
{0, 1},
|
||||
{maxDatagramSize, 1},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "second-last-no-split-last-split",
|
||||
name: "first_no_split_last_split",
|
||||
msgs: []ipv6.Message{
|
||||
newMsg(0, 0),
|
||||
newMsg(0, 0),
|
||||
newMsg(1, 0),
|
||||
newMsg(3, 1),
|
||||
},
|
||||
firstMsgAt: 2,
|
||||
wantNumEval: 4,
|
||||
wantMsgLens: []int{1, 1, 1, 1},
|
||||
wantErr: false,
|
||||
wantPackets: [][2]int{
|
||||
{0, 1},
|
||||
{maxDatagramSize, 1},
|
||||
{maxDatagramSize + 1, 1},
|
||||
{maxDatagramSize + 2, 1},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "second-last-split-last-split",
|
||||
name: "first_split_last_split",
|
||||
msgs: []ipv6.Message{
|
||||
newMsg(0, 0),
|
||||
newMsg(0, 0),
|
||||
newMsg(2, 1),
|
||||
newMsg(2, 1),
|
||||
},
|
||||
firstMsgAt: 2,
|
||||
wantNumEval: 4,
|
||||
wantMsgLens: []int{1, 1, 1, 1},
|
||||
wantErr: false,
|
||||
wantPackets: [][2]int{
|
||||
{0, 1},
|
||||
{1, 1},
|
||||
{maxDatagramSize, 1},
|
||||
{maxDatagramSize + 1, 1},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "second-last-no-split-last-split-overflow",
|
||||
name: "first_no_split_last_split_overflow",
|
||||
msgs: []ipv6.Message{
|
||||
newMsg(0, 0),
|
||||
newMsg(0, 0),
|
||||
newMsg(1, 0),
|
||||
newMsg(4, 1),
|
||||
},
|
||||
firstMsgAt: 2,
|
||||
wantNumEval: 4,
|
||||
wantMsgLens: []int{1, 1, 1, 1},
|
||||
wantErr: true,
|
||||
wantPackets: [][2]int{
|
||||
{0, 1},
|
||||
{maxDatagramSize, 1},
|
||||
{maxDatagramSize + 1, 1},
|
||||
{maxDatagramSize + 2, 1},
|
||||
},
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "split_with_short_tail",
|
||||
msgs: []ipv6.Message{
|
||||
newMsg(5, 2),
|
||||
},
|
||||
wantNumEval: 3,
|
||||
wantPackets: [][2]int{
|
||||
{0, 2},
|
||||
{2, 2},
|
||||
{4, 1},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range cases {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := c.splitCoalescedMessages(tt.msgs, 2)
|
||||
if err != nil && !tt.wantErr {
|
||||
t.Fatalf("err: %v", err)
|
||||
packets := make([]ReceivedPacket, len(tt.wantPackets))
|
||||
got, err := fillReceivedPackets(
|
||||
tt.msgs,
|
||||
maxDatagramSize,
|
||||
packets,
|
||||
true,
|
||||
)
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Fatalf("err: %v, wantErr: %v", err, tt.wantErr)
|
||||
}
|
||||
if got != tt.wantNumEval {
|
||||
t.Fatalf("got to eval: %d want: %d", got, tt.wantNumEval)
|
||||
}
|
||||
for i, msg := range tt.msgs {
|
||||
if msg.N != tt.wantMsgLens[i] {
|
||||
t.Fatalf("msg[%d].N: %d want: %d", i, msg.N, tt.wantMsgLens[i])
|
||||
if len(packets) != len(tt.wantPackets) {
|
||||
t.Fatalf("got %d packets, want %d", len(packets), len(tt.wantPackets))
|
||||
}
|
||||
for i, want := range tt.wantPackets {
|
||||
got := packets[i]
|
||||
if got.Offset != want[0] ||
|
||||
got.Size != want[1] ||
|
||||
got.Source != source {
|
||||
t.Errorf(
|
||||
"packets[%d] = {Offset: %d, Size: %d, Source: %v}, want {Offset: %d, Size: %d, Source: %v}",
|
||||
i, got.Offset, got.Size, got.Source, want[0], want[1], source,
|
||||
)
|
||||
}
|
||||
}
|
||||
})
|
||||
@@ -466,7 +498,7 @@ func Test_linuxBatchingConn_WriteBatchTo_resetsBuffersOnGSORetry(t *testing.T) {
|
||||
c := &linuxBatchingConn{
|
||||
pc: uc,
|
||||
xpc: xpc,
|
||||
sendBatchPool: sync.Pool{New: func() any {
|
||||
msgsPool: sync.Pool{New: func() any {
|
||||
ua := &net.UDPAddr{IP: make([]byte, 16)}
|
||||
msgs := make([]ipv6.Message, 8)
|
||||
for i := range msgs {
|
||||
@@ -474,7 +506,7 @@ func Test_linuxBatchingConn_WriteBatchTo_resetsBuffersOnGSORetry(t *testing.T) {
|
||||
msgs[i].Addr = ua
|
||||
msgs[i].OOB = make([]byte, controlMessageSize)
|
||||
}
|
||||
return &sendBatch{ua: ua, msgs: msgs}
|
||||
return &msgsBatch{writeBatchToUDPAddr: ua, msgs: msgs}
|
||||
}},
|
||||
}
|
||||
c.txOffload.Store(true) // force the coalesce path on the first pass
|
||||
@@ -529,7 +561,7 @@ func Test_linuxBatchingConn_WriteBatchTo_offsetStableOnNonCoalesceRetry(t *testi
|
||||
c := &linuxBatchingConn{
|
||||
pc: uc,
|
||||
xpc: xpc,
|
||||
sendBatchPool: sync.Pool{New: func() any {
|
||||
msgsPool: sync.Pool{New: func() any {
|
||||
ua := &net.UDPAddr{IP: make([]byte, 16)}
|
||||
msgs := make([]ipv6.Message, appendSentinelTailBatchSizeThreshold)
|
||||
for i := range msgs {
|
||||
@@ -537,7 +569,7 @@ func Test_linuxBatchingConn_WriteBatchTo_offsetStableOnNonCoalesceRetry(t *testi
|
||||
msgs[i].Addr = ua
|
||||
msgs[i].OOB = make([]byte, controlMessageSize)
|
||||
}
|
||||
return &sendBatch{ua: ua, msgs: msgs}
|
||||
return &msgsBatch{writeBatchToUDPAddr: ua, msgs: msgs}
|
||||
}},
|
||||
}
|
||||
c.txOffload.Store(true)
|
||||
@@ -583,8 +615,8 @@ func TestMinReadBatchMsgsLen(t *testing.T) {
|
||||
// So long as magicsock uses [Conn], and [wireguard-go/conn.Bind] API is
|
||||
// shaped for wireguard-go to control packet memory, these values should be
|
||||
// aligned.
|
||||
if IdealBatchSize != conn.IdealBatchSize {
|
||||
t.Fatalf("IdealBatchSize: %d != conn.IdealBatchSize(): %d", IdealBatchSize, conn.IdealBatchSize)
|
||||
if MaximumWriteBatchSize != conn.IdealBatchSize {
|
||||
t.Fatalf("MaximumWriteBatchSize: %d != conn.IdealBatchSize(): %d", MaximumWriteBatchSize, conn.IdealBatchSize)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -741,12 +773,12 @@ func Test_linuxBatchingConn_handleRXQOverflowCounter(t *testing.T) {
|
||||
}
|
||||
conn.rxqOverflowsMetric.Set(0) // test count > 1 will accumulate, reset
|
||||
|
||||
// n == 0
|
||||
conn.handleRXQOverflowCounter([]ipv6.Message{{}}, 0, nil)
|
||||
// len(msgs) == 0
|
||||
conn.handleRXQOverflowCounter([]ipv6.Message{}, nil)
|
||||
c.Assert(conn.rxqOverflowsMetric.Value(), qt.Equals, int64(0))
|
||||
|
||||
// rxErr non-nil
|
||||
conn.handleRXQOverflowCounter([]ipv6.Message{{}}, 0, io.EOF)
|
||||
conn.handleRXQOverflowCounter([]ipv6.Message{{}}, io.EOF)
|
||||
c.Assert(conn.rxqOverflowsMetric.Value(), qt.Equals, int64(0))
|
||||
|
||||
// nonzero counter
|
||||
@@ -754,14 +786,14 @@ func Test_linuxBatchingConn_handleRXQOverflowCounter(t *testing.T) {
|
||||
conn.handleRXQOverflowCounter([]ipv6.Message{{
|
||||
OOB: control,
|
||||
NN: len(control),
|
||||
}}, 1, nil)
|
||||
}}, nil)
|
||||
c.Assert(conn.rxqOverflowsMetric.Value(), qt.Equals, int64(1))
|
||||
|
||||
// nonzero counter, no change
|
||||
conn.handleRXQOverflowCounter([]ipv6.Message{{
|
||||
OOB: control,
|
||||
NN: len(control),
|
||||
}}, 1, nil)
|
||||
}}, nil)
|
||||
c.Assert(conn.rxqOverflowsMetric.Value(), qt.Equals, int64(1))
|
||||
|
||||
// counter rollover
|
||||
@@ -769,6 +801,6 @@ func Test_linuxBatchingConn_handleRXQOverflowCounter(t *testing.T) {
|
||||
conn.handleRXQOverflowCounter([]ipv6.Message{{
|
||||
OOB: control,
|
||||
NN: len(control),
|
||||
}}, 1, nil)
|
||||
}}, nil)
|
||||
c.Assert(conn.rxqOverflowsMetric.Value(), qt.Equals, int64(1+math.MaxUint32))
|
||||
}
|
||||
+1
-1
@@ -33,7 +33,7 @@ func (t *fakeTUN) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *fakeTUN) Read(out [][]byte, sizes []int, offset int) (int, error) {
|
||||
func (t *fakeTUN) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
|
||||
<-t.closechan
|
||||
return 0, io.EOF
|
||||
}
|
||||
|
||||
+127
-116
@@ -24,7 +24,6 @@ import (
|
||||
"tailscale.com/disco"
|
||||
"tailscale.com/feature"
|
||||
"tailscale.com/feature/buildfeatures"
|
||||
"tailscale.com/net/batching"
|
||||
"tailscale.com/net/packet"
|
||||
"tailscale.com/net/packet/checksum"
|
||||
"tailscale.com/net/routemanager"
|
||||
@@ -44,12 +43,16 @@ import (
|
||||
"tailscale.com/wgengine/wgcfg"
|
||||
)
|
||||
|
||||
const maxBufferSize = device.MaxMessageSize
|
||||
|
||||
// PacketStartOffset is the minimal amount of leading space that must exist
|
||||
// before &packet[offset] in a packet passed to Read, Write, or InjectInboundDirect.
|
||||
// This is necessary to avoid reallocation in wireguard-go internals.
|
||||
const PacketStartOffset = device.MessageTransportHeaderSize
|
||||
// WritePacketStartOffset is the minimal amount of leading space that must exist
|
||||
// before &packet[offset] in a packet passed to [tun.Device.Write], which includes
|
||||
// "injected" packet code paths. This leading space is required by [tun.Device]
|
||||
// implementations.
|
||||
//
|
||||
// TODO(jwhited): consider collapsing with and renaming [tun.ReadPacketSpacing]
|
||||
// for simplicity's sake. Exact math between [tun.Device] implementation and
|
||||
// direction varies, but we can just pick the maximum value to reduce cognitive
|
||||
// burden.
|
||||
const WritePacketStartOffset = device.MessageTransportHeaderSize
|
||||
|
||||
// MaxPacketSize is the maximum size (in bytes)
|
||||
// of a packet that can be injected into a tstun.Wrapper.
|
||||
@@ -68,7 +71,7 @@ var (
|
||||
var (
|
||||
errPacketTooBig = errors.New("packet too big")
|
||||
errOffsetTooBig = errors.New("offset larger than buffer length")
|
||||
errOffsetTooSmall = errors.New("offset smaller than PacketStartOffset")
|
||||
errOffsetTooSmall = errors.New("offset smaller than WritePacketStartOffset")
|
||||
)
|
||||
|
||||
// parsedPacketPool holds a pool of Parsed structs for use in filtering.
|
||||
@@ -119,17 +122,17 @@ type Wrapper struct {
|
||||
// peerConfig stores the current NAT configuration.
|
||||
peerConfig atomic.Pointer[peerConfigTable]
|
||||
|
||||
// vectorBuffer stores the oldest unconsumed packet vector from tdev. It is
|
||||
// allocated in wrap() and the underlying arrays should never grow.
|
||||
vectorBuffer [][]byte
|
||||
// startPollingOnce is used to start a [Wrapper.pollVector] goroutine at the
|
||||
// first call to [Wrapper.Read].
|
||||
startPollingOnce sync.Once
|
||||
// bufferConsumedMu protects bufferConsumed from concurrent sends, closes,
|
||||
// and send-after-close (by way of bufferConsumedClosed).
|
||||
bufferConsumedMu sync.Mutex
|
||||
// bufferConsumedClosed is true when bufferConsumed has been closed. This is
|
||||
// read by bufferConsumed writers to prevent send-after-close.
|
||||
bufferConsumedClosed bool
|
||||
// bufferConsumed synchronizes access to vectorBuffer (shared by Read() and
|
||||
// pollVector()).
|
||||
// bufferConsumed synchronizes access to packet bufs and descriptors shared
|
||||
// by [Wrapper.Read] and [Wrapper.pollVector].
|
||||
//
|
||||
// Close closes bufferConsumed and sets bufferConsumedClosed to true.
|
||||
bufferConsumed chan struct{}
|
||||
@@ -253,17 +256,25 @@ type tunInjectedRead struct {
|
||||
data []byte
|
||||
}
|
||||
|
||||
// tunVectorReadResult is the result of a tun.Read(), or an injected packet
|
||||
// pretending to be a tun.Read().
|
||||
// tunVectorReadResult is the result of a [tun.Device.Read], or an injected
|
||||
// packet pretending to be a [tun.Device.Read].
|
||||
type tunVectorReadResult struct {
|
||||
// When err AND data are nil, injected will be set with meaningful data
|
||||
// (injected packet). If either err OR data is non-nil, injected should be
|
||||
// ignored (a "real" tun.Read).
|
||||
err error
|
||||
data [][]byte
|
||||
injected tunInjectedRead
|
||||
// isInjected indicates if tunVectorReadResult contains a "real" [tun.Device.Read]
|
||||
// result, or an injected packet. When true, injected will be set with meaningful
|
||||
// data, otherwise real will be set with meaningful data.
|
||||
isInjected bool
|
||||
// Result consumer ([Wrapper.Read]) must call a non-nil doneHandlingFn when
|
||||
// they are done with the result. The consumer must not access [tunVectorReadResult]
|
||||
// fields once this func has been called, as it provides synchronization
|
||||
// around shared memory.
|
||||
doneHandlingFn func()
|
||||
|
||||
dataOffset int
|
||||
real struct {
|
||||
err error
|
||||
slab []byte
|
||||
packets []tun.ReadPacket
|
||||
}
|
||||
injected tunInjectedRead
|
||||
}
|
||||
|
||||
// Start unblocks any Wrapper.Read calls that have already started
|
||||
@@ -318,12 +329,6 @@ func wrap(logf logger.Logf, tdev tun.Device, isTAP bool, m *usermetric.Registry,
|
||||
w.eventClient = bus.Client("net.tstun")
|
||||
w.discoKeyAdvertisementPub = eventbus.Publish[events.DiscoKeyAdvertisement](w.eventClient)
|
||||
|
||||
w.vectorBuffer = make([][]byte, tdev.BatchSize())
|
||||
for i := range w.vectorBuffer {
|
||||
w.vectorBuffer[i] = make([]byte, maxBufferSize)
|
||||
}
|
||||
go w.pollVector()
|
||||
|
||||
go w.pumpEvents()
|
||||
// The buffer starts out consumed.
|
||||
w.bufferConsumed <- struct{}{}
|
||||
@@ -461,52 +466,45 @@ func (t *Wrapper) Name() (string, error) {
|
||||
return t.tdev.Name()
|
||||
}
|
||||
|
||||
const ethernetFrameSize = 14 // 2 six byte MACs, 2 bytes ethertype
|
||||
|
||||
// pollVector polls t.tdev.Read(), placing the oldest unconsumed packet vector
|
||||
// into t.vectorBuffer. This is needed because t.tdev.Read() in general may
|
||||
// block (it does on Windows), so packets may be stuck in t.vectorOutbound if
|
||||
// t.Read() called t.tdev.Read() directly.
|
||||
func (t *Wrapper) pollVector() {
|
||||
sizes := make([]int, len(t.vectorBuffer))
|
||||
readOffset := PacketStartOffset
|
||||
reader := t.tdev.Read
|
||||
if t.isTAP {
|
||||
type tapReader interface {
|
||||
ReadEthernet(buffs [][]byte, sizes []int, offset int) (int, error)
|
||||
}
|
||||
if r, ok := t.tdev.(tapReader); ok {
|
||||
readOffset = PacketStartOffset - ethernetFrameSize
|
||||
reader = r.ReadEthernet
|
||||
}
|
||||
}
|
||||
|
||||
// pollVector polls [Wrapper.tdev.Read], writing the oldest unconsumed packet
|
||||
// slab and packet descriptors into the [Wrapper.vectorOutbound] channel.
|
||||
// slabLen and packetsLen should originate from the first call to [Wrapper.Read],
|
||||
// and are used for sizing the equivalent arguments pollVector passes to
|
||||
// [Wrapper.tdev.Read].
|
||||
//
|
||||
// [Wrapper.tdev.Read] can block, so we poll tdev in a goroutine independent of
|
||||
// wireguard-go's calls to [Wrapper.Read], in order to support native tdev reads
|
||||
// alongside packets we inject.
|
||||
//
|
||||
// pollVector returns when [t.bufferConsumed] is closed, or when [Wrapper.isClosed]
|
||||
// returns true.
|
||||
func (t *Wrapper) pollVector(slabLen, packetsLen int) {
|
||||
slab := make([]byte, slabLen)
|
||||
packets := make([]tun.ReadPacket, packetsLen)
|
||||
for range t.bufferConsumed {
|
||||
for i := range t.vectorBuffer {
|
||||
t.vectorBuffer[i] = t.vectorBuffer[i][:cap(t.vectorBuffer[i])]
|
||||
}
|
||||
var n int
|
||||
var err error
|
||||
for n == 0 && err == nil {
|
||||
if t.isClosed() {
|
||||
return
|
||||
}
|
||||
n, err = reader(t.vectorBuffer[:], sizes, readOffset)
|
||||
n, err = t.tdev.Read(slab, packets)
|
||||
if t.isTAP && TAPDebug {
|
||||
s := fmt.Sprintf("% x", t.vectorBuffer[0][:])
|
||||
s := fmt.Sprintf("% x", slab)
|
||||
for strings.HasSuffix(s, " 00") {
|
||||
s = strings.TrimSuffix(s, " 00")
|
||||
}
|
||||
t.logf("TAP read %v, %v: %s", n, err, s)
|
||||
}
|
||||
}
|
||||
for i := range sizes[:n] {
|
||||
t.vectorBuffer[i] = t.vectorBuffer[i][:readOffset+sizes[i]]
|
||||
}
|
||||
t.sendVectorOutbound(tunVectorReadResult{
|
||||
data: t.vectorBuffer[:n],
|
||||
dataOffset: PacketStartOffset,
|
||||
err: err,
|
||||
isInjected: false,
|
||||
doneHandlingFn: t.sendBufferConsumed,
|
||||
real: struct {
|
||||
err error
|
||||
slab []byte
|
||||
packets []tun.ReadPacket
|
||||
}{err: err, slab: slab, packets: packets[:n]},
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -529,7 +527,7 @@ func (t *Wrapper) injectOutbound(r tunInjectedRead) {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case t.vectorOutbound <- tunVectorReadResult{injected: r}:
|
||||
case t.vectorOutbound <- tunVectorReadResult{injected: r, isInjected: true}:
|
||||
case <-t.closed:
|
||||
}
|
||||
}
|
||||
@@ -548,6 +546,7 @@ func (t *Wrapper) sendVectorOutbound(r tunVectorReadResult) {
|
||||
}
|
||||
|
||||
// snat does SNAT on p if the destination address requires a different source address.
|
||||
// snat never mutates packet length or IP address family.
|
||||
func (pc *peerConfigTable) snat(p *packet.Parsed) {
|
||||
oldSrc := p.Src.Addr()
|
||||
newSrc := pc.selectSrcIP(oldSrc, p.Dst.Addr())
|
||||
@@ -556,7 +555,8 @@ func (pc *peerConfigTable) snat(p *packet.Parsed) {
|
||||
}
|
||||
}
|
||||
|
||||
// dnat does destination NAT on p.
|
||||
// dnat does destination NAT on p. dnat never mutates packet length or IP address
|
||||
// family.
|
||||
func (pc *peerConfigTable) dnat(p *packet.Parsed) {
|
||||
oldDst := p.Dst.Addr()
|
||||
newDst := pc.mapDstIP(p.Src.Addr(), oldDst)
|
||||
@@ -868,32 +868,42 @@ func (t *Wrapper) awaitStart() {
|
||||
}
|
||||
}
|
||||
|
||||
func (t *Wrapper) Read(buffs [][]byte, sizes []int, offset int) (int, error) {
|
||||
// Read implements [tun.Device.Read].
|
||||
func (t *Wrapper) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
|
||||
if !t.started.Load() {
|
||||
t.awaitStart()
|
||||
}
|
||||
t.startPollingOnce.Do(func() {
|
||||
go t.pollVector(len(slab), len(packets))
|
||||
})
|
||||
// packet from OS read and sent to WG
|
||||
res, ok := <-t.vectorOutbound
|
||||
if !ok {
|
||||
return 0, io.EOF
|
||||
}
|
||||
if res.err != nil && len(res.data) == 0 {
|
||||
return 0, res.err
|
||||
defer func() {
|
||||
if res.doneHandlingFn != nil {
|
||||
res.doneHandlingFn()
|
||||
}
|
||||
}()
|
||||
if res.isInjected {
|
||||
return t.injectedRead(res.injected, slab, packets, tun.ReadPacketSpacing)
|
||||
}
|
||||
if res.data == nil {
|
||||
return t.injectedRead(res.injected, buffs, sizes, offset)
|
||||
if res.real.err != nil && len(res.real.packets) == 0 {
|
||||
return 0, res.real.err
|
||||
}
|
||||
|
||||
metricPacketOut.Add(int64(len(res.data)))
|
||||
metricPacketOut.Add(int64(len(res.real.packets)))
|
||||
|
||||
var buffsPos int
|
||||
var numPackets int
|
||||
p := parsedPacketPool.Get().(*packet.Parsed)
|
||||
defer parsedPacketPool.Put(p)
|
||||
captHook := t.captureHook.Load()
|
||||
pc := t.peerConfig.Load()
|
||||
var buffsGRO *gro.GRO
|
||||
for _, data := range res.data {
|
||||
p.Decode(data[res.dataOffset:])
|
||||
for _, meta := range res.real.packets {
|
||||
data := res.real.slab[meta.Offset : meta.Offset+meta.Size]
|
||||
p.Decode(data)
|
||||
|
||||
if buildfeatures.HasCapture && captHook != nil {
|
||||
captHook(packet.FromLocal, t.now(), p.Buffer(), p.CaptureMeta)
|
||||
@@ -915,27 +925,19 @@ func (t *Wrapper) Read(buffs [][]byte, sizes []int, offset int) (int, error) {
|
||||
// Make sure to do SNAT after filtering, so that any flow tracking in
|
||||
// the filter sees the original source address. See #12133.
|
||||
pc.snat(p)
|
||||
n := copy(buffs[buffsPos][offset:], p.Buffer())
|
||||
if n != len(data)-res.dataOffset {
|
||||
panic(fmt.Sprintf("short copy: %d != %d", n, len(data)-res.dataOffset))
|
||||
n := copy(slab[meta.Offset:meta.Offset+meta.Size], p.Buffer())
|
||||
if n != len(data) {
|
||||
panic(fmt.Sprintf("short copy: %d != %d", n, len(data)))
|
||||
}
|
||||
sizes[buffsPos] = n
|
||||
buffsPos++
|
||||
packets[numPackets] = meta
|
||||
numPackets++
|
||||
}
|
||||
if buffsGRO != nil {
|
||||
buffsGRO.Flush()
|
||||
}
|
||||
|
||||
// t.vectorBuffer has a fixed location in memory.
|
||||
// TODO(raggi): add an explicit field and possibly method to the tunVectorReadResult
|
||||
// to signal when sendBufferConsumed should be called.
|
||||
if &res.data[0] == &t.vectorBuffer[0] {
|
||||
// We are done with t.buffer. Let poll() re-use it.
|
||||
t.sendBufferConsumed()
|
||||
}
|
||||
|
||||
t.noteActivity()
|
||||
return buffsPos, res.err
|
||||
return numPackets, res.real.err
|
||||
}
|
||||
|
||||
const (
|
||||
@@ -1000,10 +1002,10 @@ func invertGSOChecksum(pkt []byte, gso netstack_GSO) {
|
||||
// filter rules, but UDP/SCTP flow state is still recorded via
|
||||
// [filter.Filter.UpdateOutboundFlowState] so inbound replies are admitted by
|
||||
// [filter.Filter.RunIn].
|
||||
func (t *Wrapper) injectedRead(res tunInjectedRead, outBuffs [][]byte, sizes []int, offset int) (n int, err error) {
|
||||
func (t *Wrapper) injectedRead(res tunInjectedRead, slab []byte, packets []tun.ReadPacket, spacing int) (n int, err error) {
|
||||
var gso netstack_GSO
|
||||
|
||||
pkt := outBuffs[0][offset:]
|
||||
pkt := slab[spacing : len(slab)-spacing]
|
||||
if res.packet != nil {
|
||||
if !buildfeatures.HasNetstack {
|
||||
panic("unreachable")
|
||||
@@ -1015,8 +1017,11 @@ func (t *Wrapper) injectedRead(res tunInjectedRead, outBuffs [][]byte, sizes []i
|
||||
pkt = pkt[:bufN]
|
||||
defer res.packet.DecRef() // defer DecRef so we may continue to reference it
|
||||
} else {
|
||||
sizes[0] = copy(pkt, res.data)
|
||||
pkt = pkt[:sizes[0]]
|
||||
packets[0] = tun.ReadPacket{
|
||||
Offset: spacing,
|
||||
Size: copy(pkt, res.data),
|
||||
}
|
||||
pkt = pkt[:packets[0].Size]
|
||||
n = 1
|
||||
}
|
||||
|
||||
@@ -1078,13 +1083,15 @@ func (t *Wrapper) injectedRead(res tunInjectedRead, outBuffs [][]byte, sizes []i
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n, err = tun.GSOSplit(pkt, gsoOptions, outBuffs, sizes, offset)
|
||||
n, err = tun.GSOSplit(pkt, gsoOptions, slab, packets, spacing)
|
||||
}
|
||||
|
||||
if buildfeatures.HasNetLog {
|
||||
if update := t.connCounter.Load(); update != nil {
|
||||
for i := 0; i < n; i++ {
|
||||
updateConnCounter(update, outBuffs[i][offset:offset+sizes[i]], false)
|
||||
start := packets[i].Offset
|
||||
end := packets[i].Offset + packets[i].Size
|
||||
updateConnCounter(update, slab[start:end], false)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1298,27 +1305,36 @@ func (t *Wrapper) SetJailedFilter(filt *filter.Filter) {
|
||||
t.jailedFilter.Store(filt)
|
||||
}
|
||||
|
||||
// InjectInboundPacketBuffer makes the Wrapper device behave as if a packet
|
||||
// InjectInboundPacketBuffer makes the [Wrapper] device behave as if a packet
|
||||
// (pkt) with the given contents was received from the network.
|
||||
// It takes ownership of one reference count on pkt. The injected
|
||||
// packet will not pass through inbound filters.
|
||||
//
|
||||
// pkt will be copied into buffs before writing to the underlying tun.Device.
|
||||
// Therefore, callers must allocate and pass a buffs slice that is sized
|
||||
// appropriately for holding pkt.Size() + PacketStartOffset as a single
|
||||
// element (buffs[0]) and split across multiple elements if the originating
|
||||
// stack supports GSO. sizes must be sized with similar consideration,
|
||||
// len(buffs) should be equal to len(sizes). If any len(buffs[<index>]) was
|
||||
// mutated by InjectInboundPacketBuffer it will be reset to cap(buffs[<index>])
|
||||
// before returning.
|
||||
// pkt will be copied into slab, and potentially GSO split, before writing to
|
||||
// the underlying [tun.Device]. Therefore, callers must allocate and pass a slab
|
||||
// slice that is sized for holding:
|
||||
//
|
||||
// pkt.Data.Size() +
|
||||
// (N * network headers len) +
|
||||
// ((N+1) * [WritePacketStartOffset])
|
||||
//
|
||||
// N is 1 when the originating netstack does not support GSO, otherwise N should
|
||||
// be [conn.IdealBatchSize]. Similarly, callers must pass packets and writeBufs
|
||||
// sized to N.
|
||||
//
|
||||
// [tun.ReadPacket] is the shape required by [tun.GSOSplit], but is
|
||||
// admittedly confusing in the context of the direction of our eventual
|
||||
// i/o op, [tun.Device.Write]. This is purely historical, as [tun.GSOSplit]
|
||||
// was initially only used post-[tun.Device.Read] heading out to the network,
|
||||
// but we are using it in the opposite direction here.
|
||||
//
|
||||
// This path is typically used to deliver synthesized packets to the
|
||||
// host networking stack.
|
||||
func (t *Wrapper) InjectInboundPacketBuffer(pkt *netstack_PacketBuffer, buffs [][]byte, sizes []int) error {
|
||||
func (t *Wrapper) InjectInboundPacketBuffer(pkt *netstack_PacketBuffer, slab []byte, packets []tun.ReadPacket, writeBufs [][]byte) error {
|
||||
if !buildfeatures.HasNetstack {
|
||||
panic("unreachable")
|
||||
}
|
||||
buf := buffs[0][PacketStartOffset:]
|
||||
buf := slab[WritePacketStartOffset:]
|
||||
|
||||
bufN := copy(buf, pkt.NetworkHeader().Slice())
|
||||
bufN += copy(buf[bufN:], pkt.TransportHeader().Slice())
|
||||
@@ -1351,7 +1367,7 @@ func (t *Wrapper) InjectInboundPacketBuffer(pkt *netstack_PacketBuffer, buffs []
|
||||
// TODO(jwhited): support GSO passthrough to t.tdev. If t.tdev supports
|
||||
// GSO we don't need to split here and coalesce inside wireguard-go,
|
||||
// we can pass a coalesced segment all the way through.
|
||||
n, err := tun.GSOSplit(buf, gso, buffs, sizes, PacketStartOffset)
|
||||
n, err := tun.GSOSplit(buf, gso, slab, packets, WritePacketStartOffset)
|
||||
if err != nil {
|
||||
if errors.Is(err, tun.ErrTooManySegments) {
|
||||
t.limitedLogf("InjectInboundPacketBuffer: GSO split overflows buffs")
|
||||
@@ -1359,15 +1375,10 @@ func (t *Wrapper) InjectInboundPacketBuffer(pkt *netstack_PacketBuffer, buffs []
|
||||
return err
|
||||
}
|
||||
}
|
||||
for i := range n {
|
||||
buffs[i] = buffs[i][:PacketStartOffset+sizes[i]]
|
||||
for i, meta := range packets[:n] {
|
||||
writeBufs[i] = slab[meta.Offset-WritePacketStartOffset : meta.Offset+meta.Size]
|
||||
}
|
||||
defer func() {
|
||||
for i := range n {
|
||||
buffs[i] = buffs[i][:cap(buffs[i])]
|
||||
}
|
||||
}()
|
||||
_, err = t.tdevWrite(buffs[:n], PacketStartOffset)
|
||||
_, err = t.tdevWrite(writeBufs[:n], WritePacketStartOffset)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -1377,7 +1388,7 @@ func (t *Wrapper) InjectInboundPacketBuffer(pkt *netstack_PacketBuffer, buffs []
|
||||
// The injected packet will not pass through inbound filters.
|
||||
//
|
||||
// The packet contents are to start at &buf[offset].
|
||||
// offset must be greater or equal to PacketStartOffset.
|
||||
// offset must be greater or equal to WritePacketStartOffset.
|
||||
// The space before &buf[offset] will be used by WireGuard.
|
||||
func (t *Wrapper) InjectInboundDirect(buf []byte, offset int) error {
|
||||
if len(buf) > MaxPacketSize {
|
||||
@@ -1386,7 +1397,7 @@ func (t *Wrapper) InjectInboundDirect(buf []byte, offset int) error {
|
||||
if len(buf) < offset {
|
||||
return errOffsetTooBig
|
||||
}
|
||||
if offset < PacketStartOffset {
|
||||
if offset < WritePacketStartOffset {
|
||||
return errOffsetTooSmall
|
||||
}
|
||||
|
||||
@@ -1408,10 +1419,10 @@ func (t *Wrapper) InjectInboundCopy(packet []byte) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
buf := make([]byte, PacketStartOffset+len(packet))
|
||||
copy(buf[PacketStartOffset:], packet)
|
||||
buf := make([]byte, WritePacketStartOffset+len(packet))
|
||||
copy(buf[WritePacketStartOffset:], packet)
|
||||
|
||||
return t.InjectInboundDirect(buf, PacketStartOffset)
|
||||
return t.InjectInboundDirect(buf, WritePacketStartOffset)
|
||||
}
|
||||
|
||||
func (t *Wrapper) injectOutboundPong(pp *packet.Parsed, req packet.TSMPPingRequest) {
|
||||
@@ -1486,7 +1497,7 @@ func (t *Wrapper) BatchSize() int {
|
||||
// Linux, and we cannot make a determination on gVisor usage at
|
||||
// wireguard-go.Device startup, which is when this value matters for
|
||||
// packet memory init.
|
||||
return batching.BatchSizeFromEnv(conn.IdealBatchSize)
|
||||
return conn.IdealBatchSize
|
||||
}
|
||||
return t.tdev.BatchSize()
|
||||
}
|
||||
|
||||
@@ -81,8 +81,8 @@ func probeTCPGRO(dev tun.GRODevice) error {
|
||||
ipAs4 := ipPort.Addr().As4()
|
||||
bufs := make([][]byte, 2)
|
||||
for i := range bufs {
|
||||
bufs[i] = make([]byte, PacketStartOffset+totalLen, PacketStartOffset+(totalLen*2))
|
||||
ipv4H := header.IPv4(bufs[i][PacketStartOffset:])
|
||||
bufs[i] = make([]byte, WritePacketStartOffset+totalLen, WritePacketStartOffset+(totalLen*2))
|
||||
ipv4H := header.IPv4(bufs[i][WritePacketStartOffset:])
|
||||
ipv4H.Encode(&header.IPv4Fields{
|
||||
SrcAddr: tcpip.AddrFromSlice(ipAs4[:]),
|
||||
DstAddr: tcpip.AddrFromSlice(ipAs4[:]),
|
||||
@@ -92,7 +92,7 @@ func probeTCPGRO(dev tun.GRODevice) error {
|
||||
TTL: 0,
|
||||
TotalLength: uint16(totalLen),
|
||||
})
|
||||
tcpH := header.TCP(bufs[i][PacketStartOffset+iphLen:])
|
||||
tcpH := header.TCP(bufs[i][WritePacketStartOffset+iphLen:])
|
||||
tcpH.Encode(&header.TCPFields{
|
||||
SrcPort: ipPort.Port(),
|
||||
DstPort: ipPort.Port(),
|
||||
@@ -102,12 +102,12 @@ func probeTCPGRO(dev tun.GRODevice) error {
|
||||
Flags: header.TCPFlagAck,
|
||||
WindowSize: 3000,
|
||||
})
|
||||
copy(bufs[i][PacketStartOffset+iphLen+tcphLen:], fingerprint)
|
||||
copy(bufs[i][WritePacketStartOffset+iphLen+tcphLen:], fingerprint)
|
||||
ipv4H.SetChecksum(^ipv4H.CalculateChecksum())
|
||||
pseudoCsum := header.PseudoHeaderChecksum(unix.IPPROTO_TCP, ipv4H.SourceAddress(), ipv4H.DestinationAddress(), uint16(tcphLen+segmentSize))
|
||||
pseudoCsum = checksum.Checksum(bufs[i][PacketStartOffset+iphLen+tcphLen:], pseudoCsum)
|
||||
pseudoCsum = checksum.Checksum(bufs[i][WritePacketStartOffset+iphLen+tcphLen:], pseudoCsum)
|
||||
tcpH.SetChecksum(^tcpH.CalculateChecksum(pseudoCsum))
|
||||
}
|
||||
_, err := dev.Write(bufs, PacketStartOffset)
|
||||
_, err := dev.Write(bufs, WritePacketStartOffset)
|
||||
return err
|
||||
}
|
||||
+31
-31
@@ -21,6 +21,7 @@ import (
|
||||
"github.com/gaissmai/bart"
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/google/go-cmp/cmp/cmpopts"
|
||||
wgtun "github.com/tailscale/wireguard-go/tun"
|
||||
"github.com/tailscale/wireguard-go/tun/tuntest"
|
||||
"go4.org/mem"
|
||||
"go4.org/netipx"
|
||||
@@ -199,6 +200,13 @@ func newFakeTUN(logf logger.Logf, bus *eventbus.Bus, secure bool) (*fakeTUN, *Wr
|
||||
return ftun.(*fakeTUN), tun
|
||||
}
|
||||
|
||||
// getSinglePacketReadArgs returns a slab and packets slice sized for passing to
|
||||
// [tun.Device.Read] where [tun.Device.BatchSize] returns 1.
|
||||
func getSinglePacketReadArgs() (slab []byte, packets []wgtun.ReadPacket) {
|
||||
return make([]byte, MaxPacketSize+(2*wgtun.ReadPacketSpacing)),
|
||||
make([]wgtun.ReadPacket, 1)
|
||||
}
|
||||
|
||||
func TestReadAndInject(t *testing.T) {
|
||||
bus := eventbustest.NewBus(t)
|
||||
chtun, tun := newChannelTUN(t.Logf, bus, false)
|
||||
@@ -225,21 +233,18 @@ func TestReadAndInject(t *testing.T) {
|
||||
}(packet)
|
||||
}
|
||||
|
||||
var buf [MaxPacketSize]byte
|
||||
seen := make(map[string]bool)
|
||||
sizes := make([]int, 1)
|
||||
slab, packets := getSinglePacketReadArgs()
|
||||
// We expect the same packets back, in no particular order.
|
||||
for i := range len(written) + len(injected) {
|
||||
packet := buf[:]
|
||||
buffs := [][]byte{packet}
|
||||
numPackets, err := tun.Read(buffs, sizes, 0)
|
||||
numPackets, err := tun.Read(slab, packets)
|
||||
if err != nil {
|
||||
t.Errorf("read %d: error: %v", i, err)
|
||||
}
|
||||
if numPackets != 1 {
|
||||
t.Fatalf("read %d packets, expected %d", numPackets, 1)
|
||||
}
|
||||
packet = packet[:sizes[0]]
|
||||
packet := slab[packets[0].Offset : packets[0].Offset+packets[0].Size]
|
||||
packetLen := len(packet)
|
||||
if packetLen != size {
|
||||
t.Errorf("read %d: got size %d; want %d", i, packetLen, size)
|
||||
@@ -373,7 +378,6 @@ func TestFilter(t *testing.T) {
|
||||
}
|
||||
}()
|
||||
|
||||
var buf [MaxPacketSize]byte
|
||||
var stats netlogtype.CountsByConnection
|
||||
tun.SetConnectionCounter(stats.Add)
|
||||
for _, tt := range tests {
|
||||
@@ -381,7 +385,6 @@ func TestFilter(t *testing.T) {
|
||||
var n int
|
||||
var err error
|
||||
var filtered bool
|
||||
sizes := make([]int, 1)
|
||||
|
||||
tunStats := stats.Clone()
|
||||
stats.Reset()
|
||||
@@ -399,8 +402,10 @@ func TestFilter(t *testing.T) {
|
||||
_, err = tun.Write([][]byte{tt.data}, 0)
|
||||
filtered = tun.lastActivityAtomic.LoadAtomic() == 0
|
||||
} else {
|
||||
chtun.Outbound <- tt.data
|
||||
n, err = tun.Read([][]byte{buf[:]}, sizes, 0)
|
||||
// chtun.Outbound is unbuffered, and won't be drained until the
|
||||
// first Read call.
|
||||
go func() { chtun.Outbound <- tt.data }()
|
||||
n, err = tun.Read(getSinglePacketReadArgs())
|
||||
// In the read direction, errors are fatal, so we return n = 0 instead.
|
||||
filtered = (n == 0)
|
||||
}
|
||||
@@ -487,9 +492,7 @@ func TestInjectOutboundRecordsUDPFlowState(t *testing.T) {
|
||||
|
||||
// Drain the injected packet via Read. This drives injectedRead, which
|
||||
// is what records the reverse-flow tuple in filter state.
|
||||
var buf [MaxPacketSize]byte
|
||||
sizes := make([]int, 1)
|
||||
if n, err := tun.Read([][]byte{buf[:]}, sizes, 0); err != nil {
|
||||
if n, err := tun.Read(getSinglePacketReadArgs()); err != nil {
|
||||
t.Fatalf("Read: %v", err)
|
||||
} else if n != 1 {
|
||||
t.Fatalf("Read returned %d packets, want 1", n)
|
||||
@@ -972,12 +975,9 @@ func TestCaptureHook(t *testing.T) {
|
||||
// Loop reading and discarding packets; this ensures that we don't have
|
||||
// packets stuck in vectorOutbound
|
||||
go func() {
|
||||
var (
|
||||
buf [MaxPacketSize]byte
|
||||
sizes = make([]int, 1)
|
||||
)
|
||||
slab, packets := getSinglePacketReadArgs()
|
||||
for {
|
||||
_, err := w.Read([][]byte{buf[:]}, sizes, 0)
|
||||
_, err := w.Read(slab, packets)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -992,10 +992,10 @@ func TestCaptureHook(t *testing.T) {
|
||||
packetBuf := stack.NewPacketBuffer(stack.PacketBufferOptions{
|
||||
Payload: buffer.MakeWithData([]byte("InjectInboundPacketBuffer")),
|
||||
})
|
||||
buffs := make([][]byte, 1)
|
||||
buffs[0] = make([]byte, PacketStartOffset+packetBuf.Size())
|
||||
sizes := make([]int, 1)
|
||||
w.InjectInboundPacketBuffer(packetBuf, buffs, sizes)
|
||||
slab := make([]byte, packetBuf.Size()+(2*WritePacketStartOffset))
|
||||
packets := make([]wgtun.ReadPacket, 1)
|
||||
writeBufs := make([][]byte, 1)
|
||||
w.InjectInboundPacketBuffer(packetBuf, slab, packets, writeBufs)
|
||||
|
||||
packetBuf = stack.NewPacketBuffer(stack.PacketBufferOptions{
|
||||
Payload: buffer.MakeWithData([]byte("InjectOutboundPacketBuffer")),
|
||||
@@ -1113,11 +1113,12 @@ func TestInterceptOrdering(t *testing.T) {
|
||||
tun.PreFilterPacketOutboundToWireGuardAppConnectorIntercept = orderedFilterFn(3)
|
||||
tun.PostFilterPacketOutboundToWireGuard = orderedFilterFn(4)
|
||||
|
||||
// Read the packet.
|
||||
var buf [MaxPacketSize]byte
|
||||
sizes := make([]int, 1)
|
||||
chtun.Outbound <- udp4("1.2.3.4", "5.6.7.8", 98, 98) // Simulate tun device sending.
|
||||
tun.Read([][]byte{buf[:]}, sizes, 0)
|
||||
// Read the packet. chtun.Outbound is unbuffered, and won't be drained until
|
||||
// first Read call.
|
||||
go func() {
|
||||
chtun.Outbound <- udp4("1.2.3.4", "5.6.7.8", 98, 98) // Simulate tun device sending.
|
||||
}()
|
||||
tun.Read(getSinglePacketReadArgs())
|
||||
|
||||
if seq != numOutboundIntercepts {
|
||||
t.Errorf("got number of intercepts run in Read(): %d; want: %d", seq, numOutboundIntercepts)
|
||||
@@ -1142,16 +1143,15 @@ func TestInjectedReadCallsAppConnectorHook(t *testing.T) {
|
||||
t.Fatalf("InjectOutbound error: %v", err)
|
||||
}
|
||||
|
||||
var buf [MaxPacketSize]byte
|
||||
sizes := make([]int, 1)
|
||||
tun.Read([][]byte{buf[:]}, sizes, 0)
|
||||
slab, packets := getSinglePacketReadArgs()
|
||||
tun.Read(slab, packets)
|
||||
|
||||
if !called {
|
||||
t.Error("app connector hook was not called in InjectOutbound")
|
||||
}
|
||||
|
||||
wantPkt := udp4("169.254.0.1", "100.25.63.57", 80, 12345)
|
||||
gotPkt := buf[:sizes[0]]
|
||||
gotPkt := slab[packets[0].Offset : packets[0].Offset+packets[0].Size]
|
||||
if !bytes.Equal(wantPkt, gotPkt) {
|
||||
t.Errorf("packet mismatch\nwant:\t% x\ngot:\t% x", wantPkt, gotPkt)
|
||||
}
|
||||
|
||||
+17
-26
@@ -24,7 +24,6 @@ import (
|
||||
|
||||
"go4.org/mem"
|
||||
"golang.org/x/crypto/blake2s"
|
||||
"golang.org/x/net/ipv6"
|
||||
"tailscale.com/control/controlknobs"
|
||||
"tailscale.com/disco"
|
||||
"tailscale.com/net/batching"
|
||||
@@ -569,13 +568,14 @@ type singlePacketConn struct {
|
||||
*net.UDPConn
|
||||
}
|
||||
|
||||
func (c *singlePacketConn) ReadBatch(msgs []ipv6.Message, _ int) (int, error) {
|
||||
n, ap, err := c.UDPConn.ReadFromUDPAddrPort(msgs[0].Buffers[0])
|
||||
func (c *singlePacketConn) ReadBatch(slab []byte, packets []batching.ReceivedPacket) (int, error) {
|
||||
n, ap, err := c.UDPConn.ReadFromUDPAddrPort(slab)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
msgs[0].N = n
|
||||
msgs[0].Addr = net.UDPAddrFromAddrPort(netaddr.Unmap(ap))
|
||||
packets[0].Offset = 0
|
||||
packets[0].Size = n
|
||||
packets[0].Source = netaddr.Unmap(ap)
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
@@ -692,7 +692,7 @@ func (s *Server) bindSockets(desiredPort uint16) error {
|
||||
break SocketsLoop
|
||||
}
|
||||
}
|
||||
pc := batching.TryUpgradeToConn(uc, network, batching.IdealBatchSize, "udprelay_rxq_overflows", s.controlKnobs)
|
||||
pc := batching.TryUpgradeToConn(uc, network, "udprelay_rxq_overflows", s.controlKnobs)
|
||||
bc, ok := pc.(batching.Conn)
|
||||
if !ok {
|
||||
bc = &singlePacketConn{uc}
|
||||
@@ -874,19 +874,11 @@ func (s *Server) packetReadLoop(readFromSocket, otherSocket batching.Conn, readF
|
||||
s.Close()
|
||||
}()
|
||||
|
||||
msgs := make([]ipv6.Message, batching.IdealBatchSize)
|
||||
for i := range msgs {
|
||||
msgs[i].OOB = make([]byte, batching.MinControlMessageSize())
|
||||
msgs[i].Buffers = make([][]byte, 1)
|
||||
msgs[i].Buffers[0] = make([]byte, 1<<16-1)
|
||||
}
|
||||
writeBuffsByDest := make(map[netip.AddrPort][][]byte, batching.IdealBatchSize)
|
||||
slab := make([]byte, 2*batching.ReadSlabMultiple)
|
||||
packets := make([]batching.ReceivedPacket, 2*batching.MinimumReadBatchSize)
|
||||
writeBuffsByDest := make(map[netip.AddrPort][][]byte, batching.MaximumWriteBatchSize)
|
||||
|
||||
for {
|
||||
for i := range msgs {
|
||||
msgs[i] = ipv6.Message{Buffers: msgs[i].Buffers, OOB: msgs[i].OOB[:cap(msgs[i].OOB)]}
|
||||
}
|
||||
|
||||
// TODO: extract laddr from IP_PKTINFO for use in reply
|
||||
// ReadBatch will split coalesced datagrams before returning, which
|
||||
// WriteBatchTo will re-coalesce further down. We _could_ be more
|
||||
@@ -894,7 +886,7 @@ func (s *Server) packetReadLoop(readFromSocket, otherSocket batching.Conn, readF
|
||||
// are non-control/handshake packets. We pay the memmove/memcopy
|
||||
// performance penalty for now in the interest of simple single packet
|
||||
// handlers.
|
||||
n, err := readFromSocket.ReadBatch(msgs, 0)
|
||||
n, err := readFromSocket.ReadBatch(slab, packets)
|
||||
if err != nil {
|
||||
s.logf("error reading from socket(%v): %v", readFromSocket.LocalAddr(), err)
|
||||
return
|
||||
@@ -907,13 +899,12 @@ func (s *Server) packetReadLoop(readFromSocket, otherSocket batching.Conn, readF
|
||||
bytes6 int64
|
||||
packets6 int64
|
||||
}{}
|
||||
for _, msg := range msgs[:n] {
|
||||
if msg.N == 0 {
|
||||
for _, packet := range packets[:n] {
|
||||
if packet.Size == 0 {
|
||||
continue
|
||||
}
|
||||
buf := msg.Buffers[0][:msg.N]
|
||||
from := msg.Addr.(*net.UDPAddr).AddrPort()
|
||||
write, to, isDataPacket := s.handlePacket(from, buf)
|
||||
buf := slab[packet.Offset : packet.Offset+packet.Size]
|
||||
write, to, isDataPacket := s.handlePacket(packet.Source, buf)
|
||||
if !to.IsValid() {
|
||||
continue
|
||||
}
|
||||
@@ -926,10 +917,10 @@ func (s *Server) packetReadLoop(readFromSocket, otherSocket batching.Conn, readF
|
||||
forwardedByOutAF.packets6++
|
||||
}
|
||||
}
|
||||
if from.Addr().Is4() == to.Addr().Is4() || otherSocket != nil {
|
||||
if packet.Source.Addr().Is4() == to.Addr().Is4() || otherSocket != nil {
|
||||
buffs, ok := writeBuffsByDest[to]
|
||||
if !ok {
|
||||
buffs = make([][]byte, 0, batching.IdealBatchSize)
|
||||
buffs = make([][]byte, 0, batching.MaximumWriteBatchSize)
|
||||
}
|
||||
buffs = append(buffs, write)
|
||||
writeBuffsByDest[to] = buffs
|
||||
@@ -939,7 +930,7 @@ func (s *Server) packetReadLoop(readFromSocket, otherSocket batching.Conn, readF
|
||||
// [server.handlePacket] has to see a packet from a particular
|
||||
// address family at least once in order for it to return a
|
||||
// packet to write towards a dest for the same address family.
|
||||
s.logf("[unexpected] packet from: %v produced packet to: %v while otherSocket is nil", from, to)
|
||||
s.logf("[unexpected] packet from: %v produced packet to: %v while otherSocket is nil", packet.Source, to)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+13
-19
@@ -28,7 +28,6 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
wgconn "github.com/tailscale/wireguard-go/conn"
|
||||
"github.com/tailscale/wireguard-go/device"
|
||||
"github.com/tailscale/wireguard-go/tun"
|
||||
"go4.org/netipx"
|
||||
@@ -944,11 +943,6 @@ func derpProbeBandwidthTUN(ctx context.Context, transferTimeSeconds, totalBytesT
|
||||
return fmt.Errorf("failed to configure tun: %w", err)
|
||||
}
|
||||
|
||||
// Depending on platform, we need some space for headers at the front
|
||||
// of TUN I/O op buffers. The below constant is more than enough space
|
||||
// for any platform that this might run on.
|
||||
tunStartOffset := device.MessageTransportHeaderSize
|
||||
|
||||
// This goroutine reads packets from the TUN device and evaluates if they
|
||||
// are IPv4 packets destined for loopback via DERP. If so, it performs L3 NAT
|
||||
// (swap src/dst) and writes them towards DERP in order to loopback via the
|
||||
@@ -958,25 +952,21 @@ func derpProbeBandwidthTUN(ctx context.Context, transferTimeSeconds, totalBytesT
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
numBufs := wgconn.IdealBatchSize
|
||||
bufs := make([][]byte, 0, numBufs)
|
||||
sizes := make([]int, numBufs)
|
||||
for range numBufs {
|
||||
bufs = append(bufs, make([]byte, mtu+tunStartOffset))
|
||||
}
|
||||
slab := make([]byte, 2*(1<<16-1)+(2*tun.ReadPacketSpacing))
|
||||
packets := make([]tun.ReadPacket, dev.BatchSize())
|
||||
|
||||
destinationAddrBytes := destinationAddr.AsSlice()
|
||||
scratch := make([]byte, 4)
|
||||
toDERPPubKey := toc.SelfPublicKey()
|
||||
for {
|
||||
n, err := dev.Read(bufs, sizes, tunStartOffset)
|
||||
n, err := dev.Read(slab, packets)
|
||||
if err != nil {
|
||||
tunReadErrC <- err
|
||||
return
|
||||
}
|
||||
|
||||
for i := range n {
|
||||
pkt := bufs[i][tunStartOffset : sizes[i]+tunStartOffset]
|
||||
for _, metadata := range packets[:n] {
|
||||
pkt := slab[metadata.Offset : metadata.Offset+metadata.Size]
|
||||
// Skip everything except valid IPv4 packets
|
||||
if len(pkt) < 20 {
|
||||
// Doesn't even have a full IPv4 header
|
||||
@@ -1011,7 +1001,11 @@ func derpProbeBandwidthTUN(ctx context.Context, transferTimeSeconds, totalBytesT
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
buf := make([]byte, mtu+tunStartOffset)
|
||||
// Depending on platform, we need some space for headers at the front
|
||||
// of TUN I/O op buffers. The below constant is more than enough space
|
||||
// for any platform that this might run on.
|
||||
tunWriteStartOffset := device.MessageTransportHeaderSize
|
||||
buf := make([]byte, mtu+tunWriteStartOffset)
|
||||
bufs := make([][]byte, 1)
|
||||
|
||||
fromDERPPubKey := fromc.SelfPublicKey()
|
||||
@@ -1028,9 +1022,9 @@ func derpProbeBandwidthTUN(ctx context.Context, transferTimeSeconds, totalBytesT
|
||||
return
|
||||
}
|
||||
pkt := v.Data
|
||||
copy(buf[tunStartOffset:], pkt)
|
||||
bufs[0] = buf[:len(pkt)+tunStartOffset]
|
||||
if _, err := dev.Write(bufs, tunStartOffset); err != nil {
|
||||
copy(buf[tunWriteStartOffset:], pkt)
|
||||
bufs[0] = buf[:len(pkt)+tunWriteStartOffset]
|
||||
if _, err := dev.Write(bufs, tunWriteStartOffset); err != nil {
|
||||
recvErrC <- fmt.Errorf("failed to write to TUN device: %w", err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -16,4 +16,4 @@
|
||||
) {
|
||||
src = ./.;
|
||||
}).shellNix
|
||||
# nix-direnv cache busting line: sha256-w7VuEoXsnL7wmHbKlwd6qgUqyRj09JAqHw073Zgnidc=
|
||||
# nix-direnv cache busting line: sha256-edmocquubPBqaakW8/TfQTLZS48HR7SOcWjYiLeJGLY=
|
||||
+1
-1
@@ -177,7 +177,7 @@ tailscale.com/tsnet dependencies: (generated by github.com/tailscale/depaware)
|
||||
tailscale.com/logtail/filch from tailscale.com/log/sockstatlog+
|
||||
tailscale.com/metrics from tailscale.com/tsweb+
|
||||
tailscale.com/net/bakedroots from tailscale.com/net/tlsdial+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock+
|
||||
💣 tailscale.com/net/batching from tailscale.com/wgengine/magicsock
|
||||
tailscale.com/net/dns from tailscale.com/ipn/ipnlocal+
|
||||
tailscale.com/net/dns/publicdns from tailscale.com/net/dns+
|
||||
tailscale.com/net/dns/resolvconffile from tailscale.com/net/dns+
|
||||
|
||||
+3
-2
@@ -2721,12 +2721,13 @@ func (t *chanTUN) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *chanTUN) Read(bufs [][]byte, sizes []int, offset int) (int, error) {
|
||||
func (t *chanTUN) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
|
||||
select {
|
||||
case <-t.closed:
|
||||
return 0, io.EOF
|
||||
case pkt := <-t.Outbound:
|
||||
sizes[0] = copy(bufs[0][offset:], pkt)
|
||||
packets[0].Offset = tun.ReadPacketSpacing
|
||||
packets[0].Size = copy(slab[tun.ReadPacketSpacing:len(slab)-tun.ReadPacketSpacing], pkt)
|
||||
return 1, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -48,16 +48,12 @@ import (
|
||||
// http://go/ios-memory or `go tool pprof` the heap) or, if the growth is
|
||||
// genuinely necessary and accounted for, raise the budget in the same change
|
||||
// with a justification.
|
||||
//
|
||||
// Note that these tests run with TS_DEBUG_WG_BATCH_SIZE=1 (see
|
||||
// measureNodeCost), so wireguard-go packet-pool memory is measured at its
|
||||
// mobile configuration, not the Linux server default.
|
||||
const (
|
||||
// startupBudget is the live-heap budget for bringing up a backend
|
||||
// connected to a tailnet with zero peers: feature/extension init,
|
||||
// engine, netstack, DNS, control client, etc.
|
||||
//
|
||||
// As of 2026-08 the measured cost is ~1.3 MiB; the budget leaves ~6x
|
||||
// As of 2026-09 the measured cost is ~1.73 MiB; the budget leaves ~5x
|
||||
// headroom for organic growth while still catching multi-MiB
|
||||
// regressions (the conn25 flow table pre-allocation this test was
|
||||
// written for measured 17 MiB here).
|
||||
@@ -135,15 +131,6 @@ func measureNodeCost(t *testing.T, peers int) uint64 {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Use the minimum wireguard-go batch size, matching iOS and other
|
||||
// memory-constrained platforms (which are what these budgets protect).
|
||||
// Without this, on Linux each of wireguard-go's four reader goroutines
|
||||
// (three UDP receive paths plus the TUN reader) pins batch-size (128) ×
|
||||
// 64 KiB message buffers at startup: ~32 MiB of live heap that exists
|
||||
// only in the Linux configuration and would drown out the regressions
|
||||
// this test exists to catch.
|
||||
t.Setenv("TS_DEBUG_WG_BATCH_SIZE", "1")
|
||||
|
||||
controlURL := startControl(t, peers)
|
||||
baseline := liveHeap()
|
||||
s := startNode(t, ctx, controlURL)
|
||||
@@ -192,6 +179,6 @@ func TestBackendPerPeerMemory(t *testing.T) {
|
||||
float64(base)/(1<<20), nPeers, float64(loaded)/(1<<20),
|
||||
float64(perPeer)/(1<<10), float64(perPeerBudget)/(1<<10))
|
||||
if perPeer > perPeerBudget {
|
||||
t.Errorf("per-peer live heap %v exceeds budget %v", perPeer, int(perPeerBudget))
|
||||
t.Errorf("per-peer live heap %v exceeds budget %v", perPeer, perPeerBudget)
|
||||
}
|
||||
}
|
||||
+9
-10
@@ -165,16 +165,15 @@ func (t *sourceTun) Write(b [][]byte, ofs int) (int, error) {
|
||||
return len(b), nil
|
||||
}
|
||||
|
||||
func (t *sourceTun) Read(b [][]byte, sizes []int, ofs int) (int, error) {
|
||||
for i, b := range b {
|
||||
// Continually generate "input" packets
|
||||
n := t.traf.Generate(b, ofs)
|
||||
sizes[i] = n
|
||||
if n == 0 {
|
||||
return 0, io.EOF
|
||||
}
|
||||
func (t *sourceTun) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
|
||||
// Continually generate "input" packets.
|
||||
n := t.traf.Generate(slab, tun.ReadPacketSpacing)
|
||||
packets[0].Size = n
|
||||
packets[0].Offset = tun.ReadPacketSpacing
|
||||
if n == 0 {
|
||||
return 0, io.EOF
|
||||
}
|
||||
return len(b), nil
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
type sinkTun struct {
|
||||
@@ -189,7 +188,7 @@ func (t *sinkTun) Flush() error { return nil }
|
||||
func (t *sinkTun) MTU() (int, error) { return 1500, nil }
|
||||
func (t *sinkTun) Name() (string, error) { return "sink", nil }
|
||||
|
||||
func (t *sinkTun) Read(b [][]byte, sizes []int, ofs int) (int, error) {
|
||||
func (t *sinkTun) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
|
||||
// Never returns
|
||||
select {}
|
||||
}
|
||||
|
||||
@@ -708,7 +708,7 @@ func (c *Conn) runDerpWriter(ctx context.Context, dc *derphttp.Client, ch <-chan
|
||||
}
|
||||
}
|
||||
|
||||
func (c *connBind) receiveDERP(buffs [][]byte, sizes []int, eps []conn.Endpoint) (int, error) {
|
||||
func (c *connBind) receiveDERP(slab []byte, packets []conn.ReceivedPacket) (int, error) {
|
||||
if s := c.Conn.health.ReceiveFuncStats(health.ReceiveDERP); s != nil {
|
||||
s.Enter()
|
||||
defer s.Exit()
|
||||
@@ -718,13 +718,14 @@ func (c *connBind) receiveDERP(buffs [][]byte, sizes []int, eps []conn.Endpoint)
|
||||
if c.isClosed() {
|
||||
break
|
||||
}
|
||||
n, ep := c.processDERPReadResult(dm, buffs[0])
|
||||
n, ep := c.processDERPReadResult(dm, slab)
|
||||
if n == 0 {
|
||||
// No data read occurred. Wait for another packet.
|
||||
continue
|
||||
}
|
||||
sizes[0] = n
|
||||
eps[0] = ep
|
||||
packets[0].Size = n
|
||||
packets[0].Offset = 0
|
||||
packets[0].Endpoint = ep
|
||||
return 1, nil
|
||||
}
|
||||
return 0, net.ErrClosed
|
||||
|
||||
@@ -28,7 +28,6 @@ import (
|
||||
"github.com/tailscale/wireguard-go/conn"
|
||||
"github.com/tailscale/wireguard-go/device"
|
||||
"go4.org/mem"
|
||||
"golang.org/x/net/ipv6"
|
||||
"tailscale.com/control/controlknobs"
|
||||
"tailscale.com/disco"
|
||||
"tailscale.com/envknob"
|
||||
@@ -193,8 +192,6 @@ type Conn struct {
|
||||
pconn4 RebindingUDPConn
|
||||
pconn6 RebindingUDPConn
|
||||
|
||||
receiveBatchPool sync.Pool
|
||||
|
||||
// closeDisco4 and closeDisco6 are io.Closers to shut down the raw
|
||||
// disco packet receivers. If nil, no raw disco receiver is
|
||||
// running for the given family.
|
||||
@@ -593,17 +590,6 @@ func newConn(logf logger.Logf) *Conn {
|
||||
}
|
||||
c.discoAtomic.Set(discoPrivate)
|
||||
c.bind = &connBind{Conn: c, closed: true}
|
||||
c.receiveBatchPool = sync.Pool{New: func() any {
|
||||
msgs := make([]ipv6.Message, c.bind.BatchSize())
|
||||
for i := range msgs {
|
||||
msgs[i].Buffers = make([][]byte, 1)
|
||||
msgs[i].OOB = make([]byte, batching.MinControlMessageSize())
|
||||
}
|
||||
batch := &receiveBatch{
|
||||
msgs: msgs,
|
||||
}
|
||||
return batch
|
||||
}}
|
||||
c.muCond = sync.NewCond(&c.mu)
|
||||
c.networkUp.Store(true) // assume up until told otherwise
|
||||
return c
|
||||
@@ -1705,26 +1691,6 @@ func (c *Conn) sendAddr(addr netip.AddrPort, pubKey key.NodePublic, b []byte, is
|
||||
return false, errDropDerpPacket
|
||||
}
|
||||
|
||||
type receiveBatch struct {
|
||||
msgs []ipv6.Message
|
||||
}
|
||||
|
||||
func (c *Conn) getReceiveBatchForBuffs(buffs [][]byte) *receiveBatch {
|
||||
batch := c.receiveBatchPool.Get().(*receiveBatch)
|
||||
for i := range buffs {
|
||||
batch.msgs[i].Buffers[0] = buffs[i]
|
||||
batch.msgs[i].OOB = batch.msgs[i].OOB[:cap(batch.msgs[i].OOB)]
|
||||
}
|
||||
return batch
|
||||
}
|
||||
|
||||
func (c *Conn) putReceiveBatch(batch *receiveBatch) {
|
||||
for i := range batch.msgs {
|
||||
batch.msgs[i] = ipv6.Message{Buffers: batch.msgs[i].Buffers, OOB: batch.msgs[i].OOB}
|
||||
}
|
||||
c.receiveBatchPool.Put(batch)
|
||||
}
|
||||
|
||||
func (c *Conn) receiveIPv4() conn.ReceiveFunc {
|
||||
return c.mkReceiveFunc(&c.pconn4, c.health.ReceiveFuncStats(health.ReceiveIPv4),
|
||||
&c.metrics.inboundPacketsIPv4Total,
|
||||
@@ -1749,8 +1715,9 @@ func (c *Conn) receiveIPv6() conn.ReceiveFunc {
|
||||
func (c *Conn) mkReceiveFunc(ruc *RebindingUDPConn, healthItem *health.ReceiveFuncStats, directPacketMetric, peerRelayPacketMetric, directBytesMetric, peerRelayBytesMetric *expvar.Int) conn.ReceiveFunc {
|
||||
// epCache caches an epAddr->endpoint for hot flows.
|
||||
var epCache epAddrEndpointCache
|
||||
var batchingPackets []batching.ReceivedPacket
|
||||
|
||||
return func(buffs [][]byte, sizes []int, eps []conn.Endpoint) (_ int, retErr error) {
|
||||
return func(slab []byte, packets []conn.ReceivedPacket) (_ int, retErr error) {
|
||||
if buildfeatures.HasHealth && healthItem != nil {
|
||||
healthItem.Enter()
|
||||
defer healthItem.Exit()
|
||||
@@ -1763,11 +1730,12 @@ func (c *Conn) mkReceiveFunc(ruc *RebindingUDPConn, healthItem *health.ReceiveFu
|
||||
if ruc == nil {
|
||||
panic("nil RebindingUDPConn")
|
||||
}
|
||||
if len(batchingPackets) != len(packets) {
|
||||
batchingPackets = make([]batching.ReceivedPacket, len(packets))
|
||||
}
|
||||
|
||||
batch := c.getReceiveBatchForBuffs(buffs)
|
||||
defer c.putReceiveBatch(batch)
|
||||
for {
|
||||
numMsgs, err := ruc.ReadBatch(batch.msgs[:len(buffs)], 0)
|
||||
numMsgs, err := ruc.ReadBatch(slab, batchingPackets)
|
||||
if err != nil {
|
||||
if neterror.PacketWasTruncated(err) {
|
||||
continue
|
||||
@@ -1776,33 +1744,33 @@ func (c *Conn) mkReceiveFunc(ruc *RebindingUDPConn, healthItem *health.ReceiveFu
|
||||
}
|
||||
|
||||
reportToCaller := false
|
||||
for i, msg := range batch.msgs[:numMsgs] {
|
||||
if msg.N == 0 {
|
||||
sizes[i] = 0
|
||||
for i, batchingPacket := range batchingPackets[:numMsgs] {
|
||||
if batchingPacket.Size == 0 {
|
||||
packets[i].Size = 0
|
||||
continue
|
||||
}
|
||||
ipp := msg.Addr.(*net.UDPAddr).AddrPort()
|
||||
if ep, size, isGeneveEncap, ok := c.receiveIP(msg.Buffers[0][:msg.N], ipp, &epCache); ok {
|
||||
buf := slab[batchingPacket.Offset : batchingPacket.Offset+batchingPacket.Size]
|
||||
packets[i].Size = batchingPacket.Size
|
||||
packets[i].Offset = batchingPacket.Offset
|
||||
if isGeneveEncap, ok := c.receiveIP(buf, batchingPacket.Source, &epCache, &packets[i]); ok {
|
||||
if isGeneveEncap {
|
||||
if peerRelayPacketMetric != nil {
|
||||
peerRelayPacketMetric.Add(1)
|
||||
}
|
||||
if peerRelayBytesMetric != nil {
|
||||
peerRelayBytesMetric.Add(int64(msg.N))
|
||||
peerRelayBytesMetric.Add(int64(len(buf)))
|
||||
}
|
||||
} else {
|
||||
if directPacketMetric != nil {
|
||||
directPacketMetric.Add(1)
|
||||
}
|
||||
if directBytesMetric != nil {
|
||||
directBytesMetric.Add(int64(msg.N))
|
||||
directBytesMetric.Add(int64(len(buf)))
|
||||
}
|
||||
}
|
||||
eps[i] = ep
|
||||
sizes[i] = size
|
||||
reportToCaller = true
|
||||
} else {
|
||||
sizes[i] = 0
|
||||
packets[i].Size = 0
|
||||
}
|
||||
}
|
||||
if reportToCaller {
|
||||
@@ -1821,17 +1789,17 @@ func looksLikeInitiationMsg(b []byte) bool {
|
||||
|
||||
// receiveIP is the shared bits of ReceiveIPv4 and ReceiveIPv6.
|
||||
//
|
||||
// size is the length of 'b' to report up to wireguard-go (only relevant if
|
||||
// 'ok' is true).
|
||||
// rp.Size & rp.Offset are pre-populated by callers. receiveIP must keep them
|
||||
// up to date if b's length is mutated, and ok will be true. receiveIP must set
|
||||
// rp.Endpoint if ok is true.
|
||||
//
|
||||
// isGeneveEncap is whether 'b' is encapsulated by a Geneve header (only
|
||||
// relevant if 'ok' is true).
|
||||
//
|
||||
// ok is whether this read should be reported up to wireguard-go (our
|
||||
// caller).
|
||||
func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCache) (_ conn.Endpoint, size int, isGeneveEncap bool, ok bool) {
|
||||
func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCache, rp *conn.ReceivedPacket) (isGeneveEncap bool, ok bool) {
|
||||
var ep *endpoint
|
||||
size = len(b)
|
||||
|
||||
var geneve packet.GeneveHeader
|
||||
pt, isGeneveEncap := packetLooksLike(b)
|
||||
@@ -1842,7 +1810,7 @@ func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCach
|
||||
// Decode only returns an error when 'b' is too short, and
|
||||
// 'isGeneveEncap' indicates it's a sufficient length.
|
||||
c.logf("[unexpected] geneve header decoding error: %v", err)
|
||||
return nil, 0, false, false
|
||||
return false, false
|
||||
}
|
||||
src.vni = geneve.VNI
|
||||
}
|
||||
@@ -1857,10 +1825,10 @@ func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCach
|
||||
// [disco.MessageType], but we assert it should be handshake-related.
|
||||
shouldByRelayHandshakeMsg := geneve.Control == true
|
||||
c.handleDiscoMessage(b, src, shouldByRelayHandshakeMsg, key.NodePublic{}, discoRXPathUDP)
|
||||
return nil, 0, false, false
|
||||
return false, false
|
||||
case packetLooksLikeSTUNBinding:
|
||||
c.netChecker.ReceiveSTUNPacket(b, ipp)
|
||||
return nil, 0, false, false
|
||||
return false, false
|
||||
default:
|
||||
// Fall through for all other packet types as they are assumed to
|
||||
// be potentially WireGuard.
|
||||
@@ -1870,7 +1838,7 @@ func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCach
|
||||
// If we have no private key, we're logged out or
|
||||
// stopped. Don't try to pass these wireguard packets
|
||||
// up to wireguard-go; it'll just complain (issue 1167).
|
||||
return nil, 0, false, false
|
||||
return false, false
|
||||
}
|
||||
|
||||
// geneveInclusivePacketLen holds the packet length prior to any potential
|
||||
@@ -1879,13 +1847,10 @@ func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCach
|
||||
if src.vni.IsSet() {
|
||||
// Strip away the Geneve header before returning the packet to
|
||||
// wireguard-go.
|
||||
//
|
||||
// TODO(jwhited): update [github.com/tailscale/wireguard-go/conn.ReceiveFunc]
|
||||
// to support returning start offset in order to get rid of this memmove perf
|
||||
// penalty.
|
||||
size = copy(b, b[packet.GeneveFixedHeaderLength:])
|
||||
b = b[:size]
|
||||
b = b[packet.GeneveFixedHeaderLength:]
|
||||
rp.Offset += packet.GeneveFixedHeaderLength
|
||||
}
|
||||
rp.Size = len(b)
|
||||
|
||||
if cache.epAddr == src && cache.de != nil && cache.gen == cache.de.numStopAndReset() {
|
||||
ep = cache.de
|
||||
@@ -1896,7 +1861,8 @@ func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCach
|
||||
if !ok {
|
||||
// TODO(jwhited): reuse [lazyEndpoint] across calls to receiveIP()
|
||||
// for the same batch & [epAddr] src.
|
||||
return &lazyEndpoint{c: c, src: src}, size, isGeneveEncap, true
|
||||
rp.Endpoint = &lazyEndpoint{c: c, src: src}
|
||||
return isGeneveEncap, true
|
||||
}
|
||||
cache.epAddr = src
|
||||
cache.de = de
|
||||
@@ -1917,9 +1883,11 @@ func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *epAddrEndpointCach
|
||||
// unlucky and fail to JIT configure the "correct" peer.
|
||||
// TODO(jwhited): relax this to include direct connections
|
||||
// See http://go/corp/29422 & http://go/corp/30042
|
||||
return &lazyEndpoint{c: c, maybeEP: ep, src: src}, size, isGeneveEncap, true
|
||||
rp.Endpoint = &lazyEndpoint{c: c, maybeEP: ep, src: src}
|
||||
return isGeneveEncap, true
|
||||
}
|
||||
return ep, size, isGeneveEncap, true
|
||||
rp.Endpoint = ep
|
||||
return isGeneveEncap, true
|
||||
}
|
||||
|
||||
// discoLogLevel controls the verbosity of discovery log messages.
|
||||
@@ -3483,7 +3451,7 @@ func (c *connBind) BatchSize() int {
|
||||
// TODO(raggi): determine by properties rather than hardcoding platform behavior
|
||||
switch runtime.GOOS {
|
||||
case "linux":
|
||||
return batching.BatchSizeFromEnv(conn.IdealBatchSize)
|
||||
return conn.IdealBatchSize
|
||||
default:
|
||||
return 1
|
||||
}
|
||||
@@ -3731,13 +3699,13 @@ func (c *Conn) bindSocket(ruc *RebindingUDPConn, network string, curPortFate cur
|
||||
defer ruc.mu.Unlock()
|
||||
|
||||
if runtime.GOOS == "js" {
|
||||
ruc.setConnLocked(newBlockForeverConn(), "", c.bind.BatchSize(), c.controlKnobs)
|
||||
ruc.setConnLocked(newBlockForeverConn(), "", c.controlKnobs)
|
||||
return nil
|
||||
}
|
||||
|
||||
if debugAlwaysDERP() {
|
||||
c.logf("disabled %v per TS_DEBUG_ALWAYS_USE_DERP", network)
|
||||
ruc.setConnLocked(newBlockForeverConn(), "", c.bind.BatchSize(), c.controlKnobs)
|
||||
ruc.setConnLocked(newBlockForeverConn(), "", c.controlKnobs)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -3796,7 +3764,7 @@ func (c *Conn) bindSocket(ruc *RebindingUDPConn, network string, curPortFate cur
|
||||
if debugBindSocket() {
|
||||
c.logf("magicsock: bindSocket: successfully listened %v port %d", network, port)
|
||||
}
|
||||
ruc.setConnLocked(pconn, network, c.bind.BatchSize(), c.controlKnobs)
|
||||
ruc.setConnLocked(pconn, network, c.controlKnobs)
|
||||
if network == "udp4" {
|
||||
c.health.SetUDP4Unbound(false)
|
||||
}
|
||||
@@ -3807,7 +3775,7 @@ func (c *Conn) bindSocket(ruc *RebindingUDPConn, network string, curPortFate cur
|
||||
// Set pconn to a dummy conn whose reads block until closed.
|
||||
// This keeps the receive funcs alive for a future in which
|
||||
// we get a link change and we can try binding again.
|
||||
ruc.setConnLocked(newBlockForeverConn(), "", c.bind.BatchSize(), c.controlKnobs)
|
||||
ruc.setConnLocked(newBlockForeverConn(), "", c.controlKnobs)
|
||||
if network == "udp4" {
|
||||
c.health.SetUDP4Unbound(true)
|
||||
}
|
||||
|
||||
@@ -49,6 +49,7 @@ import (
|
||||
"tailscale.com/envknob"
|
||||
"tailscale.com/health"
|
||||
"tailscale.com/ipn/ipnstate"
|
||||
"tailscale.com/net/batching"
|
||||
"tailscale.com/net/netaddr"
|
||||
"tailscale.com/net/netcheck"
|
||||
"tailscale.com/net/netmon"
|
||||
@@ -511,13 +512,11 @@ func TestNewConn(t *testing.T) {
|
||||
conn.SetPrivateKey(key.NewNode())
|
||||
|
||||
go func() {
|
||||
pkts := make([][]byte, 1)
|
||||
sizes := make([]int, 1)
|
||||
eps := make([]wgconn.Endpoint, 1)
|
||||
pkts[0] = make([]byte, 64<<10)
|
||||
slab := make([]byte, batching.ReadSlabMultiple)
|
||||
packets := make([]wgconn.ReceivedPacket, batching.MinimumReadBatchSize)
|
||||
receiveIPv4 := conn.receiveIPv4()
|
||||
for {
|
||||
_, err := receiveIPv4(pkts, sizes, eps)
|
||||
_, err := receiveIPv4(slab, packets)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -1683,16 +1682,14 @@ func setUpReceiveFrom(tb testing.TB) (roundTrip func()) {
|
||||
for i := range sendBuf {
|
||||
sendBuf[i] = 'x'
|
||||
}
|
||||
buffs := make([][]byte, 1)
|
||||
buffs[0] = make([]byte, 2<<10)
|
||||
sizes := make([]int, 1)
|
||||
eps := make([]wgconn.Endpoint, 1)
|
||||
slab := make([]byte, batching.ReadSlabMultiple)
|
||||
packets := make([]wgconn.ReceivedPacket, batching.MinimumReadBatchSize)
|
||||
receiveIPv4 := conn.receiveIPv4()
|
||||
return func() {
|
||||
if _, err := sendConn.WriteTo(sendBuf, dstAddr); err != nil {
|
||||
tb.Fatalf("WriteTo: %v", err)
|
||||
}
|
||||
n, err := receiveIPv4(buffs, sizes, eps)
|
||||
n, err := receiveIPv4(slab, packets)
|
||||
if err != nil {
|
||||
tb.Fatal(err)
|
||||
}
|
||||
@@ -1919,13 +1916,11 @@ func TestRebindStress(t *testing.T) {
|
||||
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
buffs := make([][]byte, 1)
|
||||
sizes := make([]int, 1)
|
||||
eps := make([]wgconn.Endpoint, 1)
|
||||
buffs[0] = make([]byte, 1500)
|
||||
slab := make([]byte, batching.ReadSlabMultiple)
|
||||
packets := make([]wgconn.ReceivedPacket, batching.MinimumReadBatchSize)
|
||||
receiveIPv4 := conn.receiveIPv4()
|
||||
for {
|
||||
_, err := receiveIPv4(buffs, sizes, eps)
|
||||
_, err := receiveIPv4(slab, packets)
|
||||
if ctx.Err() != nil {
|
||||
errc <- nil
|
||||
return
|
||||
@@ -2356,8 +2351,8 @@ func TestRebindingUDPConn(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer realConn.Close()
|
||||
c.setConnLocked(realConn.(nettype.PacketConn), "udp4", 1, nil)
|
||||
c.setConnLocked(newBlockForeverConn(), "", 1, nil)
|
||||
c.setConnLocked(realConn.(nettype.PacketConn), "udp4", nil)
|
||||
c.setConnLocked(newBlockForeverConn(), "", nil)
|
||||
}
|
||||
|
||||
// https://github.com/tailscale/tailscale/issues/6680: don't ignore
|
||||
@@ -4219,7 +4214,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
// [*lazyEndpoint] and [*lazyEndpoint.maybeEP] is non-nil, we expect
|
||||
// got.maybeEP to also be non-nil. Must not be reused across tests.
|
||||
wantEndpointType wgconn.Endpoint
|
||||
wantSize int
|
||||
wantIsGeneveEncap bool
|
||||
wantOk bool
|
||||
wantMetricInc *clientmetric.Metric
|
||||
@@ -4230,7 +4224,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
ipp: netip.MustParseAddrPort("127.0.0.1:7777"),
|
||||
cache: &epAddrEndpointCache{},
|
||||
wantEndpointType: nil,
|
||||
wantSize: 0,
|
||||
wantIsGeneveEncap: false,
|
||||
wantOk: false,
|
||||
wantMetricInc: metricRecvDiscoBadPeer,
|
||||
@@ -4241,7 +4234,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
ipp: netip.MustParseAddrPort("127.0.0.1:7777"),
|
||||
cache: &epAddrEndpointCache{},
|
||||
wantEndpointType: nil,
|
||||
wantSize: 0,
|
||||
wantIsGeneveEncap: false,
|
||||
wantOk: false,
|
||||
wantMetricInc: metricRecvDiscoBadPeer,
|
||||
@@ -4252,7 +4244,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
ipp: netip.MustParseAddrPort("127.0.0.1:7777"),
|
||||
cache: &epAddrEndpointCache{},
|
||||
wantEndpointType: nil,
|
||||
wantSize: 0,
|
||||
wantIsGeneveEncap: false,
|
||||
wantOk: false,
|
||||
wantMetricInc: findMetricByName("netcheck_stun_recv_ipv4"),
|
||||
@@ -4263,7 +4254,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
ipp: netip.MustParseAddrPort("127.0.0.1:7777"),
|
||||
cache: &epAddrEndpointCache{},
|
||||
wantEndpointType: &lazyEndpoint{},
|
||||
wantSize: len(looksLikeNakedWireGuardInit),
|
||||
wantIsGeneveEncap: false,
|
||||
wantOk: true,
|
||||
wantMetricInc: nil,
|
||||
@@ -4276,7 +4266,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
insertWantEndpointTypeInPeerMap: true,
|
||||
peerMapEpAddr: epAddr{ap: netip.MustParseAddrPort("127.0.0.1:7777")},
|
||||
wantEndpointType: newPeerMapInsertableEndpoint(0),
|
||||
wantSize: len(looksLikeNakedWireGuardInit),
|
||||
wantIsGeneveEncap: false,
|
||||
wantOk: true,
|
||||
wantMetricInc: nil,
|
||||
@@ -4287,7 +4276,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
ipp: netip.MustParseAddrPort("127.0.0.1:7777"),
|
||||
cache: &epAddrEndpointCache{},
|
||||
wantEndpointType: &lazyEndpoint{},
|
||||
wantSize: len(looksLikeGeneveWireGuardInit) - packet.GeneveFixedHeaderLength,
|
||||
wantIsGeneveEncap: true,
|
||||
wantOk: true,
|
||||
wantMetricInc: nil,
|
||||
@@ -4302,7 +4290,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
wantEndpointType: &lazyEndpoint{
|
||||
maybeEP: newPeerMapInsertableEndpoint(0),
|
||||
},
|
||||
wantSize: len(looksLikeGeneveWireGuardInit) - packet.GeneveFixedHeaderLength,
|
||||
wantIsGeneveEncap: true,
|
||||
wantOk: true,
|
||||
wantMetricInc: nil,
|
||||
@@ -4317,7 +4304,6 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
wantEndpointType: &lazyEndpoint{
|
||||
maybeEP: newPeerMapInsertableEndpoint(mono.Now().Add(time.Hour * 24)),
|
||||
},
|
||||
wantSize: len(looksLikeGeneveWireGuardInit) - packet.GeneveFixedHeaderLength,
|
||||
wantIsGeneveEncap: true,
|
||||
wantOk: true,
|
||||
wantMetricInc: nil,
|
||||
@@ -4356,44 +4342,66 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
c.peerMap.setNodeKeyForEpAddr(tt.peerMapEpAddr, insertEPIntoPeerMap.publicKey)
|
||||
}
|
||||
|
||||
// Allow the same input packet to be used across tests, receiveIP()
|
||||
// may mutate.
|
||||
inputPacket := make([]byte, len(tt.b))
|
||||
copy(inputPacket, tt.b)
|
||||
const initialOffset = 13
|
||||
slab := make([]byte, initialOffset+len(tt.b))
|
||||
copy(slab[initialOffset:], tt.b)
|
||||
before := bytes.Clone(slab)
|
||||
|
||||
got, gotSize, gotIsGeneveEncap, gotOk := c.receiveIP(inputPacket, tt.ipp, tt.cache)
|
||||
if (tt.wantEndpointType == nil) != (got == nil) {
|
||||
t.Errorf("receiveIP() (tt.wantEndpointType == nil): %v != (got == nil): %v", tt.wantEndpointType == nil, got == nil)
|
||||
rp := wgconn.ReceivedPacket{Size: len(tt.b), Offset: initialOffset}
|
||||
gotIsGeneveEncap, gotOk := c.receiveIP(rp.Bytes(slab), tt.ipp, tt.cache, &rp)
|
||||
gotEndpoint := rp.Endpoint
|
||||
if (tt.wantEndpointType == nil) != (gotEndpoint == nil) {
|
||||
t.Errorf("receiveIP() (tt.wantEndpointType == nil): %v != (got == nil): %v", tt.wantEndpointType == nil, gotEndpoint == nil)
|
||||
}
|
||||
if tt.wantEndpointType != nil && reflect.TypeOf(got).String() != reflect.TypeOf(tt.wantEndpointType).String() {
|
||||
t.Errorf("receiveIP() got = %v, want %v", reflect.TypeOf(got).String(), reflect.TypeOf(tt.wantEndpointType).String())
|
||||
if tt.wantEndpointType != nil && reflect.TypeOf(gotEndpoint).String() != reflect.TypeOf(tt.wantEndpointType).String() {
|
||||
t.Errorf("receiveIP() got = %v, want %v", reflect.TypeOf(gotEndpoint).String(), reflect.TypeOf(tt.wantEndpointType).String())
|
||||
} else {
|
||||
switch ep := tt.wantEndpointType.(type) {
|
||||
case *endpoint:
|
||||
if ep != got.(*endpoint) {
|
||||
t.Errorf("receiveIP() want [*endpoint]: %p != got [*endpoint]: %p", ep, got)
|
||||
if ep != gotEndpoint.(*endpoint) {
|
||||
t.Errorf("receiveIP() want [*endpoint]: %p != got [*endpoint]: %p", ep, gotEndpoint)
|
||||
}
|
||||
case *lazyEndpoint:
|
||||
if ep.maybeEP != nil && ep.maybeEP != got.(*lazyEndpoint).maybeEP {
|
||||
t.Errorf("receiveIP() want [*lazyEndpoint.maybeEP]: %p != got [*lazyEndpoint.maybeEP] %p", ep, got)
|
||||
if ep.maybeEP != nil && ep.maybeEP != gotEndpoint.(*lazyEndpoint).maybeEP {
|
||||
t.Errorf("receiveIP() want [*lazyEndpoint.maybeEP]: %p != got [*lazyEndpoint.maybeEP] %p", ep, gotEndpoint)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if gotSize != tt.wantSize {
|
||||
t.Errorf("receiveIP() gotSize = %v, want %v", gotSize, tt.wantSize)
|
||||
}
|
||||
if gotIsGeneveEncap != tt.wantIsGeneveEncap {
|
||||
t.Errorf("receiveIP() gotIsGeneveEncap = %v, want %v", gotIsGeneveEncap, tt.wantIsGeneveEncap)
|
||||
}
|
||||
if gotOk != tt.wantOk {
|
||||
t.Errorf("receiveIP() gotOk = %v, want %v", gotOk, tt.wantOk)
|
||||
}
|
||||
|
||||
if tt.wantOk {
|
||||
wantOffset := initialOffset
|
||||
wantBytes := tt.b
|
||||
if tt.wantIsGeneveEncap {
|
||||
wantOffset += packet.GeneveFixedHeaderLength
|
||||
wantBytes = tt.b[packet.GeneveFixedHeaderLength:]
|
||||
}
|
||||
|
||||
if rp.Offset != wantOffset {
|
||||
t.Errorf("receiveIP() Offset = %d, want %d", rp.Offset, wantOffset)
|
||||
}
|
||||
if rp.Size != len(wantBytes) {
|
||||
t.Errorf("receiveIP() Size = %d, want %d", rp.Size, len(wantBytes))
|
||||
}
|
||||
if got := rp.Bytes(slab); !bytes.Equal(got, wantBytes) {
|
||||
t.Errorf("receiveIP() bytes = %x, want %x", got, wantBytes)
|
||||
}
|
||||
if !bytes.Equal(slab, before) {
|
||||
t.Error("receiveIP() mutated the receive slab")
|
||||
}
|
||||
}
|
||||
|
||||
if tt.wantMetricInc != nil && tt.wantMetricInc.Value() != metricBefore+1 {
|
||||
t.Errorf("receiveIP() metric %v not incremented", tt.wantMetricInc.Name())
|
||||
}
|
||||
if tt.cache.de != nil {
|
||||
switch ep := got.(type) {
|
||||
switch ep := gotEndpoint.(type) {
|
||||
case *endpoint:
|
||||
if tt.cache.de != ep {
|
||||
t.Errorf("receiveIP() cache populated with [*endpoint] %p, want %p", tt.cache.de, ep)
|
||||
@@ -4419,10 +4427,7 @@ func TestConn_receiveIP(t *testing.T) {
|
||||
wantNonzeroRxStats = true
|
||||
}
|
||||
if tt.wantOk && wantNonzeroRxStats {
|
||||
wantRxBytes := uint64(tt.wantSize)
|
||||
if tt.wantIsGeneveEncap {
|
||||
wantRxBytes += packet.GeneveFixedHeaderLength
|
||||
}
|
||||
wantRxBytes := uint64(len(tt.b))
|
||||
wantPhy := map[netlogtype.Connection]netlogtype.Counts{
|
||||
{Dst: tt.ipp}: {
|
||||
RxPackets: 1,
|
||||
|
||||
@@ -11,7 +11,6 @@ import (
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
|
||||
"golang.org/x/net/ipv6"
|
||||
"tailscale.com/control/controlknobs"
|
||||
"tailscale.com/net/batching"
|
||||
"tailscale.com/net/netaddr"
|
||||
@@ -43,8 +42,8 @@ type RebindingUDPConn struct {
|
||||
// intentionally pushed closest to where read/write ops occur in order to avoid
|
||||
// disrupting surrounding code that assumes nettype.PacketConn is a
|
||||
// *net.UDPConn. knobs may be nil.
|
||||
func (c *RebindingUDPConn) setConnLocked(p nettype.PacketConn, network string, batchSize int, knobs *controlknobs.Knobs) {
|
||||
upc := batching.TryUpgradeToConn(p, network, batchSize, "magicsock_udp_rxq_overflows", knobs)
|
||||
func (c *RebindingUDPConn) setConnLocked(p nettype.PacketConn, network string, knobs *controlknobs.Knobs) {
|
||||
upc := batching.TryUpgradeToConn(p, network, "magicsock_udp_rxq_overflows", knobs)
|
||||
c.pconn = upc
|
||||
c.pconnAtomic.Store(&upc)
|
||||
c.port = uint16(c.localAddrLocked().Port)
|
||||
@@ -119,20 +118,21 @@ func (c *RebindingUDPConn) WriteWireGuardBatchTo(buffs [][]byte, addr epAddr, of
|
||||
|
||||
// ReadBatch is an alias for [batching.Conn.ReadBatch] with fallback to single
|
||||
// packet operations if c.pconn is not a [batching.Conn].
|
||||
func (c *RebindingUDPConn) ReadBatch(msgs []ipv6.Message, flags int) (int, error) {
|
||||
func (c *RebindingUDPConn) ReadBatch(slab []byte, batchingPackets []batching.ReceivedPacket) (int, error) {
|
||||
for {
|
||||
pconn := *c.pconnAtomic.Load()
|
||||
b, ok := pconn.(batching.Conn)
|
||||
if !ok {
|
||||
n, ap, err := c.readFromWithInitPconn(pconn, msgs[0].Buffers[0])
|
||||
n, ap, err := c.readFromWithInitPconn(pconn, slab)
|
||||
if err == nil {
|
||||
msgs[0].N = n
|
||||
msgs[0].Addr = net.UDPAddrFromAddrPort(netaddr.Unmap(ap))
|
||||
batchingPackets[0].Offset = 0
|
||||
batchingPackets[0].Size = n
|
||||
batchingPackets[0].Source = netaddr.Unmap(ap)
|
||||
return 1, nil
|
||||
}
|
||||
return 0, err
|
||||
}
|
||||
n, err := b.ReadBatch(msgs, flags)
|
||||
n, err := b.ReadBatch(slab, batchingPackets)
|
||||
if err != nil && pconn != c.currentConn() {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/tailscale/wireguard-go/conn"
|
||||
"github.com/tailscale/wireguard-go/tun"
|
||||
"gvisor.dev/gvisor/pkg/refs"
|
||||
"gvisor.dev/gvisor/pkg/tcpip"
|
||||
"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
|
||||
@@ -1020,32 +1021,27 @@ func (ns *Impl) DialContextUDPWithBind(ctx context.Context, localAddr netip.Addr
|
||||
return gonet.DialUDP(ns.ipstack, localAddress, remoteAddress, ipType)
|
||||
}
|
||||
|
||||
// getInjectInboundBuffsSizes returns packet memory and a sizes slice for usage
|
||||
// when calling tstun.Wrapper.InjectInboundPacketBuffer(). These are sized with
|
||||
// consideration for MTU and GSO support on ns.linkEP. They should be recycled
|
||||
// getInjectInboundPacketSlab returns packet memory and related descriptors to
|
||||
// be used when calling [tstun.Wrapper.InjectInboundPacketBuffer]. The returned
|
||||
// elements are sized with consideration for MTU and GSO support on [Impl.linkEP],
|
||||
// and inter-packet spacing requirements of [tun.GSOSplit]. It should be recycled
|
||||
// across subsequent inbound packet injection calls.
|
||||
func (ns *Impl) getInjectInboundBuffsSizes() (buffs [][]byte, sizes []int) {
|
||||
batchSize := 1
|
||||
gsoEnabled := ns.linkEP.SupportedGSO() == stack.HostGSOSupported
|
||||
if gsoEnabled {
|
||||
batchSize = conn.IdealBatchSize
|
||||
func (ns *Impl) getInjectInboundPacketSlab() (slab []byte, packets []tun.ReadPacket, writeBufs [][]byte) {
|
||||
payloadLen := int(tstun.DefaultTUNMTU())
|
||||
packetCount := 1
|
||||
if ns.linkEP.SupportedGSO() == stack.HostGSOSupported {
|
||||
payloadLen = 2 * int(ns.linkEP.GSOMaxSize())
|
||||
packetCount = conn.IdealBatchSize
|
||||
}
|
||||
buffs = make([][]byte, batchSize)
|
||||
sizes = make([]int, batchSize)
|
||||
for i := 0; i < batchSize; i++ {
|
||||
if i == 0 && gsoEnabled {
|
||||
buffs[i] = make([]byte, tstun.PacketStartOffset+ns.linkEP.GSOMaxSize())
|
||||
} else {
|
||||
buffs[i] = make([]byte, tstun.PacketStartOffset+tstun.DefaultTUNMTU())
|
||||
}
|
||||
}
|
||||
return buffs, sizes
|
||||
return make([]byte, tstun.WritePacketStartOffset+payloadLen+tstun.WritePacketStartOffset),
|
||||
make([]tun.ReadPacket, packetCount),
|
||||
make([][]byte, packetCount)
|
||||
}
|
||||
|
||||
// The inject goroutine reads in packets that netstack generated, and delivers
|
||||
// them to the correct path.
|
||||
func (ns *Impl) inject() {
|
||||
inboundBuffs, inboundBuffsSizes := ns.getInjectInboundBuffsSizes()
|
||||
inboundSlab, packets, writeBufs := ns.getInjectInboundPacketSlab()
|
||||
for {
|
||||
pkt := ns.linkEP.ReadContext(ns.ctx)
|
||||
if pkt == nil {
|
||||
@@ -1071,7 +1067,7 @@ func (ns *Impl) inject() {
|
||||
// pkt has a non-zero refcount, so injection methods takes
|
||||
// ownership of one count and will decrement on completion.
|
||||
if sendToHost {
|
||||
if err := ns.tundev.InjectInboundPacketBuffer(pkt, inboundBuffs, inboundBuffsSizes); err != nil {
|
||||
if err := ns.tundev.InjectInboundPacketBuffer(pkt, inboundSlab, packets, writeBufs); err != nil {
|
||||
ns.logf("netstack inject inbound: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -100,7 +100,7 @@ func (t *nilTun) MTU() (int, error) { return 1420, nil }
|
||||
func (t *nilTun) Name() (string, error) { return "niltun", nil }
|
||||
func (t *nilTun) Events() <-chan tun.Event { return t.events }
|
||||
|
||||
func (t *nilTun) Read(data [][]byte, sizes []int, offset int) (int, error) {
|
||||
func (t *nilTun) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
|
||||
<-t.closed
|
||||
return 0, io.EOF
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user