add std.event.Future

This is like a promise, but it's for multiple getters, and uses an event loop.
ziglang · Jul 11, 2018 · 9bdcd2a · 9bdcd2a
1 parent 5954c94
commit 9bdcd2a
Show file tree

Hide file tree

Showing 7 changed files with 132 additions and 11 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -460,6 +460,7 @@ set(ZIG_STD_FILES
     "empty.zig"
     "event.zig"
     "event/channel.zig"
+    "event/future.zig"
     "event/group.zig"
     "event/lock.zig"
     "event/locked.zig"

diff --git a/src-self-hosted/module.zig b/src-self-hosted/module.zig
@@ -381,6 +381,7 @@ pub const Module = struct {
 
         if (is_export) {
             try self.build_group.call(verifyUniqueSymbol, self, parsed_file, decl);
+            try self.build_group.call(generateDecl, self, parsed_file, decl);
         }
     }
 
@@ -429,6 +430,22 @@ pub const Module = struct {
         }
     }
 
+    /// This declaration has been blessed as going into the final code generation.
+    async fn generateDecl(self: *Module, parsed_file: *ParsedFile, decl: *Decl) void {
+        switch (decl.id) {
+            Decl.Id.Var => @panic("TODO"),
+            Decl.Id.Fn => {
+                const fn_decl = @fieldParentPtr(Decl.Fn, "base", decl);
+                return await (async self.generateDeclFn(parsed_file, fn_decl) catch unreachable);
+            },
+            Decl.Id.CompTime => @panic("TODO"),
+        }
+    }
+
+    async fn generateDeclFn(self: *Module, parsed_file: *ParsedFile, fn_decl: *Decl.Fn) void {
+        fn_decl.value = Decl.Fn.Val{ .Ok = Value.Fn{} };
+    }
+
     pub fn link(self: *Module, out_file: ?[]const u8) !void {
         warn("TODO link");
         return error.Todo;
@@ -589,7 +606,7 @@ pub const Decl = struct {
         // TODO https://github.com/ziglang/zig/issues/683 and then make this anonymous
         pub const Val = union {
             Unresolved: void,
-            Ok: *Value.Fn,
+            Ok: Value.Fn,
         };
 
         pub fn externLibName(self: Fn, tree: *ast.Tree) ?[]const u8 {

diff --git a/src-self-hosted/test.zig b/src-self-hosted/test.zig
@@ -12,22 +12,15 @@ test "compile errors" {
     try ctx.init();
     defer ctx.deinit();
 
-    try ctx.testCompileError(
-        \\export fn entry() void {}
-        \\export fn entry() void {}
-    , file1, 2, 8, "exported symbol collision: 'entry'");
-
-    try ctx.testCompileError(
-        \\fn() void {}
-    , file1, 1, 1, "missing function name");
+    try @import("../test/stage2/compile_errors.zig").addCases(&ctx);
 
     try ctx.run();
 }
 
 const file1 = "1.zig";
 const allocator = std.heap.c_allocator;
 
-const TestContext = struct {
+pub const TestContext = struct {
     loop: std.event.Loop,
     zig_lib_dir: []u8,
     zig_cache_dir: []u8,

diff --git a/std/event.zig b/std/event.zig
@@ -4,6 +4,7 @@ pub const Lock = @import("event/lock.zig").Lock;
 pub const tcp = @import("event/tcp.zig");
 pub const Channel = @import("event/channel.zig").Channel;
 pub const Group = @import("event/group.zig").Group;
+pub const Future = @import("event/future.zig").Group;
 
 test "import event tests" {
     _ = @import("event/locked.zig");
@@ -12,4 +13,5 @@ test "import event tests" {
     _ = @import("event/tcp.zig");
     _ = @import("event/channel.zig");
     _ = @import("event/group.zig");
+    _ = @import("event/future.zig");
 }
diff --git a/std/event/future.zig b/std/event/future.zig
@@ -0,0 +1,87 @@
+const std = @import("../index.zig");
+const assert = std.debug.assert;
+const builtin = @import("builtin");
+const AtomicRmwOp = builtin.AtomicRmwOp;
+const AtomicOrder = builtin.AtomicOrder;
+const Lock = std.event.Lock;
+const Loop = std.event.Loop;
+
+/// This is a value that starts out unavailable, until a value is put().
+/// While it is unavailable, coroutines suspend when they try to get() it,
+/// and then are resumed when the value is put().
+/// At this point the value remains forever available, and another put() is not allowed.
+pub fn Future(comptime T: type) type {
+    return struct {
+        lock: Lock,
+        data: T,
+        available: u8, // TODO make this a bool
+
+        const Self = this;
+        const Queue = std.atomic.QueueMpsc(promise);
+
+        pub fn init(loop: *Loop) Self {
+            return Self{
+                .lock = Lock.initLocked(loop),
+                .available = 0,
+                .data = undefined,
+            };
+        }
+
+        /// Obtain the value. If it's not available, wait until it becomes
+        /// available.
+        /// Thread-safe.
+        pub async fn get(self: *Self) T {
+            if (@atomicLoad(u8, &self.available, AtomicOrder.SeqCst) == 1) {
+                return self.data;
+            }
+            const held = await (async self.lock.acquire() catch unreachable);
+            defer held.release();
+
+            return self.data;
+        }
+
+        /// Make the data become available. May be called only once.
+        pub fn put(self: *Self, value: T) void {
+            self.data = value;
+            const prev = @atomicRmw(u8, &self.available, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
+            assert(prev == 0); // put() called twice
+            Lock.Held.release(Lock.Held{ .lock = &self.lock });
+        }
+    };
+}
+
+test "std.event.Future" {
+    var da = std.heap.DirectAllocator.init();
+    defer da.deinit();
+
+    const allocator = &da.allocator;
+
+    var loop: Loop = undefined;
+    try loop.initMultiThreaded(allocator);
+    defer loop.deinit();
+
+    const handle = try async<allocator> testFuture(&loop);
+    defer cancel handle;
+
+    loop.run();
+}
+
+async fn testFuture(loop: *Loop) void {
+    var future = Future(i32).init(loop);
+
+    const a = async waitOnFuture(&future) catch @panic("memory");
+    const b = async waitOnFuture(&future) catch @panic("memory");
+    const c = async resolveFuture(&future) catch @panic("memory");
+
+    const result = (await a) + (await b);
+    cancel c;
+    assert(result == 12);
+}
+
+async fn waitOnFuture(future: *Future(i32)) i32 {
+    return await (async future.get() catch @panic("memory"));
+}
+
+async fn resolveFuture(future: *Future(i32)) void {
+    future.put(6);
+}
diff --git a/std/event/lock.zig b/std/event/lock.zig
@@ -73,6 +73,15 @@ pub const Lock = struct {
         };
     }
 
+    pub fn initLocked(loop: *Loop) Lock {
+        return Lock{
+            .loop = loop,
+            .shared_bit = 1,
+            .queue = Queue.init(),
+            .queue_empty_bit = 1,
+        };
+    }
+
     /// Must be called when not locked. Not thread safe.
     /// All calls to acquire() and release() must complete before calling deinit().
     pub fn deinit(self: *Lock) void {
@@ -81,7 +90,7 @@ pub const Lock = struct {
     }
 
     pub async fn acquire(self: *Lock) Held {
-        s: suspend |handle| {
+        suspend |handle| {
             // TODO explicitly put this memory in the coroutine frame #1194
             var my_tick_node = Loop.NextTickNode{
                 .data = handle,

diff --git a/test/stage2/compile_errors.zig b/test/stage2/compile_errors.zig
@@ -0,0 +1,12 @@
+const TestContext = @import("../../src-self-hosted/test.zig").TestContext;
+
+pub fn addCases(ctx: *TestContext) !void {
+    try ctx.testCompileError(
+        \\export fn entry() void {}
+        \\export fn entry() void {}
+    , "1.zig", 2, 8, "exported symbol collision: 'entry'");
+
+    try ctx.testCompileError(
+        \\fn() void {}
+    , "1.zig", 1, 1, "missing function name");
+}