atproto pds in zig pds.zat.dev
pds atproto
Something went wrong. Try again.
Zig
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586const std = @import("std");
const Io = std.Io;
var store_io: Io = undefined;var initialized = false;var mutex: Io.Mutex = .init;var changed: Io.Condition = .init;var generation: usize = 0;var latest_seq: u64 = 0;
pub fn init(io: Io) void { store_io = io; initialized = true;}
pub fn publish(seq: u64) void { if (!initialized) return;
mutex.lockUncancelable(store_io); defer mutex.unlock(store_io);
_ = @atomicRmw(usize, &generation, .Add, 1, .release); latest_seq = @max(latest_seq, seq); changed.broadcast(store_io);}
pub fn currentGeneration() usize { return @atomicLoad(usize, &generation, .acquire);}
fn snapshotLocked() Snapshot { return .{ .generation = currentGeneration(), .latest_seq = latest_seq, };}
pub fn waitForChange(observed: usize, stopped: *const bool) !?Snapshot { try mutex.lock(store_io); defer mutex.unlock(store_io);
while (@atomicLoad(usize, &generation, .acquire) == observed and !@atomicLoad(bool, stopped, .acquire)) { try changed.wait(store_io, &mutex); } if (@atomicLoad(bool, stopped, .acquire)) return null; return snapshotLocked();}
pub fn wakeWaiters() void { if (!initialized) return;
mutex.lockUncancelable(store_io); defer mutex.unlock(store_io); changed.broadcast(store_io);}
pub const Snapshot = struct { generation: usize, latest_seq: u64,};
test "wakeWaiters releases a stopped idle waiter" { init(std.Options.debug_io);
var stopped = false; var returned = false; const observed = currentGeneration(); const thread = try std.Thread.spawn(.{}, struct { fn run(seen: usize, stop: *bool, done: *bool) void { const result = waitForChange(seen, stop) catch unreachable; std.debug.assert(result == null); @atomicStore(bool, done, true, .release); } }.run, .{ observed, &stopped, &returned });
std.Options.debug_io.sleep(.fromMilliseconds(10), .awake) catch unreachable; @atomicStore(bool, &stopped, true, .release); wakeWaiters(); thread.join();
try std.testing.expect(@atomicLoad(bool, &returned, .acquire));}