From 2217d43cc91145c4f569108594f6d246d4733752 Mon Sep 17 00:00:00 2001 From: Isaac Freund Date: Thu, 1 Oct 2026 10:07:01 +0200 Subject: [PATCH] Revert "std.Io.net: improve reception of control data" It is not clear to me that this is the right path to take, using a shared buffer for control data makes managing the alignment trickier, and would require the Io interface itself to deal with alignment. For an example of what I consider clean design, see net_read/net_write. There is no need for the Io interface/operations to be aware of control buffer alignment and it is sufficient for the net.Stream.Reader abstraction to use an aligned buffer. Also this has no test coverage and is (to my knowledege) not currently blocking any real world Zig project. This reverts commit 7c0347fce0b6f510b43ca503ad9c1cd2f8ac9a6b. --- 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, 23 insertions(+), 106 deletions(-) diff --git a/lib/std/Io.zig b/lib/std/Io.zig index ab0f8514b407cb567c048db502eb3b52d4a27fd4..14c7306ca6673d99bad8f3b8b5eafc150f935f74 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -346,7 +346,6 @@ 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 f663ab366a157ea562809d58c4b03ef9f6861c76..8f8be91d59bd3f3131ad5c08955da9de09b90e26 100644 --- a/lib/std/Io/Kqueue.zig +++ b/lib/std/Io/Kqueue.zig @@ -1183,7 +1183,6 @@ 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 } { @@ -1192,7 +1191,6 @@ 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 77f14f6f991faa3ec821c8ad784445f1c2dd8d09..2c7cdbc8e99e37081200900866b94edf7672e272 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.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) { + 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) { error.Canceled => |e| return e, error.WouldBlock => unreachable, else => |e| break :o .{ e, 0 }, @@ -2825,11 +2825,9 @@ 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..]; - 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) { + netReceivePosix(o.socket_handle, msg, remaining_data_buffer, o.flags, true) catch |err| switch (err) { error.Canceled => |e| return e, error.WouldBlock => { if (msg_i != 0) break .{ null, msg_i }; @@ -2839,7 +2837,6 @@ 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, @@ -3282,7 +3279,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.control_buffer, o.flags), + .net_receive = netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.flags), }); }, .net_send => |*o| { @@ -13233,11 +13230,10 @@ fn netSendManyPosix( } } -fn netReceiveOnePosix( +fn netReceivePosix( 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 { @@ -13260,24 +13256,22 @@ fn netReceiveOnePosix( .namelen = @sizeOf(PosixAddress), .iov = (&iov)[0..1], .iovlen = 1, - .control = control_buffer.ptr, - .controllen = @intCast(control_buffer.len), + .control = message.control.ptr, + .controllen = @intCast(message.control.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 = control, + .control = if (msg.control) |ptr| @as([*]u8, @ptrCast(ptr))[0..msg.controllen] else message.control, .flags = .{ .eor = (msg.flags & posix.MSG.EOR) != 0, .trunc = (msg.flags & posix.MSG.TRUNC) != 0, @@ -13318,12 +13312,8 @@ 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 8e0e0c138479d04785de62bd03e23a7f4c908ffc..907d96c6fc7d12264d6a1615ff1f46133dc5914a 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.control_buffer, o.flags); + const opt_err, const n = ev.netReceive(&maybe_sync.cancel_region, o.socket_handle, o.message_buffer, o.data_buffer, o.flags); break :r .{ if (opt_err) |err| switch (err) { error.Canceled => |e| return e, @@ -5075,18 +5075,15 @@ 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 = .{ @@ -5094,8 +5091,8 @@ fn netReceive( .namelen = @sizeOf(PosixAddress), .iov = (&iov)[0..1], .iovlen = 1, - .control = remaining_control_buffer.ptr, - .controllen = @intCast(remaining_control_buffer.len), + .control = message.control.ptr, + .controllen = @intCast(message.control.len), .flags = undefined, }; @@ -5124,13 +5121,11 @@ 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 = control, + .control = if (msg.control) |ptr| @as([*]u8, @ptrCast(ptr))[0..msg.controllen] else message.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 9fe2661af17a69649629252ea200407a0151b798..7c5ddbde5e9a711f39d250587c1a4fa95922fd3e 100644 --- a/lib/std/Io/net.zig +++ b/lib/std/Io/net.zig @@ -924,20 +924,22 @@ pub const ReceiveFlags = packed struct(u8) { _: u5 = 0, }; -/// Fields are populated by receive functions. pub const IncomingMessage = struct { + /// Populated by receive functions. from: IpAddress, - /// Points into caller-supplied buffer. + /// Populated by receive functions, points into the caller-supplied buffer. data: []u8, - /// Points into caller-supplied buffer. + /// Supplied by caller before calling receive functions; mutated by receive + /// functions. control: []u8, + /// Populated by receive functions. flags: Flags, - /// Deprecated. Initialize with `undefined`. + /// Useful for initializing before calling `receiveManyTimeout`. pub const init: IncomingMessage = .{ .from = undefined, .data = undefined, - .control = undefined, + .control = &.{}, .flags = undefined, }; @@ -1160,30 +1162,13 @@ 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 { - 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; + var message: IncomingMessage = .init; 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; @@ -1199,42 +1184,18 @@ 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 { - 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; + var message: IncomingMessage = .init; 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; @@ -1250,47 +1211,21 @@ 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