| ... | @@ -201,7 +201,7 @@ const Closure = struct { | ... | @@ -201,7 +201,7 @@ const Closure = struct { |
| 201 | const Start = *const fn (*Closure, *Threaded) void; | 201 | const Start = *const fn (*Closure, *Threaded) void; |
| 202 | | 202 | |
| 203 | fn requestCancel(closure: *Closure, t: *Threaded) void { | 203 | fn requestCancel(closure: *Closure, t: *Threaded) void { |
| 204 | var signal_id = switch (@atomicRmw(CancelStatus, &closure.cancel_status, .Xchg, .requested, .monotonic).unpack()) { | 204 | const signal_id = switch (@atomicRmw(CancelStatus, &closure.cancel_status, .Xchg, .requested, .monotonic).unpack()) { |
| 205 | .none, .acknowledged, .requested => return, | 205 | .none, .acknowledged, .requested => return, |
| 206 | .signal_id => |signal_id| signal_id, | 206 | .signal_id => |signal_id| signal_id, |
| 207 | }; | 207 | }; |
| ... | @@ -214,54 +214,32 @@ const Closure = struct { | ... | @@ -214,54 +214,32 @@ const Closure = struct { |
| 214 | | 214 | |
| 215 | // The task will enter a blocking syscall before checking for cancellation again. | 215 | // The task will enter a blocking syscall before checking for cancellation again. |
| 216 | // We can send a signal to interrupt the syscall, but if it arrives before | 216 | // We can send a signal to interrupt the syscall, but if it arrives before |
| 217 | // the syscall instruction, it will be missed. Therefore, this code tries | 217 | // the syscall instruction, it will be missed. |
| 218 | // again until the cancellation request is acknowledged. | | |
| 219 | | | |
| 220 | // 1 << 10 ns is about 1 microsecond, approximately syscall overhead. | | |
| 221 | // 1 << 20 ns is about 1 millisecond. | | |
| 222 | // 1 << 30 ns is about 1 second. | | |
| 223 | // | | |
| 224 | // On a heavily loaded Linux 6.17.5, I observed a maximum of 20 | | |
| 225 | // attempts not acknowledged before the timeout (including exponential | | |
| 226 | // backoff) was sufficient, despite the heavy load. | | |
| 227 | // | 218 | // |
| 228 | // The time wasted here sleeping is mitigated by the fact that, later | 219 | // Unfortunately, trying again until the cancellation request is |
| 229 | // on, the system will likely wait for the canceled task, causing it | 220 | // acknowledged has been observed to incur a large amount of overhead, |
| 230 | // to indefinitely yield until the canceled task finishes, and the | 221 | // and usually strong cancellation guarantees are not needed, so the |
| 231 | // task must acknowledge the cancel before it proceeds to that point. | 222 | // race condition is not handled here. Users who want to avoid this |
| 232 | const max_attempts = 22; | 223 | // have this menu of options instead: |
| 233 | | 224 | // * Use no libc, in which case Zig std lib can avoid the race (tracking |
| 234 | for (0..max_attempts) |attempt_index| { | 225 | // issue: https://codeberg.org/ziglang/zig/issues/30049) |
| 235 | if (std.Thread.use_pthreads) { | 226 | // * Use musl libc instead of glibc |
| 236 | if (std.c.pthread_kill(signal_id, .IO) != 0) return; | 227 | // * Use `std.Io.Evented`. But this is not implemented yet. Tracked by |
| 237 | } else if (native_os == .linux) { | 228 | // - https://codeberg.org/ziglang/zig/issues/30050 |
| 238 | const pid: posix.pid_t = p: { | 229 | // - https://codeberg.org/ziglang/zig/issues/30051 |
| 239 | const cached_pid = @atomicLoad(Pid, &t.pid, .monotonic); | 230 | if (std.Thread.use_pthreads) { |
| 240 | if (cached_pid != .unknown) break :p @intFromEnum(cached_pid); | 231 | if (std.c.pthread_kill(signal_id, .IO) != 0) return; |
| 241 | const pid = std.os.linux.getpid(); | 232 | } else if (native_os == .linux) { |
| 242 | @atomicStore(Pid, &t.pid, @enumFromInt(pid), .monotonic); | 233 | const pid: posix.pid_t = p: { |
| 243 | break :p pid; | 234 | const cached_pid = @atomicLoad(Pid, &t.pid, .monotonic); |
| 244 | }; | 235 | if (cached_pid != .unknown) break :p @intFromEnum(cached_pid); |
| 245 | if (std.os.linux.tgkill(pid, @bitCast(signal_id), .IO) != 0) return; | 236 | const pid = std.os.linux.getpid(); |
| 246 | } else { | 237 | @atomicStore(Pid, &t.pid, @enumFromInt(pid), .monotonic); |
| 247 | return; | 238 | break :p pid; |
| 248 | } | | |
| 249 | | | |
| 250 | var timespec: posix.timespec = .{ | | |
| 251 | .sec = 0, | | |
| 252 | .nsec = @as(isize, 1) << @intCast(attempt_index), | | |
| 253 | }; | 239 | }; |
| 254 | if (native_os == .linux) { | 240 | if (std.os.linux.tgkill(pid, @bitCast(signal_id), .IO) != 0) return; |
| 255 | _ = std.os.linux.clock_nanosleep(posix.CLOCK.MONOTONIC, .{ .ABSTIME = false }, ×pec, ×pec); | 241 | } else { |
| 256 | } else { | 242 | return; |
| 257 | _ = posix.system.nanosleep(×pec, ×pec); | | |
| 258 | } | | |
| 259 | | | |
| 260 | switch (@atomicRmw(CancelStatus, &closure.cancel_status, .Xchg, .requested, .monotonic).unpack()) { | | |
| 261 | .requested => continue, // Retry needed in case other thread hasn't yet entered the syscall. | | |
| 262 | .none, .acknowledged => return, | | |
| 263 | .signal_id => |new_signal_id| signal_id = new_signal_id, | | |
| 264 | } | | |
| 265 | } | 243 | } |
| 266 | } | 244 | } |
| 267 | }; | 245 | }; |
| ... | @@ -303,7 +281,7 @@ pub fn init( | ... | @@ -303,7 +281,7 @@ pub fn init( |
| 303 | .mask = posix.sigemptyset(), | 281 | .mask = posix.sigemptyset(), |
| 304 | .flags = 0, | 282 | .flags = 0, |
| 305 | }; | 283 | }; |
| 306 | if (have_sig_io) posix.sigaction(.IO, &act, &t.old_sig_io); | 284 | if (!is_musl and have_sig_io) posix.sigaction(.IO, &act, &t.old_sig_io); |
| 307 | if (have_sig_pipe) posix.sigaction(.PIPE, &act, &t.old_sig_pipe); | 285 | if (have_sig_pipe) posix.sigaction(.PIPE, &act, &t.old_sig_pipe); |
| 308 | t.have_signal_handler = true; | 286 | t.have_signal_handler = true; |
| 309 | } | 287 | } |
| ... | @@ -341,7 +319,7 @@ pub fn deinit(t: *Threaded) void { | ... | @@ -341,7 +319,7 @@ pub fn deinit(t: *Threaded) void { |
| 341 | if (ws2_32.WSACleanup() != 0) recoverableOsBugDetected(); | 319 | if (ws2_32.WSACleanup() != 0) recoverableOsBugDetected(); |
| 342 | } | 320 | } |
| 343 | if (posix.Sigaction != void and t.have_signal_handler) { | 321 | if (posix.Sigaction != void and t.have_signal_handler) { |
| 344 | if (have_sig_io) posix.sigaction(.IO, &t.old_sig_io, null); | 322 | if (!is_musl and have_sig_io) posix.sigaction(.IO, &t.old_sig_io, null); |
| 345 | if (have_sig_pipe) posix.sigaction(.PIPE, &t.old_sig_pipe, null); | 323 | if (have_sig_pipe) posix.sigaction(.PIPE, &t.old_sig_pipe, null); |
| 346 | } | 324 | } |
| 347 | t.* = undefined; | 325 | t.* = undefined; |
Andrew Kelley <andrew@ziglang.org> 2025-11-29 08:43:57-08:00
lib