authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-27 21:17:44-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:52-07:00
loga28d3059e60ccbed2b2b159c41899488330bc671
tree23ec1390a300a498e24e191cf2b1d5dfafb46816
parent30448d92af596724dc227c96803961bc41de9253

std.Io.Threaded: implement ResetEvent in terms of pthreads

needed for NetBSD

2 files changed, 123 insertions(+), 34 deletions(-)

lib/std/Io/Threaded.zig+121-32
......@@ -5787,7 +5787,13 @@ pub fn futexWake(ptr: *const std.atomic.Value(u32), max_waiters: u32) void {
57875787///
57885788/// It can also block threads until the value is set with cancelation via timed
57895789/// waits. Statically initializable; four bytes on all targets.
5790pub const ResetEvent = enum(u32) {
5790pub const ResetEvent = switch (native_os) {
5791 .netbsd => ResetEventPosix,
5792 else => ResetEventFutex,
5793};
5794
5795/// A `ResetEvent` implementation based on futexes.
5796const ResetEventFutex = enum(u32) {
57915797 unset = 0,
57925798 waiting = 1,
57935799 is_set = 2,
......@@ -5798,15 +5804,15 @@ pub const ResetEvent = enum(u32) {
57985804 ///
57995805 /// The memory accesses before the `set` can be said to happen before
58005806 /// `isSet` returns true.
5801 pub fn isSet(re: *const ResetEvent) bool {
5802 if (builtin.single_threaded) return switch (re.*) {
5807 pub fn isSet(ref: *const ResetEventFutex) bool {
5808 if (builtin.single_threaded) return switch (ref.*) {
58035809 .unset => false,
58045810 .waiting => unreachable,
58055811 .is_set => true,
58065812 };
58075813 // Acquire barrier ensures memory accesses before `set` happen before
58085814 // returning true.
5809 return @atomicLoad(ResetEvent, re, .acquire) == .is_set;
5815 return @atomicLoad(ResetEventFutex, ref, .acquire) == .is_set;
58105816 }
58115817
58125818 /// Blocks the calling thread until `set` is called.
......@@ -5814,51 +5820,51 @@ pub const ResetEvent = enum(u32) {
58145820 /// This is effectively a more efficient version of `while (!isSet()) {}`.
58155821 ///
58165822 /// The memory accesses before the `set` can be said to happen before `wait` returns.
5817 pub fn wait(re: *ResetEvent, t: *Threaded) Io.Cancelable!void {
5818 if (builtin.single_threaded) switch (re.*) {
5823 pub fn wait(ref: *ResetEventFutex, t: *Threaded) Io.Cancelable!void {
5824 if (builtin.single_threaded) switch (ref.*) {
58195825 .unset => unreachable, // Deadlock, no other threads to wake us up.
58205826 .waiting => unreachable, // Invalid state.
58215827 .is_set => return,
58225828 };
5823 if (re.isSet()) {
5829 // Try to set the state from `unset` to `waiting` to indicate to the
5830 // `set` thread that others are blocked on the ResetEventFutex. Avoid using
5831 // any strict barriers until we know the ResetEventFutex is set.
5832 var state = @atomicLoad(ResetEventFutex, ref, .acquire);
5833 if (state == .is_set) {
58245834 @branchHint(.likely);
58255835 return;
58265836 }
5827 // Try to set the state from `unset` to `waiting` to indicate to the
5828 // `set` thread that others are blocked on the ResetEvent. Avoid using
5829 // any strict barriers until we know the ResetEvent is set.
5830 var state = @atomicLoad(ResetEvent, re, .acquire);
58315837 if (state == .unset) {
5832 state = @cmpxchgStrong(ResetEvent, re, state, .waiting, .acquire, .acquire) orelse .waiting;
5838 state = @cmpxchgStrong(ResetEventFutex, ref, state, .waiting, .acquire, .acquire) orelse .waiting;
58335839 }
58345840 while (state == .waiting) {
5835 try futexWait(t, @ptrCast(re), @intFromEnum(ResetEvent.waiting));
5836 state = @atomicLoad(ResetEvent, re, .acquire);
5841 try futexWait(t, @ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
5842 state = @atomicLoad(ResetEventFutex, ref, .acquire);
58375843 }
58385844 assert(state == .is_set);
58395845 }
58405846
58415847 /// Same as `wait` except uninterruptible.
5842 pub fn waitUncancelable(re: *ResetEvent) void {
5843 if (builtin.single_threaded) switch (re.*) {
5848 pub fn waitUncancelable(ref: *ResetEventFutex) void {
5849 if (builtin.single_threaded) switch (ref.*) {
58445850 .unset => unreachable, // Deadlock, no other threads to wake us up.
58455851 .waiting => unreachable, // Invalid state.
58465852 .is_set => return,
58475853 };
5848 if (re.isSet()) {
5854 // Try to set the state from `unset` to `waiting` to indicate to the
5855 // `set` thread that others are blocked on the ResetEventFutex. Avoid using
5856 // any strict barriers until we know the ResetEventFutex is set.
5857 var state = @atomicLoad(ResetEventFutex, ref, .acquire);
5858 if (state == .is_set) {
58495859 @branchHint(.likely);
58505860 return;
58515861 }
5852 // Try to set the state from `unset` to `waiting` to indicate to the
5853 // `set` thread that others are blocked on the ResetEvent. Avoid using
5854 // any strict barriers until we know the ResetEvent is set.
5855 var state = @atomicLoad(ResetEvent, re, .acquire);
58565862 if (state == .unset) {
5857 state = @cmpxchgStrong(ResetEvent, re, state, .waiting, .acquire, .acquire) orelse .waiting;
5863 state = @cmpxchgStrong(ResetEventFutex, ref, state, .waiting, .acquire, .acquire) orelse .waiting;
58585864 }
58595865 while (state == .waiting) {
5860 futexWaitUncancelable(@ptrCast(re), @intFromEnum(ResetEvent.waiting));
5861 state = @atomicLoad(ResetEvent, re, .acquire);
5866 futexWaitUncancelable(@ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
5867 state = @atomicLoad(ResetEventFutex, ref, .acquire);
58625868 }
58635869 assert(state == .is_set);
58645870 }
......@@ -5871,26 +5877,109 @@ pub const ResetEvent = enum(u32) {
58715877 ///
58725878 /// The memory accesses before `set` can be said to happen before `isSet`
58735879 /// returns true or `wait`/`timedWait` return successfully.
5874 pub fn set(re: *ResetEvent) void {
5880 pub fn set(ref: *ResetEventFutex) void {
58755881 if (builtin.single_threaded) {
5876 re.* = .is_set;
5882 ref.* = .is_set;
58775883 return;
58785884 }
5879 if (@atomicRmw(ResetEvent, re, .Xchg, .is_set, .release) == .waiting) {
5880 futexWake(@ptrCast(re), std.math.maxInt(u32));
5885 if (@atomicRmw(ResetEventFutex, ref, .Xchg, .is_set, .release) == .waiting) {
5886 futexWake(@ptrCast(ref), std.math.maxInt(u32));
58815887 }
58825888 }
58835889
5884 /// Unmarks the ResetEvent as if `set` was never called.
5890 /// Unmarks the ResetEventFutex as if `set` was never called.
58855891 ///
58865892 /// Assumes no threads are blocked in `wait` or `timedWait`. Concurrent
58875893 /// calls to `set`, `isSet` and `reset` are allowed.
5888 pub fn reset(re: *ResetEvent) void {
5894 pub fn reset(ref: *ResetEventFutex) void {
5895 if (builtin.single_threaded) {
5896 ref.* = .unset;
5897 return;
5898 }
5899 @atomicStore(ResetEventFutex, ref, .unset, .monotonic);
5900 }
5901};
5902
5903/// A `ResetEvent` implementation based on pthreads API.
5904const ResetEventPosix = struct {
5905 cond: std.c.pthread_cond_t,
5906 mutex: std.c.pthread_mutex_t,
5907 state: ResetEventFutex,
5908
5909 pub const unset: ResetEventPosix = .{
5910 .cond = std.c.PTHREAD_COND_INITIALIZER,
5911 .mutex = std.c.PTHREAD_MUTEX_INITIALIZER,
5912 .state = .unset,
5913 };
5914
5915 pub fn isSet(rep: *const ResetEventPosix) bool {
5916 if (builtin.single_threaded) return switch (rep.state) {
5917 .unset => false,
5918 .waiting => unreachable,
5919 .is_set => true,
5920 };
5921 return @atomicLoad(ResetEventFutex, &rep.state, .acquire) == .is_set;
5922 }
5923
5924 pub fn wait(rep: *ResetEventPosix, t: *Threaded) Io.Cancelable!void {
5925 if (builtin.single_threaded) switch (rep.*) {
5926 .unset => unreachable, // Deadlock, no other threads to wake us up.
5927 .waiting => unreachable, // Invalid state.
5928 .is_set => return,
5929 };
5930 assert(std.c.pthread_mutex_lock(&rep.mutex) == .SUCCESS);
5931 defer assert(std.c.pthread_mutex_unlock(&rep.mutex) == .SUCCESS);
5932 sw: switch (rep.state) {
5933 .unset => {
5934 rep.state = .waiting;
5935 continue :sw .waiting;
5936 },
5937 .waiting => {
5938 try t.checkCancel();
5939 assert(std.c.pthread_cond_wait(&rep.cond, &rep.mutex) == .SUCCESS);
5940 continue :sw rep.state;
5941 },
5942 .is_set => return,
5943 }
5944 }
5945
5946 pub fn waitUncancelable(rep: *ResetEventPosix) void {
5947 if (builtin.single_threaded) switch (rep.*) {
5948 .unset => unreachable, // Deadlock, no other threads to wake us up.
5949 .waiting => unreachable, // Invalid state.
5950 .is_set => return,
5951 };
5952 assert(std.c.pthread_mutex_lock(&rep.mutex) == .SUCCESS);
5953 defer assert(std.c.pthread_mutex_unlock(&rep.mutex) == .SUCCESS);
5954 sw: switch (rep.state) {
5955 .unset => {
5956 rep.state = .waiting;
5957 continue :sw .waiting;
5958 },
5959 .waiting => {
5960 assert(std.c.pthread_cond_wait(&rep.cond, &rep.mutex) == .SUCCESS);
5961 continue :sw rep.state;
5962 },
5963 .is_set => return,
5964 }
5965 }
5966
5967 pub fn set(rep: *ResetEventPosix) void {
5968 if (builtin.single_threaded) {
5969 rep.* = .is_set;
5970 return;
5971 }
5972 if (@atomicRmw(ResetEventFutex, &rep.state, .Xchg, .is_set, .release) == .waiting) {
5973 assert(std.c.pthread_cond_broadcast(&rep.cond) == .SUCCESS);
5974 }
5975 }
5976
5977 pub fn reset(rep: *ResetEventPosix) void {
58895978 if (builtin.single_threaded) {
5890 re.* = .unset;
5979 rep.* = .unset;
58915980 return;
58925981 }
5893 @atomicStore(ResetEvent, re, .unset, .monotonic);
5982 @atomicStore(ResetEventFutex, &rep.state, .unset, .monotonic);
58945983 }
58955984};
58965985
lib/std/c.zig+2-2
......@@ -10874,13 +10874,13 @@ pub extern "c" fn dn_expand(
1087410874 length: c_int,
1087510875) c_int;
1087610876
10877pub const PTHREAD_MUTEX_INITIALIZER = pthread_mutex_t{};
10877pub const PTHREAD_MUTEX_INITIALIZER: pthread_mutex_t = .{};
1087810878pub extern "c" fn pthread_mutex_lock(mutex: *pthread_mutex_t) E;
1087910879pub extern "c" fn pthread_mutex_unlock(mutex: *pthread_mutex_t) E;
1088010880pub extern "c" fn pthread_mutex_trylock(mutex: *pthread_mutex_t) E;
1088110881pub extern "c" fn pthread_mutex_destroy(mutex: *pthread_mutex_t) E;
1088210882
10883pub const PTHREAD_COND_INITIALIZER = pthread_cond_t{};
10883pub const PTHREAD_COND_INITIALIZER: pthread_cond_t = .{};
1088410884pub extern "c" fn pthread_cond_wait(noalias cond: *pthread_cond_t, noalias mutex: *pthread_mutex_t) E;
1088510885pub extern "c" fn pthread_cond_timedwait(noalias cond: *pthread_cond_t, noalias mutex: *pthread_mutex_t, noalias abstime: *const timespec) E;
1088610886pub extern "c" fn pthread_cond_signal(cond: *pthread_cond_t) E;