authorgravatar for hemisputnik@proton.mehemisputnik <hemisputnik@proton.me> 2026-07-22 11:18:38+03:00
committergravatar for mail@isaacfreund.comIsaac Freund <mail@isaacfreund.com> 2026-10-01 10:55:12+02:00
log875cb2ff2f321ec40b02f8de40230a28e56d6844
tree6ca545a4e9eac344eef1b46a6865afac174c171e
parent20355494085a6eb59b73816038a5d0c56b5492c6
signature Commit is signed but in an unrecognized format.

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.

3 files changed, 87 insertions(+), 15 deletions(-)

lib/std/Io.zig+2-1
......@@ -431,6 +431,7 @@ pub const Operation = union(enum) {
431431 pub const NetRead = struct {
432432 socket_handle: net.Socket.Handle,
433433 data: [][]u8,
434 control: []u8 = &.{},
434435
435436 pub const Error = error{
436437 SystemResources,
......@@ -444,7 +445,7 @@ pub const Operation = union(enum) {
444445 ConnectionTimedOut,
445446 } || Io.UnexpectedError;
446447
447 pub const Result = Error!usize;
448 pub const Result = Error!net.Stream.ReadResult;
448449 };
449450
450451 pub const NetWrite = struct {
lib/std/Io/Threaded.zig+26-8
......@@ -2583,7 +2583,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper
25832583 },
25842584 },
25852585 .net_read => |o| return .{
2586 .net_read = netRead(o.socket_handle, o.data) catch |err| switch (err) {
2586 .net_read = netRead(o.socket_handle, o.data, o.control) catch |err| switch (err) {
25872587 error.Canceled => |e| return e,
25882588 else => |e| e,
25892589 },
......@@ -3296,7 +3296,7 @@ fn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io
32963296 // TODO integrate with overlapped I/O or equivalent to avoid this error
32973297 if (concurrency) return error.ConcurrencyUnavailable;
32983298 batchCompleteBlockingWindows(b, operation_userdata, .{
3299 .net_read = netRead(o.socket_handle, o.data) catch |err| switch (err) {
3299 .net_read = netRead(o.socket_handle, o.data, o.control) catch |err| switch (err) {
33003300 error.Canceled => |e| return e,
33013301 else => |e| e,
33023302 },
......@@ -12886,14 +12886,14 @@ fn deferAcceptAfd(t: *Threaded, listen_handle: net.Socket.Handle, info: windows.
1288612886 }
1288712887}
1288812888
12889fn netRead(socket_handle: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize {
12889fn netRead(socket_handle: net.Socket.Handle, data: [][]u8, control: []u8) net.Stream.Reader.Error!net.Stream.ReadResult {
1289012890 if (!have_networking) return error.NetworkDown;
1289112891
12892 if (is_windows) return netReadWindows(socket_handle, data);
12893 return netReadPosix(socket_handle, data);
12892 if (is_windows) return .{ .data_len = try netReadWindows(socket_handle, data) };
12893 return netReadPosix(socket_handle, data, control);
1289412894}
1289512895
12896fn netReadPosix(fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize {
12896fn netReadPosix(fd: net.Socket.Handle, data: [][]u8, control: []u8) net.Stream.Reader.Error!net.Stream.ReadResult {
1289712897 var iovecs_buffer: [max_iovecs_len]posix.iovec = undefined;
1289812898 var i: usize = 0;
1289912899 for (data) |buf| {
......@@ -12939,13 +12939,31 @@ fn netReadPosix(fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usi
1293912939 }
1294012940 }
1294112941
12942 var msg: posix.msghdr = .{
12943 .name = null,
12944 .namelen = 0,
12945 .iov = dest.ptr,
12946 .iovlen = @intCast(dest.len),
12947 .control = control.ptr,
12948 .controllen = @intCast(control.len),
12949 .flags = 0,
12950 };
12951
12952 const flags: u32 =
12953 @as(u32, if (@hasDecl(posix.MSG, "CMSG_CLOEXEC")) posix.MSG.CMSG_CLOEXEC else 0);
12954
1294212955 const syscall: Syscall = try .start();
1294312956 while (true) {
12944 const rc = posix.system.readv(fd, dest.ptr, @intCast(dest.len));
12957 const rc = posix.system.recvmsg(fd, &msg, flags);
12958 const rc_control = msg.controllen;
1294512959 switch (posix.errno(rc)) {
1294612960 .SUCCESS => {
1294712961 syscall.finish();
12948 return @intCast(rc);
12962 return .{
12963 .data_len = @intCast(rc),
12964 .control_len = @intCast(rc_control),
12965 .control_truncated = (msg.flags & posix.MSG.CTRUNC) != 0,
12966 };
1294912967 },
1295012968 .INTR => {
1295112969 try syscall.checkCancel();
lib/std/Io/net.zig+59-6
......@@ -1334,12 +1334,36 @@ pub const Stream = struct {
13341334
13351335 const max_iovecs_len = 8;
13361336
1337 pub const ReadResult = struct {
1338 data_len: usize,
1339 control_len: usize = 0,
1340 /// Whether only some of the control data was received.
1341 ///
1342 /// When control data is truncated, the extra data essentially
1343 /// disappears into the ether. This is indicative of a design problem,
1344 /// such as the control buffer being too small.
1345 control_truncated: bool = false,
1346 };
1347
13371348 /// This is a low-level API that calls the `Io` interface function directly.
13381349 /// For a higher level API, see `reader`.
13391350 pub fn read(s: *const Stream, io: Io, data: [][]u8) Reader.Error!usize {
1340 return (try io.operate(.{ .net_read = .{
1351 const rc, _ = try (try io.operate(.{ .net_read = .{
1352 .socket_handle = s.socket.handle,
1353 .data = data,
1354 } })).net_read;
1355 return rc;
1356 }
1357
1358 /// Read with control data.
1359 ///
1360 /// This is a low-level API that calls the `Io` interface function directly.
1361 /// For a higher level API, see `reader`.
1362 pub fn readWithControl(s: *const Stream, io: Io, data: [][]u8, control: []u8) Reader.Error!ReadResult {
1363 return try (try io.operate(.{ .net_read = .{
13411364 .socket_handle = s.socket.handle,
13421365 .data = data,
1366 .control = control,
13431367 } })).net_read;
13441368 }
13451369
......@@ -1354,12 +1378,20 @@ pub const Stream = struct {
13541378 pub const Reader = struct {
13551379 io: Io,
13561380 interface: Io.Reader,
1381 control_buffer: []u8,
1382 control_len: usize,
1383 control_truncated: bool,
13571384 stream: Stream,
13581385 err: ?Error,
13591386
13601387 pub const Error = Io.Operation.NetRead.Error || Io.Cancelable;
13611388
13621389 pub fn init(stream: Stream, io: Io, buffer: []u8) Reader {
1390 return initWithControl(stream, io, buffer, &.{});
1391 }
1392
1393 /// Same as `init`, but also provides a buffer for storing control data.
1394 pub fn initWithControl(stream: Stream, io: Io, buffer: []u8, control_buffer: []u8) Reader {
13631395 return .{
13641396 .io = io,
13651397 .interface = .{
......@@ -1371,6 +1403,9 @@ pub const Stream = struct {
13711403 .seek = 0,
13721404 .end = 0,
13731405 },
1406 .control_buffer = control_buffer,
1407 .control_len = 0,
1408 .control_truncated = false,
13741409 .stream = stream,
13751410 .err = null,
13761411 };
......@@ -1391,18 +1426,31 @@ pub const Stream = struct {
13911426 const dest_n, const data_size = try io_r.writableVector(&iovecs_buffer, data);
13921427 const dest = iovecs_buffer[0..dest_n];
13931428 assert(dest[0].len > 0);
1394 const n = r.stream.read(io, dest) catch |err| {
1429 const result = r.stream.readWithControl(io, dest, r.control_buffer[r.control_len..]) catch |err| {
13951430 r.err = err;
13961431 return error.ReadFailed;
13971432 };
1398 if (n == 0) {
1433 r.control_len += result.control_len;
1434 r.control_truncated = r.control_truncated or result.control_truncated;
1435 if (result.data_len == 0) {
13991436 return error.EndOfStream;
14001437 }
1401 if (n > data_size) {
1402 r.interface.end += n - data_size;
1438 if (result.data_len > data_size) {
1439 r.interface.end += result.data_len - data_size;
14031440 return data_size;
14041441 }
1405 return n;
1442 return result.data_len;
1443 }
1444
1445 /// Get buffered control data.
1446 pub fn controlSlice(r: *const Reader) []u8 {
1447 return r.control_buffer[0..r.control_len];
1448 }
1449
1450 /// Clear buffered control data.
1451 pub fn clearControl(r: *Reader) void {
1452 r.control_len = 0;
1453 r.control_truncated = false;
14061454 }
14071455 };
14081456
......@@ -1489,6 +1537,11 @@ pub const Stream = struct {
14891537 return .init(stream, io, buffer);
14901538 }
14911539
1540 /// Same as `reader`, but also provides a buffer for storing control data.
1541 pub fn readerWithControl(stream: Stream, io: Io, buffer: []u8, control_buffer: []u8) Reader {
1542 return .initWithControl(stream, io, buffer, control_buffer);
1543 }
1544
14921545 pub fn writer(stream: Stream, io: Io, buffer: []u8) Writer {
14931546 return .init(stream, io, buffer);
14941547 }