udp_linux_fixes_test.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708
  1. //go:build linux && !android && !e2e_testing
  2. package udp
  3. import (
  4. "fmt"
  5. "log/slog"
  6. "net"
  7. "net/netip"
  8. "slices"
  9. "testing"
  10. "time"
  11. "unsafe"
  12. "golang.org/x/sys/unix"
  13. )
  14. // TestGSOMaxSegmentsKernelGate pins the corrected kernel-version gate: the
  15. // 128-segment cap (127 usable) only lands in Linux v6.9 (commit 1382e3b6a350),
  16. // not 5.5. Everything older stays at the conservative 63.
  17. func TestGSOMaxSegmentsKernelGate(t *testing.T) {
  18. cases := []struct {
  19. release string
  20. want int
  21. }{
  22. {"5.4.0", 63},
  23. {"5.5.0-generic", 63}, // the old bug bumped here — it must not now
  24. {"5.15.0", 63},
  25. {"6.1.0", 63},
  26. {"6.8.0-generic", 63},
  27. {"6.9.0", 127},
  28. {"6.10.1-arch1-1", 127},
  29. {"7.0.5-arch1-1", 127},
  30. {"garbage", 63},
  31. {"", 63},
  32. }
  33. for _, c := range cases {
  34. if got := gsoMaxSegments(c.release); got != c.want {
  35. t.Errorf("gsoMaxSegments(%q) = %d, want %d", c.release, got, c.want)
  36. }
  37. }
  38. }
  39. // buildCmsg lays out a single ancillary cmsg (header + data) in a fresh buffer
  40. // the way the kernel would deliver it, so parseRecvCmsg can be exercised
  41. // without a live socket.
  42. func buildCmsg(level, typ int32, data []byte) []byte {
  43. buf := make([]byte, unix.CmsgSpace(len(data)))
  44. h := (*unix.Cmsghdr)(unsafe.Pointer(&buf[0]))
  45. h.Level = level
  46. h.Type = typ
  47. setCmsgLen(h, unix.CmsgLen(len(data)))
  48. copy(buf[unix.CmsgLen(0):], data)
  49. return buf
  50. }
  51. func testLogger() *slog.Logger {
  52. return slog.New(slog.DiscardHandler)
  53. }
  54. // TestWriteBatchBadFamilyDeliversOthers is the H3 regression: a batch that
  55. // contains one destination the socket can't reach (an IPv6 remote on a
  56. // v4-bound socket) must still deliver every other packet. Before the fix the
  57. // writeSockaddr error returned early and dropped the whole chunk.
  58. func TestWriteBatchBadFamilyDeliversOthers(t *testing.T) {
  59. rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
  60. if err != nil {
  61. t.Skipf("cannot open v4 receiver (sandbox?): %v", err)
  62. }
  63. defer rx.Close()
  64. rxPort := rx.LocalAddr().(*net.UDPAddr).Port
  65. // Bind a *non-wildcard* v4 address so Go gives us a genuine AF_INET
  66. // socket. A wildcard v4 bind (0.0.0.0) via network "udp" comes up as a
  67. // dual-stack AF_INET6 socket on Linux, for which a v6 dest is not a bad
  68. // family — which would defeat the point of this test.
  69. udpSettings := Settings{
  70. Listen: netip.MustParseAddrPort("127.0.0.1:0"),
  71. Multi: false,
  72. Batch: 1,
  73. Offloads: false,
  74. }
  75. c, err := NewListener(testLogger(), udpSettings)
  76. if err != nil {
  77. t.Skipf("cannot open v4 sender (sandbox?): %v", err)
  78. }
  79. defer c.Close()
  80. sender := c.(*StdConn)
  81. if !sender.isV4 {
  82. t.Fatalf("expected a v4-bound sender socket, got isV4=false")
  83. }
  84. good := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(rxPort))
  85. bad := netip.MustParseAddrPort("[2001:db8::1]:9999") // genuine v6, unreachable on v4 socket
  86. bufs := [][]byte{[]byte("AAA"), []byte("BBB"), []byte("CCC")}
  87. addrs := []netip.AddrPort{good, bad, good}
  88. n, err := sender.WriteBatch(bufs, addrs)
  89. if err != nil {
  90. t.Fatalf("WriteBatch returned error, want nil (bad dest should be isolated): %v", err)
  91. }
  92. if n != 2 {
  93. t.Errorf("WriteBatch wrote %d packets, want 2 of 3 (the bad-family dest is the only casualty)", n)
  94. }
  95. got := map[string]bool{}
  96. rx.SetReadDeadline(time.Now().Add(2 * time.Second))
  97. buf := make([]byte, 64)
  98. for i := range 2 {
  99. n, _, rerr := rx.ReadFromUDPAddrPort(buf)
  100. if rerr != nil {
  101. t.Fatalf("expected 2 delivered packets, read #%d failed: %v", i+1, rerr)
  102. }
  103. got[string(buf[:n])] = true
  104. }
  105. if !got["AAA"] || !got["CCC"] {
  106. t.Errorf("delivered set = %v, want AAA and CCC both present", got)
  107. }
  108. if got["BBB"] {
  109. t.Errorf("the bad-family packet BBB was somehow delivered")
  110. }
  111. }
  112. // TestWriteBatchUnreachableDestDeliversOthers is the kernel-rejection twin of
  113. // TestWriteBatchBadFamilyDeliversOthers. A destination the kernel refuses outright (240.0.0.0/4 is reserved, so
  114. // the send returns EINVAL) fails its sendmmsg entry; WriteBatch must drop only that entry and still deliver
  115. // every other packet rather than abandoning the batch at the first failure.
  116. func TestWriteBatchUnreachableDestDeliversOthers(t *testing.T) {
  117. rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
  118. if err != nil {
  119. t.Skipf("cannot open v4 receiver (sandbox?): %v", err)
  120. }
  121. defer rx.Close()
  122. rxPort := rx.LocalAddr().(*net.UDPAddr).Port
  123. udpSettings := Settings{
  124. Listen: netip.MustParseAddrPort("127.0.0.1:0"),
  125. Multi: false,
  126. Batch: 1,
  127. Offloads: false,
  128. }
  129. c, err := NewListener(testLogger(), udpSettings)
  130. if err != nil {
  131. t.Skipf("cannot open v4 sender (sandbox?): %v", err)
  132. }
  133. defer c.Close()
  134. sender := c.(*StdConn)
  135. good := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(rxPort))
  136. bad := netip.MustParseAddrPort("240.0.0.1:9999") // reserved space, the kernel refuses it
  137. bufs := [][]byte{[]byte("P0"), []byte("P1"), []byte("BAD"), []byte("P3"), []byte("P4")}
  138. addrs := []netip.AddrPort{good, good, bad, good, good}
  139. // The bad destination is reported, but only after every other packet has been attempted.
  140. if _, err := sender.WriteBatch(bufs, addrs); err == nil {
  141. t.Log("WriteBatch returned nil; kernel accepted the reserved address, delivery assertions still apply")
  142. }
  143. got := map[string]bool{}
  144. rx.SetReadDeadline(time.Now().Add(2 * time.Second))
  145. buf := make([]byte, 64)
  146. for i := range 4 {
  147. n, _, rerr := rx.ReadFromUDPAddrPort(buf)
  148. if rerr != nil {
  149. t.Fatalf("expected 4 delivered packets, read #%d failed: %v (got so far: %v)", i+1, rerr, got)
  150. }
  151. got[string(buf[:n])] = true
  152. }
  153. for _, want := range []string{"P0", "P1", "P3", "P4"} {
  154. if !got[want] {
  155. t.Errorf("packet %s was not delivered; delivered set = %v", want, got)
  156. }
  157. }
  158. }
  159. // TestParseRecvCmsgCorruptLenNoPanic: a cmsg Len near max-int used to wrap
  160. // off+clen negative, slip past the bounds check, and drive the walk offset
  161. // negative -- a panic on the next ctrl[off]. The guard must compare Len
  162. // against the remaining bytes instead. Also pins the plain truncated-Len
  163. // cases (too small, larger than the buffer) to a clean early return.
  164. func TestParseRecvCmsgCorruptLenNoPanic(t *testing.T) {
  165. // First cmsg: a valid empty one so the walk advances past off=0
  166. // (off+clen can't overflow while off is still zero).
  167. valid := buildCmsg(int32(unix.SOL_UDP), int32(unix.UDP_GRO), make([]byte, 4))
  168. corrupt := func(lenVal int) []byte {
  169. buf := make([]byte, len(valid)+unix.CmsgSpace(4))
  170. copy(buf, valid)
  171. h := (*unix.Cmsghdr)(unsafe.Pointer(&buf[len(valid)]))
  172. h.Level = int32(unix.IPPROTO_IP)
  173. h.Type = int32(unix.IP_TOS)
  174. setCmsgLen(h, lenVal)
  175. return buf
  176. }
  177. cases := []struct {
  178. name string
  179. ctrl []byte
  180. }{
  181. {"len_near_max_int", corrupt(int(^uint(0)>>1) - 8)},
  182. {"len_too_small", corrupt(unix.SizeofCmsghdr - 1)},
  183. {"len_past_buffer", corrupt(1 << 20)},
  184. }
  185. for _, c := range cases {
  186. t.Run(c.name, func(t *testing.T) {
  187. hdr := &msghdr{Control: &c.ctrl[0]}
  188. setMsgControllen(hdr, len(c.ctrl))
  189. gso := parseRecvCmsg(hdr)
  190. // The valid leading UDP_GRO cmsg (payload 0) must still parse;
  191. // the corrupt trailer just ends the walk.
  192. if gso != 0 {
  193. t.Errorf("parseRecvCmsg = %d, want 0", gso)
  194. }
  195. })
  196. }
  197. }
  198. // TestDeliverSegments pins the GRO RX splitting: a kernel-coalesced buffer
  199. // must come back out as the exact pre-coalesce packets -- every boundary
  200. // error here shreds encrypted packets and every decrypt downstream fails.
  201. func TestDeliverSegments(t *testing.T) {
  202. from := netip.MustParseAddrPort("192.0.2.1:4242")
  203. // Spare backing capacity mimics the recvmmsg row a real payload sits in;
  204. // the cap checks below prove none of it leaks to a delivered segment.
  205. pay := func(n int) []byte {
  206. b := make([]byte, n, n+512)
  207. for i := range b {
  208. b[i] = byte(i)
  209. }
  210. return b
  211. }
  212. cases := []struct {
  213. name string
  214. payload []byte
  215. segSize int
  216. wantLens []int
  217. }{
  218. {"no-gro", pay(1400), 0, []int{1400}},
  219. {"negative-segsize", pay(1400), -5, []int{1400}},
  220. {"segsize-equals-payload", pay(1400), 1400, []int{1400}},
  221. {"segsize-past-payload", pay(1400), 2000, []int{1400}},
  222. {"even-split", pay(4200), 1400, []int{1400, 1400, 1400}},
  223. {"short-tail", pay(3000), 1400, []int{1400, 1400, 200}},
  224. {"single-byte-tail", pay(2801), 1400, []int{1400, 1400, 1}},
  225. {"segsize-one", pay(3), 1, []int{1, 1, 1}},
  226. {"empty-payload", pay(0), 1400, []int{0}},
  227. {"max-coalesce", pay(65500), 1372, nil}, // lens derived below
  228. }
  229. for _, c := range cases {
  230. t.Run(c.name, func(t *testing.T) {
  231. wantLens := c.wantLens
  232. if wantLens == nil {
  233. for rem := len(c.payload); rem > 0; rem -= c.segSize {
  234. wantLens = append(wantLens, min(c.segSize, rem))
  235. }
  236. }
  237. var got [][]byte
  238. deliverSegments(func(a netip.AddrPort, seg []byte) {
  239. if a != from {
  240. t.Errorf("from = %v, want %v", a, from)
  241. }
  242. got = append(got, seg)
  243. }, from, c.payload, c.segSize)
  244. if len(got) != len(wantLens) {
  245. t.Fatalf("delivered %d segments, want %d", len(got), len(wantLens))
  246. }
  247. // Segments must tile the payload in order with no gap, overlap,
  248. // or copy: each must alias the payload at the right offset.
  249. off := 0
  250. for i, seg := range got {
  251. if len(seg) != wantLens[i] {
  252. t.Fatalf("segment %d len=%d want %d", i, len(seg), wantLens[i])
  253. }
  254. if cap(seg) != len(seg) {
  255. // EncReader contract: an append into spare capacity would
  256. // scribble into the next segment of the shared row.
  257. t.Errorf("segment %d cap=%d, want %d (capacity must not reach into the row)", i, cap(seg), len(seg))
  258. }
  259. if len(seg) > 0 && &seg[0] != &c.payload[off] {
  260. t.Errorf("segment %d does not alias payload at offset %d", i, off)
  261. }
  262. off += len(seg)
  263. }
  264. if off != len(c.payload) {
  265. t.Errorf("segments cover %d bytes, payload has %d", off, len(c.payload))
  266. }
  267. })
  268. }
  269. }
  270. // newRewindTestWriter builds a batchWriter with no socket: GSO planning on,
  271. // sendFn left for the test to script. fd is invalid on purpose -- any path
  272. // that actually hits the kernel fails loudly.
  273. func newRewindTestWriter() *batchWriter {
  274. w := &batchWriter{fd: -1, isV4: true, l: testLogger()}
  275. // gsoSupported must be set before prepareWriteMessages: the cmsg slab is
  276. // only allocated when GSO is already known to be supported.
  277. w.gsoSupported = true
  278. w.maxGSOSegments = 63
  279. w.prepareWriteMessages(MaxWriteBatch, true)
  280. return w
  281. }
  282. // capturePrepared decodes n prepared mmsghdr entries beginning at start
  283. // straight from their iovecs -- ground truth, deliberately not the entryEnd
  284. // bookkeeping the resume logic itself relies on. Returns one []byte per
  285. // packed packet, in entry order.
  286. func capturePrepared(w *batchWriter, start, n int) [][]byte {
  287. var out [][]byte
  288. for e := start; e < start+n; e++ {
  289. hdr := &w.msgs[e].Hdr
  290. iovs := unsafe.Slice(hdr.Iov, int(hdr.Iovlen))
  291. for _, iov := range iovs {
  292. b := make([]byte, int(iov.Len))
  293. if iov.Len > 0 {
  294. copy(b, unsafe.Slice(iov.Base, int(iov.Len)))
  295. }
  296. out = append(out, b)
  297. }
  298. }
  299. return out
  300. }
  301. // TestWriteBatchPartialSendRewind drives WriteBatch through scripted
  302. // partial sendmmsg results and asserts the rewind resumes exactly where
  303. // the kernel stopped: every packet on the wire exactly once, in order,
  304. // no duplicate, no loss. This is the hairiest logic in the write path
  305. // and a rewind bug means silent packet duplication or loss under EAGAIN-
  306. // style backpressure.
  307. func TestWriteBatchPartialSendRewind(t *testing.T) {
  308. dstA := netip.MustParseAddrPort("127.0.0.1:4242")
  309. dstB := netip.MustParseAddrPort("127.0.0.2:4242")
  310. mkBuf := func(tag byte, n int) []byte {
  311. b := make([]byte, n)
  312. for i := range b {
  313. b[i] = tag
  314. }
  315. b[0] = tag // tag identifies the packet uniquely below
  316. return b
  317. }
  318. // Mixed shape: a 3-packet GSO run to A, a lone short packet to A (run
  319. // tail), then two to B. The planner packs this as multiple entries with
  320. // multi-iovec runs, which is what makes the rewind arithmetic hairy.
  321. bufs := [][]byte{
  322. mkBuf(1, 1200), mkBuf(2, 1200), mkBuf(3, 1200), // run to A
  323. mkBuf(4, 600), // short tail to A
  324. mkBuf(5, 900), mkBuf(6, 900), // run to B
  325. }
  326. addrs := []netip.AddrPort{dstA, dstA, dstA, dstA, dstB, dstB}
  327. scripts := [][]int{
  328. {99}, // accept everything first call
  329. {1, 99}, // one entry per call, then the rest
  330. {1, 1, 1, 99}, // strictly one entry per call
  331. {2, 99}, // two entries, then the rest
  332. }
  333. for si, script := range scripts {
  334. t.Run(fmt.Sprintf("script_%d", si), func(t *testing.T) {
  335. w := newRewindTestWriter()
  336. var wire [][]byte
  337. call := 0
  338. w.sendFn = func(start, n int) (int, error) {
  339. accept := n
  340. if call < len(script) && script[call] < n {
  341. accept = script[call]
  342. }
  343. call++
  344. wire = append(wire, capturePrepared(w, start, accept)...)
  345. return accept, nil
  346. }
  347. written, err := w.WriteBatch(bufs, addrs)
  348. if err != nil {
  349. t.Fatalf("WriteBatch: %v", err)
  350. }
  351. if written != len(bufs) {
  352. t.Errorf("written = %d, want %d", written, len(bufs))
  353. }
  354. if len(wire) != len(bufs) {
  355. t.Fatalf("wire got %d packets, want %d (dup or loss in rewind)", len(wire), len(bufs))
  356. }
  357. for i, b := range wire {
  358. if len(b) != len(bufs[i]) || b[0] != bufs[i][0] {
  359. t.Errorf("wire[%d] = tag %d len %d, want tag %d len %d (reorder/dup)",
  360. i, b[0], len(b), bufs[i][0], len(bufs[i]))
  361. }
  362. }
  363. })
  364. }
  365. }
  366. // TestWriteBatchSkipUnroutableRunAccounting: an unroutable destination mid-
  367. // batch is skipped without committing an entry, leaving a hole in the bufs
  368. // index space. The written count must tally packets per sent entry -- the
  369. // index span would count the hole -- across both full and partial sendmmsg
  370. // success.
  371. func TestWriteBatchSkipUnroutableRunAccounting(t *testing.T) {
  372. dstA := netip.MustParseAddrPort("127.0.0.1:4242")
  373. dstB := netip.MustParseAddrPort("127.0.0.2:4242")
  374. bad := netip.MustParseAddrPort("[2001:db8::1]:9999") // v6 dest, v4 writer
  375. mk := func(tag byte, n int) []byte {
  376. b := make([]byte, n)
  377. b[0] = tag
  378. return b
  379. }
  380. bufs := [][]byte{mk(1, 1200), mk(2, 1200), mk(3, 500), mk(4, 900), mk(5, 900)}
  381. addrs := []netip.AddrPort{dstA, dstA, bad, dstB, dstB}
  382. for si, script := range [][]int{{99}, {1, 99}} {
  383. t.Run(fmt.Sprintf("script_%d", si), func(t *testing.T) {
  384. w := newRewindTestWriter()
  385. var wire [][]byte
  386. call := 0
  387. w.sendFn = func(start, n int) (int, error) {
  388. accept := n
  389. if call < len(script) && script[call] < n {
  390. accept = script[call]
  391. }
  392. call++
  393. wire = append(wire, capturePrepared(w, start, accept)...)
  394. return accept, nil
  395. }
  396. written, err := w.WriteBatch(bufs, addrs)
  397. if err != nil {
  398. t.Fatalf("WriteBatch: %v", err)
  399. }
  400. if written != 4 {
  401. t.Errorf("written = %d, want 4 (the unroutable run is the only casualty)", written)
  402. }
  403. wantTags := []byte{1, 2, 4, 5}
  404. if len(wire) != len(wantTags) {
  405. t.Fatalf("wire got %d packets, want %d (dup or loss around the skip)", len(wire), len(wantTags))
  406. }
  407. for i, b := range wire {
  408. if b[0] != wantTags[i] {
  409. t.Errorf("wire[%d] tag = %d, want %d", i, b[0], wantTags[i])
  410. }
  411. }
  412. })
  413. }
  414. }
  415. // TestWriteBatchMidChunkRejectResumes: after a partial success, a zero-sent
  416. // error on the FIRST REMAINING entry (done > 0) must drop only that entry's
  417. // run and resume the rest of the chunk in place -- no repacking, no packets
  418. // lost from entries before or after the rejected one.
  419. func TestWriteBatchMidChunkRejectResumes(t *testing.T) {
  420. dstA := netip.MustParseAddrPort("127.0.0.1:4242")
  421. dstB := netip.MustParseAddrPort("127.0.0.2:4242")
  422. dstC := netip.MustParseAddrPort("127.0.0.3:4242")
  423. mk := func(tag byte, n int) []byte {
  424. b := make([]byte, n)
  425. b[0] = tag
  426. return b
  427. }
  428. // Three entries: a 2-packet GSO run to A, a 2-packet run to B, one to C.
  429. bufs := [][]byte{mk(1, 1200), mk(2, 1200), mk(3, 900), mk(4, 900), mk(5, 600)}
  430. addrs := []netip.AddrPort{dstA, dstA, dstB, dstB, dstC}
  431. w := newRewindTestWriter()
  432. var wire [][]byte
  433. var starts []int
  434. call := 0
  435. w.sendFn = func(start, n int) (int, error) {
  436. starts = append(starts, start)
  437. call++
  438. switch call {
  439. case 1: // accept only entry 0 (the run to A)
  440. wire = append(wire, capturePrepared(w, start, 1)...)
  441. return 1, nil
  442. case 2: // reject entry 1 (the run to B) outright
  443. return -1, &net.OpError{Op: "sendmmsg", Err: unix.EPERM}
  444. default: // accept the rest
  445. wire = append(wire, capturePrepared(w, start, n)...)
  446. return n, nil
  447. }
  448. }
  449. written, err := w.WriteBatch(bufs, addrs)
  450. if err != nil {
  451. t.Fatalf("WriteBatch: %v", err)
  452. }
  453. if written != 3 {
  454. t.Errorf("written = %d, want 3 (B's rejected run is the only casualty)", written)
  455. }
  456. wantTags := []byte{1, 2, 5}
  457. if len(wire) != len(wantTags) {
  458. t.Fatalf("wire got %d packets, want %d (dup or loss around the mid-chunk reject)", len(wire), len(wantTags))
  459. }
  460. for i, b := range wire {
  461. if b[0] != wantTags[i] {
  462. t.Errorf("wire[%d] tag = %d, want %d", i, b[0], wantTags[i])
  463. }
  464. }
  465. // The resume must reuse the prepared entries: same chunk, advancing
  466. // start offsets, no repack (which would restart at 0 with fresh entries).
  467. if want := []int{0, 1, 2}; !slices.Equal(starts, want) {
  468. t.Errorf("sendFn start offsets = %v, want %v", starts, want)
  469. }
  470. }
  471. // TestWriteBatchMidChunkEIODisablesGSOWithoutDup: an EIO on a GSO entry
  472. // after earlier entries in the chunk already went out must replay ONLY from
  473. // the failed run (replanned as single-packet entries) -- the already-sent
  474. // entries must not be duplicated.
  475. func TestWriteBatchMidChunkEIODisablesGSOWithoutDup(t *testing.T) {
  476. dstA := netip.MustParseAddrPort("127.0.0.1:4242")
  477. dstB := netip.MustParseAddrPort("127.0.0.2:4242")
  478. mk := func(tag byte, n int) []byte {
  479. b := make([]byte, n)
  480. b[0] = tag
  481. return b
  482. }
  483. // Entry 0: single packet to A. Entry 1: 2-packet GSO run to B.
  484. bufs := [][]byte{mk(1, 600), mk(2, 1200), mk(3, 1200)}
  485. addrs := []netip.AddrPort{dstA, dstB, dstB}
  486. w := newRewindTestWriter()
  487. var wire [][]byte
  488. call := 0
  489. w.sendFn = func(start, n int) (int, error) {
  490. call++
  491. switch call {
  492. case 1: // accept entry 0 only
  493. wire = append(wire, capturePrepared(w, start, 1)...)
  494. return 1, nil
  495. case 2: // EIO on the GSO run to B
  496. return -1, &net.OpError{Op: "sendmmsg", Err: unix.EIO}
  497. default: // replanned single-packet replay
  498. wire = append(wire, capturePrepared(w, start, n)...)
  499. return n, nil
  500. }
  501. }
  502. written, err := w.WriteBatch(bufs, addrs)
  503. if err != nil {
  504. t.Fatalf("WriteBatch: %v", err)
  505. }
  506. if w.gsoSupported {
  507. t.Error("gsoSupported still true after EIO on a GSO entry")
  508. }
  509. if written != len(bufs) {
  510. t.Errorf("written = %d, want %d", written, len(bufs))
  511. }
  512. wantTags := []byte{1, 2, 3}
  513. if len(wire) != len(wantTags) {
  514. t.Fatalf("wire got %d packets, want %d (packet 1 duplicated, or B's run lost)", len(wire), len(wantTags))
  515. }
  516. for i, b := range wire {
  517. if b[0] != wantTags[i] {
  518. t.Errorf("wire[%d] tag = %d, want %d", i, b[0], wantTags[i])
  519. }
  520. }
  521. }
  522. // TestWriteBatchZeroProgress: sent == 0 with no error must abort with an
  523. // error rather than spin forever replaying the same chunk.
  524. func TestWriteBatchZeroProgress(t *testing.T) {
  525. w := newRewindTestWriter()
  526. w.sendFn = func(start, n int) (int, error) { return 0, nil }
  527. bufs := [][]byte{make([]byte, 100)}
  528. addrs := []netip.AddrPort{netip.MustParseAddrPort("127.0.0.1:4242")}
  529. if _, err := w.WriteBatch(bufs, addrs); err == nil {
  530. t.Fatal("WriteBatch = nil error on zero progress, want error")
  531. }
  532. }
  533. // TestWriteBatchEIODisablesGSOAndReplays pins the runtime GSO give-up: a
  534. // sendmmsg rejected with EIO on a GSO superpacket entry must clear
  535. // gsoSupported and replay the same packets as per-packet entries through
  536. // sendmmsg (keeping batching), not fall back to per-packet sendto.
  537. func TestWriteBatchEIODisablesGSOAndReplays(t *testing.T) {
  538. dst := netip.MustParseAddrPort("127.0.0.1:4242")
  539. bufs := [][]byte{make([]byte, 1200), make([]byte, 1200), make([]byte, 1200)}
  540. addrs := []netip.AddrPort{dst, dst, dst}
  541. w := newRewindTestWriter()
  542. var entryCounts []int
  543. call := 0
  544. w.sendFn = func(start, n int) (int, error) {
  545. entryCounts = append(entryCounts, n)
  546. call++
  547. if call == 1 {
  548. return -1, &net.OpError{Op: "sendmmsg", Err: unix.EIO}
  549. }
  550. return n, nil
  551. }
  552. written, err := w.WriteBatch(bufs, addrs)
  553. if err != nil {
  554. t.Fatalf("WriteBatch: %v", err)
  555. }
  556. if w.gsoSupported {
  557. t.Error("gsoSupported still true after EIO on a GSO entry")
  558. }
  559. if written != len(bufs) {
  560. t.Errorf("written = %d, want %d", written, len(bufs))
  561. }
  562. // First call: one GSO entry carrying the whole run. Replay: one entry
  563. // per packet, still via sendmmsg.
  564. want := []int{1, 3}
  565. if len(entryCounts) != len(want) || entryCounts[0] != want[0] || entryCounts[1] != want[1] {
  566. t.Errorf("sendmmsg entry counts = %v, want %v", entryCounts, want)
  567. }
  568. }
  569. // TestGSOEngagesOnLoopback is the offload smoke test: real sockets, real
  570. // UDP_SEGMENT cmsg, real kernel segmentation over loopback. It asserts
  571. // both that GSO *engaged* (the whole batch left in a single sendmmsg
  572. // entry -- a silent fallback to per-packet entries fails the test) and
  573. // that the kernel carved the superpacket back into the exact original
  574. // datagrams on the receive side. Runs in CI (make test on ubuntu-latest),
  575. // which is what guards against the offload path silently degrading.
  576. func TestGSOEngagesOnLoopback(t *testing.T) {
  577. rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
  578. if err != nil {
  579. t.Fatalf("listen rx: %v", err)
  580. }
  581. defer rx.Close()
  582. dst := rx.LocalAddr().(*net.UDPAddr).AddrPort()
  583. udpSettings := Settings{
  584. Listen: netip.MustParseAddrPort("127.0.0.1:0"),
  585. Multi: false,
  586. Batch: 8,
  587. Offloads: true,
  588. }
  589. uc, err := NewListener(testLogger(), udpSettings)
  590. if err != nil {
  591. t.Fatalf("NewListener: %v", err)
  592. }
  593. sc := uc.(*StdConn)
  594. defer sc.Close()
  595. if !sc.bw.gsoSupported {
  596. var un unix.Utsname
  597. _ = unix.Uname(&un)
  598. release := string(un.Release[:])
  599. if major, minor := parseRelease(release); major > 4 || (major == 4 && minor >= 18) {
  600. t.Fatalf("kernel %q supports UDP_SEGMENT but the GSO probe failed", release)
  601. }
  602. t.Skipf("kernel %q predates UDP_SEGMENT (4.18)", release)
  603. }
  604. // Spy on the real syscall to count entries per sendmmsg without
  605. // changing what hits the kernel.
  606. var entryCounts []int
  607. real := sc.bw.sendFn
  608. sc.bw.sendFn = func(start, n int) (int, error) {
  609. entryCounts = append(entryCounts, n)
  610. return real(start, n)
  611. }
  612. const numPkts = 8
  613. const pktLen = 1200
  614. bufs := make([][]byte, numPkts)
  615. addrs := make([]netip.AddrPort, numPkts)
  616. for i := range bufs {
  617. bufs[i] = make([]byte, pktLen)
  618. for j := range bufs[i] {
  619. bufs[i][j] = byte(i)
  620. }
  621. addrs[i] = dst
  622. }
  623. written, err := sc.WriteBatch(bufs, addrs)
  624. if err != nil {
  625. t.Fatalf("WriteBatch: %v", err)
  626. }
  627. if written != numPkts {
  628. t.Fatalf("written = %d, want %d", written, numPkts)
  629. }
  630. // GSO engaged means the run went out as ONE sendmmsg entry carrying a
  631. // UDP_SEGMENT superpacket. Per-packet entries mean it silently fell
  632. // back -- exactly the regression this test exists to catch.
  633. if len(entryCounts) != 1 || entryCounts[0] != 1 {
  634. t.Fatalf("sendmmsg entry counts = %v, want [1]: GSO did not engage", entryCounts)
  635. }
  636. // The kernel must deliver the original datagram boundaries and bytes.
  637. _ = rx.SetReadDeadline(time.Now().Add(5 * time.Second))
  638. got := make([]byte, pktLen+1)
  639. for i := range numPkts {
  640. n, _, err := rx.ReadFromUDP(got)
  641. if err != nil {
  642. t.Fatalf("rx read %d: %v", i, err)
  643. }
  644. if n != pktLen {
  645. t.Fatalf("rx read %d: len=%d want %d (kernel segmented at wrong boundary)", i, n, pktLen)
  646. }
  647. for j := range n {
  648. if got[j] != byte(i) {
  649. t.Fatalf("rx read %d: byte %d = %#x, want %#x", i, j, got[j], byte(i))
  650. }
  651. }
  652. }
  653. }