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