const std = @import("../std.zig");
const builtin = @import("builtin");
const assert = std.debug.assert;
const testing = std.testing;
const Lock = std.event.Lock;
pub fn Future(comptime T: type) type {
return struct {
lock: Lock,
data: T,
available: Available,
const Available = enum(u8) {
NotStarted,
Started,
Finished,
};
const Self = @This();
const Queue = std.atomic.Queue(anyframe);
pub fn init() Self {
return Self{
.lock = Lock.initLocked(),
.available = .NotStarted,
.data = undefined,
};
}
pub fn get(self: *Self) callconv(.Async) *T {
if (@atomicLoad(Available, &self.available, .SeqCst) == .Finished) {
return &self.data;
}
const held = self.lock.acquire();
held.release();
return &self.data;
}
pub fn getOrNull(self: *Self) ?*T {
if (@atomicLoad(Available, &self.available, .SeqCst) == .Finished) {
return &self.data;
} else {
return null;
}
}
pub fn start(self: *Self) callconv(.Async) ?*T {
const state = @cmpxchgStrong(Available, &self.available, .NotStarted, .Started, .SeqCst, .SeqCst) orelse return null;
switch (state) {
.Started => {
const held = self.lock.acquire();
held.release();
return &self.data;
},
.Finished => return &self.data,
else => unreachable,
}
}
pub fn resolve(self: *Self) void {
const prev = @atomicRmw(Available, &self.available, .Xchg, .Finished, .SeqCst);
assert(prev != .Finished);
Lock.Held.release(Lock.Held{ .lock = &self.lock });
}
};
}
test "std.event.Future" {
if (builtin.single_threaded) return error.SkipZigTest;
if (builtin.os.tag == .freebsd) return error.SkipZigTest;
if (!std.io.is_async) return error.SkipZigTest;
testFuture();
}
fn testFuture() void {
var future = Future(i32).init();
var a = async waitOnFuture(&future);
var b = async waitOnFuture(&future);
resolveFuture(&future);
const result = (await a) + (await b);
try testing.expect(result == 12);
}
fn waitOnFuture(future: *Future(i32)) i32 {
return future.get().*;
}
fn resolveFuture(future: *Future(i32)) void {
future.data = 6;
future.resolve();
}