ReviewOS
Reload

also looking at this

home-lang/home

fix(async): preserve cancellation and add revision-aware job gate

#914
Merged chris wants to merge feat/camera-jobs into main
6 files +217 -20
.github/workflows/async-job-state.ymladded+33-0
@@ -0,0 +1,33 @@
1name: Async job terminal states
2on:
3 push:
4 paths:
5 - 'packages/async/src/task.zig'
6 - 'packages/async/src/job_state.zig'
7 - 'packages/async/src/completion_gate.zig'
8 - '.github/workflows/async-job-state.yml'
9 pull_request:
10 paths:
11 - 'packages/async/src/task.zig'
12 - 'packages/async/src/job_state.zig'
13 - 'packages/async/src/completion_gate.zig'
14 - '.github/workflows/async-job-state.yml'
15 workflow_dispatch:
16permissions:
17 contents: read
18jobs:
19 state:
20 strategy:
21 matrix:
22 os: [ubuntu-24.04, ubuntu-24.04-arm]
23 runs-on: ${{ matrix.os }}
24 steps:
25 - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
26 - uses: mlugg/setup-zig@d1434d08867e3ee9daa34448df10607b98908d29
27 with:
28 version: 0.15.2
29 - run: |
30 for mode in Debug ReleaseSafe; do
31 zig test packages/async/src/task.zig -lc -O "$mode"
32 zig test packages/async/src/completion_gate.zig -O "$mode"
33 done
packages/async/README.mdmodified+25-0
@@ -96,3 +96,28 @@ scripts/async-runtime-tsan.sh 1 10000 1024
9696## License
9797
9898Part of the Home programming language project.
99
100## Terminal job state and event completion
101
102Task cancellation is terminal: an in-flight future poll returning Ready or Pending
103cannot overwrite Cancelled, and cancelling an already completed/failed task does
104not replace its result. JoinHandle.await returns TaskFailed immediately for a
105failed task. Cancellation does not interrupt a currently executing poll or make
106its storage safe to free; quiesce/join the polling owner before deinit.
107
108`CompletionGate` is a small allocation-free two-phase event-job helper. It ignores
109events from older revisions and terminal outcomes before an active phase has been
110observed. Success, failure, cancellation and stop are immutable terminal states.
111The caller owns synchronization; it is not an executor or a generic Future.
112zig-utils/zig-camera uses this Home module for autofocus scan gating under its
113native session mutex without allocating or running consumer work in callbacks.
114
115Regression tests cover cancellation between poll reservation and both Ready and
116Pending results, late cancellation, failed joins, stale event revisions, and
117terminal immutability. Focused task/gate tests pass on Zig 0.15.2 and Home's current
1180.17 development toolchain. The complete async root suite also passes locally with
119its threading_futex module supplied:
120
121```sh
122zig test -lc --dep threading_futex -Mroot=packages/async/src/async.zig -Mthreading_futex=packages/runtime/src/threading/Futex.zig
123```
packages/async/src/async.zigmodified+4-0
@@ -16,6 +16,7 @@ pub const RawTask = task.RawTask;
1616pub const Runtime = runtime.Runtime;
1717
1818test {
19 _ = ;
1920 _ = future;
2021 _ = task;
2122 _ = runtime;
@@ -23,3 +24,6 @@ test {
2324 _ = work_stealing_deque;
2425 _ = parker;
2526}
27
28/// Revision-aware event job gate; caller owns synchronization.
29pub const CompletionGate = .Gate;
packages/async/src/completion_gate.zigadded+57-0
@@ -0,0 +1,57 @@
1//! Revision-aware two-phase job completion. Small allocation-free adaptation of
2//! Home task states for event-driven jobs; caller owns synchronization.
3//! A terminal outcome is immutable, including under cancellation/late events.
4const std = ;
5pub const State = enum(u32) { pending = 1, active = 2, succeeded = 3, failed = 4, cancelled = 5, stopped = 6 };
6pub const Gate = struct {
7 revision: u64,
8 state: State = .pending,
9 pub fn terminal(self: Gate) bool {
10 return switch (self.state) {
11 .pending, .active => false,
12 else => true,
13 };
14 }
15 pub fn observe(self: *Gate, revision: u64, active: bool, outcome: ?bool) void {
16 if (self.terminal() or revision < self.revision) return;
17 if (active) self.state = .active;
18 if (self.state == .active) {
19 if (outcome) |success| {
20 self.state = if (success) .succeeded else .failed;
21 }
22 }
23 }
24 pub fn cancel(self: *Gate) void {
25 if (!self.terminal()) self.state = .cancelled;
26 }
27 pub fn stop(self: *Gate) void {
28 if (!self.terminal()) self.state = .stopped;
29 }
30};
31test "gate ignores stale and pre-active results; cancellation is terminal" {
32 var g: Gate = .{ .revision = 9 };
33 g.observe(8, true, null);
34 g.observe(9, false, true);
35 try std.testing.expectEqual(State.pending, g.state);
36 g.observe(9, true, null);
37 try std.testing.expectEqual(State.active, g.state);
38 g.cancel();
39 g.observe(10, false, true);
40 try std.testing.expectEqual(State.cancelled, g.state);
41}
42test "success failure and stop never overwrite a terminal result" {
43 var g: Gate = .{ .revision = 1 };
44 g.observe(1, true, null);
45 g.observe(2, false, true);
46 g.cancel();
47 g.stop();
48 try std.testing.expectEqual(State.succeeded, g.state);
49 g = .{ .revision = 1 };
50 g.observe(1, true, null);
51 g.observe(1, false, false);
52 try std.testing.expectEqual(State.failed, g.state);
53 g = .{ .revision = 1 };
54 g.stop();
55 g.observe(1, true, true);
56 try std.testing.expectEqual(State.stopped, g.state);
57}
packages/async/src/job_state.zigadded+44-0
@@ -0,0 +1,44 @@
1//! Shared atomic task transitions. Cancellation cannot be undone by an in-flight
2//! poll. Terminal states are immutable. Storage must outlive the polling owner.
3const std = ;
4pub const State = enum(u8) { Pending, Running, Completed, Failed, Cancelled };
5pub fn terminal(state: State) bool {
6 return switch (state) {
7 .Completed, .Failed, .Cancelled => true,
8 else => false,
9 };
10}
11pub fn cancel(state: *std.atomic.Value(State)) bool {
12 var current = state.load(.acquire);
13 while (!terminal(current)) {
14 current = state.cmpxchgStrong(current, .Cancelled, .acq_rel, .acquire) orelse return true;
15 }
16 return false;
17}
18/// Called only by the poll owner, after publishing any result payload.
19pub fn finishPoll(state: *std.atomic.Value(State), ready: bool) bool {
20 const next: State = if (ready) .Completed else .Pending;
21 if (state.cmpxchgStrong(.Running, next, .release, .monotonic)) |actual| return terminal(actual);
22 return ready;
23}
24test "cancel between poll reservation and completion cannot be overwritten" {
25 for ([_]bool{ false, true }) |ready| {
26 var state = std.atomic.Value(State).init(.Running);
27 try std.testing.expect(cancel(&state));
28 try std.testing.expect(finishPoll(&state, ready));
29 try std.testing.expectEqual(State.Cancelled, state.load(.acquire));
30 }
31}
32test "completion and failure are immutable and pending polls remain resumable" {
33 for ([_]State{ .Completed, .Failed, .Cancelled }) |terminal_state| {
34 var state = std.atomic.Value(State).init(terminal_state);
35 try std.testing.expect(!cancel(&state));
36 try std.testing.expectEqual(terminal_state, state.load(.acquire));
37 }
38 var state = std.atomic.Value(State).init(.Running);
39 try std.testing.expect(!finishPoll(&state, false));
40 try std.testing.expectEqual(State.Pending, state.load(.acquire));
41 state.store(.Running, .release);
42 try std.testing.expect(finishPoll(&state, true));
43 try std.testing.expectEqual(State.Completed, state.load(.acquire));
44}
packages/async/src/task.zigmodified+54-20
@@ -16,18 +16,8 @@ pub const TaskId = struct {
1616};
1717
1818/// Task state
19pub const TaskState = enum(u8) {
20 /// Task is ready to run
21 Pending,
22 /// Task is currently running
23 Running,
24 /// Task completed successfully
25 Completed,
26 /// Task failed with an error
27 Failed,
28 /// Task was cancelled
29 Cancelled,
30};
19const job_state = ;
20pub const TaskState = job_state.State;
3121
3222/// A spawned task that can be awaited
3323///
@@ -58,6 +48,8 @@ pub fn Task(comptime T: type) type {
5848 return task;
5949 }
6050
51 /// Call only after the poll owner/executor has quiesced. Cancellation
52 /// marks an outcome; it does not stop a currently executing poll.
6153 pub fn deinit(self: *Self) void {
6254 self.future.deinit();
6355 self.allocator.destroy(self);
@@ -88,19 +80,17 @@ pub fn Task(comptime T: type) type {
8880 switch (result) {
8981 .Ready => |value| {
9082 self.result = value;
91 self.state.store(.Completed, .release);
92 return true;
83 return job_state.finishPoll(&self.state, true);
9384 },
9485 .Pending => {
95 self.state.store(.Pending, .release);
96 return false;
86 return job_state.finishPoll(&self.state, false);
9787 },
9888 }
9989 }
10090
10191 /// Cancel the task
10292 pub fn cancel(self: *Self) void {
103 self.state.store(.Cancelled, .release);
93 _ = job_state.cancel(&self.state);
10494 }
10595
10696 /// Check if task is completed
@@ -137,9 +127,9 @@ pub fn JoinHandle(comptime T: type) type {
137127 /// Only use this from non-async code or the runtime's block_on.
138128 pub fn await(self: Self) !T {
139129 while (!self.task.isCompleted()) {
140 if (self.task.isCancelled()) {
141 return error.TaskCancelled;
142 }
130 const state = self.task.state.load(.acquire);
131 if (state == .Cancelled) return error.TaskCancelled;
132 if (state == .Failed) return error.TaskFailed;
143133
144134 // Spin or yield
145135 std.Thread.yield() catch {};
@@ -381,3 +371,47 @@ test "RawTask - type erasure" {
381371 try testing.expect(completed);
382372 try testing.expectEqual(@as(i32, 77), task.result.?);
383373}
374
375fn regressionContext() Context {
376 const Noop = struct {
377 fn wake(_: *anyopaque) void {}
378 fn clone(ptr: *anyopaque) *anyopaque {
379 return ptr;
380 }
381 };
382 return Context.init(.{ .data = undefined, .vtable = &.{ .wake = Noop.wake, .wake_by_ref = Noop.wake, .clone = Noop.clone, .drop = Noop.wake } });
383}
384test "cancel during a future poll survives both Ready and Pending returns" {
385 const Poller = struct {
386 task: *Task(u32) = undefined,
387 ready: bool,
388 fn poll(ptr: *anyopaque, _: *Context) future_mod.PollResult(u32) {
389 const self: *@This() = @ptrCast(@alignCast(ptr));
390 self.task.cancel();
391 return if (self.ready) .{ .Ready = 42 } else .Pending;
392 }
393 };
394 for ([_]bool{ false, true }) |ready| {
395 var poller: Poller = .{ .ready = ready };
396 const task = try Task(u32).init(std.testing.allocator, .{ .state = &poller, .poll_fn = Poller.poll });
397 defer task.deinit();
398 poller.task = task;
399 var ctx = regressionContext();
400 try std.testing.expect(task.poll(&ctx));
401 try std.testing.expect(task.isCancelled());
402 try std.testing.expect(!task.isCompleted());
403 const handle: JoinHandle(u32) = .{ .task = task };
404 try std.testing.expectError(error.TaskCancelled, handle.await());
405 }
406}
407test "completed tasks resist late cancellation and failed joins return immediately" {
408 const task = try Task(u32).init(std.testing.allocator, try future_mod.ready(u32, 42, std.testing.allocator));
409 defer task.deinit();
410 var ctx = regressionContext();
411 try std.testing.expect(task.poll(&ctx));
412 task.cancel();
413 const handle: JoinHandle(u32) = .{ .task = task };
414 try std.testing.expectEqual(@as(u32, 42), try handle.await());
415 task.state.store(.Failed, .release);
416 try std.testing.expectError(error.TaskFailed, handle.await());
417}