drop ECN support for this release

This commit is contained in:
JackDoan
2026-07-31 13:38:16 -05:00
parent d3779b6a39
commit 93946faf7a
24 changed files with 148 additions and 1101 deletions
+6 -34
View File
@@ -14,29 +14,9 @@ const MTU = 9001
// only costs additional sendmmsg chunks within a single WriteBatch call.
const MaxWriteBatch = 128
// RxMeta carries per-packet metadata extracted from the RX path (ancillary
// data, kernel offload state, etc.) and passed to EncReader callbacks.
// Backends that do not produce a particular signal leave its zero value.
//
// OuterECN is the 2-bit IP-level ECN codepoint stamped on the carrier
// datagram (extracted from IP_TOS / IPV6_TCLASS cmsg on Linux). Zero
// means Not-ECT, which is also the value backends without ECN RX support
// supply on every packet.
type RxMeta struct {
OuterECN byte
// QueueCongested is set when the receiving socket's kernel queue depth
// exceeded the configured AQM marking threshold (tunnels.ecn_mark_threshold)
// when this batch was pulled. The decap path treats it like an outer CE
// mark on ECT inner packets — nebula acting as the AQM for the one queue
// on the tunnel path no kernel AQM can see. Backends without queue
// introspection leave it false.
QueueCongested bool
}
type EncReader func(
addr netip.AddrPort,
payload []byte,
meta RxMeta,
)
type Conn interface {
@@ -47,23 +27,15 @@ type Conn interface {
// 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 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
WriteTo(b []byte, addr netip.AddrPort) 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
// same length as bufs, and outerECNs[i] is the 2-bit IP-level ECN
// codepoint to set on packet i's outer header. Linux uses sendmmsg(2)
// for a single syscall and attaches the value as IP_TOS / IPV6_TCLASS
// cmsg; other backends ignore it.
// destination. bufs and addrs must have the same length. Linux uses
// sendmmsg(2) for a single syscall.
//
// Returns the number of packets successfully written. A destination the kernel rejects costs only
// its own packet, so a short count means some peers were undeliverable, not that the batch failed.
// Not safe for concurrent use on the same Conn.
WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) (int, error)
WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error)
ReloadConfig(c *config.C)
SupportsMultipleReaders() bool
Close() error
@@ -83,10 +55,10 @@ func (NoopConn) ListenOut(_ EncReader, _ func()) error {
func (NoopConn) SupportsMultipleReaders() bool {
return false
}
func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort, _ byte) error {
func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort) error {
return nil
}
func (NoopConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) {
func (NoopConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort) (int, error) {
return len(bufs), nil
}
func (NoopConn) ReloadConfig(_ *config.C) {
+4 -5
View File
@@ -89,8 +89,7 @@ func NewListenConfig(multi bool) net.ListenConfig {
//go:noescape
func sendto(s int, buf []byte, flags int, to unsafe.Pointer, addrlen int32) (err error)
// WriteTo ignores outerECN; per-packet ECN marking is not implemented on darwin.
func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort, _ byte) error {
func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error {
var sa unsafe.Pointer
var addrLen int32
@@ -141,14 +140,14 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort, _ byte) error {
}
}
func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
// An un-sendable destination costs its own packet, never the ones behind it in the batch.
// TODO: WriteTo maps EWOULDBLOCK to an error, so a full send buffer
// silently drops the rest of a burst (linux blocks instead). Poll for
// writability on EAGAIN before giving up on the remainder.
written := 0
for i, b := range bufs {
if err := u.WriteTo(b, addrs[i], 0); err == nil {
if err := u.WriteTo(b, addrs[i]); err == nil {
written++
} else {
u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
@@ -196,7 +195,7 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
continue
}
r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n], RxMeta{})
r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])
flush()
}
}
-191
View File
@@ -1,191 +0,0 @@
//go:build linux && !android && !e2e_testing
package udp
import (
"encoding/binary"
"net/netip"
"testing"
"golang.org/x/sys/unix"
)
// TestPlanRunBreaksOnECNChange confirms that two same-destination, same-size
// packets with different outer ECN end up in separate sendmmsg entries (the
// kernel stamps one outer codepoint per entry, so a run that straddled the
// boundary would silently lose information).
func TestPlanRunBreaksOnECNChange(t *testing.T) {
u := &batchWriter{gsoSupported: true, maxGSOSegments: 63}
dst := netip.MustParseAddrPort("10.0.0.1:4242")
bufs := [][]byte{
make([]byte, 1200),
make([]byte, 1200),
make([]byte, 1200),
}
addrs := []netip.AddrPort{dst, dst, dst}
t.Run("uniform_ecn_runs_together", func(t *testing.T) {
ecns := []byte{0x02, 0x02, 0x02}
runLen, segSize := u.planRun(bufs, addrs, ecns, 0, 64)
if runLen != 3 {
t.Errorf("runLen=%d want 3 (uniform ECT(0))", runLen)
}
if segSize != 1200 {
t.Errorf("segSize=%d want 1200", segSize)
}
})
t.Run("ecn_change_truncates_run", func(t *testing.T) {
// 0,0,3: first two run together, CE seeds a fresh entry.
ecns := []byte{0x00, 0x00, 0x03}
runLen, _ := u.planRun(bufs, addrs, ecns, 0, 64)
if runLen != 2 {
t.Errorf("runLen=%d want 2 (ECN changes at index 2)", runLen)
}
})
t.Run("nil_ecns_runs_full", func(t *testing.T) {
runLen, _ := u.planRun(bufs, addrs, nil, 0, 64)
if runLen != 3 {
t.Errorf("runLen=%d want 3 (nil ecns means no break)", runLen)
}
})
t.Run("first_ecn_is_singleton", func(t *testing.T) {
// Second packet has different ECN from the first → run halts at 1
// (the first packet alone forms the run).
ecns := []byte{0x00, 0x03, 0x03}
runLen, _ := u.planRun(bufs, addrs, ecns, 0, 64)
if runLen != 1 {
t.Errorf("runLen=%d want 1 (different ECN immediately)", runLen)
}
})
}
// 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)
}
})
}
}
+3 -4
View File
@@ -39,13 +39,12 @@ func NewGenericListener(l *slog.Logger, ip netip.Addr, port int, multi bool, bat
return nil, fmt.Errorf("Unexpected PacketConn: %T %#v", pc, pc)
}
// WriteTo ignores outerECN; the stdlib UDPConn offers no per-packet TOS control.
func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort, _ byte) error {
func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort) error {
_, err := u.UDPConn.WriteToUDPAddrPort(b, addr)
return err
}
func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
// An un-sendable destination costs its own packet, never the ones behind it in the batch.
written := 0
for i, b := range bufs {
@@ -107,7 +106,7 @@ func (u *GenericConn) ListenOut(r EncReader, flush func()) error {
continue
}
r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n], RxMeta{})
r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])
flush()
}
}
+24 -168
View File
@@ -8,10 +8,8 @@ import (
"errors"
"fmt"
"log/slog"
"math"
"net"
"net/netip"
"strconv"
"sync/atomic"
"syscall"
"unsafe"
@@ -39,19 +37,6 @@ type StdConn struct {
// consecutive same-flow datagrams into a single recvmmsg entry; the
// delivered cmsg carries the gso_size used to split them back apart.
groSupported bool
// ecnRecvSupported is true when IP_RECVTOS / IPV6_RECVTCLASS was
// successfully enabled — the kernel will deliver the outer IP-ECN of
// each arriving datagram as a per-slot cmsg, and ListenOut passes
// the parsed value to the EncReader callback for RFC 6040 combine.
ecnRecvSupported bool
// ecnMarkThreshold holds tunnels.ecn_mark_threshold as float64 bits: the
// fraction of the socket receive buffer above which listenOutBatch flags
// the batch QueueCongested (decap then CE-marks ECT inner packets). Zero
// disables sampling entirely. Atomic because ReloadConfig may update it
// while the reader runs.
ecnMarkThreshold atomic.Uint64
}
func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
@@ -102,11 +87,6 @@ func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int)
if batch > 1 {
out.prepareGRO()
}
// Best-effort: ask the kernel to deliver outer IP-ECN as ancillary data
// on every recvmmsg slot so the decap side can apply RFC 6040 combine.
// On older kernels these may not exist; failing here just means we get
// 0 (Not-ECT) on every slot, which is the same as ecn_mode=disable.
out.prepareECNRecv()
return out, nil
}
@@ -138,34 +118,6 @@ func (u *StdConn) prepareGRO() {
recordCapability("udp.gro.enabled", true)
}
// prepareECNRecv turns on IP_RECVTOS / IPV6_RECVTCLASS so the outer IP-ECN
// field of each arriving datagram is delivered as ancillary data alongside
// the payload. ListenOut reads it via parseRecvCmsg and passes the codepoint
// through the EncReader for RFC 6040 combine on the decap side. Best-effort:
// we keep going on failure, and each family degrades independently — a peer
// whose family's probe failed just delivers no cmsg and lands as Not-ECT.
// Only a failure of every family the socket speaks turns the parsing off.
func (u *StdConn) prepareECNRecv() {
v4err := unix.SetsockoptInt(u.sysFd, unix.IPPROTO_IP, unix.IP_RECVTOS, 1)
var v6err error
if !u.isV4 {
v6err = unix.SetsockoptInt(u.sysFd, unix.IPPROTO_IPV6, unix.IPV6_RECVTCLASS, 1)
}
switch {
case v4err != nil && (u.isV4 || v6err != nil):
u.l.Info("udp: outer-ECN RX disabled", "reason", "kernel rejected probe", "error", errors.Join(v4err, v6err))
recordCapability("udp.ecn_rx.enabled", false)
return
case v4err != nil:
u.l.Debug("udp: outer-ECN RX degraded", "reason", "kernel rejected probe on IPv4", "error", v4err)
case v6err != nil:
u.l.Debug("udp: outer-ECN RX degraded", "reason", "kernel rejected probe on IPv6", "error", v6err)
}
u.ecnRecvSupported = true
u.l.Info("udp: outer-ECN RX enabled")
recordCapability("udp.ecn_rx.enabled", true)
}
// recordCapability registers (or updates) a boolean gauge for one of the
// kernel-feature probes. Gauges go to 1 when the feature is enabled, 0 when
// it is not — dashboards can show degraded state on partially-supported
@@ -312,12 +264,6 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
bufSize = udpGROBufferSize
cmsgSpace = unix.CmsgSpace(udpGROCmsgPayload)
}
if u.ecnRecvSupported {
// IP_TOS arrives as 1 byte; IPV6_TCLASS arrives as a 4-byte int.
// Reserve enough for the wider of the two so the same buffer fits
// either family alongside any UDP_GRO cmsg.
cmsgSpace += unix.CmsgSpace(4)
}
msgs, buffers, names, _ := prepareRawMessages(u.batch, bufSize, cmsgSpace)
for {
@@ -330,23 +276,6 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
}
}
// AQM sample: one getsockopt per recvmmsg batch (skipped entirely at
// threshold 0). Sampled BEFORE the read: a single recvmmsg can drain
// more than the whole receive buffer (64 GRO superpackets ≈ 4MB), so
// post-read residue is ~always zero; the pre-read depth is the
// backlog that accumulated while the previous batch was processed —
// the actual standing-queue signal. Depth beyond the configured
// fraction of the receive buffer flags every packet in the batch so
// decap CE-marks ECT inner packets: the ECN substitute for the
// tail-drop this queue otherwise regulates with.
congested := false
if frac := math.Float64frombits(u.ecnMarkThreshold.Load()); frac > 0 {
var mi [unix.SK_MEMINFO_VARS]uint32
if err := u.getMemInfo(&mi); err == nil {
congested = float64(mi[unix.SK_MEMINFO_RMEM_ALLOC]) >= frac*float64(mi[unix.SK_MEMINFO_RCVBUF])
}
}
n, err := u.recvmmsg(msgs)
if err != nil {
if errors.Is(err, unix.EINTR) {
@@ -362,12 +291,11 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
payload := buffers[i][:msgs[i].Len]
segSize := 0
outerECN := byte(0)
if cmsgSpace > 0 {
segSize, outerECN = parseRecvCmsg(&msgs[i].Hdr, u.groSupported, u.ecnRecvSupported)
segSize = parseRecvCmsg(&msgs[i].Hdr)
}
deliverSegments(r, from, payload, segSize, RxMeta{OuterECN: outerECN, QueueCongested: congested})
deliverSegments(r, from, payload, segSize)
}
flush()
@@ -375,9 +303,9 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
}
// deliverSegments hands a received superdatagram to r, splitting it back into pre-coalesce packets
func deliverSegments(r EncReader, from netip.AddrPort, payload []byte, segSize int, meta RxMeta) {
func deliverSegments(r EncReader, from netip.AddrPort, payload []byte, segSize int) {
if segSize <= 0 || segSize >= len(payload) { //avoid bogus values
r(from, payload, meta)
r(from, payload)
return
}
for off := 0; off < len(payload); off += segSize {
@@ -385,25 +313,16 @@ func deliverSegments(r EncReader, from netip.AddrPort, payload []byte, segSize i
if end > len(payload) {
end = len(payload)
}
r(from, payload[off:end], meta)
r(from, payload[off:end])
}
}
// parseRecvCmsg walks the per-slot ancillary buffer once and extracts up to
// two values of interest in a single pass: the UDP_GRO gso_size (when
// wantGRO is true) and the outer IP-level ECN codepoint stamped on the
// carrier (when wantECN is true). Returns zeros for whichever field is not
// requested or not present.
//
// The outer ECN is accepted from EITHER an IP_TOS (IPPROTO_IP, 1-byte) or an
// IPV6_TCLASS (IPPROTO_IPV6, 4-byte int) cmsg, regardless of the socket's
// family: a dual-stack v6 socket (isV4 == false) delivers IPv4 peers' outer
// ECN as an IP_TOS cmsg — gating on socket family here dropped v4-underlay
// ECN entirely. Whichever cmsg the kernel delivered carries the value.
func parseRecvCmsg(hdr *msghdr, wantGRO, wantECN bool) (gso int, ecn byte) {
// parseRecvCmsg walks the per-slot ancillary buffer and extracts the UDP_GRO
// gso_size, or 0 when no UDP_GRO cmsg is present.
func parseRecvCmsg(hdr *msghdr) (gso int) {
controllen := int(hdr.Controllen)
if controllen < unix.SizeofCmsghdr || hdr.Control == nil {
return 0, 0
return 0
}
ctrl := unsafe.Slice(hdr.Control, controllen)
off := 0
@@ -412,37 +331,25 @@ func parseRecvCmsg(hdr *msghdr, wantGRO, wantECN bool) (gso int, ecn byte) {
clen := int(ch.Len)
// Compare against the remaining bytes rather than off+clen
if clen < unix.SizeofCmsghdr || clen > len(ctrl)-off {
return gso, ecn
return gso
}
dataOff := off + unix.CmsgLen(0)
switch {
case wantGRO && ch.Level == unix.SOL_UDP && ch.Type == unix.UDP_GRO:
if ch.Level == unix.SOL_UDP && ch.Type == unix.UDP_GRO {
if dataOff+udpGROCmsgPayload <= len(ctrl) {
gso = int(int32(binary.NativeEndian.Uint32(ctrl[dataOff : dataOff+udpGROCmsgPayload])))
}
case wantECN && ch.Level == unix.IPPROTO_IP && ch.Type == unix.IP_TOS:
// IP_TOS arrives as a single byte; only the low 2 bits are ECN.
// A dual-stack v6 socket carries v4 peers' outer ECN here.
if dataOff+1 <= len(ctrl) {
ecn = ctrl[dataOff] & 0x03
}
case wantECN && ch.Level == unix.IPPROTO_IPV6 && ch.Type == unix.IPV6_TCLASS:
// IPV6_TCLASS arrives as a 4-byte int; ECN is the low 2 bits.
if dataOff+4 <= len(ctrl) {
ecn = byte(binary.NativeEndian.Uint32(ctrl[dataOff:dataOff+4])) & 0x03
}
}
// Advance by the aligned cmsg space.
off += unix.CmsgSpace(clen - unix.CmsgLen(0))
}
return gso, ecn
return gso
}
func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort, ecn byte) error {
return sendmsg(u.sysFd, b, ip, u.isV4, ecn)
func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
return sendto(u.sysFd, b, ip, u.isV4)
}
func sendmsg(fd int, b []byte, addr netip.AddrPort, isV4 bool, ecn byte) error {
func sendto(fd int, b []byte, addr netip.AddrPort, isV4 bool) error {
var rsa [unix.SizeofSockaddrInet6]byte
nlen, err := writeSockaddr(rsa[:], addr, isV4)
if err != nil {
@@ -452,43 +359,25 @@ func sendmsg(fd int, b []byte, addr netip.AddrPort, isV4 bool, ecn byte) 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_SENDMSG,
unix.SYS_SENDTO,
uintptr(fd),
uintptr(unsafe.Pointer(&hdr)),
0, 0, 0, 0,
uintptr(unsafe.Pointer(base)),
uintptr(len(b)),
0,
uintptr(unsafe.Pointer(&rsa[0])),
uintptr(nlen),
)
if errno != 0 {
return &net.OpError{Op: "sendmsg", Err: errno}
return &net.OpError{Op: "sendto", 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.
func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) (int, error) {
return u.bw.WriteBatch(bufs, addrs, ecns)
func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
return u.bw.WriteBatch(bufs, addrs)
}
// writeSockaddr encodes addr into buf (which must be at least SizeofSockaddrInet6 bytes).
@@ -522,8 +411,6 @@ func writeSockaddr(buf []byte, addr netip.AddrPort, isV4 bool) (int, error) {
}
func (u *StdConn) ReloadConfig(c *config.C) {
u.reloadECNMarkThreshold(c)
b := c.GetInt("listen.read_buffer", 0)
if b > 0 {
if err := u.SetRecvBuffer(b); err == nil {
@@ -565,37 +452,6 @@ func (u *StdConn) ReloadConfig(c *config.C) {
}
}
// reloadECNMarkThreshold parses tunnels.ecn_mark_threshold: the fraction
// (0..1] of the receive buffer above which decap CE-marks ECT inner packets.
// 0 (the default) disables the AQM sampling. Reloadable.
func (u *StdConn) reloadECNMarkThreshold(c *config.C) {
var frac float64
switch v := c.Get("tunnels.ecn_mark_threshold").(type) {
case nil:
case float64:
frac = v
case int:
frac = float64(v)
case string:
f, err := strconv.ParseFloat(v, 64)
if err != nil {
u.l.Warn("tunnels.ecn_mark_threshold is not a number; disabling", "value", v)
} else {
frac = f
}
default:
u.l.Warn("tunnels.ecn_mark_threshold is not a number; disabling", "value", v)
}
if frac < 0 || frac > 1 {
u.l.Warn("tunnels.ecn_mark_threshold must be within [0, 1]; disabling", "value", frac)
frac = 0
}
old := math.Float64frombits(u.ecnMarkThreshold.Swap(math.Float64bits(frac)))
if old != frac {
u.l.Info("tunnels.ecn_mark_threshold set", "fraction", frac)
}
}
func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
var vallen uint32 = 4 * unix.SK_MEMINFO_VARS
_, _, err := unix.Syscall6(unix.SYS_GETSOCKOPT, uintptr(u.sysFd), uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
+14 -140
View File
@@ -3,13 +3,11 @@
package udp
import (
"encoding/binary"
"fmt"
"log/slog"
"net"
"net/netip"
"slices"
"syscall"
"testing"
"time"
"unsafe"
@@ -56,39 +54,6 @@ func buildCmsg(level, typ int32, data []byte) []byte {
return buf
}
// TestParseRecvCmsgOuterECNFamily is the RX half of the dual-stack ECN fix:
// parseRecvCmsg must read the outer ECN from whichever family the kernel
// delivered, not from the socket family. On the default `::` dual-stack bind
// a v4 peer's outer ECN arrives as an IP_TOS cmsg, which the old socket-family
// gate ignored entirely.
func TestParseRecvCmsgOuterECNFamily(t *testing.T) {
tc := make([]byte, 4)
binary.NativeEndian.PutUint32(tc, 0x02)
cases := []struct {
name string
ctrl []byte
want byte
}{
{"ip_tos_ce", buildCmsg(int32(unix.IPPROTO_IP), int32(unix.IP_TOS), []byte{0x03}), 0x03},
{"ip_tos_ect0", buildCmsg(int32(unix.IPPROTO_IP), int32(unix.IP_TOS), []byte{0x02}), 0x02},
{"ipv6_tclass_ect0", buildCmsg(int32(unix.IPPROTO_IPV6), int32(unix.IPV6_TCLASS), tc), 0x02},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
hdr := &msghdr{Control: &c.ctrl[0]}
setMsgControllen(hdr, len(c.ctrl))
gso, ecn := parseRecvCmsg(hdr, false, true)
if gso != 0 {
t.Errorf("gso = %d, want 0 (no UDP_GRO cmsg present)", gso)
}
if ecn != c.want {
t.Errorf("ecn = 0x%02x, want 0x%02x", ecn, c.want)
}
})
}
}
func testLogger() *slog.Logger {
return slog.New(slog.DiscardHandler)
}
@@ -125,7 +90,7 @@ func TestWriteBatchBadFamilyDeliversOthers(t *testing.T) {
bufs := [][]byte{[]byte("AAA"), []byte("BBB"), []byte("CCC")}
addrs := []netip.AddrPort{good, bad, good}
n, err := sender.WriteBatch(bufs, addrs, nil)
n, err := sender.WriteBatch(bufs, addrs)
if err != nil {
t.Fatalf("WriteBatch returned error, want nil (bad dest should be isolated): %v", err)
}
@@ -151,93 +116,6 @@ func TestWriteBatchBadFamilyDeliversOthers(t *testing.T) {
}
}
// TestWriteBatchOuterTOSToV4Mapped is the TX half of the dual-stack ECN fix,
// verified against a live kernel: WriteBatch on the default `::` dual-stack
// socket, sending to a v4-mapped destination, must stamp the outer ECN via an
// IP_TOS cmsg (not IPV6_TCLASS, which the kernel's v4 path ignores) so a v4
// receiver actually sees it.
func TestWriteBatchOuterTOSToV4Mapped(t *testing.T) {
rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
if err != nil {
t.Skipf("cannot open v4 receiver (sandbox?): %v", err)
}
defer rx.Close()
rxPort := rx.LocalAddr().(*net.UDPAddr).Port
// Ask the kernel to deliver the received outer TOS as ancillary data.
rxRaw, err := rx.SyscallConn()
if err != nil {
t.Fatalf("SyscallConn: %v", err)
}
var soErr error
if err := rxRaw.Control(func(fd uintptr) {
soErr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_RECVTOS, 1)
}); err != nil || soErr != nil {
t.Skipf("cannot enable IP_RECVTOS (sandbox/kernel?): ctrl=%v so=%v", err, soErr)
}
c, err := NewListener(testLogger(), netip.IPv6Unspecified(), 0, false, 1)
if err != nil {
t.Skipf("cannot open dual-stack sender (sandbox?): %v", err)
}
defer c.Close()
sender := c.(*StdConn)
if sender.isV4 {
t.Skipf("sender came up v4-only; need a dual-stack v6 socket for this test")
}
// v4-mapped-in-v6 destination: routed through the kernel's IPv4 path.
dst := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(rxPort))
const wantECN = byte(0x02) // ECT(0)
if _, err := sender.WriteBatch([][]byte{[]byte("tos-probe")}, []netip.AddrPort{dst}, []byte{wantECN}); err != nil {
t.Fatalf("WriteBatch: %v", err)
}
// Read the datagram plus its ancillary TOS.
rx.SetReadDeadline(time.Now().Add(3 * time.Second))
payload := make([]byte, 128)
oob := make([]byte, 512)
var n, oobn int
var rerr error
if err := rxRaw.Read(func(fd uintptr) bool {
n, oobn, _, _, rerr = unix.Recvmsg(int(fd), payload, oob, 0)
if rerr == syscall.EAGAIN || rerr == syscall.EWOULDBLOCK {
return false
}
return true
}); err != nil {
t.Fatalf("waiting for datagram failed (no delivery?): %v", err)
}
if rerr != nil {
t.Fatalf("Recvmsg: %v", rerr)
}
if string(payload[:n]) != "tos-probe" {
t.Fatalf("payload = %q, want %q", string(payload[:n]), "tos-probe")
}
cmsgs, err := unix.ParseSocketControlMessage(oob[:oobn])
if err != nil {
t.Fatalf("ParseSocketControlMessage: %v", err)
}
found := false
var gotTOS byte
for _, m := range cmsgs {
if m.Header.Level == unix.IPPROTO_IP && m.Header.Type == unix.IP_TOS && len(m.Data) >= 1 {
found = true
gotTOS = m.Data[0]
}
}
if !found {
t.Fatalf("no IP_TOS cmsg delivered to v4 receiver — outer ECN did not land (%d cmsgs)", len(cmsgs))
}
if gotTOS&0x03 != wantECN {
t.Errorf("received outer TOS = 0x%02x, want low-2-bits = 0x%02x", gotTOS, wantECN)
} else {
t.Logf("verified: v4 receiver saw outer TOS 0x%02x (ECN=0x%02x) from dual-stack sender", gotTOS, gotTOS&0x03)
}
}
// TestWriteBatchUnreachableDestDeliversOthers is the kernel-rejection twin of
// TestWriteBatchBadFamilyDeliversOthers. A destination the kernel refuses outright (240.0.0.0/4 is reserved, so
// the send returns EINVAL) fails its sendmmsg entry; WriteBatch must drop only that entry and still deliver
@@ -264,7 +142,7 @@ func TestWriteBatchUnreachableDestDeliversOthers(t *testing.T) {
addrs := []netip.AddrPort{good, good, bad, good, good}
// The bad destination is reported, but only after every other packet has been attempted.
if _, err := sender.WriteBatch(bufs, addrs, nil); err == nil {
if _, err := sender.WriteBatch(bufs, addrs); err == nil {
t.Log("WriteBatch returned nil; kernel accepted the reserved address, delivery assertions still apply")
}
@@ -317,11 +195,11 @@ func TestParseRecvCmsgCorruptLenNoPanic(t *testing.T) {
t.Run(c.name, func(t *testing.T) {
hdr := &msghdr{Control: &c.ctrl[0]}
setMsgControllen(hdr, len(c.ctrl))
gso, ecn := parseRecvCmsg(hdr, true, true)
gso := parseRecvCmsg(hdr)
// The valid leading UDP_GRO cmsg (payload 0) must still parse;
// the corrupt trailer just ends the walk.
if gso != 0 || ecn != 0 {
t.Errorf("parseRecvCmsg = (%d, %#x), want (0, 0)", gso, ecn)
if gso != 0 {
t.Errorf("parseRecvCmsg = %d, want 0", gso)
}
})
}
@@ -367,16 +245,12 @@ func TestDeliverSegments(t *testing.T) {
}
var got [][]byte
meta := RxMeta{OuterECN: 0x2}
deliverSegments(func(a netip.AddrPort, seg []byte, m RxMeta) {
deliverSegments(func(a netip.AddrPort, seg []byte) {
if a != from {
t.Errorf("from = %v, want %v", a, from)
}
if m != meta {
t.Errorf("meta = %+v, want %+v", m, meta)
}
got = append(got, seg)
}, from, c.payload, c.segSize, meta)
}, from, c.payload, c.segSize)
if len(got) != len(wantLens) {
t.Fatalf("delivered %d segments, want %d", len(got), len(wantLens))
@@ -481,7 +355,7 @@ func TestWriteBatchPartialSendRewind(t *testing.T) {
return accept, nil
}
written, err := w.WriteBatch(bufs, addrs, nil)
written, err := w.WriteBatch(bufs, addrs)
if err != nil {
t.Fatalf("WriteBatch: %v", err)
}
@@ -534,7 +408,7 @@ func TestWriteBatchSkipUnroutableRunAccounting(t *testing.T) {
return accept, nil
}
written, err := w.WriteBatch(bufs, addrs, nil)
written, err := w.WriteBatch(bufs, addrs)
if err != nil {
t.Fatalf("WriteBatch: %v", err)
}
@@ -591,7 +465,7 @@ func TestWriteBatchMidChunkRejectResumes(t *testing.T) {
}
}
written, err := w.WriteBatch(bufs, addrs, nil)
written, err := w.WriteBatch(bufs, addrs)
if err != nil {
t.Fatalf("WriteBatch: %v", err)
}
@@ -648,7 +522,7 @@ func TestWriteBatchMidChunkEIODisablesGSOWithoutDup(t *testing.T) {
}
}
written, err := w.WriteBatch(bufs, addrs, nil)
written, err := w.WriteBatch(bufs, addrs)
if err != nil {
t.Fatalf("WriteBatch: %v", err)
}
@@ -676,7 +550,7 @@ func TestWriteBatchZeroProgress(t *testing.T) {
w.sendFn = func(start, n int) (int, error) { return 0, nil }
bufs := [][]byte{make([]byte, 100)}
addrs := []netip.AddrPort{netip.MustParseAddrPort("127.0.0.1:4242")}
if _, err := w.WriteBatch(bufs, addrs, nil); err == nil {
if _, err := w.WriteBatch(bufs, addrs); err == nil {
t.Fatal("WriteBatch = nil error on zero progress, want error")
}
}
@@ -702,7 +576,7 @@ func TestWriteBatchEIODisablesGSOAndReplays(t *testing.T) {
return n, nil
}
written, err := w.WriteBatch(bufs, addrs, nil)
written, err := w.WriteBatch(bufs, addrs)
if err != nil {
t.Fatalf("WriteBatch: %v", err)
}
@@ -773,7 +647,7 @@ func TestGSOEngagesOnLoopback(t *testing.T) {
addrs[i] = dst
}
written, err := sc.WriteBatch(bufs, addrs, nil)
written, err := sc.WriteBatch(bufs, addrs)
if err != nil {
t.Fatalf("WriteBatch: %v", err)
}
+1 -1
View File
@@ -128,7 +128,7 @@ func runTeardownCase(t *testing.T, batch int, name string, traffic func(send net
var received atomic.Int64
loopDone := make(chan error, 1)
go func() {
loopDone <- sc.ListenOut(func(netip.AddrPort, []byte, RxMeta) {
loopDone <- sc.ListenOut(func(netip.AddrPort, []byte) {
received.Add(1)
}, func() {})
}()
+25 -90
View File
@@ -27,17 +27,17 @@ import (
// packet one element of bufs: a single UDP datagram. The unit of the
// returned written count.
// run consecutive packets planRun groups into one entry: same
// destination and outer ECN, equal sizes (a shorter packet only
// last), within maxGSOBytes and maxGSOSegments. Without GSO a run
// is always one packet. Runs are atomic: packed whole into one
// entry, or skipped whole if the socket cannot address their
// destination, leaving a hole (bufs indices covered by no entry).
// destination, equal sizes (a shorter packet only last), within
// maxGSOBytes and maxGSOSegments. Without GSO a run is always one
// packet. Runs are atomic: packed whole into one entry, or
// skipped whole if the socket cannot address their destination,
// leaving a hole (bufs indices covered by no entry).
// entry one mmsghdr slot of the sendmmsg array; the kernel's unit of
// success and failure. A multi-packet entry carries a UDP_SEGMENT
// cmsg and is sent as one superpacket the kernel segments into
// gso_size-byte datagrams. Entries never split.
// chunk the entries packed for one sendmmsg call, at most MaxWriteBatch.
// batch the caller's whole bufs/addrs/ecns triple, processed as one or
// batch the caller's whole bufs/addrs pair, processed as one or
// more chunks.
type batchWriter struct {
fd int
@@ -59,13 +59,10 @@ type batchWriter struct {
names [][]byte
// Per-entry cmsg scratch: one contiguous slab of
// MaxWriteBatch * cmsgSpace bytes holding two cmsg headers per entry
// (UDP_SEGMENT, then IP_TOS / IPV6_TCLASS). Layout in
// prepareWriteMessages.
cmsg []byte
cmsgSpace int
cmsgSegSpace int
cmsgEcnSpace int
// MaxWriteBatch * cmsgSpace bytes holding one UDP_SEGMENT cmsg per
// entry. Layout in prepareWriteMessages.
cmsg []byte
cmsgSpace int
// entryEnd[e] is the bufs index after the last packet packed into entry
// e. entryEnd[e]-entryPkts[e] recovers the bufs index the entry's run
@@ -91,17 +88,11 @@ func newBatchWriter(fd int, isV4 bool, l *slog.Logger) *batchWriter {
// prepareWriteMessages allocates the per-entry mmsghdr/iovec/sockaddr/cmsg
// scratch. Hdr.Iov/Iovlen/Control/Controllen are wired per call, since an
// entry spans a variable number of iovecs and may or may not carry cmsgs.
// entry spans a variable number of iovecs and may or may not carry a cmsg.
//
// Each entry's cmsg slot holds up to two headers at fixed offsets:
//
// [0 .. cmsgSegSpace) UDP_SEGMENT (gso_size, uint16)
// [cmsgSegSpace .. cmsgSpace) IP_TOS or IPV6_TCLASS (int32)
//
// The UDP_SEGMENT header is pre-filled here; only its payload is rewritten
// per call. The ECN header is written per entry by writeEntryCmsg because
// its Level/Type follow the destination's family. Hdr.Control/Controllen
// select whichever subset applies (none / segment / ecn / both).
// Each entry's cmsg slot holds one UDP_SEGMENT (gso_size, uint16) header,
// pre-filled here; only its payload is rewritten per call.
// Hdr.Control/Controllen select whether it applies (none / segment).
func (w *batchWriter) prepareWriteMessages(n int) {
w.msgs = make([]rawMessage, n)
w.iovs = make([]iovec, n)
@@ -109,9 +100,7 @@ func (w *batchWriter) prepareWriteMessages(n int) {
w.entryEnd = make([]int, n)
w.entryPkts = make([]int, n)
w.cmsgSegSpace = unix.CmsgSpace(2)
w.cmsgEcnSpace = unix.CmsgSpace(4)
w.cmsgSpace = w.cmsgSegSpace + w.cmsgEcnSpace
w.cmsgSpace = unix.CmsgSpace(2)
w.cmsg = make([]byte, n*w.cmsgSpace)
for k := 0; k < n; k++ {
@@ -197,13 +186,10 @@ func parseRelease(r string) (major, minor int) {
//
// Returns the number of packets sent. An error means the call itself
// failed; a short count means some destinations were undeliverable.
func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) (int, error) {
func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
if len(bufs) != len(addrs) {
return 0, fmt.Errorf("WriteBatch: len(bufs)=%d != len(addrs)=%d", len(bufs), len(addrs))
}
if ecns != nil && len(ecns) != len(bufs) {
return 0, fmt.Errorf("WriteBatch: len(ecns)=%d != len(bufs)=%d", len(ecns), len(bufs))
}
// Callers deliver same-destination packets contiguously and in counter order, so we run the GSO planner directly without a pre-sort.
// A sorting pass measurably hurt throughput in microbenchmarks while providing no observed reordering benefit.
@@ -221,7 +207,7 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b
if iovBudget < 1 {
break
}
runLen, segSize := w.planRun(bufs, addrs, ecns, i, iovBudget)
runLen, segSize := w.planRun(bufs, addrs, i, iovBudget)
if runLen == 0 {
break
}
@@ -254,13 +240,7 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b
setMsgIovlen(hdr, runLen)
hdr.Namelen = uint32(nlen)
var ecn byte
if ecns != nil {
ecn = ecns[i]
}
// ECN cmsg family follows the destination, not the socket
dstIsV4 := addrs[i].Addr().Unmap().Is4()
w.writeEntryCmsg(entry, runLen, segSize, ecn, dstIsV4)
w.writeEntryCmsg(entry, runLen, segSize)
i += runLen
iovIdx += runLen
@@ -340,9 +320,8 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b
// planRun returns the length of the run starting at start and its segment
// size (len(bufs[start])). A run of length 1 carries no UDP_SEGMENT cmsg
// and is sent as a plain datagram; without GSO support planRun always
// returns 1. Outer ECN is a run boundary: the kernel stamps one codepoint per entry.
func (w *batchWriter) planRun(bufs [][]byte, addrs []netip.AddrPort, ecns []byte, start, iovBudget int) (int, int) {
// and is sent as a plain datagram; without GSO support planRun always returns 1.
func (w *batchWriter) planRun(bufs [][]byte, addrs []netip.AddrPort, start, iovBudget int) (int, int) {
if start >= len(bufs) || iovBudget < 1 {
return 0, 0
}
@@ -351,10 +330,6 @@ func (w *batchWriter) planRun(bufs [][]byte, addrs []netip.AddrPort, ecns []byte
return 1, segSize
}
dst := addrs[start]
var ecn byte
if ecns != nil {
ecn = ecns[start]
}
maxLen := w.maxGSOSegments
if iovBudget < maxLen {
maxLen = iovBudget
@@ -369,9 +344,6 @@ func (w *batchWriter) planRun(bufs [][]byte, addrs []netip.AddrPort, ecns []byte
if addrs[start+runLen] != dst {
break
}
if ecns != nil && ecns[start+runLen] != ecn {
break
}
if total+nextLen > maxGSOBytes {
break
}
@@ -385,55 +357,18 @@ 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.
func (w *batchWriter) writeEntryCmsg(entry, runLen, segSize int, ecn byte, dstIsV4 bool) {
// writeEntryCmsg writes one entry's UDP_SEGMENT payload when runLen >= 2 and
// points Hdr.Control at it; a single-packet entry carries no cmsg.
func (w *batchWriter) writeEntryCmsg(entry, runLen, segSize int) {
hdr := &w.msgs[entry].Hdr
useSeg := runLen >= 2
useEcn := ecn != 0
base := entry * w.cmsgSpace
if useSeg {
if runLen >= 2 {
dataOff := base + unix.CmsgLen(0)
binary.NativeEndian.PutUint16(w.cmsg[dataOff:dataOff+2], uint16(segSize))
}
if useEcn {
writeECNCmsg(w.cmsg[base+w.cmsgSegSpace:], dstIsV4, ecn)
}
switch {
case useSeg && useEcn:
hdr.Control = &w.cmsg[base]
setMsgControllen(hdr, w.cmsgSpace)
case useSeg:
hdr.Control = &w.cmsg[base]
setMsgControllen(hdr, w.cmsgSegSpace)
case useEcn:
hdr.Control = &w.cmsg[base+w.cmsgSegSpace]
setMsgControllen(hdr, w.cmsgEcnSpace)
default:
} else {
hdr.Control = nil
setMsgControllen(hdr, 0)
}
+19 -70
View File
@@ -11,7 +11,7 @@ import (
// no per-packet heap allocations on the happy path: all mmsghdr/iovec/cmsg
// scratch is preallocated in newBatchWriter and WriteBatch may only rewrite
// it. The batch deliberately mixes a GSO-eligible run, a short tail segment,
// destination changes, and zero/nonzero outer ECN so the planner, sockaddr,
// destination changes, so the planner, sockaddr,
// and cmsg paths are all exercised.
func TestWriteBatchNoAllocs(t *testing.T) {
for _, tc := range []struct {
@@ -53,91 +53,40 @@ func TestWriteBatchNoAllocs(t *testing.T) {
var bufs [][]byte
var addrs []netip.AddrPort
var ecns []byte
add := func(b []byte, dst netip.AddrPort, ecn byte) {
add := func(b []byte, dst netip.AddrPort) {
bufs = append(bufs, b)
addrs = append(addrs, dst)
ecns = append(ecns, ecn)
}
// GSO-eligible run with a short tail, all ECT(0).
// GSO-eligible run with a short tail.
for k := 0; k < 8; k++ {
add(payload, dstA, 0b10)
add(payload, dstA)
}
add(short, dstA, 0b10)
// ECN change on the same destination forces a run boundary.
add(payload, dstA, 0)
add(short, dstA)
add(payload, dstA)
// Alternating destinations defeat coalescing entirely.
for k := 0; k < 4; k++ {
dst := dstA
if k%2 == 0 {
dst = dstB
}
add(payload, dst, 0)
add(payload, dst)
}
send := func(ecns []byte) {
t.Helper()
var werr error
// Warm-up outside the measured runs.
if _, err := tx.WriteBatch(bufs, addrs, ecns); err != nil {
t.Fatalf("WriteBatch warm-up: %v", err)
}
allocs := testing.AllocsPerRun(100, func() {
if _, err := tx.WriteBatch(bufs, addrs, ecns); err != nil {
werr = err
}
})
if werr != nil {
t.Fatalf("WriteBatch: %v", werr)
}
if allocs != 0 {
t.Fatalf("WriteBatch allocated %.1f times per call, want 0", allocs)
}
var werr error
// Warm-up outside the measured runs.
if _, err := tx.WriteBatch(bufs, addrs); err != nil {
t.Fatalf("WriteBatch warm-up: %v", err)
}
send(ecns)
send(nil)
})
}
}
// 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)
allocs := testing.AllocsPerRun(100, func() {
if _, err := tx.WriteBatch(bufs, addrs); err != nil {
werr = err
}
t.Cleanup(func() { _ = c.Close() })
return c
})
if werr != nil {
t.Fatalf("WriteBatch: %v", werr)
}
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)
}
if allocs != 0 {
t.Fatalf("WriteBatch allocated %.1f times per call, want 0", allocs)
}
})
}
+4 -5
View File
@@ -161,7 +161,7 @@ func (u *RIOConn) ListenOut(r EncReader, flush func()) error {
continue
}
r(netip.AddrPortFrom(netip.AddrFrom16(rua.Addr).Unmap(), (rua.Port>>8)|((rua.Port&0xff)<<8)), buffer[:n], RxMeta{})
r(netip.AddrPortFrom(netip.AddrFrom16(rua.Addr).Unmap(), (rua.Port>>8)|((rua.Port&0xff)<<8)), buffer[:n])
flush()
}
}
@@ -254,8 +254,7 @@ retry:
return n, ep, nil
}
// WriteTo ignores outerECN; per-packet ECN marking is not implemented on windows.
func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort, _ byte) error {
func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort) error {
if !u.isOpen.Load() {
return net.ErrClosed
}
@@ -318,11 +317,11 @@ func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort, _ byte) error {
return winrio.SendEx(u.rq, dataBuffer, 1, nil, addressBuffer, nil, nil, 0, 0)
}
func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
// 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], 0); err == nil {
if err := u.WriteTo(b, addrs[i]); err == nil {
written++
} else {
u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
+4 -5
View File
@@ -153,8 +153,7 @@ func (u *TesterConn) Get(block bool) *Packet {
// Below this is boilerplate implementation to make nebula actually work
//********************************************************************************************************************//
// WriteTo ignores outerECN; the in-memory tester carries no IP headers.
func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort, _ byte) error {
func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
p := acquirePacket()
if cap(p.Data) < len(b) {
p.Data = make([]byte, len(b))
@@ -172,10 +171,10 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort, _ byte) error {
return nil
}
}
func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
written := 0
for i, b := range bufs {
if err := u.WriteTo(b, addrs[i], 0); err == nil {
if err := u.WriteTo(b, addrs[i]); err == nil {
written++
} else {
u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
@@ -190,7 +189,7 @@ func (u *TesterConn) ListenOut(r EncReader, flush func()) error {
case <-u.done:
return os.ErrClosed
case p := <-u.RxPackets:
r(p.From, p.Data, RxMeta{})
r(p.From, p.Data)
// The batcher borrows plaintext decrypted in place inside p.Data
// until Flush, so the packet must stay alive across flush()
flush()