Merge pull request #229 from lightpanda-io/cancellation

Cancellation

Author: krichprollsch

src/loop.zig

@@ -20,6 +20,8 @@ pub const IO = @import("tigerbeetle-io").IO;
 const public = @import("api.zig");
 const JSCallback = public.Callback;
 
+const log = std.log.scoped(.loop);
+
 fn report(comptime fmt: []const u8, args: anytype) void {
     const max_len = 200;
     var buf: [max_len]u8 = undefined;
@@ -113,7 +115,13 @@ pub const SingleThreaded = struct {
         defer ctx.loop.freeCbk(completion, ctx);
 
         // TODO: return the error to the callback
-        result catch |err| @panic(@errorName(err));
+        result catch |err| {
+            switch (err) {
+                error.Canceled => {},
+                else => log.err("timeout callback: {any}", .{err}),
+            }
+            return;
+        };
 
         const old_events_nb = ctx.loop.removeEvent();
         if (builtin.is_test) {
@@ -129,7 +137,7 @@ pub const SingleThreaded = struct {
         }
     }
 
-    pub fn timeout(self: *Self, nanoseconds: u63, js_cbk: ?JSCallback) void {
+    pub fn timeout(self: *Self, nanoseconds: u63, js_cbk: ?JSCallback) usize {
         const completion = self.alloc.create(IO.Completion) catch unreachable;
         completion.* = undefined;
         const ctx = self.alloc.create(ContextTimeout) catch unreachable;
@@ -142,6 +150,54 @@ pub const SingleThreaded = struct {
         if (builtin.is_test) {
             report("start timeout {d} for {d} nanoseconds", .{ old_events_nb + 1, nanoseconds });
         }
+
+        return @intFromPtr(completion);
+    }
+
+    const ContextCancel = struct {
+        loop: *Self,
+        js_cbk: ?JSCallback,
+    };
+
+    fn cancelCallback(
+        ctx: *ContextCancel,
+        completion: *IO.Completion,
+        result: IO.CancelError!void,
+    ) void {
+        defer ctx.loop.freeCbk(completion, ctx);
+
+        // TODO: return the error to the callback
+        result catch |err| {
+            log.err("cancel callback: {any}", .{err});
+            return;
+        };
+
+        const old_events_nb = ctx.loop.removeEvent();
+        if (builtin.is_test) {
+            report("timeout done, remaining events: {d}", .{old_events_nb - 1});
+        }
+
+        // js callback
+        if (ctx.js_cbk) |js_cbk| {
+            defer js_cbk.deinit(ctx.loop.alloc);
+            js_cbk.call(null) catch {
+                ctx.loop.cbk_error = true;
+            };
+        }
+    }
+
+    pub fn cancel(self: *Self, id: usize, js_cbk: ?JSCallback) void {
+        const comp_cancel: *IO.Completion = @ptrFromInt(id);
+
+        const completion = self.alloc.create(IO.Completion) catch unreachable;
+        completion.* = undefined;
+        const ctx = self.alloc.create(ContextCancel) catch unreachable;
+        ctx.* = ContextCancel{
+            .loop = self,
+            .js_cbk = js_cbk,
+        };
+
+        self.io.cancel(*ContextCancel, ctx, cancelCallback, completion, comp_cancel);
     }
 
     // Yield

src/tests/cbk_test.zig

@@ -52,7 +52,7 @@ pub const Window = struct {
     ) void {
         const n: u63 = @intCast(milliseconds);
         // TODO: check this value can be holded in u63
-        loop.timeout(n * std.time.ns_per_ms, callback);
+        _ = loop.timeout(n * std.time.ns_per_ms, callback);
     }
 
     pub fn _cbkAsyncWithJSArg(
@@ -64,7 +64,7 @@ pub const Window = struct {
     ) void {
         const n: u63 = @intCast(milliseconds);
         // TODO: check this value can be holded in u63
-        loop.timeout(n * std.time.ns_per_ms, callback);
+        _ = loop.timeout(n * std.time.ns_per_ms, callback);
     }
 
     pub fn _cbkAsyncWithNatArg(_: Window, callback: Callback) !void {