authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-15 00:36:02-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:49-07:00
logd3c4158a1053e5322d137f61bf4bc643650a4741
treee9b3e9e3b2ba91c9b74d44c0fd314461941e995c
parent35ce907c06d5758adab276927ad8dbe730d6130d

std.Io: implement Select

and finish implementation of HostName.connect

5 files changed, 144 insertions(+), 36 deletions(-)

lib/std/Io.zig+82-2
......@@ -639,7 +639,7 @@ pub const VTable = struct {
639639 /// Copied and then passed to `start`.
640640 context: []const u8,
641641 context_alignment: std.mem.Alignment,
642 start: *const fn (context: *const anyopaque) void,
642 start: *const fn (*Group, context: *const anyopaque) void,
643643 ) void,
644644 groupWait: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
645645 groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
......@@ -1005,7 +1005,8 @@ pub const Group = struct {
10051005 pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {
10061006 const Args = @TypeOf(args);
10071007 const TypeErased = struct {
1008 fn start(context: *const anyopaque) void {
1008 fn start(group: *Group, context: *const anyopaque) void {
1009 _ = group;
10091010 const args_casted: *const Args = @ptrCast(@alignCast(context));
10101011 @call(.auto, function, args_casted.*);
10111012 }
......@@ -1033,6 +1034,85 @@ pub const Group = struct {
10331034 }
10341035};
10351036
1037pub fn Select(comptime U: type) type {
1038 return struct {
1039 io: Io,
1040 group: Group,
1041 queue: Queue(U),
1042 outstanding: usize,
1043
1044 const S = @This();
1045
1046 pub const Union = U;
1047
1048 pub const Field = std.meta.FieldEnum(U);
1049
1050 pub fn init(io: Io, buffer: []U) S {
1051 return .{
1052 .io = io,
1053 .queue = .init(buffer),
1054 .group = .init,
1055 .outstanding = 0,
1056 };
1057 }
1058
1059 /// Calls `function` with `args` asynchronously. The resource spawned is
1060 /// owned by the select.
1061 ///
1062 /// `function` must have return type matching the `field` field of `Union`.
1063 ///
1064 /// `function` *may* be called immediately, before `async` returns.
1065 ///
1066 /// After this is called, `wait` or `cancel` must be called before the
1067 /// select is deinitialized.
1068 ///
1069 /// Threadsafe.
1070 ///
1071 /// Related:
1072 /// * `Io.async`
1073 /// * `Group.async`
1074 pub fn async(
1075 s: *S,
1076 comptime field: Field,
1077 function: anytype,
1078 args: std.meta.ArgsTuple(@TypeOf(function)),
1079 ) void {
1080 const Args = @TypeOf(args);
1081 const TypeErased = struct {
1082 fn start(group: *Group, context: *const anyopaque) void {
1083 const args_casted: *const Args = @ptrCast(@alignCast(context));
1084 const unerased_select: *S = @fieldParentPtr("group", group);
1085 const elem = @unionInit(U, @tagName(field), @call(.auto, function, args_casted.*));
1086 unerased_select.queue.putOneUncancelable(unerased_select.io, elem);
1087 }
1088 };
1089 _ = @atomicRmw(usize, &s.outstanding, .Add, 1, .monotonic);
1090 s.io.vtable.groupAsync(s.io.userdata, &s.group, @ptrCast((&args)[0..1]), .of(Args), TypeErased.start);
1091 }
1092
1093 /// Blocks until another task of the select finishes.
1094 ///
1095 /// Asserts there is at least one more `outstanding` task.
1096 ///
1097 /// Not threadsafe.
1098 pub fn wait(s: *S) Io.Cancelable!U {
1099 s.outstanding -= 1;
1100 return s.queue.getOne(s.io);
1101 }
1102
1103 /// Equivalent to `wait` but requests cancellation on all remaining
1104 /// tasks owned by the select.
1105 ///
1106 /// It is illegal to call `wait` after this.
1107 ///
1108 /// Idempotent. Not threadsafe.
1109 pub fn cancel(s: *S) void {
1110 s.outstanding = 0;
1111 s.group.cancel(s.io);
1112 }
1113 };
1114}
1115
10361116pub const Mutex = struct {
10371117 state: State,
10381118
lib/std/Io/Threaded.zig+7-6
......@@ -458,7 +458,7 @@ const GroupClosure = struct {
458458 group: *Io.Group,
459459 /// Points to sibling `GroupClosure`. Used for walking the group to cancel all.
460460 node: std.SinglyLinkedList.Node,
461 func: *const fn (context: *anyopaque) void,
461 func: *const fn (*Io.Group, context: *anyopaque) void,
462462 context_alignment: std.mem.Alignment,
463463 context_len: usize,
464464
......@@ -476,7 +476,7 @@ const GroupClosure = struct {
476476 return;
477477 }
478478 current_closure = closure;
479 gc.func(gc.contextPointer());
479 gc.func(group, gc.contextPointer());
480480 current_closure = null;
481481
482482 // In case a cancel happens after successful task completion, prevents
......@@ -512,7 +512,7 @@ fn groupAsync(
512512 group: *Io.Group,
513513 context: []const u8,
514514 context_alignment: std.mem.Alignment,
515 start: *const fn (context: *const anyopaque) void,
515 start: *const fn (*Io.Group, context: *const anyopaque) void,
516516) void {
517517 if (builtin.single_threaded) return start(context.ptr);
518518 const pool: *Pool = @ptrCast(@alignCast(userdata));
......@@ -520,7 +520,7 @@ fn groupAsync(
520520 const gpa = pool.allocator;
521521 const n = GroupClosure.contextEnd(context_alignment, context.len);
522522 const gc: *GroupClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(GroupClosure), n) catch {
523 return start(context.ptr);
523 return start(group, context.ptr);
524524 }));
525525 gc.* = .{
526526 .closure = .{
......@@ -548,7 +548,7 @@ fn groupAsync(
548548 pool.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {
549549 pool.mutex.unlock();
550550 gc.free(gpa);
551 return start(context.ptr);
551 return start(group, context.ptr);
552552 };
553553
554554 pool.run_queue.prepend(&gc.closure.node);
......@@ -558,7 +558,7 @@ fn groupAsync(
558558 assert(pool.run_queue.popFirst() == &gc.closure.node);
559559 pool.mutex.unlock();
560560 gc.free(gpa);
561 return start(context.ptr);
561 return start(group, context.ptr);
562562 };
563563 pool.threads.appendAssumeCapacity(thread);
564564 }
......@@ -2662,6 +2662,7 @@ fn netLookupFallible(
26622662 .{ .address = addr },
26632663 .{ .canonical_name = copyCanon(options.canonical_name_buffer, name) },
26642664 });
2665 return;
26652666 } else |_| {}
26662667 }
26672668
lib/std/Io/net.zig+2-2
......@@ -315,8 +315,8 @@ pub const IpAddress = union(enum) {
315315 };
316316
317317 /// Initiates a connection-oriented network stream.
318 pub fn connect(address: *const IpAddress, io: Io, options: ConnectOptions) ConnectError!Stream {
319 return io.vtable.netConnectIp(io.userdata, address, options);
318 pub fn connect(address: IpAddress, io: Io, options: ConnectOptions) ConnectError!Stream {
319 return io.vtable.netConnectIp(io.userdata, &address, options);
320320 }
321321};
322322
lib/std/Io/net/HostName.zig+8-6
......@@ -88,7 +88,7 @@ pub const LookupResult = union(enum) {
8888/// Adds any number of `IpAddress` into resolved, exactly one canonical_name,
8989/// and then always finishes by adding one `LookupResult.end` entry.
9090///
91/// Guaranteed not to block if provided queue has capacity at least 8.
91/// Guaranteed not to block if provided queue has capacity at least 16.
9292pub fn lookup(
9393 host_name: HostName,
9494 io: Io,
......@@ -216,11 +216,13 @@ pub fn connect(
216216 } });
217217 defer lookup_task.cancel(io);
218218
219 var select: Io.Select(union(enum) { ip_connect: IpAddress.ConnectError!Stream }) = .init;
220 defer select.cancel(io);
219 const Result = union(enum) { connect_result: IpAddress.ConnectError!Stream };
220 var finished_task_buffer: [results_buffer.len]Result = undefined;
221 var select: Io.Select(Result) = .init(io, &finished_task_buffer);
222 defer select.cancel();
221223
222224 while (results.getOne(io)) |result| switch (result) {
223 .address => |address| select.async(io, .ip_connect, IpAddress.connect, .{ address, io, options }),
225 .address => |address| select.async(.connect_result, IpAddress.connect, .{ address, io, options }),
224226 .canonical_name => continue,
225227 .end => |lookup_result| {
226228 try lookup_result;
......@@ -230,8 +232,8 @@ pub fn connect(
230232
231233 var aggregate_error: ConnectError = error.UnknownHostName;
232234
233 while (select.remaining != 0) switch (select.wait(io)) {
234 .ip_connect => |ip_connect| if (ip_connect) |stream| return stream else |err| switch (err) {
235 while (select.outstanding != 0) switch (try select.wait()) {
236 .connect_result => |connect_result| if (connect_result) |stream| return stream else |err| switch (err) {
235237 error.SystemResources => |e| return e,
236238 error.OptionUnsupported => |e| return e,
237239 error.ProcessFdQuotaExceeded => |e| return e,
lib/std/Io/net/test.zig+45-20
......@@ -1,5 +1,7 @@
1const std = @import("std");
21const builtin = @import("builtin");
2
3const std = @import("std");
4const Io = std.Io;
35const net = std.Io.net;
46const mem = std.mem;
57const testing = std.testing;
......@@ -126,33 +128,56 @@ test "resolve DNS" {
126128 const localhost_v4 = try net.IpAddress.parse("127.0.0.1", 80);
127129 const localhost_v6 = try net.IpAddress.parse("::2", 80);
128130
129 var addresses_buffer: [8]net.IpAddress = undefined;
130 var canon_name_buffer: [net.HostName.max_len]u8 = undefined;
131 const result = try net.HostName.lookup(try .init("localhost"), io, .{
131 var canonical_name_buffer: [net.HostName.max_len]u8 = undefined;
132 var results_buffer: [32]net.HostName.LookupResult = undefined;
133 var results: Io.Queue(net.HostName.LookupResult) = .init(&results_buffer);
134
135 net.HostName.lookup(try .init("localhost"), io, &results, .{
132136 .port = 80,
133 .addresses_buffer = &addresses_buffer,
134 .canonical_name_buffer = &canon_name_buffer,
137 .canonical_name_buffer = &canonical_name_buffer,
135138 });
136 for (addresses_buffer[0..result.addresses_len]) |addr| {
137 if (addr.eql(&localhost_v4) or addr.eql(&localhost_v6)) break;
138 } else @panic("unexpected address for localhost");
139
140 var addresses_found: usize = 0;
141
142 while (results.getOne(io)) |result| switch (result) {
143 .address => |address| {
144 if (address.eql(&localhost_v4) or address.eql(&localhost_v6))
145 addresses_found += 1;
146 },
147 .canonical_name => |canonical_name| try testing.expectEqualStrings("localhost", canonical_name.bytes),
148 .end => |end| {
149 try end;
150 break;
151 },
152 } else |err| return err;
153
154 try testing.expect(addresses_found != 0);
139155 }
140156
141157 {
142158 // The tests are required to work even when there is no Internet connection,
143159 // so some of these errors we must accept and skip the test.
144 var addresses_buffer: [8]net.IpAddress = undefined;
145 var canon_name_buffer: [net.HostName.max_len]u8 = undefined;
146 const result = net.HostName.lookup(try .init("example.com"), io, .{
160 var canonical_name_buffer: [net.HostName.max_len]u8 = undefined;
161 var results_buffer: [16]net.HostName.LookupResult = undefined;
162 var results: Io.Queue(net.HostName.LookupResult) = .init(&results_buffer);
163
164 net.HostName.lookup(try .init("example.com"), io, &results, .{
147165 .port = 80,
148 .addresses_buffer = &addresses_buffer,
149 .canonical_name_buffer = &canon_name_buffer,
150 }) catch |err| switch (err) {
151 error.UnknownHostName => return error.SkipZigTest,
152 error.NameServerFailure => return error.SkipZigTest,
153 else => return err,
154 };
155 _ = result;
166 .canonical_name_buffer = &canonical_name_buffer,
167 });
168
169 while (results.getOne(io)) |result| switch (result) {
170 .address => {},
171 .canonical_name => {},
172 .end => |end| {
173 end catch |err| switch (err) {
174 error.UnknownHostName => return error.SkipZigTest,
175 error.NameServerFailure => return error.SkipZigTest,
176 else => return err,
177 };
178 break;
179 },
180 } else |err| return err;
156181 }
157182}
158183