authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-31 02:10:50-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:39-07:00
logebf92042e3a081ce84668a7edc1aa74c9ad7e9e5
tree377ed810f6e21ad79e6ea5dd6380ca0f28775417
parenta7790bd32e1e8caf6f2f0bedede8a7cb7b35c443

std.Io: add detached async

2 files changed, 113 insertions(+), 4 deletions(-)

lib/std/Io.zig+43-4
......@@ -933,6 +933,18 @@ pub const VTable = struct {
933933 context_alignment: std.mem.Alignment,
934934 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
935935 ) ?*AnyFuture,
936 /// Executes `start` asynchronously in a manner such that it cleans itself
937 /// up. This mode does not support results, await, or cancel.
938 ///
939 /// Thread-safe.
940 go: *const fn (
941 /// Corresponds to `Io.userdata`.
942 userdata: ?*anyopaque,
943 /// Copied and then passed to `start`.
944 context: []const u8,
945 context_alignment: std.mem.Alignment,
946 start: *const fn (context: *const anyopaque) void,
947 ) void,
936948 /// This function is only called when `async` returns a non-null value.
937949 ///
938950 /// Thread-safe.
......@@ -946,7 +958,6 @@ pub const VTable = struct {
946958 result: []u8,
947959 result_alignment: std.mem.Alignment,
948960 ) void,
949
950961 /// Equivalent to `await` but initiates cancel request.
951962 ///
952963 /// This function is only called when `async` returns a non-null value.
......@@ -1024,14 +1035,24 @@ pub fn Future(Result: type) type {
10241035 /// Idempotent.
10251036 pub fn cancel(f: *@This(), io: Io) Result {
10261037 const any_future = f.any_future orelse return f.result;
1027 io.vtable.cancel(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));
1038 io.vtable.cancel(
1039 io.userdata,
1040 any_future,
1041 if (@sizeOf(Result) == 0) &.{} else @ptrCast((&f.result)[0..1]), // work around compiler bug
1042 .of(Result),
1043 );
10281044 f.any_future = null;
10291045 return f.result;
10301046 }
10311047
10321048 pub fn await(f: *@This(), io: Io) Result {
10331049 const any_future = f.any_future orelse return f.result;
1034 io.vtable.await(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));
1050 io.vtable.await(
1051 io.userdata,
1052 any_future,
1053 if (@sizeOf(Result) == 0) &.{} else @ptrCast((&f.result)[0..1]), // work around compiler bug
1054 .of(Result),
1055 );
10351056 f.any_future = null;
10361057 return f.result;
10371058 }
......@@ -1349,7 +1370,7 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(
13491370 var future: Future(Result) = undefined;
13501371 future.any_future = io.vtable.async(
13511372 io.userdata,
1352 @ptrCast((&future.result)[0..1]),
1373 if (@sizeOf(Result) == 0) &.{} else @ptrCast((&future.result)[0..1]), // work around compiler bug
13531374 .of(Result),
13541375 if (@sizeOf(Args) == 0) &.{} else @ptrCast((&args)[0..1]), // work around compiler bug
13551376 .of(Args),
......@@ -1358,6 +1379,24 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(
13581379 return future;
13591380}
13601381
1382/// Calls `function` with `args` asynchronously. The resource cleans itself up
1383/// when the function returns. Does not support await, cancel, or a return value.
1384pub fn go(io: Io, function: anytype, args: anytype) void {
1385 const Args = @TypeOf(args);
1386 const TypeErased = struct {
1387 fn start(context: *const anyopaque) void {
1388 const args_casted: *const Args = @alignCast(@ptrCast(context));
1389 @call(.auto, function, args_casted.*);
1390 }
1391 };
1392 io.vtable.go(
1393 io.userdata,
1394 if (@sizeOf(Args) == 0) &.{} else @ptrCast((&args)[0..1]), // work around compiler bug
1395 .of(Args),
1396 TypeErased.start,
1397 );
1398}
1399
13611400pub fn openFile(io: Io, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File {
13621401 return io.vtable.openFile(io.userdata, dir, sub_path, flags);
13631402}
lib/std/Thread/Pool.zig+70
......@@ -332,6 +332,7 @@ pub fn io(pool: *Pool) Io {
332332 .vtable = &.{
333333 .@"async" = @"async",
334334 .@"await" = @"await",
335 .go = go,
335336 .cancel = cancel,
336337 .cancelRequested = cancelRequested,
337338 .mutexLock = mutexLock,
......@@ -472,6 +473,75 @@ fn @"async"(
472473 return @ptrCast(closure);
473474}
474475
476const DetachedClosure = struct {
477 pool: *Pool,
478 func: *const fn (context: *anyopaque) void,
479 run_node: std.Thread.Pool.RunQueue.Node = .{ .data = .{ .runFn = runFn } },
480 context_alignment: std.mem.Alignment,
481 context_len: usize,
482
483 fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void {
484 const run_node: *std.Thread.Pool.RunQueue.Node = @fieldParentPtr("data", runnable);
485 const closure: *DetachedClosure = @alignCast(@fieldParentPtr("run_node", run_node));
486 closure.func(closure.contextPointer());
487 const gpa = closure.pool.allocator;
488 const base: [*]align(@alignOf(DetachedClosure)) u8 = @ptrCast(closure);
489 gpa.free(base[0..contextEnd(closure.context_alignment, closure.context_len)]);
490 }
491
492 fn contextOffset(context_alignment: std.mem.Alignment) usize {
493 return context_alignment.forward(@sizeOf(DetachedClosure));
494 }
495
496 fn contextEnd(context_alignment: std.mem.Alignment, context_len: usize) usize {
497 return contextOffset(context_alignment) + context_len;
498 }
499
500 fn contextPointer(closure: *DetachedClosure) [*]u8 {
501 const base: [*]u8 = @ptrCast(closure);
502 return base + contextOffset(closure.context_alignment);
503 }
504};
505
506fn go(
507 userdata: ?*anyopaque,
508 context: []const u8,
509 context_alignment: std.mem.Alignment,
510 start: *const fn (context: *const anyopaque) void,
511) void {
512 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
513 pool.mutex.lock();
514
515 const gpa = pool.allocator;
516 const n = DetachedClosure.contextEnd(context_alignment, context.len);
517 const closure: *DetachedClosure = @alignCast(@ptrCast(gpa.alignedAlloc(u8, @alignOf(DetachedClosure), n) catch {
518 pool.mutex.unlock();
519 start(context.ptr);
520 return;
521 }));
522 closure.* = .{
523 .pool = pool,
524 .func = start,
525 .context_alignment = context_alignment,
526 .context_len = context.len,
527 };
528 @memcpy(closure.contextPointer()[0..context.len], context);
529 pool.run_queue.prepend(&closure.run_node);
530
531 if (pool.threads.items.len < pool.threads.capacity) {
532 pool.threads.addOneAssumeCapacity().* = std.Thread.spawn(.{
533 .stack_size = pool.stack_size,
534 .allocator = gpa,
535 }, worker, .{pool}) catch t: {
536 pool.threads.items.len -= 1;
537 break :t undefined;
538 };
539 }
540
541 pool.mutex.unlock();
542 pool.cond.signal();
543}
544
475545fn @"await"(
476546 userdata: ?*anyopaque,
477547 any_future: *std.Io.AnyFuture,