personal_interest_mirrors/nebula
1
0
Fork 0
mirror of https://github.com/slackhq/nebula.git synced 2026-08-15 07:17:00 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

plumb ECN through via relays

Browse Source
This commit is contained in:
JackDoan
2026-07-31 12:14:56 -05:00
parent b3002c2d13
commit b7bf32240b
18 changed files with 282 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
control_lifecycle_test.go
+4 -4
View File
@@ -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
}
handshake_manager.go
+3 -3
View File
@@ -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])...)
}
}
handshake_manager_test.go
+1 -1
View File
@@ -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
}
inside.go
+13 -11
View File
@@ -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
}
}
interface.go
+1 -7
View File
@@ -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)
lighthouse_test.go
+1 -1
View File
@@ -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) {
}
outside.go
+6 -2
View File
@@ -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,
punchy.go
+3 -3
View File
@@ -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.
relay_manager.go
+1 -1
View File
@@ -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)
udp/conn.go
+10 -7
View File
@@ -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) {
udp/udp_darwin.go
+3 -2
View File
@@ -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)
udp/udp_ecn_outer_linux_test.go
+130
View File
@@ -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)
}
})
}
}
udp/udp_generic.go
+2 -1
View File
@@ -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
}
udp/udp_linux.go
+33 -16
View File
@@ -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 {
udp/udp_linux_writebatch.go
+22 -16
View File
@@ -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 {
udp/udp_linux_writebatch_alloc_test.go
+43
View File
@@ -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)
}
}
})
}
}
udp/udp_rio_windows.go
+3 -2
View File
@@ -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)
udp/udp_tester.go
+3 -2
View File
@@ -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)