authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-09-30 23:56:52-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log47f18ee6a058966bbe06e83e26a5b6ed67f368cb
treeca322e3903ce23fea8488e330b820af39730b2c5
parent46e5068e48fdc7ecdc6121b978c87fad680cbe40

std.Io.net: make netSend support multiple messages

this lowers to sendmmsg on linux, and means Io.Group is no longer needed, resulting in a more efficient implementation.

4 files changed, 39 insertions(+), 15 deletions(-)

lib/std/Io.zig+1-1
...@@ -671,7 +671,7 @@ pub const VTable = struct {...@@ -671,7 +671,7 @@ pub const VTable = struct {
671 listen: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Server,671 listen: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Server,
672 accept: *const fn (?*anyopaque, server: *net.Server) net.Server.AcceptError!net.Stream,672 accept: *const fn (?*anyopaque, server: *net.Server) net.Server.AcceptError!net.Stream,
673 ipBind: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket,673 ipBind: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket,
674 netSend: *const fn (?*anyopaque, handle: net.Socket.Handle, address: *const net.IpAddress, data: []const u8) net.Socket.SendError!void,674 netSend: *const fn (?*anyopaque, net.Socket.Handle, []const net.OutgoingMessage, net.SendFlags) net.Socket.SendError!void,
675 netReceive: *const fn (?*anyopaque, handle: net.Socket.Handle, buffer: []u8, timeout: Timeout) net.Socket.ReceiveTimeoutError!net.ReceivedMessage,675 netReceive: *const fn (?*anyopaque, handle: net.Socket.Handle, buffer: []u8, timeout: Timeout) net.Socket.ReceiveTimeoutError!net.ReceivedMessage,
676 netRead: *const fn (?*anyopaque, src: net.Stream, data: [][]u8) net.Stream.Reader.Error!usize,676 netRead: *const fn (?*anyopaque, src: net.Stream, data: [][]u8) net.Stream.Reader.Error!usize,
677 netWrite: *const fn (?*anyopaque, dest: net.Stream, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize,677 netWrite: *const fn (?*anyopaque, dest: net.Stream, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize,
lib/std/Io/Threaded.zig+4-4
...@@ -1309,15 +1309,15 @@ fn netReadPosix(userdata: ?*anyopaque, stream: Io.net.Stream, data: [][]u8) Io.n...@@ -1309,15 +1309,15 @@ fn netReadPosix(userdata: ?*anyopaque, stream: Io.net.Stream, data: [][]u8) Io.n
1309fn netSend(1309fn netSend(
1310 userdata: ?*anyopaque,1310 userdata: ?*anyopaque,
1311 handle: Io.net.Socket.Handle,1311 handle: Io.net.Socket.Handle,
1312 address: *const Io.net.IpAddress,1312 messages: []const Io.net.OutgoingMessage,
1313 data: []const u8,1313 flags: Io.net.SendFlags,
1314) Io.net.Socket.SendError!void {1314) Io.net.Socket.SendError!void {
1315 const pool: *Pool = @ptrCast(@alignCast(userdata));1315 const pool: *Pool = @ptrCast(@alignCast(userdata));
1316 try pool.checkCancel();1316 try pool.checkCancel();
13171317
1318 _ = handle;1318 _ = handle;
1319 _ = address;1319 _ = messages;
1320 _ = data;1320 _ = flags;
1321 @panic("TODO");1321 @panic("TODO");
1322}1322}
13231323
lib/std/Io/net.zig+21-1
...@@ -700,6 +700,21 @@ pub const ReceivedMessage = struct {...@@ -700,6 +700,21 @@ pub const ReceivedMessage = struct {
700 len: usize,700 len: usize,
701};701};
702702
703pub const OutgoingMessage = struct {
704 address: *const IpAddress,
705 data: []const u8,
706 control: []const u8 = &.{},
707};
708
709pub const SendFlags = packed struct(u8) {
710 confirm: bool = false,
711 dont_route: bool = false,
712 eor: bool = false,
713 oob: bool = false,
714 fastopen: bool = false,
715 _: u3 = 0,
716};
717
703pub const Interface = struct {718pub const Interface = struct {
704 /// Value 0 indicates `none`.719 /// Value 0 indicates `none`.
705 index: u32,720 index: u32,
...@@ -818,7 +833,12 @@ pub const Socket = struct {...@@ -818,7 +833,12 @@ pub const Socket = struct {
818833
819 /// Transfers `data` to `dest`, connectionless.834 /// Transfers `data` to `dest`, connectionless.
820 pub fn send(s: *const Socket, io: Io, dest: *const IpAddress, data: []const u8) SendError!void {835 pub fn send(s: *const Socket, io: Io, dest: *const IpAddress, data: []const u8) SendError!void {
821 return io.vtable.netSend(io.userdata, s.handle, dest, data);836 const message: OutgoingMessage = .{ .address = dest, .data = data };
837 return io.vtable.netSend(io.userdata, s.handle, &.{message}, .{});
838 }
839
840 pub fn sendMany(s: *const Socket, io: Io, messages: []const OutgoingMessage, flags: SendFlags) SendError!void {
841 return io.vtable.netSend(io.userdata, s.handle, messages, flags);
822 }842 }
823843
824 pub const ReceiveError = error{} || Io.UnexpectedError || Io.Cancelable;844 pub const ReceiveError = error{} || Io.UnexpectedError || Io.Cancelable;
lib/std/Io/net/HostName.zig+13-9
...@@ -271,15 +271,19 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio...@@ -271,15 +271,19 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio
271 };271 };
272272
273 send: while (now_ts.compare(.lt, final_ts)) : (now_ts = try io.now(.MONOTONIC)) {273 send: while (now_ts.compare(.lt, final_ts)) : (now_ts = try io.now(.MONOTONIC)) {
274 var group: Io.Group = .init;274 var message_buffer: [queries_buffer.len * ResolvConf.max_nameservers]Io.net.OutgoingMessage = undefined;
275 defer group.cancel(io);275 var message_i: usize = 0;
276
277 for (queries, answers) |query, *answer| {276 for (queries, answers) |query, *answer| {
278 if (answer.len != 0) continue;277 if (answer.len != 0) continue;
279 for (mapped_nameservers) |*ns| {278 for (mapped_nameservers) |*ns| {
280 group.async(io, sendIgnoringResult, .{ io, socket.handle, ns, query });279 message_buffer[message_i] = .{
280 .address = ns,
281 .data = query,
282 };
283 message_i += 1;
281 }284 }
282 }285 }
286 io.vtable.netSend(io.userdata, socket.handle, message_buffer[0..message_i], .{}) catch {};
283287
284 const timeout: Io.Timeout = .{ .deadline = now_ts.addDuration(attempt_duration) };288 const timeout: Io.Timeout = .{ .deadline = now_ts.addDuration(attempt_duration) };
285289
...@@ -320,7 +324,11 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio...@@ -320,7 +324,11 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio
320 if (next_answer_buffer == answers.len) break :send;324 if (next_answer_buffer == answers.len) break :send;
321 },325 },
322 2 => {326 2 => {
323 group.async(io, sendIgnoringResult, .{ io, socket.handle, ns, query });327 const message: Io.net.OutgoingMessage = .{
328 .address = ns,
329 .data = query,
330 };
331 io.vtable.netSend(io.userdata, socket.handle, &.{message}, .{}) catch {};
324 continue;332 continue;
325 },333 },
326 else => continue,334 else => continue,
...@@ -382,10 +390,6 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio...@@ -382,10 +390,6 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio
382 return error.NameServerFailure;390 return error.NameServerFailure;
383}391}
384392
385fn sendIgnoringResult(io: Io, socket_handle: Io.net.Socket.Handle, dest: *const IpAddress, msg: []const u8) void {
386 _ = io.vtable.netSend(io.userdata, socket_handle, dest, msg) catch {};
387}
388
389fn lookupHosts(host_name: HostName, io: Io, options: LookupOptions) !LookupResult {393fn lookupHosts(host_name: HostName, io: Io, options: LookupOptions) !LookupResult {
390 const file = Io.File.openAbsolute(io, "/etc/hosts", .{}) catch |err| switch (err) {394 const file = Io.File.openAbsolute(io, "/etc/hosts", .{}) catch |err| switch (err) {
391 error.FileNotFound,395 error.FileNotFound,