authorgravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-20 13:10:10+00:00
committergravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-21 13:07:04+00:00
log9bf65f6e05467496780ede466cebc9ed8a0e17f3
tree145199235f4c1e565a9de225b1cd1c968362114f
parent0649f96da3e89c1397824e3b42702024b70ec332
signaturelock-open Commit is signed but in an unrecognized format.

std.Io.Threaded: replace ResetEvent with Io.Event

1 files changed, 24 insertions(+), 227 deletions(-)

lib/std/Io/Threaded.zig+24-227
......@@ -783,13 +783,13 @@ const preadv_sym = if (posix.lfs64_abi) posix.system.preadv64 else posix.system.
783783const AsyncClosure = struct {
784784 closure: Closure,
785785 func: *const fn (context: *anyopaque, result: *anyopaque) void,
786 reset_event: ResetEvent,
787 select_condition: ?*ResetEvent,
786 event: Io.Event,
787 select_condition: ?*Io.Event,
788788 context_alignment: Alignment,
789789 result_offset: usize,
790790 alloc_len: usize,
791791
792 const done_reset_event: *ResetEvent = @ptrFromInt(@alignOf(ResetEvent));
792 const done_event: *Io.Event = @ptrFromInt(@alignOf(Io.Event));
793793
794794 fn start(closure: *Closure, t: *Threaded) void {
795795 const ac: *AsyncClosure = @alignCast(@fieldParentPtr("closure", closure));
......@@ -798,11 +798,11 @@ const AsyncClosure = struct {
798798 ac.func(ac.contextPointer(), ac.resultPointer());
799799 current_thread.current_closure = null;
800800
801 if (@atomicRmw(?*ResetEvent, &ac.select_condition, .Xchg, done_reset_event, .release)) |select_reset| {
802 assert(select_reset != done_reset_event);
803 select_reset.set();
801 if (@atomicRmw(?*Io.Event, &ac.select_condition, .Xchg, done_event, .release)) |select_event| {
802 assert(select_event != done_event);
803 select_event.set(ioBasic(t));
804804 }
805 ac.reset_event.set();
805 ac.event.set(ioBasic(t));
806806 }
807807
808808 fn resultPointer(ac: *AsyncClosure) [*]u8 {
......@@ -844,7 +844,7 @@ const AsyncClosure = struct {
844844 .context_alignment = context_alignment,
845845 .result_offset = actual_result_offset,
846846 .alloc_len = alloc_len,
847 .reset_event = .unset,
847 .event = .unset,
848848 .select_condition = null,
849849 };
850850 @memcpy(ac.contextPointer()[0..context.len], context);
......@@ -852,10 +852,10 @@ const AsyncClosure = struct {
852852 }
853853
854854 fn waitAndDeinit(ac: *AsyncClosure, t: *Threaded, result: []u8) void {
855 ac.reset_event.wait(t) catch |err| switch (err) {
855 ac.event.wait(ioBasic(t)) catch |err| switch (err) {
856856 error.Canceled => {
857857 ac.closure.requestCancel(t);
858 ac.reset_event.waitUncancelable();
858 ac.event.waitUncancelable(ioBasic(t));
859859 },
860860 };
861861 @memcpy(result, ac.resultPointer()[0..result.len]);
......@@ -977,14 +977,14 @@ const GroupClosure = struct {
977977 const current_thread = Thread.getCurrent(t);
978978 const group = gc.group;
979979 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
980 const reset_event: *ResetEvent = @ptrCast(&group.context);
980 const event: *Io.Event = @ptrCast(&group.context);
981981 current_thread.current_closure = closure;
982982 gc.func(group, gc.contextPointer());
983983 current_thread.current_closure = null;
984984
985985 const prev_state = group_state.fetchSub(sync_one_pending, .acq_rel);
986986 assert((prev_state / sync_one_pending) > 0);
987 if (prev_state == (sync_one_pending | sync_is_waiting)) reset_event.set();
987 if (prev_state == (sync_one_pending | sync_is_waiting)) event.set(ioBasic(t));
988988 }
989989
990990 fn contextPointer(gc: *GroupClosure) [*]u8 {
......@@ -1145,10 +1145,10 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
11451145 if (builtin.single_threaded) return;
11461146
11471147 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
1148 const reset_event: *ResetEvent = @ptrCast(&group.context);
1148 const event: *Io.Event = @ptrCast(&group.context);
11491149 const prev_state = group_state.fetchAdd(GroupClosure.sync_is_waiting, .acquire);
11501150 assert(prev_state & GroupClosure.sync_is_waiting == 0);
1151 if ((prev_state / GroupClosure.sync_one_pending) > 0) reset_event.wait(t) catch |err| switch (err) {
1151 if ((prev_state / GroupClosure.sync_one_pending) > 0) event.wait(ioBasic(t)) catch |err| switch (err) {
11521152 error.Canceled => {
11531153 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
11541154 while (true) {
......@@ -1156,7 +1156,7 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
11561156 gc.closure.requestCancel(t);
11571157 node = node.next orelse break;
11581158 }
1159 reset_event.waitUncancelable();
1159 event.waitUncancelable(ioBasic(t));
11601160 },
11611161 };
11621162
......@@ -1185,10 +1185,10 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void
11851185 }
11861186
11871187 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
1188 const reset_event: *ResetEvent = @ptrCast(&group.context);
1188 const event: *Io.Event = @ptrCast(&group.context);
11891189 const prev_state = group_state.fetchAdd(GroupClosure.sync_is_waiting, .acquire);
11901190 assert(prev_state & GroupClosure.sync_is_waiting == 0);
1191 if ((prev_state / GroupClosure.sync_one_pending) > 0) reset_event.waitUncancelable();
1191 if ((prev_state / GroupClosure.sync_one_pending) > 0) event.waitUncancelable(ioBasic(t));
11921192
11931193 {
11941194 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
......@@ -3684,28 +3684,28 @@ fn sleepPosix(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void {
36843684fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) Io.Cancelable!usize {
36853685 const t: *Threaded = @ptrCast(@alignCast(userdata));
36863686
3687 var reset_event: ResetEvent = .unset;
3687 var event: Io.Event = .unset;
36883688
36893689 for (futures, 0..) |future, i| {
36903690 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
3691 if (@atomicRmw(?*ResetEvent, &closure.select_condition, .Xchg, &reset_event, .seq_cst) == AsyncClosure.done_reset_event) {
3691 if (@atomicRmw(?*Io.Event, &closure.select_condition, .Xchg, &event, .seq_cst) == AsyncClosure.done_event) {
36923692 for (futures[0..i]) |cleanup_future| {
36933693 const cleanup_closure: *AsyncClosure = @ptrCast(@alignCast(cleanup_future));
3694 if (@atomicRmw(?*ResetEvent, &cleanup_closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
3695 cleanup_closure.reset_event.waitUncancelable(); // Ensure no reference to our stack-allocated reset_event.
3694 if (@atomicRmw(?*Io.Event, &cleanup_closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_event) {
3695 cleanup_closure.event.waitUncancelable(ioBasic(t)); // Ensure no reference to our stack-allocated event.
36963696 }
36973697 }
36983698 return i;
36993699 }
37003700 }
37013701
3702 try reset_event.wait(t);
3702 try event.wait(ioBasic(t));
37033703
37043704 var result: ?usize = null;
37053705 for (futures, 0..) |future, i| {
37063706 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
3707 if (@atomicRmw(?*ResetEvent, &closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
3708 closure.reset_event.waitUncancelable(); // Ensure no reference to our stack-allocated reset_event.
3707 if (@atomicRmw(?*Io.Event, &closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_event) {
3708 closure.event.waitUncancelable(ioBasic(t)); // Ensure no reference to our stack-allocated event.
37093709 if (result == null) result = i; // In case multiple are ready, return first.
37103710 }
37113711 }
......@@ -6666,209 +6666,6 @@ fn copyCanon(canonical_name_buffer: *[HostName.max_len]u8, name: []const u8) Hos
66666666/// ulock_wait2() uses 64-bit nano-second timeouts (with the same convention)
66676667const darwin_supports_ulock_wait2 = builtin.os.version_range.semver.min.major >= 11;
66686668
6669/// A thread-safe logical boolean value which can be `set` and `unset`.
6670///
6671/// It can also block threads until the value is set with cancelation via timed
6672/// waits. Statically initializable; four bytes on all targets.
6673pub const ResetEvent = switch (native_os) {
6674 .illumos, .netbsd, .openbsd => ResetEventPosix,
6675 else => ResetEventFutex,
6676};
6677
6678/// A `ResetEvent` implementation based on futexes.
6679const ResetEventFutex = enum(u32) {
6680 unset = 0,
6681 waiting = 1,
6682 is_set = 2,
6683
6684 /// Returns whether the logical boolean is `set`.
6685 ///
6686 /// Once `reset` is called, this returns false until the next `set`.
6687 ///
6688 /// The memory accesses before the `set` can be said to happen before
6689 /// `isSet` returns true.
6690 pub fn isSet(ref: *const ResetEventFutex) bool {
6691 if (builtin.single_threaded) return switch (ref.*) {
6692 .unset => false,
6693 .waiting => unreachable,
6694 .is_set => true,
6695 };
6696 // Acquire barrier ensures memory accesses before `set` happen before
6697 // returning true.
6698 return @atomicLoad(ResetEventFutex, ref, .acquire) == .is_set;
6699 }
6700
6701 /// Blocks the calling thread until `set` is called.
6702 ///
6703 /// This is effectively a more efficient version of `while (!isSet()) {}`.
6704 ///
6705 /// The memory accesses before the `set` can be said to happen before `wait` returns.
6706 pub fn wait(ref: *ResetEventFutex, t: *Threaded) Io.Cancelable!void {
6707 if (builtin.single_threaded) switch (ref.*) {
6708 .unset => unreachable, // Deadlock, no other threads to wake us up.
6709 .waiting => unreachable, // Invalid state.
6710 .is_set => return,
6711 };
6712 // Try to set the state from `unset` to `waiting` to indicate to the
6713 // `set` thread that others are blocked on the ResetEventFutex. Avoid using
6714 // any strict barriers until we know the ResetEventFutex is set.
6715 var state = @atomicLoad(ResetEventFutex, ref, .acquire);
6716 if (state == .is_set) {
6717 @branchHint(.likely);
6718 return;
6719 }
6720 if (state == .unset) {
6721 state = @cmpxchgStrong(ResetEventFutex, ref, state, .waiting, .acquire, .acquire) orelse .waiting;
6722 }
6723 const current_thread = Thread.getCurrent(t);
6724 while (state == .waiting) {
6725 try current_thread.futexWait(@ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
6726 state = @atomicLoad(ResetEventFutex, ref, .acquire);
6727 }
6728 assert(state == .is_set);
6729 }
6730
6731 /// Same as `wait` except uninterruptible.
6732 pub fn waitUncancelable(ref: *ResetEventFutex) void {
6733 if (builtin.single_threaded) switch (ref.*) {
6734 .unset => unreachable, // Deadlock, no other threads to wake us up.
6735 .waiting => unreachable, // Invalid state.
6736 .is_set => return,
6737 };
6738 // Try to set the state from `unset` to `waiting` to indicate to the
6739 // `set` thread that others are blocked on the ResetEventFutex. Avoid using
6740 // any strict barriers until we know the ResetEventFutex is set.
6741 var state = @atomicLoad(ResetEventFutex, ref, .acquire);
6742 if (state == .is_set) {
6743 @branchHint(.likely);
6744 return;
6745 }
6746 if (state == .unset) {
6747 state = @cmpxchgStrong(ResetEventFutex, ref, state, .waiting, .acquire, .acquire) orelse .waiting;
6748 }
6749 while (state == .waiting) {
6750 Thread.futexWaitUncancelable(@ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
6751 state = @atomicLoad(ResetEventFutex, ref, .acquire);
6752 }
6753 assert(state == .is_set);
6754 }
6755
6756 /// Marks the logical boolean as `set` and unblocks any threads in `wait`
6757 /// or `timedWait` to observe the new state.
6758 ///
6759 /// The logical boolean stays `set` until `reset` is called, making future
6760 /// `set` calls do nothing semantically.
6761 ///
6762 /// The memory accesses before `set` can be said to happen before `isSet`
6763 /// returns true or `wait`/`timedWait` return successfully.
6764 pub fn set(ref: *ResetEventFutex) void {
6765 if (builtin.single_threaded) {
6766 ref.* = .is_set;
6767 return;
6768 }
6769 if (@atomicRmw(ResetEventFutex, ref, .Xchg, .is_set, .release) == .waiting) {
6770 Thread.futexWake(@ptrCast(ref), std.math.maxInt(u32));
6771 }
6772 }
6773
6774 /// Unmarks the ResetEventFutex as if `set` was never called.
6775 ///
6776 /// Assumes no threads are blocked in `wait` or `timedWait`. Concurrent
6777 /// calls to `set`, `isSet` and `reset` are allowed.
6778 pub fn reset(ref: *ResetEventFutex) void {
6779 if (builtin.single_threaded) {
6780 ref.* = .unset;
6781 return;
6782 }
6783 @atomicStore(ResetEventFutex, ref, .unset, .monotonic);
6784 }
6785};
6786
6787/// A `ResetEvent` implementation based on pthreads API.
6788const ResetEventPosix = struct {
6789 cond: std.c.pthread_cond_t,
6790 mutex: std.c.pthread_mutex_t,
6791 state: ResetEventFutex,
6792
6793 pub const unset: ResetEventPosix = .{
6794 .cond = std.c.PTHREAD_COND_INITIALIZER,
6795 .mutex = std.c.PTHREAD_MUTEX_INITIALIZER,
6796 .state = .unset,
6797 };
6798
6799 pub fn isSet(rep: *const ResetEventPosix) bool {
6800 if (builtin.single_threaded) return switch (rep.state) {
6801 .unset => false,
6802 .waiting => unreachable,
6803 .is_set => true,
6804 };
6805 return @atomicLoad(ResetEventFutex, &rep.state, .acquire) == .is_set;
6806 }
6807
6808 pub fn wait(rep: *ResetEventPosix, t: *Threaded) Io.Cancelable!void {
6809 if (builtin.single_threaded) switch (rep.*) {
6810 .unset => unreachable, // Deadlock, no other threads to wake us up.
6811 .waiting => unreachable, // Invalid state.
6812 .is_set => return,
6813 };
6814 const current_thread = Thread.getCurrent(t);
6815 assert(std.c.pthread_mutex_lock(&rep.mutex) == .SUCCESS);
6816 defer assert(std.c.pthread_mutex_unlock(&rep.mutex) == .SUCCESS);
6817 sw: switch (rep.state) {
6818 .unset => {
6819 rep.state = .waiting;
6820 continue :sw .waiting;
6821 },
6822 .waiting => {
6823 try current_thread.beginSyscall();
6824 assert(std.c.pthread_cond_wait(&rep.cond, &rep.mutex) == .SUCCESS);
6825 current_thread.endSyscall();
6826 continue :sw rep.state;
6827 },
6828 .is_set => return,
6829 }
6830 }
6831
6832 pub fn waitUncancelable(rep: *ResetEventPosix) void {
6833 if (builtin.single_threaded) switch (rep.*) {
6834 .unset => unreachable, // Deadlock, no other threads to wake us up.
6835 .waiting => unreachable, // Invalid state.
6836 .is_set => return,
6837 };
6838 assert(std.c.pthread_mutex_lock(&rep.mutex) == .SUCCESS);
6839 defer assert(std.c.pthread_mutex_unlock(&rep.mutex) == .SUCCESS);
6840 sw: switch (rep.state) {
6841 .unset => {
6842 rep.state = .waiting;
6843 continue :sw .waiting;
6844 },
6845 .waiting => {
6846 assert(std.c.pthread_cond_wait(&rep.cond, &rep.mutex) == .SUCCESS);
6847 continue :sw rep.state;
6848 },
6849 .is_set => return,
6850 }
6851 }
6852
6853 pub fn set(rep: *ResetEventPosix) void {
6854 if (builtin.single_threaded) {
6855 rep.* = .is_set;
6856 return;
6857 }
6858 if (@atomicRmw(ResetEventFutex, &rep.state, .Xchg, .is_set, .release) == .waiting) {
6859 assert(std.c.pthread_cond_broadcast(&rep.cond) == .SUCCESS);
6860 }
6861 }
6862
6863 pub fn reset(rep: *ResetEventPosix) void {
6864 if (builtin.single_threaded) {
6865 rep.* = .unset;
6866 return;
6867 }
6868 @atomicStore(ResetEventFutex, &rep.state, .unset, .monotonic);
6869 }
6870};
6871
68726669fn closeSocketWindows(s: ws2_32.SOCKET) void {
68736670 const rc = ws2_32.closesocket(s);
68746671 if (is_debug) switch (rc) {