diff --git a/lib/std/Io.zig b/lib/std/Io.zig index 666d00f01bc86f48783ff8206225139140912bbd..424f5a8e7006d3bb8f2d0e6675673ae8e9a30396 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -1386,6 +1386,109 @@ pub const Condition = struct { } }; +/// Logical boolean flag which can be set and unset and supports a "wait until set" operation. +pub const Event = enum(u32) { + unset, + waiting, + is_set, + + /// Returns whether the logical boolean is `true`. + pub fn isSet(event: *const Event) bool { + return switch (@atomicLoad(Event, event, .acquire)) { + .unset, .waiting => false, + .is_set => true, + }; + } + + /// Blocks until the logical boolean is `true`. + pub fn wait(event: *Event, io: Io) Io.Cancelable!void { + if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) { + .unset => unreachable, + .waiting => {}, + .is_set => return, + }; + errdefer { + // Ideally we would restore the event back to `.unset` instead of `.waiting`, but there + // might be other threads waiting on the event. In theory we could track the *number* of + // waiting threads in the unused bits of the `Event`, but that has its own problem: the + // waiters would wake up when a *new waiter* was added. So it's easiest to just leave + // the state at `.waiting`---at worst it causes one redundant call to `futexWake`. + } + while (true) { + try io.futexWait(Event, event, .waiting); + switch (@atomicLoad(Event, event, .acquire)) { + .unset => unreachable, // `reset` called before pending `wait` returned + .waiting => continue, + .is_set => return, + } + } + } + + /// Same as `wait` except uninterruptible. + pub fn waitUncancelable(event: *Event, io: Io) void { + if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) { + .unset => unreachable, + .waiting => {}, + .is_set => return, + }; + while (true) { + io.futexWaitUncancelable(Event, event, .waiting); + switch (@atomicLoad(Event, event, .acquire)) { + .unset => unreachable, // `reset` called before pending `wait` returned + .waiting => continue, + .is_set => return, + } + } + } + + /// Blocks the calling thread until either the logical boolean is set, the timeout expires, or a + /// spurious wakeup occurs. If the timeout expires or a spurious wakeup occurs, `error.Timeout` + /// is returned. + pub fn waitTimeout(event: *Event, io: Io, timeout: Timeout) (error{Timeout} || Cancelable)!void { + if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) { + .unset => unreachable, + .waiting => assert(!builtin.single_threaded), // invalid state + .is_set => return, + }; + errdefer { + // Ideally we would restore the event back to `.unset` instead of `.waiting`, but there + // might be other threads waiting on the event. In theory we could track the *number* of + // waiting threads in the unused bits of the `Event`, but that has its own problem: the + // waiters would wake up when a *new waiter* was added. So it's easiest to just leave + // the state at `.waiting`---at worst it causes one redundant call to `futexWake`. + } + io.futexWaitTimeout(Event, event, .waiting, timeout); + switch (@atomicLoad(Event, event, .acquire)) { + .unset => unreachable, // `reset` called before pending `wait` returned + .waiting => return error.Timeout, + .is_set => return, + } + } + + /// Sets the logical boolean to true, and hence unblocks any pending calls to `wait`. The + /// logical boolean remains true until `reset` is called, so future calls to `set` have no + /// semantic effect. + /// + /// Any memory accesses prior to a `set` call are "released", so that if this `set` call causes + /// `isSet` to return `true` or a wait to finish, those tasks will be able to observe those + /// memory accesses. + pub fn set(e: *Event, io: Io) void { + switch (@atomicRmw(Event, e, .Xchg, .is_set, .release)) { + .unset, .is_set => {}, + .waiting => io.futexWake(Event, e, std.math.maxInt(u32)), + } + } + + /// Sets the logical boolean to false. + /// + /// Assumes that there is no pending call to `wait` or `waitUncancelable`. + /// + /// However, concurrent calls to `isSet`, `set`, and `reset` are allowed. + pub fn reset(e: *Event) void { + @atomicStore(Event, e, .unset, .monotonic); + } +}; + pub const TypeErasedQueue = struct { mutex: Mutex, diff --git a/lib/std/Io/test.zig b/lib/std/Io/test.zig index 0ac6c333c9fb0bdd75f0d3a3231deedeb53ebe3a..c0a4445c1e34993406c39a642e7982d78d01a46a 100644 --- a/lib/std/Io/test.zig +++ b/lib/std/Io/test.zig @@ -255,3 +255,39 @@ test "Queue" { try testQueue(4); try testQueue(5); } + +test "Event" { + const global = struct { + fn waitAndRead(io: Io, event: *Io.Event, ptr: *const u32) Io.Cancelable!u32 { + try event.wait(io); + return ptr.*; + } + }; + + const io = std.testing.io; + + var event: Io.Event = .unset; + var buffer: u32 = undefined; + + { + var future = io.concurrent(global.waitAndRead, .{ io, &event, &buffer }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + + buffer = 123; + event.set(io); + + const result = try future.await(io); + + try std.testing.expectEqual(123, result); + } + + event.reset(); + + { + var future = io.concurrent(global.waitAndRead, .{ io, &event, &buffer }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + try std.testing.expectError(error.Canceled, future.cancel(io)); + } +}