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

make it nicer

Browse Source
This commit is contained in:
JackDoan
2026-07-18 00:01:08 -05:00
parent 9b37b2023d
commit 59ecea92ce
7 changed files with 645 additions and 599 deletions
Download Patch File Download Diff File Expand all files Collapse all files
udp/udp_ecn_outer_linux_test.go
+1 -1
View File
@@ -12,7 +12,7 @@ import (
// kernel stamps one outer codepoint per entry, so a run that straddled the
// boundary would silently lose information).
func TestPlanRunBreaksOnECNChange(t *testing.T) {
u := &StdConn{gsoSupported: true, maxGSOSegments: 63}
u := &batchWriter{gsoSupported: true, maxGSOSegments: 63}
dst := netip.MustParseAddrPort("10.0.0.1:4242")
bufs := [][]byte{
udp/udp_linux.go
+92 -529
View File
@@ -10,8 +10,6 @@ import (
"log/slog"
"net"
"net/netip"
"strconv"
"strings"
"sync/atomic"
"syscall"
"unsafe"
@@ -28,37 +26,13 @@ type StdConn struct {
l *slog.Logger
batch int
// sendmmsg scratch. Each queue has its own StdConn, so no locking is
// needed. Sized to MaxWriteBatch at construction; WriteBatch chunks
// larger inputs.
writeMsgs []rawMessage
writeIovs []iovec
writeNames [][]byte
// Per-entry cmsg scratch. writeCmsg is one contiguous slab of
// MaxWriteBatch * writeCmsgSpace bytes; each entry holds two cmsg
// headers (UDP_SEGMENT then IP_TOS / IPV6_TCLASS) pre-filled once in
// prepareWriteMessages. WriteBatch only rewrites the per-call data
// payloads and toggles Hdr.Control / Hdr.Controllen to point at
// whichever subset of the two cmsgs applies.
writeCmsg []byte
writeCmsgSpace int
writeCmsgSegSpace int
writeCmsgEcnSpace int
// writeEntryEnd[e] is the bufs index *after* the last packet packed
// into mmsghdr entry e. Used to rewind `i` on partial sendmmsg success.
writeEntryEnd []int
// UDP GSO (sendmsg with UDP_SEGMENT cmsg) support. gsoSupported is
// probed once at socket creation. When true, WriteBatch packs same-
// destination consecutive packets into a single sendmmsg entry with a
// UDP_SEGMENT cmsg; otherwise each packet is its own entry.
gsoSupported bool
maxGSOSegments int
// bw owns the sendmmsg/UDP-GSO transmit path: the per-queue write
// scratch and the GSO capability state probed at socket creation. See
// udp_linux_writebatch.go.
bw *batchWriter
// UDP GRO (recvmsg with UDP_GRO cmsg) support. groSupported is probed
// once at socket creation. When true, listenOutBatch allocates larger
// once at socket creation. When true, ListenOut allocates larger
// RX buffers and a per-entry cmsg slot so the kernel can coalesce
// consecutive same-flow datagrams into a single recvmmsg entry; the
// delivered cmsg carries the gso_size used to split them back apart.
@@ -66,7 +40,7 @@ type StdConn struct {
// 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 listenOutBatch passes
// each arriving datagram as a per-slot cmsg, and ListenOut passes
// the parsed value to the EncReader callback for RFC 6040 combine.
ecnRecvSupported bool
}
@@ -110,12 +84,12 @@ func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int)
out := &StdConn{sysFd: fd, isV4: ip.Is4(), l: l, batch: batch}
out.prepareWriteMessages(MaxWriteBatch)
out.bw = newBatchWriter(fd, out.isV4, l)
out.prepareGSO()
// GRO delivers coalesced superpackets that need a cmsg to split back
// into segments. The single-packet RX path uses ReadFromUDPAddrPort
// and cannot see that cmsg, so only enable GRO for the batch path.
// GRO coalesces same-flow datagrams into superpackets that must be split
// back apart via the delivered gso_size cmsg. batch == 1 means the caller
// wants plain single-datagram reads with MTU-sized buffers, so leave it
// off there.
if batch > 1 {
out.prepareGRO()
}
@@ -128,110 +102,6 @@ func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int)
return out, nil
}
// prepareWriteMessages allocates one mmsghdr/iovec/sockaddr/cmsg scratch
// slot per sendmmsg entry. The iovec slab is sized to n so all entries'
// iovecs share one allocation; per-entry fan-out is further capped at
// maxGSOSegments. Hdr.Iov / Hdr.Iovlen / Hdr.Control / Hdr.Controllen are
// wired per call since each entry can span a variable number of iovecs
// and may or may not carry a cmsg.
//
// Per-mmsghdr cmsg layout. Each entry's slot of length writeCmsgSpace holds
// up to two cmsg headers placed at fixed offsets:
//
// [0 .. writeCmsgSegSpace) UDP_SEGMENT (gso_size, uint16)
// [writeCmsgSegSpace .. writeCmsgSpace) IP_TOS or IPV6_TCLASS (int32)
//
// Both headers are pre-filled once here; per-call we only rewrite the data
// payload and toggle Hdr.Control / Hdr.Controllen to point at whichever
// subset applies (none / segment-only / ecn-only / both).
func (u *StdConn) prepareWriteMessages(n int) {
u.writeMsgs = make([]rawMessage, n)
u.writeIovs = make([]iovec, n)
u.writeNames = make([][]byte, n)
u.writeEntryEnd = make([]int, n)
u.writeCmsgSegSpace = unix.CmsgSpace(2)
u.writeCmsgEcnSpace = unix.CmsgSpace(4)
u.writeCmsgSpace = u.writeCmsgSegSpace + u.writeCmsgEcnSpace
u.writeCmsg = make([]byte, n*u.writeCmsgSpace)
// Default the ECN header to the socket's own family. writeEntryCmsg
// finalizes Level/Type per entry from the destination address (a v4-mapped
// dst on a dual-stack v6 socket needs IP_TOS, not IPV6_TCLASS), so this is
// only the value used before the first per-entry rewrite.
ecnLevel := int32(unix.IPPROTO_IP)
ecnType := int32(unix.IP_TOS)
if !u.isV4 {
ecnLevel = unix.IPPROTO_IPV6
ecnType = unix.IPV6_TCLASS
}
for k := 0; k < n; k++ {
base := k * u.writeCmsgSpace
seg := (*unix.Cmsghdr)(unsafe.Pointer(&u.writeCmsg[base]))
seg.Level = unix.SOL_UDP
seg.Type = unix.UDP_SEGMENT
setCmsgLen(seg, unix.CmsgLen(2))
ecn := (*unix.Cmsghdr)(unsafe.Pointer(&u.writeCmsg[base+u.writeCmsgSegSpace]))
ecn.Level = ecnLevel
ecn.Type = ecnType
setCmsgLen(ecn, unix.CmsgLen(4))
}
for i := range u.writeMsgs {
u.writeNames[i] = make([]byte, unix.SizeofSockaddrInet6)
u.writeMsgs[i].Hdr.Name = &u.writeNames[i][0]
}
}
// maxGSOBytes bounds the total payload per sendmsg() when UDP_SEGMENT is
// set. The kernel stitches all iovecs into a single skb whose length the
// UDP length field can represent, and also enforces sk_gso_max_size (which
// on most devices is 65536). We use 65000 to leave headroom under the
// 65535 UDP-length cap, avoiding EMSGSIZE on large TSO superpackets.
const maxGSOBytes = 65000
// prepareGSO probes UDP_SEGMENT support and sets u.gsoSupported on success.
// Best-effort; failure leaves it false.
func (u *StdConn) prepareGSO() {
u.maxGSOSegments = 63 //gotta be one less than the max so we can still attach a header
if err := unix.SetsockoptInt(u.sysFd, unix.IPPROTO_UDP, unix.UDP_SEGMENT, 0); err != nil {
u.l.Info("udp: GSO disabled", "reason", "rawconn control failed", "error", err)
recordCapability("udp.gso.enabled", false)
return
}
var un unix.Utsname
if err := unix.Uname(&un); err != nil {
u.l.Info("udp: GSO disabled", "reason", "kernel uname probe failed", "error", err)
recordCapability("udp.gso.enabled", false)
return
}
u.maxGSOSegments = gsoMaxSegments(string(un.Release[:]))
u.gsoSupported = true
u.l.Info("udp: GSO enabled", "maxGSOSegments", u.maxGSOSegments)
recordCapability("udp.gso.enabled", true)
}
// gsoMaxSegments returns the largest number of UDP_SEGMENT segments a single
// sendmsg may carry on the running kernel, reserving one segment for the
// header. UDP_MAX_SEGMENTS was 64 until Linux v6.9 (commit 1382e3b6a350,
// "udp: change maximum number of UDP segments to 128") raised it to 128;
// nothing about this changed in 5.5. On kernels older than 6.9 packing more
// than 64 segments gets the sendmsg rejected with EINVAL, so cap at 63 there
// and only use 127 from 6.9 on. (Maintainer stance: update your kernel if you
// want to go fast — this is a plain version gate, not a runtime probe.)
func gsoMaxSegments(release string) int {
major, minor := parseRelease(release)
if major > 6 || (major == 6 && minor >= 9) {
return 127
}
return 63
}
// udpGROBufferSize sizes the per-entry recvmmsg buffer when UDP_GRO is on.
// The kernel stitches a run of same-flow datagrams into a single skb whose
// length is bounded by sk_gso_max_size (typically 65535); anything larger
@@ -261,40 +131,30 @@ func (u *StdConn) prepareGRO() {
// 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. listenOutBatch 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.
// 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. Only the socket's own family gates support; on a
// dual-stack v6 socket a failed IPv4 probe just degrades v4 peers to Not-ECT
// (could be a v6-specific bind).
func (u *StdConn) prepareECNRecv() {
var v4err, v6err error
v4err = unix.SetsockoptInt(u.sysFd, unix.IPPROTO_IP, unix.IP_RECVTOS, 1)
v4err := unix.SetsockoptInt(u.sysFd, unix.IPPROTO_IP, unix.IP_RECVTOS, 1)
err := v4err
if !u.isV4 {
v6err = unix.SetsockoptInt(u.sysFd, unix.IPPROTO_IPV6, unix.IPV6_RECVTCLASS, 1)
}
if u.isV4 { //only check the V4 attempt
if v4err != nil {
u.l.Info("udp: outer-ECN RX disabled", "reason", "kernel rejected probe", "error", v4err)
recordCapability("udp.ecn_rx.enabled", false)
} else {
u.ecnRecvSupported = true
u.l.Info("udp: outer-ECN RX enabled")
recordCapability("udp.ecn_rx.enabled", true)
}
return
} else {
if v6err != nil { //no V6 ECN? disable it.
u.l.Info("udp: outer-ECN RX disabled", "reason", "kernel rejected probe", "error", errors.Join(v4err, v6err))
recordCapability("udp.ecn_rx.enabled", false)
return
} else if v4err != nil { //no V4, but yes V6? Low level warning. Could be a V6-specific bind.
err = unix.SetsockoptInt(u.sysFd, unix.IPPROTO_IPV6, unix.IPV6_RECVTCLASS, 1)
if err != nil {
err = errors.Join(v4err, err)
} else if v4err != nil {
u.l.Debug("udp: outer-ECN RX degraded", "reason", "kernel rejected probe on IPv4", "error", v4err)
}
// all good
u.ecnRecvSupported = true
u.l.Info("udp: outer-ECN RX enabled")
recordCapability("udp.ecn_rx.enabled", true)
}
if err != nil {
u.l.Info("udp: outer-ECN RX disabled", "reason", "kernel rejected probe", "error", err)
recordCapability("udp.ecn_rx.enabled", false)
return
}
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
@@ -358,7 +218,9 @@ func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
}
}
// recvmmsg does one blocking recvmmsg (MSG_WAITFORONE), reading up to len(msgs) datagrams
// recvmmsg does one blocking recvmmsg (MSG_WAITFORONE), reading up to len(msgs)
// datagrams. With len(msgs) == 1 it degenerates to a plain single-datagram
// read (the kernel implements recvmmsg as a recvmsg loop).
func (u *StdConn) recvmmsg(msgs []rawMessage) (int, error) {
r, _, errno := unix.Syscall6(
unix.SYS_RECVMMSG,
@@ -382,28 +244,40 @@ func (u *StdConn) recvmmsg(msgs []rawMessage) (int, error) {
return n, nil
}
// recvmsg does one blocking recvmsg into msgs[0]
func (u *StdConn) recvmsg(msgs []rawMessage) (int, error) {
r, _, errno := unix.Syscall6(
unix.SYS_RECVMSG,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&msgs[0].Hdr)),
0,
0,
0,
0,
)
if errno != 0 {
if u.closed.Load() {
return 0, net.ErrClosed
// prepareRawMessages allocates the recvmmsg scratch: n rawMessages, each
// wired to its own bufSize receive buffer, sockaddr name slot and — when
// cmsgSpace > 0 — a slice of one contiguous ancillary-data slab. All iovecs
// share a single slab kept alive by the msghdrs that point into it.
func prepareRawMessages(n, bufSize, cmsgSpace int) ([]rawMessage, [][]byte, [][]byte, []byte) {
msgs := make([]rawMessage, n)
buffers := make([][]byte, n)
names := make([][]byte, n)
iovs := make([]iovec, n)
var cmsgs []byte
if cmsgSpace > 0 {
cmsgs = make([]byte, n*cmsgSpace)
}
for i := range msgs {
buffers[i] = make([]byte, bufSize)
names[i] = make([]byte, unix.SizeofSockaddrInet6)
iovs[i].Base = &buffers[i][0]
setIovLen(&iovs[i], bufSize)
msgs[i].Hdr.Iov = &iovs[i]
setMsgIovlen(&msgs[i].Hdr, 1)
msgs[i].Hdr.Name = &names[i][0]
msgs[i].Hdr.Namelen = uint32(len(names[i]))
if cmsgSpace > 0 {
msgs[i].Hdr.Control = &cmsgs[i*cmsgSpace]
setMsgControllen(&msgs[i].Hdr, cmsgSpace)
}
return 0, &net.OpError{Op: "recvmsg", Err: errno}
}
if r == 0 && u.closed.Load() {
return 0, net.ErrClosed
}
msgs[0].Len = uint32(r)
return 1, nil
return msgs, buffers, names, cmsgs
}
func getFrom(names [][]byte, i int, isV4 bool) netip.AddrPort {
@@ -430,12 +304,7 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
// either family alongside any UDP_GRO cmsg.
cmsgSpace += unix.CmsgSpace(4)
}
msgs, buffers, names, _ := u.PrepareRawMessages(u.batch, bufSize, cmsgSpace)
read := u.recvmmsg
if u.batch == 1 {
read = u.recvmsg
}
msgs, buffers, names, _ := prepareRawMessages(u.batch, bufSize, cmsgSpace)
for {
if cmsgSpace > 0 {
@@ -443,7 +312,7 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
setMsgControllen(&msgs[i].Hdr, cmsgSpace)
}
}
n, err := read(msgs)
n, err := u.recvmmsg(msgs)
if err != nil {
if errors.Is(err, unix.EINTR) {
continue // interrupted by a signal, retry the read
@@ -530,326 +399,39 @@ func parseRecvCmsg(hdr *msghdr, wantGRO, wantECN bool) (gso int, ecn byte) {
}
func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
if u.isV4 {
return u.writeTo4(b, ip)
}
return u.writeTo6(b, ip)
return sendto(u.sysFd, b, ip, u.isV4)
}
func (u *StdConn) writeTo6(b []byte, ip netip.AddrPort) error {
var rsa unix.RawSockaddrInet6
rsa.Family = unix.AF_INET6
rsa.Addr = ip.Addr().As16()
binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], ip.Port())
for {
_, _, err := unix.Syscall6(
unix.SYS_SENDTO,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&b[0])),
uintptr(len(b)),
uintptr(0),
uintptr(unsafe.Pointer(&rsa)),
uintptr(unix.SizeofSockaddrInet6),
)
if err != 0 {
return &net.OpError{Op: "sendto", Err: err}
}
return nil
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 {
return err
}
}
func (u *StdConn) writeTo4(b []byte, ip netip.AddrPort) error {
if !ip.Addr().Is4() {
return ErrInvalidIPv6RemoteForSocket
var base *byte
if len(b) > 0 {
base = &b[0]
}
var rsa unix.RawSockaddrInet4
rsa.Family = unix.AF_INET
rsa.Addr = ip.Addr().As4()
binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], ip.Port())
for {
_, _, err := unix.Syscall6(
unix.SYS_SENDTO,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&b[0])),
uintptr(len(b)),
uintptr(0),
uintptr(unsafe.Pointer(&rsa)),
uintptr(unix.SizeofSockaddrInet4),
)
if err != 0 {
return &net.OpError{Op: "sendto", Err: err}
}
return nil
}
}
// WriteBatch sends bufs via sendmmsg(2) using the preallocated scratch on
// StdConn. Consecutive packets to the same destination with matching segment
// sizes (all but possibly the last) are coalesced into a single mmsghdr entry
// carrying a UDP_SEGMENT cmsg, so one syscall can mix runs of GSO superpackets
// with plain one-off datagrams. Without GSO support every packet is its own
// entry, matching the prior behaviour.
//
// Chunks larger than the scratch are processed across multiple syscalls. If
// sendmmsg returns an error AND zero entries went out we fall back to
// per-packet WriteTo for that chunk so the caller still gets best-effort
// delivery; on a partial-success error we just replay the remainder.
func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) error {
if len(bufs) != len(addrs) {
return fmt.Errorf("WriteBatch: len(bufs)=%d != len(addrs)=%d", len(bufs), len(addrs))
}
if ecns != nil && len(ecns) != len(bufs) {
return 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.
i := 0
sendChunks:
for i < len(bufs) {
baseI := i
entry := 0
iovIdx := 0
for entry < len(u.writeMsgs) && i < len(bufs) {
iovBudget := len(u.writeIovs) - iovIdx
if iovBudget < 1 {
break
}
runLen, segSize := u.planRun(bufs, addrs, ecns, i, iovBudget)
if runLen == 0 {
break
}
for k := 0; k < runLen; k++ {
b := bufs[i+k]
if len(b) == 0 {
u.writeIovs[iovIdx+k].Base = nil
setIovLen(&u.writeIovs[iovIdx+k], 0)
} else {
u.writeIovs[iovIdx+k].Base = &b[0]
setIovLen(&u.writeIovs[iovIdx+k], len(b))
}
}
nlen, err := writeSockaddr(u.writeNames[entry], addrs[i], u.isV4)
if err != nil {
// One destination in this chunk has an address family the
// socket can't send to (e.g. an IPv6 remote on a v4-bound
// socket → ErrInvalidIPv6RemoteForSocket). Abandoning the whole
// sendmmsg here would drop every packet already packed for this
// chunk plus every packet still ahead of us in bufs. Instead
// fall back to per-packet WriteTo for the packets packed so far
// in this chunk and the offending one: WriteTo delivers each
// good destination and only errors on the bad one, which we
// drop and keep going. One bad destination costs one packet,
// never the batch. (Same fallback the zero-sent sendmmsg path
// below uses, extended to cover the misaddressed packet.)
for k := baseI; k <= i; k++ {
if werr := u.WriteTo(bufs[k], addrs[k]); werr != nil && k != i {
return werr
}
}
i++
continue sendChunks
}
hdr := &u.writeMsgs[entry].Hdr
hdr.Iov = &u.writeIovs[iovIdx]
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: a
// v4-mapped dst on a dual-stack v6 socket must be stamped via
// IP_TOS. addrs[i] is this run's destination (i advances below).
dstIsV4 := addrs[i].Addr().Unmap().Is4()
u.writeEntryCmsg(entry, runLen, segSize, ecn, dstIsV4)
i += runLen
iovIdx += runLen
u.writeEntryEnd[entry] = i
entry++
}
if entry == 0 {
return fmt.Errorf("sendmmsg: no progress")
}
sent, serr := u.sendmmsg(entry)
if serr != nil && sent <= 0 {
// Nothing went out for this chunk; fall back to WriteTo for each
// packet that was queued this iteration. We only enter this path
// when sendmmsg returned an error AND zero entries succeeded —
// otherwise the partial-success advance below replays only the
// remainder, avoiding duplicates of already-sent packets.
//
// sent=-1 from sendmmsg means message 0 itself failed (partial
// success returns the count instead), so log entry 0's parameters
// — that's the entry the kernel rejected.
hdr0 := &u.writeMsgs[0].Hdr
runLen0 := u.writeEntryEnd[0] - baseI
seg0 := len(bufs[baseI])
ecn0 := byte(0)
if ecns != nil {
ecn0 = ecns[baseI]
}
u.l.Warn("sendmmsg had problem",
"sent", sent, "err", serr,
"entries", entry,
"entry0_runLen", runLen0,
"entry0_segSize", seg0,
"entry0_iovlen", hdr0.Iovlen,
"entry0_controllen", hdr0.Controllen,
"entry0_namelen", hdr0.Namelen,
"entry0_ecn", ecn0,
"entry0_dst", addrs[baseI],
"isV4", u.isV4,
"gso", u.gsoSupported,
"gro", u.groSupported,
)
for k := baseI; k < i; k++ {
if werr := u.WriteTo(bufs[k], addrs[k]); werr != nil {
return werr
}
}
continue
}
if sent == 0 {
return fmt.Errorf("sendmmsg made no progress")
}
// Rewind i to the end of the last successfully sent entry. For a
// full-success send this leaves i unchanged; for a partial send it
// replays the remainder on the next outer-loop iteration.
i = u.writeEntryEnd[sent-1]
_, _, errno := unix.Syscall6(
unix.SYS_SENDTO,
uintptr(fd),
uintptr(unsafe.Pointer(base)),
uintptr(len(b)),
0,
uintptr(unsafe.Pointer(&rsa[0])),
uintptr(nlen),
)
if errno != 0 {
return &net.OpError{Op: "sendto", Err: errno}
}
return nil
}
// planRun groups consecutive packets starting at `start` that can be sent as
// a single UDP GSO superpacket (one sendmmsg entry with UDP_SEGMENT cmsg).
// A run of length 1 means the entry carries no UDP_SEGMENT cmsg and the
// kernel treats it as a plain datagram. Returns the run length and the
// per-segment size (which equals len(bufs[start])). Without GSO support
// every call returns runLen=1. Outer ECN (when ecns != nil) is also a run
// boundary — the kernel stamps one outer codepoint per sendmsg entry, so
// mixing values inside a run would lose information.
func (u *StdConn) planRun(bufs [][]byte, addrs []netip.AddrPort, ecns []byte, start, iovBudget int) (int, int) {
if start >= len(bufs) || iovBudget < 1 {
return 0, 0
}
segSize := len(bufs[start])
if !u.gsoSupported || segSize == 0 || segSize > maxGSOBytes {
return 1, segSize
}
dst := addrs[start]
var ecn byte
if ecns != nil {
ecn = ecns[start]
}
maxLen := u.maxGSOSegments
if iovBudget < maxLen {
maxLen = iovBudget
}
runLen := 1
total := segSize
for runLen < maxLen && start+runLen < len(bufs) {
nextLen := len(bufs[start+runLen])
if nextLen == 0 || nextLen > segSize {
break
}
if addrs[start+runLen] != dst {
break
}
if ecns != nil && ecns[start+runLen] != ecn {
break
}
if total+nextLen > maxGSOBytes {
break
}
total += nextLen
runLen++
if nextLen < segSize {
// A short packet must be the last in the run.
break
}
}
return runLen, segSize
}
// writeEntryCmsg sets up the per-mmsghdr Hdr.Control / Hdr.Controllen for one
// entry. It writes the UDP_SEGMENT payload when runLen >= 2 and the
// IP_TOS/IPV6_TCLASS payload when ecn != 0, then points hdr.Control at the
// smallest contiguous span that covers whichever cmsg(s) actually apply.
//
// The outer-ECN cmsg family must match the *destination*, not the socket: on
// the default dual-stack v6 bind, a v4-mapped destination is routed through
// the kernel's IPv4 path, which parses IP_TOS (IPPROTO_IP) and ignores an
// IPV6_TCLASS cmsg. prepareWriteMessages pre-fills a default header; here we
// rewrite its Level/Type (and Len) per entry from dstIsV4 so v4 peers get
// IP_TOS and v6 peers get IPV6_TCLASS. The data payload is a 4-byte int for
// both families, so the pre-computed cmsg space is unchanged.
func (u *StdConn) writeEntryCmsg(entry, runLen, segSize int, ecn byte, dstIsV4 bool) {
hdr := &u.writeMsgs[entry].Hdr
useSeg := runLen >= 2
useEcn := ecn != 0
base := entry * u.writeCmsgSpace
if useSeg {
dataOff := base + unix.CmsgLen(0)
binary.NativeEndian.PutUint16(u.writeCmsg[dataOff:dataOff+2], uint16(segSize))
}
if useEcn {
ecnHdr := (*unix.Cmsghdr)(unsafe.Pointer(&u.writeCmsg[base+u.writeCmsgSegSpace]))
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 + u.writeCmsgSegSpace + unix.CmsgLen(0)
binary.NativeEndian.PutUint32(u.writeCmsg[dataOff:dataOff+4], uint32(ecn))
}
switch {
case useSeg && useEcn:
hdr.Control = &u.writeCmsg[base]
setMsgControllen(hdr, u.writeCmsgSpace)
case useSeg:
hdr.Control = &u.writeCmsg[base]
setMsgControllen(hdr, u.writeCmsgSegSpace)
case useEcn:
hdr.Control = &u.writeCmsg[base+u.writeCmsgSegSpace]
setMsgControllen(hdr, u.writeCmsgEcnSpace)
default:
hdr.Control = nil
setMsgControllen(hdr, 0)
}
}
// sendmmsg issues sendmmsg(2) over u.rawConn against the first n entries
// of u.writeMsgs.
func (u *StdConn) sendmmsg(n int) (int, error) {
r1, _, errno := unix.Syscall6(unix.SYS_SENDMMSG, uintptr(u.sysFd),
uintptr(unsafe.Pointer(&u.writeMsgs[0])), uintptr(n),
0, 0, 0,
)
sent := int(r1)
if errno != 0 {
return sent, &net.OpError{Op: "sendmmsg", Err: errno}
}
return sent, 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) error {
return u.bw.WriteBatch(bufs, addrs, ecns)
}
// writeSockaddr encodes addr into buf (which must be at least
@@ -935,7 +517,7 @@ func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
func (u *StdConn) Close() error {
u.closed.Store(true)
// Wake the reader parked in recvmmsg/recvmsg. shutdown(2) on an unconnected socket
// Wake the reader parked in recvmmsg. shutdown(2) on an unconnected socket
// returns ENOTCONN but still wakes it, so ignore the error.
// The reader then sees closed and stops touching the fd, making the Close below safe.
_ = unix.Shutdown(u.sysFd, unix.SHUT_RDWR)
@@ -973,22 +555,3 @@ func NewUDPStatsEmitter(udpConns []Conn) func() {
}
}
}
func parseRelease(r string) (major, minor int) {
// strip anything after the second dot or any non-digit
parts := strings.SplitN(r, ".", 3)
if len(parts) < 2 {
return 0, 0
}
major, _ = strconv.Atoi(parts[0])
// minor may have trailing junk like "15-generic"
mp := parts[1]
for i, c := range mp {
if c < '0' || c > '9' {
mp = mp[:i]
break
}
}
minor, _ = strconv.Atoi(mp)
return
}
udp/udp_linux_32.go
-33
View File
@@ -30,39 +30,6 @@ type rawMessage struct {
Len uint32
}
func (u *StdConn) PrepareRawMessages(n, bufSize, cmsgSpace int) ([]rawMessage, [][]byte, [][]byte, []byte) {
msgs := make([]rawMessage, n)
buffers := make([][]byte, n)
names := make([][]byte, n)
var cmsgs []byte
if cmsgSpace > 0 {
cmsgs = make([]byte, n*cmsgSpace)
}
for i := range msgs {
buffers[i] = make([]byte, bufSize)
names[i] = make([]byte, unix.SizeofSockaddrInet6)
vs := []iovec{
{Base: &buffers[i][0], Len: uint32(len(buffers[i]))},
}
msgs[i].Hdr.Iov = &vs[0]
msgs[i].Hdr.Iovlen = uint32(len(vs))
msgs[i].Hdr.Name = &names[i][0]
msgs[i].Hdr.Namelen = uint32(len(names[i]))
if cmsgSpace > 0 {
msgs[i].Hdr.Control = &cmsgs[i*cmsgSpace]
msgs[i].Hdr.Controllen = uint32(cmsgSpace)
}
}
return msgs, buffers, names, cmsgs
}
func setIovLen(v *iovec, n int) {
v.Len = uint32(n)
}
udp/udp_linux_64.go
-33
View File
@@ -33,39 +33,6 @@ type rawMessage struct {
Pad0 [4]byte
}
func (u *StdConn) PrepareRawMessages(n, bufSize, cmsgSpace int) ([]rawMessage, [][]byte, [][]byte, []byte) {
msgs := make([]rawMessage, n)
buffers := make([][]byte, n)
names := make([][]byte, n)
var cmsgs []byte
if cmsgSpace > 0 {
cmsgs = make([]byte, n*cmsgSpace)
}
for i := range msgs {
buffers[i] = make([]byte, bufSize)
names[i] = make([]byte, unix.SizeofSockaddrInet6)
vs := []iovec{
{Base: &buffers[i][0], Len: uint64(len(buffers[i]))},
}
msgs[i].Hdr.Iov = &vs[0]
msgs[i].Hdr.Iovlen = uint64(len(vs))
msgs[i].Hdr.Name = &names[i][0]
msgs[i].Hdr.Namelen = uint32(len(names[i]))
if cmsgSpace > 0 {
msgs[i].Hdr.Control = &cmsgs[i*cmsgSpace]
msgs[i].Hdr.Controllen = uint64(cmsgSpace)
}
}
return msgs, buffers, names, cmsgs
}
func setIovLen(v *iovec, n int) {
v.Len = uint64(n)
}
udp/udp_linux_test.go
+3 -3
View File
@@ -29,7 +29,7 @@ func TestShutdownWakesAfterRx_Mechanism(t *testing.T) {
if err != nil {
t.Fatalf("LocalAddr: %v", err)
}
msgs, _, _, _ := sc.PrepareRawMessages(sc.batch, 0xffff, 16)
msgs, _, _, _ := prepareRawMessages(sc.batch, 0xffff, 16)
// Receive a real packet so the socket has carried data.
send, err := net.Dial("udp", addr.String())
@@ -103,8 +103,8 @@ func TestListenOutTeardown_TrafficPatterns(t *testing.T) {
}},
}
// batch 1 exercises the recvmsg path, batch 64 the recvmmsg path; both must
// tear down cleanly.
// batch 1 exercises single-message reads, batch 64 a full recvmmsg batch;
// both must tear down cleanly.
for _, batch := range []int{1, 64} {
for _, tc := range cases {
t.Run(fmt.Sprintf("batch%d/%s", batch, tc.name), func(t *testing.T) {
udp/udp_linux_writebatch.go
+448
View File
@@ -0,0 +1,448 @@
//go:build linux && !android && !e2e_testing
package udp
import (
"encoding/binary"
"fmt"
"log/slog"
"net"
"net/netip"
"strconv"
"strings"
"unsafe"
"golang.org/x/sys/unix"
)
// batchWriter owns the sendmmsg(2)/UDP-GSO transmit path for a StdConn: the
// per-queue scratch WriteBatch packs mmsghdr entries into, plus the GSO
// capability state probed once at socket creation. Each queue has its own
// StdConn and therefore its own batchWriter, so no locking is needed.
type batchWriter struct {
fd int
isV4 bool
l *slog.Logger
// UDP GSO (sendmsg with UDP_SEGMENT cmsg) support. gsoSupported is
// probed once at socket creation. When true, WriteBatch packs same-
// destination consecutive packets into a single sendmmsg entry with a
// UDP_SEGMENT cmsg; otherwise each packet is its own entry.
gsoSupported bool
maxGSOSegments int
// sendmmsg scratch, sized to MaxWriteBatch at construction; WriteBatch
// chunks larger inputs.
msgs []rawMessage
iovs []iovec
names [][]byte
// Per-entry cmsg scratch. cmsg is one contiguous slab of
// MaxWriteBatch * cmsgSpace bytes; each entry holds two cmsg headers
// (UDP_SEGMENT then IP_TOS / IPV6_TCLASS) pre-filled once in
// prepareWriteMessages. WriteBatch only rewrites the per-call data
// payloads and toggles Hdr.Control / Hdr.Controllen to point at
// whichever subset of the two cmsgs applies.
cmsg []byte
cmsgSpace int
cmsgSegSpace int
cmsgEcnSpace int
// entryEnd[e] is the bufs index *after* the last packet packed into
// mmsghdr entry e. Used to rewind `i` on partial sendmmsg success.
entryEnd []int
}
func newBatchWriter(fd int, isV4 bool, l *slog.Logger) *batchWriter {
w := &batchWriter{fd: fd, isV4: isV4, l: l}
w.prepareWriteMessages(MaxWriteBatch)
w.prepareGSO()
return w
}
// prepareWriteMessages allocates one mmsghdr/iovec/sockaddr/cmsg scratch
// slot per sendmmsg entry. The iovec slab is sized to n so all entries'
// iovecs share one allocation; per-entry fan-out is further capped at
// maxGSOSegments. Hdr.Iov / Hdr.Iovlen / Hdr.Control / Hdr.Controllen are
// wired per call since each entry can span a variable number of iovecs
// and may or may not carry a cmsg.
//
// Per-mmsghdr cmsg layout. Each entry's slot of length cmsgSpace holds
// up to two cmsg headers placed at fixed offsets:
//
// [0 .. cmsgSegSpace) UDP_SEGMENT (gso_size, uint16)
// [cmsgSegSpace .. cmsgSpace) IP_TOS or IPV6_TCLASS (int32)
//
// Both headers are pre-filled once here; per-call we only rewrite the data
// payload and toggle Hdr.Control / Hdr.Controllen to point at whichever
// subset applies (none / segment-only / ecn-only / both).
func (w *batchWriter) prepareWriteMessages(n int) {
w.msgs = make([]rawMessage, n)
w.iovs = make([]iovec, n)
w.names = make([][]byte, n)
w.entryEnd = make([]int, n)
w.cmsgSegSpace = unix.CmsgSpace(2)
w.cmsgEcnSpace = unix.CmsgSpace(4)
w.cmsgSpace = w.cmsgSegSpace + w.cmsgEcnSpace
w.cmsg = make([]byte, n*w.cmsgSpace)
// Default the ECN header to the socket's own family. writeEntryCmsg
// finalizes Level/Type per entry from the destination address (a v4-mapped
// dst on a dual-stack v6 socket needs IP_TOS, not IPV6_TCLASS), so this is
// only the value used before the first per-entry rewrite.
ecnLevel := int32(unix.IPPROTO_IP)
ecnType := int32(unix.IP_TOS)
if !w.isV4 {
ecnLevel = unix.IPPROTO_IPV6
ecnType = unix.IPV6_TCLASS
}
for k := 0; k < n; k++ {
base := k * w.cmsgSpace
seg := (*unix.Cmsghdr)(unsafe.Pointer(&w.cmsg[base]))
seg.Level = unix.SOL_UDP
seg.Type = unix.UDP_SEGMENT
setCmsgLen(seg, unix.CmsgLen(2))
ecn := (*unix.Cmsghdr)(unsafe.Pointer(&w.cmsg[base+w.cmsgSegSpace]))
ecn.Level = ecnLevel
ecn.Type = ecnType
setCmsgLen(ecn, unix.CmsgLen(4))
}
for i := range w.msgs {
w.names[i] = make([]byte, unix.SizeofSockaddrInet6)
w.msgs[i].Hdr.Name = &w.names[i][0]
}
}
// maxGSOBytes bounds the total payload per sendmsg() when UDP_SEGMENT is
// set. The kernel stitches all iovecs into a single skb whose length the
// UDP length field can represent, and also enforces sk_gso_max_size (which
// on most devices is 65536). We use 65000 to leave headroom under the
// 65535 UDP-length cap, avoiding EMSGSIZE on large TSO superpackets.
const maxGSOBytes = 65000
// prepareGSO probes UDP_SEGMENT support and sets w.gsoSupported on success.
// Best-effort; failure leaves it false.
func (w *batchWriter) prepareGSO() {
w.maxGSOSegments = 63 //gotta be one less than the max so we can still attach a header
if err := unix.SetsockoptInt(w.fd, unix.IPPROTO_UDP, unix.UDP_SEGMENT, 0); err != nil {
w.l.Info("udp: GSO disabled", "reason", "rawconn control failed", "error", err)
recordCapability("udp.gso.enabled", false)
return
}
var un unix.Utsname
if err := unix.Uname(&un); err != nil {
w.l.Info("udp: GSO disabled", "reason", "kernel uname probe failed", "error", err)
recordCapability("udp.gso.enabled", false)
return
}
w.maxGSOSegments = gsoMaxSegments(string(un.Release[:]))
w.gsoSupported = true
w.l.Info("udp: GSO enabled", "maxGSOSegments", w.maxGSOSegments)
recordCapability("udp.gso.enabled", true)
}
// gsoMaxSegments returns the largest number of UDP_SEGMENT segments a single
// sendmsg may carry on the running kernel, reserving one segment for the
// header. UDP_MAX_SEGMENTS was 64 until Linux v6.9 (commit 1382e3b6a350,
// "udp: change maximum number of UDP segments to 128") raised it to 128;
// nothing about this changed in 5.5. On kernels older than 6.9 packing more
// than 64 segments gets the sendmsg rejected with EINVAL, so cap at 63 there
// and only use 127 from 6.9 on. (Maintainer stance: update your kernel if you
// want to go fast — this is a plain version gate, not a runtime probe.)
func gsoMaxSegments(release string) int {
major, minor := parseRelease(release)
if major > 6 || (major == 6 && minor >= 9) {
return 127
}
return 63
}
func parseRelease(r string) (major, minor int) {
// strip anything after the second dot or any non-digit
parts := strings.SplitN(r, ".", 3)
if len(parts) < 2 {
return 0, 0
}
major, _ = strconv.Atoi(parts[0])
// minor may have trailing junk like "15-generic"
mp := parts[1]
for i, c := range mp {
if c < '0' || c > '9' {
mp = mp[:i]
break
}
}
minor, _ = strconv.Atoi(mp)
return
}
// WriteBatch sends bufs via sendmmsg(2) using the preallocated scratch on
// batchWriter. Consecutive packets to the same destination with matching
// segment sizes (all but possibly the last) are coalesced into a single
// mmsghdr entry carrying a UDP_SEGMENT cmsg, so one syscall can mix runs of
// GSO superpackets with plain one-off datagrams. Without GSO support every
// packet is its own entry, matching the prior behaviour.
//
// Chunks larger than the scratch are processed across multiple syscalls. If
// sendmmsg returns an error AND zero entries went out we fall back to
// per-packet sendto for that chunk so the caller still gets best-effort
// delivery; on a partial-success error we just replay the remainder.
func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) error {
if len(bufs) != len(addrs) {
return fmt.Errorf("WriteBatch: len(bufs)=%d != len(addrs)=%d", len(bufs), len(addrs))
}
if ecns != nil && len(ecns) != len(bufs) {
return 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.
i := 0
sendChunks:
for i < len(bufs) {
baseI := i
entry := 0
iovIdx := 0
for entry < len(w.msgs) && i < len(bufs) {
iovBudget := len(w.iovs) - iovIdx
if iovBudget < 1 {
break
}
runLen, segSize := w.planRun(bufs, addrs, ecns, i, iovBudget)
if runLen == 0 {
break
}
for k := 0; k < runLen; k++ {
b := bufs[i+k]
if len(b) == 0 {
w.iovs[iovIdx+k].Base = nil
setIovLen(&w.iovs[iovIdx+k], 0)
} else {
w.iovs[iovIdx+k].Base = &b[0]
setIovLen(&w.iovs[iovIdx+k], len(b))
}
}
nlen, err := writeSockaddr(w.names[entry], addrs[i], w.isV4)
if err != nil {
// One destination in this chunk has an address family the
// socket can't send to (e.g. an IPv6 remote on a v4-bound
// socket → ErrInvalidIPv6RemoteForSocket). Abandoning the whole
// sendmmsg here would drop every packet already packed for this
// chunk plus every packet still ahead of us in bufs. Instead
// fall back to per-packet sendto for the packets packed so far
// in this chunk and the offending one: sendto delivers each
// good destination and only errors on the bad one, which we
// drop and keep going. One bad destination costs one packet,
// never the batch. (Same fallback the zero-sent sendmmsg path
// below uses, extended to cover the misaddressed packet.)
for k := baseI; k <= i; k++ {
if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr != nil && k != i {
return werr
}
}
i++
continue sendChunks
}
hdr := &w.msgs[entry].Hdr
hdr.Iov = &w.iovs[iovIdx]
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: a
// v4-mapped dst on a dual-stack v6 socket must be stamped via
// IP_TOS. addrs[i] is this run's destination (i advances below).
dstIsV4 := addrs[i].Addr().Unmap().Is4()
w.writeEntryCmsg(entry, runLen, segSize, ecn, dstIsV4)
i += runLen
iovIdx += runLen
w.entryEnd[entry] = i
entry++
}
if entry == 0 {
return fmt.Errorf("sendmmsg: no progress")
}
sent, serr := w.sendmmsg(entry)
if serr != nil && sent <= 0 {
// Nothing went out for this chunk; fall back to sendto for each
// packet that was queued this iteration. We only enter this path
// when sendmmsg returned an error AND zero entries succeeded —
// otherwise the partial-success advance below replays only the
// remainder, avoiding duplicates of already-sent packets.
//
// sent=-1 from sendmmsg means message 0 itself failed (partial
// success returns the count instead), so log entry 0's parameters
// — that's the entry the kernel rejected.
hdr0 := &w.msgs[0].Hdr
runLen0 := w.entryEnd[0] - baseI
seg0 := len(bufs[baseI])
ecn0 := byte(0)
if ecns != nil {
ecn0 = ecns[baseI]
}
w.l.Warn("sendmmsg had problem",
"sent", sent, "err", serr,
"entries", entry,
"entry0_runLen", runLen0,
"entry0_segSize", seg0,
"entry0_iovlen", hdr0.Iovlen,
"entry0_controllen", hdr0.Controllen,
"entry0_namelen", hdr0.Namelen,
"entry0_ecn", ecn0,
"entry0_dst", addrs[baseI],
"isV4", w.isV4,
"gso", w.gsoSupported,
)
for k := baseI; k < i; k++ {
if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr != nil {
return werr
}
}
continue
}
if sent == 0 {
return fmt.Errorf("sendmmsg made no progress")
}
// Rewind i to the end of the last successfully sent entry. For a
// full-success send this leaves i unchanged; for a partial send it
// replays the remainder on the next outer-loop iteration.
i = w.entryEnd[sent-1]
}
return nil
}
// planRun groups consecutive packets starting at `start` that can be sent as
// a single UDP GSO superpacket (one sendmmsg entry with UDP_SEGMENT cmsg).
// A run of length 1 means the entry carries no UDP_SEGMENT cmsg and the
// kernel treats it as a plain datagram. Returns the run length and the
// per-segment size (which equals len(bufs[start])). Without GSO support
// every call returns runLen=1. Outer ECN (when ecns != nil) is also a run
// boundary — the kernel stamps one outer codepoint per sendmsg entry, so
// mixing values inside a run would lose information.
func (w *batchWriter) planRun(bufs [][]byte, addrs []netip.AddrPort, ecns []byte, start, iovBudget int) (int, int) {
if start >= len(bufs) || iovBudget < 1 {
return 0, 0
}
segSize := len(bufs[start])
if !w.gsoSupported || segSize == 0 || segSize > maxGSOBytes {
return 1, segSize
}
dst := addrs[start]
var ecn byte
if ecns != nil {
ecn = ecns[start]
}
maxLen := w.maxGSOSegments
if iovBudget < maxLen {
maxLen = iovBudget
}
runLen := 1
total := segSize
for runLen < maxLen && start+runLen < len(bufs) {
nextLen := len(bufs[start+runLen])
if nextLen == 0 || nextLen > segSize {
break
}
if addrs[start+runLen] != dst {
break
}
if ecns != nil && ecns[start+runLen] != ecn {
break
}
if total+nextLen > maxGSOBytes {
break
}
total += nextLen
runLen++
if nextLen < segSize {
// A short packet must be the last in the run.
break
}
}
return runLen, segSize
}
// writeEntryCmsg sets up the per-mmsghdr Hdr.Control / Hdr.Controllen for one
// entry. It writes the UDP_SEGMENT payload when runLen >= 2 and the
// IP_TOS/IPV6_TCLASS payload when ecn != 0, then points hdr.Control at the
// smallest contiguous span that covers whichever cmsg(s) actually apply.
//
// The outer-ECN cmsg family must match the *destination*, not the socket: on
// the default dual-stack v6 bind, a v4-mapped destination is routed through
// the kernel's IPv4 path, which parses IP_TOS (IPPROTO_IP) and ignores an
// IPV6_TCLASS cmsg. prepareWriteMessages pre-fills a default header; here we
// rewrite its Level/Type (and Len) per entry from dstIsV4 so v4 peers get
// IP_TOS and v6 peers get IPV6_TCLASS. The data payload is a 4-byte int for
// both families, so the pre-computed cmsg space is unchanged.
func (w *batchWriter) writeEntryCmsg(entry, runLen, segSize int, ecn byte, dstIsV4 bool) {
hdr := &w.msgs[entry].Hdr
useSeg := runLen >= 2
useEcn := ecn != 0
base := entry * w.cmsgSpace
if useSeg {
dataOff := base + unix.CmsgLen(0)
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))
}
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:
hdr.Control = nil
setMsgControllen(hdr, 0)
}
}
// sendmmsg issues sendmmsg(2) against the first n entries of w.msgs.
func (w *batchWriter) sendmmsg(n int) (int, error) {
r1, _, errno := unix.Syscall6(unix.SYS_SENDMMSG, uintptr(w.fd),
uintptr(unsafe.Pointer(&w.msgs[0])), uintptr(n),
0, 0, 0,
)
sent := int(r1)
if errno != 0 {
return sent, &net.OpError{Op: "sendmmsg", Err: errno}
}
return sent, nil
}
udp/udp_linux_writebatch_alloc_test.go
+101
View File
@@ -0,0 +1,101 @@
//go:build linux && !android && !e2e_testing
package udp
import (
"net/netip"
"testing"
)
// TestWriteBatchNoAllocs verifies the sendmmsg/UDP-GSO transmit path performs
// 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,
// and cmsg paths are all exercised.
func TestWriteBatchNoAllocs(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()
rxA := newConn()
rxB := newConn()
if sc, ok := tx.(*StdConn); ok {
// Records which planner path the measurement covered; GSO
// support depends on the running kernel.
t.Logf("gsoSupported=%v maxGSOSegments=%d", sc.bw.gsoSupported, sc.bw.maxGSOSegments)
}
dstA, err := rxA.LocalAddr()
if err != nil {
t.Fatalf("LocalAddr: %v", err)
}
dstB, err := rxB.LocalAddr()
if err != nil {
t.Fatalf("LocalAddr: %v", err)
}
payload := make([]byte, 1200)
short := make([]byte, 900)
var bufs [][]byte
var addrs []netip.AddrPort
var ecns []byte
add := func(b []byte, dst netip.AddrPort, ecn byte) {
bufs = append(bufs, b)
addrs = append(addrs, dst)
ecns = append(ecns, ecn)
}
// GSO-eligible run with a short tail, all ECT(0).
for k := 0; k < 8; k++ {
add(payload, dstA, 0b10)
}
add(short, dstA, 0b10)
// ECN change on the same destination forces a run boundary.
add(payload, dstA, 0)
// Alternating destinations defeat coalescing entirely.
for k := 0; k < 4; k++ {
dst := dstA
if k%2 == 0 {
dst = dstB
}
add(payload, dst, 0)
}
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)
}
}
send(ecns)
send(nil)
})
}
}