Commit 20355494085a6eb59b73816038a5d0c56b5492c6 | zig

author	hemisputnik <hemisputnik@proton.me> 2026-07-22 09:06:26+03:00
committer	Isaac Freund <mail@isaacfreund.com> 2026-10-01 10:55:12+02:00
log	20355494085a6eb59b73816038a5d0c56b5492c6
tree	07234e26640c2f1626046fb5853d199e8af45a9a
parent	8909d25b365de94d4b37c613ce3276634765a86b
signature	Commit is signed but in an unrecognized format.

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.

5 files changed, 106 insertions(+), 23 deletions(-)

lib
- std
  - Io.zig
  - Io
    Kqueue.zig
    Threaded.zig
    Uring.zig
    net.zig

lib/std/Io.zig+1

...	@@ -346,6 +346,7 @@ pub const Operation = union(enum) {	...	@@ -346,6 +346,7 @@ pub const Operation = union(enum) {
346	socket_handle: net.Socket.Handle,	346	socket_handle: net.Socket.Handle,
347	message_buffer: []net.IncomingMessage,	347	message_buffer: []net.IncomingMessage,
348	data_buffer: []u8,	348	data_buffer: []u8,
		349	control_buffer: []u8 = &.{},
349	flags: net.ReceiveFlags,	350	flags: net.ReceiveFlags,
350		351
351	pub const Error = error{	352	pub const Error = error{

lib/std/Io/Kqueue.zig+2

...	@@ -1183,6 +1183,7 @@ fn netReceive(	...	@@ -1183,6 +1183,7 @@ fn netReceive(
1183	handle: net.Socket.Handle,	1183	handle: net.Socket.Handle,
1184	message_buffer: []net.IncomingMessage,	1184	message_buffer: []net.IncomingMessage,
1185	data_buffer: []u8,	1185	data_buffer: []u8,
		1186	control_buffer: []u8,
1186	flags: net.ReceiveFlags,	1187	flags: net.ReceiveFlags,
1187	timeout: Io.Timeout,	1188	timeout: Io.Timeout,
1188	) struct { ?net.Socket.ReceiveTimeoutError, usize } {	1189	) struct { ?net.Socket.ReceiveTimeoutError, usize } {
...	@@ -1191,6 +1192,7 @@ fn netReceive(	...	@@ -1191,6 +1192,7 @@ fn netReceive(
1191	_ = handle;	1192	_ = handle;
1192	_ = message_buffer;	1193	_ = message_buffer;
1193	_ = data_buffer;	1194	_ = data_buffer;
		1195	_ = control_buffer;
1194	_ = flags;	1196	_ = flags;
1195	_ = timeout;	1197	_ = timeout;
1196	@panic("TODO");	1198	@panic("TODO");

lib/std/Io/Threaded.zig+18-8

 @@ -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);
+            if (is_windows) break :o netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.control_buffer, o.flags);
-            netReceivePosix(o.socket_handle, &o.message_buffer[0], o.data_buffer, o.flags, false) catch |err| switch (err) {
+            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,
+        .control = control_buffer.ptr,
-        .controllen = @intCast(message.control.len),
+        .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 };

lib/std/Io/Uring.zig+9-4

 @@ -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,
+            .control = remaining_control_buffer.ptr,
-            .controllen = @intCast(message.control.len),
+            .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,

lib/std/Io/net.zig+76-11

 @@ -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
+    /// Points into caller-supplied buffer.
-    /// functions.
     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;