90 lines
3.3 KiB
Zig
90 lines
3.3 KiB
Zig
//! A futex word that counts releases, so a task can wait for "something was
|
|
//! released since I looked" with a deadline on its own thread.
|
|
//!
|
|
//! `std.Io.Condition` and `std.Io.Semaphore` have no timed wait in 0.16, and
|
|
//! racing one against a timer costs two extra tasks per waiter. A waiter reads
|
|
//! `observe` *before* it checks for a free resource; a release bumps the word
|
|
//! *after* freeing one. A release that lands between the check and the wait
|
|
//! therefore changes the word, and the wait returns at once instead of
|
|
//! sleeping through it.
|
|
//!
|
|
//! Like any futex, a wait can return spuriously, so callers loop on their own
|
|
//! condition. A waiter that leaves without taking what it was woken for — a
|
|
//! timeout or a cancel that consumed a `wakeOne` — must call `wakeOne` again
|
|
//! if something is still free, or a peer sleeps through it.
|
|
|
|
const std = @import("std");
|
|
|
|
pub const WakeWord = struct {
|
|
releases: std.atomic.Value(u32) = .init(0),
|
|
|
|
pub fn observe(self: *const WakeWord) u32 {
|
|
return self.releases.load(.acquire);
|
|
}
|
|
|
|
/// Returns on a wake, a spurious wake, or `deadline`; `null` waits for a
|
|
/// wake only. Returns at once when the word has moved past `seen`.
|
|
pub fn wait(
|
|
self: *const WakeWord,
|
|
io: std.Io,
|
|
seen: u32,
|
|
deadline: ?std.Io.Clock.Timestamp,
|
|
) std.Io.Cancelable!void {
|
|
const timeout: std.Io.Timeout = if (deadline) |at| .{ .deadline = at } else .none;
|
|
return io.futexWaitTimeout(u32, &self.releases.raw, seen, timeout);
|
|
}
|
|
|
|
pub fn wakeOne(self: *WakeWord, io: std.Io) void {
|
|
_ = self.releases.fetchAdd(1, .release);
|
|
io.futexWake(u32, &self.releases.raw, 1);
|
|
}
|
|
|
|
pub fn wakeAll(self: *WakeWord, io: std.Io) void {
|
|
_ = self.releases.fetchAdd(1, .release);
|
|
io.futexWake(u32, &self.releases.raw, std.math.maxInt(u32));
|
|
}
|
|
};
|
|
|
|
const testing = std.testing;
|
|
|
|
test "a release between observe and wait is not slept through" {
|
|
var word: WakeWord = .{};
|
|
const seen = word.observe();
|
|
word.wakeOne(testing.io);
|
|
// No deadline: a lost wake would hang this test.
|
|
try word.wait(testing.io, seen, null);
|
|
}
|
|
|
|
test "a wait with nothing released returns at its deadline" {
|
|
const io = testing.io;
|
|
var word: WakeWord = .{};
|
|
const deadline: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromMilliseconds(20), .clock = .awake });
|
|
try word.wait(io, word.observe(), deadline);
|
|
try testing.expect(std.Io.Clock.Timestamp.now(io, .awake).compare(.gte, deadline));
|
|
}
|
|
|
|
fn waitForever(word: *const WakeWord, io: std.Io, seen: u32) std.Io.Cancelable!void {
|
|
return word.wait(io, seen, null);
|
|
}
|
|
|
|
test "wakeAll releases every waiter" {
|
|
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
|
defer threaded.deinit();
|
|
const io = threaded.io();
|
|
|
|
var word: WakeWord = .{};
|
|
const seen = word.observe();
|
|
var first = io.concurrent(waitForever, .{ &word, io, seen }) catch |err| switch (err) {
|
|
error.ConcurrencyUnavailable => return error.SkipZigTest,
|
|
};
|
|
var second = io.concurrent(waitForever, .{ &word, io, seen }) catch |err| switch (err) {
|
|
error.ConcurrencyUnavailable => {
|
|
_ = first.cancel(io) catch {};
|
|
return error.SkipZigTest;
|
|
},
|
|
};
|
|
word.wakeAll(io);
|
|
try first.await(io);
|
|
try second.await(io);
|
|
}
|