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); }