| ... | @@ -27,7 +27,6 @@ const Thread = struct { | ... | @@ -27,7 +27,6 @@ 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, | | |
| 31 | io_uring: IoUring, | 30 | io_uring: IoUring, |
| 32 | idle_search_index: u32, | 31 | idle_search_index: u32, |
| 33 | steal_ready_search_index: u32, | 32 | steal_ready_search_index: u32, |
| ... | @@ -209,7 +208,6 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void { | ... | @@ -209,7 +208,6 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void { |
| 209 | .current_context = &main_fiber.context, | 208 | .current_context = &main_fiber.context, |
| 210 | .ready_queue = null, | 209 | .ready_queue = null, |
| 211 | .free_queue = null, | 210 | .free_queue = null, |
| 212 | .detached_queue = null, | | |
| 213 | .io_uring = try IoUring.init(io_uring_entries, 0), | 211 | .io_uring = try IoUring.init(io_uring_entries, 0), |
| 214 | .idle_search_index = 1, | 212 | .idle_search_index = 1, |
| 215 | .steal_ready_search_index = 1, | 213 | .steal_ready_search_index = 1, |
| ... | @@ -220,16 +218,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void { | ... | @@ -220,16 +218,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void { |
| 220 | } | 218 | } |
| 221 | | 219 | |
| 222 | pub fn deinit(el: *EventLoop) void { | 220 | pub fn deinit(el: *EventLoop) void { |
| 223 | // Wait for detached fibers. | | |
| 224 | const active_threads = @atomicLoad(u32, &el.threads.active, .acquire); | 221 | 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 | | | |
| 233 | for (el.threads.allocated[0..active_threads]) |*thread| { | 222 | for (el.threads.allocated[0..active_threads]) |*thread| { |
| 234 | const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic); | 223 | const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic); |
| 235 | assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async | 224 | assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async |
| ... | @@ -347,7 +336,6 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void { | ... | @@ -347,7 +336,6 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void { |
| 347 | .current_context = &new_thread.idle_context, | 336 | .current_context = &new_thread.idle_context, |
| 348 | .ready_queue = ready_queue.head, | 337 | .ready_queue = ready_queue.head, |
| 349 | .free_queue = null, | 338 | .free_queue = null, |
| 350 | .detached_queue = null, | | |
| 351 | .io_uring = IoUring.init(io_uring_entries, 0) catch |err| { | 339 | .io_uring = IoUring.init(io_uring_entries, 0) catch |err| { |
| 352 | @atomicStore(u32, &el.threads.reserved, new_thread_index, .release); | 340 | @atomicStore(u32, &el.threads.reserved, new_thread_index, .release); |
| 353 | // no more access to `thread` after giving up reservation | 341 | // no more access to `thread` after giving up reservation |
| ... | @@ -673,8 +661,6 @@ const DetachedClosure = struct { | ... | @@ -673,8 +661,6 @@ const DetachedClosure = struct { |
| 673 | message.handle(closure.event_loop); | 661 | message.handle(closure.event_loop); |
| 674 | std.log.debug("{*} performing async detached", .{closure.fiber}); | 662 | std.log.debug("{*} performing async detached", .{closure.fiber}); |
| 675 | closure.start(closure.contextPointer()); | 663 | 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); | 664 | const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel); |
| 679 | if (awaiter) |a| { | 665 | if (awaiter) |a| { |
| 680 | closure.event_loop.yield(a, .nothing); | 666 | closure.event_loop.yield(a, .nothing); |
| ... | @@ -766,11 +752,10 @@ fn go( | ... | @@ -766,11 +752,10 @@ fn go( |
| 766 | else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)), | 752 | else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)), |
| 767 | }, | 753 | }, |
| 768 | .awaiter = null, | 754 | .awaiter = null, |
| 769 | .queue_next = current_thread.detached_queue, | 755 | .queue_next = null, |
| 770 | .cancel_thread = null, | 756 | .cancel_thread = null, |
| 771 | .awaiting_completions = .initEmpty(), | 757 | .awaiting_completions = .initEmpty(), |
| 772 | }; | 758 | }; |
| 773 | current_thread.detached_queue = fiber; | | |
| 774 | closure.* = .{ | 759 | closure.* = .{ |
| 775 | .event_loop = event_loop, | 760 | .event_loop = event_loop, |
| 776 | .fiber = fiber, | 761 | .fiber = fiber, |
Andrew Kelley <andrew@ziglang.org> 2025-03-31 17:06:05-07:00
lib