diff --git a/cmd/k8s-operator/depaware.txt b/cmd/k8s-operator/depaware.txt index 76b90e956..b8b7e7612 100644 --- a/cmd/k8s-operator/depaware.txt +++ b/cmd/k8s-operator/depaware.txt @@ -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+ diff --git a/cmd/tailscaled/depaware-min.txt b/cmd/tailscaled/depaware-min.txt index 1cdfa8395..8b9803b39 100644 --- a/cmd/tailscaled/depaware-min.txt +++ b/cmd/tailscaled/depaware-min.txt @@ -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+ diff --git a/cmd/tailscaled/depaware-minbox.txt b/cmd/tailscaled/depaware-minbox.txt index f8b3f4859..9f9da0ad9 100644 --- a/cmd/tailscaled/depaware-minbox.txt +++ b/cmd/tailscaled/depaware-minbox.txt @@ -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+ diff --git a/cmd/tsidp/depaware.txt b/cmd/tsidp/depaware.txt index 2905a37bb..62312238f 100644 --- a/cmd/tsidp/depaware.txt +++ b/cmd/tsidp/depaware.txt @@ -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+ diff --git a/feature/conn25/datapath_test.go b/feature/conn25/datapath_test.go index 2505bb3da..546a7e15f 100644 --- a/feature/conn25/datapath_test.go +++ b/feature/conn25/datapath_test.go @@ -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) } diff --git a/feature/tap/tap_linux.go b/feature/tap/tap_linux.go index e66f74ba4..2b4210761 100644 --- a/feature/tap/tap_linux.go +++ b/feature/tap/tap_linux.go @@ -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 } diff --git a/feature/tundevstats/tundevstats_linux_test.go b/feature/tundevstats/tundevstats_linux_test.go index 05468039b..f87a287f4 100644 --- a/feature/tundevstats/tundevstats_linux_test.go +++ b/feature/tundevstats/tundevstats_linux_test.go @@ -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 diff --git a/flake.nix b/flake.nix index 9abd91fdc..94e665483 100644 --- a/flake.nix +++ b/flake.nix @@ -173,4 +173,4 @@ }); }; } -# nix-direnv cache busting line: sha256-w7VuEoXsnL7wmHbKlwd6qgUqyRj09JAqHw073Zgnidc= +# nix-direnv cache busting line: sha256-edmocquubPBqaakW8/TfQTLZS48HR7SOcWjYiLeJGLY= \ No newline at end of file diff --git a/flakehashes.json b/flakehashes.json index 3aeefb3b7..9b8902e10 100644 --- a/flakehashes.json +++ b/flakehashes.json @@ -4,7 +4,7 @@ "sri": "sha256-+cTlDLusJthxTayVx6yGoqZHRN4ylvs/9Qpy4P2dlDI=" }, "vendor": { - "goModSum": "sha256-aYPi3BUkgTUZ3ESpK33W2CtHruPSgb7gMJlAP/tUL0I=", - "sri": "sha256-w7VuEoXsnL7wmHbKlwd6qgUqyRj09JAqHw073Zgnidc=" + "goModSum": "sha256-PGvkYIOJdO2bTiy4Rsp+82idPQuvqqPW5XSFutoEaqQ=", + "sri": "sha256-edmocquubPBqaakW8/TfQTLZS48HR7SOcWjYiLeJGLY=" } } diff --git a/go.mod b/go.mod index 65eb35638..68d5e8012 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 4b3d02f52..c3f608daa 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/net/batching/conn.go b/net/batching/conn.go index dfe44763d..5c573aad0 100644 --- a/net/batching/conn.go +++ b/net/batching/conn.go @@ -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. diff --git a/net/batching/conn_default.go b/net/batching/conn_default.go index 922a6a358..c9208523d 100644 --- a/net/batching/conn_default.go +++ b/net/batching/conn_default.go @@ -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 diff --git a/net/batching/conn_linux.go b/net/batching/conn_linux.go index f094f4b65..df4c7ce94 100644 --- a/net/batching/conn_linux.go +++ b/net/batching/conn_linux.go @@ -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 diff --git a/net/batching/conn_linux_test.go b/net/batching/conn_linux_test.go index 633e5d607..25e8d8b46 100644 --- a/net/batching/conn_linux_test.go +++ b/net/batching/conn_linux_test.go @@ -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)) } diff --git a/net/tstun/fake.go b/net/tstun/fake.go index f7925116e..d6a981576 100644 --- a/net/tstun/fake.go +++ b/net/tstun/fake.go @@ -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 } diff --git a/net/tstun/wrap.go b/net/tstun/wrap.go index 171347f6f..523cb1d65 100644 --- a/net/tstun/wrap.go +++ b/net/tstun/wrap.go @@ -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[]) was -// mutated by InjectInboundPacketBuffer it will be reset to cap(buffs[]) -// 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() } diff --git a/net/tstun/wrap_linux.go b/net/tstun/wrap_linux.go index bb193266f..3fbeb3525 100644 --- a/net/tstun/wrap_linux.go +++ b/net/tstun/wrap_linux.go @@ -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 } diff --git a/net/tstun/wrap_test.go b/net/tstun/wrap_test.go index 76edf7563..876288b42 100644 --- a/net/tstun/wrap_test.go +++ b/net/tstun/wrap_test.go @@ -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) } diff --git a/net/udprelay/server.go b/net/udprelay/server.go index 9052873b0..f846eb68d 100644 --- a/net/udprelay/server.go +++ b/net/udprelay/server.go @@ -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) } } diff --git a/prober/derp.go b/prober/derp.go index 0e9ecf49c..ddb7647d2 100644 --- a/prober/derp.go +++ b/prober/derp.go @@ -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 } diff --git a/shell.nix b/shell.nix index 29c9dd9c2..2d839da4e 100644 --- a/shell.nix +++ b/shell.nix @@ -16,4 +16,4 @@ ) { src = ./.; }).shellNix -# nix-direnv cache busting line: sha256-w7VuEoXsnL7wmHbKlwd6qgUqyRj09JAqHw073Zgnidc= +# nix-direnv cache busting line: sha256-edmocquubPBqaakW8/TfQTLZS48HR7SOcWjYiLeJGLY= diff --git a/tsnet/depaware.txt b/tsnet/depaware.txt index c174e31a9..9e16e8722 100644 --- a/tsnet/depaware.txt +++ b/tsnet/depaware.txt @@ -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+ diff --git a/tsnet/tsnet_test.go b/tsnet/tsnet_test.go index efbfb2d2a..bd242fca3 100644 --- a/tsnet/tsnet_test.go +++ b/tsnet/tsnet_test.go @@ -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 } } diff --git a/tstest/membudget/membudget_test.go b/tstest/membudget/membudget_test.go index 39dc23997..e33baa64b 100644 --- a/tstest/membudget/membudget_test.go +++ b/tstest/membudget/membudget_test.go @@ -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) } } diff --git a/wgengine/bench/wg.go b/wgengine/bench/wg.go index 8d2ad0451..bfffc3f15 100644 --- a/wgengine/bench/wg.go +++ b/wgengine/bench/wg.go @@ -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 {} } diff --git a/wgengine/magicsock/derp.go b/wgengine/magicsock/derp.go index 1209ea54c..020116228 100644 --- a/wgengine/magicsock/derp.go +++ b/wgengine/magicsock/derp.go @@ -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 diff --git a/wgengine/magicsock/magicsock.go b/wgengine/magicsock/magicsock.go index b42ccad6a..89b3bce96 100644 --- a/wgengine/magicsock/magicsock.go +++ b/wgengine/magicsock/magicsock.go @@ -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) } diff --git a/wgengine/magicsock/magicsock_test.go b/wgengine/magicsock/magicsock_test.go index bf705c80a..9619ee2c9 100644 --- a/wgengine/magicsock/magicsock_test.go +++ b/wgengine/magicsock/magicsock_test.go @@ -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, diff --git a/wgengine/magicsock/rebinding_conn.go b/wgengine/magicsock/rebinding_conn.go index c361a6b09..aaad1d6be 100644 --- a/wgengine/magicsock/rebinding_conn.go +++ b/wgengine/magicsock/rebinding_conn.go @@ -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 } diff --git a/wgengine/netstack/netstack.go b/wgengine/netstack/netstack.go index 17ba3dfa6..28da7c06e 100644 --- a/wgengine/netstack/netstack.go +++ b/wgengine/netstack/netstack.go @@ -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 } diff --git a/wgengine/wgcfg/device_test.go b/wgengine/wgcfg/device_test.go index 1a0222057..4c94eb25e 100644 --- a/wgengine/wgcfg/device_test.go +++ b/wgengine/wgcfg/device_test.go @@ -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 }