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-10-02 16:30:59-07:00
log663611773ca5e6438fab82c4697bac6cc9b8e538
tree089fa894fb8560027085825af916415d26b61a5a
parent0f083f24ff9ed532576cb6e4a476bfdb57427c35

EventLoop: implement detached async

data races on deinit tho

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

lib/std/Io.zig+31-26
...@@ -626,7 +626,7 @@ pub const VTable = struct {...@@ -626,7 +626,7 @@ pub const VTable = struct {
626 /// Thread-safe.626 /// Thread-safe.
627 cancelRequested: *const fn (?*anyopaque) bool,627 cancelRequested: *const fn (?*anyopaque) bool,
628628
629 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) error{Canceled}!void,629 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,
630 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,630 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
631631
632 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex, timeout_ns: ?u64) Condition.WaitError!void,632 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex, timeout_ns: ?u64) Condition.WaitError!void,
...@@ -645,11 +645,11 @@ pub const VTable = struct {...@@ -645,11 +645,11 @@ pub const VTable = struct {
645pub const OpenFlags = fs.File.OpenFlags;645pub const OpenFlags = fs.File.OpenFlags;
646pub const CreateFlags = fs.File.CreateFlags;646pub const CreateFlags = fs.File.CreateFlags;
647647
648pub const FileOpenError = fs.File.OpenError || error{Canceled};648pub const FileOpenError = fs.File.OpenError || Cancelable;
649pub const FileReadError = fs.File.ReadError || error{Canceled};649pub const FileReadError = fs.File.ReadError || Cancelable;
650pub const FilePReadError = fs.File.PReadError || error{Canceled};650pub const FilePReadError = fs.File.PReadError || Cancelable;
651pub const FileWriteError = fs.File.WriteError || error{Canceled};651pub const FileWriteError = fs.File.WriteError || Cancelable;
652pub const FilePWriteError = fs.File.PWriteError || error{Canceled};652pub const FilePWriteError = fs.File.PWriteError || Cancelable;
653653
654pub const Timestamp = enum(i96) {654pub const Timestamp = enum(i96) {
655 _,655 _,
...@@ -666,7 +666,7 @@ pub const Deadline = union(enum) {...@@ -666,7 +666,7 @@ pub const Deadline = union(enum) {
666 nanoseconds: i96,666 nanoseconds: i96,
667 timestamp: Timestamp,667 timestamp: Timestamp,
668};668};
669pub const ClockGetTimeError = std.posix.ClockGetTimeError || error{Canceled};669pub const ClockGetTimeError = std.posix.ClockGetTimeError || Cancelable;
670pub const SleepError = error{ UnsupportedClock, Unexpected, Canceled };670pub const SleepError = error{ UnsupportedClock, Unexpected, Canceled };
671671
672pub const AnyFuture = opaque {};672pub const AnyFuture = opaque {};
...@@ -734,7 +734,7 @@ pub const Mutex = if (true) struct {...@@ -734,7 +734,7 @@ pub const Mutex = if (true) struct {
734 return prev_state.isUnlocked();734 return prev_state.isUnlocked();
735 }735 }
736736
737 pub fn lock(mutex: *Mutex, io: std.Io) error{Canceled}!void {737 pub fn lock(mutex: *Mutex, io: std.Io) Cancelable!void {
738 const prev_state: State = @enumFromInt(@atomicRmw(738 const prev_state: State = @enumFromInt(@atomicRmw(
739 usize,739 usize,
740 @as(*usize, @ptrCast(&mutex.state)),740 @as(*usize, @ptrCast(&mutex.state)),
...@@ -783,7 +783,7 @@ pub const Mutex = if (true) struct {...@@ -783,7 +783,7 @@ pub const Mutex = if (true) struct {
783 }783 }
784784
785 /// Avoids the vtable for uncontended locks.785 /// Avoids the vtable for uncontended locks.
786 pub fn lock(m: *Mutex, io: Io) error{Canceled}!void {786 pub fn lock(m: *Mutex, io: Io) Cancelable!void {
787 if (!m.tryLock()) {787 if (!m.tryLock()) {
788 @branchHint(.unlikely);788 @branchHint(.unlikely);
789 try io.vtable.mutexLock(io.userdata, {}, m);789 try io.vtable.mutexLock(io.userdata, {}, m);
...@@ -809,10 +809,10 @@ pub const Condition = struct {...@@ -809,10 +809,10 @@ pub const Condition = struct {
809 all,809 all,
810 };810 };
811811
812 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) void {812 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {
813 io.vtable.conditionWait(io.userdata, cond, mutex, null) catch |err| switch (err) {813 io.vtable.conditionWait(io.userdata, cond, mutex, null) catch |err| switch (err) {
814 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out814 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out
815 error.Canceled => return, // handled as spurious wakeup815 error.Canceled => return error.Canceled,
816 };816 };
817 }817 }
818818
...@@ -829,6 +829,11 @@ pub const Condition = struct {...@@ -829,6 +829,11 @@ pub const Condition = struct {
829 }829 }
830};830};
831831
832pub const Cancelable = error{
833 /// Caller has requested the async operation to stop.
834 Canceled,
835};
836
832pub const TypeErasedQueue = struct {837pub const TypeErasedQueue = struct {
833 mutex: Mutex,838 mutex: Mutex,
834839
...@@ -852,7 +857,7 @@ pub const TypeErasedQueue = struct {...@@ -852,7 +857,7 @@ pub const TypeErasedQueue = struct {
852857
853 pub fn init(buffer: []u8) TypeErasedQueue {858 pub fn init(buffer: []u8) TypeErasedQueue {
854 return .{859 return .{
855 .mutex = .{},860 .mutex = .init,
856 .buffer = buffer,861 .buffer = buffer,
857 .put_index = 0,862 .put_index = 0,
858 .get_index = 0,863 .get_index = 0,
...@@ -861,10 +866,10 @@ pub const TypeErasedQueue = struct {...@@ -861,10 +866,10 @@ pub const TypeErasedQueue = struct {
861 };866 };
862 }867 }
863868
864 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) usize {869 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) Cancelable!usize {
865 assert(elements.len >= min);870 assert(elements.len >= min);
866871
867 q.mutex.lock(io);872 try q.mutex.lock(io);
868 defer q.mutex.unlock(io);873 defer q.mutex.unlock(io);
869874
870 // Getters have first priority on the data, and only when the getters875 // Getters have first priority on the data, and only when the getters
...@@ -911,15 +916,15 @@ pub const TypeErasedQueue = struct {...@@ -911,15 +916,15 @@ pub const TypeErasedQueue = struct {
911 .data = .{ .remaining = remaining, .condition = .{} },916 .data = .{ .remaining = remaining, .condition = .{} },
912 };917 };
913 q.putters.append(&node);918 q.putters.append(&node);
914 node.data.condition.wait(io, &q.mutex);919 try node.data.condition.wait(io, &q.mutex);
915 remaining = node.data.remaining;920 remaining = node.data.remaining;
916 }921 }
917 }922 }
918923
919 pub fn get(q: *@This(), io: Io, buffer: []u8, min: usize) usize {924 pub fn get(q: *@This(), io: Io, buffer: []u8, min: usize) Cancelable!usize {
920 assert(buffer.len >= min);925 assert(buffer.len >= min);
921926
922 q.mutex.lock(io);927 try q.mutex.lock(io);
923 defer q.mutex.unlock(io);928 defer q.mutex.unlock(io);
924929
925 // The ring buffer gets first priority, then data should come from any930 // The ring buffer gets first priority, then data should come from any
...@@ -976,7 +981,7 @@ pub const TypeErasedQueue = struct {...@@ -976,7 +981,7 @@ pub const TypeErasedQueue = struct {
976 .data = .{ .remaining = remaining, .condition = .{} },981 .data = .{ .remaining = remaining, .condition = .{} },
977 };982 };
978 q.getters.append(&node);983 q.getters.append(&node);
979 node.data.condition.wait(io, &q.mutex);984 try node.data.condition.wait(io, &q.mutex);
980 remaining = node.data.remaining;985 remaining = node.data.remaining;
981 }986 }
982 }987 }
...@@ -1030,8 +1035,8 @@ pub fn Queue(Elem: type) type {...@@ -1030,8 +1035,8 @@ pub fn Queue(Elem: type) type {
1030 /// Returns how many elements have been added to the queue.1035 /// Returns how many elements have been added to the queue.
1031 ///1036 ///
1032 /// Asserts that `elements.len >= min`.1037 /// Asserts that `elements.len >= min`.
1033 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) usize {1038 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) Cancelable!usize {
1034 return @divExact(q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));1039 return @divExact(try q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
1035 }1040 }
10361041
1037 /// Receives elements from the beginning of the queue. The function1042 /// Receives elements from the beginning of the queue. The function
...@@ -1041,17 +1046,17 @@ pub fn Queue(Elem: type) type {...@@ -1041,17 +1046,17 @@ pub fn Queue(Elem: type) type {
1041 /// Returns how many elements of `buffer` have been populated.1046 /// Returns how many elements of `buffer` have been populated.
1042 ///1047 ///
1043 /// Asserts that `buffer.len >= min`.1048 /// Asserts that `buffer.len >= min`.
1044 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) usize {1049 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) Cancelable!usize {
1045 return @divExact(q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));1050 return @divExact(try q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
1046 }1051 }
10471052
1048 pub fn putOne(q: *@This(), io: Io, item: Elem) void {1053 pub fn putOne(q: *@This(), io: Io, item: Elem) Cancelable!void {
1049 assert(q.put(io, &.{item}, 1) == 1);1054 assert(try q.put(io, &.{item}, 1) == 1);
1050 }1055 }
10511056
1052 pub fn getOne(q: *@This(), io: Io) Elem {1057 pub fn getOne(q: *@This(), io: Io) Cancelable!Elem {
1053 var buf: [1]Elem = undefined;1058 var buf: [1]Elem = undefined;
1054 assert(q.get(io, &buf, 1) == 1);1059 assert(try q.get(io, &buf, 1) == 1);
1055 return buf[0];1060 return buf[0];
1056 }1061 }
1057 };1062 };
lib/std/Io/EventLoop.zig+102-14
...@@ -27,6 +27,7 @@ const Thread = struct {...@@ -27,6 +27,7 @@ const Thread = struct {
27 current_context: *Context,27 current_context: *Context,
28 ready_queue: ?*Fiber,28 ready_queue: ?*Fiber,
29 free_queue: ?*Fiber,29 free_queue: ?*Fiber,
30 detached_queue: ?*Fiber,
30 io_uring: IoUring,31 io_uring: IoUring,
31 idle_search_index: u32,32 idle_search_index: u32,
32 steal_ready_search_index: u32,33 steal_ready_search_index: u32,
...@@ -208,6 +209,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -208,6 +209,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
208 .current_context = &main_fiber.context,209 .current_context = &main_fiber.context,
209 .ready_queue = null,210 .ready_queue = null,
210 .free_queue = null,211 .free_queue = null,
212 .detached_queue = null,
211 .io_uring = try IoUring.init(io_uring_entries, 0),213 .io_uring = try IoUring.init(io_uring_entries, 0),
212 .idle_search_index = 1,214 .idle_search_index = 1,
213 .steal_ready_search_index = 1,215 .steal_ready_search_index = 1,
...@@ -218,7 +220,16 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -218,7 +220,16 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
218}220}
219221
220pub fn deinit(el: *EventLoop) void {222pub fn deinit(el: *EventLoop) void {
223 // Wait for detached fibers.
221 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);224 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
222 for (el.threads.allocated[0..active_threads]) |*thread| {233 for (el.threads.allocated[0..active_threads]) |*thread| {
223 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);234 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
224 assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async235 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 {...@@ -336,6 +347,7 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
336 .current_context = &new_thread.idle_context,347 .current_context = &new_thread.idle_context,
337 .ready_queue = ready_queue.head,348 .ready_queue = ready_queue.head,
338 .free_queue = null,349 .free_queue = null,
350 .detached_queue = null,
339 .io_uring = IoUring.init(io_uring_entries, 0) catch |err| {351 .io_uring = IoUring.init(io_uring_entries, 0) catch |err| {
340 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);352 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
341 // no more access to `thread` after giving up reservation353 // no more access to `thread` after giving up reservation
...@@ -470,6 +482,7 @@ const SwitchMessage = struct {...@@ -470,6 +482,7 @@ const SwitchMessage = struct {
470 const PendingTask = union(enum) {482 const PendingTask = union(enum) {
471 nothing,483 nothing,
472 reschedule,484 reschedule,
485 recycle: *Fiber,
473 register_awaiter: *?*Fiber,486 register_awaiter: *?*Fiber,
474 lock_mutex: struct {487 lock_mutex: struct {
475 prev_state: Io.Mutex.State,488 prev_state: Io.Mutex.State,
...@@ -488,6 +501,9 @@ const SwitchMessage = struct {...@@ -488,6 +501,9 @@ const SwitchMessage = struct {
488 assert(prev_fiber.queue_next == null);501 assert(prev_fiber.queue_next == null);
489 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });502 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
490 },503 },
504 .recycle => |fiber| {
505 fiber.recycle();
506 },
491 .register_awaiter => |awaiter| {507 .register_awaiter => |awaiter| {
492 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));508 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
493 assert(prev_fiber.queue_next == null);509 assert(prev_fiber.queue_next == null);
...@@ -612,6 +628,18 @@ fn fiberEntry() callconv(.naked) void {...@@ -612,6 +628,18 @@ fn fiberEntry() callconv(.naked) void {
612 }628 }
613}629}
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
615const AsyncClosure = struct {643const AsyncClosure = struct {
616 event_loop: *EventLoop,644 event_loop: *EventLoop,
617 fiber: *Fiber,645 fiber: *Fiber,
...@@ -632,6 +660,31 @@ const AsyncClosure = struct {...@@ -632,6 +660,31 @@ const AsyncClosure = struct {
632 }660 }
633};661};
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
635fn @"async"(688fn @"async"(
636 userdata: ?*anyopaque,689 userdata: ?*anyopaque,
637 result: []u8,690 result: []u8,
...@@ -682,6 +735,53 @@ fn @"async"(...@@ -682,6 +735,53 @@ fn @"async"(
682 return @ptrCast(fiber);735 return @ptrCast(fiber);
683}736}
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
685fn @"await"(785fn @"await"(
686 userdata: ?*anyopaque,786 userdata: ?*anyopaque,
687 any_future: *std.Io.AnyFuture,787 any_future: *std.Io.AnyFuture,
...@@ -690,24 +790,12 @@ fn @"await"(...@@ -690,24 +790,12 @@ fn @"await"(
690) void {790) void {
691 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));791 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
692 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));792 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 });
694 @memcpy(result, future_fiber.resultBytes(result_alignment));795 @memcpy(result, future_fiber.resultBytes(result_alignment));
695 future_fiber.recycle();796 future_fiber.recycle();
696}797}
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
711fn cancel(799fn cancel(
712 userdata: ?*anyopaque,800 userdata: ?*anyopaque,
713 any_future: *std.Io.AnyFuture,801 any_future: *std.Io.AnyFuture,