authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-16 23:40:32-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:50-07:00
logcf6fa219fd05b9f2c01e85557bcd140e72802459
tree7384deba8832c91ae257180835cc3c957912927f
parentd4215ffaa04b976400bd597cca0cca8182068bf6

std.Io.Threaded: fix netWrite cancellation

Move std.posix logic over rather than calling into it.

2 files changed, 87 insertions(+), 32 deletions(-)

lib/std/Io/Threaded.zig+61-25
......@@ -250,6 +250,19 @@ pub fn io(t: *Threaded) Io {
250250 };
251251}
252252
253const socket_flags_unsupported = builtin.os.tag.isDarwin() or native_os == .haiku; // 💩💩
254const have_accept4 = !socket_flags_unsupported;
255const have_flock_open_flags = @hasField(posix.O, "EXLOCK");
256const have_networking = builtin.os.tag != .wasi;
257const have_flock = @TypeOf(posix.system.flock) != void;
258const have_sendmmsg = builtin.os.tag == .linux;
259
260const openat_sym = if (posix.lfs64_abi) posix.system.openat64 else posix.system.openat;
261const fstat_sym = if (posix.lfs64_abi) posix.system.fstat64 else posix.system.fstat;
262const fstatat_sym = if (posix.lfs64_abi) posix.system.fstatat64 else posix.system.fstatat;
263const lseek_sym = if (posix.lfs64_abi) posix.system.lseek64 else posix.system.lseek;
264const preadv_sym = if (posix.lfs64_abi) posix.system.preadv64 else posix.system.preadv;
265
253266/// Trailing data:
254267/// 1. context
255268/// 2. result
......@@ -1143,13 +1156,6 @@ fn fileStatWasi(userdata: ?*anyopaque, file: Io.File) Io.File.StatError!Io.File.
11431156 }
11441157}
11451158
1146const have_flock = @TypeOf(posix.system.flock) != void;
1147const openat_sym = if (posix.lfs64_abi) posix.system.openat64 else posix.system.openat;
1148const fstat_sym = if (posix.lfs64_abi) posix.system.fstat64 else posix.system.fstat;
1149const fstatat_sym = if (posix.lfs64_abi) posix.system.fstatat64 else posix.system.fstatat;
1150const lseek_sym = if (posix.lfs64_abi) posix.system.lseek64 else posix.system.lseek;
1151const preadv_sym = if (posix.lfs64_abi) posix.system.preadv64 else posix.system.preadv;
1152
11531159fn dirAccessPosix(
11541160 userdata: ?*anyopaque,
11551161 dir: Io.Dir,
......@@ -1277,8 +1283,7 @@ fn dirCreateFilePosix(
12771283 // Use the O locking flags if the os supports them to acquire the lock
12781284 // atomically. Note that the NONBLOCK flag is removed after the openat()
12791285 // call is successful.
1280 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
1281 if (has_flock_open_flags) switch (flags.lock) {
1286 if (have_flock_open_flags) switch (flags.lock) {
12821287 .none => {},
12831288 .shared => {
12841289 os_flags.SHLOCK = true;
......@@ -1328,7 +1333,7 @@ fn dirCreateFilePosix(
13281333 };
13291334 errdefer posix.close(fd);
13301335
1331 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
1336 if (have_flock and !have_flock_open_flags and flags.lock != .none) {
13321337 const lock_nonblocking: i32 = if (flags.lock_nonblocking) posix.LOCK.NB else 0;
13331338 const lock_flags = switch (flags.lock) {
13341339 .none => unreachable,
......@@ -1352,7 +1357,7 @@ fn dirCreateFilePosix(
13521357 }
13531358 }
13541359
1355 if (has_flock_open_flags and flags.lock_nonblocking) {
1360 if (have_flock_open_flags and flags.lock_nonblocking) {
13561361 var fl_flags: usize = while (true) {
13571362 try t.checkCancel();
13581363 const rc = posix.system.fcntl(fd, posix.F.GETFL, @as(usize, 0));
......@@ -1476,8 +1481,7 @@ fn dirOpenFile(
14761481
14771482 // Use the O locking flags if the os supports them to acquire the lock
14781483 // atomically.
1479 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
1480 if (has_flock_open_flags) {
1484 if (have_flock_open_flags) {
14811485 // Note that the NONBLOCK flag is removed after the openat() call
14821486 // is successful.
14831487 switch (flags.lock) {
......@@ -1530,7 +1534,7 @@ fn dirOpenFile(
15301534 };
15311535 errdefer posix.close(fd);
15321536
1533 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
1537 if (have_flock and !have_flock_open_flags and flags.lock != .none) {
15341538 const lock_nonblocking: i32 = if (flags.lock_nonblocking) posix.LOCK.NB else 0;
15351539 const lock_flags = switch (flags.lock) {
15361540 .none => unreachable,
......@@ -1554,7 +1558,7 @@ fn dirOpenFile(
15541558 }
15551559 }
15561560
1557 if (has_flock_open_flags and flags.lock_nonblocking) {
1561 if (have_flock_open_flags and flags.lock_nonblocking) {
15581562 var fl_flags: usize = while (true) {
15591563 try t.checkCancel();
15601564 const rc = posix.system.fcntl(fd, posix.F.GETFL, @as(usize, 0));
......@@ -1954,7 +1958,7 @@ fn nowWasi(userdata: ?*anyopaque, clock: Io.Clock) Io.Clock.Error!Io.Timestamp {
19541958 var ns: std.os.wasi.timestamp_t = undefined;
19551959 const err = std.os.wasi.clock_time_get(clockToWasi(clock), 1, &ns);
19561960 if (err != .SUCCESS) return error.Unexpected;
1957 return ns;
1961 return .fromNanoseconds(ns);
19581962}
19591963
19601964fn sleepLinux(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void {
......@@ -2004,7 +2008,7 @@ fn sleepWasi(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void {
20042008
20052009 const clock: w.subscription_clock_t = if (try timeout.toDurationFromNow(t.io())) |d| .{
20062010 .id = clockToWasi(d.clock),
2007 .timeout = std.math.lossyCast(u64, d.duration.nanoseconds),
2011 .timeout = std.math.lossyCast(u64, d.raw.nanoseconds),
20082012 .precision = 0,
20092013 .flags = 0,
20102014 } else .{
......@@ -2083,6 +2087,7 @@ fn netListenIpPosix(
20832087 address: IpAddress,
20842088 options: IpAddress.ListenOptions,
20852089) IpAddress.ListenError!net.Server {
2090 if (!have_networking) return error.NetworkDown;
20862091 const t: *Threaded = @ptrCast(@alignCast(userdata));
20872092 const family = posixAddressFamily(&address);
20882093 const socket_fd = try openSocketPosix(t, family, .{
......@@ -2230,6 +2235,7 @@ fn posixConnect(t: *Threaded, socket_fd: posix.socket_t, addr: *const posix.sock
22302235 .ACCES => return error.AccessDenied,
22312236 .PERM => |err| return errnoBug(err),
22322237 .NOENT => |err| return errnoBug(err),
2238 .NETDOWN => return error.NetworkDown,
22332239 else => |err| return posix.unexpectedErrno(err),
22342240 }
22352241 }
......@@ -2306,6 +2312,7 @@ fn netConnectIpPosix(
23062312 address: *const IpAddress,
23072313 options: IpAddress.ConnectOptions,
23082314) IpAddress.ConnectError!net.Stream {
2315 if (!have_networking) return error.NetworkDown;
23092316 if (options.timeout != .none) @panic("TODO");
23102317 const t: *Threaded = @ptrCast(@alignCast(userdata));
23112318 const family = posixAddressFamily(address);
......@@ -2346,6 +2353,7 @@ fn netBindIpPosix(
23462353 address: *const IpAddress,
23472354 options: IpAddress.BindOptions,
23482355) IpAddress.BindError!net.Socket {
2356 if (!have_networking) return error.NetworkDown;
23492357 const t: *Threaded = @ptrCast(@alignCast(userdata));
23502358 const family = posixAddressFamily(address);
23512359 const socket_fd = try openSocketPosix(t, family, options);
......@@ -2421,9 +2429,6 @@ fn openSocketPosix(
24212429 return socket_fd;
24222430}
24232431
2424const socket_flags_unsupported = builtin.os.tag.isDarwin() or native_os == .haiku; // 💩💩
2425const have_accept4 = !socket_flags_unsupported;
2426
24272432fn netAcceptPosix(userdata: ?*anyopaque, listen_fd: net.Socket.Handle) net.Server.AcceptError!net.Stream {
24282433 const t: *Threaded = @ptrCast(@alignCast(userdata));
24292434 var storage: PosixAddress = undefined;
......@@ -2534,14 +2539,13 @@ fn netReadPosix(userdata: ?*anyopaque, fd: net.Socket.Handle, data: [][]u8) net.
25342539 }
25352540}
25362541
2537const have_sendmmsg = builtin.os.tag == .linux;
2538
25392542fn netSend(
25402543 userdata: ?*anyopaque,
25412544 handle: net.Socket.Handle,
25422545 messages: []net.OutgoingMessage,
25432546 flags: net.SendFlags,
25442547) struct { ?net.Socket.SendError, usize } {
2548 if (!have_networking) return .{ error.NetworkDown, 0 };
25452549 const t: *Threaded = @ptrCast(@alignCast(userdata));
25462550
25472551 const posix_flags: u32 =
......@@ -2703,7 +2707,7 @@ fn netSendMany(
27032707 .OPNOTSUPP => |err| return errnoBug(err), // Some bit in the flags argument is inappropriate for the socket type.
27042708 .PIPE => return error.SocketUnconnected,
27052709 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
2706 .HOSTUNREACH => return error.NetworkUnreachable,
2710 .HOSTUNREACH => return error.HostUnreachable,
27072711 .NETUNREACH => return error.NetworkUnreachable,
27082712 .NOTCONN => return error.SocketUnconnected,
27092713 .NETDOWN => return error.NetworkDown,
......@@ -2720,6 +2724,7 @@ fn netReceive(
27202724 flags: net.ReceiveFlags,
27212725 timeout: Io.Timeout,
27222726) struct { ?net.Socket.ReceiveTimeoutError, usize } {
2727 if (!have_networking) return .{ error.NetworkDown, 0 };
27232728 const t: *Threaded = @ptrCast(@alignCast(userdata));
27242729
27252730 // recvmmsg is useless, here's why:
......@@ -2847,8 +2852,8 @@ fn netWritePosix(
28472852 data: []const []const u8,
28482853 splat: usize,
28492854) net.Stream.Writer.Error!usize {
2855 if (!have_networking) return error.NetworkDown;
28502856 const t: *Threaded = @ptrCast(@alignCast(userdata));
2851 try t.checkCancel();
28522857
28532858 var iovecs: [max_iovecs_len]posix.iovec_const = undefined;
28542859 var msg: posix.msghdr_const = .{
......@@ -2889,7 +2894,37 @@ fn netWritePosix(
28892894 },
28902895 };
28912896 const flags = posix.MSG.NOSIGNAL;
2892 return posix.sendmsg(fd, &msg, flags);
2897 while (true) {
2898 try t.checkCancel();
2899 const rc = posix.system.sendmsg(fd, &msg, flags);
2900 switch (posix.errno(rc)) {
2901 .SUCCESS => return @intCast(rc),
2902 .INTR => continue,
2903 .CANCELED => return error.Canceled,
2904
2905 .ACCES => |err| return errnoBug(err),
2906 .AGAIN => |err| return errnoBug(err),
2907 .ALREADY => return error.FastOpenAlreadyInProgress,
2908 .BADF => |err| return errnoBug(err), // always a race condition
2909 .CONNRESET => return error.ConnectionResetByPeer,
2910 .DESTADDRREQ => |err| return errnoBug(err), // The socket is not connection-mode, and no peer address is set.
2911 .FAULT => |err| return errnoBug(err), // An invalid user space address was specified for an argument.
2912 .INVAL => |err| return errnoBug(err), // Invalid argument passed.
2913 .ISCONN => |err| return errnoBug(err), // connection-mode socket was connected already but a recipient was specified
2914 .MSGSIZE => |err| return errnoBug(err),
2915 .NOBUFS => return error.SystemResources,
2916 .NOMEM => return error.SystemResources,
2917 .NOTSOCK => |err| return errnoBug(err), // The file descriptor sockfd does not refer to a socket.
2918 .OPNOTSUPP => |err| return errnoBug(err), // Some bit in the flags argument is inappropriate for the socket type.
2919 .PIPE => return error.SocketUnconnected,
2920 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
2921 .HOSTUNREACH => return error.HostUnreachable,
2922 .NETUNREACH => return error.NetworkUnreachable,
2923 .NOTCONN => return error.SocketUnconnected,
2924 .NETDOWN => return error.NetworkDown,
2925 else => |err| return posix.unexpectedErrno(err),
2926 }
2927 }
28932928}
28942929
28952930fn addBuf(v: []posix.iovec_const, i: *@FieldType(posix.msghdr_const, "iovlen"), bytes: []const u8) void {
......@@ -2913,6 +2948,7 @@ fn netInterfaceNameResolve(
29132948 userdata: ?*anyopaque,
29142949 name: *const net.Interface.Name,
29152950) net.Interface.Name.ResolveError!net.Interface {
2951 if (!have_networking) return error.InterfaceNotFound;
29162952 const t: *Threaded = @ptrCast(@alignCast(userdata));
29172953
29182954 if (native_os == .linux) {
lib/std/Io/net.zig+26-7
......@@ -309,6 +309,7 @@ pub const IpAddress = union(enum) {
309309 AccessDenied,
310310 /// Non-blocking was requested and the operation cannot return immediately.
311311 WouldBlock,
312 NetworkDown,
312313 } || Io.Timeout.Error || Io.UnexpectedError || Io.Cancelable;
313314
314315 pub const ConnectOptions = struct {
......@@ -1062,7 +1063,7 @@ pub const Socket = struct {
10621063 AddressFamilyUnsupported,
10631064 /// Another TCP Fast Open is already in progress.
10641065 FastOpenAlreadyInProgress,
1065 /// Network connection was unexpectedly closed by recipient.
1066 /// Network session was unexpectedly closed by recipient.
10661067 ConnectionResetByPeer,
10671068 /// Local end has been shut down on a connection-oriented socket, or
10681069 /// the socket was never connected.
......@@ -1242,15 +1243,33 @@ pub const Stream = struct {
12421243 stream: Stream,
12431244 err: ?Error = null,
12441245
1245 pub const Error = std.posix.SendMsgError || error{
1246 pub const Error = error{
1247 /// Another TCP Fast Open is already in progress.
1248 FastOpenAlreadyInProgress,
1249 /// Network session was unexpectedly closed by recipient.
12461250 ConnectionResetByPeer,
1247 SocketNotBound,
1248 MessageOversize,
1249 NetworkDown,
1251 /// The output queue for a network interface was full. This generally indicates that the
1252 /// interface has stopped sending, but may be caused by transient congestion. (Normally,
1253 /// this does not occur in Linux. Packets are just silently dropped when a device queue
1254 /// overflows.)
1255 ///
1256 /// This is also caused when there is not enough kernel memory available.
12501257 SystemResources,
1258 /// No route to network.
1259 NetworkUnreachable,
1260 /// Network reached but no route to host.
1261 HostUnreachable,
1262 /// The local network interface used to reach the destination is down.
1263 NetworkDown,
1264 /// The destination address is not listening.
1265 ConnectionRefused,
1266 /// The passed address didn't have the correct address family in its sa_family field.
1267 AddressFamilyUnsupported,
1268 /// Local end has been shut down on a connection-oriented socket, or
1269 /// the socket was never connected.
12511270 SocketUnconnected,
1252 Unexpected,
1253 } || Io.Cancelable;
1271 SocketNotBound,
1272 } || Io.UnexpectedError || Io.Cancelable;
12541273
12551274 pub fn init(stream: Stream, io: Io, buffer: []u8) Writer {
12561275 return .{