authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-31 16:25:11-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:39-07:00
log74969d1c40fdc31c8d8df6df2619304aaa1a533b
tree57320a5f7269278eaca239cc37ddba945a4a8f63
parent59270425aa78e0bd1d6fff50e4a7e55469547a63

std.Io.Condition: change primitive to support only one

and no timer

3 files changed, 37 insertions(+), 80 deletions(-)

lib/std/Io.zig+16-36
...@@ -982,8 +982,8 @@ pub const VTable = struct {...@@ -982,8 +982,8 @@ pub const VTable = struct {
982 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,982 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,
983 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,983 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
984984
985 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex, timeout_ns: ?u64) Condition.WaitError!void,985 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,
986 conditionWake: *const fn (?*anyopaque, cond: *Condition, notify: Condition.Notify) void,986 conditionWake: *const fn (?*anyopaque, cond: *Condition) void,
987987
988 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,988 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
989 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,989 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
...@@ -995,6 +995,11 @@ pub const VTable = struct {...@@ -995,6 +995,11 @@ pub const VTable = struct {
995 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,995 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,
996};996};
997997
998pub const Cancelable = error{
999 /// Caller has requested the async operation to stop.
1000 Canceled,
1001};
1002
998pub const OpenFlags = fs.File.OpenFlags;1003pub const OpenFlags = fs.File.OpenFlags;
999pub const CreateFlags = fs.File.CreateFlags;1004pub const CreateFlags = fs.File.CreateFlags;
10001005
...@@ -1148,43 +1153,18 @@ pub const Mutex = if (true) struct {...@@ -1148,43 +1153,18 @@ pub const Mutex = if (true) struct {
1148 }1153 }
1149};1154};
11501155
1156/// Supports exactly 1 waiter. More than 1 simultaneous wait on the same
1157/// condition is illegal.
1151pub const Condition = struct {1158pub const Condition = struct {
1152 state: u64 = 0,1159 state: u64 = 0,
11531160
1154 pub const WaitError = error{
1155 Timeout,
1156 Canceled,
1157 };
1158
1159 /// How many waiters to wake up.
1160 pub const Notify = enum {
1161 one,
1162 all,
1163 };
1164
1165 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {1161 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {
1166 io.vtable.conditionWait(io.userdata, cond, mutex, null) catch |err| switch (err) {1162 return io.vtable.conditionWait(io.userdata, cond, mutex);
1167 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out
1168 error.Canceled => return error.Canceled,
1169 };
1170 }1163 }
11711164
1172 pub fn timedWait(cond: *Condition, io: Io, mutex: *Mutex, timeout_ns: u64) WaitError!void {1165 pub fn wake(cond: *Condition, io: Io) void {
1173 return io.vtable.conditionWait(io.userdata, cond, mutex, timeout_ns);1166 io.vtable.conditionWake(io.userdata, cond);
1174 }1167 }
1175
1176 pub fn signal(cond: *Condition, io: Io) void {
1177 io.vtable.conditionWake(io.userdata, cond, .one);
1178 }
1179
1180 pub fn broadcast(cond: *Condition, io: Io) void {
1181 io.vtable.conditionWake(io.userdata, cond, .all);
1182 }
1183};
1184
1185pub const Cancelable = error{
1186 /// Caller has requested the async operation to stop.
1187 Canceled,
1188};1168};
11891169
1190pub const TypeErasedQueue = struct {1170pub const TypeErasedQueue = struct {
...@@ -1236,7 +1216,7 @@ pub const TypeErasedQueue = struct {...@@ -1236,7 +1216,7 @@ pub const TypeErasedQueue = struct {
1236 remaining = remaining[copy_len..];1216 remaining = remaining[copy_len..];
1237 getter.data.remaining = getter.data.remaining[copy_len..];1217 getter.data.remaining = getter.data.remaining[copy_len..];
1238 if (getter.data.remaining.len == 0) {1218 if (getter.data.remaining.len == 0) {
1239 getter.data.condition.signal(io);1219 getter.data.condition.wake(io);
1240 continue;1220 continue;
1241 }1221 }
1242 q.getters.prepend(getter);1222 q.getters.prepend(getter);
...@@ -1319,7 +1299,7 @@ pub const TypeErasedQueue = struct {...@@ -1319,7 +1299,7 @@ pub const TypeErasedQueue = struct {
1319 putter.data.remaining = putter.data.remaining[copy_len..];1299 putter.data.remaining = putter.data.remaining[copy_len..];
1320 remaining = remaining[copy_len..];1300 remaining = remaining[copy_len..];
1321 if (putter.data.remaining.len == 0) {1301 if (putter.data.remaining.len == 0) {
1322 putter.data.condition.signal(io);1302 putter.data.condition.wake(io);
1323 } else {1303 } else {
1324 assert(remaining.len == 0);1304 assert(remaining.len == 0);
1325 q.putters.prepend(putter);1305 q.putters.prepend(putter);
...@@ -1352,7 +1332,7 @@ pub const TypeErasedQueue = struct {...@@ -1352,7 +1332,7 @@ pub const TypeErasedQueue = struct {
1352 putter.data.remaining = putter.data.remaining[copy_len..];1332 putter.data.remaining = putter.data.remaining[copy_len..];
1353 q.put_index += copy_len;1333 q.put_index += copy_len;
1354 if (putter.data.remaining.len == 0) {1334 if (putter.data.remaining.len == 0) {
1355 putter.data.condition.signal(io);1335 putter.data.condition.wake(io);
1356 continue;1336 continue;
1357 }1337 }
1358 const second_available = q.buffer[0..q.get_index];1338 const second_available = q.buffer[0..q.get_index];
...@@ -1361,7 +1341,7 @@ pub const TypeErasedQueue = struct {...@@ -1361,7 +1341,7 @@ pub const TypeErasedQueue = struct {
1361 putter.data.remaining = putter.data.remaining[copy_len..];1341 putter.data.remaining = putter.data.remaining[copy_len..];
1362 q.put_index = copy_len;1342 q.put_index = copy_len;
1363 if (putter.data.remaining.len == 0) {1343 if (putter.data.remaining.len == 0) {
1364 putter.data.condition.signal(io);1344 putter.data.condition.wake(io);
1365 continue;1345 continue;
1366 }1346 }
1367 q.putters.prepend(putter);1347 q.putters.prepend(putter);
lib/std/Io/EventLoop.zig+16-17
...@@ -781,7 +781,6 @@ fn go(...@@ -781,7 +781,6 @@ fn go(
781 event_loop.schedule(current_thread, .{ .head = fiber, .tail = fiber });781 event_loop.schedule(current_thread, .{ .head = fiber, .tail = fiber });
782}782}
783783
784
785fn @"await"(784fn @"await"(
786 userdata: ?*anyopaque,785 userdata: ?*anyopaque,
787 any_future: *std.Io.AnyFuture,786 any_future: *std.Io.AnyFuture,
...@@ -1277,24 +1276,24 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut...@@ -1277,24 +1276,24 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut
1277 el.yield(maybe_waiting_fiber.?, .reschedule);1276 el.yield(maybe_waiting_fiber.?, .reschedule);
1278}1277}
12791278
1280fn conditionWait(1279fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {
1281 userdata: ?*anyopaque,1280 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1282 cond: *Io.Condition,1281 const cond_state: *?*Fiber = @ptrCast(&cond.state);
1283 mutex: *Io.Mutex,1282 const thread: *Thread = .current();
1284 timeout: ?u64,1283 const fiber = thread.currentFiber();
1285) Io.Condition.WaitError!void {1284 const prev = @atomicRmw(?*Fiber, cond_state, .Xchg, fiber, .acquire);
1286 _ = userdata;1285 assert(prev == null); // More than one wait on same Condition is illegal.
1287 _ = cond;1286 mutex.unlock(io(el));
1288 _ = mutex;1287 el.yield(null, .nothing);
1289 _ = timeout;1288 try mutex.lock(io(el));
1290 @panic("TODO");
1291}1289}
12921290
1293fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Condition.Notify) void {1291fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {
1294 _ = userdata;1292 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1295 _ = cond;1293 const cond_state: *?*Fiber = @ptrCast(&cond.state);
1296 _ = notify;1294 if (@atomicRmw(?*Fiber, cond_state, .Xchg, null, .acquire)) |fiber| {
1297 @panic("TODO");1295 el.yield(fiber, .reschedule);
1296 }
1298}1297}
12991298
1300fn errno(signed: i32) std.os.linux.E {1299fn errno(signed: i32) std.os.linux.E {
lib/std/Thread/Pool.zig+5-27
...@@ -619,12 +619,7 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut...@@ -619,12 +619,7 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut
619 }619 }
620}620}
621621
622fn conditionWait(622fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {
623 userdata: ?*anyopaque,
624 cond: *Io.Condition,
625 mutex: *Io.Mutex,
626 timeout: ?u64,
627) Io.Condition.WaitError!void {
628 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));623 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
629 comptime assert(@TypeOf(cond.state) == u64);624 comptime assert(@TypeOf(cond.state) == u64);
630 const ints: *[2]std.atomic.Value(u32) = @ptrCast(&cond.state);625 const ints: *[2]std.atomic.Value(u32) = @ptrCast(&cond.state);
...@@ -652,25 +647,11 @@ fn conditionWait(...@@ -652,25 +647,11 @@ fn conditionWait(
652 mutex.unlock(pool.io());647 mutex.unlock(pool.io());
653 defer mutex.lock(pool.io()) catch @panic("TODO");648 defer mutex.lock(pool.io()) catch @panic("TODO");
654649
655 var futex_deadline = std.Thread.Futex.Deadline.init(timeout);650 var futex_deadline = std.Thread.Futex.Deadline.init(null);
656651
657 while (true) {652 while (true) {
658 futex_deadline.wait(cond_epoch, epoch) catch |err| switch (err) {653 futex_deadline.wait(cond_epoch, epoch) catch |err| switch (err) {
659 // On timeout, we must decrement the waiter we added above.654 error.Timeout => unreachable,
660 error.Timeout => {
661 while (true) {
662 // If there's a signal when we're timing out, consume it and report being woken up instead.
663 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
664 while (state & signal_mask != 0) {
665 const new_state = state - one_waiter - one_signal;
666 state = cond_state.cmpxchgWeak(state, new_state, .acquire, .monotonic) orelse return;
667 }
668
669 // Remove the waiter we added and officially return timed out.
670 const new_state = state - one_waiter;
671 state = cond_state.cmpxchgWeak(state, new_state, .monotonic, .monotonic) orelse return err;
672 }
673 },
674 };655 };
675656
676 epoch = cond_epoch.load(.acquire);657 epoch = cond_epoch.load(.acquire);
...@@ -685,7 +666,7 @@ fn conditionWait(...@@ -685,7 +666,7 @@ fn conditionWait(
685 }666 }
686}667}
687668
688fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Condition.Notify) void {669fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {
689 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));670 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
690 _ = pool;671 _ = pool;
691 comptime assert(@TypeOf(cond.state) == u64);672 comptime assert(@TypeOf(cond.state) == u64);
...@@ -709,10 +690,7 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Conditio...@@ -709,10 +690,7 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Conditio
709 return;690 return;
710 }691 }
711692
712 const to_wake = switch (notify) {693 const to_wake = 1;
713 .one => 1,
714 .all => wakeable,
715 };
716694
717 // Reserve the amount of waiters to wake by incrementing the signals count.695 // Reserve the amount of waiters to wake by incrementing the signals count.
718 // Release barrier ensures code before the wake() happens before the signal it posted and consumed by the wait() threads.696 // Release barrier ensures code before the wake() happens before the signal it posted and consumed by the wait() threads.