diff --git a/control_lifecycle_test.go b/control_lifecycle_test.go index 4ec331a7..1eef98c8 100644 --- a/control_lifecycle_test.go +++ b/control_lifecycle_test.go @@ -144,11 +144,11 @@ type fakeConn struct { rebinds int } -func (c *fakeConn) Rebind() error { c.rebinds++; return nil } -func (c *fakeConn) LocalAddr() (netip.AddrPort, error) { return netip.AddrPort{}, nil } -func (c *fakeConn) ListenOut(_ udp.EncReader, _ func()) error { return nil } -func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort, _ byte) error { return nil } -func (c *fakeConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) { +func (c *fakeConn) Rebind() error { c.rebinds++; return nil } +func (c *fakeConn) LocalAddr() (netip.AddrPort, error) { return netip.AddrPort{}, nil } +func (c *fakeConn) ListenOut(_ udp.EncReader, _ func()) error { return nil } +func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort) error { return nil } +func (c *fakeConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort) (int, error) { return len(bufs), nil } func (c *fakeConn) ReloadConfig(_ *config.C) {} diff --git a/ecn_inner_test.go b/ecn_inner_test.go deleted file mode 100644 index 18dc94d4..00000000 --- a/ecn_inner_test.go +++ /dev/null @@ -1,188 +0,0 @@ -package nebula - -import ( - "encoding/binary" - "log/slog" - "testing" - - "golang.org/x/net/ipv4" -) - -func TestInnerECN(t *testing.T) { - cases := []struct { - name string - pkt []byte - want byte - }{ - {"empty", nil, 0}, - {"v4_NotECT", v4WithToS(0x00), 0x00}, - {"v4_ECT0", v4WithToS(0x02), 0x02}, - {"v4_ECT1", v4WithToS(0x01), 0x01}, - {"v4_CE", v4WithToS(0x03), 0x03}, - {"v4_DSCP_then_NotECT", v4WithToS(0x88 | 0x00), 0x00}, - {"v4_DSCP_then_CE", v4WithToS(0x88 | 0x03), 0x03}, - {"v6_NotECT", v6WithTC(0x00), 0x00}, - {"v6_ECT0", v6WithTC(0x02), 0x02}, - {"v6_CE", v6WithTC(0x03), 0x03}, - {"v6_DSCP_then_CE", v6WithTC(0x88 | 0x03), 0x03}, - {"unknown_version", []byte{0xa5, 0xff}, 0}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - got := innerECN(c.pkt) - if got != c.want { - t.Errorf("innerECN=0x%02x want 0x%02x", got, c.want) - } - }) - } -} - -// v4WithToS returns a 2-byte slice tall enough for innerECN: byte 0 carries -// version=4 in the high nibble, byte 1 is the full ToS so we exercise both -// the DSCP and ECN portions through the byte 1 mask. -func v4WithToS(tos byte) []byte { - return []byte{0x45, tos} -} - -// v6WithTC builds a 2-byte slice that places a known traffic class value -// across bytes 0 (high nibble of TC) and 1 (low nibble of TC). innerECN -// extracts ECN as (b[1]>>4)&0x03, which corresponds to TC[1:0]. -func v6WithTC(tc byte) []byte { - return []byte{0x60 | (tc>>4)&0x0f, (tc & 0x0f) << 4} -} - -func TestApplyOuterECN(t *testing.T) { - silent := slog.New(slog.DiscardHandler) - hi := &HostInfo{} - - // Build a v4 packet helper with a given inner ECN field. - v4 := func(innerECN byte) []byte { - // 20-byte minimal IPv4 header with ToS = innerECN (DSCP zeroed). - return []byte{ - 0x45, innerECN, 0, 28, - 0, 0, 0x40, 0, - 64, 6, 0, 0, - 10, 0, 0, 1, - 10, 0, 0, 2, - } - } - // Build a v6 packet helper with a given inner ECN field. ECN occupies - // TC[1:0] which sit at byte 1 mask 0x30. - v6 := func(innerECN byte) []byte { - // 40-byte minimal IPv6 header with TC[1:0] = innerECN. - pkt := make([]byte, 40) - pkt[0] = 0x60 // version=6, TC[7:4]=0 - pkt[1] = (innerECN & 0x03) << 4 // TC[3:0]: low 2 bits = ECN, top 2 = DSCP-low (0) - return pkt - } - - type cell struct { - outer byte - inner byte - wantECN byte - wantSame bool // expect inner unchanged (true => verify the byte didn't move) - } - - // RFC 6040 normal-mode combine table. Only outer==CE causes mutation. - table := []cell{ - {ecnNotECT, ecnNotECT, ecnNotECT, true}, - {ecnNotECT, ecnECT0, ecnECT0, true}, - {ecnNotECT, ecnECT1, ecnECT1, true}, - {ecnNotECT, ecnCE, ecnCE, true}, - - {ecnECT0, ecnNotECT, ecnNotECT, true}, - {ecnECT0, ecnECT0, ecnECT0, true}, - {ecnECT0, ecnECT1, ecnECT1, true}, - {ecnECT0, ecnCE, ecnCE, true}, - - {ecnECT1, ecnNotECT, ecnNotECT, true}, - {ecnECT1, ecnECT0, ecnECT0, true}, - {ecnECT1, ecnECT1, ecnECT1, true}, - {ecnECT1, ecnCE, ecnCE, true}, - - {ecnCE, ecnNotECT, ecnNotECT, true}, // legacy: log, leave alone - {ecnCE, ecnECT0, ecnCE, false}, // CE folded in - {ecnCE, ecnECT1, ecnCE, false}, - {ecnCE, ecnCE, ecnCE, true}, - } - - for _, c := range table { - t.Run("v4", func(t *testing.T) { - pkt := v4(c.inner) - applyOuterECN(pkt, c.outer, hi, silent) - got := pkt[1] & 0x03 - if got != c.wantECN { - t.Errorf("v4 outer=0x%02x inner=0x%02x: got 0x%02x want 0x%02x", c.outer, c.inner, got, c.wantECN) - } - }) - t.Run("v6", func(t *testing.T) { - pkt := v6(c.inner) - applyOuterECN(pkt, c.outer, hi, silent) - got := (pkt[1] >> 4) & 0x03 - if got != c.wantECN { - t.Errorf("v6 outer=0x%02x inner=0x%02x: got 0x%02x want 0x%02x", c.outer, c.inner, got, c.wantECN) - } - }) - } -} - -// TestApplyOuterECN_IPv4ChecksumStaysValid guards against H1: folding an outer -// CE mark into the inner IPv4 ToS byte must keep the IPv4 header checksum valid. -// The passthrough emit paths write the packet verbatim, so a stale checksum -// turns an underlay congestion mark into packet loss at the receiver. -func TestApplyOuterECN_IPv4ChecksumStaysValid(t *testing.T) { - silent := slog.New(slog.DiscardHandler) - hi := &HostInfo{} - - // 20-byte IPv4 header with DSCP=0x88 and inner ECN = ECT(0). Folding CE - // flips only the low two bits of the ToS byte while leaving DSCP intact. - pkt := []byte{ - 0x45, 0x88 | ecnECT0, 0, 40, - 0x1c, 0x46, 0x40, 0x00, - 64, 6, 0, 0, - 10, 0, 0, 1, - 10, 0, 0, 2, - } - // Stamp a correct header checksum before the fold. - binary.BigEndian.PutUint16(pkt[10:12], ipv4HeaderChecksum(pkt[:ipv4.HeaderLen])) - if !ipv4HeaderChecksumValid(pkt[:ipv4.HeaderLen]) { - t.Fatal("test setup: initial header checksum invalid") - } - - applyOuterECN(pkt, ecnCE, hi, silent) - - // CE folded in, DSCP preserved. - if got, want := pkt[1], byte(0x88|ecnCE); got != want { - t.Fatalf("ToS after fold = 0x%02x, want 0x%02x", got, want) - } - // The incremental RFC 1624 update must leave the checksum valid and equal - // to a full recompute over the mutated header. - if !ipv4HeaderChecksumValid(pkt[:ipv4.HeaderLen]) { - t.Fatalf("IPv4 header checksum invalid after CE fold: 0x%04x", binary.BigEndian.Uint16(pkt[10:12])) - } - if got, want := binary.BigEndian.Uint16(pkt[10:12]), ipv4HeaderChecksum(pkt[:ipv4.HeaderLen]); got != want { - t.Fatalf("checksum = 0x%04x, full recompute = 0x%04x", got, want) - } -} - -// ipv4HeaderChecksum computes the RFC 1071 IPv4 header checksum over hdr, -// treating the checksum field (bytes 10:12) as zero. -func ipv4HeaderChecksum(hdr []byte) uint16 { - var sum uint32 - for i := 0; i+1 < len(hdr); i += 2 { - if i == 10 { - continue // checksum field - } - sum += uint32(hdr[i])<<8 | uint32(hdr[i+1]) - } - for sum > 0xffff { - sum = (sum >> 16) + (sum & 0xffff) - } - return ^uint16(sum) -} - -// ipv4HeaderChecksumValid reports whether the stored checksum matches a fresh -// computation over the header. -func ipv4HeaderChecksumValid(hdr []byte) bool { - return binary.BigEndian.Uint16(hdr[10:12]) == ipv4HeaderChecksum(hdr) -} diff --git a/handshake_manager.go b/handshake_manager.go index d252f315..4942d6de 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -293,7 +293,7 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered var sentTo []netip.AddrPort hostinfo.remotes.ForEach(hm.mainHostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) { hm.messageMetrics.Tx(header.Handshake, hh.machine.Subtype(), 1) - err := hm.outside.WriteTo(stage0, addr, 0) + err := hm.outside.WriteTo(stage0, addr) if err != nil { // These repeat every attempt, so match the success log below and only shout when the remotes changed level := slog.LevelDebug @@ -1074,7 +1074,7 @@ func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hos if !via.IsRelayed { fields := append(logFields, "from", via) - err := f.outside.WriteTo(msg, via.UdpAddr, 0) + err := f.outside.WriteTo(msg, via.UdpAddr) if err != nil { f.l.Error("Failed to send handshake message", append(fields, "error", err)...) } else { @@ -1089,7 +1089,7 @@ func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hos // We received a valid handshake on this relay, so make sure the relay // state reflects that, in case it had been marked Disestablished. via.relayHI.relayState.UpdateRelayForByIdxState(via.relay.LocalIndex, Established) - f.SendVia(via.relayHI, via.relay, msg, make([]byte, 12), make([]byte, mtu), false, 0, 0) + f.SendVia(via.relayHI, via.relay, msg, make([]byte, 12), make([]byte, mtu), false, 0) f.l.Info("Handshake message sent", append(logFields, "relay", via.relayHI.vpnAddrs[0])...) } } diff --git a/handshake_manager_test.go b/handshake_manager_test.go index 7915c97c..03483d87 100644 --- a/handshake_manager_test.go +++ b/handshake_manager_test.go @@ -84,7 +84,7 @@ func (mw *mockEncWriter) SendMessageToVpnAddr(_ header.MessageType, _ header.Mes return } -func (mw *mockEncWriter) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) { +func (mw *mockEncWriter) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, q int) { return } diff --git a/inside.go b/inside.go index 30468093..cee7a787 100644 --- a/inside.go +++ b/inside.go @@ -164,7 +164,6 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b f.connectionManager.Out(hostinfo) remote := hostinfo.GetRemote() - ecnEnabled := f.ecnEnabled.Load() if hostinfo.lastRebindCount != f.rebindCount { //NOTE: there is an update hole if a tunnel isn't used and exactly 256 rebinds occur before the tunnel is // finally used again. This tunnel would eventually be torn down and recreated if this action didn't help. @@ -215,11 +214,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return nil } - var ecn byte - if ecnEnabled { - ecn = innerECN(seg) - } - sendBatch.Commit(toSend, relayHostInfo.GetRemote(), ecn) + sendBatch.Commit(toSend, relayHostInfo.GetRemote()) return nil }) if err != nil { @@ -237,11 +232,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return nil } - var ecn byte - if ecnEnabled { - ecn = innerECN(seg) - } - sendBatch.Commit(out, remote, ecn) + sendBatch.Commit(out, remote) return nil }) if err != nil { @@ -251,22 +242,6 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b } } -// innerECN returns the 2-bit IP-level ECN codepoint of an inner IPv4 or IPv6 -// packet, or 0 if pkt is too short or its IP version is unrecognized. Used at -// encap to copy the inner codepoint onto the outer carrier per RFC 6040. -func innerECN(pkt []byte) byte { - if len(pkt) < 2 { - return 0 - } - switch pkt[0] >> 4 { - case 4: - return pkt[1] & 0x03 - case 6: - return (pkt[1] >> 4) & 0x03 - } - return 0 -} - func (f *Interface) rejectInside(packet []byte, out []byte, q int) { if !f.firewall.OutboundSendReject { return @@ -520,16 +495,15 @@ func (f *Interface) prepareSendVia(via *HostInfo, // ad is the plaintext data to authenticate, but not encrypt // nb is a buffer used to store the nonce value, re-used for performance reasons. // out is a buffer used to store the result of the Encrypt operation -// outerECN is the 2-bit codepoint to stamp on the carrier datagram (0 for control traffic). // q indicates which writer to use to send the packet. -func (f *Interface) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) { +func (f *Interface) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, q int) { toSend, err := f.prepareSendVia(via, relay, ad, nb, out, nocopy) if err != nil { // already logged by prepareSendVia return } - err = f.writers[q].WriteTo(toSend, via.GetRemote(), outerECN) + err = f.writers[q].WriteTo(toSend, via.GetRemote()) if err != nil { via.logger(f.l).Info("Failed to WriteTo in sendVia", "error", err) } @@ -590,15 +564,8 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType return } - // Data packets copy the inner packet's ECN codepoint onto the outer - // carrier per RFC 6040; control traffic stays Not-ECT. - var outerECN byte - if t == header.Message && f.ecnEnabled.Load() { - outerECN = innerECN(p) - } - if remote.IsValid() { - err = f.writers[q].WriteTo(out, remote, outerECN) + err = f.writers[q].WriteTo(out, remote) if err != nil { hostinfo.logger(f.l).Error("Failed to write outgoing packet", "error", err, @@ -606,7 +573,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType ) } } else if hr := hostinfo.GetRemote(); hr.IsValid() { - err = f.writers[q].WriteTo(out, hr, outerECN) + err = f.writers[q].WriteTo(out, hr) if err != nil { hostinfo.logger(f.l).Error("Failed to write outgoing packet", "error", err, @@ -625,7 +592,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType ) continue } - f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true, outerECN, q) + f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true, q) break } } diff --git a/interface.go b/interface.go index 85003492..89fc95e9 100644 --- a/interface.go +++ b/interface.go @@ -96,12 +96,7 @@ type Interface struct { // pinThreads controls whether listenIn pins each TUN reader OS thread to // a CPU at all (tun.pin_threads, default true). When false, threads are // left free to migrate as on stock nebula. - pinThreads bool - // ecnEnabled gates RFC 6040 underlay ECN propagation. When true, - // inside.go copies the inner ECN onto the outer carrier on encap and - // decryptToTun folds outer CE into the inner header on decap. Toggle - // via tunnels.ecn (default false; see reloadEcn for why). - ecnEnabled atomic.Bool + pinThreads bool relayManager *relayManager tryPromoteEvery atomic.Uint32 @@ -143,7 +138,7 @@ type Interface struct { } type EncWriter interface { - SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) + SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, q int) SendMessageToVpnAddr(t header.MessageType, st header.MessageSubType, vpnAddr netip.Addr, p, nb, out []byte) SendMessageToHostInfo(t header.MessageType, st header.MessageSubType, hostinfo *HostInfo, p, nb, out []byte) Handshake(vpnAddr netip.Addr) @@ -369,8 +364,8 @@ func (f *Interface) listenOut(i int) { nb := make([]byte, 12, 12) scratch := make([]byte, mtu) - listener := func(fromUdpAddr netip.AddrPort, payload []byte, meta udp.RxMeta) { - f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, scratch, payload, h, fwPacket, lhh, nb, i, ctCache.Get(), meta) + listener := func(fromUdpAddr netip.AddrPort, payload []byte) { + f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, scratch, payload, h, fwPacket, lhh, nb, i, ctCache.Get()) } flusher := func() { @@ -472,7 +467,6 @@ func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) { c.RegisterReloadCallback(f.reloadAcceptRecvError) c.RegisterReloadCallback(f.reloadDisconnectInvalid) c.RegisterReloadCallback(f.reloadMisc) - c.RegisterReloadCallback(f.reloadEcn) for _, udpConn := range f.writers { c.RegisterReloadCallback(udpConn.ReloadConfig) @@ -605,32 +599,6 @@ func (f *Interface) reloadMisc(c *config.C) { } } -// reloadEcn syncs Interface.ecnEnabled with the tunnels.ecn config knob. -// -// Default is disabled (RFC 6040 compatibility mode). There is no in-band -// capability negotiation, and RFC 6040 §4.3 forbids marking the outer -// header ECT unless the ingress knows the egress propagates CE inward: a -// receiver that cannot read outer ECN (any pre-ECN nebula) silently -// discards AQM CE marks, so the inner flow is advertised as -// congestion-responsive but never sees the signal and never backs off. -// Setting tunnels.ecn=true is the operator's assertion that every peer -// this host tunnels with runs an ECN-capable nebula with the knob enabled; -// enable it fleet-wide or not at all. It is also the escape hatch for -// underlay middleboxes that rewrite or drop ECN bits unpredictably. -func (f *Interface) reloadEcn(c *config.C) { - initial := c.InitialLoad() - if initial || c.HasChanged("tunnels.ecn") { - v := c.GetBool("tunnels.ecn", true) //todo!!! - changed := f.ecnEnabled.Swap(v) != v - if !initial { - f.l.Info("tunnels.ecn changed", "enabled", v) - if changed { - f.l.Warn("tunnels.ecn datapath toggled, but route-level ECN negotiation (RTAX_FEATURE_ECN) retains its previous state until nebula is restarted", "enabled", v) - } - } - } -} - func (f *Interface) emitStats(ctx context.Context, i time.Duration) { ticker := time.NewTicker(i) defer ticker.Stop() diff --git a/lighthouse_test.go b/lighthouse_test.go index 6b15e162..7a81e5d2 100644 --- a/lighthouse_test.go +++ b/lighthouse_test.go @@ -498,7 +498,7 @@ type testEncWriter struct { protocolVersion cert.Version } -func (tw *testEncWriter) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, outerECN byte, q int) { +func (tw *testEncWriter) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, q int) { } func (tw *testEncWriter) Handshake(vpnIp netip.Addr) { } diff --git a/outside.go b/outside.go index 5f1b24a4..28be3520 100644 --- a/outside.go +++ b/outside.go @@ -13,7 +13,6 @@ import ( "github.com/slackhq/nebula/firewall" "github.com/slackhq/nebula/header" - "github.com/slackhq/nebula/udp" "golang.org/x/net/ipv4" ) @@ -26,7 +25,7 @@ var ErrOutOfWindow = errors.New("out of window packet") // readOutsidePackets processes one received underlay packet. // Message payloads are decrypted IN PLACE, so packet must stay untouched // by the caller until the batcher for queue q has been flushed -func (f *Interface) readOutsidePackets(via ViaSender, scratch []byte, packet []byte, h *header.H, fwPacket *firewall.Packet, lhf *LightHouseHandler, nb []byte, q int, localCache firewall.ConntrackCache, meta udp.RxMeta) { +func (f *Interface) readOutsidePackets(via ViaSender, scratch []byte, packet []byte, h *header.H, fwPacket *firewall.Packet, lhf *LightHouseHandler, nb []byte, q int, localCache firewall.ConntrackCache) { err := h.Parse(packet) if err != nil { // Hole punch packets are 0 or 1 byte big, so lets ignore printing those errors @@ -124,7 +123,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, scratch []byte, packet []b } return } - f.handleOutsideRelayPacket(hostinfo, via, scratch, packet, h, fwPacket, lhf, nb, q, localCache, meta) + f.handleOutsideRelayPacket(hostinfo, via, scratch, packet, h, fwPacket, lhf, nb, q, localCache) return } @@ -144,7 +143,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, scratch []byte, packet []b case header.Message: switch h.Subtype { case header.MessageNone: - f.handleOutsideMessagePacket(hostinfo, out, scratch, fwPacket, nb, q, localCache, meta) + f.handleOutsideMessagePacket(hostinfo, out, scratch, fwPacket, nb, q, localCache) default: hostinfo.logger(f.l).Error("IsValidSubType was true, but unexpected message subtype seen", "from", via, "header", h) return @@ -186,7 +185,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, scratch []byte, packet []b } } -func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, scratch []byte, packet []byte, h *header.H, fwPacket *firewall.Packet, lhf *LightHouseHandler, nb []byte, q int, localCache firewall.ConntrackCache, meta udp.RxMeta) { +func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, scratch []byte, packet []byte, h *header.H, fwPacket *firewall.Packet, lhf *LightHouseHandler, nb []byte, q int, localCache firewall.ConntrackCache) { // Successfully validated the thing. Get rid of the Relay header and the AEAD tag signedPayload := packet[header.Len : len(packet)-hostinfo.ConnectionState.dKey.Overhead()] // Pull the Roaming parts up here, and return in all call paths. @@ -213,7 +212,7 @@ func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, relay: relay, IsRelayed: true, } - f.readOutsidePackets(via, scratch, signedPayload, h, fwPacket, lhf, nb, q, localCache, meta) + f.readOutsidePackets(via, scratch, signedPayload, h, fwPacket, lhf, nb, q, localCache) case ForwardingType: // Find the target HostInfo relay object targetHI, targetRelay, err := f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relay.PeerAddr) @@ -233,12 +232,8 @@ func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, // Forward this packet through the relay tunnel, rebuilding it in place. // Encode overwrites the old outer header, and the new AEAD tag lands where the old one was fwdBuf := packet[:0:len(packet)] // Cap to len(packet) to protect memory from a larger parent buffer - var fwdECN byte - if f.ecnEnabled.Load() { - fwdECN = meta.OuterECN - } //todo it would potentially be nice to batch these - f.SendVia(targetHI, targetRelay, signedPayload, nb, fwdBuf, true, fwdECN, q) + f.SendVia(targetHI, targetRelay, signedPayload, nb, fwdBuf, true, q) case TerminalType: hostinfo.logger(f.l).Error("Unexpected Relay Type of Terminal") return @@ -505,85 +500,7 @@ func parseV4(data []byte, incoming bool, fp *firewall.Packet) error { return nil } -// 2-bit IP-level ECN codepoints (lower bits of IPv4 ToS / IPv6 TC). -const ( - ecnNotECT = 0x00 - ecnECT1 = 0x01 - ecnECT0 = 0x02 - ecnCE = 0x03 -) - -// applyOuterECN folds an outer CE mark from the underlay into the inner -// IP header per RFC 6040 normal mode. It mutates pkt[1] in place. Other -// codepoints are advisory only and leave the inner unchanged. -// -// Merge cases (outer × inner → action): -// -// outer != CE : no-op (inner is authoritative) -// outer == CE, inner Not-ECT : log; cannot propagate to a non-ECN host -// outer == CE, inner ECT/CE : rewrite inner ECN to CE -func applyOuterECN(pkt []byte, outerECN byte, hostinfo *HostInfo, l *slog.Logger) { - if outerECN&ecnCE != ecnCE || len(pkt) < 2 { - return - } - switch pkt[0] >> 4 { - case 4: - switch pkt[1] & 0x03 { - case ecnNotECT: - if l.Enabled(context.Background(), slog.LevelDebug) { - hostinfo.logger(l).Debug("RFC 6040: outer CE on inner Not-ECT, leaving inner unchanged") - } - case ecnCE: - // Already CE. - default: - // Rewriting the ToS byte invalidates the IPv4 header checksum, so - // patch it incrementally per RFC 1624 (HC' = ~(~HC + ~m + m')). The - // ToS is the low byte of the 16-bit word at pkt[0:2]; the header - // checksum lives at pkt[10:12]. A header too short to carry a - // checksum can't be fixed up here, so leave it for newPacket to - // reject rather than emit a mangled packet. - if len(pkt) < ipv4.HeaderLen { - return - } - m := binary.BigEndian.Uint16(pkt[0:2]) - pkt[1] = (pkt[1] &^ 0x03) | ecnCE - mNew := binary.BigEndian.Uint16(pkt[0:2]) - sum := uint32(^binary.BigEndian.Uint16(pkt[10:12])) + uint32(^m) + uint32(mNew) - for sum > 0xffff { - sum = (sum >> 16) + (sum & 0xffff) - } - binary.BigEndian.PutUint16(pkt[10:12], ^uint16(sum)) - } - case 6: - switch (pkt[1] >> 4) & 0x03 { - case ecnNotECT: - if l.Enabled(context.Background(), slog.LevelDebug) { - hostinfo.logger(l).Debug("RFC 6040: outer CE on inner Not-ECT, leaving inner unchanged") - } - case ecnCE: - // Already CE. - default: - pkt[1] = (pkt[1] &^ 0x30) | (ecnCE << 4) - } - } -} - -func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, scratch []byte, fwPacket *firewall.Packet, nb []byte, q int, localCache firewall.ConntrackCache, meta udp.RxMeta) { - // RFC 6040 normal-mode combine: fold any outer CE mark stamped by the - // underlay into the inner header before firewall + TUN write. Other - // outer codepoints are advisory only — we keep the inner unchanged. - if f.ecnEnabled.Load() { - outerECN := meta.OuterECN - if meta.QueueCongested { - // nebula-as-AQM: our own receive queue is the congested hop on - // this path and no kernel AQM can see it. Depth beyond the - // marking threshold is treated as CE so ECT senders back off - // before the queue regulates by tail-drop instead. - outerECN = ecnCE - } - applyOuterECN(out, outerECN, hostinfo, f.l) - } - +func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, scratch []byte, fwPacket *firewall.Packet, nb []byte, q int, localCache firewall.ConntrackCache) { err := newPacket(out, true, fwPacket) if err != nil { hostinfo.logger(f.l).Warn("Error while validating inbound packet", @@ -621,7 +538,7 @@ func (f *Interface) sendRecvError(endpoint netip.AddrPort, index uint32) { f.messageMetrics.Tx(header.RecvError, 0, 1) b := header.Encode(make([]byte, header.Len), header.Version, header.RecvError, 0, index, 0) - _ = f.outside.WriteTo(b, endpoint, 0) + _ = f.outside.WriteTo(b, endpoint) if f.l.Enabled(context.Background(), slog.LevelDebug) { f.l.Debug("Recv error sent", "index", index, diff --git a/overlay/batch/batch.go b/overlay/batch/batch.go index d34f7751..5cb6cbe4 100644 --- a/overlay/batch/batch.go +++ b/overlay/batch/batch.go @@ -19,11 +19,9 @@ type RxBatcher interface { type TxBatcher interface { // Reserve creates a pkt to borrow Reserve(sz int) []byte - // Commit borrows pkt and records its destination plus the 2-bit - // IP-level ECN codepoint to set on the outer (carrier) header. The - // caller must keep pkt valid until the next Flush. Pass 0 (Not-ECT) - // to leave the outer ECN field unset. - Commit(pkt []byte, dst netip.AddrPort, outerECN byte) + // Commit borrows pkt and records its destination. The caller must + // keep pkt valid until the next Flush. + Commit(pkt []byte, dst netip.AddrPort) // Flush emits every queued packet via the underlying batch writer in arrival order and reports how many were // actually written. A short count means some destinations were undeliverable, not that the batch failed. // After Flush returns, borrowed payload slices may be recycled. diff --git a/overlay/batch/tx_batch.go b/overlay/batch/tx_batch.go index a78943a5..4f6f7da2 100644 --- a/overlay/batch/tx_batch.go +++ b/overlay/batch/tx_batch.go @@ -6,7 +6,7 @@ const SendBatchCap = 128 // batchWriter is the minimal subset of udp.Conn needed by SendBatch to flush. type batchWriter interface { - WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) (int, error) + WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) } // SendBatch accumulates encrypted UDP packets and flushes them via WriteBatch. @@ -16,7 +16,6 @@ type SendBatch struct { out batchWriter bufs [][]byte dsts []netip.AddrPort - ecns []byte arena *Arena } @@ -26,7 +25,6 @@ func NewSendBatch(out batchWriter, batchCap, arenaSize int) *SendBatch { out: out, bufs: make([][]byte, 0, batchCap), dsts: make([]netip.AddrPort, 0, batchCap), - ecns: make([]byte, 0, batchCap), arena: NewArena(arenaSize), } } @@ -40,10 +38,9 @@ func (b *SendBatch) Reserve(sz int) []byte { // bounding how long the first packet of a large read batch waits. func (b *SendBatch) Len() int { return len(b.bufs) } -func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort, outerECN byte) { +func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort) { b.bufs = append(b.bufs, pkt) b.dsts = append(b.dsts, dst) - b.ecns = append(b.ecns, outerECN) } // Flush writes every queued packet and reports how many actually went out. A short count means some destinations @@ -52,12 +49,11 @@ func (b *SendBatch) Flush() (int, error) { var err error written := 0 if len(b.bufs) > 0 { - written, err = b.out.WriteBatch(b.bufs, b.dsts, b.ecns) + written, err = b.out.WriteBatch(b.bufs, b.dsts) } clear(b.bufs) b.bufs = b.bufs[:0] b.dsts = b.dsts[:0] - b.ecns = b.ecns[:0] b.arena.Reset() return written, err } diff --git a/overlay/batch/tx_batch_test.go b/overlay/batch/tx_batch_test.go index d314784a..9a2a75b4 100644 --- a/overlay/batch/tx_batch_test.go +++ b/overlay/batch/tx_batch_test.go @@ -8,10 +8,9 @@ import ( type fakeBatchWriter struct { bufs [][]byte addrs []netip.AddrPort - ecns []byte } -func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) (int, error) { +func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { // Snapshot — SendBatch.Flush nils its slot pointers right after WriteBatch // returns, so tests must capture data before that happens. w.bufs = make([][]byte, len(bufs)) @@ -21,7 +20,6 @@ func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns w.bufs[i] = cp } w.addrs = append(w.addrs[:0], addrs...) - w.ecns = append(w.ecns[:0], ecns...) return len(bufs), nil } @@ -36,7 +34,7 @@ func TestSendBatchReserveCommitFlush(t *testing.T) { t.Fatalf("slot %d: cap=%d want 32", i, cap(slot)) } pkt := append(slot[:0], byte(i), byte(i+1), byte(i+2)) - b.Commit(pkt, ap, 0) + b.Commit(pkt, ap) } if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) @@ -77,7 +75,7 @@ func TestSendBatchSlotsDoNotOverlap(t *testing.T) { for i := 0; i < 3; i++ { s := b.Reserve(8) pkt := append(s[:0], byte(0xA0+i), byte(0xB0+i)) - b.Commit(pkt, ap, 0) + b.Commit(pkt, ap) } if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) @@ -98,11 +96,11 @@ func TestSendBatchGrowPreservesCommitted(t *testing.T) { s1 := b.Reserve(4) pkt1 := append(s1[:0], 0x11, 0x22, 0x33, 0x44) - b.Commit(pkt1, ap, 0) + b.Commit(pkt1, ap) s2 := b.Reserve(8) // exceeds remaining cap, triggers grow pkt2 := append(s2[:0], 0xA, 0xB, 0xC, 0xD, 0xE) - b.Commit(pkt2, ap, 0) + b.Commit(pkt2, ap) // pkt1 must still be intact even though backing reallocated. if pkt1[0] != 0x11 || pkt1[3] != 0x44 { diff --git a/punchy.go b/punchy.go index 915942fa..4bce4392 100644 --- a/punchy.go +++ b/punchy.go @@ -176,7 +176,7 @@ func (p *Punchy) SendPunch(hostinfo *HostInfo) { p.sendPunchToAllRemotes(hostinfo) } else if hr := hostinfo.GetRemote(); hr.IsValid() { p.metricPunchyTx.Inc(1) - p.punchConn.WriteTo([]byte{1}, hr, 0) + p.punchConn.WriteTo([]byte{1}, hr) } } @@ -200,7 +200,7 @@ func (p *Punchy) SendPunchToAll(hostinfo *HostInfo) { func (p *Punchy) sendPunchToAllRemotes(hostinfo *HostInfo) { hostinfo.remotes.ForEach(p.hm.GetPreferredRanges(), func(addr netip.AddrPort, preferred bool) { p.metricPunchyTx.Inc(1) - p.punchConn.WriteTo([]byte{1}, addr, 0) + p.punchConn.WriteTo([]byte{1}, addr) }) } @@ -222,7 +222,7 @@ func (p *Punchy) Start(ctx context.Context, ifce EncWriter, hm *HostMap, lh ligh p.l.Debug("Punching", "target", job.target, "vpnAddr", job.vpnAddr) } p.metricHolepunchTx.Inc(1) - p.punchConn.WriteTo(empty, job.target, 0) + p.punchConn.WriteTo(empty, job.target) case job.vpnAddr.IsValid(): // A nebula test packet to the host trying to contact us. // In the case of a double nat or other difficult scenario, this may help establish a tunnel. diff --git a/relay_manager.go b/relay_manager.go index 0aa7ff4b..46d7a2bb 100644 --- a/relay_manager.go +++ b/relay_manager.go @@ -161,7 +161,7 @@ func (rm *relayManager) StartRelays(f *Interface, vpnIp netip.Addr, hh *Handshak switch existingRelay.State { case Established: hl.Log(context.Background(), level, "Send handshake via relay", "relay", relay.String()) - f.SendVia(relayHostInfo, existingRelay, stage0, make([]byte, 12), make([]byte, mtu), false, 0, 0) + f.SendVia(relayHostInfo, existingRelay, stage0, make([]byte, 12), make([]byte, mtu), false, 0) case Disestablished: // Mark this relay as 'requested' relayHostInfo.relayState.UpdateRelayForByIpState(vpnIp, Requested) diff --git a/udp/conn.go b/udp/conn.go index ab517b20..2d407981 100644 --- a/udp/conn.go +++ b/udp/conn.go @@ -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) { diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index 60c0d065..b381920f 100644 --- a/udp/udp_darwin.go +++ b/udp/udp_darwin.go @@ -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() } } diff --git a/udp/udp_ecn_outer_linux_test.go b/udp/udp_ecn_outer_linux_test.go deleted file mode 100644 index c8e4a916..00000000 --- a/udp/udp_ecn_outer_linux_test.go +++ /dev/null @@ -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) - } - }) - } -} diff --git a/udp/udp_generic.go b/udp/udp_generic.go index 975e6549..9735592c 100644 --- a/udp/udp_generic.go +++ b/udp/udp_generic.go @@ -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() } } diff --git a/udp/udp_linux.go b/udp/udp_linux.go index 0b00a2f2..001555a3 100644 --- a/udp/udp_linux.go +++ b/udp/udp_linux.go @@ -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) diff --git a/udp/udp_linux_fixes_test.go b/udp/udp_linux_fixes_test.go index 03696b76..4ed3514d 100644 --- a/udp/udp_linux_fixes_test.go +++ b/udp/udp_linux_fixes_test.go @@ -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) } diff --git a/udp/udp_linux_test.go b/udp/udp_linux_test.go index cbc73b87..5eca0938 100644 --- a/udp/udp_linux_test.go +++ b/udp/udp_linux_test.go @@ -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() {}) }() diff --git a/udp/udp_linux_writebatch.go b/udp/udp_linux_writebatch.go index 00bc95cb..bc94464c 100644 --- a/udp/udp_linux_writebatch.go +++ b/udp/udp_linux_writebatch.go @@ -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) } diff --git a/udp/udp_linux_writebatch_alloc_test.go b/udp/udp_linux_writebatch_alloc_test.go index ffc641cb..5f0c2cd5 100644 --- a/udp/udp_linux_writebatch_alloc_test.go +++ b/udp/udp_linux_writebatch_alloc_test.go @@ -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) } }) } diff --git a/udp/udp_rio_windows.go b/udp/udp_rio_windows.go index f9a555cf..566a44d9 100644 --- a/udp/udp_rio_windows.go +++ b/udp/udp_rio_windows.go @@ -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) diff --git a/udp/udp_tester.go b/udp/udp_tester.go index 8d1ddc34..13d0bc15 100644 --- a/udp/udp_tester.go +++ b/udp/udp_tester.go @@ -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()