From bb4625008744b1b2d43e63923ecedda55824491e Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Thu, 2 Jul 2026 11:54:07 -0400 Subject: [PATCH] feat(zingela): M7 AF_XDP TX backend behind -Dxdp - pure-syscall UMEM + 4 rings, zero-copy/SKB/AF_PACKET selection ladder, asymmetric AF_XDP-TX + AF_PACKET-RX --- .../advanced/zig-stateless-scanner/build.zig | 36 ++- .../zig-stateless-scanner/src/afxdp.zig | 220 +++++++++++++ .../zig-stateless-scanner/src/cli.zig | 1 + .../zig-stateless-scanner/src/packet_io.zig | 112 +++++++ .../zig-stateless-scanner/src/scancmd.zig | 17 +- .../zig-stateless-scanner/src/txcmd.zig | 15 +- .../zig-stateless-scanner/src/xdp.zig | 296 ++++++++++++++++++ 7 files changed, 687 insertions(+), 10 deletions(-) create mode 100644 PROJECTS/advanced/zig-stateless-scanner/src/afxdp.zig create mode 100644 PROJECTS/advanced/zig-stateless-scanner/src/packet_io.zig create mode 100644 PROJECTS/advanced/zig-stateless-scanner/src/xdp.zig diff --git a/PROJECTS/advanced/zig-stateless-scanner/build.zig b/PROJECTS/advanced/zig-stateless-scanner/build.zig index 68e49199..09bd2b57 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/build.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/build.zig @@ -7,8 +7,12 @@ pub fn build(b: *std.Build) void { const target = b.standardTargetOptions(.{}); const optimize = b.standardOptimizeOption(.{}); + const xdp_enabled = b.option(bool, "xdp", "Enable the AF_XDP TX backend (pure-syscall, no libxdp; needs CAP_NET_ADMIN at runtime)") orelse false; + const opts = b.addOptions(); - opts.addOption([]const u8, "version", "0.0.0-m6"); + opts.addOption([]const u8, "version", "0.0.0-m7"); + opts.addOption(bool, "xdp", xdp_enabled); + const build_config_mod = opts.createModule(); const packet_mod = b.createModule(.{ .root_source_file = b.path("src/packet.zig"), @@ -21,7 +25,7 @@ pub fn build(b: *std.Build) void { .target = target, .optimize = optimize, }); - cli_mod.addOptions("build_config", opts); + cli_mod.addImport("build_config", build_config_mod); const smoke_mod = b.createModule(.{ .root_source_file = b.path("src/smoke.zig"), @@ -85,6 +89,28 @@ pub fn build(b: *std.Build) void { }); afpacket_mod.addImport("packet", packet_mod); + const xdp_mod = b.createModule(.{ + .root_source_file = b.path("src/xdp.zig"), + .target = target, + .optimize = optimize, + }); + + const afxdp_mod = b.createModule(.{ + .root_source_file = b.path("src/afxdp.zig"), + .target = target, + .optimize = optimize, + }); + afxdp_mod.addImport("xdp", xdp_mod); + + const packet_io_mod = b.createModule(.{ + .root_source_file = b.path("src/packet_io.zig"), + .target = target, + .optimize = optimize, + }); + packet_io_mod.addImport("afpacket", afpacket_mod); + packet_io_mod.addImport("afxdp", afxdp_mod); + packet_io_mod.addImport("build_config", build_config_mod); + const tx_mod = b.createModule(.{ .root_source_file = b.path("src/tx.zig"), .target = target, @@ -141,7 +167,7 @@ pub fn build(b: *std.Build) void { txcmd_mod.addImport("targets", targets_mod); txcmd_mod.addImport("template", template_mod); txcmd_mod.addImport("ratelimit", ratelimit_mod); - txcmd_mod.addImport("afpacket", afpacket_mod); + txcmd_mod.addImport("packet_io", packet_io_mod); txcmd_mod.addImport("cookie", cookie_mod); txcmd_mod.addImport("tx", tx_mod); txcmd_mod.addImport("netutil", netutil_mod); @@ -155,7 +181,7 @@ pub fn build(b: *std.Build) void { scancmd_mod.addImport("template", template_mod); scancmd_mod.addImport("udp", udp_mod); scancmd_mod.addImport("ratelimit", ratelimit_mod); - scancmd_mod.addImport("afpacket", afpacket_mod); + scancmd_mod.addImport("packet_io", packet_io_mod); scancmd_mod.addImport("cookie", cookie_mod); scancmd_mod.addImport("tx", tx_mod); scancmd_mod.addImport("rx", rx_mod); @@ -192,7 +218,7 @@ pub fn build(b: *std.Build) void { smoke_step.dependOn(&smoke_cmd.step); const test_step = b.step("test", "Run unit tests"); - const test_mods = [_]*std.Build.Module{ packet_mod, cli_mod, smoke_mod, cookie_mod, numtheory_mod, targets_mod, ratelimit_mod, template_mod, payloads_mod, udp_mod, afpacket_mod, tx_mod, txcmd_mod, classify_mod, dedup_mod, rx_mod, netutil_mod, output_mod, scancmd_mod }; + const test_mods = [_]*std.Build.Module{ packet_mod, cli_mod, smoke_mod, cookie_mod, numtheory_mod, targets_mod, ratelimit_mod, template_mod, payloads_mod, udp_mod, afpacket_mod, xdp_mod, afxdp_mod, packet_io_mod, tx_mod, txcmd_mod, classify_mod, dedup_mod, rx_mod, netutil_mod, output_mod, scancmd_mod }; for (test_mods) |mod| { const t = b.addTest(.{ .root_module = mod }); const rt = b.addRunArtifact(t); diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/afxdp.zig b/PROJECTS/advanced/zig-stateless-scanner/src/afxdp.zig new file mode 100644 index 00000000..e17135e1 --- /dev/null +++ b/PROJECTS/advanced/zig-stateless-scanner/src/afxdp.zig @@ -0,0 +1,220 @@ +// ©AngelaMos | 2026 +// afxdp.zig + +const std = @import("std"); +const xdp = @import("xdp"); +const linux = std.os.linux; + +const page = std.heap.page_size_min; + +pub const Mode = enum { zerocopy, copy }; + +pub const OpenError = error{ + NeedCapNetRaw, + SocketFailed, + IfIndexFailed, + UmemAllocFailed, + UmemRegFailed, + RingSetupFailed, + MmapOffsetsFailed, + RingMmapFailed, + BindFailed, + BadConfig, + OutOfMemory, +}; + +pub const Config = struct { + frame_size: u32 = 2048, + num_frames: u32 = 4096, + tx_size: u32 = 2048, + comp_size: u32 = 2048, + fill_size: u32 = 2048, + queue_id: u32 = 0, +}; + +fn setU32(fd: i32, name: u32, val: u32) OpenError!void { + var v = val; + if (linux.errno(linux.setsockopt(fd, xdp.SOL_XDP, name, std.mem.asBytes(&v), @sizeOf(u32))) != .SUCCESS) + return error.RingSetupFailed; +} + +fn mapRing(fd: i32, len: usize, pg: u64) OpenError![]align(page) u8 { + const rc = linux.mmap(null, len, .{ .READ = true, .WRITE = true }, .{ .TYPE = .SHARED, .POPULATE = true }, fd, @intCast(pg)); + if (linux.errno(rc) != .SUCCESS) return error.RingMmapFailed; + const ptr: [*]align(page) u8 = @ptrFromInt(rc); + return ptr[0..len]; +} + +fn ringPtr(comptime T: type, map: []align(page) u8, offset: u64) T { + return @ptrCast(@alignCast(map.ptr + @as(usize, @intCast(offset)))); +} + +pub const Backend = struct { + allocator: std.mem.Allocator, + fd: i32, + mode: Mode, + frame_size: u32, + umem: []align(page) u8, + tx_map: []align(page) u8, + comp_map: []align(page) u8, + tx: xdp.Prod, + comp: xdp.Comp, + frames: xdp.FrameStack, + frame_backing: []u64, + + pub fn open(allocator: std.mem.Allocator, ifname: []const u8, mode: Mode, cfg: Config) OpenError!Backend { + if (!std.math.isPowerOfTwo(cfg.tx_size) or !std.math.isPowerOfTwo(cfg.comp_size) or !std.math.isPowerOfTwo(cfg.fill_size)) + return error.BadConfig; + if (cfg.frame_size != 2048 and cfg.frame_size != 4096) return error.BadConfig; + if (cfg.num_frames < cfg.tx_size or cfg.num_frames < cfg.comp_size) return error.BadConfig; + + const rc_sock = linux.socket(xdp.AF_XDP, linux.SOCK.RAW, 0); + switch (linux.errno(rc_sock)) { + .SUCCESS => {}, + .PERM, .ACCES => return error.NeedCapNetRaw, + else => return error.SocketFailed, + } + const fd: i32 = @intCast(rc_sock); + errdefer _ = linux.close(fd); + + var ifr = std.mem.zeroes(linux.ifreq); + if (ifname.len >= ifr.ifrn.name.len) return error.IfIndexFailed; + @memcpy(ifr.ifrn.name[0..ifname.len], ifname); + if (linux.errno(linux.ioctl(fd, linux.SIOCGIFINDEX, @intFromPtr(&ifr))) != .SUCCESS) + return error.IfIndexFailed; + const ifindex: u32 = @intCast(ifr.ifru.ivalue); + + const umem_len: usize = @as(usize, cfg.num_frames) * cfg.frame_size; + const rc_umem = linux.mmap(null, umem_len, .{ .READ = true, .WRITE = true }, .{ .TYPE = .PRIVATE, .ANONYMOUS = true }, -1, 0); + if (linux.errno(rc_umem) != .SUCCESS) return error.UmemAllocFailed; + const umem_ptr: [*]align(page) u8 = @ptrFromInt(rc_umem); + const umem = umem_ptr[0..umem_len]; + errdefer _ = linux.munmap(umem.ptr, umem.len); + + var reg = xdp.UmemReg{ + .addr = @intFromPtr(umem.ptr), + .len = umem_len, + .chunk_size = cfg.frame_size, + .headroom = 0, + .flags = 0, + .tx_metadata_len = 0, + }; + if (linux.errno(linux.setsockopt(fd, xdp.SOL_XDP, xdp.sockopt.UMEM_REG, std.mem.asBytes(®), @sizeOf(xdp.UmemReg))) != .SUCCESS) + return error.UmemRegFailed; + + try setU32(fd, xdp.sockopt.UMEM_FILL_RING, cfg.fill_size); + try setU32(fd, xdp.sockopt.UMEM_COMPLETION_RING, cfg.comp_size); + try setU32(fd, xdp.sockopt.TX_RING, cfg.tx_size); + + var off = std.mem.zeroes(xdp.MmapOffsets); + var off_len: linux.socklen_t = @sizeOf(xdp.MmapOffsets); + if (linux.errno(linux.getsockopt(fd, xdp.SOL_XDP, xdp.sockopt.MMAP_OFFSETS, std.mem.asBytes(&off), &off_len)) != .SUCCESS) + return error.MmapOffsetsFailed; + + const comp_len: usize = @as(usize, @intCast(off.cr.desc)) + @as(usize, cfg.comp_size) * @sizeOf(u64); + const comp_map = try mapRing(fd, comp_len, xdp.pgoff.COMPLETION_RING); + errdefer _ = linux.munmap(comp_map.ptr, comp_map.len); + + const tx_len: usize = @as(usize, @intCast(off.tx.desc)) + @as(usize, cfg.tx_size) * @sizeOf(xdp.Desc); + const tx_map = try mapRing(fd, tx_len, xdp.pgoff.TX_RING); + errdefer _ = linux.munmap(tx_map.ptr, tx_map.len); + + const frame_backing = try allocator.alloc(u64, cfg.num_frames); + errdefer allocator.free(frame_backing); + + const tx = xdp.Prod{ + .producer = ringPtr(*u32, tx_map, off.tx.producer), + .consumer = ringPtr(*u32, tx_map, off.tx.consumer), + .ring = ringPtr([*]xdp.Desc, tx_map, off.tx.desc), + .mask = cfg.tx_size - 1, + .size = cfg.tx_size, + }; + const comp = xdp.Comp{ + .producer = ringPtr(*u32, comp_map, off.cr.producer), + .consumer = ringPtr(*u32, comp_map, off.cr.consumer), + .ring = ringPtr([*]u64, comp_map, off.cr.desc), + .mask = cfg.comp_size - 1, + .size = cfg.comp_size, + }; + + const zc: u16 = if (mode == .zerocopy) xdp.bind_flags.ZEROCOPY else xdp.bind_flags.COPY; + var sxdp = linux.sockaddr.xdp{ + .flags = xdp.bind_flags.USE_NEED_WAKEUP | zc, + .ifindex = ifindex, + .queue_id = cfg.queue_id, + .shared_umem_fd = 0, + }; + switch (linux.errno(linux.bind(fd, @ptrCast(&sxdp), @sizeOf(linux.sockaddr.xdp)))) { + .SUCCESS => {}, + .PERM, .ACCES => return error.NeedCapNetRaw, + else => return error.BindFailed, + } + + var actual_mode = mode; + var xopt = xdp.Options{ .flags = 0 }; + var xopt_len: linux.socklen_t = @sizeOf(xdp.Options); + if (linux.errno(linux.getsockopt(fd, xdp.SOL_XDP, xdp.sockopt.OPTIONS, std.mem.asBytes(&xopt), &xopt_len)) == .SUCCESS) { + actual_mode = if ((xopt.flags & xdp.OPTIONS_ZEROCOPY) != 0) .zerocopy else .copy; + } + + return .{ + .allocator = allocator, + .fd = fd, + .mode = actual_mode, + .frame_size = cfg.frame_size, + .umem = umem, + .tx_map = tx_map, + .comp_map = comp_map, + .tx = tx, + .comp = comp, + .frames = xdp.FrameStack.init(frame_backing, cfg.num_frames, cfg.frame_size), + .frame_backing = frame_backing, + }; + } + + pub fn submit(self: *Backend, frame: []const u8) bool { + if (frame.len > self.frame_size) return false; + const off = self.frames.pop() orelse return false; + const slot = self.tx.reserve() orelse { + self.frames.push(off); + return false; + }; + const o: usize = @intCast(off); + @memcpy(self.umem[o..][0..frame.len], frame); + self.tx.write(slot, .{ .addr = off, .len = @intCast(frame.len), .options = 0 }); + return true; + } + + pub fn kick(self: *Backend) void { + self.tx.publish(); + var dummy: u8 = 0; + _ = linux.sendto(self.fd, @ptrCast(&dummy), 0, linux.MSG.DONTWAIT, null, 0); + const n = self.comp.peek(); + var i: u32 = 0; + while (i < n) : (i += 1) { + self.frames.push(self.comp.addrAt(i)); + } + if (n > 0) self.comp.release(n); + } + + pub fn close(self: *Backend) void { + _ = linux.munmap(self.tx_map.ptr, self.tx_map.len); + _ = linux.munmap(self.comp_map.ptr, self.comp_map.len); + _ = linux.munmap(self.umem.ptr, self.umem.len); + self.allocator.free(self.frame_backing); + _ = linux.close(self.fd); + } + + pub fn zerocopy(self: *const Backend) bool { + return self.mode == .zerocopy; + } +}; + +test "Config rejects non-power-of-two rings and illegal chunk sizes" { + const bad_ring = Backend.open(std.testing.allocator, "lo", .copy, .{ .tx_size = 1000 }); + try std.testing.expectError(error.BadConfig, bad_ring); + const bad_chunk = Backend.open(std.testing.allocator, "lo", .copy, .{ .frame_size = 1024 }); + try std.testing.expectError(error.BadConfig, bad_chunk); + const too_few = Backend.open(std.testing.allocator, "lo", .copy, .{ .num_frames = 512, .tx_size = 2048 }); + try std.testing.expectError(error.BadConfig, too_few); +} diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig b/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig index 70c28542..09f62450 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig @@ -56,6 +56,7 @@ pub fn printHelp(io: std.Io, env: *std.process.Environ.Map) !void { \\ --src-port source port; UDP uses it as the cookie-range base (default 40000) \\ --gw-mac gateway/dst MAC aa:bb:cc:dd:ee:ff (default 00:..:00) \\ --seed permutation seed (default: per-scan CSPRNG) + \\ --backend TX path: auto | xdp | afpacket (default auto; xdp needs a -Dxdp build) \\ \\scan-only options: \\ --udp UDP scan: per-protocol payloads, ICMP type3/code3 = closed, diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/packet_io.zig b/PROJECTS/advanced/zig-stateless-scanner/src/packet_io.zig new file mode 100644 index 00000000..34ab9bc3 --- /dev/null +++ b/PROJECTS/advanced/zig-stateless-scanner/src/packet_io.zig @@ -0,0 +1,112 @@ +// ©AngelaMos | 2026 +// packet_io.zig + +const std = @import("std"); +const build_config = @import("build_config"); +const afpacket = @import("afpacket"); +const afxdp = @import("afxdp"); + +pub const Kind = enum { afpacket, afxdp_copy, afxdp_zerocopy }; +pub const Choice = enum { auto, xdp, afpacket }; + +pub const Backend = union(enum) { + afpacket: afpacket.Backend, + afxdp: afxdp.Backend, + + pub fn submit(self: *Backend, frame: []const u8) bool { + return switch (self.*) { + inline else => |*b| b.submit(frame), + }; + } + + pub fn kick(self: *Backend) void { + switch (self.*) { + inline else => |*b| b.kick(), + } + } + + pub fn close(self: *Backend) void { + switch (self.*) { + inline else => |*b| b.close(), + } + } + + pub fn kind(self: *const Backend) Kind { + return switch (self.*) { + .afpacket => .afpacket, + .afxdp => |*b| if (b.zerocopy()) .afxdp_zerocopy else .afxdp_copy, + }; + } +}; + +pub const SelectError = error{XdpNotCompiledIn} || afxdp.OpenError || afpacket.OpenError; + +pub fn parseChoice(text: ?[]const u8) ?Choice { + const t = text orelse return .auto; + if (std.mem.eql(u8, t, "auto")) return .auto; + if (std.mem.eql(u8, t, "xdp")) return .xdp; + if (std.mem.eql(u8, t, "afpacket")) return .afpacket; + return null; +} + +pub fn kindLabel(k: Kind) []const u8 { + return switch (k) { + .afpacket => "AF_PACKET (PACKET_TX_RING)", + .afxdp_copy => "AF_XDP (copy / XDP_SKB mode)", + .afxdp_zerocopy => "AF_XDP (zero-copy)", + }; +} + +fn note(diag: ?*std.Io.Writer, comptime fmt: []const u8, args: anytype) void { + if (diag) |w| { + w.print(" backend: " ++ fmt ++ "\n", args) catch {}; + } +} + +pub fn select( + allocator: std.mem.Allocator, + ifname: []const u8, + choice: Choice, + xdp_cfg: afxdp.Config, + afp_cfg: afpacket.RingConfig, + diag: ?*std.Io.Writer, +) SelectError!Backend { + if (choice == .xdp and !build_config.xdp) return error.XdpNotCompiledIn; + + const want_xdp = build_config.xdp and choice != .afpacket; + if (want_xdp) { + if (afxdp.Backend.open(allocator, ifname, .zerocopy, xdp_cfg)) |b| { + return .{ .afxdp = b }; + } else |err| { + note(diag, "AF_XDP zero-copy unavailable ({s})", .{@errorName(err)}); + } + if (afxdp.Backend.open(allocator, ifname, .copy, xdp_cfg)) |b| { + return .{ .afxdp = b }; + } else |err| { + if (choice == .xdp) return err; + note(diag, "AF_XDP copy mode unavailable ({s}); using AF_PACKET", .{@errorName(err)}); + } + } + return .{ .afpacket = try afpacket.Backend.open(ifname, afp_cfg) }; +} + +test "parseChoice maps flag text, defaults to auto, and rejects unknown values" { + try std.testing.expectEqual(Choice.auto, parseChoice(null).?); + try std.testing.expectEqual(Choice.auto, parseChoice("auto").?); + try std.testing.expectEqual(Choice.xdp, parseChoice("xdp").?); + try std.testing.expectEqual(Choice.afpacket, parseChoice("afpacket").?); + try std.testing.expect(parseChoice("nonsense") == null); +} + +test "forcing --backend xdp without a -Dxdp build is rejected up front" { + if (!build_config.xdp) { + const r = select(std.testing.allocator, "lo", .xdp, .{}, .{}, null); + try std.testing.expectError(error.XdpNotCompiledIn, r); + } +} + +test "kindLabel covers every backend kind" { + try std.testing.expect(kindLabel(.afpacket).len > 0); + try std.testing.expect(kindLabel(.afxdp_copy).len > 0); + try std.testing.expect(kindLabel(.afxdp_zerocopy).len > 0); +} diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig b/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig index 3d9bbb91..a37d9a26 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig @@ -6,7 +6,7 @@ const targets = @import("targets"); const template = @import("template"); const udp = @import("udp"); const ratelimit = @import("ratelimit"); -const afpacket = @import("afpacket"); +const packet_io = @import("packet_io"); const cookie = @import("cookie"); const tx = @import("tx"); const rx = @import("rx"); @@ -40,7 +40,7 @@ const concurrency_hint = "scan: this system cannot launch concurrent TX/RX (needs >= 2 worker threads).\n"; const TxSink = struct { - backend: *afpacket.Backend, + backend: *packet_io.Backend, sent: *output.Counter, pub fn submit(self: *TxSink, frame: []const u8) bool { @@ -151,6 +151,11 @@ pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8, e const wait_ms = if (netutil.getFlag(args, "--wait")) |w| try std.fmt.parseInt(i32, w, 10) else default_wait_ms; const json = netutil.hasFlag(args, "--json"); const is_udp = netutil.hasFlag(args, "--udp"); + const backend_choice = packet_io.parseChoice(netutil.getFlag(args, "--backend")) orelse { + try derr.writeAll("scan: --backend must be one of auto, xdp, afpacket\n"); + try derr.flush(); + return; + }; const udp_base: u16 = src_port; const udp_span: u16 = @intCast(@min(@as(u32, default_udp_src_span), 65536 - @as(u32, udp_base))); const proto_json: []const u8 = if (is_udp) "udp" else "tcp"; @@ -206,15 +211,21 @@ pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8, e }); var bucket = ratelimit.TokenBucket.init(rate, rate); - var backend = afpacket.Backend.open(ifname, .{}) catch |err| switch (err) { + var backend = packet_io.select(allocator, ifname, backend_choice, .{}, .{}, derr) catch |err| switch (err) { error.NeedCapNetRaw => { try derr.writeAll(need_cap_hint); try derr.flush(); return; }, + error.XdpNotCompiledIn => { + try derr.writeAll("scan: --backend xdp needs a build with -Dxdp\n"); + try derr.flush(); + return; + }, else => return err, }; defer backend.close(); + try derr.print(" using {s}\n", .{packet_io.kindLabel(backend.kind())}); var tx_done = std.atomic.Value(bool).init(false); var rx_done = std.atomic.Value(bool).init(false); diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig b/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig index e22f5f30..786dcb4c 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig @@ -5,7 +5,7 @@ const std = @import("std"); const targets = @import("targets"); const template = @import("template"); const ratelimit = @import("ratelimit"); -const afpacket = @import("afpacket"); +const packet_io = @import("packet_io"); const cookie = @import("cookie"); const tx = @import("tx"); const netutil = @import("netutil"); @@ -69,15 +69,26 @@ pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8) ! }); var bucket = ratelimit.TokenBucket.init(rate, rate); - var backend = afpacket.Backend.open(ifname, .{}) catch |err| switch (err) { + const backend_choice = packet_io.parseChoice(getFlag(args, "--backend")) orelse { + try out.writeAll("tx: --backend must be one of auto, xdp, afpacket\n"); + try out.flush(); + return; + }; + var backend = packet_io.select(allocator, ifname, backend_choice, .{}, .{}, out) catch |err| switch (err) { error.NeedCapNetRaw => { try out.writeAll("tx: need CAP_NET_RAW + CAP_NET_ADMIN. Grant once, then re-run (no sudo):\n sudo setcap cap_net_raw,cap_net_admin=eip ./zig-out/bin/zingela\nSkipping.\n"); try out.flush(); return; }, + error.XdpNotCompiledIn => { + try out.writeAll("tx: --backend xdp needs a build with -Dxdp\n"); + try out.flush(); + return; + }, else => return err, }; defer backend.close(); + try out.print("tx: using {s}\n", .{packet_io.kindLabel(backend.kind())}); var clock = RealClock{}; const t0 = clock.now(); diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/xdp.zig b/PROJECTS/advanced/zig-stateless-scanner/src/xdp.zig new file mode 100644 index 00000000..29a0c5a8 --- /dev/null +++ b/PROJECTS/advanced/zig-stateless-scanner/src/xdp.zig @@ -0,0 +1,296 @@ +// ©AngelaMos | 2026 +// xdp.zig + +const std = @import("std"); +const linux = std.os.linux; + +pub const AF_XDP: u32 = linux.AF.XDP; +pub const SOL_XDP: i32 = linux.SOL.XDP; + +pub const bind_flags = struct { + pub const SHARED_UMEM: u16 = 1 << 0; + pub const COPY: u16 = 1 << 1; + pub const ZEROCOPY: u16 = 1 << 2; + pub const USE_NEED_WAKEUP: u16 = 1 << 3; +}; + +pub const umem_flags = struct { + pub const UNALIGNED_CHUNK: u32 = 1 << 0; +}; + +pub const sockopt = struct { + pub const MMAP_OFFSETS: u32 = 1; + pub const RX_RING: u32 = 2; + pub const TX_RING: u32 = 3; + pub const UMEM_REG: u32 = 4; + pub const UMEM_FILL_RING: u32 = 5; + pub const UMEM_COMPLETION_RING: u32 = 6; + pub const STATISTICS: u32 = 7; + pub const OPTIONS: u32 = 8; +}; + +pub const RING_NEED_WAKEUP: u32 = 1 << 0; +pub const OPTIONS_ZEROCOPY: u32 = 1 << 0; + +pub const pgoff = struct { + pub const RX_RING: u64 = 0; + pub const TX_RING: u64 = 0x80000000; + pub const FILL_RING: u64 = 0x100000000; + pub const COMPLETION_RING: u64 = 0x180000000; +}; + +comptime { + std.debug.assert(bind_flags.COPY == linux.XDP.COPY); + std.debug.assert(bind_flags.ZEROCOPY == linux.XDP.ZEROCOPY); + std.debug.assert(bind_flags.USE_NEED_WAKEUP == linux.XDP.USE_NEED_WAKEUP); + std.debug.assert(sockopt.MMAP_OFFSETS == linux.XDP.MMAP_OFFSETS); + std.debug.assert(sockopt.TX_RING == linux.XDP.TX_RING); + std.debug.assert(sockopt.UMEM_REG == linux.XDP.UMEM_REG); + std.debug.assert(sockopt.UMEM_FILL_RING == linux.XDP.UMEM_FILL_RING); + std.debug.assert(sockopt.UMEM_COMPLETION_RING == linux.XDP.UMEM_COMPLETION_RING); + std.debug.assert(sockopt.OPTIONS == linux.XDP.OPTIONS); + std.debug.assert(OPTIONS_ZEROCOPY == linux.XDP.OPTIONS_ZEROCOPY); + std.debug.assert(pgoff.TX_RING == linux.XDP.PGOFF_TX_RING); + std.debug.assert(pgoff.FILL_RING == linux.XDP.UMEM_PGOFF_FILL_RING); + std.debug.assert(pgoff.COMPLETION_RING == linux.XDP.UMEM_PGOFF_COMPLETION_RING); +} + +pub const UmemReg = extern struct { + addr: u64, + len: u64, + chunk_size: u32, + headroom: u32, + flags: u32, + tx_metadata_len: u32, +}; + +pub const RingOffset = extern struct { + producer: u64, + consumer: u64, + desc: u64, + flags: u64, +}; + +pub const MmapOffsets = extern struct { + rx: RingOffset, + tx: RingOffset, + fr: RingOffset, + cr: RingOffset, +}; + +pub const Desc = extern struct { + addr: u64, + len: u32, + options: u32, +}; + +pub const Options = extern struct { + flags: u32, +}; + +comptime { + std.debug.assert(@sizeOf(UmemReg) == 32); + std.debug.assert(@sizeOf(RingOffset) == 32); + std.debug.assert(@sizeOf(MmapOffsets) == 128); + std.debug.assert(@sizeOf(Desc) == 16); + std.debug.assert(@sizeOf(Options) == 4); + std.debug.assert(@sizeOf(linux.sockaddr.xdp) == 16); +} + +pub const Prod = struct { + producer: *u32, + consumer: *u32, + ring: [*]Desc, + mask: u32, + size: u32, + cached_prod: u32 = 0, + cached_cons: u32 = 0, + + pub fn reserve(self: *Prod) ?u32 { + if (self.cached_prod -% self.cached_cons == self.size) { + self.cached_cons = @atomicLoad(u32, self.consumer, .acquire); + if (self.cached_prod -% self.cached_cons == self.size) return null; + } + const slot = self.cached_prod & self.mask; + self.cached_prod +%= 1; + return slot; + } + + pub fn write(self: *Prod, slot: u32, desc: Desc) void { + self.ring[slot] = desc; + } + + pub fn publish(self: *Prod) void { + @atomicStore(u32, self.producer, self.cached_prod, .release); + } +}; + +pub const Comp = struct { + producer: *u32, + consumer: *u32, + ring: [*]u64, + mask: u32, + size: u32, + cached_prod: u32 = 0, + cached_cons: u32 = 0, + + pub fn peek(self: *Comp) u32 { + var avail = self.cached_prod -% self.cached_cons; + if (avail == 0) { + self.cached_prod = @atomicLoad(u32, self.producer, .acquire); + avail = self.cached_prod -% self.cached_cons; + } + return avail; + } + + pub fn addrAt(self: *const Comp, i: u32) u64 { + return self.ring[(self.cached_cons +% i) & self.mask]; + } + + pub fn release(self: *Comp, n: u32) void { + self.cached_cons +%= n; + @atomicStore(u32, self.consumer, self.cached_cons, .release); + } +}; + +pub const FrameStack = struct { + free: []u64, + top: usize, + + pub fn init(buf: []u64, num_frames: u32, frame_size: u32) FrameStack { + var i: u32 = 0; + while (i < num_frames) : (i += 1) { + buf[i] = @as(u64, i) * frame_size; + } + return .{ .free = buf, .top = num_frames }; + } + + pub fn pop(self: *FrameStack) ?u64 { + if (self.top == 0) return null; + self.top -= 1; + return self.free[self.top]; + } + + pub fn push(self: *FrameStack, addr: u64) void { + std.debug.assert(self.top < self.free.len); + self.free[self.top] = addr; + self.top += 1; + } + + pub fn available(self: *const FrameStack) usize { + return self.top; + } +}; + +test "FrameStack lays out frame offsets, pops LIFO, and reports empty" { + var buf: [4]u64 = undefined; + var fs = FrameStack.init(&buf, 4, 2048); + try std.testing.expectEqual(@as(usize, 4), fs.available()); + try std.testing.expectEqual(@as(u64, 6144), fs.pop().?); + try std.testing.expectEqual(@as(u64, 4096), fs.pop().?); + fs.push(4096); + try std.testing.expectEqual(@as(u64, 4096), fs.pop().?); + try std.testing.expectEqual(@as(u64, 2048), fs.pop().?); + try std.testing.expectEqual(@as(u64, 0), fs.pop().?); + try std.testing.expect(fs.pop() == null); +} + +test "Prod.reserve backpressures at ring size and recovers as the kernel consumes" { + const size: u32 = 4; + var producer: u32 = 0; + var consumer: u32 = 0; + var ring: [size]Desc = undefined; + var p = Prod{ .producer = &producer, .consumer = &consumer, .ring = &ring, .mask = size - 1, .size = size }; + + var got: u32 = 0; + while (p.reserve()) |slot| { + p.write(slot, .{ .addr = got, .len = 1, .options = 0 }); + got += 1; + } + try std.testing.expectEqual(size, got); + try std.testing.expect(p.reserve() == null); + + p.publish(); + try std.testing.expectEqual(size, @atomicLoad(u32, &producer, .acquire)); + + @atomicStore(u32, &consumer, 2, .release); + try std.testing.expect(p.reserve() != null); + try std.testing.expect(p.reserve() != null); + try std.testing.expect(p.reserve() == null); +} + +test "Comp.peek sees kernel completions and release advances the consumer" { + const size: u32 = 4; + var producer: u32 = 0; + var consumer: u32 = 0; + var ring: [size]u64 = undefined; + var c = Comp{ .producer = &producer, .consumer = &consumer, .ring = &ring, .mask = size - 1, .size = size }; + + try std.testing.expectEqual(@as(u32, 0), c.peek()); + + ring[0] = 0; + ring[1] = 2048; + ring[2] = 4096; + @atomicStore(u32, &producer, 3, .release); + + try std.testing.expectEqual(@as(u32, 3), c.peek()); + try std.testing.expectEqual(@as(u64, 0), c.addrAt(0)); + try std.testing.expectEqual(@as(u64, 2048), c.addrAt(1)); + try std.testing.expectEqual(@as(u64, 4096), c.addrAt(2)); + c.release(3); + try std.testing.expectEqual(@as(u32, 3), @atomicLoad(u32, &consumer, .acquire)); + try std.testing.expectEqual(@as(u32, 0), c.peek()); +} + +test "TX and completion rings recycle a bounded UMEM frame pool across kicks" { + const size: u32 = 8; + const num_frames: u32 = 8; + const frame_size: u32 = 2048; + + var tx_prod: u32 = 0; + var tx_cons: u32 = 0; + var tx_ring: [size]Desc = undefined; + var cq_prod: u32 = 0; + var cq_cons: u32 = 0; + var cq_ring: [size]u64 = undefined; + + var tx = Prod{ .producer = &tx_prod, .consumer = &tx_cons, .ring = &tx_ring, .mask = size - 1, .size = size }; + var cq = Comp{ .producer = &cq_prod, .consumer = &cq_cons, .ring = &cq_ring, .mask = size - 1, .size = size }; + + var fb: [num_frames]u64 = undefined; + var frames = FrameStack.init(&fb, num_frames, frame_size); + + var total_sent: u32 = 0; + var round: u32 = 0; + while (round < 4) : (round += 1) { + var i: u32 = 0; + while (i < size) : (i += 1) { + const off = frames.pop() orelse break; + const slot = tx.reserve() orelse { + frames.push(off); + break; + }; + tx.write(slot, .{ .addr = off, .len = 54, .options = 0 }); + total_sent += 1; + } + tx.publish(); + + const published = @atomicLoad(u32, &tx_prod, .acquire); + var kc = @atomicLoad(u32, &tx_cons, .monotonic); + var kp = @atomicLoad(u32, &cq_prod, .monotonic); + while (kc != published) : (kc +%= 1) { + cq_ring[kc & (size - 1)] = tx_ring[kc & (size - 1)].addr; + kp +%= 1; + } + @atomicStore(u32, &tx_cons, kc, .release); + @atomicStore(u32, &cq_prod, kp, .release); + + const n = cq.peek(); + var k: u32 = 0; + while (k < n) : (k += 1) frames.push(cq.addrAt(k)); + if (n > 0) cq.release(n); + } + + try std.testing.expect(total_sent > num_frames); + try std.testing.expectEqual(@as(usize, num_frames), frames.available()); +}