From 875cb2ff2f321ec40b02f8de40230a28e56d6844 Mon Sep 17 00:00:00 2001 From: hemisputnik Date: Wed, 22 Jul 2026 11:18:38 +0300 Subject: [PATCH] std.Io.net: read control data with .net_read Also adds a way to read control data using Stream.Reader. Stream.Reader will buffer control data until it can't no more, in which case it will be truncated. The readv syscall in netReadPosix has been replaced with recvmsg. --- lib/std/Io.zig | 3 +- lib/std/Io/Threaded.zig | 34 ++++++++++++++++----- lib/std/Io/net.zig | 65 +++++++++++++++++++++++++++++++++++++---- 3 files changed, 87 insertions(+), 15 deletions(-) diff --git a/lib/std/Io.zig b/lib/std/Io.zig index 7d43bef3bdd52f6e4bcbc17351b1e2c9107f25f0..e7bef51fce5b8160af6b5d1deef4869edd6e594a 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -431,6 +431,7 @@ pub const Operation = union(enum) { pub const NetRead = struct { socket_handle: net.Socket.Handle, data: [][]u8, + control: []u8 = &.{}, pub const Error = error{ SystemResources, @@ -444,7 +445,7 @@ pub const Operation = union(enum) { ConnectionTimedOut, } || Io.UnexpectedError; - pub const Result = Error!usize; + pub const Result = Error!net.Stream.ReadResult; }; pub const NetWrite = struct { diff --git a/lib/std/Io/Threaded.zig b/lib/std/Io/Threaded.zig index 33f9083bc54896d1c1741293eb38eb28033e8046..4ae0a35668728c0a17b1708d6cba113dfe843e60 100644 --- a/lib/std/Io/Threaded.zig +++ b/lib/std/Io/Threaded.zig @@ -2583,7 +2583,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper }, }, .net_read => |o| return .{ - .net_read = netRead(o.socket_handle, o.data) catch |err| switch (err) { + .net_read = netRead(o.socket_handle, o.data, o.control) catch |err| switch (err) { error.Canceled => |e| return e, else => |e| e, }, @@ -3296,7 +3296,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_read = netRead(o.socket_handle, o.data) catch |err| switch (err) { + .net_read = netRead(o.socket_handle, o.data, o.control) catch |err| switch (err) { error.Canceled => |e| return e, else => |e| e, }, @@ -12886,14 +12886,14 @@ fn deferAcceptAfd(t: *Threaded, listen_handle: net.Socket.Handle, info: windows. } } -fn netRead(socket_handle: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize { +fn netRead(socket_handle: net.Socket.Handle, data: [][]u8, control: []u8) net.Stream.Reader.Error!net.Stream.ReadResult { if (!have_networking) return error.NetworkDown; - if (is_windows) return netReadWindows(socket_handle, data); - return netReadPosix(socket_handle, data); + if (is_windows) return .{ .data_len = try netReadWindows(socket_handle, data) }; + return netReadPosix(socket_handle, data, control); } -fn netReadPosix(fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize { +fn netReadPosix(fd: net.Socket.Handle, data: [][]u8, control: []u8) net.Stream.Reader.Error!net.Stream.ReadResult { var iovecs_buffer: [max_iovecs_len]posix.iovec = undefined; var i: usize = 0; for (data) |buf| { @@ -12939,13 +12939,31 @@ fn netReadPosix(fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usi } } + var msg: posix.msghdr = .{ + .name = null, + .namelen = 0, + .iov = dest.ptr, + .iovlen = @intCast(dest.len), + .control = control.ptr, + .controllen = @intCast(control.len), + .flags = 0, + }; + + const flags: u32 = + @as(u32, if (@hasDecl(posix.MSG, "CMSG_CLOEXEC")) posix.MSG.CMSG_CLOEXEC else 0); + const syscall: Syscall = try .start(); while (true) { - const rc = posix.system.readv(fd, dest.ptr, @intCast(dest.len)); + const rc = posix.system.recvmsg(fd, &msg, flags); + const rc_control = msg.controllen; switch (posix.errno(rc)) { .SUCCESS => { syscall.finish(); - return @intCast(rc); + return .{ + .data_len = @intCast(rc), + .control_len = @intCast(rc_control), + .control_truncated = (msg.flags & posix.MSG.CTRUNC) != 0, + }; }, .INTR => { try syscall.checkCancel(); diff --git a/lib/std/Io/net.zig b/lib/std/Io/net.zig index a7e726d8a3b718f0f58729e6c194f05c7045681a..82bbcc0e707836d807cc5f7dea739f0668a583b5 100644 --- a/lib/std/Io/net.zig +++ b/lib/std/Io/net.zig @@ -1334,13 +1334,37 @@ pub const Stream = struct { const max_iovecs_len = 8; + pub const ReadResult = struct { + data_len: usize, + control_len: usize = 0, + /// Whether only some of the control data was received. + /// + /// When control data is truncated, the extra data essentially + /// disappears into the ether. This is indicative of a design problem, + /// such as the control buffer being too small. + control_truncated: bool = false, + }; + /// This is a low-level API that calls the `Io` interface function directly. /// For a higher level API, see `reader`. pub fn read(s: *const Stream, io: Io, data: [][]u8) Reader.Error!usize { - return (try io.operate(.{ .net_read = .{ + const rc, _ = try (try io.operate(.{ .net_read = .{ .socket_handle = s.socket.handle, .data = data, } })).net_read; + return rc; + } + + /// Read with control data. + /// + /// This is a low-level API that calls the `Io` interface function directly. + /// For a higher level API, see `reader`. + pub fn readWithControl(s: *const Stream, io: Io, data: [][]u8, control: []u8) Reader.Error!ReadResult { + return try (try io.operate(.{ .net_read = .{ + .socket_handle = s.socket.handle, + .data = data, + .control = control, + } })).net_read; } pub fn close(s: *const Stream, io: Io) void { @@ -1354,12 +1378,20 @@ pub const Stream = struct { pub const Reader = struct { io: Io, interface: Io.Reader, + control_buffer: []u8, + control_len: usize, + control_truncated: bool, stream: Stream, err: ?Error, pub const Error = Io.Operation.NetRead.Error || Io.Cancelable; pub fn init(stream: Stream, io: Io, buffer: []u8) Reader { + return initWithControl(stream, io, buffer, &.{}); + } + + /// Same as `init`, but also provides a buffer for storing control data. + pub fn initWithControl(stream: Stream, io: Io, buffer: []u8, control_buffer: []u8) Reader { return .{ .io = io, .interface = .{ @@ -1371,6 +1403,9 @@ pub const Stream = struct { .seek = 0, .end = 0, }, + .control_buffer = control_buffer, + .control_len = 0, + .control_truncated = false, .stream = stream, .err = null, }; @@ -1391,18 +1426,31 @@ pub const Stream = struct { const dest_n, const data_size = try io_r.writableVector(&iovecs_buffer, data); const dest = iovecs_buffer[0..dest_n]; assert(dest[0].len > 0); - const n = r.stream.read(io, dest) catch |err| { + const result = r.stream.readWithControl(io, dest, r.control_buffer[r.control_len..]) catch |err| { r.err = err; return error.ReadFailed; }; - if (n == 0) { + r.control_len += result.control_len; + r.control_truncated = r.control_truncated or result.control_truncated; + if (result.data_len == 0) { return error.EndOfStream; } - if (n > data_size) { - r.interface.end += n - data_size; + if (result.data_len > data_size) { + r.interface.end += result.data_len - data_size; return data_size; } - return n; + return result.data_len; + } + + /// Get buffered control data. + pub fn controlSlice(r: *const Reader) []u8 { + return r.control_buffer[0..r.control_len]; + } + + /// Clear buffered control data. + pub fn clearControl(r: *Reader) void { + r.control_len = 0; + r.control_truncated = false; } }; @@ -1489,6 +1537,11 @@ pub const Stream = struct { return .init(stream, io, buffer); } + /// Same as `reader`, but also provides a buffer for storing control data. + pub fn readerWithControl(stream: Stream, io: Io, buffer: []u8, control_buffer: []u8) Reader { + return .initWithControl(stream, io, buffer, control_buffer); + } + pub fn writer(stream: Stream, io: Io, buffer: []u8) Writer { return .init(stream, io, buffer); } -- 2.54.0