Something went wrong. Try again.
fork of gitlab.com/devnw/zig/thread-pool for zig 0.17
Something went wrong. Try again.
6.8 kB · 175 lines
Zig
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176//! Internal sync primitives that preserve the zero-arg API of 0.15.2//! (`Mutex.lock/unlock`, `Condition.wait/signal/broadcast`, `Semaphore.wait/post`)//! on top of 0.16.0's `std.Io`-based primitives.//!//! Implementation: call the `*Uncancelable` variants on a static//! `std.Io.Threaded.global_single_threaded` instance. The futex vtable//! functions in `std.Io.Threaded` ignore their `userdata` parameter for the//! uncancelable path, so this is safe to call from any thread regardless of//! whether it is registered with a `Threaded` runtime.
const std = @import("std");
inline fn io() std.Io { return std.Io.Threaded.global_single_threaded.io();}
pub const Mutex = struct { inner: std.Io.Mutex = .init,
/// Blocks until the mutex is acquired (uncancelable). pub fn lock(self: *Mutex) void { self.inner.lockUncancelable(io()); }
/// Releases the mutex. Must be called by the thread that holds it. pub fn unlock(self: *Mutex) void { self.inner.unlock(io()); }
/// Attempts to acquire the mutex without blocking. Returns `true` on success. pub fn tryLock(self: *Mutex) bool { return self.inner.tryLock(); }};
pub const Condition = struct { state: std.atomic.Value(State) = .init(.{ .waiters = 0, .signals = 0 }), epoch: std.atomic.Value(u32) = .init(0),
const State = packed struct(u32) { waiters: u16, signals: u16, };
/// Atomically releases `m` and blocks until signalled. Reacquires `m` before returning. pub fn wait(self: *Condition, m: *Mutex) void { self.waitInner(m, .none) catch |err| switch (err) { error.Timeout => unreachable, }; }
/// Like `wait` but returns `error.Timeout` if `timeout_ns` elapses before a signal. /// A signal that arrives after the deadline but before the waiter leaves the /// wait is still consumed and reported as a wakeup (return without error), so /// no signal is ever orphaned by a timed-out waiter. pub fn timedWait(self: *Condition, m: *Mutex, timeout_ns: u64) error{Timeout}!void { return self.waitInner(m, .{ .duration = .{ .raw = .fromNanoseconds(@intCast(timeout_ns)), .clock = .awake, } }); }
fn waitInner(self: *Condition, m: *Mutex, timeout: std.Io.Timeout) error{Timeout}!void { var epoch = self.epoch.load(.acquire); { const prev_state = self.state.fetchAdd(.{ .waiters = 1, .signals = 0 }, .monotonic); std.debug.assert(prev_state.waiters < std.math.maxInt(u16)); }
m.inner.unlock(io()); defer m.inner.lockUncancelable(io());
const deadline: std.Io.Timeout = switch (timeout) { .none => .none, else => timeout.toDeadline(io()), };
while (true) { switch (deadline) { .none => io().futexWaitUncancelable(u32, &self.epoch.raw, epoch), else => io().futexWaitTimeout(u32, &self.epoch.raw, epoch, deadline) catch {}, }
epoch = self.epoch.load(.acquire);
// Try to consume a pending signal first. { var prev_state = self.state.load(.monotonic); while (prev_state.signals > 0) { prev_state = self.state.cmpxchgWeak(prev_state, .{ .waiters = prev_state.waiters - 1, .signals = prev_state.signals - 1, }, .acquire, .monotonic) orelse return; } }
// No signal: decide whether to loop or give up on timeout. switch (deadline) { .none => {}, // spurious wake; loop else => { const remaining = deadline.toDurationFromNow(io()) orelse break; if (remaining.raw.nanoseconds <= 0) break; }, } }
// Timed out. Leave the waiter count with a CAS, never a blind // fetchSub: signal()/broadcast() reserve a wake by bumping // `signals` while `waiters > signals`, so a reservation landing // after the check above would be orphaned (signals > waiters) and // later signal()/broadcast() calls would skip the next waiter // (SONNY-2427). If the CAS fails because a signal arrived, consume // it and report a wakeup — std.Thread.Condition's timeout path in // Zig 0.15. var state = self.state.load(.monotonic); while (true) { if (state.signals > 0) { state = self.state.cmpxchgWeak(state, .{ .waiters = state.waiters - 1, .signals = state.signals - 1, }, .acquire, .monotonic) orelse return; } else { std.debug.assert(state.waiters > 0); state = self.state.cmpxchgWeak(state, .{ .waiters = state.waiters - 1, .signals = 0, }, .monotonic, .monotonic) orelse return error.Timeout; } } }
/// Wakes at most one waiter blocked in `wait` or `timedWait`. pub fn signal(self: *Condition) void { var prev_state = self.state.load(.monotonic); while (prev_state.waiters > prev_state.signals) { prev_state = self.state.cmpxchgWeak(prev_state, .{ .waiters = prev_state.waiters, .signals = prev_state.signals + 1, }, .release, .monotonic) orelse { _ = self.epoch.fetchAdd(1, .release); io().futexWake(u32, &self.epoch.raw, 1); return; }; } }
/// Wakes all waiters blocked in `wait` or `timedWait`. pub fn broadcast(self: *Condition) void { var prev_state = self.state.load(.monotonic); while (prev_state.waiters > prev_state.signals) { prev_state = self.state.cmpxchgWeak(prev_state, .{ .waiters = prev_state.waiters, .signals = prev_state.waiters, }, .release, .monotonic) orelse { _ = self.epoch.fetchAdd(1, .release); io().futexWake(u32, &self.epoch.raw, prev_state.waiters - prev_state.signals); return; }; } }};
pub const Semaphore = struct { inner: std.Io.Semaphore = .{},
/// Blocks until the semaphore count is greater than zero, then decrements it (uncancelable). pub fn wait(self: *Semaphore) void { self.inner.waitUncancelable(io()); }
/// Increments the semaphore count, waking one waiter if any are blocked. pub fn post(self: *Semaphore) void { self.inner.post(io()); }};