authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-01 16:07:50-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log96cf75977bd46ccf4d0626183bc1d9e3e5d80eac
tree768c9a67603ddb8213169294b9beea22881360a9
parent29d4de53d6bb970050981d6c0f5fd38372b31345

std.Io: implement netSend

5 files changed, 127 insertions(+), 37 deletions(-)

lib/std/Io.zig+1-1
......@@ -671,7 +671,7 @@ pub const VTable = struct {
671671 listen: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Server,
672672 accept: *const fn (?*anyopaque, server: *net.Server) net.Server.AcceptError!net.Stream,
673673 ipBind: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket,
674 netSend: *const fn (?*anyopaque, net.Socket.Handle, []const net.OutgoingMessage, net.SendFlags) net.Socket.SendError!void,
674 netSend: *const fn (?*anyopaque, net.Socket.Handle, []net.OutgoingMessage, net.SendFlags) net.Socket.SendError!void,
675675 netReceive: *const fn (?*anyopaque, handle: net.Socket.Handle, buffer: []u8, timeout: Timeout) net.Socket.ReceiveTimeoutError!net.ReceivedMessage,
676676 netRead: *const fn (?*anyopaque, src: net.Stream, data: [][]u8) net.Stream.Reader.Error!usize,
677677 netWrite: *const fn (?*anyopaque, dest: net.Stream, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize,
lib/std/Io/Threaded.zig+96-15
......@@ -1108,8 +1108,8 @@ fn listenPosix(
11081108 }
11091109
11101110 var storage: PosixAddress = undefined;
1111 var socklen = addressToPosix(address, &storage);
1112 try posixBind(pool, socket_fd, &storage.any, socklen);
1111 var addr_len = addressToPosix(&address, &storage);
1112 try posixBind(pool, socket_fd, &storage.any, addr_len);
11131113
11141114 while (true) {
11151115 try pool.checkCancel();
......@@ -1121,7 +1121,7 @@ fn listenPosix(
11211121 }
11221122 }
11231123
1124 try posixGetSockName(pool, socket_fd, &storage.any, &socklen);
1124 try posixGetSockName(pool, socket_fd, &storage.any, &addr_len);
11251125 return .{
11261126 .socket = .{
11271127 .handle = socket_fd,
......@@ -1226,9 +1226,9 @@ fn ipBindPosix(
12261226 }
12271227
12281228 var storage: PosixAddress = undefined;
1229 var socklen = addressToPosix(address, &storage);
1230 try posixBind(pool, socket_fd, &storage.any, socklen);
1231 try posixGetSockName(pool, socket_fd, &storage.any, &socklen);
1229 var addr_len = addressToPosix(&address, &storage);
1230 try posixBind(pool, socket_fd, &storage.any, addr_len);
1231 try posixGetSockName(pool, socket_fd, &storage.any, &addr_len);
12321232 return .{
12331233 .handle = socket_fd,
12341234 .address = addressFromPosix(&storage),
......@@ -1306,21 +1306,102 @@ fn netReadPosix(userdata: ?*anyopaque, stream: Io.net.Stream, data: [][]u8) Io.n
13061306 return n;
13071307}
13081308
1309const have_sendmmsg = builtin.os.tag == .linux;
1310
13091311fn netSend(
13101312 userdata: ?*anyopaque,
13111313 handle: Io.net.Socket.Handle,
1312 messages: []const Io.net.OutgoingMessage,
1314 messages: []Io.net.OutgoingMessage,
13131315 flags: Io.net.SendFlags,
13141316) Io.net.Socket.SendError!void {
13151317 const pool: *Pool = @ptrCast(@alignCast(userdata));
1316 try pool.checkCancel();
13171318
1318 _ = handle;
1319 _ = messages;
1320 _ = flags;
1319 if (have_sendmmsg) {
1320 var i: usize = 0;
1321 while (messages.len - i != 0) {
1322 i += try netSendMany(pool, handle, messages[i..], flags);
1323 }
1324 return;
1325 }
1326
1327 try pool.checkCancel();
13211328 @panic("TODO");
13221329}
13231330
1331fn netSendMany(
1332 pool: *Pool,
1333 handle: Io.net.Socket.Handle,
1334 messages: []Io.net.OutgoingMessage,
1335 flags: Io.net.SendFlags,
1336) Io.net.Socket.SendError!usize {
1337 var msg_buffer: [64]std.os.linux.mmsghdr = undefined;
1338 var addr_buffer: [msg_buffer.len]PosixAddress = undefined;
1339 var iovecs_buffer: [msg_buffer.len]posix.iovec = undefined;
1340 const min_len: usize = @min(messages.len, msg_buffer.len);
1341 const clamped_messages = messages[0..min_len];
1342 const clamped_msgs = (&msg_buffer)[0..min_len];
1343 const clamped_addrs = (&addr_buffer)[0..min_len];
1344 const clamped_iovecs = (&iovecs_buffer)[0..min_len];
1345
1346 for (clamped_messages, clamped_msgs, clamped_addrs, clamped_iovecs) |*message, *msg, *addr, *iovec| {
1347 iovec.* = .{ .base = @constCast(message.data_ptr), .len = message.data_len };
1348 msg.* = .{
1349 .hdr = .{
1350 .name = &addr.any,
1351 .namelen = addressToPosix(message.address, addr),
1352 .iov = iovec[0..1],
1353 .iovlen = 1,
1354 .control = @constCast(message.control.ptr),
1355 .controllen = message.control.len,
1356 .flags = 0,
1357 },
1358 .len = undefined, // Populated by calling sendmmsg below.
1359 };
1360 }
1361
1362 const posix_flags: u32 =
1363 @as(u32, if (flags.confirm) posix.MSG.CONFIRM else 0) |
1364 @as(u32, if (flags.dont_route) posix.MSG.DONTROUTE else 0) |
1365 @as(u32, if (flags.eor) posix.MSG.EOR else 0) |
1366 @as(u32, if (flags.oob) posix.MSG.OOB else 0) |
1367 @as(u32, if (flags.fastopen) posix.MSG.FASTOPEN else 0) |
1368 posix.MSG.NOSIGNAL;
1369
1370 while (true) {
1371 try pool.checkCancel();
1372 const rc = posix.system.sendmmsg(handle, clamped_msgs.ptr, @intCast(clamped_msgs.len), posix_flags);
1373 switch (posix.errno(rc)) {
1374 .SUCCESS => {
1375 for (clamped_messages[0..rc], clamped_msgs[0..rc]) |*message, *msg| {
1376 message.data_len = msg.len;
1377 }
1378 return rc;
1379 },
1380 .AGAIN => |err| return errnoBug(err),
1381 .ALREADY => return error.FastOpenAlreadyInProgress,
1382 .BADF => |err| return errnoBug(err), // Always a race condition.
1383 .CONNRESET => return error.ConnectionResetByPeer,
1384 .DESTADDRREQ => |err| return errnoBug(err), // The socket is not connection-mode, and no peer address is set.
1385 .FAULT => |err| return errnoBug(err), // An invalid user space address was specified for an argument.
1386 .INTR => continue,
1387 .INVAL => |err| return errnoBug(err), // Invalid argument passed.
1388 .ISCONN => |err| return errnoBug(err), // connection-mode socket was connected already but a recipient was specified
1389 .MSGSIZE => return error.MessageOversize,
1390 .NOBUFS => return error.SystemResources,
1391 .NOMEM => return error.SystemResources,
1392 .NOTSOCK => |err| return errnoBug(err), // The file descriptor sockfd does not refer to a socket.
1393 .OPNOTSUPP => |err| return errnoBug(err), // Some bit in the flags argument is inappropriate for the socket type.
1394 .PIPE => return error.SocketNotConnected,
1395 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1396 .HOSTUNREACH => return error.NetworkUnreachable,
1397 .NETUNREACH => return error.NetworkUnreachable,
1398 .NOTCONN => return error.SocketNotConnected,
1399 .NETDOWN => return error.NetworkDown,
1400 else => |err| return posix.unexpectedErrno(err),
1401 }
1402 }
1403}
1404
13241405fn netReceive(
13251406 userdata: ?*anyopaque,
13261407 handle: Io.net.Socket.Handle,
......@@ -1503,13 +1584,13 @@ fn addressFromPosix(posix_address: *PosixAddress) Io.net.IpAddress {
15031584 };
15041585}
15051586
1506fn addressToPosix(a: Io.net.IpAddress, storage: *PosixAddress) posix.socklen_t {
1507 return switch (a) {
1587fn addressToPosix(a: *const Io.net.IpAddress, storage: *PosixAddress) posix.socklen_t {
1588 return switch (a.*) {
15081589 .ip4 => |ip4| {
15091590 storage.in = address4ToPosix(ip4);
15101591 return @sizeOf(posix.sockaddr.in);
15111592 },
1512 .ip6 => |ip6| {
1593 .ip6 => |*ip6| {
15131594 storage.in6 = address6ToPosix(ip6);
15141595 return @sizeOf(posix.sockaddr.in6);
15151596 },
......@@ -1539,7 +1620,7 @@ fn address4ToPosix(a: Io.net.Ip4Address) posix.sockaddr.in {
15391620 };
15401621}
15411622
1542fn address6ToPosix(a: Io.net.Ip6Address) posix.sockaddr.in6 {
1623fn address6ToPosix(a: *const Io.net.Ip6Address) posix.sockaddr.in6 {
15431624 return .{
15441625 .port = std.mem.nativeToBig(u16, a.port),
15451626 .flowinfo = a.flow,
lib/std/Io/net.zig+23-11
......@@ -154,7 +154,7 @@ pub const IpAddress = union(enum) {
154154 /// A nonexistent interface was requested or the requested address was not local.
155155 AddressUnavailable,
156156 /// The local network interface used to reach the destination is offline.
157 NetworkSubsystemDown,
157 NetworkDown,
158158 /// Insufficient memory or other resource internal to the operating system.
159159 SystemResources,
160160 /// Per-process limit on the number of open file descriptors has been reached.
......@@ -192,7 +192,7 @@ pub const IpAddress = union(enum) {
192192 /// Insufficient memory or other resource internal to the operating system.
193193 SystemResources,
194194 /// The local network interface used to reach the destination is offline.
195 NetworkSubsystemDown,
195 NetworkDown,
196196 ProtocolUnsupportedBySystem,
197197 ProtocolUnsupportedByAddressFamily,
198198 /// Per-process limit on the number of open file descriptors has been reached.
......@@ -702,7 +702,10 @@ pub const ReceivedMessage = struct {
702702
703703pub const OutgoingMessage = struct {
704704 address: *const IpAddress,
705 data: []const u8,
705 data_ptr: [*]const u8,
706 /// Initialized with how many bytes of `data_ptr` to send. After sending
707 /// succeeds, replaced with how many bytes were actually sent.
708 data_len: usize,
706709 control: []const u8 = &.{},
707710};
708711
......@@ -808,9 +811,10 @@ pub const Socket = struct {
808811 }
809812
810813 pub const SendError = error{
811 /// The socket type requires that message be sent atomically, and the size of the message
812 /// to be sent made this impossible. The message is not transmitted.
813 MessageTooBig,
814 /// The socket type requires that message be sent atomically, and the
815 /// size of the message to be sent made this impossible. The message
816 /// was not transmitted, or was partially transmitted.
817 MessageOversize,
814818 /// The output queue for a network interface was full. This generally indicates that the
815819 /// interface has stopped sending, but may be caused by transient congestion. (Normally,
816820 /// this does not occur in Linux. Packets are just silently dropped when a device queue
......@@ -823,21 +827,29 @@ pub const Socket = struct {
823827 /// Network reached but no route to host.
824828 HostUnreachable,
825829 /// The local network interface used to reach the destination is offline.
826 NetworkSubsystemDown,
830 NetworkDown,
827831 /// The destination address is not listening. Can still occur for
828832 /// connectionless messages.
829833 ConnectionRefused,
830834 /// Operating system or protocol does not support the address family.
831835 AddressFamilyUnsupported,
836 /// Another TCP Fast Open is already in progress.
837 FastOpenAlreadyInProgress,
838 /// Network connection was unexpectedly closed by recipient.
839 ConnectionResetByPeer,
840 /// Local end has been shut down on a connection-oriented socket, or
841 /// the socket was never connected.
842 SocketNotConnected,
832843 } || Io.UnexpectedError || Io.Cancelable;
833844
834 /// Transfers `data` to `dest`, connectionless.
845 /// Transfers `data` to `dest`, connectionless, in one packet.
835846 pub fn send(s: *const Socket, io: Io, dest: *const IpAddress, data: []const u8) SendError!void {
836 const message: OutgoingMessage = .{ .address = dest, .data = data };
837 return io.vtable.netSend(io.userdata, s.handle, &.{message}, .{});
847 var message: OutgoingMessage = .{ .address = dest, .data_ptr = data.ptr, .data_len = data.len };
848 try io.vtable.netSend(io.userdata, s.handle, &message, .{});
849 if (message.data_len != data.len) return error.MessageOversize;
838850 }
839851
840 pub fn sendMany(s: *const Socket, io: Io, messages: []const OutgoingMessage, flags: SendFlags) SendError!void {
852 pub fn sendMany(s: *const Socket, io: Io, messages: []OutgoingMessage, flags: SendFlags) SendError!void {
841853 return io.vtable.netSend(io.userdata, s.handle, messages, flags);
842854 }
843855
lib/std/Io/net/HostName.zig+6-4
......@@ -278,7 +278,8 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio
278278 for (mapped_nameservers) |*ns| {
279279 message_buffer[message_i] = .{
280280 .address = ns,
281 .data = query,
281 .data_ptr = query.ptr,
282 .data_len = query.len,
282283 };
283284 message_i += 1;
284285 }
......@@ -324,11 +325,12 @@ fn lookupDns(io: Io, lookup_canon_name: []const u8, rc: *const ResolvConf, optio
324325 if (next_answer_buffer == answers.len) break :send;
325326 },
326327 2 => {
327 const message: Io.net.OutgoingMessage = .{
328 var message: Io.net.OutgoingMessage = .{
328329 .address = ns,
329 .data = query,
330 .data_ptr = query.ptr,
331 .data_len = query.len,
330332 };
331 io.vtable.netSend(io.userdata, socket.handle, &.{message}, .{}) catch {};
333 io.vtable.netSend(io.userdata, socket.handle, (&message)[0..1], .{}) catch {};
332334 continue;
333335 },
334336 else => continue,
lib/std/os/linux.zig+1-6
......@@ -2011,7 +2011,7 @@ pub fn sendmsg(fd: i32, msg: *const msghdr_const, flags: u32) usize {
20112011 }
20122012}
20132013
2014pub fn sendmmsg(fd: i32, msgvec: [*]mmsghdr_const, vlen: u32, flags: u32) usize {
2014pub fn sendmmsg(fd: i32, msgvec: [*]mmsghdr, vlen: u32, flags: u32) usize {
20152015 return syscall4(.sendmmsg, @as(usize, @bitCast(@as(isize, fd))), @intFromPtr(msgvec), vlen, flags);
20162016}
20172017
......@@ -5884,11 +5884,6 @@ pub const mmsghdr = extern struct {
58845884 len: u32,
58855885};
58865886
5887pub const mmsghdr_const = extern struct {
5888 hdr: msghdr_const,
5889 len: u32,
5890};
5891
58925887pub const epoll_data = extern union {
58935888 ptr: usize,
58945889 fd: i32,