authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-30 19:56:30-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:38-07:00
loga7790bd32e1e8caf6f2f0bedede8a7cb7b35c443
tree968158158976a01daf6dccf9bc002ecd60906e5e
parent012ef81b8ba63f757366384a7721647481665762

implement Mutex, Condition, and Queue

3 files changed, 496 insertions(+), 27 deletions(-)

lib/std/Io.zig+310-12
...@@ -1,7 +1,5 @@...@@ -1,7 +1,5 @@
1const std = @import("std.zig");
2const builtin = @import("builtin");1const builtin = @import("builtin");
3const root = @import("root");2const std = @import("std.zig");
4const c = std.c;
5const is_windows = builtin.os.tag == .windows;3const is_windows = builtin.os.tag == .windows;
6const windows = std.os.windows;4const windows = std.os.windows;
7const posix = std.posix;5const posix = std.posix;
...@@ -9,8 +7,6 @@ const math = std.math;...@@ -9,8 +7,6 @@ const math = std.math;
9const assert = std.debug.assert;7const assert = std.debug.assert;
10const fs = std.fs;8const fs = std.fs;
11const mem = std.mem;9const mem = std.mem;
12const meta = std.meta;
13const File = std.fs.File;
14const Allocator = std.mem.Allocator;10const Allocator = std.mem.Allocator;
15const Alignment = std.mem.Alignment;11const Alignment = std.mem.Alignment;
1612
...@@ -972,6 +968,12 @@ pub const VTable = struct {...@@ -972,6 +968,12 @@ pub const VTable = struct {
972 /// Thread-safe.968 /// Thread-safe.
973 cancelRequested: *const fn (?*anyopaque) bool,969 cancelRequested: *const fn (?*anyopaque) bool,
974970
971 mutexLock: *const fn (?*anyopaque, mutex: *Mutex) void,
972 mutexUnlock: *const fn (?*anyopaque, mutex: *Mutex) void,
973
974 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex, timeout_ns: ?u64) Condition.WaitError!void,
975 conditionWake: *const fn (?*anyopaque, cond: *Condition, notify: Condition.Notify) void,
976
975 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,977 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
976 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,978 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
977 closeFile: *const fn (?*anyopaque, fs.File) void,979 closeFile: *const fn (?*anyopaque, fs.File) void,
...@@ -985,11 +987,11 @@ pub const VTable = struct {...@@ -985,11 +987,11 @@ pub const VTable = struct {
985pub const OpenFlags = fs.File.OpenFlags;987pub const OpenFlags = fs.File.OpenFlags;
986pub const CreateFlags = fs.File.CreateFlags;988pub const CreateFlags = fs.File.CreateFlags;
987989
988pub const FileOpenError = fs.File.OpenError || error{AsyncCancel};990pub const FileOpenError = fs.File.OpenError || error{Canceled};
989pub const FileReadError = fs.File.ReadError || error{AsyncCancel};991pub const FileReadError = fs.File.ReadError || error{Canceled};
990pub const FilePReadError = fs.File.PReadError || error{AsyncCancel};992pub const FilePReadError = fs.File.PReadError || error{Canceled};
991pub const FileWriteError = fs.File.WriteError || error{AsyncCancel};993pub const FileWriteError = fs.File.WriteError || error{Canceled};
992pub const FilePWriteError = fs.File.PWriteError || error{AsyncCancel};994pub const FilePWriteError = fs.File.PWriteError || error{Canceled};
993995
994pub const Timestamp = enum(i96) {996pub const Timestamp = enum(i96) {
995 _,997 _,
...@@ -1006,8 +1008,8 @@ pub const Deadline = union(enum) {...@@ -1006,8 +1008,8 @@ pub const Deadline = union(enum) {
1006 nanoseconds: i96,1008 nanoseconds: i96,
1007 timestamp: Timestamp,1009 timestamp: Timestamp,
1008};1010};
1009pub const ClockGetTimeError = std.posix.ClockGetTimeError || error{AsyncCancel};1011pub const ClockGetTimeError = std.posix.ClockGetTimeError || error{Canceled};
1010pub const SleepError = error{ UnsupportedClock, Unexpected, AsyncCancel };1012pub const SleepError = error{ UnsupportedClock, Unexpected, Canceled };
10111013
1012pub const AnyFuture = opaque {};1014pub const AnyFuture = opaque {};
10131015
...@@ -1036,6 +1038,302 @@ pub fn Future(Result: type) type {...@@ -1036,6 +1038,302 @@ pub fn Future(Result: type) type {
1036 };1038 };
1037}1039}
10381040
1041pub const Mutex = struct {
1042 state: std.atomic.Value(u32) = std.atomic.Value(u32).init(unlocked),
1043
1044 pub const unlocked: u32 = 0b00;
1045 pub const locked: u32 = 0b01;
1046 pub const contended: u32 = 0b11; // must contain the `locked` bit for x86 optimization below
1047
1048 pub fn tryLock(m: *Mutex) bool {
1049 // On x86, use `lock bts` instead of `lock cmpxchg` as:
1050 // - they both seem to mark the cache-line as modified regardless: https://stackoverflow.com/a/63350048
1051 // - `lock bts` is smaller instruction-wise which makes it better for inlining
1052 if (builtin.target.cpu.arch.isX86()) {
1053 const locked_bit = @ctz(locked);
1054 return m.state.bitSet(locked_bit, .acquire) == 0;
1055 }
1056
1057 // Acquire barrier ensures grabbing the lock happens before the critical section
1058 // and that the previous lock holder's critical section happens before we grab the lock.
1059 return m.state.cmpxchgWeak(unlocked, locked, .acquire, .monotonic) == null;
1060 }
1061
1062 /// Avoids the vtable for uncontended locks.
1063 pub fn lock(m: *Mutex, io: Io) void {
1064 if (!m.tryLock()) {
1065 @branchHint(.unlikely);
1066 io.vtable.mutexLock(io.userdata, m);
1067 }
1068 }
1069
1070 pub fn unlock(m: *Mutex, io: Io) void {
1071 io.vtable.mutexUnlock(io.userdata, m);
1072 }
1073};
1074
1075pub const Condition = struct {
1076 state: u64 = 0,
1077
1078 pub const WaitError = error{
1079 Timeout,
1080 Canceled,
1081 };
1082
1083 /// How many waiters to wake up.
1084 pub const Notify = enum {
1085 one,
1086 all,
1087 };
1088
1089 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) void {
1090 io.vtable.conditionWait(io.userdata, cond, mutex, null) catch |err| switch (err) {
1091 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out
1092 error.Canceled => return, // handled as spurious wakeup
1093 };
1094 }
1095
1096 pub fn timedWait(cond: *Condition, io: Io, mutex: *Mutex, timeout_ns: u64) WaitError!void {
1097 return io.vtable.conditionWait(io.userdata, cond, mutex, timeout_ns);
1098 }
1099
1100 pub fn signal(cond: *Condition, io: Io) void {
1101 io.vtable.conditionWake(io.userdata, cond, .one);
1102 }
1103
1104 pub fn broadcast(cond: *Condition, io: Io) void {
1105 io.vtable.conditionWake(io.userdata, cond, .all);
1106 }
1107};
1108
1109pub const TypeErasedQueue = struct {
1110 mutex: Mutex,
1111
1112 /// Ring buffer. This data is logically *after* queued getters.
1113 buffer: []u8,
1114 put_index: usize,
1115 get_index: usize,
1116
1117 putters: std.DoublyLinkedList(PutNode),
1118 getters: std.DoublyLinkedList(GetNode),
1119
1120 const PutNode = struct {
1121 remaining: []const u8,
1122 condition: Condition,
1123 };
1124
1125 const GetNode = struct {
1126 remaining: []u8,
1127 condition: Condition,
1128 };
1129
1130 pub fn init(buffer: []u8) TypeErasedQueue {
1131 return .{
1132 .mutex = .{},
1133 .buffer = buffer,
1134 .put_index = 0,
1135 .get_index = 0,
1136 .putters = .{},
1137 .getters = .{},
1138 };
1139 }
1140
1141 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) usize {
1142 assert(elements.len >= min);
1143
1144 q.mutex.lock(io);
1145 defer q.mutex.unlock(io);
1146
1147 // Getters have first priority on the data, and only when the getters
1148 // queue is empty do we start populating the buffer.
1149
1150 var remaining = elements;
1151 while (true) {
1152 const getter = q.getters.popFirst() orelse break;
1153 const copy_len = @min(getter.data.remaining.len, remaining.len);
1154 @memcpy(getter.data.remaining[0..copy_len], remaining[0..copy_len]);
1155 remaining = remaining[copy_len..];
1156 getter.data.remaining = getter.data.remaining[copy_len..];
1157 if (getter.data.remaining.len == 0) {
1158 getter.data.condition.signal(io);
1159 continue;
1160 }
1161 q.getters.prepend(getter);
1162 assert(remaining.len == 0);
1163 return elements.len;
1164 }
1165
1166 while (true) {
1167 {
1168 const available = q.buffer[q.put_index..];
1169 const copy_len = @min(available.len, remaining.len);
1170 @memcpy(available[0..copy_len], remaining[0..copy_len]);
1171 remaining = remaining[copy_len..];
1172 q.put_index += copy_len;
1173 if (remaining.len == 0) return elements.len;
1174 }
1175 {
1176 const available = q.buffer[0..q.get_index];
1177 const copy_len = @min(available.len, remaining.len);
1178 @memcpy(available[0..copy_len], remaining[0..copy_len]);
1179 remaining = remaining[copy_len..];
1180 q.put_index = copy_len;
1181 if (remaining.len == 0) return elements.len;
1182 }
1183
1184 const total_filled = elements.len - remaining.len;
1185 if (total_filled >= min) return total_filled;
1186
1187 var node: std.DoublyLinkedList(PutNode).Node = .{
1188 .data = .{ .remaining = remaining, .condition = .{} },
1189 };
1190 q.putters.append(&node);
1191 node.data.condition.wait(io, &q.mutex);
1192 remaining = node.data.remaining;
1193 }
1194 }
1195
1196 pub fn get(q: *@This(), io: Io, buffer: []u8, min: usize) usize {
1197 assert(buffer.len >= min);
1198
1199 q.mutex.lock(io);
1200 defer q.mutex.unlock(io);
1201
1202 // The ring buffer gets first priority, then data should come from any
1203 // queued putters, then finally the ring buffer should be filled with
1204 // data from putters so they can be resumed.
1205
1206 var remaining = buffer;
1207 while (true) {
1208 if (q.get_index <= q.put_index) {
1209 const available = q.buffer[q.get_index..q.put_index];
1210 const copy_len = @min(available.len, remaining.len);
1211 @memcpy(remaining[0..copy_len], available[0..copy_len]);
1212 q.get_index += copy_len;
1213 remaining = remaining[copy_len..];
1214 if (remaining.len == 0) return fillRingBufferFromPutters(q, io, buffer.len);
1215 } else {
1216 {
1217 const available = q.buffer[q.get_index..];
1218 const copy_len = @min(available.len, remaining.len);
1219 @memcpy(remaining[0..copy_len], available[0..copy_len]);
1220 q.get_index += copy_len;
1221 remaining = remaining[copy_len..];
1222 if (remaining.len == 0) return fillRingBufferFromPutters(q, io, buffer.len);
1223 }
1224 {
1225 const available = q.buffer[0..q.put_index];
1226 const copy_len = @min(available.len, remaining.len);
1227 @memcpy(remaining[0..copy_len], available[0..copy_len]);
1228 q.get_index = copy_len;
1229 remaining = remaining[copy_len..];
1230 if (remaining.len == 0) return fillRingBufferFromPutters(q, io, buffer.len);
1231 }
1232 }
1233 // Copy directly from putters into buffer.
1234 while (remaining.len > 0) {
1235 const putter = q.putters.popFirst() orelse break;
1236 const copy_len = @min(putter.data.remaining.len, remaining.len);
1237 @memcpy(remaining[0..copy_len], putter.data.remaining[0..copy_len]);
1238 putter.data.remaining = putter.data.remaining[copy_len..];
1239 remaining = remaining[copy_len..];
1240 if (putter.data.remaining.len == 0) {
1241 putter.data.condition.signal(io);
1242 } else {
1243 assert(remaining.len == 0);
1244 q.putters.prepend(putter);
1245 return fillRingBufferFromPutters(q, io, buffer.len);
1246 }
1247 }
1248 // Both ring buffer and putters queue is empty.
1249 const total_filled = buffer.len - remaining.len;
1250 if (total_filled >= min) return total_filled;
1251
1252 var node: std.DoublyLinkedList(GetNode).Node = .{
1253 .data = .{ .remaining = remaining, .condition = .{} },
1254 };
1255 q.getters.append(&node);
1256 node.data.condition.wait(io, &q.mutex);
1257 remaining = node.data.remaining;
1258 }
1259 }
1260
1261 /// Called when there is nonzero space available in the ring buffer and
1262 /// potentially putters waiting. The mutex is already held and the task is
1263 /// to copy putter data to the ring buffer and signal any putters whose
1264 /// buffers been fully copied.
1265 fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io, len: usize) usize {
1266 while (true) {
1267 const putter = q.putters.popFirst() orelse return len;
1268 const available = q.buffer[q.put_index..];
1269 const copy_len = @min(available.len, putter.data.remaining.len);
1270 @memcpy(available[0..copy_len], putter.data.remaining[0..copy_len]);
1271 putter.data.remaining = putter.data.remaining[copy_len..];
1272 q.put_index += copy_len;
1273 if (putter.data.remaining.len == 0) {
1274 putter.data.condition.signal(io);
1275 continue;
1276 }
1277 const second_available = q.buffer[0..q.get_index];
1278 const second_copy_len = @min(second_available.len, putter.data.remaining.len);
1279 @memcpy(second_available[0..second_copy_len], putter.data.remaining[0..second_copy_len]);
1280 putter.data.remaining = putter.data.remaining[copy_len..];
1281 q.put_index = copy_len;
1282 if (putter.data.remaining.len == 0) {
1283 putter.data.condition.signal(io);
1284 continue;
1285 }
1286 q.putters.prepend(putter);
1287 return len;
1288 }
1289 }
1290};
1291
1292/// Many producer, many consumer, thread-safe, runtime configurable buffer size.
1293/// When buffer is empty, consumers suspend and are resumed by producers.
1294/// When buffer is full, producers suspend and are resumed by consumers.
1295pub fn Queue(Elem: type) type {
1296 return struct {
1297 type_erased: TypeErasedQueue,
1298
1299 pub fn init(buffer: []Elem) @This() {
1300 return .{ .type_erased = .init(@ptrCast(buffer)) };
1301 }
1302
1303 /// Appends elements to the end of the queue. The function returns when
1304 /// at least `min` elements have been added to the buffer or sent
1305 /// directly to a consumer.
1306 ///
1307 /// Returns how many elements have been added to the queue.
1308 ///
1309 /// Asserts that `elements.len >= min`.
1310 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) usize {
1311 return @divExact(q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
1312 }
1313
1314 /// Receives elements from the beginning of the queue. The function
1315 /// returns when at least `min` elements have been populated inside
1316 /// `buffer`.
1317 ///
1318 /// Returns how many elements of `buffer` have been populated.
1319 ///
1320 /// Asserts that `buffer.len >= min`.
1321 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) usize {
1322 return @divExact(q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
1323 }
1324
1325 pub fn putOne(q: *@This(), io: Io, item: Elem) void {
1326 assert(q.put(io, &.{item}, 1) == 1);
1327 }
1328
1329 pub fn getOne(q: *@This(), io: Io) Elem {
1330 var buf: [1]Elem = undefined;
1331 assert(q.get(io, &buf, 1) == 1);
1332 return buf[0];
1333 }
1334 };
1335}
1336
1039/// Calls `function` with `args`, such that the return value of the function is1337/// Calls `function` with `args`, such that the return value of the function is
1040/// not guaranteed to be available until `await` is called.1338/// not guaranteed to be available until `await` is called.
1041pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {1339pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {
lib/std/Io/EventLoop.zig+7-7
...@@ -102,7 +102,7 @@ const Fiber = struct {...@@ -102,7 +102,7 @@ const Fiber = struct {
102 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));102 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
103 }103 }
104104
105 fn enterCancelRegion(fiber: *Fiber, thread: *Thread) error{AsyncCancel}!void {105 fn enterCancelRegion(fiber: *Fiber, thread: *Thread) error{Canceled}!void {
106 if (@cmpxchgStrong(106 if (@cmpxchgStrong(
107 ?*Thread,107 ?*Thread,
108 &fiber.cancel_thread,108 &fiber.cancel_thread,
...@@ -112,7 +112,7 @@ const Fiber = struct {...@@ -112,7 +112,7 @@ const Fiber = struct {
112 .acquire,112 .acquire,
113 )) |cancel_thread| {113 )) |cancel_thread| {
114 assert(cancel_thread == Thread.canceling);114 assert(cancel_thread == Thread.canceling);
115 return error.AsyncCancel;115 return error.Canceled;
116 }116 }
117 }117 }
118118
...@@ -746,7 +746,7 @@ pub fn createFile(...@@ -746,7 +746,7 @@ pub fn createFile(
746 switch (errno(completion.result)) {746 switch (errno(completion.result)) {
747 .SUCCESS => return .{ .handle = completion.result },747 .SUCCESS => return .{ .handle = completion.result },
748 .INTR => unreachable,748 .INTR => unreachable,
749 .CANCELED => return error.AsyncCancel,749 .CANCELED => return error.Canceled,
750750
751 .FAULT => unreachable,751 .FAULT => unreachable,
752 .INVAL => return error.BadPathName,752 .INVAL => return error.BadPathName,
...@@ -854,7 +854,7 @@ pub fn openFile(...@@ -854,7 +854,7 @@ pub fn openFile(
854 switch (errno(completion.result)) {854 switch (errno(completion.result)) {
855 .SUCCESS => return .{ .handle = completion.result },855 .SUCCESS => return .{ .handle = completion.result },
856 .INTR => unreachable,856 .INTR => unreachable,
857 .CANCELED => return error.AsyncCancel,857 .CANCELED => return error.Canceled,
858858
859 .FAULT => unreachable,859 .FAULT => unreachable,
860 .INVAL => return error.BadPathName,860 .INVAL => return error.BadPathName,
...@@ -950,7 +950,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std...@@ -950,7 +950,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std
950 switch (errno(completion.result)) {950 switch (errno(completion.result)) {
951 .SUCCESS => return @as(u32, @bitCast(completion.result)),951 .SUCCESS => return @as(u32, @bitCast(completion.result)),
952 .INTR => unreachable,952 .INTR => unreachable,
953 .CANCELED => return error.AsyncCancel,953 .CANCELED => return error.Canceled,
954954
955 .INVAL => unreachable,955 .INVAL => unreachable,
956 .FAULT => unreachable,956 .FAULT => unreachable,
...@@ -1002,7 +1002,7 @@ pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offs...@@ -1002,7 +1002,7 @@ pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offs
1002 switch (errno(completion.result)) {1002 switch (errno(completion.result)) {
1003 .SUCCESS => return @as(u32, @bitCast(completion.result)),1003 .SUCCESS => return @as(u32, @bitCast(completion.result)),
1004 .INTR => unreachable,1004 .INTR => unreachable,
1005 .CANCELED => return error.AsyncCancel,1005 .CANCELED => return error.Canceled,
10061006
1007 .INVAL => return error.InvalidArgument,1007 .INVAL => return error.InvalidArgument,
1008 .FAULT => unreachable,1008 .FAULT => unreachable,
...@@ -1080,7 +1080,7 @@ pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.D...@@ -1080,7 +1080,7 @@ pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.D
1080 switch (errno(completion.result)) {1080 switch (errno(completion.result)) {
1081 .SUCCESS, .TIME => return,1081 .SUCCESS, .TIME => return,
1082 .INTR => unreachable,1082 .INTR => unreachable,
1083 .CANCELED => return error.AsyncCancel,1083 .CANCELED => return error.Canceled,
10841084
1085 else => |err| return std.posix.unexpectedErrno(err),1085 else => |err| return std.posix.unexpectedErrno(err),
1086 }1086 }
lib/std/Thread/Pool.zig+179-8
...@@ -332,9 +332,12 @@ pub fn io(pool: *Pool) Io {...@@ -332,9 +332,12 @@ pub fn io(pool: *Pool) Io {
332 .vtable = &.{332 .vtable = &.{
333 .@"async" = @"async",333 .@"async" = @"async",
334 .@"await" = @"await",334 .@"await" = @"await",
335
336 .cancel = cancel,335 .cancel = cancel,
337 .cancelRequested = cancelRequested,336 .cancelRequested = cancelRequested,
337 .mutexLock = mutexLock,
338 .mutexUnlock = mutexUnlock,
339 .conditionWait = conditionWait,
340 .conditionWake = conditionWake,
338341
339 .createFile = createFile,342 .createFile = createFile,
340 .openFile = openFile,343 .openFile = openFile,
...@@ -517,11 +520,179 @@ fn cancelRequested(userdata: ?*anyopaque) bool {...@@ -517,11 +520,179 @@ fn cancelRequested(userdata: ?*anyopaque) bool {
517 return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == AsyncClosure.canceling_tid;520 return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == AsyncClosure.canceling_tid;
518}521}
519522
520fn checkCancel(pool: *Pool) error{AsyncCancel}!void {523fn checkCancel(pool: *Pool) error{Canceled}!void {
521 if (cancelRequested(pool)) return error.AsyncCancel;524 if (cancelRequested(pool)) return error.Canceled;
525}
526
527fn mutexLock(userdata: ?*anyopaque, m: *Io.Mutex) void {
528 @branchHint(.cold);
529 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
530 _ = pool;
531
532 // Avoid doing an atomic swap below if we already know the state is contended.
533 // An atomic swap unconditionally stores which marks the cache-line as modified unnecessarily.
534 if (m.state.load(.monotonic) == Io.Mutex.contended) {
535 std.Thread.Futex.wait(&m.state, Io.Mutex.contended);
536 }
537
538 // Try to acquire the lock while also telling the existing lock holder that there are threads waiting.
539 //
540 // Once we sleep on the Futex, we must acquire the mutex using `contended` rather than `locked`.
541 // If not, threads sleeping on the Futex wouldn't see the state change in unlock and potentially deadlock.
542 // The downside is that the last mutex unlocker will see `contended` and do an unnecessary Futex wake
543 // but this is better than having to wake all waiting threads on mutex unlock.
544 //
545 // Acquire barrier ensures grabbing the lock happens before the critical section
546 // and that the previous lock holder's critical section happens before we grab the lock.
547 while (m.state.swap(Io.Mutex.contended, .acquire) != Io.Mutex.unlocked) {
548 std.Thread.Futex.wait(&m.state, Io.Mutex.contended);
549 }
550}
551
552fn mutexUnlock(userdata: ?*anyopaque, m: *Io.Mutex) void {
553 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
554 _ = pool;
555 // Needs to also wake up a waiting thread if any.
556 //
557 // A waiting thread will acquire with `contended` instead of `locked`
558 // which ensures that it wakes up another thread on the next unlock().
559 //
560 // Release barrier ensures the critical section happens before we let go of the lock
561 // and that our critical section happens before the next lock holder grabs the lock.
562 const state = m.state.swap(Io.Mutex.unlocked, .release);
563 assert(state != Io.Mutex.unlocked);
564
565 if (state == Io.Mutex.contended) {
566 std.Thread.Futex.wake(&m.state, 1);
567 }
568}
569
570fn mutexLockInternal(pool: *std.Thread.Pool, m: *Io.Mutex) void {
571 if (!m.tryLock()) {
572 @branchHint(.unlikely);
573 mutexLock(pool, m);
574 }
575}
576
577fn conditionWait(
578 userdata: ?*anyopaque,
579 cond: *Io.Condition,
580 mutex: *Io.Mutex,
581 timeout: ?u64,
582) Io.Condition.WaitError!void {
583 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
584 comptime assert(@TypeOf(cond.state) == u64);
585 const ints: *[2]std.atomic.Value(u32) = @ptrCast(&cond.state);
586 const cond_state = &ints[0];
587 const cond_epoch = &ints[1];
588 const one_waiter = 1;
589 const waiter_mask = 0xffff;
590 const one_signal = 1 << 16;
591 const signal_mask = 0xffff << 16;
592 // Observe the epoch, then check the state again to see if we should wake up.
593 // The epoch must be observed before we check the state or we could potentially miss a wake() and deadlock:
594 //
595 // - T1: s = LOAD(&state)
596 // - T2: UPDATE(&s, signal)
597 // - T2: UPDATE(&epoch, 1) + FUTEX_WAKE(&epoch)
598 // - T1: e = LOAD(&epoch) (was reordered after the state load)
599 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed the state update + the epoch change)
600 //
601 // Acquire barrier to ensure the epoch load happens before the state load.
602 var epoch = cond_epoch.load(.acquire);
603 var state = cond_state.fetchAdd(one_waiter, .monotonic);
604 assert(state & waiter_mask != waiter_mask);
605 state += one_waiter;
606
607 mutexUnlock(pool, mutex);
608 defer mutexLockInternal(pool, mutex);
609
610 var futex_deadline = std.Thread.Futex.Deadline.init(timeout);
611
612 while (true) {
613 futex_deadline.wait(cond_epoch, epoch) catch |err| switch (err) {
614 // On timeout, we must decrement the waiter we added above.
615 error.Timeout => {
616 while (true) {
617 // If there's a signal when we're timing out, consume it and report being woken up instead.
618 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
619 while (state & signal_mask != 0) {
620 const new_state = state - one_waiter - one_signal;
621 state = cond_state.cmpxchgWeak(state, new_state, .acquire, .monotonic) orelse return;
622 }
623
624 // Remove the waiter we added and officially return timed out.
625 const new_state = state - one_waiter;
626 state = cond_state.cmpxchgWeak(state, new_state, .monotonic, .monotonic) orelse return err;
627 }
628 },
629 };
630
631 epoch = cond_epoch.load(.acquire);
632 state = cond_state.load(.monotonic);
633
634 // Try to wake up by consuming a signal and decremented the waiter we added previously.
635 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
636 while (state & signal_mask != 0) {
637 const new_state = state - one_waiter - one_signal;
638 state = cond_state.cmpxchgWeak(state, new_state, .acquire, .monotonic) orelse return;
639 }
640 }
641}
642
643fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Condition.Notify) void {
644 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
645 _ = pool;
646 comptime assert(@TypeOf(cond.state) == u64);
647 const ints: *[2]std.atomic.Value(u32) = @ptrCast(&cond.state);
648 const cond_state = &ints[0];
649 const cond_epoch = &ints[1];
650 const one_waiter = 1;
651 const waiter_mask = 0xffff;
652 const one_signal = 1 << 16;
653 const signal_mask = 0xffff << 16;
654 var state = cond_state.load(.monotonic);
655 while (true) {
656 const waiters = (state & waiter_mask) / one_waiter;
657 const signals = (state & signal_mask) / one_signal;
658
659 // Reserves which waiters to wake up by incrementing the signals count.
660 // Therefore, the signals count is always less than or equal to the waiters count.
661 // We don't need to Futex.wake if there's nothing to wake up or if other wake() threads have reserved to wake up the current waiters.
662 const wakeable = waiters - signals;
663 if (wakeable == 0) {
664 return;
665 }
666
667 const to_wake = switch (notify) {
668 .one => 1,
669 .all => wakeable,
670 };
671
672 // Reserve the amount of waiters to wake by incrementing the signals count.
673 // Release barrier ensures code before the wake() happens before the signal it posted and consumed by the wait() threads.
674 const new_state = state + (one_signal * to_wake);
675 state = cond_state.cmpxchgWeak(state, new_state, .release, .monotonic) orelse {
676 // Wake up the waiting threads we reserved above by changing the epoch value.
677 // NOTE: a waiting thread could miss a wake up if *exactly* ((1<<32)-1) wake()s happen between it observing the epoch and sleeping on it.
678 // This is very unlikely due to how many precise amount of Futex.wake() calls that would be between the waiting thread's potential preemption.
679 //
680 // Release barrier ensures the signal being added to the state happens before the epoch is changed.
681 // If not, the waiting thread could potentially deadlock from missing both the state and epoch change:
682 //
683 // - T2: UPDATE(&epoch, 1) (reordered before the state change)
684 // - T1: e = LOAD(&epoch)
685 // - T1: s = LOAD(&state)
686 // - T2: UPDATE(&state, signal) + FUTEX_WAKE(&epoch)
687 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed both epoch change and state change)
688 _ = cond_epoch.fetchAdd(1, .release);
689 std.Thread.Futex.wake(cond_epoch, to_wake);
690 return;
691 };
692 }
522}693}
523694
524pub fn createFile(695fn createFile(
525 userdata: ?*anyopaque,696 userdata: ?*anyopaque,
526 dir: std.fs.Dir,697 dir: std.fs.Dir,
527 sub_path: []const u8,698 sub_path: []const u8,
...@@ -532,7 +703,7 @@ pub fn createFile(...@@ -532,7 +703,7 @@ pub fn createFile(
532 return dir.createFile(sub_path, flags);703 return dir.createFile(sub_path, flags);
533}704}
534705
535pub fn openFile(706fn openFile(
536 userdata: ?*anyopaque,707 userdata: ?*anyopaque,
537 dir: std.fs.Dir,708 dir: std.fs.Dir,
538 sub_path: []const u8,709 sub_path: []const u8,
...@@ -543,13 +714,13 @@ pub fn openFile(...@@ -543,13 +714,13 @@ pub fn openFile(
543 return dir.openFile(sub_path, flags);714 return dir.openFile(sub_path, flags);
544}715}
545716
546pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {717fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
547 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));718 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
548 _ = pool;719 _ = pool;
549 return file.close();720 return file.close();
550}721}
551722
552pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {723fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {
553 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));724 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
554 try pool.checkCancel();725 try pool.checkCancel();
555 return switch (offset) {726 return switch (offset) {
...@@ -558,7 +729,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std...@@ -558,7 +729,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std
558 };729 };
559}730}
560731
561pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {732fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {
562 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));733 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
563 try pool.checkCancel();734 try pool.checkCancel();
564 return switch (offset) {735 return switch (offset) {