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-10-02 16:30:59-07:00
loge5b2df0c9b84da0daac5166a1b6170ccbf00e0b7
treea7a8875a1f9c82bf2a879370d082999cbd551809
parent63b3a3d11c389a2cf617ed6d1a7b5a6b36991974

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
......@@ -629,8 +629,8 @@ pub const VTable = struct {
629629 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,
630630 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,
633 conditionWake: *const fn (?*anyopaque, cond: *Condition, notify: Condition.Notify) void,
632 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,
633 conditionWake: *const fn (?*anyopaque, cond: *Condition) void,
634634
635635 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
636636 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
......@@ -642,6 +642,11 @@ pub const VTable = struct {
642642 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,
643643};
644644
645pub const Cancelable = error{
646 /// Caller has requested the async operation to stop.
647 Canceled,
648};
649
645650pub const OpenFlags = fs.File.OpenFlags;
646651pub const CreateFlags = fs.File.CreateFlags;
647652
......@@ -795,43 +800,18 @@ pub const Mutex = if (true) struct {
795800 }
796801};
797802
803/// Supports exactly 1 waiter. More than 1 simultaneous wait on the same
804/// condition is illegal.
798805pub const Condition = struct {
799806 state: u64 = 0,
800807
801 pub const WaitError = error{
802 Timeout,
803 Canceled,
804 };
805
806 /// How many waiters to wake up.
807 pub const Notify = enum {
808 one,
809 all,
810 };
811
812808 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {
813 io.vtable.conditionWait(io.userdata, cond, mutex, null) catch |err| switch (err) {
814 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out
815 error.Canceled => return error.Canceled,
816 };
809 return io.vtable.conditionWait(io.userdata, cond, mutex);
817810 }
818811
819 pub fn timedWait(cond: *Condition, io: Io, mutex: *Mutex, timeout_ns: u64) WaitError!void {
820 return io.vtable.conditionWait(io.userdata, cond, mutex, timeout_ns);
812 pub fn wake(cond: *Condition, io: Io) void {
813 io.vtable.conditionWake(io.userdata, cond);
821814 }
822
823 pub fn signal(cond: *Condition, io: Io) void {
824 io.vtable.conditionWake(io.userdata, cond, .one);
825 }
826
827 pub fn broadcast(cond: *Condition, io: Io) void {
828 io.vtable.conditionWake(io.userdata, cond, .all);
829 }
830};
831
832pub const Cancelable = error{
833 /// Caller has requested the async operation to stop.
834 Canceled,
835815};
836816
837817pub const TypeErasedQueue = struct {
......@@ -883,7 +863,7 @@ pub const TypeErasedQueue = struct {
883863 remaining = remaining[copy_len..];
884864 getter.data.remaining = getter.data.remaining[copy_len..];
885865 if (getter.data.remaining.len == 0) {
886 getter.data.condition.signal(io);
866 getter.data.condition.wake(io);
887867 continue;
888868 }
889869 q.getters.prepend(getter);
......@@ -966,7 +946,7 @@ pub const TypeErasedQueue = struct {
966946 putter.data.remaining = putter.data.remaining[copy_len..];
967947 remaining = remaining[copy_len..];
968948 if (putter.data.remaining.len == 0) {
969 putter.data.condition.signal(io);
949 putter.data.condition.wake(io);
970950 } else {
971951 assert(remaining.len == 0);
972952 q.putters.prepend(putter);
......@@ -999,7 +979,7 @@ pub const TypeErasedQueue = struct {
999979 putter.data.remaining = putter.data.remaining[copy_len..];
1000980 q.put_index += copy_len;
1001981 if (putter.data.remaining.len == 0) {
1002 putter.data.condition.signal(io);
982 putter.data.condition.wake(io);
1003983 continue;
1004984 }
1005985 const second_available = q.buffer[0..q.get_index];
......@@ -1008,7 +988,7 @@ pub const TypeErasedQueue = struct {
1008988 putter.data.remaining = putter.data.remaining[copy_len..];
1009989 q.put_index = copy_len;
1010990 if (putter.data.remaining.len == 0) {
1011 putter.data.condition.signal(io);
991 putter.data.condition.wake(io);
1012992 continue;
1013993 }
1014994 q.putters.prepend(putter);
lib/std/Io/EventLoop.zig+16-17
......@@ -781,7 +781,6 @@ fn go(
781781 event_loop.schedule(current_thread, .{ .head = fiber, .tail = fiber });
782782}
783783
784
785784fn @"await"(
786785 userdata: ?*anyopaque,
787786 any_future: *std.Io.AnyFuture,
......@@ -1277,24 +1276,24 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut
12771276 el.yield(maybe_waiting_fiber.?, .reschedule);
12781277}
12791278
1280fn conditionWait(
1281 userdata: ?*anyopaque,
1282 cond: *Io.Condition,
1283 mutex: *Io.Mutex,
1284 timeout: ?u64,
1285) Io.Condition.WaitError!void {
1286 _ = userdata;
1287 _ = cond;
1288 _ = mutex;
1289 _ = timeout;
1290 @panic("TODO");
1279fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {
1280 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1281 const cond_state: *?*Fiber = @ptrCast(&cond.state);
1282 const thread: *Thread = .current();
1283 const fiber = thread.currentFiber();
1284 const prev = @atomicRmw(?*Fiber, cond_state, .Xchg, fiber, .acquire);
1285 assert(prev == null); // More than one wait on same Condition is illegal.
1286 mutex.unlock(io(el));
1287 el.yield(null, .nothing);
1288 try mutex.lock(io(el));
12911289}
12921290
1293fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Condition.Notify) void {
1294 _ = userdata;
1295 _ = cond;
1296 _ = notify;
1297 @panic("TODO");
1291fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {
1292 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1293 const cond_state: *?*Fiber = @ptrCast(&cond.state);
1294 if (@atomicRmw(?*Fiber, cond_state, .Xchg, null, .acquire)) |fiber| {
1295 el.yield(fiber, .reschedule);
1296 }
12981297}
12991298
13001299fn 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
619619 }
620620}
621621
622fn conditionWait(
623 userdata: ?*anyopaque,
624 cond: *Io.Condition,
625 mutex: *Io.Mutex,
626 timeout: ?u64,
627) Io.Condition.WaitError!void {
622fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {
628623 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
629624 comptime assert(@TypeOf(cond.state) == u64);
630625 const ints: *[2]std.atomic.Value(u32) = @ptrCast(&cond.state);
......@@ -652,25 +647,11 @@ fn conditionWait(
652647 mutex.unlock(pool.io());
653648 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
657652 while (true) {
658653 futex_deadline.wait(cond_epoch, epoch) catch |err| switch (err) {
659 // On timeout, we must decrement the waiter we added above.
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 },
654 error.Timeout => unreachable,
674655 };
675656
676657 epoch = cond_epoch.load(.acquire);
......@@ -685,7 +666,7 @@ fn conditionWait(
685666 }
686667}
687668
688fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Condition.Notify) void {
669fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {
689670 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
690671 _ = pool;
691672 comptime assert(@TypeOf(cond.state) == u64);
......@@ -709,10 +690,7 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Conditio
709690 return;
710691 }
711692
712 const to_wake = switch (notify) {
713 .one => 1,
714 .all => wakeable,
715 };
693 const to_wake = 1;
716694
717695 // Reserve the amount of waiters to wake by incrementing the signals count.
718696 // Release barrier ensures code before the wake() happens before the signal it posted and consumed by the wait() threads.