diff --git a/CHANGELOG.md b/CHANGELOG.md index 962e2b6..6c7218c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,10 @@ # Changelog ## Unreleased +- Pin coapz v0.4.0 with frontend-inline option decoding and public Observe values +- Use cooperative io_uring task work, a single pending tick timer and a registered socket +- Return receive buffers through a shared descriptor ring; account its locked-memory overhead +- Index completed CON exchanges by address hash without changing eviction/cache policy - Add configurable standalone `bench-server` build target and trusted-network remote benchmark protocol - Account unique remote NON receipts/loss, isolate run leases, and load DTLS PSKs from bounded files - Add per-client multi-thread CON windows; fix plain benchmark MID-wrap/duplicate accounting and startup errors diff --git a/README.md b/README.md index c91a8f5..4c5e486 100644 --- a/README.md +++ b/README.md @@ -674,7 +674,13 @@ processed, but during bursts the pool must absorb all arrivals between processing cycles. Set this to at least 2x your expected concurrent clients' send window. Default: `512`. -Higher values require more kernel memory per io_uring instance. +Higher values require more kernel memory per io_uring instance. Buffers are +returned through a shared provided-buffer ring, without a return-buffer SQE/CQE. +The payload pool is not pinned; the descriptor ring adds page-rounded locked +memory, included in the automatic memlock budget estimate. + +See [server I/O implementation notes](docs/IO_OPTIMIZATIONS.md) for ownership, +locked-memory costs and performance limitations. ### `buffer_size` @@ -955,7 +961,7 @@ See [remote benchmark setup and accounting](docs/REMOTE_BENCHMARKS.md). ## Requirements -- Linux (io_uring support, kernel 5.13+ for multishot recvmsg) +- Linux kernel 6.0+ (multishot recvmsg; shared provided-buffer rings and cooperative task running) - Zig 0.16.0 (the currently supported toolchain) ### Migrating from Zig 0.15 diff --git a/build.zig.zon b/build.zig.zon index 40a7e25..f7c427a 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -5,8 +5,8 @@ .fingerprint = 0x67bf20367bf50f8b, .dependencies = .{ .coapz = .{ - .url = "git+https://github.com/cvik/coapz#v0.3.0", - .hash = "coapz-0.3.0-kDItyJkUAQDoQvLuHUX2cOWPIygn2rBDEa8MskzHnCKr", + .url = "git+https://github.com/cvik/coapz#v0.4.0", + .hash = "coapz-0.4.0-kDItyCgdAQBe-JIBtd-Clos6G21bs9MirsRRpswizQ69", }, }, .paths = .{ diff --git a/docs/IO_OPTIMIZATIONS.md b/docs/IO_OPTIMIZATIONS.md new file mode 100644 index 0000000..90d3a56 --- /dev/null +++ b/docs/IO_OPTIMIZATIONS.md @@ -0,0 +1,57 @@ +# Server I/O optimizations + +This change ports the measured five-part native io_uring prototype without +changing CoAP response-cache policy or the public server API. A follow-up commit +pins the independently released coapz v0.4.0 parser optimization. + +- `COOP_TASKRUN` reduces cross-CPU task-work notifications. No `SINGLE_ISSUER`, + `DEFER_TASKRUN`, or SQPOLL restriction is introduced; existing thread-handoff + fixtures remain supported. +- One pending 50-ms tick timeout preserves idle shutdown/maintenance wakeups + without queuing a fresh timeout for every busy tick. Timeout CQEs clear the + pending state; submission errors propagate instead of being silently ignored. +- A shared provided-buffer descriptor ring returns buffers without a return SQE + or CQE. Incremental consumption is disabled: one datagram consumes one buffer. + Plaintext packets are decoded into the existing arena, and DTLS datagrams are + copied, before returning their receive buffer. Close io_uring before unmapping + the descriptor ring; never unregister through a potentially reused fd. +- The bound socket occupies registered-file slot zero for receive and send. + The ordinary descriptor remains owned until teardown; closing the ring releases + its registered reference. This was a marginal performance improvement, retained + as part of the explicitly selected full chain. +- Completed exchanges have preallocated address-hash buckets and intrusive links. + Peer eviction visits its bucket instead of scanning every slot. All removal + paths unlink both indexes. Matching still uses the existing full `addr_key`; + hash grouping, rejection, expiry, cached responses and retransmission behavior + are unchanged. This is not a new RFC exchange-lifetime guarantee. + +The descriptor ring adds page-rounded locked memory, accounted for in the +existing approximate memlock clamp. Payload buffers are not pinned. Peer indexing +adds two links per slot (using existing padding on the measured x86_64 target), +a slice header and `2 * table_size` bytes of bucket heads: 1,024 bytes at the +default 256 exchanges. Init allocates this storage; peer eviction does not allocate. + +## Evidence and limits + +Independent, alternating native loopback experiments used Zig 0.16.0, Linux +7.1.11, Ryzen AI MAX+ 395, one server thread, 100-byte payloads and W8/client. +The five parts were evaluated incrementally: COOP improved a single-client CON +case about 5.5%; one timer was about 1%; provided buffers improved four-client +CON about 7.9%; registered socket about 1%; peer indexing about 6.4%. These are +**different matched comparisons, not additive gains or a universal speedup**. +Four clients are not server-worker scaling. No libcoap performance ranking is +claimed here. Raw exploratory archives remain separate from this focused PR. + +Coverage includes repeated idle timer wakeups, held/exhausted/resumed receive +buffers, no SQE publication on buffer return, registered-socket operation after +closing the original descriptor, allocation-failure cleanup, index collision/ +unlink/expiry/reuse and 5,000 operations against a scan model. Existing benchmark +integration covers IPv4/IPv6, native UDP/DTLS, two server workers and client windows. + +Linux 6.0+ is required by the existing multishot recvmsg path. Shared descriptor +rings and COOP are supported within that floor. Older kernels, overload/lifecycle +patterns and different CPUs still warrant workload-specific validation. No empty +submission skipping or CQE-clearing removal is included. The I/O measurements +above used the original coapz v0.3.0 dependency. The subsequent v0.4.0 pin also +includes its independently released inline-helper fix; parser/end-to-end gains +are not attributed to the five I/O changes. diff --git a/src/Io.zig b/src/Io.zig index 925c239..abc3acf 100644 --- a/src/Io.zig +++ b/src/Io.zig @@ -30,6 +30,8 @@ iovecs: []posix.iovec, buffer_count: u16, buffer_size: u32, fd_socket: ?posix.socket_t, +buffer_ring: *align(std.heap.page_size_min) linux.io_uring_buf_ring, +buffer_ring_entries: u16, pub fn init( allocator: std.mem.Allocator, @@ -38,12 +40,10 @@ pub fn init( ) !Io { if (buffer_count == 0 or buffer_size == 0) return error.InvalidConfig; - // Ring must accommodate send + release_buffer per recv, plus overhead. - // Each received packet may generate up to 3 SQEs: release_buffer, - // send_msg response, and periodic provide_buffers. The 4x multiplier - // covers this worst case with headroom for the multishot recv SQE. + // Retain the existing SQ/CQ sizing for sends, receive bursts and timers. + // Buffer returns now publish descriptors directly, without an SQE/CQE. const ring_entries: u16 = buffer_count *| 4; - var ring = try linux.IoUring.init(ring_entries, 0); + var ring = try linux.IoUring.init(ring_entries, linux.IORING_SETUP_COOP_TASKRUN); errdefer ring.deinit(); const total = @as(usize, buffer_count) * buffer_size; @@ -53,8 +53,20 @@ pub fn init( const iovecs = try allocator.alloc(posix.iovec, buffer_count); errdefer allocator.free(iovecs); + const entries = std.math.ceilPowerOfTwo(u16, buffer_count) catch return error.InvalidConfig; + const br = try linux.IoUring.setup_buf_ring(ring.fd, entries, constants.buffer_group_id, .{ .inc = false }); + linux.IoUring.buf_ring_init(br); + const mask = linux.IoUring.buf_ring_mask(entries); + for (0..buffer_count) |i| { + const offset = i * buffer_size; + linux.IoUring.buf_ring_add(br, buffers[offset..][0..buffer_size], @intCast(i), mask, @intCast(i)); + } + linux.IoUring.buf_ring_advance(br, buffer_count); + return .{ .ring = ring, + .buffer_ring = br, + .buffer_ring_entries = entries, .buffers = buffers, .iovecs = iovecs, .buffer_count = buffer_count, @@ -64,7 +76,11 @@ pub fn init( } pub fn deinit(io: *Io, allocator: std.mem.Allocator) void { + // Closing the ring first cancels its receive and releases kernel access. + // Do not unregister via a potentially reused fd after closing it. io.ring.deinit(); + const ring_bytes: [*]align(std.heap.page_size_min) u8 = @ptrCast(io.buffer_ring); + std.posix.munmap(ring_bytes[0 .. @as(usize, io.buffer_ring_entries) * @sizeOf(linux.io_uring_buf)]); allocator.free(io.buffers); allocator.free(io.iovecs); if (io.fd_socket) |fd| { @@ -72,7 +88,7 @@ pub fn deinit(io: *Io, allocator: std.mem.Allocator) void { } } -/// Bind a UDP socket and provide buffers to the kernel pool. +/// Bind and register a UDP socket. Buffers were published during init. pub fn setup(io: *Io, port: u16, bind_address: []const u8) !void { const address = try platform.parseIp(bind_address, port); const family = platform.family(address); @@ -108,24 +124,8 @@ pub fn setup(io: *Io, port: u16, bind_address: []const u8) !void { }; } - // Provide initial buffer pool to the kernel. - // Note: register_buffers is intentionally not called — the server - // uses provided buffers (IOSQE_BUFFER_SELECT) not fixed buffer I/O, - // and register_buffers pins pages against RLIMIT_MEMLOCK which limits - // the number of io_uring instances that can run concurrently. - _ = try io.ring.provide_buffers( - @intFromEnum(UserData.provide_buffers), - @ptrCast(io.buffers.ptr), - io.buffer_size, - io.buffer_count, - constants.buffer_group_id, - 0, - ); - _ = try io.ring.submit(); - const cqe = try io.ring.copy_cqe(); - if (cqe.err() != .SUCCESS) { - return error.BufferSetupFailed; - } + try io.ring.register_files(&.{fd}); + // The initial buffers were published in the shared ring during init. } /// Queue a multishot recvmsg operation. @@ -133,10 +133,10 @@ pub fn recv_multishot( io: *Io, msg_header: *linux.msghdr, ) !void { - const fd = io.fd_socket orelse return error.SocketNotReady; + _ = io.fd_socket orelse return error.SocketNotReady; const sqe = try io.ring.get_sqe(); - sqe.prep_recvmsg_multishot(fd, msg_header, MSG_TRUNC); - sqe.flags |= linux.IOSQE_BUFFER_SELECT; + sqe.prep_recvmsg_multishot(0, msg_header, MSG_TRUNC); + sqe.flags |= linux.IOSQE_BUFFER_SELECT | linux.IOSQE_FIXED_FILE; sqe.user_data = @intFromEnum(UserData.recv_msg); sqe.buf_index = constants.buffer_group_id; } @@ -146,9 +146,10 @@ pub fn send_msg( io: *Io, msg_header: *linux.msghdr_const, ) !void { - const fd = io.fd_socket orelse return error.SocketNotReady; + _ = io.fd_socket orelse return error.SocketNotReady; const sqe = try io.ring.get_sqe(); - sqe.prep_sendmsg(fd, msg_header, 0); + sqe.prep_sendmsg(0, msg_header, 0); + sqe.flags |= linux.IOSQE_FIXED_FILE; sqe.user_data = @intFromEnum(UserData.send_msg); } @@ -166,18 +167,19 @@ pub fn submit(io: *Io) !u32 { return io.ring.submit(); } -/// Return a provided buffer to the kernel pool. +/// Return a provided buffer once all references into it have been consumed/copied. +/// A single producer publishes one whole datagram buffer; incremental mode is off. pub fn release_buffer(io: *Io, buffer_id: u16) !void { if (buffer_id >= io.buffer_count) return error.InvalidBufferId; const offset = @as(usize, buffer_id) * io.buffer_size; - _ = try io.ring.provide_buffers( - @intFromEnum(UserData.provide_buffers), - @ptrCast(io.buffers.ptr + offset), - io.buffer_size, - 1, - constants.buffer_group_id, + linux.IoUring.buf_ring_add( + io.buffer_ring, + io.buffers[offset..][0..io.buffer_size], buffer_id, + linux.IoUring.buf_ring_mask(io.buffer_ring_entries), + 0, ); + linux.IoUring.buf_ring_advance(io.buffer_ring, 1); } /// Extract the peer address and payload from a recv CQE. @@ -250,6 +252,15 @@ pub fn is_success(cqe: *const Cqe) bool { const testing = std.testing; +fn checkInitAllocation(allocator: std.mem.Allocator) !void { + var io = try Io.init(allocator, 4, 256); + defer io.deinit(allocator); +} + +test "provided ring init cleans up after allocation failures" { + try testing.checkAllAllocationFailures(testing.allocator, checkInitAllocation, .{}); +} + test "SO_REUSEPORT allows two instances on same port" { const port: u16 = 19691; const allocator = testing.allocator; diff --git a/src/Server.zig b/src/Server.zig index aa813e0..9de3354 100644 --- a/src/Server.zig +++ b/src/Server.zig @@ -161,6 +161,7 @@ buffers_outstanding: u16, buffers_peak: u16, rate_limiter: ?RateLimiter, tick_now_ns: i64, +tick_timeout_pending: bool = false, /// Pre-allocated RST buffers for rate-limited/shed CON packets. rate_limit_rst: []u8, @@ -216,8 +217,9 @@ pub fn initContext( } /// Clamp buffer_count so that all io_uring instances fit within -/// RLIMIT_MEMLOCK. Only the ring structures are locked (provided buffers -/// are not pinned). Each ring locks approximately: +/// RLIMIT_MEMLOCK. The payload pool is not pinned. Ring structures and +/// the page-rounded provided-buffer descriptor ring are accounted for. +/// Each io_uring locks approximately: /// SQ: ring_entries × 64 bytes (io_uring_sqe) /// CQ: ring_entries × 2 × 16 bytes (io_uring_cqe) /// ring_entries = next_pow2(buffer_count × 4) @@ -235,7 +237,13 @@ fn clampBufferCount(requested: u16, thread_count: u16) u16 { while (candidate > 4) { const ring_entries = std.math.ceilPowerOfTwo(u32, @as(u32, candidate) *| 4) catch break; // SQE: 64 bytes each, CQE: 16 bytes each (2× ring_entries). - const locked: u64 = @as(u64, ring_entries) * 96; + const buffer_entries = std.math.ceilPowerOfTwo(u32, candidate) catch break; + const descriptor_bytes = std.mem.alignForward( + u64, + @as(u64, buffer_entries) * @sizeOf(linux.io_uring_buf), + std.heap.pageSize(), + ); + const locked: u64 = @as(u64, ring_entries) * 96 + descriptor_bytes; if (locked <= per_thread) break; candidate /= 2; } @@ -712,8 +720,11 @@ pub fn tick(server: *Server) !void { // Submit a timeout so wait_cqes unblocks periodically, allowing // the run loop to check the shutdown flag even when idle. const tick_ts = linux.kernel_timespec{ .sec = 0, .nsec = 50 * std.time.ns_per_ms }; - server.io.queue_timeout(&tick_ts) catch {}; - _ = server.io.submit() catch {}; + if (!server.tick_timeout_pending) { + try server.io.queue_timeout(&tick_ts); + server.tick_timeout_pending = true; + } + _ = try server.io.submit(); const count = try server.io.wait_cqes(cqes[0..], 1); var recv_failed = false; @@ -721,7 +732,10 @@ pub fn tick(server: *Server) !void { var processed: u32 = 0; for (cqes[0..count], 0..) |cqe, index| { - if (Io.is_timeout(&cqe)) continue; + if (Io.is_timeout(&cqe)) { + server.tick_timeout_pending = false; + continue; + } if (Io.is_recv(&cqe) and !Io.is_success(&cqe)) { recv_failed = true; recv_fail_count += 1; @@ -2635,6 +2649,112 @@ test "init and deinit with null handler" { server.deinit(); } +test "registered socket remains usable after original descriptor closes" { + var server = try Server.init(testing.allocator, .{ + .port = 19794, + .buffer_count = 8, + .buffer_size = 1280, + }, echo_handler); + defer server.deinit(); + try setup_for_test(&server); + posix.close(server.io.fd_socket.?); + defer server.io.fd_socket = null; // Avoid closing a reused numeric fd. + const fd = try test_client(19794); + defer posix.close(fd); + const packet = coapz.Packet{ + .kind = .confirmable, + .code = .get, + .msg_id = 0x4242, + .token = &.{ 1, 2 }, + .options = &.{}, + .payload = "registered", + .data_buf = &.{}, + }; + const wire = try packet.write(testing.allocator); + defer testing.allocator.free(wire); + const raw = try send_tick_recv(&server, fd, wire); + defer testing.allocator.free(raw); + const response = try coapz.Packet.read(testing.allocator, raw); + defer response.deinit(testing.allocator); + try testing.expectEqual(.acknowledgement, response.kind); + try testing.expectEqual(.content, response.code); + try testing.expectEqual(packet.msg_id, response.msg_id); + try testing.expectEqualSlices(u8, packet.token, response.token); + try testing.expectEqualStrings("registered", response.payload); +} + +test "provided ring holds buffers, exhausts, and resumes after release" { + var server = try Server.init(testing.allocator, .{ + .port = 19793, + .buffer_count = 4, + .buffer_size = 256, + }, echo_handler); + defer server.deinit(); + try setup_for_test(&server); + const fd = try test_client(19793); + defer posix.close(fd); + const messages = [_][]const u8{ "one", "two", "three", "four", "five" }; + for (messages) |message| _ = try posix.send(fd, message, 0); + var ids: [4]u16 = undefined; + var held: usize = 0; + var exhausted = false; + while (!exhausted) { + var cqes: [8]Cqe = undefined; + const count = try server.io.wait_cqes(&cqes, 1); + for (cqes[0..count]) |cqe| { + if (!Io.is_recv(&cqe)) continue; + if (cqe.err() == .NOBUFS) { + exhausted = true; + continue; + } + try testing.expect(Io.is_success(&cqe)); + try testing.expect(held < ids.len); + const recv = try server.io.decode_recv(&cqe); + try testing.expectEqualStrings(messages[held], recv.payload); + ids[held] = recv.buffer_id; + held += 1; + } + } + try testing.expectEqual(ids.len, held); + for (ids, 0..) |id, i| { + for (ids[0..i]) |previous| try testing.expect(id != previous); + } + const sq_tail = server.io.ring.sq.sqe_tail; + for (ids) |id| try server.io.release_buffer(id); + try testing.expectEqual(sq_tail, server.io.ring.sq.sqe_tail); + try server.io.recv_multishot(&server.msg_recv); + _ = try server.io.submit(); + var cqes: [8]Cqe = undefined; + const count = try server.io.wait_cqes(&cqes, 1); + var resumed = false; + for (cqes[0..count]) |cqe| { + if (!Io.is_recv(&cqe)) continue; + try testing.expect(Io.is_success(&cqe)); + const recv = try server.io.decode_recv(&cqe); + try testing.expectEqualStrings("five", recv.payload); + try server.io.release_buffer(recv.buffer_id); + resumed = true; + } + try testing.expect(resumed); +} + +test "single pending tick timer repeatedly wakes an idle server" { + var server = try Server.init(testing.allocator, .{ + .port = 19791, + .buffer_count = 8, + .buffer_size = 1280, + }, echo_handler); + defer server.deinit(); + try setup_for_test(&server); + const start = platform.nanoTimestamp(); + try server.tick(); + try testing.expect(!server.tick_timeout_pending); + try server.tick(); + try testing.expect(!server.tick_timeout_pending); + const elapsed = platform.nanoTimestamp() - start; + try testing.expect(elapsed >= 80 * std.time.ns_per_ms); +} + test "round-trip: NON echo via UDP" { const port: u16 = 19683; diff --git a/src/exchange.zig b/src/exchange.zig index 81ec947..11df6ab 100644 --- a/src/exchange.zig +++ b/src/exchange.zig @@ -33,6 +33,8 @@ pub const Slot = struct { completed_at_ns: i64, /// Index of next free slot (when state == .free). next_free: u16, + peer_prev: u16, + peer_next: u16, }; slots: []Slot, @@ -44,6 +46,8 @@ response_buffer: []u8, /// 0xFFFF = empty sentinel. table: []u16, table_mask: u16, +/// Address-hash bucket heads; slots form intrusive doubly linked lists. +peer_heads: []u16, /// Head of intrusive free list through slots. free_head: u16, /// Count of active (non-free) exchanges. @@ -75,6 +79,9 @@ pub fn init(allocator: std.mem.Allocator, config: Config) !Exchange { const table = try allocator.alloc(u16, table_size); errdefer allocator.free(table); + const peer_heads = try allocator.alloc(u16, table_size); + errdefer allocator.free(peer_heads); + @memset(peer_heads, empty_sentinel); // Initialize free list and table. for (slots, 0..) |*slot, i| { @@ -85,6 +92,8 @@ pub fn init(allocator: std.mem.Allocator, config: Config) !Exchange { .message_id = 0, .response_length = 0, .completed_at_ns = 0, + .peer_prev = empty_sentinel, + .peer_next = empty_sentinel, .next_free = if (i + 1 < config.exchange_count) @intCast(i + 1) else @@ -99,6 +108,7 @@ pub fn init(allocator: std.mem.Allocator, config: Config) !Exchange { .response_buffer = response_buffer, .table = table, .table_mask = table_size - 1, + .peer_heads = peer_heads, .free_head = 0, .count_active = 0, .config = config, @@ -109,6 +119,7 @@ pub fn deinit(exchange: *Exchange, allocator: std.mem.Allocator) void { allocator.free(exchange.slots); allocator.free(exchange.response_buffer); allocator.free(exchange.table); + allocator.free(exchange.peer_heads); } /// Hash peer address only (no message ID) for peer-based eviction. @@ -191,7 +202,14 @@ pub fn insert( .response_length = @intCast(response_data.len), .completed_at_ns = now_ns, .next_free = empty_sentinel, + .peer_prev = empty_sentinel, + .peer_next = empty_sentinel, }; + const bucket = exchange.peerBucket(addr_key); + const head = exchange.peer_heads[bucket]; + slot.peer_next = head; + if (head != empty_sentinel) exchange.slots[head].peer_prev = slot_idx; + exchange.peer_heads[bucket] = slot_idx; // Cache the response. const offset = @as(usize, slot_idx) * exchange.config.response_size_max; @@ -221,19 +239,26 @@ pub fn cachedResponse(exchange: *const Exchange, slot_idx: u16) []const u8 { } /// Evict all completed exchanges for a given peer address. -/// Called when a new (non-duplicate) request arrives from a peer, -/// proving the peer received all prior responses. +/// Called by the existing server policy on a new non-duplicate request. +/// This does not prove reception of every older response. pub fn evictPeer(exchange: *Exchange, ak: u32) u16 { var evicted: u16 = 0; - for (exchange.slots, 0..) |*slot, i| { - if (slot.state != .completed) continue; - if (slot.addr_key != ak) continue; - exchange.remove(@intCast(i)); - evicted += 1; + var idx = exchange.peer_heads[exchange.peerBucket(ak)]; + while (idx != empty_sentinel) { + const next = exchange.slots[idx].peer_next; + if (exchange.slots[idx].addr_key == ak) { + exchange.remove(idx); + evicted += 1; + } + idx = next; } return evicted; } +fn peerBucket(exchange: *const Exchange, ak: u32) u16 { + return @as(u16, @truncate(ak)) & exchange.table_mask; +} + /// Evict exchanges that have expired past the exchange lifetime. /// Returns the number of evicted exchanges. pub fn evictExpired(exchange: *Exchange, now_ns: i64, lifetime_ms: u32) u16 { @@ -261,6 +286,16 @@ pub fn remove(exchange: *Exchange, slot_idx: u16) void { // Remove from hash table. exchange.removeFromTable(slot.peerKey); + if (slot.peer_prev == empty_sentinel) { + exchange.peer_heads[exchange.peerBucket(slot.addr_key)] = slot.peer_next; + } else { + exchange.slots[slot.peer_prev].peer_next = slot.peer_next; + } + if (slot.peer_next != empty_sentinel) { + exchange.slots[slot.peer_next].peer_prev = slot.peer_prev; + } + slot.peer_prev = empty_sentinel; + slot.peer_next = empty_sentinel; // Return to free list. slot.state = .free; @@ -322,6 +357,151 @@ fn wrappingDistance(from: u16, to: u16, mask: u16) u16 { const testing = std.testing; +fn checkPeerIndex(pool: *const Exchange) !void { + var linked: [32]bool = @splat(false); + var free: [32]bool = @splat(false); + try testing.expect(pool.slots.len <= linked.len); + var members: usize = 0; + for (pool.peer_heads, 0..) |head, bucket| { + var current = head; + var previous = empty_sentinel; + while (current != empty_sentinel) { + try testing.expect(current < pool.slots.len); + try testing.expect(!linked[current]); + linked[current] = true; + const slot = pool.slots[current]; + try testing.expectEqual(State.completed, slot.state); + try testing.expectEqual(previous, slot.peer_prev); + try testing.expectEqual(bucket, @as(usize, pool.peerBucket(slot.addr_key))); + try testing.expectEqual(current, pool.find(slot.peerKey).?); + previous = current; + current = slot.peer_next; + members += 1; + } + } + try testing.expectEqual(@as(usize, pool.count_active), members); + var current = pool.free_head; + var free_count: usize = 0; + while (current != empty_sentinel) { + try testing.expect(current < pool.slots.len); + try testing.expect(!free[current]); + free[current] = true; + try testing.expectEqual(State.free, pool.slots[current].state); + current = pool.slots[current].next_free; + free_count += 1; + } + try testing.expectEqual(pool.slots.len, members + free_count); + for (pool.slots, 0..) |slot, i| { + try testing.expectEqual(slot.state == .completed, linked[i]); + try testing.expectEqual(slot.state == .free, free[i]); + } +} + +fn checkIndexAllocation(allocator: std.mem.Allocator) !void { + var pool = try Exchange.init(allocator, .{ .exchange_count = 8, .response_size_max = 16 }); + defer pool.deinit(allocator); +} + +test "peer index allocation failures are leak free" { + try testing.checkAllAllocationFailures(testing.allocator, checkIndexAllocation, .{}); +} + +test "peer index handles bucket collisions, middle removal, expiry and reuse" { + var pool = try Exchange.init(testing.allocator, .{ .exchange_count = 8, .response_size_max = 16 }); + defer pool.deinit(testing.allocator); + // Same bucket, different address hashes; key table also wraps/probes. + const first = pool.insert(15, 1, 1, "first", 0).?; + const middle = pool.insert(31, 17, 2, "middle", 20 * std.time.ns_per_ms).?; + _ = pool.insert(47, 1, 3, "last", 20 * std.time.ns_per_ms).?; + try checkPeerIndex(&pool); + pool.remove(middle); + pool.remove(middle); // Already-free removal must not corrupt links. + try checkPeerIndex(&pool); + _ = pool.insert(63, 33, 4, "reuse", 20 * std.time.ns_per_ms).?; + try testing.expectEqual(@as(u16, 1), pool.evictExpired(31 * std.time.ns_per_ms, 30)); + try testing.expect(pool.find(15) == null); + try testing.expectEqual(State.free, pool.slots[first].state); + try checkPeerIndex(&pool); + try testing.expectEqual(@as(u16, 1), pool.evictPeer(1)); + try testing.expect(pool.find(63) != null); + try testing.expectEqualStrings("reuse", pool.cachedResponse(pool.find(63).?)); + try testing.expectEqual(@as(u16, 0), pool.evictPeer(17)); + try testing.expectEqual(@as(u16, 1), pool.evictPeer(33)); + try checkPeerIndex(&pool); +} + +test "peer index retains address hash grouping and rejected insert semantics" { + var pool = try Exchange.init(testing.allocator, .{ .exchange_count = 2, .response_size_max = 4 }); + defer pool.deinit(testing.allocator); + _ = pool.insert(1, 0, 1, "one", 0).?; + try testing.expect(pool.insert(1, 99, 1, "dup", 1) == null); + try testing.expect(pool.insert(2, 7, 2, "large", 1) == null); + _ = pool.insert(2, 0, 2, "two", 1).?; + try testing.expect(pool.insert(3, 7, 3, "full", 2) == null); + try checkPeerIndex(&pool); + // Identical addr_key still evicts both, regardless of actual address. + try testing.expectEqual(@as(u16, 2), pool.evictPeer(0)); + try checkPeerIndex(&pool); + _ = pool.insert(3, std.math.maxInt(u32), 3, "new", 3).?; + try testing.expectEqual(@as(u16, 1), pool.evictPeer(std.math.maxInt(u32))); + try checkPeerIndex(&pool); +} + +test "peer index randomized churn matches a scan model" { + var pool = try Exchange.init(testing.allocator, .{ .exchange_count = 32, .response_size_max = 16 }); + defer pool.deinit(testing.allocator); + const Entry = struct { active: bool = false, addr: u32 = 0, time: i64 = 0 }; + var model: [64]Entry = @splat(.{}); + var seed: u64 = 0x8db39427a21; + var now: i64 = 0; + for (0..5000) |_| { + seed = seed *% 6364136223846793005 +% 1442695040888963407; + const id: usize = @intCast((seed >> 16) % model.len); + const key: u64 = 0x100000000 + id; + const addr: u32 = @intCast((id % 8) * pool.peer_heads.len + id % 2); + now += std.time.ns_per_ms; + switch ((seed >> 32) % 6) { + 0, 1, 2 => { + const capacity = pool.count_active < pool.config.exchange_count; + const inserted = pool.insert(key, addr, @intCast(id), "cache", now); + try testing.expectEqual(!model[id].active and capacity, inserted != null); + if (inserted != null) model[id] = .{ .active = true, .addr = addr, .time = now }; + }, + 3 => { + if (pool.find(key)) |slot| pool.remove(slot); + model[id].active = false; + }, + 4 => { + var expected: u16 = 0; + for (&model) |*entry| { + if (entry.active and entry.addr == addr) { + entry.active = false; + expected += 1; + } + } + try testing.expectEqual(expected, pool.evictPeer(addr)); + }, + 5 => { + var expected: u16 = 0; + for (&model) |*entry| { + if (entry.active and now - entry.time >= 10 * std.time.ns_per_ms) { + entry.active = false; + expected += 1; + } + } + try testing.expectEqual(expected, pool.evictExpired(now, 10)); + }, + else => unreachable, + } + try checkPeerIndex(&pool); + for (model, 0..) |entry, i| { + const found = pool.find(0x100000000 + i); + try testing.expectEqual(entry.active, found != null); + if (found) |slot| try testing.expectEqualStrings("cache", pool.cachedResponse(slot)); + } + } +} + test "init and deinit" { var pool = try Exchange.init(testing.allocator, .{ .exchange_count = 8,