authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-09-26 21:58:51-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log3c9fdf810f6f1517193786998ac5efe9b2b2c276
tree93236f61e8ac1c74b508b7299febe7440124bcf7
parentd37d795cba27c9793436449bfd783fc504c69cbb

std.Io: implement Group API

3 files changed, 102 insertions(+), 49 deletions(-)

lib/std/Io.zig+57-32
......@@ -593,18 +593,6 @@ pub const VTable = struct {
593593 context_alignment: std.mem.Alignment,
594594 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
595595 ) error{OutOfMemory}!*AnyFuture,
596 /// Executes `start` asynchronously in a manner such that it cleans itself
597 /// up. This mode does not support results, await, or cancel.
598 ///
599 /// Thread-safe.
600 asyncDetached: *const fn (
601 /// Corresponds to `Io.userdata`.
602 userdata: ?*anyopaque,
603 /// Copied and then passed to `start`.
604 context: []const u8,
605 context_alignment: std.mem.Alignment,
606 start: *const fn (context: *const anyopaque) void,
607 ) void,
608596 /// This function is only called when `async` returns a non-null value.
609597 ///
610598 /// Thread-safe.
......@@ -639,6 +627,23 @@ pub const VTable = struct {
639627 /// Thread-safe.
640628 cancelRequested: *const fn (?*anyopaque) bool,
641629
630 /// Executes `start` asynchronously in a manner such that it cleans itself
631 /// up. This mode does not support results, await, or cancel.
632 ///
633 /// Thread-safe.
634 groupAsync: *const fn (
635 /// Corresponds to `Io.userdata`.
636 userdata: ?*anyopaque,
637 /// Owner of the spawned async task.
638 group: *Group,
639 /// Copied and then passed to `start`.
640 context: []const u8,
641 context_alignment: std.mem.Alignment,
642 start: *const fn (context: *const anyopaque) void,
643 ) void,
644 groupWait: *const fn (?*anyopaque, *Group) void,
645 groupCancel: *const fn (?*anyopaque, *Group) void,
646
642647 /// Blocks until one of the futures from the list has a result ready, such
643648 /// that awaiting it will not block. Returns that index.
644649 select: *const fn (?*anyopaque, futures: []const *AnyFuture) usize,
......@@ -751,6 +756,45 @@ pub fn Future(Result: type) type {
751756 };
752757}
753758
759pub const Group = struct {
760 state: usize,
761 context: ?*anyopaque,
762
763 pub const init: Group = .{ .state = 0, .context = null };
764
765 /// Calls `function` with `args` asynchronously. The resource spawned is
766 /// owned by the group.
767 ///
768 /// `function` *may* be called immediately, before `async` returns.
769 ///
770 /// After this is called, `wait` must be called before the group is
771 /// deinitialized.
772 ///
773 /// See also:
774 /// * `async`
775 /// * `concurrent`
776 pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {
777 const Args = @TypeOf(args);
778 const TypeErased = struct {
779 fn start(context: *const anyopaque) void {
780 const args_casted: *const Args = @ptrCast(@alignCast(context));
781 @call(.auto, function, args_casted.*);
782 }
783 };
784 io.vtable.groupAsync(io.userdata, g, @ptrCast((&args)[0..1]), .of(Args), TypeErased.start);
785 }
786
787 /// Idempotent.
788 pub fn wait(g: *Group, io: Io) void {
789 io.vtable.groupWait(io.userdata, g);
790 }
791
792 /// Idempotent.
793 pub fn cancel(g: *Group, io: Io) void {
794 io.vtable.groupCancel(io.userdata, g);
795 }
796};
797
754798pub const Mutex = if (true) struct {
755799 state: State,
756800
......@@ -1099,7 +1143,7 @@ pub fn Queue(Elem: type) type {
10991143/// reusable.
11001144///
11011145/// See also:
1102/// * `asyncDetached`
1146/// * `Group`
11031147pub fn async(
11041148 io: Io,
11051149 function: anytype,
......@@ -1159,25 +1203,6 @@ pub fn concurrent(
11591203 return future;
11601204}
11611205
1162/// Calls `function` with `args` asynchronously. The resource cleans itself up
1163/// when the function returns. Does not support await, cancel, or a return value.
1164///
1165/// `function` *may* be called immediately, before `async` returns.
1166///
1167/// See also:
1168/// * `async`
1169/// * `concurrent`
1170pub fn asyncDetached(io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {
1171 const Args = @TypeOf(args);
1172 const TypeErased = struct {
1173 fn start(context: *const anyopaque) void {
1174 const args_casted: *const Args = @ptrCast(@alignCast(context));
1175 @call(.auto, function, args_casted.*);
1176 }
1177 };
1178 io.vtable.asyncDetached(io.userdata, @ptrCast((&args)[0..1]), .of(Args), TypeErased.start);
1179}
1180
11811206pub fn cancelRequested(io: Io) bool {
11821207 return io.vtable.cancelRequested(io.userdata);
11831208}
lib/std/Io/Threaded.zig+43-15
......@@ -8,7 +8,6 @@ const windows = std.os.windows;
88const std = @import("../std.zig");
99const Allocator = std.mem.Allocator;
1010const assert = std.debug.assert;
11const WaitGroup = std.Thread.WaitGroup;
1211const posix = std.posix;
1312const Io = std.Io;
1413
......@@ -101,10 +100,12 @@ pub fn io(pool: *Pool) Io {
101100 .async = async,
102101 .concurrent = concurrent,
103102 .await = await,
104 .asyncDetached = asyncDetached,
105103 .cancel = cancel,
106104 .cancelRequested = cancelRequested,
107105 .select = select,
106 .groupAsync = groupAsync,
107 .groupWait = groupWait,
108 .groupCancel = groupCancel,
108109
109110 .mutexLock = mutexLock,
110111 .mutexUnlock = mutexUnlock,
......@@ -279,7 +280,7 @@ fn async(
279280 .func = start,
280281 .context_offset = context_offset,
281282 .result_offset = result_offset,
282 .reset_event = .{},
283 .reset_event = .unset,
283284 .cancel_tid = 0,
284285 .select_condition = null,
285286 .runnable = .{
......@@ -347,7 +348,7 @@ fn concurrent(
347348 .func = start,
348349 .context_offset = context_offset,
349350 .result_offset = result_offset,
350 .reset_event = .{},
351 .reset_event = .unset,
351352 .cancel_tid = 0,
352353 .select_condition = null,
353354 .runnable = .{
......@@ -385,41 +386,47 @@ fn concurrent(
385386 return @ptrCast(closure);
386387}
387388
388const DetachedClosure = struct {
389const GroupClosure = struct {
389390 pool: *Pool,
391 group: *Io.Group,
390392 func: *const fn (context: *anyopaque) void,
391393 runnable: Runnable,
392394 context_alignment: std.mem.Alignment,
393395 context_len: usize,
394396
395397 fn start(runnable: *Runnable) void {
396 const closure: *DetachedClosure = @alignCast(@fieldParentPtr("runnable", runnable));
398 const closure: *GroupClosure = @alignCast(@fieldParentPtr("runnable", runnable));
397399 closure.func(closure.contextPointer());
400 const group = closure.group;
398401 const gpa = closure.pool.allocator;
399402 free(closure, gpa);
403 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
404 const reset_event: *std.Thread.ResetEvent = @ptrCast(&group.context);
405 std.Thread.WaitGroup.finishStateless(group_state, reset_event);
400406 }
401407
402 fn free(closure: *DetachedClosure, gpa: Allocator) void {
403 const base: [*]align(@alignOf(DetachedClosure)) u8 = @ptrCast(closure);
408 fn free(closure: *GroupClosure, gpa: Allocator) void {
409 const base: [*]align(@alignOf(GroupClosure)) u8 = @ptrCast(closure);
404410 gpa.free(base[0..contextEnd(closure.context_alignment, closure.context_len)]);
405411 }
406412
407413 fn contextOffset(context_alignment: std.mem.Alignment) usize {
408 return context_alignment.forward(@sizeOf(DetachedClosure));
414 return context_alignment.forward(@sizeOf(GroupClosure));
409415 }
410416
411417 fn contextEnd(context_alignment: std.mem.Alignment, context_len: usize) usize {
412418 return contextOffset(context_alignment) + context_len;
413419 }
414420
415 fn contextPointer(closure: *DetachedClosure) [*]u8 {
421 fn contextPointer(closure: *GroupClosure) [*]u8 {
416422 const base: [*]u8 = @ptrCast(closure);
417423 return base + contextOffset(closure.context_alignment);
418424 }
419425};
420426
421fn asyncDetached(
427fn groupAsync(
422428 userdata: ?*anyopaque,
429 group: *Io.Group,
423430 context: []const u8,
424431 context_alignment: std.mem.Alignment,
425432 start: *const fn (context: *const anyopaque) void,
......@@ -428,17 +435,18 @@ fn asyncDetached(
428435 const pool: *Pool = @ptrCast(@alignCast(userdata));
429436 const cpu_count = pool.cpu_count catch 1;
430437 const gpa = pool.allocator;
431 const n = DetachedClosure.contextEnd(context_alignment, context.len);
432 const closure: *DetachedClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(DetachedClosure), n) catch {
438 const n = GroupClosure.contextEnd(context_alignment, context.len);
439 const closure: *GroupClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(GroupClosure), n) catch {
433440 return start(context.ptr);
434441 }));
435442 closure.* = .{
436443 .pool = pool,
444 .group = group,
437445 .func = start,
438446 .context_alignment = context_alignment,
439447 .context_len = context.len,
440448 .runnable = .{
441 .start = DetachedClosure.start,
449 .start = GroupClosure.start,
442450 .is_parallel = false,
443451 },
444452 };
......@@ -466,10 +474,30 @@ fn asyncDetached(
466474 pool.threads.appendAssumeCapacity(thread);
467475 }
468476
477 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
478 std.Thread.WaitGroup.startStateless(group_state);
479
469480 pool.mutex.unlock();
470481 pool.cond.signal();
471482}
472483
484fn groupWait(userdata: ?*anyopaque, group: *Io.Group) void {
485 if (builtin.single_threaded) return;
486 const pool: *Pool = @ptrCast(@alignCast(userdata));
487 _ = pool;
488 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
489 const reset_event: *std.Thread.ResetEvent = @ptrCast(&group.context);
490 std.Thread.WaitGroup.waitStateless(group_state, reset_event);
491}
492
493fn groupCancel(userdata: ?*anyopaque, group: *Io.Group) void {
494 if (builtin.single_threaded) return;
495 const pool: *Pool = @ptrCast(@alignCast(userdata));
496 _ = pool;
497 _ = group;
498 @panic("TODO threaded group cancel");
499}
500
473501fn await(
474502 userdata: ?*anyopaque,
475503 any_future: *Io.AnyFuture,
......@@ -968,7 +996,7 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
968996 const pool: *Pool = @ptrCast(@alignCast(userdata));
969997 _ = pool;
970998
971 var reset_event: std.Thread.ResetEvent = .{};
999 var reset_event: std.Thread.ResetEvent = .unset;
9721000
9731001 for (futures, 0..) |future, i| {
9741002 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
lib/std/Io/net/HostName.zig+2-2
......@@ -653,7 +653,7 @@ pub const ResolvConf = struct {
653653
654654 const mapped_nameservers = if (any_ip6) ip4_mapped[0..rc.nameservers_len] else rc.nameservers();
655655
656 var group: Io.Group = .{};
656 var group: Io.Group = .init;
657657 defer group.cancel();
658658
659659 for (queries) |query| {
......@@ -702,7 +702,7 @@ test ResolvConf {
702702 .search_buffer = undefined,
703703 .search_len = 0,
704704 .ndots = 1,
705 .timeout = 5,
705 .timeout = .seconds(5),
706706 .attempts = 2,
707707 };
708708