authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-09-29 14:04:07-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
loga37c0bca2248e5d2e18c4855ee8d8d17bf26aa26
treeead0450f58070aa43a6fa3953976ff2e6a8c4f52
parent3c9fdf810f6f1517193786998ac5efe9b2b2c276

std.Io.Threaded: implement Group.cancel

2 files changed, 191 insertions(+), 174 deletions(-)

lib/std/Io.zig+15-7
...@@ -736,10 +736,9 @@ pub fn Future(Result: type) type {...@@ -736,10 +736,9 @@ pub fn Future(Result: type) type {
736 any_future: ?*AnyFuture,736 any_future: ?*AnyFuture,
737 result: Result,737 result: Result,
738738
739 /// Equivalent to `await` but sets a flag observable to application739 /// Equivalent to `await` but places a cancellation request.
740 /// code that cancellation has been requested.
741 ///740 ///
742 /// Idempotent.741 /// Idempotent. Not threadsafe.
743 pub fn cancel(f: *@This(), io: Io) Result {742 pub fn cancel(f: *@This(), io: Io) Result {
744 const any_future = f.any_future orelse return f.result;743 const any_future = f.any_future orelse return f.result;
745 io.vtable.cancel(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));744 io.vtable.cancel(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));
...@@ -747,6 +746,7 @@ pub fn Future(Result: type) type {...@@ -747,6 +746,7 @@ pub fn Future(Result: type) type {
747 return f.result;746 return f.result;
748 }747 }
749748
749 /// Idempotent. Not threadsafe.
750 pub fn await(f: *@This(), io: Io) Result {750 pub fn await(f: *@This(), io: Io) Result {
751 const any_future = f.any_future orelse return f.result;751 const any_future = f.any_future orelse return f.result;
752 io.vtable.await(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));752 io.vtable.await(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));
...@@ -759,8 +759,9 @@ pub fn Future(Result: type) type {...@@ -759,8 +759,9 @@ pub fn Future(Result: type) type {
759pub const Group = struct {759pub const Group = struct {
760 state: usize,760 state: usize,
761 context: ?*anyopaque,761 context: ?*anyopaque,
762 token: ?*anyopaque,
762763
763 pub const init: Group = .{ .state = 0, .context = null };764 pub const init: Group = .{ .state = 0, .context = null, .token = null };
764765
765 /// Calls `function` with `args` asynchronously. The resource spawned is766 /// Calls `function` with `args` asynchronously. The resource spawned is
766 /// owned by the group.767 /// owned by the group.
...@@ -771,7 +772,7 @@ pub const Group = struct {...@@ -771,7 +772,7 @@ pub const Group = struct {
771 /// deinitialized.772 /// deinitialized.
772 ///773 ///
773 /// See also:774 /// See also:
774 /// * `async`775 /// * `Io.async`
775 /// * `concurrent`776 /// * `concurrent`
776 pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {777 pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {
777 const Args = @TypeOf(args);778 const Args = @TypeOf(args);
...@@ -784,14 +785,21 @@ pub const Group = struct {...@@ -784,14 +785,21 @@ pub const Group = struct {
784 io.vtable.groupAsync(io.userdata, g, @ptrCast((&args)[0..1]), .of(Args), TypeErased.start);785 io.vtable.groupAsync(io.userdata, g, @ptrCast((&args)[0..1]), .of(Args), TypeErased.start);
785 }786 }
786787
787 /// Idempotent.788 /// Blocks until all tasks of the group finish.
789 ///
790 /// Idempotent. Not threadsafe.
788 pub fn wait(g: *Group, io: Io) void {791 pub fn wait(g: *Group, io: Io) void {
789 io.vtable.groupWait(io.userdata, g);792 io.vtable.groupWait(io.userdata, g);
790 }793 }
791794
792 /// Idempotent.795 /// Equivalent to `wait` but requests cancellation on all tasks owned by
796 /// the group.
797 ///
798 /// Idempotent. Not threadsafe.
793 pub fn cancel(g: *Group, io: Io) void {799 pub fn cancel(g: *Group, io: Io) void {
800 if (g.token == null) return;
794 io.vtable.groupCancel(io.userdata, g);801 io.vtable.groupCancel(io.userdata, g);
802 assert(g.token == null);
795 }803 }
796};804};
797805
lib/std/Io/Threaded.zig+176-167
...@@ -10,6 +10,7 @@ const Allocator = std.mem.Allocator;...@@ -10,6 +10,7 @@ const Allocator = std.mem.Allocator;
10const assert = std.debug.assert;10const assert = std.debug.assert;
11const posix = std.posix;11const posix = std.posix;
12const Io = std.Io;12const Io = std.Io;
13const ResetEvent = std.Thread.ResetEvent;
1314
14/// Thread-safe.15/// Thread-safe.
15allocator: Allocator,16allocator: Allocator,
...@@ -20,9 +21,9 @@ join_requested: bool = false,...@@ -20,9 +21,9 @@ join_requested: bool = false,
20threads: std.ArrayListUnmanaged(std.Thread),21threads: std.ArrayListUnmanaged(std.Thread),
21stack_size: usize,22stack_size: usize,
22cpu_count: std.Thread.CpuCountError!usize,23cpu_count: std.Thread.CpuCountError!usize,
23parallel_count: usize,24concurrent_count: usize,
2425
25threadlocal var current_closure: ?*AsyncClosure = null;26threadlocal var current_closure: ?*Closure = null;
2627
27const max_iovecs_len = 8;28const max_iovecs_len = 8;
28const splat_buffer_size = 64;29const splat_buffer_size = 64;
...@@ -31,12 +32,33 @@ comptime {...@@ -31,12 +32,33 @@ comptime {
31 assert(max_iovecs_len <= posix.IOV_MAX);32 assert(max_iovecs_len <= posix.IOV_MAX);
32}33}
3334
34pub const Runnable = struct {35const Closure = struct {
35 start: Start,36 start: Start,
36 node: std.SinglyLinkedList.Node = .{},37 node: std.SinglyLinkedList.Node = .{},
37 is_parallel: bool,38 cancel_tid: std.Thread.Id,
39 /// Whether this task bumps minimum number of threads in the pool.
40 is_concurrent: bool,
41
42 const Start = *const fn (*Closure) void;
43
44 const canceling_tid: std.Thread.Id = switch (@typeInfo(std.Thread.Id)) {
45 .int => |int_info| switch (int_info.signedness) {
46 .signed => -1,
47 .unsigned => std.math.maxInt(std.Thread.Id),
48 },
49 .pointer => @ptrFromInt(std.math.maxInt(usize)),
50 else => @compileError("unsupported std.Thread.Id: " ++ @typeName(std.Thread.Id)),
51 };
3852
39 pub const Start = *const fn (*Runnable) void;53 fn requestCancel(closure: *Closure) void {
54 switch (@atomicRmw(std.Thread.Id, &closure.cancel_tid, .Xchg, canceling_tid, .acq_rel)) {
55 0, canceling_tid => {},
56 else => |tid| switch (builtin.os.tag) {
57 .linux => _ = std.os.linux.tgkill(std.os.linux.getpid(), @bitCast(tid), posix.SIG.IO),
58 else => {},
59 },
60 }
61 }
40};62};
4163
42pub const InitError = std.Thread.CpuCountError || Allocator.Error;64pub const InitError = std.Thread.CpuCountError || Allocator.Error;
...@@ -47,7 +69,7 @@ pub fn init(gpa: Allocator) Pool {...@@ -47,7 +69,7 @@ pub fn init(gpa: Allocator) Pool {
47 .threads = .empty,69 .threads = .empty,
48 .stack_size = std.Thread.SpawnConfig.default_stack_size,70 .stack_size = std.Thread.SpawnConfig.default_stack_size,
49 .cpu_count = std.Thread.getCpuCount(),71 .cpu_count = std.Thread.getCpuCount(),
50 .parallel_count = 0,72 .concurrent_count = 0,
51 };73 };
52 if (pool.cpu_count) |n| {74 if (pool.cpu_count) |n| {
53 pool.threads.ensureTotalCapacityPrecise(gpa, n - 1) catch {};75 pool.threads.ensureTotalCapacityPrecise(gpa, n - 1) catch {};
...@@ -78,14 +100,15 @@ fn worker(pool: *Pool) void {...@@ -78,14 +100,15 @@ fn worker(pool: *Pool) void {
78 defer pool.mutex.unlock();100 defer pool.mutex.unlock();
79101
80 while (true) {102 while (true) {
81 while (pool.run_queue.popFirst()) |run_node| {103 while (pool.run_queue.popFirst()) |closure_node| {
82 pool.mutex.unlock();104 pool.mutex.unlock();
83 const runnable: *Runnable = @fieldParentPtr("node", run_node);105 const closure: *Closure = @fieldParentPtr("node", closure_node);
84 runnable.start(runnable);106 const is_concurrent = closure.is_concurrent;
107 closure.start(closure);
85 pool.mutex.lock();108 pool.mutex.lock();
86 if (runnable.is_parallel) {109 if (is_concurrent) {
87 // TODO also pop thread and join sometimes110 // TODO also pop thread and join sometimes
88 pool.parallel_count -= 1;111 pool.concurrent_count -= 1;
89 }112 }
90 }113 }
91 if (pool.join_requested) break;114 if (pool.join_requested) break;
...@@ -154,97 +177,71 @@ pub fn io(pool: *Pool) Io {...@@ -154,97 +177,71 @@ pub fn io(pool: *Pool) Io {
154 };177 };
155}178}
156179
180/// Trailing data:
181/// 1. context
182/// 2. result
157const AsyncClosure = struct {183const AsyncClosure = struct {
184 closure: Closure,
158 func: *const fn (context: *anyopaque, result: *anyopaque) void,185 func: *const fn (context: *anyopaque, result: *anyopaque) void,
159 runnable: Runnable,186 reset_event: ResetEvent,
160 reset_event: std.Thread.ResetEvent,187 select_condition: ?*ResetEvent,
161 select_condition: ?*std.Thread.ResetEvent,188 context_alignment: std.mem.Alignment,
162 cancel_tid: std.Thread.Id,
163 context_offset: usize,
164 result_offset: usize,189 result_offset: usize,
190 /// Whether the task has a return type with nonzero bits.
191 has_result: bool,
165192
166 const done_reset_event: *std.Thread.ResetEvent = @ptrFromInt(@alignOf(std.Thread.ResetEvent));193 const done_reset_event: *ResetEvent = @ptrFromInt(@alignOf(ResetEvent));
167194
168 const canceling_tid: std.Thread.Id = switch (@typeInfo(std.Thread.Id)) {195 fn start(closure: *Closure) void {
169 .int => |int_info| switch (int_info.signedness) {196 const ac: *AsyncClosure = @alignCast(@fieldParentPtr("closure", closure));
170 .signed => -1,
171 .unsigned => std.math.maxInt(std.Thread.Id),
172 },
173 .pointer => @ptrFromInt(std.math.maxInt(usize)),
174 else => @compileError("unsupported std.Thread.Id: " ++ @typeName(std.Thread.Id)),
175 };
176
177 fn start(runnable: *Runnable) void {
178 const closure: *AsyncClosure = @alignCast(@fieldParentPtr("runnable", runnable));
179 const tid = std.Thread.getCurrentId();197 const tid = std.Thread.getCurrentId();
180 if (@cmpxchgStrong(198 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, 0, tid, .acq_rel, .acquire)) |cancel_tid| {
181 std.Thread.Id,199 assert(cancel_tid == Closure.canceling_tid);
182 &closure.cancel_tid,200 // Even though we already know the task is canceled, we must still
183 0,201 // run the closure in order to make the return value valid - that
184 tid,202 // is, unless the result is zero bytes!
185 .acq_rel,203 if (!ac.has_result) {
186 .acquire,204 ac.reset_event.set();
187 )) |cancel_tid| {205 return;
188 assert(cancel_tid == canceling_tid);206 }
189 closure.reset_event.set();
190 return;
191 }207 }
192 current_closure = closure;208 current_closure = closure;
193 closure.func(closure.contextPointer(), closure.resultPointer());209 ac.func(ac.contextPointer(), ac.resultPointer());
194 current_closure = null;210 current_closure = null;
195 if (@cmpxchgStrong(211
196 std.Thread.Id,212 // In case a cancel happens after successful task completion, prevents
197 &closure.cancel_tid,213 // signal from being delivered to the thread in `requestCancel`.
198 tid,214 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, tid, 0, .acq_rel, .acquire)) |cancel_tid| {
199 0,215 assert(cancel_tid == Closure.canceling_tid);
200 .acq_rel,216 }
201 .acquire,217
202 )) |cancel_tid| assert(cancel_tid == canceling_tid);218 if (@atomicRmw(?*ResetEvent, &ac.select_condition, .Xchg, done_reset_event, .release)) |select_reset| {
203
204 if (@atomicRmw(
205 ?*std.Thread.ResetEvent,
206 &closure.select_condition,
207 .Xchg,
208 done_reset_event,
209 .release,
210 )) |select_reset| {
211 assert(select_reset != done_reset_event);219 assert(select_reset != done_reset_event);
212 select_reset.set();220 select_reset.set();
213 }221 }
214 closure.reset_event.set();222 ac.reset_event.set();
215 }
216
217 fn contextOffset(context_alignment: std.mem.Alignment) usize {
218 return context_alignment.forward(@sizeOf(AsyncClosure));
219 }
220
221 fn resultOffset(
222 context_alignment: std.mem.Alignment,
223 context_len: usize,
224 result_alignment: std.mem.Alignment,
225 ) usize {
226 return result_alignment.forward(contextOffset(context_alignment) + context_len);
227 }223 }
228224
229 fn resultPointer(closure: *AsyncClosure) [*]u8 {225 fn resultPointer(ac: *AsyncClosure) [*]u8 {
230 const base: [*]u8 = @ptrCast(closure);226 const base: [*]u8 = @ptrCast(ac);
231 return base + closure.result_offset;227 return base + ac.result_offset;
232 }228 }
233229
234 fn contextPointer(closure: *AsyncClosure) [*]u8 {230 fn contextPointer(ac: *AsyncClosure) [*]u8 {
235 const base: [*]u8 = @ptrCast(closure);231 const base: [*]u8 = @ptrCast(ac);
236 return base + closure.context_offset;232 return base + ac.context_alignment.forward(@sizeOf(AsyncClosure));
237 }233 }
238234
239 fn waitAndFree(closure: *AsyncClosure, gpa: Allocator, result: []u8) void {235 fn waitAndFree(ac: *AsyncClosure, gpa: Allocator, result: []u8) void {
240 closure.reset_event.wait();236 ac.reset_event.wait();
241 @memcpy(result, closure.resultPointer()[0..result.len]);237 @memcpy(result, ac.resultPointer()[0..result.len]);
242 free(closure, gpa, result.len);238 free(ac, gpa, result.len);
243 }239 }
244240
245 fn free(closure: *AsyncClosure, gpa: Allocator, result_len: usize) void {241 fn free(ac: *AsyncClosure, gpa: Allocator, result_len: usize) void {
246 const base: [*]align(@alignOf(AsyncClosure)) u8 = @ptrCast(closure);242 if (!ac.has_result) assert(result_len == 0);
247 gpa.free(base[0 .. closure.result_offset + result_len]);243 const base: [*]align(@alignOf(AsyncClosure)) u8 = @ptrCast(ac);
244 gpa.free(base[0 .. ac.result_offset + result_len]);
248 }245 }
249};246};
250247
...@@ -271,59 +268,60 @@ fn async(...@@ -271,59 +268,60 @@ fn async(
271 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));268 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));
272 const result_offset = result_alignment.forward(context_offset + context.len);269 const result_offset = result_alignment.forward(context_offset + context.len);
273 const n = result_offset + result.len;270 const n = result_offset + result.len;
274 const closure: *AsyncClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(AsyncClosure), n) catch {271 const ac: *AsyncClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(AsyncClosure), n) catch {
275 start(context.ptr, result.ptr);272 start(context.ptr, result.ptr);
276 return null;273 return null;
277 }));274 }));
278275
279 closure.* = .{276 ac.* = .{
277 .closure = .{
278 .cancel_tid = 0,
279 .start = AsyncClosure.start,
280 .is_concurrent = false,
281 },
280 .func = start,282 .func = start,
281 .context_offset = context_offset,283 .context_alignment = context_alignment,
282 .result_offset = result_offset,284 .result_offset = result_offset,
285 .has_result = result.len != 0,
283 .reset_event = .unset,286 .reset_event = .unset,
284 .cancel_tid = 0,
285 .select_condition = null,287 .select_condition = null,
286 .runnable = .{
287 .start = AsyncClosure.start,
288 .is_parallel = false,
289 },
290 };288 };
291289
292 @memcpy(closure.contextPointer()[0..context.len], context);290 @memcpy(ac.contextPointer()[0..context.len], context);
293291
294 pool.mutex.lock();292 pool.mutex.lock();
295293
296 const thread_capacity = cpu_count - 1 + pool.parallel_count;294 const thread_capacity = cpu_count - 1 + pool.concurrent_count;
297295
298 pool.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {296 pool.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {
299 pool.mutex.unlock();297 pool.mutex.unlock();
300 closure.free(gpa, result.len);298 ac.free(gpa, result.len);
301 start(context.ptr, result.ptr);299 start(context.ptr, result.ptr);
302 return null;300 return null;
303 };301 };
304302
305 pool.run_queue.prepend(&closure.runnable.node);303 pool.run_queue.prepend(&ac.closure.node);
306304
307 if (pool.threads.items.len < thread_capacity) {305 if (pool.threads.items.len < thread_capacity) {
308 const thread = std.Thread.spawn(.{ .stack_size = pool.stack_size }, worker, .{pool}) catch {306 const thread = std.Thread.spawn(.{ .stack_size = pool.stack_size }, worker, .{pool}) catch {
309 if (pool.threads.items.len == 0) {307 if (pool.threads.items.len == 0) {
310 assert(pool.run_queue.popFirst() == &closure.runnable.node);308 assert(pool.run_queue.popFirst() == &ac.closure.node);
311 pool.mutex.unlock();309 pool.mutex.unlock();
312 closure.free(gpa, result.len);310 ac.free(gpa, result.len);
313 start(context.ptr, result.ptr);311 start(context.ptr, result.ptr);
314 return null;312 return null;
315 }313 }
316 // Rely on other workers to do it.314 // Rely on other workers to do it.
317 pool.mutex.unlock();315 pool.mutex.unlock();
318 pool.cond.signal();316 pool.cond.signal();
319 return @ptrCast(closure);317 return @ptrCast(ac);
320 };318 };
321 pool.threads.appendAssumeCapacity(thread);319 pool.threads.appendAssumeCapacity(thread);
322 }320 }
323321
324 pool.mutex.unlock();322 pool.mutex.unlock();
325 pool.cond.signal();323 pool.cond.signal();
326 return @ptrCast(closure);324 return @ptrCast(ac);
327}325}
328326
329fn concurrent(327fn concurrent(
...@@ -342,40 +340,41 @@ fn concurrent(...@@ -342,40 +340,41 @@ fn concurrent(
342 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));340 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));
343 const result_offset = result_alignment.forward(context_offset + context.len);341 const result_offset = result_alignment.forward(context_offset + context.len);
344 const n = result_offset + result_len;342 const n = result_offset + result_len;
345 const closure: *AsyncClosure = @ptrCast(@alignCast(try gpa.alignedAlloc(u8, .of(AsyncClosure), n)));343 const ac: *AsyncClosure = @ptrCast(@alignCast(try gpa.alignedAlloc(u8, .of(AsyncClosure), n)));
346344
347 closure.* = .{345 ac.* = .{
346 .closure = .{
347 .cancel_tid = 0,
348 .start = AsyncClosure.start,
349 .is_concurrent = true,
350 },
348 .func = start,351 .func = start,
349 .context_offset = context_offset,352 .context_alignment = context_alignment,
350 .result_offset = result_offset,353 .result_offset = result_offset,
354 .has_result = result_len != 0,
351 .reset_event = .unset,355 .reset_event = .unset,
352 .cancel_tid = 0,
353 .select_condition = null,356 .select_condition = null,
354 .runnable = .{
355 .start = AsyncClosure.start,
356 .is_parallel = true,
357 },
358 };357 };
359 @memcpy(closure.contextPointer()[0..context.len], context);358 @memcpy(ac.contextPointer()[0..context.len], context);
360359
361 pool.mutex.lock();360 pool.mutex.lock();
362361
363 pool.parallel_count += 1;362 pool.concurrent_count += 1;
364 const thread_capacity = cpu_count - 1 + pool.parallel_count;363 const thread_capacity = cpu_count - 1 + pool.concurrent_count;
365364
366 pool.threads.ensureTotalCapacity(gpa, thread_capacity) catch {365 pool.threads.ensureTotalCapacity(gpa, thread_capacity) catch {
367 pool.mutex.unlock();366 pool.mutex.unlock();
368 closure.free(gpa, result_len);367 ac.free(gpa, result_len);
369 return error.OutOfMemory;368 return error.OutOfMemory;
370 };369 };
371370
372 pool.run_queue.prepend(&closure.runnable.node);371 pool.run_queue.prepend(&ac.closure.node);
373372
374 if (pool.threads.items.len < thread_capacity) {373 if (pool.threads.items.len < thread_capacity) {
375 const thread = std.Thread.spawn(.{ .stack_size = pool.stack_size }, worker, .{pool}) catch {374 const thread = std.Thread.spawn(.{ .stack_size = pool.stack_size }, worker, .{pool}) catch {
376 assert(pool.run_queue.popFirst() == &closure.runnable.node);375 assert(pool.run_queue.popFirst() == &ac.closure.node);
377 pool.mutex.unlock();376 pool.mutex.unlock();
378 closure.free(gpa, result_len);377 ac.free(gpa, result_len);
379 return error.OutOfMemory;378 return error.OutOfMemory;
380 };379 };
381 pool.threads.appendAssumeCapacity(thread);380 pool.threads.appendAssumeCapacity(thread);
...@@ -383,31 +382,48 @@ fn concurrent(...@@ -383,31 +382,48 @@ fn concurrent(
383382
384 pool.mutex.unlock();383 pool.mutex.unlock();
385 pool.cond.signal();384 pool.cond.signal();
386 return @ptrCast(closure);385 return @ptrCast(ac);
387}386}
388387
389const GroupClosure = struct {388const GroupClosure = struct {
389 closure: Closure,
390 pool: *Pool,390 pool: *Pool,
391 group: *Io.Group,391 group: *Io.Group,
392 /// Points to sibling `GroupClosure`. Used for walking the group to cancel all.
393 node: std.SinglyLinkedList.Node,
392 func: *const fn (context: *anyopaque) void,394 func: *const fn (context: *anyopaque) void,
393 runnable: Runnable,
394 context_alignment: std.mem.Alignment,395 context_alignment: std.mem.Alignment,
395 context_len: usize,396 context_len: usize,
396397
397 fn start(runnable: *Runnable) void {398 fn start(closure: *Closure) void {
398 const closure: *GroupClosure = @alignCast(@fieldParentPtr("runnable", runnable));399 const gc: *GroupClosure = @alignCast(@fieldParentPtr("closure", closure));
399 closure.func(closure.contextPointer());400 const tid = std.Thread.getCurrentId();
400 const group = closure.group;401 const group = gc.group;
401 const gpa = closure.pool.allocator;
402 free(closure, gpa);
403 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);402 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
404 const reset_event: *std.Thread.ResetEvent = @ptrCast(&group.context);403 const reset_event: *ResetEvent = @ptrCast(&group.context);
404 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, 0, tid, .acq_rel, .acquire)) |cancel_tid| {
405 assert(cancel_tid == Closure.canceling_tid);
406 // We already know the task is canceled before running the callback. Since all closures
407 // in a Group have void return type, we can return early.
408 std.Thread.WaitGroup.finishStateless(group_state, reset_event);
409 return;
410 }
411 current_closure = closure;
412 gc.func(gc.contextPointer());
413 current_closure = null;
414
415 // In case a cancel happens after successful task completion, prevents
416 // signal from being delivered to the thread in `requestCancel`.
417 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, tid, 0, .acq_rel, .acquire)) |cancel_tid| {
418 assert(cancel_tid == Closure.canceling_tid);
419 }
420
405 std.Thread.WaitGroup.finishStateless(group_state, reset_event);421 std.Thread.WaitGroup.finishStateless(group_state, reset_event);
406 }422 }
407423
408 fn free(closure: *GroupClosure, gpa: Allocator) void {424 fn free(gc: *GroupClosure, gpa: Allocator) void {
409 const base: [*]align(@alignOf(GroupClosure)) u8 = @ptrCast(closure);425 const base: [*]align(@alignOf(GroupClosure)) u8 = @ptrCast(gc);
410 gpa.free(base[0..contextEnd(closure.context_alignment, closure.context_len)]);426 gpa.free(base[0..contextEnd(gc.context_alignment, gc.context_len)]);
411 }427 }
412428
413 fn contextOffset(context_alignment: std.mem.Alignment) usize {429 fn contextOffset(context_alignment: std.mem.Alignment) usize {
...@@ -418,9 +434,9 @@ const GroupClosure = struct {...@@ -418,9 +434,9 @@ const GroupClosure = struct {
418 return contextOffset(context_alignment) + context_len;434 return contextOffset(context_alignment) + context_len;
419 }435 }
420436
421 fn contextPointer(closure: *GroupClosure) [*]u8 {437 fn contextPointer(gc: *GroupClosure) [*]u8 {
422 const base: [*]u8 = @ptrCast(closure);438 const base: [*]u8 = @ptrCast(gc);
423 return base + contextOffset(closure.context_alignment);439 return base + contextOffset(gc.context_alignment);
424 }440 }
425};441};
426442
...@@ -436,39 +452,42 @@ fn groupAsync(...@@ -436,39 +452,42 @@ fn groupAsync(
436 const cpu_count = pool.cpu_count catch 1;452 const cpu_count = pool.cpu_count catch 1;
437 const gpa = pool.allocator;453 const gpa = pool.allocator;
438 const n = GroupClosure.contextEnd(context_alignment, context.len);454 const n = GroupClosure.contextEnd(context_alignment, context.len);
439 const closure: *GroupClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(GroupClosure), n) catch {455 const gc: *GroupClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(GroupClosure), n) catch {
440 return start(context.ptr);456 return start(context.ptr);
441 }));457 }));
442 closure.* = .{458 gc.* = .{
459 .closure = .{
460 .cancel_tid = 0,
461 .start = GroupClosure.start,
462 .is_concurrent = false,
463 },
443 .pool = pool,464 .pool = pool,
444 .group = group,465 .group = group,
466 .node = .{ .next = @ptrCast(@alignCast(group.token)) },
445 .func = start,467 .func = start,
446 .context_alignment = context_alignment,468 .context_alignment = context_alignment,
447 .context_len = context.len,469 .context_len = context.len,
448 .runnable = .{
449 .start = GroupClosure.start,
450 .is_parallel = false,
451 },
452 };470 };
453 @memcpy(closure.contextPointer()[0..context.len], context);471 group.token = &gc.node;
472 @memcpy(gc.contextPointer()[0..context.len], context);
454473
455 pool.mutex.lock();474 pool.mutex.lock();
456475
457 const thread_capacity = cpu_count - 1 + pool.parallel_count;476 const thread_capacity = cpu_count - 1 + pool.concurrent_count;
458477
459 pool.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {478 pool.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {
460 pool.mutex.unlock();479 pool.mutex.unlock();
461 closure.free(gpa);480 gc.free(gpa);
462 return start(context.ptr);481 return start(context.ptr);
463 };482 };
464483
465 pool.run_queue.prepend(&closure.runnable.node);484 pool.run_queue.prepend(&gc.closure.node);
466485
467 if (pool.threads.items.len < thread_capacity) {486 if (pool.threads.items.len < thread_capacity) {
468 const thread = std.Thread.spawn(.{ .stack_size = pool.stack_size }, worker, .{pool}) catch {487 const thread = std.Thread.spawn(.{ .stack_size = pool.stack_size }, worker, .{pool}) catch {
469 assert(pool.run_queue.popFirst() == &closure.runnable.node);488 assert(pool.run_queue.popFirst() == &gc.closure.node);
470 pool.mutex.unlock();489 pool.mutex.unlock();
471 closure.free(gpa);490 gc.free(gpa);
472 return start(context.ptr);491 return start(context.ptr);
473 };492 };
474 pool.threads.appendAssumeCapacity(thread);493 pool.threads.appendAssumeCapacity(thread);
...@@ -486,7 +505,7 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group) void {...@@ -486,7 +505,7 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group) void {
486 const pool: *Pool = @ptrCast(@alignCast(userdata));505 const pool: *Pool = @ptrCast(@alignCast(userdata));
487 _ = pool;506 _ = pool;
488 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);507 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
489 const reset_event: *std.Thread.ResetEvent = @ptrCast(&group.context);508 const reset_event: *ResetEvent = @ptrCast(&group.context);
490 std.Thread.WaitGroup.waitStateless(group_state, reset_event);509 std.Thread.WaitGroup.waitStateless(group_state, reset_event);
491}510}
492511
...@@ -494,8 +513,14 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group) void {...@@ -494,8 +513,14 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group) void {
494 if (builtin.single_threaded) return;513 if (builtin.single_threaded) return;
495 const pool: *Pool = @ptrCast(@alignCast(userdata));514 const pool: *Pool = @ptrCast(@alignCast(userdata));
496 _ = pool;515 _ = pool;
497 _ = group;516 const token = group.token.?;
498 @panic("TODO threaded group cancel");517 group.token = null;
518 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
519 while (true) {
520 const gc: *GroupClosure = @fieldParentPtr("node", node);
521 gc.closure.requestCancel();
522 node = node.next orelse break;
523 }
499}524}
500525
501fn await(526fn await(
...@@ -518,32 +543,16 @@ fn cancel(...@@ -518,32 +543,16 @@ fn cancel(
518) void {543) void {
519 _ = result_alignment;544 _ = result_alignment;
520 const pool: *Pool = @ptrCast(@alignCast(userdata));545 const pool: *Pool = @ptrCast(@alignCast(userdata));
521 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));546 const ac: *AsyncClosure = @ptrCast(@alignCast(any_future));
522 switch (@atomicRmw(547 ac.closure.requestCancel();
523 std.Thread.Id,548 ac.waitAndFree(pool.allocator, result);
524 &closure.cancel_tid,
525 .Xchg,
526 AsyncClosure.canceling_tid,
527 .acq_rel,
528 )) {
529 0, AsyncClosure.canceling_tid => {},
530 else => |cancel_tid| switch (builtin.os.tag) {
531 .linux => _ = std.os.linux.tgkill(
532 std.os.linux.getpid(),
533 @bitCast(cancel_tid),
534 posix.SIG.IO,
535 ),
536 else => {},
537 },
538 }
539 closure.waitAndFree(pool.allocator, result);
540}549}
541550
542fn cancelRequested(userdata: ?*anyopaque) bool {551fn cancelRequested(userdata: ?*anyopaque) bool {
543 const pool: *Pool = @ptrCast(@alignCast(userdata));552 const pool: *Pool = @ptrCast(@alignCast(userdata));
544 _ = pool;553 _ = pool;
545 const closure = current_closure orelse return false;554 const closure = current_closure orelse return false;
546 return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == AsyncClosure.canceling_tid;555 return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == Closure.canceling_tid;
547}556}
548557
549fn checkCancel(pool: *Pool) error{Canceled}!void {558fn checkCancel(pool: *Pool) error{Canceled}!void {
...@@ -996,14 +1005,14 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {...@@ -996,14 +1005,14 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
996 const pool: *Pool = @ptrCast(@alignCast(userdata));1005 const pool: *Pool = @ptrCast(@alignCast(userdata));
997 _ = pool;1006 _ = pool;
9981007
999 var reset_event: std.Thread.ResetEvent = .unset;1008 var reset_event: ResetEvent = .unset;
10001009
1001 for (futures, 0..) |future, i| {1010 for (futures, 0..) |future, i| {
1002 const closure: *AsyncClosure = @ptrCast(@alignCast(future));1011 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
1003 if (@atomicRmw(?*std.Thread.ResetEvent, &closure.select_condition, .Xchg, &reset_event, .seq_cst) == AsyncClosure.done_reset_event) {1012 if (@atomicRmw(?*ResetEvent, &closure.select_condition, .Xchg, &reset_event, .seq_cst) == AsyncClosure.done_reset_event) {
1004 for (futures[0..i]) |cleanup_future| {1013 for (futures[0..i]) |cleanup_future| {
1005 const cleanup_closure: *AsyncClosure = @ptrCast(@alignCast(cleanup_future));1014 const cleanup_closure: *AsyncClosure = @ptrCast(@alignCast(cleanup_future));
1006 if (@atomicRmw(?*std.Thread.ResetEvent, &cleanup_closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {1015 if (@atomicRmw(?*ResetEvent, &cleanup_closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
1007 cleanup_closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.1016 cleanup_closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.
1008 }1017 }
1009 }1018 }
...@@ -1016,7 +1025,7 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {...@@ -1016,7 +1025,7 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
1016 var result: ?usize = null;1025 var result: ?usize = null;
1017 for (futures, 0..) |future, i| {1026 for (futures, 0..) |future, i| {
1018 const closure: *AsyncClosure = @ptrCast(@alignCast(future));1027 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
1019 if (@atomicRmw(?*std.Thread.ResetEvent, &closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {1028 if (@atomicRmw(?*ResetEvent, &closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
1020 closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.1029 closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.
1021 if (result == null) result = i; // In case multiple are ready, return first.1030 if (result == null) result = i; // In case multiple are ready, return first.
1022 }1031 }