From 20355494085a6eb59b73816038a5d0c56b5492c6 Mon Sep 17 00:00:00 2001 From: hemisputnik Date: Wed, 22 Jul 2026 09:06:26 +0300 Subject: [PATCH] std.Io.net: improve reception of control data A couple of changes to improve reception of control data when using the datagram API. Instead of having a separate control buffer per received message, there's one big buffer that all messages share (like the data buffer). This means that now IncomingMessage has no user-initialised fields, and can be initialised with undefined (which means that .init is now deprecated). Also, instead of taking a pointer to mutable IncomingMessage, netReceiveOne* functions simply return a IncomingMessage. --- lib/std/Io.zig | 1 + lib/std/Io/Kqueue.zig | 2 + lib/std/Io/Threaded.zig | 26 ++++++++---- lib/std/Io/Uring.zig | 13 ++++-- lib/std/Io/net.zig | 87 +++++++++++++++++++++++++++++++++++------ 5 files changed, 106 insertions(+), 23 deletions(-) diff --git a/lib/std/Io.zig b/lib/std/Io.zig index d83c63e7276615651ad896128e6878fcb0e8e792..7d43bef3bdd52f6e4bcbc17351b1e2c9107f25f0 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -346,6 +346,7 @@ pub const Operation = union(enum) { socket_handle: net.Socket.Handle, message_buffer: []net.IncomingMessage, data_buffer: []u8, + control_buffer: []u8 = &.{}, flags: net.ReceiveFlags, pub const Error = error{ diff --git a/lib/std/Io/Kqueue.zig b/lib/std/Io/Kqueue.zig index 8f8be91d59bd3f3131ad5c08955da9de09b90e26..f663ab366a157ea562809d58c4b03ef9f6861c76 100644 --- a/lib/std/Io/Kqueue.zig +++ b/lib/std/Io/Kqueue.zig @@ -1183,6 +1183,7 @@ fn netReceive( handle: net.Socket.Handle, message_buffer: []net.IncomingMessage, data_buffer: []u8, + control_buffer: []u8, flags: net.ReceiveFlags, timeout: Io.Timeout, ) struct { ?net.Socket.ReceiveTimeoutError, usize } { @@ -1191,6 +1192,7 @@ fn netReceive( _ = handle; _ = message_buffer; _ = data_buffer; + _ = control_buffer; _ = flags; _ = timeout; @panic("TODO"); diff --git a/lib/std/Io/Threaded.zig b/lib/std/Io/Threaded.zig index da5adae30bf6d4683ec0ee85a1ef524e2398c52c..33f9083bc54896d1c1741293eb38eb28033e8046 100644 --- a/lib/std/Io/Threaded.zig +++ b/lib/std/Io/Threaded.zig @@ -2555,8 +2555,8 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper .device_io_control => |*o| return .{ .device_io_control = try deviceIoControl(o) }, .net_receive => |*o| return .{ .net_receive = o: { if (!have_networking) break :o .{ error.NetworkDown, 0 }; - if (is_windows) break :o netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.flags); - netReceivePosix(o.socket_handle, &o.message_buffer[0], o.data_buffer, o.flags, false) catch |err| switch (err) { + if (is_windows) break :o netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.control_buffer, o.flags); + netReceiveOnePosix(o.socket_handle, &o.message_buffer[0], o.data_buffer, o.control_buffer, o.flags, false) catch |err| switch (err) { error.Canceled => |e| return e, error.WouldBlock => unreachable, else => |e| break :o .{ e, 0 }, @@ -2825,9 +2825,11 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout .device_io_control => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.OUT | posix.POLL.ERR), .net_receive => |*o| nb: { var data_i: usize = 0; + var control_i: usize = 0; const result: Io.Operation.Result = .{ .net_receive = for (o.message_buffer, 0..) |*msg, msg_i| { const remaining_data_buffer = o.data_buffer[data_i..]; - netReceivePosix(o.socket_handle, msg, remaining_data_buffer, o.flags, true) catch |err| switch (err) { + const remaining_control_buffer = o.control_buffer[control_i..]; + netReceiveOnePosix(o.socket_handle, msg, remaining_data_buffer, remaining_control_buffer, o.flags, true) catch |err| switch (err) { error.Canceled => |e| return e, error.WouldBlock => { if (msg_i != 0) break .{ null, msg_i }; @@ -2837,6 +2839,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout else => |e| break .{ e, 0 }, }; data_i += msg.data.len; + control_i += msg.control.len; } else .{ null, o.message_buffer.len } }; switch (b.completed.tail) { .none => b.completed.head = index, @@ -3279,7 +3282,7 @@ fn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io // TODO integrate with overlapped I/O or equivalent to avoid this error if (concurrency) return error.ConcurrencyUnavailable; batchCompleteBlockingWindows(b, operation_userdata, .{ - .net_receive = netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.flags), + .net_receive = netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.control_buffer, o.flags), }); }, .net_send => |*o| { @@ -13212,10 +13215,11 @@ fn netSendManyPosix( } } -fn netReceivePosix( +fn netReceiveOnePosix( socket_handle: net.Socket.Handle, message: *net.IncomingMessage, data_buffer: []u8, + control_buffer: []u8, flags: net.ReceiveFlags, nonblocking: bool, ) (net.Socket.ReceiveError || error{WouldBlock})!void { @@ -13238,22 +13242,24 @@ fn netReceivePosix( .namelen = @sizeOf(PosixAddress), .iov = (&iov)[0..1], .iovlen = 1, - .control = message.control.ptr, - .controllen = @intCast(message.control.len), + .control = control_buffer.ptr, + .controllen = @intCast(control_buffer.len), .flags = 0, }; const syscall = try Syscall.start(); while (true) { const rc = posix.system.recvmsg(socket_handle, &msg, posix_flags); + const rc_control = msg.controllen; switch (posix.errno(rc)) { .SUCCESS => { syscall.finish(); const data = data_buffer[0..@intCast(rc)]; + const control = control_buffer[0..@intCast(rc_control)]; message.* = .{ .from = addressFromPosix(&storage), .data = data, - .control = if (msg.control) |ptr| @as([*]u8, @ptrCast(ptr))[0..msg.controllen] else message.control, + .control = control, .flags = .{ .eor = (msg.flags & posix.MSG.EOR) != 0, .trunc = (msg.flags & posix.MSG.TRUNC) != 0, @@ -13294,8 +13300,12 @@ fn netReceiveWindows( socket_handle: net.Socket.Handle, message_buffer: []net.IncomingMessage, data_buffer: []u8, + control_buffer: []u8, flags: net.ReceiveFlags, ) struct { ?net.Socket.ReceiveError, usize } { + // Windows doesn't have the concept of control/ancillary data. + _ = control_buffer; + t.netReceiveOneWindows(socket_handle, &message_buffer[0], data_buffer, flags) catch |err| return .{ err, 0 }; return .{ null, 1 }; diff --git a/lib/std/Io/Uring.zig b/lib/std/Io/Uring.zig index 907d96c6fc7d12264d6a1615ff1f46133dc5914a..8e0e0c138479d04785de62bd03e23a7f4c908ffc 100644 --- a/lib/std/Io/Uring.zig +++ b/lib/std/Io/Uring.zig @@ -2095,7 +2095,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper }, .net_receive => |o| .{ .net_receive = r: { - const opt_err, const n = ev.netReceive(&maybe_sync.cancel_region, o.socket_handle, o.message_buffer, o.data_buffer, o.flags); + const opt_err, const n = ev.netReceive(&maybe_sync.cancel_region, o.socket_handle, o.message_buffer, o.data_buffer, o.control_buffer, o.flags); break :r .{ if (opt_err) |err| switch (err) { error.Canceled => |e| return e, @@ -5075,15 +5075,18 @@ fn netReceive( handle: net.Socket.Handle, message_buffer: []net.IncomingMessage, data_buffer: []u8, + control_buffer: []u8, flags: net.ReceiveFlags, ) struct { ?net.Socket.ReceiveError, usize } { var message_i: usize = 0; var data_i: usize = 0; + var control_i: usize = 0; while (true) { if (message_buffer.len - message_i == 0) return .{ null, message_i }; const message = &message_buffer[message_i]; const remaining_data_buffer = data_buffer[data_i..]; + const remaining_control_buffer = control_buffer[control_i..]; var storage: PosixAddress = undefined; var iov: iovec = .{ .base = remaining_data_buffer.ptr, .len = remaining_data_buffer.len }; var msg: linux.msghdr = .{ @@ -5091,8 +5094,8 @@ fn netReceive( .namelen = @sizeOf(PosixAddress), .iov = (&iov)[0..1], .iovlen = 1, - .control = message.control.ptr, - .controllen = @intCast(message.control.len), + .control = remaining_control_buffer.ptr, + .controllen = @intCast(remaining_control_buffer.len), .flags = undefined, }; @@ -5121,11 +5124,13 @@ fn netReceive( switch (completion.errno()) { .SUCCESS => { const data = remaining_data_buffer[0..@intCast(completion.result)]; + const control = remaining_control_buffer[0..@intCast(msg.controllen)]; data_i += data.len; + control_i += @intCast(msg.controllen); message.* = .{ .from = addressFromPosix(&storage), .data = data, - .control = if (msg.control) |ptr| @as([*]u8, @ptrCast(ptr))[0..msg.controllen] else message.control, + .control = control, .flags = .{ .eor = msg.flags & linux.MSG.EOR != 0, .trunc = msg.flags & linux.MSG.TRUNC != 0, diff --git a/lib/std/Io/net.zig b/lib/std/Io/net.zig index ebaab6de90cc3ef610e0245fe952a90011d6c9f1..a7e726d8a3b718f0f58729e6c194f05c7045681a 100644 --- a/lib/std/Io/net.zig +++ b/lib/std/Io/net.zig @@ -924,22 +924,20 @@ pub const ReceiveFlags = packed struct(u8) { _: u5 = 0, }; +/// Fields are populated by receive functions. pub const IncomingMessage = struct { - /// Populated by receive functions. from: IpAddress, - /// Populated by receive functions, points into the caller-supplied buffer. + /// Points into caller-supplied buffer. data: []u8, - /// Supplied by caller before calling receive functions; mutated by receive - /// functions. + /// Points into caller-supplied buffer. control: []u8, - /// Populated by receive functions. flags: Flags, - /// Useful for initializing before calling `receiveManyTimeout`. + /// Deprecated. Initialize with `undefined`. pub const init: IncomingMessage = .{ .from = undefined, .data = undefined, - .control = &.{}, + .control = undefined, .flags = undefined, }; @@ -1162,13 +1160,30 @@ pub const Socket = struct { /// Waits for data. Connectionless. /// /// See also: + /// * `receiveWithControl` /// * `receiveTimeout` + /// * `receiveWithControlTimeout` + /// * `receiveManyTimeout` + /// * `receiveManyWithControlTimeout` pub fn receive(s: *const Socket, io: Io, buffer: []u8) ReceiveError!IncomingMessage { - var message: IncomingMessage = .init; + return s.receiveWithControl(io, buffer, &.{}); + } + + /// Waits for data, allowing the reception of control data. Connectionless. + /// + /// See also: + /// * `receive` + /// * `receiveTimeout` + /// * `receiveWithControlTimeout` + /// * `receiveManyTimeout` + /// * `receiveManyWithControlTimeout` + pub fn receiveWithControl(s: *const Socket, io: Io, buffer: []u8, control: []u8) ReceiveError!IncomingMessage { + var message: IncomingMessage = undefined; const maybe_err, const count = (try io.operate(.{ .net_receive = .{ .socket_handle = s.handle, .message_buffer = (&message)[0..1], .data_buffer = buffer, + .control_buffer = control, .flags = .{}, } })).net_receive; if (maybe_err) |err| return err; @@ -1184,18 +1199,42 @@ pub const Socket = struct { /// /// See also: /// * `receive` + /// * `receiveWithControl` + /// * `receiveWithControlTimeout` /// * `receiveManyTimeout` + /// * `receiveManyWithControlTimeout` pub fn receiveTimeout( s: *const Socket, io: Io, buffer: []u8, timeout: Io.Timeout, ) ReceiveTimeoutError!IncomingMessage { - var message: IncomingMessage = .init; + return s.receiveWithControlTimeout(io, buffer, &.{}, timeout); + } + + /// Waits for data, allowing the reception of control data. Connectionless. + /// + /// Returns `error.Timeout` if no message arrives early enough. + /// + /// See also: + /// * `receive` + /// * `receiveWithControl` + /// * `receiveTimeout` + /// * `receiveManyTimeout` + /// * `receiveManyWithControlTimeout` + pub fn receiveWithControlTimeout( + s: *const Socket, + io: Io, + buffer: []u8, + control: []u8, + timeout: Io.Timeout, + ) ReceiveTimeoutError!IncomingMessage { + var message: IncomingMessage = undefined; const maybe_err, const count = (try io.operateTimeout(.{ .net_receive = .{ .socket_handle = s.handle, .message_buffer = (&message)[0..1], .data_buffer = buffer, + .control_buffer = control, .flags = .{}, } }, timeout)).net_receive; if (maybe_err) |err| return err; @@ -1211,21 +1250,47 @@ pub const Socket = struct { /// /// See also: /// * `receive` + /// * `receiveWithControl` /// * `receiveTimeout` + /// * `receiveWithControlTimeout` + /// * `receiveManyWithControlTimeout` pub fn receiveManyTimeout( s: *const Socket, io: Io, - /// Function assumes each element has initialized `control` field. - /// Initializing with `IncomingMessage.init` may be helpful. message_buffer: []IncomingMessage, data_buffer: []u8, flags: ReceiveFlags, timeout: Io.Timeout, + ) struct { ?ReceiveTimeoutError, usize } { + return s.receiveManyWithControlTimeout(io, message_buffer, data_buffer, &.{}, flags, timeout); + } + + /// Waits until at least one message is delivered, possibly returning more + /// than one message. Allows reception of control data. Connectionless. + /// + /// Returns number of messages received, or `error.Timeout` if no message + /// arrives early enough. + /// + /// See also: + /// * `receive` + /// * `receiveWithControl` + /// * `receiveTimeout` + /// * `receiveWithControlTimeout` + /// * `receiveManyTimeout` + pub fn receiveManyWithControlTimeout( + s: *const Socket, + io: Io, + message_buffer: []IncomingMessage, + data_buffer: []u8, + control_buffer: []u8, + flags: ReceiveFlags, + timeout: Io.Timeout, ) struct { ?ReceiveTimeoutError, usize } { const result = io.operateTimeout(.{ .net_receive = .{ .socket_handle = s.handle, .message_buffer = message_buffer, .data_buffer = data_buffer, + .control_buffer = control_buffer, .flags = flags, } }, timeout) catch |err| return .{ err, 0 }; return result.net_receive; -- 2.54.0