From 56ccf388f68633814a963e6db54da5b0eca2d404 Mon Sep 17 00:00:00 2001 From: Endel Dreyer Date: Fri, 18 Sep 2026 19:20:39 -0300 Subject: [PATCH 1/6] kqueue: keep a rearmed completion active when submit() reported it When a completion's fd is already ready at submission, the kevent() call in submit() returns its event and the callback runs from the completion queue. If that callback returns .rearm, the kqueue filter stays registered but the completion is left in the .dead state. A later cancel then sees a finished completion and does nothing, so the callback keeps firing after the owner believes it cancelled, and possibly freed, the completion. Assisted-by: Claude Opus 5 --- src/backend/kqueue.zig | 66 ++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 64 insertions(+), 2 deletions(-) diff --git a/src/backend/kqueue.zig b/src/backend/kqueue.zig index a78249dd..bcac3c9c 100644 --- a/src/backend/kqueue.zig +++ b/src/backend/kqueue.zig @@ -456,8 +456,12 @@ pub const Loop = struct { if (c_active) self.active -= 1; }, - // Only resubmit if we aren't already active (in the queue) - .rearm => if (!c_active) self.submissions.push(c), + // Still registered with kqueue (reported by submit's event + // list): it must stay active, or its next event queues it twice. + .rearm => if (disarm_ev != null) { + c.flags.state = .active; + c.result = null; + } else if (!c_active) self.submissions.push(c), } // If we filled the events slice, we break to avoid overflow. @@ -2860,6 +2864,64 @@ test "kqueue: timer armed from delayed callback must not fire early" { try testing.expect(elapsed_ms >= @as(i128, @intCast(timer_delay_ms))); } +test "kqueue: cancel after a rearm of a completion reported at submit" { + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + var fds: [2]posix.fd_t = undefined; + try testing.expectEqual(@as(c_int, 0), std.c.pipe(&fds)); + defer _ = std.c.close(fds[0]); + defer _ = std.c.close(fds[1]); + + // Readable before submission, so submit's event list reports it and the + // callback runs from the completion queue. + try testing.expectEqual(@as(isize, 1), std.c.write(fds[1], "a", 1)); + + const State = struct { reads: usize = 0, canceled: bool = false }; + var state: State = .{}; + var buf: [1]u8 = undefined; + var c: Completion = .{ + .op = .{ .read = .{ .fd = fds[0], .buffer = .{ .slice = &buf } } }, + .userdata = &state, + .callback = (struct { + fn callback(ud: ?*anyopaque, _: *Loop, _: *Completion, r: Result) CallbackAction { + const s = @as(*State, @ptrCast(@alignCast(ud.?))); + _ = r.read catch |err| { + s.canceled = err == error.Canceled; + return .disarm; + }; + s.reads += 1; + return .rearm; + } + }).callback, + }; + loop.add(&c); + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 1), state.reads); + try testing.expect(c.state() == .active); + + var cancel: Completion = .{ + .op = .{ .cancel = .{ .c = &c } }, + .callback = (struct { + fn callback(_: ?*anyopaque, _: *Loop, _: *Completion, r: Result) CallbackAction { + _ = r.cancel catch unreachable; + return .disarm; + } + }).callback, + }; + loop.add(&cancel); + try loop.run(.no_wait); + try testing.expect(state.canceled); + + // Cancelled, so new data must not reach the callback. + try testing.expectEqual(@as(isize, 1), std.c.write(fds[1], "b", 1)); + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 1), state.reads); + try testing.expectEqual(@as(usize, 0), loop.active); +} + test "kqueue: socket accept/cancel cancellation should decrease active count" { const mem = std.mem; const testing = std.testing; From 6995a3ae8d8c34ed04d9425bd38851f5f4588b6d Mon Sep 17 00:00:00 2001 From: Bryan Date: Fri, 26 Jun 2026 23:36:12 -0600 Subject: [PATCH 2/6] test: reproduce kqueue tick(0) and queue re-enqueue bugs Add tests for two bugs: Bug #209: tick(0) EV_DELETE flush - kqueue tick(0) may leave stale kernel filters when disarming. Six tests added covering multiple disarms, write/read disarm on same fd, rapid cycles, no-op tick, stale filter detection, and the kevent-event-processing race path (fires without fix, passes with fix). Bug #170: rearm from non-kqueue completions - when a completion that completes without kqueue registration returns .rearm, it is pushed to submissions with .dead state which gets routed to stop_completion instead of start. Integration test added (fails without fix, passes with fix). Queue unit tests added verifying re-enqueue across queues, drain-and-refill, and rapid cycle invariants. (cherry picked from commit efdf2cbff9013ca9c79029fbef5c6e893e879d8d) --- src/backend/kqueue.zig | 438 +++++++++++++++++++++++++++++++++++++++++ src/queue.zig | 92 +++++++++ 2 files changed, 530 insertions(+) diff --git a/src/backend/kqueue.zig b/src/backend/kqueue.zig index bcac3c9c..3c1609c4 100644 --- a/src/backend/kqueue.zig +++ b/src/backend/kqueue.zig @@ -3032,3 +3032,441 @@ test "kqueue: socket accept/cancel cancellation should decrease active count" { try loop.run(.until_done); try testing.expect(ln == 0); } + +fn makePipe() ![2]posix.fd_t { + var fds: [2]posix.fd_t = undefined; + const rc = posix.system.pipe(&fds); + return switch (posix.errno(rc)) { + .SUCCESS => fds, + .MFILE => error.ProcessFdQuotaExceeded, + .NFILE => error.SystemFdQuotaExceeded, + else => |err| posix.unexpectedErrno(err), + }; +} + +test "kqueue: tick(0) flushes EV_DELETE after disarm" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const pfds = try makePipe(); + defer { + xev_posix.close(pfds[0]); + xev_posix.close(pfds[1]); + } + + const reader = pfds[0]; + const writer = pfds[1]; + + var fire_count: usize = 0; + var buf: [16]u8 = undefined; + + const disarm_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.read catch unreachable; + const ptr: *usize = @ptrCast(@alignCast(ud.?)); + ptr.* += 1; + return .disarm; + } + }).callback; + + // First read: arm, trigger, disarm. + var c1: Completion = .{ + .op = .{ .read = .{ .fd = reader, .buffer = .{ .slice = &buf } } }, + .userdata = &fire_count, + .callback = disarm_cb, + }; + loop.add(&c1); + _ = try xev_posix.write(writer, &[_]u8{42}); + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 1), fire_count); + + // Second read: if stale filter lingers, it fires alongside the new + // one and fire_count exceeds 2. + _ = try xev_posix.write(writer, &[_]u8{43, 44}); + var c2: Completion = .{ + .op = .{ .read = .{ .fd = reader, .buffer = .{ .slice = &buf } } }, + .userdata = &fire_count, + .callback = disarm_cb, + }; + loop.add(&c2); + try loop.run(.once); + try testing.expectEqual(@as(usize, 2), fire_count); + try testing.expectEqual(@as(usize, 0), loop.active); +} + +test "kqueue: tick(0) no-op when nothing pending" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 0), loop.active); +} + +test "kqueue: multiple disarms in a single tick(0)" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const pipe1 = try makePipe(); + defer { xev_posix.close(pipe1[0]); xev_posix.close(pipe1[1]); } + + const pipe2 = try makePipe(); + defer { xev_posix.close(pipe2[0]); xev_posix.close(pipe2[1]); } + + var fire_count: usize = 0; + var buf1: [16]u8 = undefined; + var buf2: [16]u8 = undefined; + + const disarm_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.read catch unreachable; + const ptr: *usize = @ptrCast(@alignCast(ud.?)); + ptr.* += 1; + return .disarm; + } + }).callback; + + var c1: Completion = .{ + .op = .{ .read = .{ .fd = pipe1[0], .buffer = .{ .slice = &buf1 } } }, + .userdata = &fire_count, + .callback = disarm_cb, + }; + loop.add(&c1); + + var c2: Completion = .{ + .op = .{ .read = .{ .fd = pipe2[0], .buffer = .{ .slice = &buf2 } } }, + .userdata = &fire_count, + .callback = disarm_cb, + }; + loop.add(&c2); + + _ = try xev_posix.write(pipe1[1], &[_]u8{1}); + _ = try xev_posix.write(pipe2[1], &[_]u8{2}); + + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 2), fire_count); + + // Verify stale filters don't linger. + _ = try xev_posix.write(pipe1[1], &[_]u8{3}); + _ = try xev_posix.write(pipe2[1], &[_]u8{4}); + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 2), fire_count); + try testing.expectEqual(@as(usize, 0), loop.active); +} + +test "kqueue: disarm write while read is armed on same fd" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const pipe_fds = try makePipe(); + defer { + xev_posix.close(pipe_fds[0]); + xev_posix.close(pipe_fds[1]); + } + + const reader = pipe_fds[0]; + const writer = pipe_fds[1]; + + var read_fired = false; + var write_fired = false; + var read_buf: [1]u8 = undefined; + + var c_write: Completion = .{ + .op = .{ .write = .{ .fd = writer, .buffer = .{ .slice = "x" } } }, + .userdata = &write_fired, + .callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.write catch unreachable; + const ptr: *bool = @ptrCast(@alignCast(ud.?)); + ptr.* = true; + return .disarm; + } + }).callback, + }; + loop.add(&c_write); + + var c_read: Completion = .{ + .op = .{ .read = .{ .fd = reader, .buffer = .{ .slice = &read_buf } } }, + .userdata = &read_fired, + .callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.read catch unreachable; + const ptr: *bool = @ptrCast(@alignCast(ud.?)); + ptr.* = true; + return .disarm; + } + }).callback, + }; + loop.add(&c_read); + + _ = try xev_posix.write(writer, "y"); + try loop.run(.no_wait); + try testing.expect(write_fired); + try testing.expect(read_fired); + try testing.expectEqual(@as(usize, 0), loop.active); +} + +test "kqueue: rapid tick(0) disarm cycles" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const pipe_fds = try makePipe(); + defer { + xev_posix.close(pipe_fds[0]); + xev_posix.close(pipe_fds[1]); + } + + const reader = pipe_fds[0]; + const writer = pipe_fds[1]; + + var fire_total: usize = 0; + var buf: [4]u8 = undefined; + + const disarm_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.read catch unreachable; + const ptr: *usize = @ptrCast(@alignCast(ud.?)); + ptr.* += 1; + return .disarm; + } + }).callback; + + for (0..5) |_| { + var c_read: Completion = .{ + .op = .{ .read = .{ .fd = reader, .buffer = .{ .slice = &buf } } }, + .userdata = &fire_total, + .callback = disarm_cb, + }; + loop.add(&c_read); + _ = try xev_posix.write(writer, &[_]u8{1}); + try loop.run(.no_wait); + } + + try testing.expectEqual(@as(usize, 5), fire_total); + try testing.expectEqual(@as(usize, 0), loop.active); + + // Verify no stale filters remain. + _ = try xev_posix.write(writer, &[_]u8{9, 9, 9}); + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 5), fire_total); +} + +test "kqueue: rearm from non-kqueue completion re-enqueues correctly" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const pfds1 = try makePipe(); + const pfds2 = try makePipe(); + defer xev_posix.close(pfds1[0]); + defer xev_posix.close(pfds2[0]); + + const State = struct { + fire_count: usize = 0, + fd1: posix.fd_t, + fd2: posix.fd_t, + }; + var state = State{ .fd1 = pfds1[1], .fd2 = pfds2[1] }; + + const rearm_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = r.close catch {}; + const s: *State = @ptrCast(@alignCast(ud.?)); + s.fire_count += 1; + if (s.fire_count == 1) { + c.op = .{ .close = .{ .fd = s.fd2 } }; + return .rearm; + } + return .disarm; + } + }).callback; + + var c1: Completion = .{ + .op = .{ .close = .{ .fd = state.fd1 } }, + .userdata = &state, + .callback = rearm_cb, + }; + loop.add(&c1); + + // First tick: close fd1, callback fires, returns .rearm. + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 1), state.fire_count); + + // Second tick: without fix, .dead state routes to stop_completion + // and callback never fires again. With fix, close fd2 succeeds. + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 2), state.fire_count); + + try testing.expectEqual(@as(usize, 0), loop.active); +} + +test "kqueue: tick(0) flushes disarm from kevent event processing path" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const pA = try makePipe(); + const pB = try makePipe(); + defer { xev_posix.close(pA[0]); xev_posix.close(pA[1]); } + defer { xev_posix.close(pB[0]); xev_posix.close(pB[1]); } + + const State = struct { + fire_count: usize = 0, + writerB: posix.fd_t, + buf: [16]u8 = undefined, + }; + var state = State{ .writerB = pB[1] }; + + // readB callback: disarm and count. + const disarm_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.read catch unreachable; + const s: *State = @ptrCast(@alignCast(ud.?)); + s.fire_count += 1; + return .disarm; + } + }).callback; + + // readA callback: write to pipeB so readB fires from the flushing + // kevent_syscall in the same tick, then disarm. + const disarmA_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.read catch unreachable; + const s: *State = @ptrCast(@alignCast(ud.?)); + s.fire_count += 1; + _ = xev_posix.write(s.writerB, &[_]u8{42}) catch unreachable; + return .disarm; + } + }).callback; + + // Arm both reads. Only pipeA has data. + _ = try xev_posix.write(pA[1], &[_]u8{1}); + + var cA: Completion = .{ + .op = .{ .read = .{ .fd = pA[0], .buffer = .{ .slice = &state.buf } } }, + .userdata = &state, + .callback = disarmA_cb, + }; + loop.add(&cA); + + var cB: Completion = .{ + .op = .{ .read = .{ .fd = pB[0], .buffer = .{ .slice = &state.buf } } }, + .userdata = &state, + .callback = disarm_cb, + }; + loop.add(&cB); + + // tick(0): readA fires, writes to pipeB, returns disarm (changes=1). + // Flushing kevent_syscall returns readB. readB fires, returns disarm + // (changes=1 again). Without fix, readB's EV_DELETE is lost. + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 2), state.fire_count); + try testing.expectEqual(@as(usize, 0), loop.active); + + // Write more data, then arm a timer to keep the loop alive so + // kevent_syscall is reached. If stale readB filter lingers, it fires. + _ = try xev_posix.write(pB[1], &[_]u8{43}); + + var timer_fired = false; + var c_timer: Completion = undefined; + loop.timer(&c_timer, 100, &timer_fired, (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.timer catch unreachable; + const ptr: *bool = @ptrCast(@alignCast(ud.?)); + ptr.* = true; + return .disarm; + } + }).callback); + + try loop.run(.once); + try testing.expectEqual(@as(usize, 2), state.fire_count); + try testing.expect(timer_fired); + try testing.expectEqual(@as(usize, 0), loop.active); +} diff --git a/src/queue.zig b/src/queue.zig index 7cafda37..21b37429 100644 --- a/src/queue.zig +++ b/src/queue.zig @@ -99,3 +99,95 @@ test Intrusive { try testing.expect(q.pop().? == &elems[1]); try testing.expect(q.pop() == null); } + +test "Intrusive: re-enqueue across queues clears next" { + const testing = std.testing; + + const Elem = struct { + const Self = @This(); + next: ?*Self = null, + }; + const Queue = Intrusive(Elem); + + var qA: Queue = .{}; + var qB: Queue = .{}; + + var elems: [5]Elem = .{Elem{}} ** 5; + + // Enqueue all into A. + for (&elems) |*e| qA.push(e); + try testing.expect(!qA.empty()); + try testing.expect(qB.empty()); + + // Dequeue one from A, enqueue into B. + const moved = qA.pop().?; + qB.push(moved); + + // B: exactly one node, tail.next is null. + try testing.expect(!qB.empty()); + try testing.expectEqual(&elems[0], qB.head.?); + try testing.expectEqual(&elems[0], qB.tail.?); + try testing.expect(qB.head.?.next == null); + + // A: remaining 4 nodes, last.next is null. + try testing.expect(!qA.empty()); + var count: usize = 0; + var cur = qA.head; + while (cur) |node| : (cur = node.next) { + count += 1; + // Verify no cross-queue contamination: no node in A points into B. + try testing.expect(node != qB.head); + } + try testing.expectEqual(@as(usize, 4), count); + try testing.expect(qA.tail.?.next == null); +} + +test "Intrusive: drain and refill" { + const testing = std.testing; + + const Elem = struct { + const Self = @This(); + next: ?*Self = null, + }; + const Queue = Intrusive(Elem); + + var qA: Queue = .{}; + var qB: Queue = .{}; + + var elems: [5]Elem = .{Elem{}} ** 5; + + // Enqueue all into A. + for (&elems) |*e| qA.push(e); + + // Move all from A to B. + while (qA.pop()) |e| qB.push(e); + try testing.expect(qA.empty()); + + // Verify B has all 5, no loops. + var count: usize = 0; + var cur = qB.head; + while (cur) |node| : (cur = node.next) count += 1; + try testing.expectEqual(@as(usize, 5), count); + try testing.expect(qB.tail.?.next == null); +} + +test "Intrusive: rapid enqueue-dequeue cycles" { + const testing = std.testing; + + const Elem = struct { + const Self = @This(); + next: ?*Self = null, + }; + const Queue = Intrusive(Elem); + + var q: Queue = .{}; + var elems: [10]Elem = .{Elem{}} ** 10; + + for (0..100) |_| { + for (&elems) |*e| q.push(e); + var count: usize = 0; + while (q.pop()) |_| count += 1; + try testing.expectEqual(@as(usize, 10), count); + try testing.expect(q.empty()); + } +} From 570e3b0b8ef13a7cf176e0aa23b02907e6aefb64 Mon Sep 17 00:00:00 2001 From: Bryan Date: Sat, 27 Jun 2026 10:19:52 -0600 Subject: [PATCH 3/6] fix: flush EV_DELETE in tick(0), set rearm state to .adding, clear next in submit Bug #209: kqueue tick(0) EV_DELETE flush - Changed early exit condition from 'if (wait == 0) break' to 'if (wait == 0 and changes == 0) break' so pending EV_DELETE kevent changes from the event-processing loop are flushed before tick(0) returns. Bug #170: rearm state correction - When a non-kqueue completion (close, shutdown, threadpool fallback) returns .rearm, set flags.state to .adding before pushing to the submissions queue. Previously .dead state caused submit() to route through stop_completion (cancel) instead of start() (resubmit). Issue #169: defensive next-pointer clear in submit() - Explicitly clear c.next before push to completions queue when processing kevent-returned events in submit(). This prevents an assertion failure if a completion is present in multiple queues concurrently (race condition). (cherry picked from mitchellh/libxev#224; the .rearm arm is merged with this fork's keep-active fix: a completion still registered with kqueue stays .active, one never registered is resubmitted as .adding) (cherry picked from commit a7502d6fe7c5bf5d5afea1d7e4dc4437f4a05184) --- src/backend/kqueue.zig | 94 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 92 insertions(+), 2 deletions(-) diff --git a/src/backend/kqueue.zig b/src/backend/kqueue.zig index 3c1609c4..80206ed5 100644 --- a/src/backend/kqueue.zig +++ b/src/backend/kqueue.zig @@ -264,6 +264,7 @@ pub const Loop = struct { } assert(c.result != null); + c.next = null; self.completions.push(c); } } @@ -458,10 +459,15 @@ pub const Loop = struct { // Still registered with kqueue (reported by submit's event // list): it must stay active, or its next event queues it twice. + // Never registered (close, shutdown, ...): resubmit as .adding, + // since submit stops a .dead completion instead of starting it. .rearm => if (disarm_ev != null) { c.flags.state = .active; c.result = null; - } else if (!c_active) self.submissions.push(c), + } else if (!c_active) { + c.flags.state = .adding; + self.submissions.push(c); + }, } // If we filled the events slice, we break to avoid overflow. @@ -567,7 +573,7 @@ pub const Loop = struct { } // If we ran through the loop once we break if we don't care. - if (wait == 0) break; + if (wait == 0 and changes == 0) break; } } @@ -3470,3 +3476,87 @@ test "kqueue: tick(0) flushes disarm from kevent event processing path" { try testing.expect(timer_fired); try testing.expectEqual(@as(usize, 0), loop.active); } + +test "kqueue: re-add armed completion from callback does not corrupt queue" { + if (builtin.os.tag != .macos) return error.SkipZigTest; + + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const p1 = try makePipe(); + const p2 = try makePipe(); + defer { xev_posix.close(p1[0]); xev_posix.close(p1[1]); } + defer { xev_posix.close(p2[0]); xev_posix.close(p2[1]); } + + const State = struct { + fire_count: usize = 0, + c2: *Completion, + buf: [16]u8 = undefined, + }; + + // Placeholder for c2, filled in after we have the pointer. + var c2: Completion = undefined; + var state = State{ .c2 = &c2 }; + + // c1 callback: re-add c2 to submissions (simulates #169 scenario), disarm. + const c1_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = c; + _ = r.read catch unreachable; + const s: *State = @ptrCast(@alignCast(ud.?)); + s.fire_count += 1; + l.add(s.c2); + return .disarm; + } + }).callback; + + // c2 callback: disarm and count. + const c2_cb: Callback = (struct { + fn callback( + ud: ?*anyopaque, + l: *Loop, + c: *Completion, + r: Result, + ) CallbackAction { + _ = l; + _ = c; + _ = r.read catch unreachable; + const s: *State = @ptrCast(@alignCast(ud.?)); + s.fire_count += 1; + return .disarm; + } + }).callback; + + // Arm c2 first (in kqueue), then arm c1. + c2 = .{ + .op = .{ .read = .{ .fd = p2[0], .buffer = .{ .slice = &state.buf } } }, + .userdata = &state, + .callback = c2_cb, + }; + loop.add(&c2); + + var c1: Completion = .{ + .op = .{ .read = .{ .fd = p1[0], .buffer = .{ .slice = &state.buf } } }, + .userdata = &state, + .callback = c1_cb, + }; + loop.add(&c1); + + // Write to p1: c1 fires, re-adds c2, then disarms. + // Write to p2: c2 fires (from the re-add OR from its original kqueue + // registration). The defensive c.next=null in submit prevents the + // queue.zig:24 assert from firing when the kevent returns c2's event. + _ = try xev_posix.write(p1[1], &[_]u8{1}); + _ = try xev_posix.write(p2[1], &[_]u8{2}); + + try loop.run(.no_wait); + try testing.expectEqual(@as(usize, 2), state.fire_count); + try testing.expectEqual(@as(usize, 0), loop.active); +} From 4d0a273683f073a71e7aa3fe93576c6004a536b2 Mon Sep 17 00:00:00 2001 From: Endel Dreyer Date: Fri, 18 Sep 2026 23:26:25 -0300 Subject: [PATCH 4/6] kqueue: zig fmt the tick(0) and rearm tests from #224 Assisted-by: Claude Opus 5 (cherry picked from commit 6f8278bd7f0ac5c5df354d10f84dafd246426d20) --- src/backend/kqueue.zig | 34 ++++++++++++++++++++++++++-------- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/src/backend/kqueue.zig b/src/backend/kqueue.zig index 80206ed5..603678ee 100644 --- a/src/backend/kqueue.zig +++ b/src/backend/kqueue.zig @@ -3099,7 +3099,7 @@ test "kqueue: tick(0) flushes EV_DELETE after disarm" { // Second read: if stale filter lingers, it fires alongside the new // one and fire_count exceeds 2. - _ = try xev_posix.write(writer, &[_]u8{43, 44}); + _ = try xev_posix.write(writer, &[_]u8{ 43, 44 }); var c2: Completion = .{ .op = .{ .read = .{ .fd = reader, .buffer = .{ .slice = &buf } } }, .userdata = &fire_count, @@ -3132,10 +3132,16 @@ test "kqueue: multiple disarms in a single tick(0)" { defer loop.deinit(); const pipe1 = try makePipe(); - defer { xev_posix.close(pipe1[0]); xev_posix.close(pipe1[1]); } + defer { + xev_posix.close(pipe1[0]); + xev_posix.close(pipe1[1]); + } const pipe2 = try makePipe(); - defer { xev_posix.close(pipe2[0]); xev_posix.close(pipe2[1]); } + defer { + xev_posix.close(pipe2[0]); + xev_posix.close(pipe2[1]); + } var fire_count: usize = 0; var buf1: [16]u8 = undefined; @@ -3306,7 +3312,7 @@ test "kqueue: rapid tick(0) disarm cycles" { try testing.expectEqual(@as(usize, 0), loop.active); // Verify no stale filters remain. - _ = try xev_posix.write(writer, &[_]u8{9, 9, 9}); + _ = try xev_posix.write(writer, &[_]u8{ 9, 9, 9 }); try loop.run(.no_wait); try testing.expectEqual(@as(usize, 5), fire_total); } @@ -3379,8 +3385,14 @@ test "kqueue: tick(0) flushes disarm from kevent event processing path" { const pA = try makePipe(); const pB = try makePipe(); - defer { xev_posix.close(pA[0]); xev_posix.close(pA[1]); } - defer { xev_posix.close(pB[0]); xev_posix.close(pB[1]); } + defer { + xev_posix.close(pA[0]); + xev_posix.close(pA[1]); + } + defer { + xev_posix.close(pB[0]); + xev_posix.close(pB[1]); + } const State = struct { fire_count: usize = 0, @@ -3487,8 +3499,14 @@ test "kqueue: re-add armed completion from callback does not corrupt queue" { const p1 = try makePipe(); const p2 = try makePipe(); - defer { xev_posix.close(p1[0]); xev_posix.close(p1[1]); } - defer { xev_posix.close(p2[0]); xev_posix.close(p2[1]); } + defer { + xev_posix.close(p1[0]); + xev_posix.close(p1[1]); + } + defer { + xev_posix.close(p2[0]); + xev_posix.close(p2[1]); + } const State = struct { fire_count: usize = 0, From 555c44d31cbc5acebb5ad65a7a87e4334e3d4265 Mon Sep 17 00:00:00 2001 From: Endel Dreyer Date: Sat, 3 Oct 2026 02:25:33 -0300 Subject: [PATCH 5/6] queue: build the #224 tests' arrays with @splat Zig 0.17 removes array multiplication. @splat builds the same array on 0.16 too. Assisted-by: Claude Opus 5.5 --- src/queue.zig | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/queue.zig b/src/queue.zig index 21b37429..ea17ae0a 100644 --- a/src/queue.zig +++ b/src/queue.zig @@ -112,7 +112,7 @@ test "Intrusive: re-enqueue across queues clears next" { var qA: Queue = .{}; var qB: Queue = .{}; - var elems: [5]Elem = .{Elem{}} ** 5; + var elems: [5]Elem = @splat(.{}); // Enqueue all into A. for (&elems) |*e| qA.push(e); @@ -154,7 +154,7 @@ test "Intrusive: drain and refill" { var qA: Queue = .{}; var qB: Queue = .{}; - var elems: [5]Elem = .{Elem{}} ** 5; + var elems: [5]Elem = @splat(.{}); // Enqueue all into A. for (&elems) |*e| qA.push(e); @@ -181,7 +181,7 @@ test "Intrusive: rapid enqueue-dequeue cycles" { const Queue = Intrusive(Elem); var q: Queue = .{}; - var elems: [10]Elem = .{Elem{}} ** 10; + var elems: [10]Elem = @splat(.{}); for (0..100) |_| { for (&elems) |*e| q.push(e); From 23d039f5f5c1031778a035b504f9f423674322ea Mon Sep 17 00:00:00 2001 From: Endel Dreyer Date: Wed, 28 Oct 2026 03:19:19 -0300 Subject: [PATCH 6/6] kqueue: register in submit() with EV_RECEIPT, so a ready completion runs once submit() took events with each batch of 256 changes. Read and write filters are level-triggered, so every batch reported again the completions earlier batches had queued: queueing one twice corrupted the intrusive queue, and a queued completion stayed registered, so the wait in tick() could run its callback a second time. When that callback freed it, the queue then read freed memory. With receipts submit() only learns which changes failed; ready completions arrive through the wait. Nothing left in the completion queue is registered, so tick() no longer deletes from it, which also removes its write past the 256-entry change list once the wait's own disarms had filled it. Assisted-by: Claude Opus 5.5 --- src/backend/kqueue.zig | 154 ++++++++++++++++++++++++----------------- 1 file changed, 91 insertions(+), 63 deletions(-) diff --git a/src/backend/kqueue.zig b/src/backend/kqueue.zig index 603678ee..76984876 100644 --- a/src/backend/kqueue.zig +++ b/src/backend/kqueue.zig @@ -229,21 +229,14 @@ pub const Loop = struct { // submission queue and we're done. if (events_len == 0) break; - // Zero timeout so that kevent returns immediately. - var timeout = std.mem.zeroes(posix.timespec); - const completed = try kevent_syscall( - self.kqueue_fd, - events[0..events_len], - events[0..events.len], - &timeout, - ); + // Receipts only: what is already ready comes through the wait + // in tick. Draining events here reported a level-triggered + // completion once per batch, queueing it twice, and left it + // registered while queued, so the wait could run it again. + const receipts = try applyChanges(self.kqueue_fd, events[0..events_len]); events_len = 0; - // Go through the completed events and queue them. - // NOTE: we currently never process completions (we set - // event list to zero length) because it was leading to - // memory corruption we need to investigate. - for (events[0..completed]) |ev| { + for (receipts) |ev| { // Zero udata values are internal events that we do nothing // on such as the mach port wakeup. if (ev.udata == 0) continue; @@ -251,19 +244,13 @@ pub const Loop = struct { // We handle deletions separately. if (ev.flags & std.c.EV.DELETE != 0) continue; - const c: *Completion = @ptrFromInt(@as(usize, @intCast(ev.udata))); + // Registered: its event comes through the wait. + if (ev.data == 0) continue; - // If EV_ERROR is set, then submission failed for this - // completion. We get the syscall errorcode from data and - // store it. - if (ev.flags & std.c.EV.ERROR != 0) { - c.result = c.syscall_result(-@as(i32, @intCast(ev.data))); - } else { - // No error, means that this completion is ready to work. - c.result = c.perform(&ev); - } - - assert(c.result != null); + // Submission failed for this completion: the syscall error + // code is in data. + const c: *Completion = @ptrFromInt(@as(usize, @intCast(ev.udata))); + c.result = c.syscall_result(-@as(i32, @intCast(ev.data))); c.next = null; self.completions.push(c); } @@ -417,20 +404,15 @@ pub const Loop = struct { } } - // Process the completions we already have completed. + // Process the completions we already have completed. None of + // them is registered with kqueue: submit() queues only the + // changes that failed, so nothing here needs deleting. while (self.completions.pop()) |c| { - // disarm_ev is the Kevent to use for disarming if the - // completion wants to disarm. We have to calculate this up - // front because c can be reused in callback. - const disarm_ev: ?Kevent = ev: { - // If we're not active then we were never part of the kqueue. - // If we are part of a threadpool we also never were part - // of the kqueue. - if (c.flags.state != .active or - c.flags.threadpool) break :ev null; - - break :ev c.kevent(); - }; + // A kevent operation whose registration failed, which a rearm + // registers again. Read up front because c can be reused in + // the callback. + const failed_kevent = c.flags.state == .active and + !c.flags.threadpool and c.kevent() != null; // We store whether this completion was active so we can decrement // the active count later @@ -443,35 +425,19 @@ pub const Loop = struct { // Completion queue items MUST have a result set. const action = c.callback(c.userdata, self, c, c.result.?); switch (action) { - // If we're active we have to schedule a delete. Otherwise - // we do nothing because we were never part of the kqueue. - .disarm => { - if (disarm_ev) |ev| { - events[changes] = ev; - events[changes].flags = std.c.EV.DELETE; - events[changes].udata = 0; - changes += 1; - assert(changes <= events.len); - } - - if (c_active) self.active -= 1; + .disarm => if (c_active) { + self.active -= 1; }, - // Still registered with kqueue (reported by submit's event - // list): it must stay active, or its next event queues it twice. - // Never registered (close, shutdown, ...): resubmit as .adding, - // since submit stops a .dead completion instead of starting it. - .rearm => if (disarm_ev != null) { - c.flags.state = .active; - c.result = null; - } else if (!c_active) { + // Resubmit as .adding, since submit stops a .dead + // completion instead of starting it. start() counts it + // active again. + .rearm => if (!c_active or failed_kevent) { + if (c_active) self.active -= 1; c.flags.state = .adding; self.submissions.push(c); }, } - - // If we filled the events slice, we break to avoid overflow. - if (changes == events.len) break; } // Determine our next timeout based on the timers @@ -1874,6 +1840,16 @@ const Kevent = switch (builtin.os.tag) { else => @compileError("kqueue not supported yet for target OS"), }; +/// Applies `changes` without taking any pending event: each change returns +/// one receipt in its own slot (EV_ERROR, with data zero on success), and an +/// event list exactly that long leaves the kernel no room to scan for more. +fn applyChanges(kq: i32, changes: []Kevent) KEventError![]Kevent { + for (changes) |*ev| ev.flags |= std.c.EV.RECEIPT; + var timeout = std.mem.zeroes(posix.timespec); + const n = try kevent_syscall(kq, changes, changes, &timeout); + return changes[0..n]; +} + /// kevent calls either kevent or kevent64 depending on the /// target platform. fn kevent_syscall( @@ -2881,8 +2857,7 @@ test "kqueue: cancel after a rearm of a completion reported at submit" { defer _ = std.c.close(fds[0]); defer _ = std.c.close(fds[1]); - // Readable before submission, so submit's event list reports it and the - // callback runs from the completion queue. + // Readable before submission, so the first wait reports it. try testing.expectEqual(@as(isize, 1), std.c.write(fds[1], "a", 1)); const State = struct { reads: usize = 0, canceled: bool = false }; @@ -3123,6 +3098,59 @@ test "kqueue: tick(0) no-op when nothing pending" { try testing.expectEqual(@as(usize, 0), loop.active); } +test "kqueue: more ready reads than one submit batch each complete once" { + // Readable before they're submitted: submit() registers them 256 at a + // time. Taking events with each batch reported the earlier batches + // again, queueing a completion twice, and left them registered while + // queued, so the wait ran them a second time. + const testing = std.testing; + const n = 600; + + var limit = try posix.getrlimit(.NOFILE); + if (limit.cur < 2 * n) { + if (limit.max < 2 * n) return error.SkipZigTest; + limit.cur = 2 * n; + try posix.setrlimit(.NOFILE, limit); + } + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const readers = try testing.allocator.alloc(posix.fd_t, n); + defer testing.allocator.free(readers); + const completions = try testing.allocator.alloc(Completion, n); + defer testing.allocator.free(completions); + const calls = try testing.allocator.alloc(u8, n); + defer testing.allocator.free(calls); + @memset(calls, 0); + var bufs: [n][1]u8 = undefined; + + for (readers, completions, calls, &bufs) |*fd, *c, *count, *buf| { + const p = try makePipe(); + _ = try xev_posix.write(p[1], &[_]u8{1}); + xev_posix.close(p[1]); + fd.* = p[0]; + c.* = .{ + .op = .{ .read = .{ .fd = p[0], .buffer = .{ .slice = buf } } }, + .userdata = count, + .callback = (struct { + fn callback(ud: ?*anyopaque, _: *Loop, _: *Completion, r: Result) CallbackAction { + _ = r.read catch unreachable; + const k: *u8 = @ptrCast(ud.?); + k.* += 1; + return .disarm; + } + }).callback, + }; + loop.add(c); + } + defer for (readers) |fd| xev_posix.close(fd); + + try loop.run(.until_done); + for (calls) |count| try testing.expectEqual(@as(u8, 1), count); + try testing.expectEqual(@as(usize, 0), loop.active); +} + test "kqueue: multiple disarms in a single tick(0)" { if (builtin.os.tag != .macos) return error.SkipZigTest;