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-07-20 10:38:39-07:00
logf3553049cb72a5ea44667e450a2ff1bf173a6b6d
treee34d681165ad71466afda1dbf8f5c444c13c32a5
parentebf92042e3a081ce84668a7edc1aa74c9ad7e9e5

Io: implement faster mutex

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

lib/std/Io.zig+62-7
......@@ -979,8 +979,8 @@ pub const VTable = struct {
979979 /// Thread-safe.
980980 cancelRequested: *const fn (?*anyopaque) bool,
981981
982 mutexLock: *const fn (?*anyopaque, mutex: *Mutex) void,
983 mutexUnlock: *const fn (?*anyopaque, mutex: *Mutex) void,
982 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) error{Canceled}!void,
983 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,
986986 conditionWake: *const fn (?*anyopaque, cond: *Condition, notify: Condition.Notify) void,
......@@ -1059,8 +1059,63 @@ pub fn Future(Result: type) type {
10591059 };
10601060}
10611061
1062pub const Mutex = struct {
1063 state: std.atomic.Value(u32) = std.atomic.Value(u32).init(unlocked),
1062pub const Mutex = if (true) struct {
1063 state: State,
1064
1065 pub const State = enum(usize) {
1066 locked_once = 0b00,
1067 unlocked = 0b01,
1068 contended = 0b10,
1069 /// contended
1070 _,
1071
1072 pub fn isUnlocked(state: State) bool {
1073 return @intFromEnum(state) & @intFromEnum(State.unlocked) == @intFromEnum(State.unlocked);
1074 }
1075 };
1076
1077 pub const init: Mutex = .{ .state = .unlocked };
1078
1079 pub fn tryLock(mutex: *Mutex) bool {
1080 const prev_state: State = @enumFromInt(@atomicRmw(
1081 usize,
1082 @as(*usize, @ptrCast(&mutex.state)),
1083 .And,
1084 ~@intFromEnum(State.unlocked),
1085 .acquire,
1086 ));
1087 return prev_state.isUnlocked();
1088 }
1089
1090 pub fn lock(mutex: *Mutex, io: std.Io) error{Canceled}!void {
1091 const prev_state: State = @enumFromInt(@atomicRmw(
1092 usize,
1093 @as(*usize, @ptrCast(&mutex.state)),
1094 .And,
1095 ~@intFromEnum(State.unlocked),
1096 .acquire,
1097 ));
1098 if (prev_state.isUnlocked()) {
1099 @branchHint(.likely);
1100 return;
1101 }
1102 return io.vtable.mutexLock(io.userdata, prev_state, mutex);
1103 }
1104
1105 pub fn unlock(mutex: *Mutex, io: std.Io) void {
1106 const prev_state = @cmpxchgWeak(State, &mutex.state, .locked_once, .unlocked, .release, .acquire) orelse {
1107 @branchHint(.likely);
1108 return;
1109 };
1110 std.debug.assert(prev_state != .unlocked); // mutex not locked
1111 return io.vtable.mutexUnlock(io.userdata, prev_state, mutex);
1112 }
1113} else struct {
1114 state: std.atomic.Value(u32),
1115
1116 pub const State = void;
1117
1118 pub const init: Mutex = .{ .state = .init(unlocked) };
10641119
10651120 pub const unlocked: u32 = 0b00;
10661121 pub const locked: u32 = 0b01;
......@@ -1081,15 +1136,15 @@ pub const Mutex = struct {
10811136 }
10821137
10831138 /// Avoids the vtable for uncontended locks.
1084 pub fn lock(m: *Mutex, io: Io) void {
1139 pub fn lock(m: *Mutex, io: Io) error{Canceled}!void {
10851140 if (!m.tryLock()) {
10861141 @branchHint(.unlikely);
1087 io.vtable.mutexLock(io.userdata, m);
1142 try io.vtable.mutexLock(io.userdata, {}, m);
10881143 }
10891144 }
10901145
10911146 pub fn unlock(m: *Mutex, io: Io) void {
1092 io.vtable.mutexUnlock(io.userdata, m);
1147 io.vtable.mutexUnlock(io.userdata, {}, m);
10931148 }
10941149};
10951150
lib/std/Io/EventLoop.zig+184-55
......@@ -10,7 +10,7 @@ const IoUring = std.os.linux.IoUring;
1010/// Must be a thread-safe allocator.
1111gpa: Allocator,
1212mutex: std.Thread.Mutex,
13main_fiber: Fiber,
13main_fiber_buffer: [@sizeOf(Fiber) + Fiber.max_result_size]u8 align(@alignOf(Fiber)),
1414threads: Thread.List,
1515
1616/// Empirically saw >128KB being used by the self-hosted backend to panic.
......@@ -51,10 +51,12 @@ const Thread = struct {
5151};
5252
5353const Fiber = struct {
54 required_align: void align(4),
5455 context: Context,
5556 awaiter: ?*Fiber,
5657 queue_next: ?*Fiber,
5758 cancel_thread: ?*Thread,
59 awaiting_completions: std.StaticBitSet(3),
5860
5961 const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
6062
......@@ -131,7 +133,7 @@ const Fiber = struct {
131133 const thread: *Thread = .current();
132134 std.log.debug("recyling {*}", .{fiber});
133135 assert(fiber.queue_next == null);
134 @memset(fiber.allocatedSlice(), undefined);
136 //@memset(fiber.allocatedSlice(), undefined); // (race)
135137 fiber.queue_next = thread.free_queue;
136138 thread.free_queue = fiber;
137139 }
......@@ -145,10 +147,17 @@ pub fn io(el: *EventLoop) Io {
145147 .vtable = &.{
146148 .@"async" = @"async",
147149 .@"await" = @"await",
150 .go = go,
148151
149152 .cancel = cancel,
150153 .cancelRequested = cancelRequested,
151154
155 .mutexLock = mutexLock,
156 .mutexUnlock = mutexUnlock,
157
158 .conditionWait = conditionWait,
159 .conditionWake = conditionWake,
160
152161 .createFile = createFile,
153162 .openFile = openFile,
154163 .closeFile = closeFile,
......@@ -169,18 +178,22 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
169178 el.* = .{
170179 .gpa = gpa,
171180 .mutex = .{},
172 .main_fiber = .{
173 .context = undefined,
174 .awaiter = null,
175 .queue_next = null,
176 .cancel_thread = null,
177 },
181 .main_fiber_buffer = undefined,
178182 .threads = .{
179183 .allocated = @ptrCast(allocated_slice[0..threads_size]),
180184 .reserved = 1,
181185 .active = 1,
182186 },
183187 };
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 };
184197 const main_thread = &el.threads.allocated[0];
185198 Thread.self = main_thread;
186199 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 {
192205 .rbp = 0,
193206 .rip = @intFromPtr(&mainIdleEntry),
194207 },
195 .current_context = &el.main_fiber.context,
208 .current_context = &main_fiber.context,
196209 .ready_queue = null,
197210 .free_queue = null,
198211 .io_uring = try IoUring.init(io_uring_entries, 0),
......@@ -201,53 +214,57 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
201214 };
202215 errdefer main_thread.io_uring.deinit();
203216 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});
205218}
206219
207220pub fn deinit(el: *EventLoop) void {
208221 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
209 for (el.threads.allocated[0..active_threads]) |*thread|
210 assert(@atomicLoad(?*Fiber, &thread.ready_queue, .acquire) == null); // pending async
222 for (el.threads.allocated[0..active_threads]) |*thread| {
223 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
224 assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async
225 }
211226 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();
212230 for (el.threads.allocated[0..active_threads]) |*thread| while (thread.free_queue) |free_fiber| {
213231 thread.free_queue = free_fiber.queue_next;
214232 free_fiber.queue_next = null;
215233 el.gpa.free(free_fiber.allocatedSlice());
216234 };
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();
220235 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
221236 el.* = undefined;
222237}
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
224262fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
225263 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|
227265 &ready_fiber.context
228 else if (thread.ready_queue) |ready_fiber| ready_context: {
229 thread.ready_queue = ready_fiber.queue_next;
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 };
266 else
267 &thread.idle_context;
251268 const message: SwitchMessage = .{
252269 .contexts = .{
253270 .prev = thread.current_context,
......@@ -270,10 +287,10 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
270287 }
271288 // shared fields of previous `Thread` must be initialized before later ones are marked as active
272289 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)) |_| {
274291 defer thread.idle_search_index += 1;
275292 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];
277294 if (idle_search_thread == thread) continue;
278295 if (@cmpxchgWeak(
279296 ?*Fiber,
......@@ -325,8 +342,8 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
325342 std.log.warn("unable to create worker thread due to io_uring init failure: {s}", .{@errorName(err)});
326343 break :spawn_thread;
327344 },
328 .idle_search_index = next_thread_index,
329 .steal_ready_search_index = next_thread_index,
345 .idle_search_index = 0,
346 .steal_ready_search_index = 0,
330347 };
331348 new_thread.thread = std.Thread.spawn(.{
332349 .stack_size = idle_stack_size,
......@@ -357,7 +374,7 @@ fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAl
357374 message.handle(el);
358375 const thread: *Thread = &el.threads.allocated[0];
359376 el.idle(thread);
360 el.yield(&el.main_fiber, .nothing);
377 el.yield(@ptrCast(&el.main_fiber_buffer), .nothing);
361378 unreachable; // switched to dead fiber
362379}
363380
......@@ -384,8 +401,10 @@ const Completion = struct {
384401fn idle(el: *EventLoop, thread: *Thread) void {
385402 var maybe_ready_fiber: ?*Fiber = null;
386403 while (true) {
387 el.yield(maybe_ready_fiber, .nothing);
388 maybe_ready_fiber = null;
404 while (maybe_ready_fiber orelse el.findReadyFiber(thread)) |ready_fiber| {
405 el.yield(ready_fiber, .nothing);
406 maybe_ready_fiber = null;
407 }
389408 _ = thread.io_uring.submit_and_wait(1) catch |err| switch (err) {
390409 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
391410 else => |e| @panic(@errorName(e)),
......@@ -450,7 +469,12 @@ const SwitchMessage = struct {
450469
451470 const PendingTask = union(enum) {
452471 nothing,
472 reschedule,
453473 register_awaiter: *?*Fiber,
474 lock_mutex: struct {
475 prev_state: Io.Mutex.State,
476 mutex: *Io.Mutex,
477 },
454478 exit,
455479 };
456480
......@@ -459,8 +483,14 @@ const SwitchMessage = struct {
459483 thread.current_context = message.contexts.ready;
460484 switch (message.pending_task) {
461485 .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 },
462491 .register_awaiter => |awaiter| {
463492 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
493 assert(prev_fiber.queue_next == null);
464494 if (@atomicRmw(
465495 ?*Fiber,
466496 awaiter,
......@@ -469,6 +499,36 @@ const SwitchMessage = struct {
469499 .acq_rel,
470500 ) == Fiber.finished) el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
471501 },
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 },
472532 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {
473533 getSqe(&thread.io_uring).* = .{
474534 .opcode = .MSG_RING,
......@@ -590,13 +650,13 @@ fn @"async"(
590650 start(context.ptr, result.ptr);
591651 return null;
592652 };
593 errdefer fiber.recycle();
594653 std.log.debug("allocated {*}", .{fiber});
595654
596655 const closure: *AsyncClosure = @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
597656 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
598657 ) - @sizeOf(AsyncClosure));
599658 fiber.* = .{
659 .required_align = {},
600660 .context = switch (builtin.cpu.arch) {
601661 .x86_64 => .{
602662 .rsp = @intFromPtr(closure) - @sizeOf(usize),
......@@ -608,6 +668,7 @@ fn @"async"(
608668 .awaiter = null,
609669 .queue_next = null,
610670 .cancel_thread = null,
671 .awaiting_completions = .initEmpty(),
611672 };
612673 closure.* = .{
613674 .event_loop = event_loop,
......@@ -634,6 +695,19 @@ fn @"await"(
634695 future_fiber.recycle();
635696}
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
637711fn cancel(
638712 userdata: ?*anyopaque,
639713 any_future: *std.Io.AnyFuture,
......@@ -673,7 +747,7 @@ fn cancelRequested(userdata: ?*anyopaque) bool {
673747 return @atomicLoad(?*Thread, &Thread.current().currentFiber().cancel_thread, .acquire) == Thread.canceling;
674748}
675749
676pub fn createFile(
750fn createFile(
677751 userdata: ?*anyopaque,
678752 dir: std.fs.Dir,
679753 sub_path: []const u8,
......@@ -775,7 +849,7 @@ pub fn createFile(
775849 }
776850}
777851
778pub fn openFile(
852fn openFile(
779853 userdata: ?*anyopaque,
780854 dir: std.fs.Dir,
781855 sub_path: []const u8,
......@@ -883,7 +957,7 @@ pub fn openFile(
883957 }
884958}
885959
886pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
960fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
887961 const el: *EventLoop = @alignCast(@ptrCast(userdata));
888962 const thread: *Thread = .current();
889963 const iou = &thread.io_uring;
......@@ -919,7 +993,7 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
919993 }
920994}
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 {
923997 const el: *EventLoop = @alignCast(@ptrCast(userdata));
924998 const thread: *Thread = .current();
925999 const iou = &thread.io_uring;
......@@ -971,7 +1045,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std
9711045 }
9721046}
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 {
9751049 const el: *EventLoop = @alignCast(@ptrCast(userdata));
9761050 const thread: *Thread = .current();
9771051 const iou = &thread.io_uring;
......@@ -1027,13 +1101,13 @@ pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offs
10271101 }
10281102}
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 {
10311105 _ = userdata;
10321106 const timespec = try std.posix.clock_gettime(clockid);
10331107 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
10341108}
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 {
10371111 const el: *EventLoop = @alignCast(@ptrCast(userdata));
10381112 const thread: *Thread = .current();
10391113 const iou = &thread.io_uring;
......@@ -1086,10 +1160,65 @@ pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.D
10861160 }
10871161}
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
10891212fn errno(signed: i32) std.os.linux.E {
10901213 return .init(@bitCast(@as(isize, signed)));
10911214}
10921215
10931216fn 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 };
10951224}
lib/std/Thread/Pool.zig+23-48
......@@ -335,8 +335,10 @@ pub fn io(pool: *Pool) Io {
335335 .go = go,
336336 .cancel = cancel,
337337 .cancelRequested = cancelRequested,
338
338339 .mutexLock = mutexLock,
339340 .mutexUnlock = mutexUnlock,
341
340342 .conditionWait = conditionWait,
341343 .conditionWake = conditionWake,
342344
......@@ -594,53 +596,26 @@ fn checkCancel(pool: *Pool) error{Canceled}!void {
594596 if (cancelRequested(pool)) return error.Canceled;
595597}
596598
597fn mutexLock(userdata: ?*anyopaque, m: *Io.Mutex) void {
598 @branchHint(.cold);
599 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
600 _ = pool;
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);
599fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) error{Canceled}!void {
600 _ = userdata;
601 if (prev_state == .contended) {
602 std.Thread.Futex.wait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
619603 }
620}
621
622fn mutexUnlock(userdata: ?*anyopaque, m: *Io.Mutex) void {
623 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
624 _ = pool;
625 // Needs to also wake up a waiting thread if any.
626 //
627 // A waiting thread will acquire with `contended` instead of `locked`
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);
604 while (@atomicRmw(
605 Io.Mutex.State,
606 &mutex.state,
607 .Xchg,
608 .contended,
609 .acquire,
610 ) != .unlocked) {
611 std.Thread.Futex.wait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
637612 }
638613}
639
640fn mutexLockInternal(pool: *std.Thread.Pool, m: *Io.Mutex) void {
641 if (!m.tryLock()) {
642 @branchHint(.unlikely);
643 mutexLock(pool, m);
614fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {
615 _ = userdata;
616 _ = prev_state;
617 if (@atomicRmw(Io.Mutex.State, &mutex.state, .Xchg, .unlocked, .release) == .contended) {
618 std.Thread.Futex.wake(@ptrCast(&mutex.state), 1);
644619 }
645620}
646621
......@@ -674,8 +649,8 @@ fn conditionWait(
674649 assert(state & waiter_mask != waiter_mask);
675650 state += one_waiter;
676651
677 mutexUnlock(pool, mutex);
678 defer mutexLockInternal(pool, mutex);
652 mutex.unlock(pool.io());
653 defer mutex.lock(pool.io()) catch @panic("TODO");
679654
680655 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:
808783 };
809784}
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 {
812787 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
813788 try pool.checkCancel();
814789 const timespec = try std.posix.clock_gettime(clockid);
815790 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
816791}
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 {
819794 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
820795 const deadline_nanoseconds: i96 = switch (deadline) {
821796 .nanoseconds => |nanoseconds| nanoseconds,