From b7bf32240b4f2afbc30f543523e187df3ba423b6 Mon Sep 17 00:00:00 2001 From: JackDoan Date: Fri, 31 Jul 2026 12:14:56 -0500 Subject: [PATCH] plumb ECN through via relays --- control_lifecycle_test.go | 8 +- handshake_manager.go | 6 +- handshake_manager_test.go | 2 +- inside.go | 24 ++--- interface.go | 8 +- lighthouse_test.go | 2 +- outside.go | 8 +- punchy.go | 6 +- relay_manager.go | 2 +- udp/conn.go | 17 ++-- udp/udp_darwin.go | 5 +- udp/udp_ecn_outer_linux_test.go | 130 +++++++++++++++++++++++++ udp/udp_generic.go | 3 +- udp/udp_linux.go | 49 +++++++--- udp/udp_linux_writebatch.go | 38 +++++--- udp/udp_linux_writebatch_alloc_test.go | 43 ++++++++ udp/udp_rio_windows.go | 5 +- udp/udp_tester.go | 5 +- 18 files changed, 282 insertions(+), 79 deletions(-) diff --git a/control_lifecycle_test.go b/control_lifecycle_test.go index 73b14b46..4ec331a7 100644 --- a/control_lifecycle_test.go +++ b/control_lifecycle_test.go @@ -144,10 +144,10 @@ type fakeConn struct { rebinds int } -func (c *fakeConn) Rebind() error { c.rebinds++; return nil } -func (c *fakeConn) LocalAddr() (netip.AddrPort, error) { return netip.AddrPort{}, nil } -func (c *fakeConn) ListenOut(_ udp.EncReader, _ func()) error { return nil } -func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort) error { return nil } +func (c *fakeConn) Rebind() error { c.rebinds++; return nil } +func (c *fakeConn) LocalAddr() (netip.AddrPort, error) { return netip.AddrPort{}, nil } +func (c *fakeConn) ListenOut(_ udp.EncReader, _ func()) error { return nil } +func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort, _ byte) error { return nil } func (c *fakeConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) { return len(bufs), nil } diff --git a/handshake_manager.go b/handshake_manager.go index 419f7b3e..d252f315 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -293,7 +293,7 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered var sentTo []netip.AddrPort hostinfo.remotes.ForEach(hm.mainHostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) { hm.messageMetrics.Tx(header.Handshake, hh.machine.Subtype(), 1) - err := hm.outside.WriteTo(stage0, addr) + err := hm.outside.WriteTo(stage0, addr, 0) if err != nil { // These repeat every attempt, so match the success log below and only shout when the remotes changed level := slog.LevelDebug @@ -1074,7 +1074,7 @@ func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hos if !via.IsRelayed { fields := append(logFields, "from", via) - err := f.outside.WriteTo(msg, via.UdpAddr) + err := f.outside.WriteTo(msg, via.UdpAddr, 0) if err != nil { f.l.Error("Failed to send handshake message", append(fields, "error", err)...) } else { @@ -1089,7 +1089,7 @@ func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hos // We received a valid handshake on this relay, so make sure the relay // state reflects that, in case it had been marked Disestablished. via.relayHI.relayState.UpdateRelayForByIdxState(via.relay.LocalIndex, Established) - f.SendVia(via.relayHI, via.relay, msg, make([]byte, 12), make([]byte, mtu), false) + f.SendVia(via.relayHI, via.relay, msg, make([]byte, 12), make([]byte, mtu), false, 0, 0) f.l.Info("Handshake message sent", append(logFields, "relay", via.relayHI.vpnAddrs[0])...) } } diff --git a/handshake_manager_test.go b/handshake_manager_test.go index 5f8383e4..7915c97c 100644 --- a/handshake_manager_test.go +++ b/handshake_manager_test.go @@ -84,7 +84,7 @@ func (mw *mockEncWriter) SendMessageToVpnAddr(_ header.MessageType, _ header.Mes return } -func (mw *mockEncWriter) SendVia(_ *HostInfo, _ *Relay, _, _, _ []byte, _ bool) { +func (mw *mockEncWriter) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) { return } diff --git a/inside.go b/inside.go index f484667c..30468093 100644 --- a/inside.go +++ b/inside.go @@ -520,21 +520,16 @@ func (f *Interface) prepareSendVia(via *HostInfo, // ad is the plaintext data to authenticate, but not encrypt // nb is a buffer used to store the nonce value, re-used for performance reasons. // out is a buffer used to store the result of the Encrypt operation +// outerECN is the 2-bit codepoint to stamp on the carrier datagram (0 for control traffic). // q indicates which writer to use to send the packet. -func (f *Interface) SendVia(via *HostInfo, - relay *Relay, - ad, - nb, - out []byte, - nocopy bool, -) { +func (f *Interface) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) { toSend, err := f.prepareSendVia(via, relay, ad, nb, out, nocopy) if err != nil { // already logged by prepareSendVia return } - err = f.writers[0].WriteTo(toSend, via.GetRemote()) + err = f.writers[q].WriteTo(toSend, via.GetRemote(), outerECN) if err != nil { via.logger(f.l).Info("Failed to WriteTo in sendVia", "error", err) } @@ -595,8 +590,15 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType return } + // Data packets copy the inner packet's ECN codepoint onto the outer + // carrier per RFC 6040; control traffic stays Not-ECT. + var outerECN byte + if t == header.Message && f.ecnEnabled.Load() { + outerECN = innerECN(p) + } + if remote.IsValid() { - err = f.writers[q].WriteTo(out, remote) + err = f.writers[q].WriteTo(out, remote, outerECN) if err != nil { hostinfo.logger(f.l).Error("Failed to write outgoing packet", "error", err, @@ -604,7 +606,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType ) } } else if hr := hostinfo.GetRemote(); hr.IsValid() { - err = f.writers[q].WriteTo(out, hr) + err = f.writers[q].WriteTo(out, hr, outerECN) if err != nil { hostinfo.logger(f.l).Error("Failed to write outgoing packet", "error", err, @@ -623,7 +625,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType ) continue } - f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true) + f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true, outerECN, q) break } } diff --git a/interface.go b/interface.go index d66feca5..7053f1ed 100644 --- a/interface.go +++ b/interface.go @@ -143,13 +143,7 @@ type Interface struct { } type EncWriter interface { - SendVia(via *HostInfo, - relay *Relay, - ad, - nb, - out []byte, - nocopy bool, - ) + SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) SendMessageToVpnAddr(t header.MessageType, st header.MessageSubType, vpnAddr netip.Addr, p, nb, out []byte) SendMessageToHostInfo(t header.MessageType, st header.MessageSubType, hostinfo *HostInfo, p, nb, out []byte) Handshake(vpnAddr netip.Addr) diff --git a/lighthouse_test.go b/lighthouse_test.go index 81c883ff..6b15e162 100644 --- a/lighthouse_test.go +++ b/lighthouse_test.go @@ -498,7 +498,7 @@ type testEncWriter struct { protocolVersion cert.Version } -func (tw *testEncWriter) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool) { +func (tw *testEncWriter) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) { } func (tw *testEncWriter) Handshake(vpnIp netip.Addr) { } diff --git a/outside.go b/outside.go index 85d5d752..bbed1659 100644 --- a/outside.go +++ b/outside.go @@ -233,8 +233,12 @@ func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, // Forward this packet through the relay tunnel, rebuilding it in place. // Encode overwrites the old outer header, and the new AEAD tag lands where the old one was fwdBuf := packet[:0:len(packet)] // Cap to len(packet) to protect memory from a larger parent buffer + var fwdECN byte + if f.ecnEnabled.Load() { + fwdECN = meta.OuterECN + } //todo it would potentially be nice to batch these - f.SendVia(targetHI, targetRelay, signedPayload, nb, fwdBuf, true) + f.SendVia(targetHI, targetRelay, signedPayload, nb, fwdBuf, true, fwdECN, q) case TerminalType: hostinfo.logger(f.l).Error("Unexpected Relay Type of Terminal") return @@ -609,7 +613,7 @@ func (f *Interface) sendRecvError(endpoint netip.AddrPort, index uint32) { f.messageMetrics.Tx(header.RecvError, 0, 1) b := header.Encode(make([]byte, header.Len), header.Version, header.RecvError, 0, index, 0) - _ = f.outside.WriteTo(b, endpoint) + _ = f.outside.WriteTo(b, endpoint, 0) if f.l.Enabled(context.Background(), slog.LevelDebug) { f.l.Debug("Recv error sent", "index", index, diff --git a/punchy.go b/punchy.go index 4bce4392..915942fa 100644 --- a/punchy.go +++ b/punchy.go @@ -176,7 +176,7 @@ func (p *Punchy) SendPunch(hostinfo *HostInfo) { p.sendPunchToAllRemotes(hostinfo) } else if hr := hostinfo.GetRemote(); hr.IsValid() { p.metricPunchyTx.Inc(1) - p.punchConn.WriteTo([]byte{1}, hr) + p.punchConn.WriteTo([]byte{1}, hr, 0) } } @@ -200,7 +200,7 @@ func (p *Punchy) SendPunchToAll(hostinfo *HostInfo) { func (p *Punchy) sendPunchToAllRemotes(hostinfo *HostInfo) { hostinfo.remotes.ForEach(p.hm.GetPreferredRanges(), func(addr netip.AddrPort, preferred bool) { p.metricPunchyTx.Inc(1) - p.punchConn.WriteTo([]byte{1}, addr) + p.punchConn.WriteTo([]byte{1}, addr, 0) }) } @@ -222,7 +222,7 @@ func (p *Punchy) Start(ctx context.Context, ifce EncWriter, hm *HostMap, lh ligh p.l.Debug("Punching", "target", job.target, "vpnAddr", job.vpnAddr) } p.metricHolepunchTx.Inc(1) - p.punchConn.WriteTo(empty, job.target) + p.punchConn.WriteTo(empty, job.target, 0) case job.vpnAddr.IsValid(): // A nebula test packet to the host trying to contact us. // In the case of a double nat or other difficult scenario, this may help establish a tunnel. diff --git a/relay_manager.go b/relay_manager.go index 1ae382a3..0aa7ff4b 100644 --- a/relay_manager.go +++ b/relay_manager.go @@ -161,7 +161,7 @@ func (rm *relayManager) StartRelays(f *Interface, vpnIp netip.Addr, hh *Handshak switch existingRelay.State { case Established: hl.Log(context.Background(), level, "Send handshake via relay", "relay", relay.String()) - f.SendVia(relayHostInfo, existingRelay, stage0, make([]byte, 12), make([]byte, mtu), false) + f.SendVia(relayHostInfo, existingRelay, stage0, make([]byte, 12), make([]byte, mtu), false, 0, 0) case Disestablished: // Mark this relay as 'requested' relayHostInfo.relayState.UpdateRelayForByIpState(vpnIp, Requested) diff --git a/udp/conn.go b/udp/conn.go index dba84c0a..4af50b01 100644 --- a/udp/conn.go +++ b/udp/conn.go @@ -35,13 +35,16 @@ type EncReader func( type Conn interface { Rebind() error LocalAddr() (netip.AddrPort, error) - // ListenOut invokes r for each received packet. On batch-capable - // backends (recvmmsg), flush is called after each batch is fully - // delivered — callers use it to flush per-batch accumulators such as - // TUN write coalescers. Single-packet backends call flush after each - // packet. flush must not be nil. + // ListenOut invokes r for each received packet. + // On batch-capable backends (recvmmsg), flush is called after each batch is fully delivered. + // Callers use it to flush per-batch accumulators such as TUN write coalescers. + // Single-packet backends call flush after each packet. flush must not be nil. ListenOut(r EncReader, flush func()) error - WriteTo(b []byte, addr netip.AddrPort) error + // WriteTo sends a single packet to addr. + // outerECN is the 2-bit IP-level ECN codepoint to stamp on the packet's outer IP header. + // 0 (Not-ECT) is the pass-through value. + // Linux attaches it as an IP_TOS / IPV6_TCLASS cmsg. Backends without per-packet ECN support ignore it. + WriteTo(b []byte, addr netip.AddrPort, outerECN byte) error // WriteBatch sends a contiguous batch of packets, each with its own // destination. bufs and addrs must have the same length. outerECNs may // be nil (treated as all-zero / Not-ECT); when non-nil it must have the @@ -73,7 +76,7 @@ func (NoopConn) ListenOut(_ EncReader, _ func()) error { func (NoopConn) SupportsMultipleReaders() bool { return false } -func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort) error { +func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort, _ byte) error { return nil } func (NoopConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) { diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index 58c4bfe2..60c0d065 100644 --- a/udp/udp_darwin.go +++ b/udp/udp_darwin.go @@ -89,7 +89,8 @@ func NewListenConfig(multi bool) net.ListenConfig { //go:noescape func sendto(s int, buf []byte, flags int, to unsafe.Pointer, addrlen int32) (err error) -func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error { +// WriteTo ignores outerECN; per-packet ECN marking is not implemented on darwin. +func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort, _ byte) error { var sa unsafe.Pointer var addrLen int32 @@ -147,7 +148,7 @@ func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (i // writability on EAGAIN before giving up on the remainder. written := 0 for i, b := range bufs { - if err := u.WriteTo(b, addrs[i]); err == nil { + if err := u.WriteTo(b, addrs[i], 0); err == nil { written++ } else { u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err) diff --git a/udp/udp_ecn_outer_linux_test.go b/udp/udp_ecn_outer_linux_test.go index 4f83c7c0..c8e4a916 100644 --- a/udp/udp_ecn_outer_linux_test.go +++ b/udp/udp_ecn_outer_linux_test.go @@ -3,8 +3,11 @@ package udp import ( + "encoding/binary" "net/netip" "testing" + + "golang.org/x/sys/unix" ) // TestPlanRunBreaksOnECNChange confirms that two same-destination, same-size @@ -59,3 +62,130 @@ func TestPlanRunBreaksOnECNChange(t *testing.T) { } }) } + +// ecnReceiver is a raw UDP socket with IP_RECVTOS / IPV6_RECVTCLASS enabled, +// used to observe the outer ECN codepoint WriteTo stamps on the wire. +type ecnReceiver struct { + fd int + addr netip.AddrPort +} + +func newEcnReceiver(t *testing.T, v6 bool) *ecnReceiver { + t.Helper() + family := unix.AF_INET + if v6 { + family = unix.AF_INET6 + } + fd, err := unix.Socket(family, unix.SOCK_DGRAM, 0) + if err != nil { + t.Fatalf("socket: %v", err) + } + t.Cleanup(func() { unix.Close(fd) }) + + var bindAddr netip.Addr + if v6 { + if err = unix.SetsockoptInt(fd, unix.IPPROTO_IPV6, unix.IPV6_RECVTCLASS, 1); err != nil { + t.Fatalf("IPV6_RECVTCLASS: %v", err) + } + if err = unix.Bind(fd, &unix.SockaddrInet6{Addr: [16]byte{15: 1}}); err != nil { + t.Fatalf("bind ::1: %v", err) + } + bindAddr = netip.MustParseAddr("::1") + } else { + if err = unix.SetsockoptInt(fd, unix.IPPROTO_IP, unix.IP_RECVTOS, 1); err != nil { + t.Fatalf("IP_RECVTOS: %v", err) + } + if err = unix.Bind(fd, &unix.SockaddrInet4{Addr: [4]byte{127, 0, 0, 1}}); err != nil { + t.Fatalf("bind 127.0.0.1: %v", err) + } + bindAddr = netip.MustParseAddr("127.0.0.1") + } + tv := unix.Timeval{Sec: 5} + if err = unix.SetsockoptTimeval(fd, unix.SOL_SOCKET, unix.SO_RCVTIMEO, &tv); err != nil { + t.Fatalf("SO_RCVTIMEO: %v", err) + } + + sa, err := unix.Getsockname(fd) + if err != nil { + t.Fatalf("getsockname: %v", err) + } + var port int + switch v := sa.(type) { + case *unix.SockaddrInet4: + port = v.Port + case *unix.SockaddrInet6: + port = v.Port + default: + t.Fatalf("unexpected sockaddr %T", sa) + } + return &ecnReceiver{fd: fd, addr: netip.AddrPortFrom(bindAddr, uint16(port))} +} + +// recvECN receives one datagram and returns the 2-bit ECN codepoint from its +// TOS / TCLASS cmsg. +func (r *ecnReceiver) recvECN(t *testing.T) byte { + t.Helper() + buf := make([]byte, 128) + oob := make([]byte, 128) + _, oobn, _, _, err := unix.Recvmsg(r.fd, buf, oob, 0) + if err != nil { + t.Fatalf("recvmsg: %v", err) + } + cmsgs, err := unix.ParseSocketControlMessage(oob[:oobn]) + if err != nil { + t.Fatalf("parse cmsg: %v", err) + } + for _, m := range cmsgs { + switch { + case m.Header.Level == unix.IPPROTO_IP && m.Header.Type == unix.IP_TOS: + return m.Data[0] & 0x03 + case m.Header.Level == unix.IPPROTO_IPV6 && m.Header.Type == unix.IPV6_TCLASS: + return byte(binary.NativeEndian.Uint32(m.Data)) & 0x03 + } + } + t.Fatal("no TOS/TCLASS cmsg received") + return 0 +} + +// TestWriteToStampsOuterECN sends single packets through StdConn.WriteTo and +// asserts the requested ECN codepoint lands on the outer IP header, for a +// v4 socket, a v6 socket, and the dual-stack case where a v4-mapped +// destination must be stamped via IP_TOS rather than IPV6_TCLASS. +func TestWriteToStampsOuterECN(t *testing.T) { + cases := []struct { + name string + bind string + recvV6 bool + sendECN byte + }{ + {"v4_socket_to_v4", "127.0.0.1", false, 0x03}, + {"v6_socket_to_v6", "::1", true, 0x01}, + {"dualstack_v6_socket_to_v4_mapped", "::", false, 0x02}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c, err := NewListener(testLogger(), netip.MustParseAddr(tc.bind), 0, false, 8) + if err != nil { + t.Fatalf("NewListener: %v", err) + } + defer c.Close() + rx := newEcnReceiver(t, tc.recvV6) + + if err = c.WriteTo([]byte("ecn"), rx.addr, tc.sendECN); err != nil { + t.Fatalf("WriteTo(ecn=%#02x): %v", tc.sendECN, err) + } + if got := rx.recvECN(t); got != tc.sendECN { + t.Errorf("outer ECN = %#02x, want %#02x", got, tc.sendECN) + } + + // The zero codepoint sends no TOS cmsg and must arrive Not-ECT + // (the socket-default TOS byte). + if err = c.WriteTo([]byte("ecn"), rx.addr, 0); err != nil { + t.Fatalf("WriteTo(ecn=0): %v", err) + } + if got := rx.recvECN(t); got != 0 { + t.Errorf("outer ECN = %#02x, want 0 (Not-ECT)", got) + } + }) + } +} diff --git a/udp/udp_generic.go b/udp/udp_generic.go index a7c83e92..975e6549 100644 --- a/udp/udp_generic.go +++ b/udp/udp_generic.go @@ -39,7 +39,8 @@ func NewGenericListener(l *slog.Logger, ip netip.Addr, port int, multi bool, bat return nil, fmt.Errorf("Unexpected PacketConn: %T %#v", pc, pc) } -func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort) error { +// WriteTo ignores outerECN; the stdlib UDPConn offers no per-packet TOS control. +func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort, _ byte) error { _, err := u.UDPConn.WriteToUDPAddrPort(b, addr) return err } diff --git a/udp/udp_linux.go b/udp/udp_linux.go index e8884ddc..356b8f88 100644 --- a/udp/udp_linux.go +++ b/udp/udp_linux.go @@ -411,11 +411,11 @@ func parseRecvCmsg(hdr *msghdr, wantGRO, wantECN bool) (gso int, ecn byte) { return gso, ecn } -func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error { - return sendto(u.sysFd, b, ip, u.isV4) +func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort, ecn byte) error { + return sendmsg(u.sysFd, b, ip, u.isV4, ecn) } -func sendto(fd int, b []byte, addr netip.AddrPort, isV4 bool) error { +func sendmsg(fd int, b []byte, addr netip.AddrPort, isV4 bool, ecn byte) error { var rsa [unix.SizeofSockaddrInet6]byte nlen, err := writeSockaddr(rsa[:], addr, isV4) if err != nil { @@ -425,31 +425,48 @@ func sendto(fd int, b []byte, addr netip.AddrPort, isV4 bool) error { if len(b) > 0 { base = &b[0] } + + var iov iovec + iov.Base = base + setIovLen(&iov, len(b)) + + var hdr msghdr + hdr.Name = &rsa[0] + hdr.Namelen = uint32(nlen) + hdr.Iov = &iov + setMsgIovlen(&hdr, 1) + + // Stack scratch for the ECN cmsg, typed as uint64s so its base is cmsg-aligned on every arch. + // CmsgSpace(4) needs 24 bytes on 64-bit linux, 16 on 32-bit. + var ctrl [3]uint64 + if ecn != 0 { + buf := (*[24]byte)(unsafe.Pointer(&ctrl[0]))[:] + writeECNCmsg(buf, addr.Addr().Unmap().Is4(), ecn) + hdr.Control = &buf[0] + setMsgControllen(&hdr, unix.CmsgSpace(4)) + } + _, _, errno := unix.Syscall6( - unix.SYS_SENDTO, + unix.SYS_SENDMSG, uintptr(fd), - uintptr(unsafe.Pointer(base)), - uintptr(len(b)), - 0, - uintptr(unsafe.Pointer(&rsa[0])), - uintptr(nlen), + uintptr(unsafe.Pointer(&hdr)), + 0, 0, 0, 0, ) if errno != 0 { - return &net.OpError{Op: "sendto", Err: errno} + return &net.OpError{Op: "sendmsg", Err: errno} } return nil } -// WriteBatch sends bufs via sendmmsg(2), coalescing same-destination runs -// into UDP-GSO superpackets when supported. See batchWriter in -// udp_linux_writebatch.go for the mechanics. +// WriteBatch sends bufs via sendmmsg(2), coalescing same-destination runs into UDP-GSO superpackets when supported. +// See batchWriter in udp_linux_writebatch.go for the mechanics. func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) (int, error) { return u.bw.WriteBatch(bufs, addrs, ecns) } -// writeSockaddr encodes addr into buf (which must be at least -// SizeofSockaddrInet6 bytes). Returns the number of bytes used. If isV4 is -// true and addr is not a v4 (or v4-in-v6) address, returns an error. +// writeSockaddr encodes addr into buf (which must be at least SizeofSockaddrInet6 bytes). +// Returns the number of bytes used. +// If isV4 is true and addr is not a v4 (or v4-in-v6) address, returns an error. func writeSockaddr(buf []byte, addr netip.AddrPort, isV4 bool) (int, error) { ap := addr.Addr().Unmap() if isV4 { diff --git a/udp/udp_linux_writebatch.go b/udp/udp_linux_writebatch.go index 912ed619..00bc95cb 100644 --- a/udp/udp_linux_writebatch.go +++ b/udp/udp_linux_writebatch.go @@ -385,14 +385,30 @@ func (w *batchWriter) planRun(bufs [][]byte, addrs []netip.AddrPort, ecns []byte return runLen, segSize } +// writeECNCmsg fills the start of buf with one IP_TOS / IPV6_TCLASS cmsg +// carrying the 2-bit ECN codepoint. buf must be cmsg-aligned (the batch +// writer's heap slab is runtime-aligned; sendmsg passes uint64-backed stack +// scratch) and at least CmsgSpace(4) bytes. The cmsg family must match the +// socket: on the default dual-stack v6 bind, a v4-mapped destination takes +// the kernel's IPv4 path, which reads IP_TOS and ignores IPV6_TCLASS. The +// payload is a 4-byte int for both families, so the cmsg space is the same. +func writeECNCmsg(buf []byte, dstIsV4 bool, ecn byte) { + h := (*unix.Cmsghdr)(unsafe.Pointer(&buf[0])) + if dstIsV4 { + h.Level = int32(unix.IPPROTO_IP) + h.Type = int32(unix.IP_TOS) + } else { + h.Level = int32(unix.IPPROTO_IPV6) + h.Type = int32(unix.IPV6_TCLASS) + } + setCmsgLen(h, unix.CmsgLen(4)) + dataOff := unix.CmsgLen(0) + binary.NativeEndian.PutUint32(buf[dataOff:dataOff+4], uint32(ecn)) +} + // writeEntryCmsg writes one entry's cmsgs: the UDP_SEGMENT payload when // runLen >= 2, the IP_TOS/IPV6_TCLASS cmsg when ecn != 0, then points // Hdr.Control at the smallest span covering the cmsgs in use. -// -// The ECN cmsg family must match the destination, not the socket: on the -// default dual-stack v6 bind, a v4-mapped destination takes the kernel's -// IPv4 path, which reads IP_TOS and ignores IPV6_TCLASS. The payload is a -// 4-byte int for both families, so the cmsg space is the same. func (w *batchWriter) writeEntryCmsg(entry, runLen, segSize int, ecn byte, dstIsV4 bool) { hdr := &w.msgs[entry].Hdr useSeg := runLen >= 2 @@ -404,17 +420,7 @@ func (w *batchWriter) writeEntryCmsg(entry, runLen, segSize int, ecn byte, dstIs binary.NativeEndian.PutUint16(w.cmsg[dataOff:dataOff+2], uint16(segSize)) } if useEcn { - ecnHdr := (*unix.Cmsghdr)(unsafe.Pointer(&w.cmsg[base+w.cmsgSegSpace])) - if dstIsV4 { - ecnHdr.Level = int32(unix.IPPROTO_IP) - ecnHdr.Type = int32(unix.IP_TOS) - } else { - ecnHdr.Level = int32(unix.IPPROTO_IPV6) - ecnHdr.Type = int32(unix.IPV6_TCLASS) - } - setCmsgLen(ecnHdr, unix.CmsgLen(4)) - dataOff := base + w.cmsgSegSpace + unix.CmsgLen(0) - binary.NativeEndian.PutUint32(w.cmsg[dataOff:dataOff+4], uint32(ecn)) + writeECNCmsg(w.cmsg[base+w.cmsgSegSpace:], dstIsV4, ecn) } switch { diff --git a/udp/udp_linux_writebatch_alloc_test.go b/udp/udp_linux_writebatch_alloc_test.go index d0b5dc4a..ffc641cb 100644 --- a/udp/udp_linux_writebatch_alloc_test.go +++ b/udp/udp_linux_writebatch_alloc_test.go @@ -99,3 +99,46 @@ func TestWriteBatchNoAllocs(t *testing.T) { }) } } + +// TestWriteToNoAllocs verifies the single-packet WriteTo path performs no +// heap allocations on the happy path, both without ancillary data and with +// an ECN cmsg (which is built in stack scratch, not a per-call slab). +func TestWriteToNoAllocs(t *testing.T) { + for _, tc := range []struct { + name string + addr string + }{ + {"v4", "127.0.0.1"}, + {"v6", "::1"}, + } { + t.Run(tc.name, func(t *testing.T) { + ip := netip.MustParseAddr(tc.addr) + newConn := func() Conn { + c, err := NewListener(testLogger(), ip, 0, false, 8) + if err != nil { + t.Fatalf("NewListener: %v", err) + } + t.Cleanup(func() { _ = c.Close() }) + return c + } + tx := newConn() + rx := newConn() + dst, err := rx.LocalAddr() + if err != nil { + t.Fatalf("LocalAddr: %v", err) + } + + payload := make([]byte, 512) + for _, ecn := range []byte{0, 0x03} { + allocs := testing.AllocsPerRun(100, func() { + if werr := tx.WriteTo(payload, dst, ecn); werr != nil { + t.Fatalf("WriteTo(ecn=%#02x): %v", ecn, werr) + } + }) + if allocs != 0 { + t.Errorf("ecn=%#02x: %v allocs per WriteTo, want 0", ecn, allocs) + } + } + }) + } +} diff --git a/udp/udp_rio_windows.go b/udp/udp_rio_windows.go index a6f097dc..f9a555cf 100644 --- a/udp/udp_rio_windows.go +++ b/udp/udp_rio_windows.go @@ -254,7 +254,8 @@ retry: return n, ep, nil } -func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort) error { +// WriteTo ignores outerECN; per-packet ECN marking is not implemented on windows. +func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort, _ byte) error { if !u.isOpen.Load() { return net.ErrClosed } @@ -321,7 +322,7 @@ func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (i // An un-sendable destination costs its own packet, never the ones behind it in the batch. written := 0 for i, b := range bufs { - if err := u.WriteTo(b, addrs[i]); err == nil { + if err := u.WriteTo(b, addrs[i], 0); err == nil { written++ } else { u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err) diff --git a/udp/udp_tester.go b/udp/udp_tester.go index aa6cd570..8d1ddc34 100644 --- a/udp/udp_tester.go +++ b/udp/udp_tester.go @@ -153,7 +153,8 @@ func (u *TesterConn) Get(block bool) *Packet { // Below this is boilerplate implementation to make nebula actually work //********************************************************************************************************************// -func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error { +// WriteTo ignores outerECN; the in-memory tester carries no IP headers. +func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort, _ byte) error { p := acquirePacket() if cap(p.Data) < len(b) { p.Data = make([]byte, len(b)) @@ -174,7 +175,7 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error { func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) { written := 0 for i, b := range bufs { - if err := u.WriteTo(b, addrs[i]); err == nil { + if err := u.WriteTo(b, addrs[i], 0); err == nil { written++ } else { u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)