authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-03-31 08:06:20-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log0f083f24ff9ed532576cb6e4a476bfdb57427c35
treeca9f06498cc2828fd23248700028847a88626eef
parent0086d315f53a76c028d3072240ebb3a3b9757a04

Io: implement faster mutex

3 files changed, 269 insertions(+), 110 deletions(-)

lib/std/Io.zig+62-7
...@@ -626,8 +626,8 @@ pub const VTable = struct {...@@ -626,8 +626,8 @@ 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, mutex: *Mutex) void,629 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) error{Canceled}!void,
630 mutexUnlock: *const fn (?*anyopaque, 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,
633 conditionWake: *const fn (?*anyopaque, cond: *Condition, notify: Condition.Notify) void,633 conditionWake: *const fn (?*anyopaque, cond: *Condition, notify: Condition.Notify) void,
...@@ -706,8 +706,63 @@ pub fn Future(Result: type) type {...@@ -706,8 +706,63 @@ pub fn Future(Result: type) type {
706 };706 };
707}707}
708708
709pub const Mutex = struct {709pub const Mutex = if (true) struct {
710 state: std.atomic.Value(u32) = std.atomic.Value(u32).init(unlocked),710 state: State,
711
712 pub const State = enum(usize) {
713 locked_once = 0b00,
714 unlocked = 0b01,
715 contended = 0b10,
716 /// contended
717 _,
718
719 pub fn isUnlocked(state: State) bool {
720 return @intFromEnum(state) & @intFromEnum(State.unlocked) == @intFromEnum(State.unlocked);
721 }
722 };
723
724 pub const init: Mutex = .{ .state = .unlocked };
725
726 pub fn tryLock(mutex: *Mutex) bool {
727 const prev_state: State = @enumFromInt(@atomicRmw(
728 usize,
729 @as(*usize, @ptrCast(&mutex.state)),
730 .And,
731 ~@intFromEnum(State.unlocked),
732 .acquire,
733 ));
734 return prev_state.isUnlocked();
735 }
736
737 pub fn lock(mutex: *Mutex, io: std.Io) error{Canceled}!void {
738 const prev_state: State = @enumFromInt(@atomicRmw(
739 usize,
740 @as(*usize, @ptrCast(&mutex.state)),
741 .And,
742 ~@intFromEnum(State.unlocked),
743 .acquire,
744 ));
745 if (prev_state.isUnlocked()) {
746 @branchHint(.likely);
747 return;
748 }
749 return io.vtable.mutexLock(io.userdata, prev_state, mutex);
750 }
751
752 pub fn unlock(mutex: *Mutex, io: std.Io) void {
753 const prev_state = @cmpxchgWeak(State, &mutex.state, .locked_once, .unlocked, .release, .acquire) orelse {
754 @branchHint(.likely);
755 return;
756 };
757 std.debug.assert(prev_state != .unlocked); // mutex not locked
758 return io.vtable.mutexUnlock(io.userdata, prev_state, mutex);
759 }
760} else struct {
761 state: std.atomic.Value(u32),
762
763 pub const State = void;
764
765 pub const init: Mutex = .{ .state = .init(unlocked) };
711766
712 pub const unlocked: u32 = 0b00;767 pub const unlocked: u32 = 0b00;
713 pub const locked: u32 = 0b01;768 pub const locked: u32 = 0b01;
...@@ -728,15 +783,15 @@ pub const Mutex = struct {...@@ -728,15 +783,15 @@ pub const Mutex = struct {
728 }783 }
729784
730 /// Avoids the vtable for uncontended locks.785 /// Avoids the vtable for uncontended locks.
731 pub fn lock(m: *Mutex, io: Io) void {786 pub fn lock(m: *Mutex, io: Io) error{Canceled}!void {
732 if (!m.tryLock()) {787 if (!m.tryLock()) {
733 @branchHint(.unlikely);788 @branchHint(.unlikely);
734 io.vtable.mutexLock(io.userdata, m);789 try io.vtable.mutexLock(io.userdata, {}, m);
735 }790 }
736 }791 }
737792
738 pub fn unlock(m: *Mutex, io: Io) void {793 pub fn unlock(m: *Mutex, io: Io) void {
739 io.vtable.mutexUnlock(io.userdata, m);794 io.vtable.mutexUnlock(io.userdata, {}, m);
740 }795 }
741};796};
742797
lib/std/Io/EventLoop.zig+184-55
...@@ -10,7 +10,7 @@ const IoUring = std.os.linux.IoUring;...@@ -10,7 +10,7 @@ const IoUring = std.os.linux.IoUring;
10/// Must be a thread-safe allocator.10/// Must be a thread-safe allocator.
11gpa: Allocator,11gpa: Allocator,
12mutex: std.Thread.Mutex,12mutex: std.Thread.Mutex,
13main_fiber: Fiber,13main_fiber_buffer: [@sizeOf(Fiber) + Fiber.max_result_size]u8 align(@alignOf(Fiber)),
14threads: Thread.List,14threads: Thread.List,
1515
16/// Empirically saw >128KB being used by the self-hosted backend to panic.16/// Empirically saw >128KB being used by the self-hosted backend to panic.
...@@ -51,10 +51,12 @@ const Thread = struct {...@@ -51,10 +51,12 @@ const Thread = struct {
51};51};
5252
53const Fiber = struct {53const Fiber = struct {
54 required_align: void align(4),
54 context: Context,55 context: Context,
55 awaiter: ?*Fiber,56 awaiter: ?*Fiber,
56 queue_next: ?*Fiber,57 queue_next: ?*Fiber,
57 cancel_thread: ?*Thread,58 cancel_thread: ?*Thread,
59 awaiting_completions: std.StaticBitSet(3),
5860
59 const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));61 const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
6062
...@@ -131,7 +133,7 @@ const Fiber = struct {...@@ -131,7 +133,7 @@ const Fiber = struct {
131 const thread: *Thread = .current();133 const thread: *Thread = .current();
132 std.log.debug("recyling {*}", .{fiber});134 std.log.debug("recyling {*}", .{fiber});
133 assert(fiber.queue_next == null);135 assert(fiber.queue_next == null);
134 @memset(fiber.allocatedSlice(), undefined);136 //@memset(fiber.allocatedSlice(), undefined); // (race)
135 fiber.queue_next = thread.free_queue;137 fiber.queue_next = thread.free_queue;
136 thread.free_queue = fiber;138 thread.free_queue = fiber;
137 }139 }
...@@ -145,10 +147,17 @@ pub fn io(el: *EventLoop) Io {...@@ -145,10 +147,17 @@ pub fn io(el: *EventLoop) Io {
145 .vtable = &.{147 .vtable = &.{
146 .@"async" = @"async",148 .@"async" = @"async",
147 .@"await" = @"await",149 .@"await" = @"await",
150 .go = go,
148151
149 .cancel = cancel,152 .cancel = cancel,
150 .cancelRequested = cancelRequested,153 .cancelRequested = cancelRequested,
151154
155 .mutexLock = mutexLock,
156 .mutexUnlock = mutexUnlock,
157
158 .conditionWait = conditionWait,
159 .conditionWake = conditionWake,
160
152 .createFile = createFile,161 .createFile = createFile,
153 .openFile = openFile,162 .openFile = openFile,
154 .closeFile = closeFile,163 .closeFile = closeFile,
...@@ -169,18 +178,22 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -169,18 +178,22 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
169 el.* = .{178 el.* = .{
170 .gpa = gpa,179 .gpa = gpa,
171 .mutex = .{},180 .mutex = .{},
172 .main_fiber = .{181 .main_fiber_buffer = undefined,
173 .context = undefined,
174 .awaiter = null,
175 .queue_next = null,
176 .cancel_thread = null,
177 },
178 .threads = .{182 .threads = .{
179 .allocated = @ptrCast(allocated_slice[0..threads_size]),183 .allocated = @ptrCast(allocated_slice[0..threads_size]),
180 .reserved = 1,184 .reserved = 1,
181 .active = 1,185 .active = 1,
182 },186 },
183 };187 };
188 const main_fiber: *Fiber = @ptrCast(&el.main_fiber_buffer);
189 main_fiber.* = .{
190 .required_align = {},
191 .context = undefined,
192 .awaiter = null,
193 .queue_next = null,
194 .cancel_thread = null,
195 .awaiting_completions = .initEmpty(),
196 };
184 const main_thread = &el.threads.allocated[0];197 const main_thread = &el.threads.allocated[0];
185 Thread.self = main_thread;198 Thread.self = main_thread;
186 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));199 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
...@@ -192,7 +205,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -192,7 +205,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
192 .rbp = 0,205 .rbp = 0,
193 .rip = @intFromPtr(&mainIdleEntry),206 .rip = @intFromPtr(&mainIdleEntry),
194 },207 },
195 .current_context = &el.main_fiber.context,208 .current_context = &main_fiber.context,
196 .ready_queue = null,209 .ready_queue = null,
197 .free_queue = null,210 .free_queue = null,
198 .io_uring = try IoUring.init(io_uring_entries, 0),211 .io_uring = try IoUring.init(io_uring_entries, 0),
...@@ -201,53 +214,57 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -201,53 +214,57 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
201 };214 };
202 errdefer main_thread.io_uring.deinit();215 errdefer main_thread.io_uring.deinit();
203 std.log.debug("created main idle {*}", .{&main_thread.idle_context});216 std.log.debug("created main idle {*}", .{&main_thread.idle_context});
204 std.log.debug("created main {*}", .{&el.main_fiber});217 std.log.debug("created main {*}", .{main_fiber});
205}218}
206219
207pub fn deinit(el: *EventLoop) void {220pub fn deinit(el: *EventLoop) void {
208 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);221 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
209 for (el.threads.allocated[0..active_threads]) |*thread|222 for (el.threads.allocated[0..active_threads]) |*thread| {
210 assert(@atomicLoad(?*Fiber, &thread.ready_queue, .acquire) == null); // pending async223 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
224 assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async
225 }
211 el.yield(null, .exit);226 el.yield(null, .exit);
227 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.allocated.ptr));
228 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
229 for (el.threads.allocated[1..active_threads]) |*thread| thread.thread.join();
212 for (el.threads.allocated[0..active_threads]) |*thread| while (thread.free_queue) |free_fiber| {230 for (el.threads.allocated[0..active_threads]) |*thread| while (thread.free_queue) |free_fiber| {
213 thread.free_queue = free_fiber.queue_next;231 thread.free_queue = free_fiber.queue_next;
214 free_fiber.queue_next = null;232 free_fiber.queue_next = null;
215 el.gpa.free(free_fiber.allocatedSlice());233 el.gpa.free(free_fiber.allocatedSlice());
216 };234 };
217 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.allocated.ptr));
218 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
219 for (el.threads.allocated[1..active_threads]) |thread| thread.thread.join();
220 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);235 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
221 el.* = undefined;236 el.* = undefined;
222}237}
223238
239fn findReadyFiber(el: *EventLoop, thread: *Thread) ?*Fiber {
240 if (@atomicRmw(?*Fiber, &thread.ready_queue, .Xchg, Fiber.finished, .acquire)) |ready_fiber| {
241 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
242 ready_fiber.queue_next = null;
243 return ready_fiber;
244 }
245 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
246 for (0..@min(max_steal_ready_search, active_threads)) |_| {
247 defer thread.steal_ready_search_index += 1;
248 if (thread.steal_ready_search_index == active_threads) thread.steal_ready_search_index = 0;
249 const steal_ready_search_thread = &el.threads.allocated[0..active_threads][thread.steal_ready_search_index];
250 if (steal_ready_search_thread == thread) continue;
251 const ready_fiber = @atomicRmw(?*Fiber, &steal_ready_search_thread.ready_queue, .And, Fiber.finished, .acquire) orelse continue;
252 if (ready_fiber == Fiber.finished) continue;
253 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
254 ready_fiber.queue_next = null;
255 return ready_fiber;
256 }
257 // couldn't find anything to do, so we are now open for business
258 @atomicStore(?*Fiber, &thread.ready_queue, null, .monotonic);
259 return null;
260}
261
224fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {262fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
225 const thread: *Thread = .current();263 const thread: *Thread = .current();
226 const ready_context: *Context = if (maybe_ready_fiber) |ready_fiber|264 const ready_context = if (maybe_ready_fiber orelse el.findReadyFiber(thread)) |ready_fiber|
227 &ready_fiber.context265 &ready_fiber.context
228 else if (thread.ready_queue) |ready_fiber| ready_context: {266 else
229 thread.ready_queue = ready_fiber.queue_next;267 &thread.idle_context;
230 ready_fiber.queue_next = null;
231 break :ready_context &ready_fiber.context;
232 } else ready_context: {
233 const ready_threads = @atomicLoad(u32, &el.threads.active, .acquire);
234 break :ready_context for (0..max_steal_ready_search) |_| {
235 defer thread.steal_ready_search_index += 1;
236 if (thread.steal_ready_search_index == ready_threads) thread.steal_ready_search_index = 0;
237 const steal_ready_search_thread = &el.threads.allocated[thread.steal_ready_search_index];
238 if (steal_ready_search_thread == thread) continue;
239 const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
240 if (@cmpxchgWeak(
241 ?*Fiber,
242 &steal_ready_search_thread.ready_queue,
243 ready_fiber,
244 @atomicLoad(?*Fiber, &ready_fiber.queue_next, .acquire),
245 .acq_rel,
246 .monotonic,
247 )) |_| continue;
248 break &ready_fiber.context;
249 } else &thread.idle_context;
250 };
251 const message: SwitchMessage = .{268 const message: SwitchMessage = .{
252 .contexts = .{269 .contexts = .{
253 .prev = thread.current_context,270 .prev = thread.current_context,
...@@ -270,10 +287,10 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {...@@ -270,10 +287,10 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
270 }287 }
271 // shared fields of previous `Thread` must be initialized before later ones are marked as active288 // shared fields of previous `Thread` must be initialized before later ones are marked as active
272 const new_thread_index = @atomicLoad(u32, &el.threads.active, .acquire);289 const new_thread_index = @atomicLoad(u32, &el.threads.active, .acquire);
273 for (0..max_idle_search) |_| {290 for (0..@min(max_idle_search, new_thread_index)) |_| {
274 defer thread.idle_search_index += 1;291 defer thread.idle_search_index += 1;
275 if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;292 if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;
276 const idle_search_thread = &el.threads.allocated[thread.idle_search_index];293 const idle_search_thread = &el.threads.allocated[0..new_thread_index][thread.idle_search_index];
277 if (idle_search_thread == thread) continue;294 if (idle_search_thread == thread) continue;
278 if (@cmpxchgWeak(295 if (@cmpxchgWeak(
279 ?*Fiber,296 ?*Fiber,
...@@ -325,8 +342,8 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {...@@ -325,8 +342,8 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
325 std.log.warn("unable to create worker thread due to io_uring init failure: {s}", .{@errorName(err)});342 std.log.warn("unable to create worker thread due to io_uring init failure: {s}", .{@errorName(err)});
326 break :spawn_thread;343 break :spawn_thread;
327 },344 },
328 .idle_search_index = next_thread_index,345 .idle_search_index = 0,
329 .steal_ready_search_index = next_thread_index,346 .steal_ready_search_index = 0,
330 };347 };
331 new_thread.thread = std.Thread.spawn(.{348 new_thread.thread = std.Thread.spawn(.{
332 .stack_size = idle_stack_size,349 .stack_size = idle_stack_size,
...@@ -357,7 +374,7 @@ fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAl...@@ -357,7 +374,7 @@ fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAl
357 message.handle(el);374 message.handle(el);
358 const thread: *Thread = &el.threads.allocated[0];375 const thread: *Thread = &el.threads.allocated[0];
359 el.idle(thread);376 el.idle(thread);
360 el.yield(&el.main_fiber, .nothing);377 el.yield(@ptrCast(&el.main_fiber_buffer), .nothing);
361 unreachable; // switched to dead fiber378 unreachable; // switched to dead fiber
362}379}
363380
...@@ -384,8 +401,10 @@ const Completion = struct {...@@ -384,8 +401,10 @@ const Completion = struct {
384fn idle(el: *EventLoop, thread: *Thread) void {401fn idle(el: *EventLoop, thread: *Thread) void {
385 var maybe_ready_fiber: ?*Fiber = null;402 var maybe_ready_fiber: ?*Fiber = null;
386 while (true) {403 while (true) {
387 el.yield(maybe_ready_fiber, .nothing);404 while (maybe_ready_fiber orelse el.findReadyFiber(thread)) |ready_fiber| {
388 maybe_ready_fiber = null;405 el.yield(ready_fiber, .nothing);
406 maybe_ready_fiber = null;
407 }
389 _ = thread.io_uring.submit_and_wait(1) catch |err| switch (err) {408 _ = thread.io_uring.submit_and_wait(1) catch |err| switch (err) {
390 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),409 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
391 else => |e| @panic(@errorName(e)),410 else => |e| @panic(@errorName(e)),
...@@ -450,7 +469,12 @@ const SwitchMessage = struct {...@@ -450,7 +469,12 @@ const SwitchMessage = struct {
450469
451 const PendingTask = union(enum) {470 const PendingTask = union(enum) {
452 nothing,471 nothing,
472 reschedule,
453 register_awaiter: *?*Fiber,473 register_awaiter: *?*Fiber,
474 lock_mutex: struct {
475 prev_state: Io.Mutex.State,
476 mutex: *Io.Mutex,
477 },
454 exit,478 exit,
455 };479 };
456480
...@@ -459,8 +483,14 @@ const SwitchMessage = struct {...@@ -459,8 +483,14 @@ const SwitchMessage = struct {
459 thread.current_context = message.contexts.ready;483 thread.current_context = message.contexts.ready;
460 switch (message.pending_task) {484 switch (message.pending_task) {
461 .nothing => {},485 .nothing => {},
486 .reschedule => {
487 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
488 assert(prev_fiber.queue_next == null);
489 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
490 },
462 .register_awaiter => |awaiter| {491 .register_awaiter => |awaiter| {
463 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));492 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
493 assert(prev_fiber.queue_next == null);
464 if (@atomicRmw(494 if (@atomicRmw(
465 ?*Fiber,495 ?*Fiber,
466 awaiter,496 awaiter,
...@@ -469,6 +499,36 @@ const SwitchMessage = struct {...@@ -469,6 +499,36 @@ const SwitchMessage = struct {
469 .acq_rel,499 .acq_rel,
470 ) == Fiber.finished) el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });500 ) == Fiber.finished) el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
471 },501 },
502 .lock_mutex => |lock_mutex| {
503 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
504 assert(prev_fiber.queue_next == null);
505 var prev_state = lock_mutex.prev_state;
506 while (switch (prev_state) {
507 else => next_state: {
508 prev_fiber.queue_next = @ptrFromInt(@intFromEnum(prev_state));
509 break :next_state @cmpxchgWeak(
510 Io.Mutex.State,
511 &lock_mutex.mutex.state,
512 prev_state,
513 @enumFromInt(@intFromPtr(prev_fiber)),
514 .release,
515 .acquire,
516 );
517 },
518 .unlocked => @cmpxchgWeak(
519 Io.Mutex.State,
520 &lock_mutex.mutex.state,
521 .unlocked,
522 .locked_once,
523 .acquire,
524 .acquire,
525 ) orelse {
526 prev_fiber.queue_next = null;
527 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
528 return;
529 },
530 }) |next_state| prev_state = next_state;
531 },
472 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {532 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {
473 getSqe(&thread.io_uring).* = .{533 getSqe(&thread.io_uring).* = .{
474 .opcode = .MSG_RING,534 .opcode = .MSG_RING,
...@@ -590,13 +650,13 @@ fn @"async"(...@@ -590,13 +650,13 @@ fn @"async"(
590 start(context.ptr, result.ptr);650 start(context.ptr, result.ptr);
591 return null;651 return null;
592 };652 };
593 errdefer fiber.recycle();
594 std.log.debug("allocated {*}", .{fiber});653 std.log.debug("allocated {*}", .{fiber});
595654
596 const closure: *AsyncClosure = @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(655 const closure: *AsyncClosure = @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
597 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,656 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
598 ) - @sizeOf(AsyncClosure));657 ) - @sizeOf(AsyncClosure));
599 fiber.* = .{658 fiber.* = .{
659 .required_align = {},
600 .context = switch (builtin.cpu.arch) {660 .context = switch (builtin.cpu.arch) {
601 .x86_64 => .{661 .x86_64 => .{
602 .rsp = @intFromPtr(closure) - @sizeOf(usize),662 .rsp = @intFromPtr(closure) - @sizeOf(usize),
...@@ -608,6 +668,7 @@ fn @"async"(...@@ -608,6 +668,7 @@ fn @"async"(
608 .awaiter = null,668 .awaiter = null,
609 .queue_next = null,669 .queue_next = null,
610 .cancel_thread = null,670 .cancel_thread = null,
671 .awaiting_completions = .initEmpty(),
611 };672 };
612 closure.* = .{673 closure.* = .{
613 .event_loop = event_loop,674 .event_loop = event_loop,
...@@ -634,6 +695,19 @@ fn @"await"(...@@ -634,6 +695,19 @@ fn @"await"(
634 future_fiber.recycle();695 future_fiber.recycle();
635}696}
636697
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
637fn cancel(711fn cancel(
638 userdata: ?*anyopaque,712 userdata: ?*anyopaque,
639 any_future: *std.Io.AnyFuture,713 any_future: *std.Io.AnyFuture,
...@@ -673,7 +747,7 @@ fn cancelRequested(userdata: ?*anyopaque) bool {...@@ -673,7 +747,7 @@ fn cancelRequested(userdata: ?*anyopaque) bool {
673 return @atomicLoad(?*Thread, &Thread.current().currentFiber().cancel_thread, .acquire) == Thread.canceling;747 return @atomicLoad(?*Thread, &Thread.current().currentFiber().cancel_thread, .acquire) == Thread.canceling;
674}748}
675749
676pub fn createFile(750fn createFile(
677 userdata: ?*anyopaque,751 userdata: ?*anyopaque,
678 dir: std.fs.Dir,752 dir: std.fs.Dir,
679 sub_path: []const u8,753 sub_path: []const u8,
...@@ -775,7 +849,7 @@ pub fn createFile(...@@ -775,7 +849,7 @@ pub fn createFile(
775 }849 }
776}850}
777851
778pub fn openFile(852fn openFile(
779 userdata: ?*anyopaque,853 userdata: ?*anyopaque,
780 dir: std.fs.Dir,854 dir: std.fs.Dir,
781 sub_path: []const u8,855 sub_path: []const u8,
...@@ -883,7 +957,7 @@ pub fn openFile(...@@ -883,7 +957,7 @@ pub fn openFile(
883 }957 }
884}958}
885959
886pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {960fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
887 const el: *EventLoop = @alignCast(@ptrCast(userdata));961 const el: *EventLoop = @alignCast(@ptrCast(userdata));
888 const thread: *Thread = .current();962 const thread: *Thread = .current();
889 const iou = &thread.io_uring;963 const iou = &thread.io_uring;
...@@ -919,7 +993,7 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {...@@ -919,7 +993,7 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
919 }993 }
920}994}
921995
922pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {996fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {
923 const el: *EventLoop = @alignCast(@ptrCast(userdata));997 const el: *EventLoop = @alignCast(@ptrCast(userdata));
924 const thread: *Thread = .current();998 const thread: *Thread = .current();
925 const iou = &thread.io_uring;999 const iou = &thread.io_uring;
...@@ -971,7 +1045,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std...@@ -971,7 +1045,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std
971 }1045 }
972}1046}
9731047
974pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {1048fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {
975 const el: *EventLoop = @alignCast(@ptrCast(userdata));1049 const el: *EventLoop = @alignCast(@ptrCast(userdata));
976 const thread: *Thread = .current();1050 const thread: *Thread = .current();
977 const iou = &thread.io_uring;1051 const iou = &thread.io_uring;
...@@ -1027,13 +1101,13 @@ pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offs...@@ -1027,13 +1101,13 @@ pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offs
1027 }1101 }
1028}1102}
10291103
1030pub fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {1104fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
1031 _ = userdata;1105 _ = userdata;
1032 const timespec = try std.posix.clock_gettime(clockid);1106 const timespec = try std.posix.clock_gettime(clockid);
1033 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);1107 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
1034}1108}
10351109
1036pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {1110fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
1037 const el: *EventLoop = @alignCast(@ptrCast(userdata));1111 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1038 const thread: *Thread = .current();1112 const thread: *Thread = .current();
1039 const iou = &thread.io_uring;1113 const iou = &thread.io_uring;
...@@ -1086,10 +1160,65 @@ pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.D...@@ -1086,10 +1160,65 @@ pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.D
1086 }1160 }
1087}1161}
10881162
1163fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) error{Canceled}!void {
1164 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1165 el.yield(null, .{ .lock_mutex = .{
1166 .prev_state = prev_state,
1167 .mutex = mutex,
1168 } });
1169}
1170fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {
1171 var maybe_waiting_fiber: ?*Fiber = @ptrFromInt(@intFromEnum(prev_state));
1172 while (if (maybe_waiting_fiber) |waiting_fiber| @cmpxchgWeak(
1173 Io.Mutex.State,
1174 &mutex.state,
1175 @enumFromInt(@intFromPtr(waiting_fiber)),
1176 @enumFromInt(@intFromPtr(waiting_fiber.queue_next)),
1177 .release,
1178 .acquire,
1179 ) else @cmpxchgWeak(
1180 Io.Mutex.State,
1181 &mutex.state,
1182 .locked_once,
1183 .unlocked,
1184 .release,
1185 .acquire,
1186 ) orelse return) |next_state| maybe_waiting_fiber = @ptrFromInt(@intFromEnum(next_state));
1187 maybe_waiting_fiber.?.queue_next = null;
1188 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1189 el.yield(maybe_waiting_fiber.?, .reschedule);
1190}
1191
1192fn conditionWait(
1193 userdata: ?*anyopaque,
1194 cond: *Io.Condition,
1195 mutex: *Io.Mutex,
1196 timeout: ?u64,
1197) Io.Condition.WaitError!void {
1198 _ = userdata;
1199 _ = cond;
1200 _ = mutex;
1201 _ = timeout;
1202 @panic("TODO");
1203}
1204
1205fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Condition.Notify) void {
1206 _ = userdata;
1207 _ = cond;
1208 _ = notify;
1209 @panic("TODO");
1210}
1211
1089fn errno(signed: i32) std.os.linux.E {1212fn errno(signed: i32) std.os.linux.E {
1090 return .init(@bitCast(@as(isize, signed)));1213 return .init(@bitCast(@as(isize, signed)));
1091}1214}
10921215
1093fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {1216fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
1094 return iou.get_sqe() catch @panic("TODO: handle submission queue full");1217 while (true) return iou.get_sqe() catch {
1218 _ = iou.submit_and_wait(0) catch |err| switch (err) {
1219 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
1220 else => |e| @panic(@errorName(e)),
1221 };
1222 continue;
1223 };
1095}1224}
lib/std/Thread/Pool.zig+23-48
...@@ -335,8 +335,10 @@ pub fn io(pool: *Pool) Io {...@@ -335,8 +335,10 @@ pub fn io(pool: *Pool) Io {
335 .go = go,335 .go = go,
336 .cancel = cancel,336 .cancel = cancel,
337 .cancelRequested = cancelRequested,337 .cancelRequested = cancelRequested,
338
338 .mutexLock = mutexLock,339 .mutexLock = mutexLock,
339 .mutexUnlock = mutexUnlock,340 .mutexUnlock = mutexUnlock,
341
340 .conditionWait = conditionWait,342 .conditionWait = conditionWait,
341 .conditionWake = conditionWake,343 .conditionWake = conditionWake,
342344
...@@ -594,53 +596,26 @@ fn checkCancel(pool: *Pool) error{Canceled}!void {...@@ -594,53 +596,26 @@ fn checkCancel(pool: *Pool) error{Canceled}!void {
594 if (cancelRequested(pool)) return error.Canceled;596 if (cancelRequested(pool)) return error.Canceled;
595}597}
596598
597fn mutexLock(userdata: ?*anyopaque, m: *Io.Mutex) void {599fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) error{Canceled}!void {
598 @branchHint(.cold);600 _ = userdata;
599 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));601 if (prev_state == .contended) {
600 _ = pool;602 std.Thread.Futex.wait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
601
602 // Avoid doing an atomic swap below if we already know the state is contended.
603 // An atomic swap unconditionally stores which marks the cache-line as modified unnecessarily.
604 if (m.state.load(.monotonic) == Io.Mutex.contended) {
605 std.Thread.Futex.wait(&m.state, Io.Mutex.contended);
606 }
607
608 // Try to acquire the lock while also telling the existing lock holder that there are threads waiting.
609 //
610 // Once we sleep on the Futex, we must acquire the mutex using `contended` rather than `locked`.
611 // If not, threads sleeping on the Futex wouldn't see the state change in unlock and potentially deadlock.
612 // The downside is that the last mutex unlocker will see `contended` and do an unnecessary Futex wake
613 // but this is better than having to wake all waiting threads on mutex unlock.
614 //
615 // Acquire barrier ensures grabbing the lock happens before the critical section
616 // and that the previous lock holder's critical section happens before we grab the lock.
617 while (m.state.swap(Io.Mutex.contended, .acquire) != Io.Mutex.unlocked) {
618 std.Thread.Futex.wait(&m.state, Io.Mutex.contended);
619 }603 }
620}604 while (@atomicRmw(
621605 Io.Mutex.State,
622fn mutexUnlock(userdata: ?*anyopaque, m: *Io.Mutex) void {606 &mutex.state,
623 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));607 .Xchg,
624 _ = pool;608 .contended,
625 // Needs to also wake up a waiting thread if any.609 .acquire,
626 //610 ) != .unlocked) {
627 // A waiting thread will acquire with `contended` instead of `locked`611 std.Thread.Futex.wait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
628 // which ensures that it wakes up another thread on the next unlock().
629 //
630 // Release barrier ensures the critical section happens before we let go of the lock
631 // and that our critical section happens before the next lock holder grabs the lock.
632 const state = m.state.swap(Io.Mutex.unlocked, .release);
633 assert(state != Io.Mutex.unlocked);
634
635 if (state == Io.Mutex.contended) {
636 std.Thread.Futex.wake(&m.state, 1);
637 }612 }
638}613}
639614fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {
640fn mutexLockInternal(pool: *std.Thread.Pool, m: *Io.Mutex) void {615 _ = userdata;
641 if (!m.tryLock()) {616 _ = prev_state;
642 @branchHint(.unlikely);617 if (@atomicRmw(Io.Mutex.State, &mutex.state, .Xchg, .unlocked, .release) == .contended) {
643 mutexLock(pool, m);618 std.Thread.Futex.wake(@ptrCast(&mutex.state), 1);
644 }619 }
645}620}
646621
...@@ -674,8 +649,8 @@ fn conditionWait(...@@ -674,8 +649,8 @@ fn conditionWait(
674 assert(state & waiter_mask != waiter_mask);649 assert(state & waiter_mask != waiter_mask);
675 state += one_waiter;650 state += one_waiter;
676651
677 mutexUnlock(pool, mutex);652 mutex.unlock(pool.io());
678 defer mutexLockInternal(pool, mutex);653 defer mutex.lock(pool.io()) catch @panic("TODO");
679654
680 var futex_deadline = std.Thread.Futex.Deadline.init(timeout);655 var futex_deadline = std.Thread.Futex.Deadline.init(timeout);
681656
...@@ -808,14 +783,14 @@ fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset:...@@ -808,14 +783,14 @@ fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset:
808 };783 };
809}784}
810785
811pub fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {786fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
812 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));787 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
813 try pool.checkCancel();788 try pool.checkCancel();
814 const timespec = try std.posix.clock_gettime(clockid);789 const timespec = try std.posix.clock_gettime(clockid);
815 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);790 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
816}791}
817792
818pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {793fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
819 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));794 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
820 const deadline_nanoseconds: i96 = switch (deadline) {795 const deadline_nanoseconds: i96 = switch (deadline) {
821 .nanoseconds => |nanoseconds| nanoseconds,796 .nanoseconds => |nanoseconds| nanoseconds,