authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-31 14:36:20-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:39-07:00
log5aa3d1425ec9ec0fbcc3926987a2127af6802049
treea1b1fcbd885231dd2275a78b1cb8ef5a03d39c8c
parentf3553049cb72a5ea44667e450a2ff1bf173a6b6d

EventLoop: implement detached async

data races on deinit tho

2 files changed, 133 insertions(+), 40 deletions(-)

lib/std/Io.zig+31-26
......@@ -979,7 +979,7 @@ pub const VTable = struct {
979979 /// Thread-safe.
980980 cancelRequested: *const fn (?*anyopaque) bool,
981981
982 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) error{Canceled}!void,
982 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,
983983 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
984984
985985 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex, timeout_ns: ?u64) Condition.WaitError!void,
......@@ -998,11 +998,11 @@ pub const VTable = struct {
998998pub const OpenFlags = fs.File.OpenFlags;
999999pub const CreateFlags = fs.File.CreateFlags;
10001000
1001pub const FileOpenError = fs.File.OpenError || error{Canceled};
1002pub const FileReadError = fs.File.ReadError || error{Canceled};
1003pub const FilePReadError = fs.File.PReadError || error{Canceled};
1004pub const FileWriteError = fs.File.WriteError || error{Canceled};
1005pub const FilePWriteError = fs.File.PWriteError || error{Canceled};
1001pub const FileOpenError = fs.File.OpenError || Cancelable;
1002pub const FileReadError = fs.File.ReadError || Cancelable;
1003pub const FilePReadError = fs.File.PReadError || Cancelable;
1004pub const FileWriteError = fs.File.WriteError || Cancelable;
1005pub const FilePWriteError = fs.File.PWriteError || Cancelable;
10061006
10071007pub const Timestamp = enum(i96) {
10081008 _,
......@@ -1019,7 +1019,7 @@ pub const Deadline = union(enum) {
10191019 nanoseconds: i96,
10201020 timestamp: Timestamp,
10211021};
1022pub const ClockGetTimeError = std.posix.ClockGetTimeError || error{Canceled};
1022pub const ClockGetTimeError = std.posix.ClockGetTimeError || Cancelable;
10231023pub const SleepError = error{ UnsupportedClock, Unexpected, Canceled };
10241024
10251025pub const AnyFuture = opaque {};
......@@ -1087,7 +1087,7 @@ pub const Mutex = if (true) struct {
10871087 return prev_state.isUnlocked();
10881088 }
10891089
1090 pub fn lock(mutex: *Mutex, io: std.Io) error{Canceled}!void {
1090 pub fn lock(mutex: *Mutex, io: std.Io) Cancelable!void {
10911091 const prev_state: State = @enumFromInt(@atomicRmw(
10921092 usize,
10931093 @as(*usize, @ptrCast(&mutex.state)),
......@@ -1136,7 +1136,7 @@ pub const Mutex = if (true) struct {
11361136 }
11371137
11381138 /// Avoids the vtable for uncontended locks.
1139 pub fn lock(m: *Mutex, io: Io) error{Canceled}!void {
1139 pub fn lock(m: *Mutex, io: Io) Cancelable!void {
11401140 if (!m.tryLock()) {
11411141 @branchHint(.unlikely);
11421142 try io.vtable.mutexLock(io.userdata, {}, m);
......@@ -1162,10 +1162,10 @@ pub const Condition = struct {
11621162 all,
11631163 };
11641164
1165 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) void {
1165 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {
11661166 io.vtable.conditionWait(io.userdata, cond, mutex, null) catch |err| switch (err) {
11671167 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out
1168 error.Canceled => return, // handled as spurious wakeup
1168 error.Canceled => return error.Canceled,
11691169 };
11701170 }
11711171
......@@ -1182,6 +1182,11 @@ pub const Condition = struct {
11821182 }
11831183};
11841184
1185pub const Cancelable = error{
1186 /// Caller has requested the async operation to stop.
1187 Canceled,
1188};
1189
11851190pub const TypeErasedQueue = struct {
11861191 mutex: Mutex,
11871192
......@@ -1205,7 +1210,7 @@ pub const TypeErasedQueue = struct {
12051210
12061211 pub fn init(buffer: []u8) TypeErasedQueue {
12071212 return .{
1208 .mutex = .{},
1213 .mutex = .init,
12091214 .buffer = buffer,
12101215 .put_index = 0,
12111216 .get_index = 0,
......@@ -1214,10 +1219,10 @@ pub const TypeErasedQueue = struct {
12141219 };
12151220 }
12161221
1217 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) usize {
1222 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) Cancelable!usize {
12181223 assert(elements.len >= min);
12191224
1220 q.mutex.lock(io);
1225 try q.mutex.lock(io);
12211226 defer q.mutex.unlock(io);
12221227
12231228 // Getters have first priority on the data, and only when the getters
......@@ -1264,15 +1269,15 @@ pub const TypeErasedQueue = struct {
12641269 .data = .{ .remaining = remaining, .condition = .{} },
12651270 };
12661271 q.putters.append(&node);
1267 node.data.condition.wait(io, &q.mutex);
1272 try node.data.condition.wait(io, &q.mutex);
12681273 remaining = node.data.remaining;
12691274 }
12701275 }
12711276
1272 pub fn get(q: *@This(), io: Io, buffer: []u8, min: usize) usize {
1277 pub fn get(q: *@This(), io: Io, buffer: []u8, min: usize) Cancelable!usize {
12731278 assert(buffer.len >= min);
12741279
1275 q.mutex.lock(io);
1280 try q.mutex.lock(io);
12761281 defer q.mutex.unlock(io);
12771282
12781283 // The ring buffer gets first priority, then data should come from any
......@@ -1329,7 +1334,7 @@ pub const TypeErasedQueue = struct {
13291334 .data = .{ .remaining = remaining, .condition = .{} },
13301335 };
13311336 q.getters.append(&node);
1332 node.data.condition.wait(io, &q.mutex);
1337 try node.data.condition.wait(io, &q.mutex);
13331338 remaining = node.data.remaining;
13341339 }
13351340 }
......@@ -1383,8 +1388,8 @@ pub fn Queue(Elem: type) type {
13831388 /// Returns how many elements have been added to the queue.
13841389 ///
13851390 /// Asserts that `elements.len >= min`.
1386 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) usize {
1387 return @divExact(q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
1391 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) Cancelable!usize {
1392 return @divExact(try q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
13881393 }
13891394
13901395 /// Receives elements from the beginning of the queue. The function
......@@ -1394,17 +1399,17 @@ pub fn Queue(Elem: type) type {
13941399 /// Returns how many elements of `buffer` have been populated.
13951400 ///
13961401 /// Asserts that `buffer.len >= min`.
1397 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) usize {
1398 return @divExact(q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
1402 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) Cancelable!usize {
1403 return @divExact(try q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
13991404 }
14001405
1401 pub fn putOne(q: *@This(), io: Io, item: Elem) void {
1402 assert(q.put(io, &.{item}, 1) == 1);
1406 pub fn putOne(q: *@This(), io: Io, item: Elem) Cancelable!void {
1407 assert(try q.put(io, &.{item}, 1) == 1);
14031408 }
14041409
1405 pub fn getOne(q: *@This(), io: Io) Elem {
1410 pub fn getOne(q: *@This(), io: Io) Cancelable!Elem {
14061411 var buf: [1]Elem = undefined;
1407 assert(q.get(io, &buf, 1) == 1);
1412 assert(try q.get(io, &buf, 1) == 1);
14081413 return buf[0];
14091414 }
14101415 };
lib/std/Io/EventLoop.zig+102-14
......@@ -27,6 +27,7 @@ const Thread = struct {
2727 current_context: *Context,
2828 ready_queue: ?*Fiber,
2929 free_queue: ?*Fiber,
30 detached_queue: ?*Fiber,
3031 io_uring: IoUring,
3132 idle_search_index: u32,
3233 steal_ready_search_index: u32,
......@@ -208,6 +209,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
208209 .current_context = &main_fiber.context,
209210 .ready_queue = null,
210211 .free_queue = null,
212 .detached_queue = null,
211213 .io_uring = try IoUring.init(io_uring_entries, 0),
212214 .idle_search_index = 1,
213215 .steal_ready_search_index = 1,
......@@ -218,7 +220,16 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
218220}
219221
220222pub fn deinit(el: *EventLoop) void {
223 // Wait for detached fibers.
221224 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
225 for (el.threads.allocated[0..active_threads]) |*thread| {
226 while (thread.detached_queue) |detached_fiber| {
227 if (@atomicLoad(?*Fiber, &detached_fiber.awaiter, .acquire) != Fiber.finished)
228 el.yield(null, .{ .register_awaiter = &detached_fiber.awaiter });
229 detached_fiber.recycle();
230 }
231 }
232
222233 for (el.threads.allocated[0..active_threads]) |*thread| {
223234 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
224235 assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async
......@@ -336,6 +347,7 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
336347 .current_context = &new_thread.idle_context,
337348 .ready_queue = ready_queue.head,
338349 .free_queue = null,
350 .detached_queue = null,
339351 .io_uring = IoUring.init(io_uring_entries, 0) catch |err| {
340352 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
341353 // no more access to `thread` after giving up reservation
......@@ -470,6 +482,7 @@ const SwitchMessage = struct {
470482 const PendingTask = union(enum) {
471483 nothing,
472484 reschedule,
485 recycle: *Fiber,
473486 register_awaiter: *?*Fiber,
474487 lock_mutex: struct {
475488 prev_state: Io.Mutex.State,
......@@ -488,6 +501,9 @@ const SwitchMessage = struct {
488501 assert(prev_fiber.queue_next == null);
489502 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
490503 },
504 .recycle => |fiber| {
505 fiber.recycle();
506 },
491507 .register_awaiter => |awaiter| {
492508 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
493509 assert(prev_fiber.queue_next == null);
......@@ -612,6 +628,18 @@ fn fiberEntry() callconv(.naked) void {
612628 }
613629}
614630
631fn fiberEntryDetached() callconv(.naked) void {
632 switch (builtin.cpu.arch) {
633 .x86_64 => asm volatile (
634 \\ leaq 8(%%rsp), %%rdi
635 \\ jmp %[DetachedClosure_call:P]
636 :
637 : [DetachedClosure_call] "X" (&DetachedClosure.call),
638 ),
639 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
640 }
641}
642
615643const AsyncClosure = struct {
616644 event_loop: *EventLoop,
617645 fiber: *Fiber,
......@@ -632,6 +660,31 @@ const AsyncClosure = struct {
632660 }
633661};
634662
663const DetachedClosure = struct {
664 event_loop: *EventLoop,
665 fiber: *Fiber,
666 start: *const fn (context: *const anyopaque) void,
667
668 fn contextPointer(closure: *DetachedClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
669 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(DetachedClosure));
670 }
671
672 fn call(closure: *DetachedClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(DetachedClosure))) noreturn {
673 message.handle(closure.event_loop);
674 std.log.debug("{*} performing async detached", .{closure.fiber});
675 closure.start(closure.contextPointer());
676 const current_thread: *Thread = .current();
677 current_thread.detached_queue = closure.fiber.queue_next;
678 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
679 if (awaiter) |a| {
680 closure.event_loop.yield(a, .nothing);
681 } else {
682 closure.event_loop.yield(null, .{ .recycle = closure.fiber });
683 }
684 unreachable; // switched to dead fiber
685 }
686};
687
635688fn @"async"(
636689 userdata: ?*anyopaque,
637690 result: []u8,
......@@ -682,6 +735,53 @@ fn @"async"(
682735 return @ptrCast(fiber);
683736}
684737
738fn go(
739 userdata: ?*anyopaque,
740 context: []const u8,
741 context_alignment: std.mem.Alignment,
742 start: *const fn (context: *const anyopaque) void,
743) void {
744 assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO
745 assert(context.len <= Fiber.max_context_size); // TODO
746
747 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
748 const fiber = Fiber.allocate(event_loop) catch {
749 start(context.ptr);
750 return;
751 };
752 std.log.debug("allocated {*}", .{fiber});
753
754 const current_thread: *Thread = .current();
755 const closure: *DetachedClosure = @ptrFromInt(Fiber.max_context_align.max(.of(DetachedClosure)).backward(
756 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
757 ) - @sizeOf(DetachedClosure));
758 fiber.* = .{
759 .required_align = {},
760 .context = switch (builtin.cpu.arch) {
761 .x86_64 => .{
762 .rsp = @intFromPtr(closure) - @sizeOf(usize),
763 .rbp = 0,
764 .rip = @intFromPtr(&fiberEntryDetached),
765 },
766 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
767 },
768 .awaiter = null,
769 .queue_next = current_thread.detached_queue,
770 .cancel_thread = null,
771 .awaiting_completions = .initEmpty(),
772 };
773 current_thread.detached_queue = fiber;
774 closure.* = .{
775 .event_loop = event_loop,
776 .fiber = fiber,
777 .start = start,
778 };
779 @memcpy(closure.contextPointer(), context);
780
781 event_loop.schedule(current_thread, .{ .head = fiber, .tail = fiber });
782}
783
784
685785fn @"await"(
686786 userdata: ?*anyopaque,
687787 any_future: *std.Io.AnyFuture,
......@@ -690,24 +790,12 @@ fn @"await"(
690790) void {
691791 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
692792 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
693 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
793 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished)
794 event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
694795 @memcpy(result, future_fiber.resultBytes(result_alignment));
695796 future_fiber.recycle();
696797}
697798
698fn go(
699 userdata: ?*anyopaque,
700 context: []const u8,
701 context_alignment: std.mem.Alignment,
702 start: *const fn (context: *const anyopaque) void,
703) void {
704 _ = userdata;
705 _ = context;
706 _ = context_alignment;
707 _ = start;
708 @panic("TODO");
709}
710
711799fn cancel(
712800 userdata: ?*anyopaque,
713801 any_future: *std.Io.AnyFuture,