diff --git a/.github/workflows/natlab-test.yml b/.github/workflows/natlab-test.yml index 4f53c4ce4..60f4070a8 100644 --- a/.github/workflows/natlab-test.yml +++ b/.github/workflows/natlab-test.yml @@ -102,7 +102,7 @@ jobs: # single-test-per-matrix-job model. They stay runnable locally. run: | set -euo pipefail - exclude='^(TestGrid)$' + exclude='^(TestGrid|TestVnetPerf.*)$' tmp=$(mktemp) for pkg_dir in tstest/natlab/vmtest tstest/integration/nat; do pkg="./${pkg_dir}/" diff --git a/tstest/natlab/vmtest/cloudinit.go b/tstest/natlab/vmtest/cloudinit.go index 841ee4ad6..7d460e1ba 100644 --- a/tstest/natlab/vmtest/cloudinit.go +++ b/tstest/natlab/vmtest/cloudinit.go @@ -165,6 +165,20 @@ func (e *Env) generateFreeBSDUserData(n *Node) string { // traffic goes through the vnet NICs. The debug NIC is only for SSH. ud.WriteString(" - \"route delete default 10.0.2.2 2>/dev/null || true\"\n") + // Grow the TCP socket buffer limits before the binary downloads + // below. FreeBSD's defaults start receive windows at 64 kB and + // autoscale in slow 16 kB steps, which caps a transfer at roughly + // 64kB per round trip. On an oversubscribed CI runner, where + // scheduling delay inflates the effective RTT of the userspace vnet + // data path to tens or hundreds of milliseconds, that works out to + // a few hundred kB/s and made the multi-megabyte binary fetches (and + // so the whole test) time out. + ud.WriteString(" - \"sysctl kern.ipc.maxsockbuf=16777216\"\n") + ud.WriteString(" - \"sysctl net.inet.tcp.recvspace=4194304\"\n") + ud.WriteString(" - \"sysctl net.inet.tcp.sendspace=1048576\"\n") + ud.WriteString(" - \"sysctl net.inet.tcp.recvbuf_max=16777216\"\n") + ud.WriteString(" - \"sysctl net.inet.tcp.sendbuf_max=16777216\"\n") + // Download binaries from the files.tailscale VIP (52.52.0.6). // FreeBSD's fetch(1) is part of the base system (no curl needed). // Retry in a loop since the file server may not be ready immediately. diff --git a/tstest/natlab/vmtest/qemu.go b/tstest/natlab/vmtest/qemu.go index 73b265078..47f4191aa 100644 --- a/tstest/natlab/vmtest/qemu.go +++ b/tstest/natlab/vmtest/qemu.go @@ -28,6 +28,9 @@ import ( // platforms (macOS, etc.) TCG is used, which allows the tests to run // without a same-architecture hypervisor at the cost of speed. func qemuAccelArgs() []string { + if os.Getenv("VMTEST_NO_KVM") == "1" { + return nil + } if runtime.GOOS == "linux" { if f, err := os.OpenFile("/dev/kvm", os.O_RDWR, 0); err == nil { f.Close() @@ -133,12 +136,13 @@ func (e *Env) startGokrazyQEMU(n *Node) error { } // Add network devices — one per NIC. + // rx_queue_size=1024: see the comment in startCloudQEMU. for i := range n.vnetNode.NumNICs() { mac := n.vnetNode.NICMac(i) netdevID := fmt.Sprintf("net%d", i) args = append(args, "-netdev", fmt.Sprintf("stream,id=%s,addr.type=unix,addr.path=%s", netdevID, e.sockAddr), - "-device", fmt.Sprintf("virtio-net-device,netdev=%s,mac=%s", netdevID, mac), + "-device", fmt.Sprintf("virtio-net-device,netdev=%s,mac=%s,rx_queue_size=1024", netdevID, mac), ) } @@ -180,12 +184,15 @@ func (e *Env) startCloudQEMU(n *Node) error { // Add network devices — one per NIC. // romfile="" disables the iPXE option ROM entirely, saving ~5s per NIC at boot // and avoiding "duplicate fw_cfg file name" errors with multiple NICs. + // rx_queue_size=1024 (up from the 256 default) gives the guest 4x the + // virtio RX ring capacity, absorbing vnet bursts that would otherwise be + // dropped while the guest is descheduled on a contended host. for i := range n.vnetNode.NumNICs() { mac := n.vnetNode.NICMac(i) netdevID := fmt.Sprintf("net%d", i) args = append(args, "-netdev", fmt.Sprintf("stream,id=%s,addr.type=unix,addr.path=%s", netdevID, e.sockAddr), - "-device", fmt.Sprintf("virtio-net-pci,netdev=%s,mac=%s,romfile=", netdevID, mac), + "-device", fmt.Sprintf("virtio-net-pci,netdev=%s,mac=%s,romfile=,rx_queue_size=1024", netdevID, mac), ) } diff --git a/tstest/natlab/vmtest/vnetperf_test.go b/tstest/natlab/vmtest/vnetperf_test.go new file mode 100644 index 000000000..5143aa0d4 --- /dev/null +++ b/tstest/natlab/vmtest/vnetperf_test.go @@ -0,0 +1,61 @@ +// Copyright (c) Tailscale Inc & contributors +// SPDX-License-Identifier: BSD-3-Clause + +package vmtest + +import ( + "flag" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "tailscale.com/tstest/natlab/vnet" +) + +var ( + runPerfTests = flag.Bool("run-perf-tests", false, "run performance measurement tests that are not pass/fail regression tests; wasteful to run in CI") + perfPCAP = flag.String("perf-pcap", "", "if non-empty, write a pcap of vnet traffic to this file during TestVnetPerfFreeBSDDownload") + perfLatency = flag.Duration("perf-latency", 0, "simulated latency to add to the vnet network in TestVnetPerfFreeBSDDownload") +) + +// TestVnetPerfFreeBSDDownload is a benchmark harness for vnet TCP +// throughput, opt-in via --run-perf-tests (in addition to +// --run-vm-tests). It boots a single FreeBSD VM on one vnet network and +// waits only for its TTA agent to connect, which requires the VM to have +// downloaded tailscaled, tailscale, and tta from the vnet's +// files.tailscale VIP. The elapsed time is dominated by that download, +// so the test duration is the benchmark metric. +func TestVnetPerfFreeBSDDownload(t *testing.T) { + if !*runPerfTests { + t.Skip("skipping perf test; set --run-perf-tests to run") + } + env := New(t) + + lan := env.AddNetwork("2.1.1.1", "192.168.1.1/24", vnet.EasyNAT) + if *perfPCAP != "" { + env.cfg.SetPCAPFile(*perfPCAP) + } + if *perfLatency > 0 { + lan.SetLatency(*perfLatency) + } + env.AddNode("fbsd", lan, OS(FreeBSD150), DontJoinTailnet()) + + t0 := time.Now() + env.Start() + t.Logf("Start took %v (boot + binary downloads over vnet)", time.Since(t0).Round(time.Second)) + + // Surface fetch(1)'s final progress lines (size, rate, duration per + // binary) from the serial console log. + console, err := os.ReadFile(filepath.Join(env.tempDir, "fbsd.log")) + if err != nil { + t.Logf("reading console log: %v", err) + return + } + for line := range strings.Lines(string(console)) { + if strings.Contains(line, "Bps") { + t.Logf("fetch: %s", strings.TrimSpace(line)) + } + } +} diff --git a/tstest/natlab/vnet/pcap.go b/tstest/natlab/vnet/pcap.go index 3a766b375..34870fd29 100644 --- a/tstest/natlab/vnet/pcap.go +++ b/tstest/natlab/vnet/pcap.go @@ -39,10 +39,14 @@ func (p *pcapWriter) WritePacket(ci gopacket.CaptureInfo, data []byte) error { if p.w == nil { return io.ErrClosedPipe } + // Flush per packet so the file is readable if the process dies + // mid-test, but don't fsync: an fsync per packet serializes every + // packet behind a disk write and caps vnet TCP throughput at a few + // hundred kB/s (each data segment pays a couple of milliseconds + // before the receiver can ACK it). return do( func() error { return p.w.WritePacket(ci, data) }, p.w.Flush, - p.f.Sync, ) } diff --git a/tstest/natlab/vnet/vnet.go b/tstest/natlab/vnet/vnet.go index 98bdf2973..4e643c6f9 100644 --- a/tstest/natlab/vnet/vnet.go +++ b/tstest/natlab/vnet/vnet.go @@ -34,6 +34,7 @@ import ( "net" "net/http" "net/netip" + "os" "os/exec" "strconv" "strings" @@ -154,7 +155,43 @@ func (n *network) initStack() error { if tcpipErr != nil { return fmt.Errorf("SetTransportProtocolOption SACK: %v", tcpipErr) } - n.linkEP = channel.New(512, 1500, tcpip.LinkAddress(n.mac.HWAddr())) + // Raise the TCP buffer limits (defaults: 1 MB send, 1 MB receive) + // so that netstack-terminated connections (the fake control plane, + // DERP, log catcher, file servers) can keep a large window's worth + // of data in flight. In slow environments (oversubscribed CI + // runners) the effective RTT of the userspace data path reaches + // tens or hundreds of milliseconds, and throughput is capped at + // window/RTT. + // + // The send buffer default deliberately stays at 1 MB: the send + // buffer caps how much un-ACKed data one connection can burst into + // the QEMU socket and the guest's virtio RX ring. Bursting more + // than the downstream path can buffer mass-drops segments, and + // netstack's loss recovery handles wide holes so poorly (one or two + // segments per 200 ms RTO, for minutes) that transfers effectively + // wedge. 1 MB of in-flight data fits within the socket buffer plus + // the (enlarged, see qemu.go) virtio ring, and still allows 10 MB/s + // at a 100 ms effective RTT. + sndBufOpt := tcpip.TCPSendBufferSizeRangeOption{Min: 4 << 10, Default: 1 << 20, Max: 4 << 20} + if err := n.ns.SetTransportProtocolOption(tcp.ProtocolNumber, &sndBufOpt); err != nil { + return fmt.Errorf("SetTransportProtocolOption send buf: %v", err) + } + rcvBufOpt := tcpip.TCPReceiveBufferSizeRangeOption{Min: 4 << 10, Default: 4 << 20, Max: 16 << 20} + if err := n.ns.SetTransportProtocolOption(tcp.ProtocolNumber, &rcvBufOpt); err != nil { + return fmt.Errorf("SetTransportProtocolOption recv buf: %v", err) + } + // Enable receive buffer moderation (auto-tuning) so idle + // connections don't hold the full 4 MB. + modRcvBufOpt := tcpip.TCPModerateReceiveBufferOption(true) + if err := n.ns.SetTransportProtocolOption(tcp.ProtocolNumber, &modRcvBufOpt); err != nil { + return fmt.Errorf("SetTransportProtocolOption moderate recv buf: %v", err) + } + // The queue is sized to hold a full TCP send buffer's worth of + // 1500-byte frames (see the send buffer sizing above) so that a + // burst from one netstack connection can't overflow it; overflow + // here is silent packet loss. It's a channel of pointers, so the + // memory cost of the headroom is trivial. + n.linkEP = channel.New(4096, 1500, tcpip.LinkAddress(n.mac.HWAddr())) if tcpipProblem := n.ns.CreateNIC(nicID, n.linkEP); tcpipProblem != nil { return fmt.Errorf("CreateNIC: %v", tcpipProblem) } @@ -280,7 +317,12 @@ func (n *network) handleIPPacketFromGvisor(ipRaw []byte) { // where the primary MAC may be on a different network). mac := node.macForNet(n) if nw, ok := n.writers.Load(mac); ok { - nw.write(resPkt) + // conditionedWrite (rather than nw.write) so that the network's + // simulated latency and packet loss also apply to traffic + // originating from the router's own netstack (the fake control + // plane, DERP, DNS, file servers, etc), not just to forwarded + // node-to-node traffic. + n.conditionedWrite(nw, resPkt) } else { n.logf("gvisor write: no writeFunc for %v (node %v on net %v)", mac, node, n.mac) } @@ -296,6 +338,32 @@ func netaddrIPFromNetstackIP(s tcpip.Address) netip.Addr { return netip.Addr{} } +// debugSampleTCPInfo periodically logs the TCP sender state (cwnd, RTO, +// RTT, congestion state) of ep plus stack-wide TCP counters, for +// debugging vnet throughput. Enabled by VNET_TCP_DEBUG=1. It returns +// when the endpoint leaves the established state. +func (n *network) debugSampleTCPInfo(ep tcpip.Endpoint) { + st := n.ns.Stats().TCP + for { + time.Sleep(500 * time.Millisecond) + var info tcpip.TCPInfoOption + if err := ep.GetSockOpt(&info); err != nil { + log.Printf("tcpdebug: GetSockOpt: %v", err) + return + } + log.Printf("tcpdebug: state=%v cc=%v cwnd=%v ssthresh=%v rtt=%v rttvar=%v rto=%v reorderSeen=%v | stack: retrans=%v rtoTimeouts=%v fastRetrans=%v fastRecovery=%v sackRecovery=%v spuriousRecovery=%v sendErrs=%v qDrops=%v", + info.State, info.CcState, info.SndCwnd, info.SndSsthresh, + info.RTT.Round(time.Microsecond), info.RTTVar.Round(time.Microsecond), info.RTO, + info.ReorderSeen, + st.Retransmits.Value(), st.Timeouts.Value(), st.FastRetransmit.Value(), + st.FastRecovery.Value(), st.SACKRecovery.Value(), st.SpuriousRecovery.Value(), + st.SegmentSendErrors.Value(), n.ns.Stats().DroppedPackets.Value()) + if info.State != tcpip.EndpointState(tcp.StateEstablished) { + return + } + } +} + func stringifyTEI(tei stack.TransportEndpointID) string { localHostPort := net.JoinHostPort(tei.LocalAddress.String(), strconv.Itoa(int(tei.LocalPort))) remoteHostPort := net.JoinHostPort(tei.RemoteAddress.String(), strconv.Itoa(int(tei.RemotePort))) @@ -447,6 +515,9 @@ func (n *network) acceptTCP(r *tcp.ForwarderRequest) { r.Complete(false) tc := gonet.NewTCPConn(&wq, ep) context.AfterFunc(n.s.shutdownCtx, func() { tc.SetDeadline(time.Now()) }) + if os.Getenv("VNET_TCP_DEBUG") == "1" { + go n.debugSampleTCPInfo(ep) + } hs := &http.Server{Handler: n.s.fileServerHandler()} n.s.wg.Go(func() { hs.Serve(netutil.NewOneConnListener(tc, nil)) @@ -865,9 +936,13 @@ type Server struct { fakeACME *fakeACMEServer pcapWriter *pcapWriter - // writeMu serializes all writes to VM clients. - writeMu sync.Mutex - scratch []byte + // vmWriteState holds per-VM-connection write state, serializing + // writes of length-prefixed frames so concurrent writers can't + // interleave them mid-frame. It is deliberately per-connection + // rather than one global lock: a VM that is slow to drain its + // socket (common on contended CI hosts) must not stall writes to + // every other VM on the server. + vmWriteState syncs.Map[*net.UnixConn, *vmWriteState] mu sync.Mutex agentConnWaiter map[*node]chan<- struct{} // signaled after added to set @@ -1184,22 +1259,31 @@ const ( ProtocolUnixDGRAM // for macOS Virtualization.Framework and VZFileHandleNetworkDeviceAttachment ) -func (s *Server) writeEthernetFrameToVM(c vmClient, ethPkt []byte, interfaceID int) { - s.writeMu.Lock() - defer s.writeMu.Unlock() +// vmWriteState is a VM connection's write serialization state. +// See the Server.vmWriteState field comment. +type vmWriteState struct { + mu sync.Mutex + scratch []byte // length-prefixed frame being written; owned by mu +} +func (s *Server) writeEthernetFrameToVM(c vmClient, ethPkt []byte, interfaceID int) { if ethPkt == nil { return } switch c.proto() { case ProtocolQEMU: - s.scratch = binary.BigEndian.AppendUint32(s.scratch[:0], uint32(len(ethPkt))) - s.scratch = append(s.scratch, ethPkt...) - if _, err := c.uc.Write(s.scratch); err != nil { + ws, _ := s.vmWriteState.LoadOrInit(c.uc, func() *vmWriteState { return new(vmWriteState) }) + ws.mu.Lock() + ws.scratch = binary.BigEndian.AppendUint32(ws.scratch[:0], uint32(len(ethPkt))) + ws.scratch = append(ws.scratch, ethPkt...) + _, err := c.uc.Write(ws.scratch) + ws.mu.Unlock() + if err != nil { s.logf("Write pkt: %v", err) } case ProtocolUnixDGRAM: + // Datagram writes are atomic; no locking needed. if _, err := c.uc.WriteToUnix(ethPkt, c.raddr); err != nil { s.logf("Write pkt : %v", err) return @@ -1257,6 +1341,15 @@ func (s *Server) ServeUnixConn(uc *net.UnixConn, proto Protocol) { s.logf("Got conn %T %p", uc, uc) defer uc.Close() + // Enlarge the socket buffers (best effort; the kernel caps these at + // net.core.{w,r}mem_max). The write buffer sits between vnet and a + // QEMU process that may be slow to drain on a contended host; the + // deeper it is, the more of a TCP flow's in-flight data queues here + // (lossless backpressure) instead of overrunning the guest's virtio + // RX ring and getting dropped. + uc.SetWriteBuffer(4 << 20) + uc.SetReadBuffer(4 << 20) + buf := make([]byte, 16<<10) didReg := map[MAC]bool{} for {