| ... | ... | @@ -37,7 +37,7 @@ cpu_count_error: ?std.Thread.CpuCountError, |
| 37 | 37 | /// available count, subtract this from either `async_limit` or |
| 38 | 38 | /// `concurrent_limit`. |
| 39 | 39 | busy_count: usize = 0, |
| 40 | | main_thread: Thread, |
| 40 | worker_threads: std.atomic.Value(?*Thread), |
| 41 | 41 | pid: Pid = .unknown, |
| 42 | 42 | robust_cancel: RobustCancel, |
| 43 | 43 | |
| ... | ... | @@ -153,107 +153,465 @@ pub const UseFchmodat2 = if (have_fchmodat2 and !have_fchmodat_flags) enum { |
| 153 | 153 | pub const default: UseFchmodat2 = .disabled; |
| 154 | 154 | }; |
| 155 | 155 | |
| 156 | | const Thread = struct { |
| 157 | | /// The value that needs to be passed to pthread_kill or tgkill in order to |
| 158 | | /// send a signal. |
| 159 | | signal_id: SignaleeId, |
| 160 | | current_closure: ?*Closure, |
| 161 | | /// Only populated if `current_closure != null`. Indicates the current cancel protection mode. |
| 162 | | cancel_protection: Io.CancelProtection, |
| 163 | | |
| 164 | | const SignaleeId = if (std.Thread.use_pthreads) std.c.pthread_t else std.Thread.Id; |
| 156 | const Runnable = struct { |
| 157 | node: std.SinglyLinkedList.Node, |
| 158 | startFn: *const fn (*Runnable, *Thread, *Threaded) void, |
| 159 | }; |
| 165 | 160 | |
| 166 | | threadlocal var current: ?*Thread = null; |
| 161 | const Group = struct { |
| 162 | ptr: *Io.Group, |
| 167 | 163 | |
| 168 | | fn getCurrent(t: *Threaded) *Thread { |
| 169 | | return current orelse return &t.main_thread; |
| 164 | /// Returns a correctly-typed pointer to the `Io.Group.token` field. |
| 165 | /// |
| 166 | /// The status indicates how many pending tasks are in the group, whether the group has been |
| 167 | /// canceled, and whether the group has been awaited. |
| 168 | /// |
| 169 | /// Note that the zero value of `Status` intentionally represents the initial group state (empty |
| 170 | /// with no awaiters). This is a requirement of `Io.Group`. |
| 171 | fn status(g: Group) *std.atomic.Value(Status) { |
| 172 | return @ptrCast(&g.ptr.token); |
| 173 | } |
| 174 | /// Returns a correctly-typed pointer to the `Io.Group.state` field. The double-pointer here is |
| 175 | /// intentional, because the `state` field itself stores a pointer, and this function returns a |
| 176 | /// pointer to that field. |
| 177 | /// |
| 178 | /// On completion of the whole group, if `status` indicates that there is an awaiter, the last |
| 179 | /// task must increment this `u32` and do a futex wake on it to signal that awaiter. |
| 180 | fn awaiter(g: Group) **std.atomic.Value(u32) { |
| 181 | return @ptrCast(&g.ptr.state); |
| 170 | 182 | } |
| 171 | 183 | |
| 172 | | fn checkCancel(thread: *Thread) error{Canceled}!void { |
| 173 | | const closure = thread.current_closure orelse return; |
| 184 | const Status = packed struct(usize) { |
| 185 | num_running: @Int(.unsigned, @bitSizeOf(usize) - 2), |
| 186 | have_awaiter: bool, |
| 187 | canceled: bool, |
| 188 | }; |
| 174 | 189 | |
| 175 | | switch (thread.cancel_protection) { |
| 176 | | .unblocked => {}, |
| 177 | | .blocked => return, |
| 190 | const Task = struct { |
| 191 | runnable: Runnable, |
| 192 | group: *Io.Group, |
| 193 | func: *const fn (*Io.Group, context: *const anyopaque) void, |
| 194 | context_alignment: Alignment, |
| 195 | alloc_len: usize, |
| 196 | |
| 197 | /// `Task.runnable.node` is `undefined` in the created `Task`. |
| 198 | fn create( |
| 199 | gpa: Allocator, |
| 200 | group: Group, |
| 201 | context: []const u8, |
| 202 | context_alignment: Alignment, |
| 203 | func: *const fn (*Io.Group, context: *const anyopaque) void, |
| 204 | ) Allocator.Error!*Task { |
| 205 | const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(Task); |
| 206 | const worst_case_context_offset = context_alignment.forward(@sizeOf(Task) + max_context_misalignment); |
| 207 | const alloc_len = worst_case_context_offset + context.len; |
| 208 | |
| 209 | const task: *Task = @ptrCast(@alignCast(try gpa.alignedAlloc(u8, .of(Task), alloc_len))); |
| 210 | errdefer comptime unreachable; |
| 211 | |
| 212 | task.* = .{ |
| 213 | .runnable = .{ |
| 214 | .node = undefined, |
| 215 | .startFn = &start, |
| 216 | }, |
| 217 | .group = group.ptr, |
| 218 | .func = func, |
| 219 | .context_alignment = context_alignment, |
| 220 | .alloc_len = alloc_len, |
| 221 | }; |
| 222 | @memcpy(task.contextPointer()[0..context.len], context); |
| 223 | return task; |
| 224 | } |
| 225 | |
| 226 | fn destroy(task: *Task, gpa: Allocator) void { |
| 227 | const base: [*]align(@alignOf(Task)) u8 = @ptrCast(task); |
| 228 | gpa.free(base[0..task.alloc_len]); |
| 229 | } |
| 230 | |
| 231 | fn contextPointer(task: *Task) [*]u8 { |
| 232 | const base: [*]u8 = @ptrCast(task); |
| 233 | const offset = task.context_alignment.forward(@intFromPtr(base) + @sizeOf(Task)) - @intFromPtr(base); |
| 234 | return base + offset; |
| 235 | } |
| 236 | |
| 237 | fn start(r: *Runnable, thread: *Thread, t: *Threaded) void { |
| 238 | const task: *Task = @fieldParentPtr("runnable", r); |
| 239 | const group: Group = .{ .ptr = task.group }; |
| 240 | |
| 241 | // This would be a simple store, but it's upgraded to an RMW so we can use `.acquire` to |
| 242 | // enforce the ordering between this and the `group.status().load` below. Paired with |
| 243 | // the `.release` rmw on `Thread.status` in `cancelThreads`, this creates a StoreLoad |
| 244 | // barrier which guarantees that when a group is canceled, either we see the cancelation |
| 245 | // in the group status, or the canceler sees our thread status so can directly notify us |
| 246 | // of the cancelation. |
| 247 | _ = thread.status.swap(.{ |
| 248 | .cancelation = .none, |
| 249 | .awaitable = .fromGroup(group.ptr), |
| 250 | }, .acquire); |
| 251 | if (group.status().load(.monotonic).canceled) { |
| 252 | thread.status.store(.{ |
| 253 | .cancelation = .canceling, |
| 254 | .awaitable = .fromGroup(group.ptr), |
| 255 | }, .monotonic); |
| 256 | } |
| 257 | |
| 258 | assertGroupResult(task.func(group.ptr, task.contextPointer())); |
| 259 | |
| 260 | thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic); |
| 261 | const old_status = group.status().fetchSub(.{ |
| 262 | .num_running = 1, |
| 263 | .have_awaiter = false, |
| 264 | .canceled = false, |
| 265 | }, .acq_rel); // acquire `group.awaiter()`, release task results |
| 266 | assert(old_status.num_running > 0); |
| 267 | if (old_status.have_awaiter and old_status.num_running == 1) { |
| 268 | const to_signal = group.awaiter().*; |
| 269 | // `awaiter` should only be modified by us. For another thread to see `num_running` |
| 270 | // drop to 0 after this point would indicate that another task started up, meaning |
| 271 | // `async`/`cancel` was racing with awaited group completion. |
| 272 | group.awaiter().* = undefined; |
| 273 | _ = to_signal.fetchAdd(1, .release); // release results |
| 274 | Thread.futexWake(&to_signal.raw, 1); |
| 275 | } |
| 276 | |
| 277 | // Task completed. Self-destruct sequence initiated. |
| 278 | task.destroy(t.allocator); |
| 178 | 279 | } |
| 280 | }; |
| 179 | 281 | |
| 180 | | switch (@cmpxchgStrong( |
| 181 | | CancelStatus, |
| 182 | | &closure.cancel_status, |
| 183 | | .requested, |
| 184 | | .acknowledged, |
| 185 | | .acq_rel, |
| 186 | | .acquire, |
| 187 | | ) orelse return error.Canceled) { |
| 188 | | .requested => unreachable, |
| 189 | | .acknowledged => unreachable, |
| 190 | | .none, _ => {}, |
| 282 | /// Assumes the caller has already atomically updated the group status to indicate cancelation, |
| 283 | /// and notifies any already-running threads of this cancelation. |
| 284 | fn cancelThreads(g: Group, t: *Threaded) bool { |
| 285 | var any_blocked = false; |
| 286 | var it = t.worker_threads.load(.acquire); // acquire `Thread` values |
| 287 | while (it) |thread| : (it = thread.next) { |
| 288 | // This non-mutating RMW exists for ordering reasons: see comment in `Group.Task.start` for reasons. |
| 289 | _ = thread.status.fetchOr(.{ .cancelation = @enumFromInt(0), .awaitable = .null }, .release); |
| 290 | if (thread.cancelAwaitable(.fromGroup(g.ptr))) any_blocked = true; |
| 191 | 291 | } |
| 292 | return any_blocked; |
| 192 | 293 | } |
| 193 | 294 | |
| 194 | | fn beginSyscall(thread: *Thread) error{Canceled}!void { |
| 195 | | const closure = thread.current_closure orelse return; |
| 295 | /// Uses `Thread.signalCanceledSyscall` to signal any threads which are still blocked in a |
| 296 | /// syscall for this group and have not observed a cancelation request yet. Returns `true` if |
| 297 | /// more signals may be necessary, in which case the caller must call this again after a delay. |
| 298 | fn signalAllCanceledSyscalls(g: Group, t: *Threaded) bool { |
| 299 | var any_signaled = false; |
| 300 | var it = t.worker_threads.load(.acquire); // acquire `Thread` values |
| 301 | while (it) |thread| : (it = thread.next) { |
| 302 | if (thread.signalCanceledSyscall(t, .fromGroup(g.ptr))) any_signaled = true; |
| 303 | } |
| 304 | return any_signaled; |
| 305 | } |
| 196 | 306 | |
| 197 | | switch (thread.cancel_protection) { |
| 198 | | .unblocked => {}, |
| 199 | | .blocked => return, |
| 307 | /// The caller has canceled `g`. Inform any threads working on that group of the cancelation if |
| 308 | /// necessary, and wait for `g` to finish (indicated by `num_completed` being incremented from 0 |
| 309 | /// to 1), while sending regular signals to threads if necessary for them to unblock from any |
| 310 | /// cancelable syscalls. |
| 311 | /// |
| 312 | /// `skip_signals` means it is already known that no threads are currently working on the group |
| 313 | /// so no notifications or signals are necessary. |
| 314 | fn waitForCancelWithSignaling( |
| 315 | g: Group, |
| 316 | t: *Threaded, |
| 317 | num_completed: *std.atomic.Value(u32), |
| 318 | skip_signals: bool, |
| 319 | ) void { |
| 320 | var need_signal: bool = !skip_signals and g.cancelThreads(t); |
| 321 | var timeout_ns: u64 = 1 << 10; |
| 322 | while (true) { |
| 323 | need_signal = need_signal and g.signalAllCanceledSyscalls(t) and t.robust_cancel == .enabled; |
| 324 | Thread.futexWaitTimed( |
| 325 | null, |
| 326 | &num_completed.raw, |
| 327 | 0, |
| 328 | if (need_signal) timeout_ns else null, |
| 329 | ) catch |err| switch (err) { |
| 330 | error.Canceled => unreachable, |
| 331 | }; |
| 332 | switch (num_completed.load(.acquire)) { // acquire task results |
| 333 | 0 => {}, |
| 334 | 1 => break, |
| 335 | else => unreachable, |
| 336 | } |
| 337 | timeout_ns <<|= 1; |
| 200 | 338 | } |
| 339 | } |
| 340 | }; |
| 201 | 341 | |
| 202 | | switch (@cmpxchgStrong( |
| 203 | | CancelStatus, |
| 204 | | &closure.cancel_status, |
| 205 | | .none, |
| 206 | | .fromSignaleeId(thread.signal_id), |
| 207 | | .acq_rel, |
| 208 | | .acquire, |
| 209 | | ) orelse return) { |
| 210 | | .none => unreachable, |
| 211 | | .requested => { |
| 212 | | @atomicStore(CancelStatus, &closure.cancel_status, .acknowledged, .release); |
| 213 | | return error.Canceled; |
| 342 | /// Trailing data: |
| 343 | /// 1. context |
| 344 | /// 2. result |
| 345 | const Future = struct { |
| 346 | runnable: Runnable, |
| 347 | func: *const fn (context: *const anyopaque, result: *anyopaque) void, |
| 348 | status: std.atomic.Value(Status), |
| 349 | /// On completion, increment this `u32` and do a futex wake on it. |
| 350 | awaiter: *std.atomic.Value(u32), |
| 351 | context_alignment: Alignment, |
| 352 | result_offset: usize, |
| 353 | alloc_len: usize, |
| 354 | |
| 355 | const Status = packed struct(usize) { |
| 356 | /// The values of this enum are chosen so that await/cancel can just OR with 0b01 and 0b11 |
| 357 | /// respectively. That *does* clobber `.done`, but that's actually fine, because if the tag |
| 358 | /// is `.done` then only the awaiter is referencing this `Future` anyway. |
| 359 | tag: enum(u2) { |
| 360 | /// The future is queued or running (depending on whether `thread` is set). |
| 361 | pending = 0b00, |
| 362 | /// Like `pending`, but the future is being awaited. `Future.awaiter` is populated. |
| 363 | pending_awaited = 0b01, |
| 364 | /// Like `pending`, but the future is being canceled. `Future.awaiter` is populated. |
| 365 | pending_canceled = 0b11, |
| 366 | /// The future has already completed. `thread` is `null`. |
| 367 | done = 0b10, |
| 368 | }, |
| 369 | /// When the future begins execution, this is atomically updated from `null` to the thread running the |
| 370 | /// `Future`, so that cancelation knows which thread to cancel. |
| 371 | thread: Thread.PackedPtr, |
| 372 | }; |
| 373 | |
| 374 | /// `Future.runnable.node` is `undefined` in the created `Future`. |
| 375 | fn create( |
| 376 | gpa: Allocator, |
| 377 | result_len: usize, |
| 378 | result_alignment: Alignment, |
| 379 | context: []const u8, |
| 380 | context_alignment: Alignment, |
| 381 | func: *const fn (context: *const anyopaque, result: *anyopaque) void, |
| 382 | ) Allocator.Error!*Future { |
| 383 | const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(Future); |
| 384 | const worst_case_context_offset = context_alignment.forward(@sizeOf(Future) + max_context_misalignment); |
| 385 | const worst_case_result_offset = result_alignment.forward(worst_case_context_offset + context.len); |
| 386 | const alloc_len = worst_case_result_offset + result_len; |
| 387 | |
| 388 | const future: *Future = @ptrCast(@alignCast(try gpa.alignedAlloc(u8, .of(Future), alloc_len))); |
| 389 | errdefer comptime unreachable; |
| 390 | |
| 391 | const actual_context_addr = context_alignment.forward(@intFromPtr(future) + @sizeOf(Future)); |
| 392 | const actual_result_addr = result_alignment.forward(actual_context_addr + context.len); |
| 393 | const actual_result_offset = actual_result_addr - @intFromPtr(future); |
| 394 | future.* = .{ |
| 395 | .runnable = .{ |
| 396 | .node = undefined, |
| 397 | .startFn = &start, |
| 214 | 398 | }, |
| 215 | | .acknowledged => return, |
| 216 | | _ => unreachable, |
| 217 | | } |
| 399 | .func = func, |
| 400 | .status = .init(.{ |
| 401 | .tag = .pending, |
| 402 | .thread = .null, |
| 403 | }), |
| 404 | .awaiter = undefined, |
| 405 | .context_alignment = context_alignment, |
| 406 | .result_offset = actual_result_offset, |
| 407 | .alloc_len = alloc_len, |
| 408 | }; |
| 409 | @memcpy(future.contextPointer()[0..context.len], context); |
| 410 | return future; |
| 218 | 411 | } |
| 219 | 412 | |
| 220 | | fn endSyscall(thread: *Thread) void { |
| 221 | | const closure = thread.current_closure orelse return; |
| 413 | fn destroy(future: *Future, gpa: Allocator) void { |
| 414 | const base: [*]align(@alignOf(Future)) u8 = @ptrCast(future); |
| 415 | gpa.free(base[0..future.alloc_len]); |
| 416 | } |
| 222 | 417 | |
| 223 | | switch (thread.cancel_protection) { |
| 224 | | .unblocked => {}, |
| 225 | | .blocked => return, |
| 226 | | } |
| 418 | fn resultPointer(future: *Future) [*]u8 { |
| 419 | const base: [*]u8 = @ptrCast(future); |
| 420 | return base + future.result_offset; |
| 421 | } |
| 227 | 422 | |
| 228 | | _ = @cmpxchgStrong( |
| 229 | | CancelStatus, |
| 230 | | &closure.cancel_status, |
| 231 | | .fromSignaleeId(thread.signal_id), |
| 232 | | .none, |
| 233 | | .acq_rel, |
| 234 | | .acquire, |
| 235 | | ) orelse return; |
| 423 | fn contextPointer(future: *Future) [*]u8 { |
| 424 | const base: [*]u8 = @ptrCast(future); |
| 425 | const context_offset = future.context_alignment.forward(@intFromPtr(future) + @sizeOf(Future)) - @intFromPtr(future); |
| 426 | return base + context_offset; |
| 236 | 427 | } |
| 237 | 428 | |
| 238 | | fn endSyscallErrnoBug(thread: *Thread, err: posix.E) Io.UnexpectedError { |
| 239 | | @branchHint(.cold); |
| 240 | | thread.endSyscall(); |
| 241 | | return errnoBug(err); |
| 429 | fn start(r: *Runnable, thread: *Thread, t: *Threaded) void { |
| 430 | _ = t; |
| 431 | const future: *Future = @fieldParentPtr("runnable", r); |
| 432 | |
| 433 | thread.status.store(.{ |
| 434 | .cancelation = .none, |
| 435 | .awaitable = .fromFuture(future), |
| 436 | }, .monotonic); |
| 437 | { |
| 438 | const old_status = future.status.fetchOr(.{ |
| 439 | .tag = .pending, |
| 440 | .thread = .pack(thread), |
| 441 | }, .release); |
| 442 | assert(old_status.thread == .null); |
| 443 | switch (old_status.tag) { |
| 444 | .pending, .pending_awaited => {}, |
| 445 | .pending_canceled => thread.status.store(.{ |
| 446 | .cancelation = .canceling, |
| 447 | .awaitable = .fromFuture(future), |
| 448 | }, .monotonic), |
| 449 | .done => unreachable, |
| 450 | } |
| 451 | } |
| 452 | |
| 453 | future.func(future.contextPointer(), future.resultPointer()); |
| 454 | |
| 455 | thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic); |
| 456 | const old_status = future.status.swap(.{ |
| 457 | .tag = .done, |
| 458 | .thread = .null, |
| 459 | }, .acq_rel); // acquire `future.awaiter`, release results |
| 460 | switch (old_status.tag) { |
| 461 | .pending => {}, |
| 462 | .pending_awaited, .pending_canceled => { |
| 463 | const to_signal = future.awaiter; |
| 464 | _ = to_signal.fetchAdd(1, .release); // release results |
| 465 | Thread.futexWake(&to_signal.raw, 1); |
| 466 | }, |
| 467 | .done => unreachable, |
| 468 | } |
| 469 | } |
| 470 | |
| 471 | /// The caller has canceled `future`. `thread` is the thread currently running that future. |
| 472 | /// Inform `thread` of the cancelation if necessary, and wait for `future` to finish (indicated |
| 473 | /// by `num_completed` being incremented from 0 to 1), while sending regular signals to `thread` |
| 474 | /// if necessary for it to unblock from a cancelable syscall. |
| 475 | fn waitForCancelWithSignaling( |
| 476 | future: *Future, |
| 477 | t: *Threaded, |
| 478 | num_completed: *std.atomic.Value(u32), |
| 479 | thread: ?*Thread, |
| 480 | ) void { |
| 481 | var need_signal: bool = thread != null and thread.?.cancelAwaitable(.fromFuture(future)); |
| 482 | var timeout_ns: u64 = 1 << 10; |
| 483 | while (true) { |
| 484 | need_signal = need_signal and thread.?.signalCanceledSyscall(t, .fromFuture(future)) and t.robust_cancel == .enabled; |
| 485 | Thread.futexWaitTimed( |
| 486 | null, |
| 487 | &num_completed.raw, |
| 488 | 0, |
| 489 | if (need_signal) timeout_ns else null, |
| 490 | ) catch |err| switch (err) { |
| 491 | error.Canceled => unreachable, |
| 492 | }; |
| 493 | switch (num_completed.load(.acquire)) { // acquire task results |
| 494 | 0 => {}, |
| 495 | 1 => break, |
| 496 | else => unreachable, |
| 497 | } |
| 498 | timeout_ns <<|= 1; |
| 499 | } |
| 242 | 500 | } |
| 501 | }; |
| 243 | 502 | |
| 244 | | fn endSyscallUnexpectedErrno(thread: *Thread, err: posix.E) Io.UnexpectedError { |
| 245 | | @branchHint(.cold); |
| 246 | | thread.endSyscall(); |
| 247 | | return posix.unexpectedErrno(err); |
| 503 | /// A sequence of (ptr_bit_width - 3) bits which uniquely identifies a group or future. The bits are |
| 504 | /// the MSBs of the `*Io.Group` or `*Future`. These things do not necessarily have 3 zero bits at |
| 505 | /// the end (they are pointer-aligned, so on 32-bit targets only have 2), but because they both have |
| 506 | /// a *size* of at least 8 bytes, no two groups/futures in memory at the same time will have the |
| 507 | /// same value for all of these bits. In other words, given a group/future pointer, the next group |
| 508 | /// or future must be at least 8 bytes later, so its address will have a different value for one of |
| 509 | /// the top (ptr_bit_width - 3) bits. |
| 510 | const AwaitableId = enum(@Int(.unsigned, @bitSizeOf(usize) - 3)) { |
| 511 | comptime { |
| 512 | assert(@sizeOf(Future) >= 8); |
| 513 | assert(@sizeOf(Io.Group) >= 8); |
| 514 | } |
| 515 | null = 0, |
| 516 | all_ones = std.math.maxInt(@Int(.unsigned, @bitSizeOf(usize) - 3)), |
| 517 | _, |
| 518 | const Split = packed struct(usize) { low: u3, high: AwaitableId }; |
| 519 | fn fromGroup(g: *Io.Group) AwaitableId { |
| 520 | const split: Split = @bitCast(@intFromPtr(g)); |
| 521 | return split.high; |
| 248 | 522 | } |
| 523 | fn fromFuture(f: *Future) AwaitableId { |
| 524 | const split: Split = @bitCast(@intFromPtr(f)); |
| 525 | return split.high; |
| 526 | } |
| 527 | }; |
| 249 | 528 | |
| 250 | | /// inline to make error return traces slightly shallower. |
| 251 | | inline fn endSyscallError(thread: *Thread, err: anytype) @TypeOf(err) { |
| 252 | | thread.endSyscall(); |
| 253 | | return err; |
| 529 | const Thread = struct { |
| 530 | next: ?*Thread, |
| 531 | /// The value that needs to be passed to pthread_kill or tgkill in order to |
| 532 | /// send a signal. |
| 533 | signalee_id: SignaleeId, |
| 534 | |
| 535 | status: std.atomic.Value(Status), |
| 536 | |
| 537 | cancel_protection: Io.CancelProtection, |
| 538 | |
| 539 | const Status = packed struct(usize) { |
| 540 | /// The specific values of these enum fields are chosen to simplify the implementation of |
| 541 | /// the transformations we need to apply to this state. |
| 542 | cancelation: enum(u3) { |
| 543 | /// The thread has not yet been canceled, and is not in a cancelable operation. |
| 544 | /// To request cancelation, just set the status to `.canceling`. |
| 545 | none = 0b000, |
| 546 | |
| 547 | /// The thread is parked in a cancelable futex wait or sleep. |
| 548 | /// Only applicable on Windows, NetBSD, and Illumos. |
| 549 | /// To request cancelation, set the status to `.canceling` and unpark the thread. |
| 550 | /// To unpark for another reason (futex wake), set the status to `.none` and unpark the thread. |
| 551 | parked = 0b001, |
| 552 | |
| 553 | /// The thread is blocked in a cancelable system call. |
| 554 | /// To request cancelation, set the status to `.blocked_canceling` and repeatedly interrupt the system call until the status changes. |
| 555 | blocked = 0b011, |
| 556 | |
| 557 | /// Windows-only: the thread is blocked on a DNS query. |
| 558 | /// To request cancelation, set the status to `.canceling` and call `DnsCancelQuery`. |
| 559 | blocked_windows_dns = 0b010, |
| 560 | |
| 561 | /// The thread has an outstanding cancelation request but is not in a cancelable operation. |
| 562 | /// When it acknowledges the cancelation, it will set the status to `.canceled`. |
| 563 | canceling = 0b110, |
| 564 | |
| 565 | /// The thread has received and acknowledged a cancelation request. |
| 566 | /// If `recancel` is called, the status will revert to `.canceling`, but otherwise, the status |
| 567 | /// will not change for the remainder of this task's execution. |
| 568 | canceled = 0b111, |
| 569 | |
| 570 | /// The thread is blocked in a cancelable system call, and is being canceled. The thread which triggered the cancelation will send signals to this thread |
| 571 | /// until its status changes. |
| 572 | blocked_canceling = 0b101, |
| 573 | }, |
| 574 | |
| 575 | /// We cannot turn this value back into a pointer. Instead, it exists so that a task can be |
| 576 | /// canceled by a cmpxchg on thread status: if it is running the task we want to cancel, |
| 577 | /// then update the `cancelation` field. |
| 578 | awaitable: AwaitableId, |
| 579 | }; |
| 580 | |
| 581 | const SignaleeId = if (std.Thread.use_pthreads) std.c.pthread_t else std.Thread.Id; |
| 582 | |
| 583 | threadlocal var current: ?*Thread = null; |
| 584 | |
| 585 | /// The thread is neither in a syscall nor entering one, but we want to check for cancelation |
| 586 | /// anyway. If there is a pending cancel request, acknowledge it and return `error.Canceled`. |
| 587 | fn checkCancel() Io.Cancelable!void { |
| 588 | const thread = Thread.current orelse return; |
| 589 | switch (thread.cancel_protection) { |
| 590 | .blocked => return, |
| 591 | .unblocked => {}, |
| 592 | } |
| 593 | // Here, unlike `Syscall.checkCancel`, it's not particularly likely that we're canceled, so |
| 594 | // it seems preferable to do a cheap atomic load and, in the unlikely case, a separate store |
| 595 | // to acknowledge. Besides, the state transitions we need here can't be done with one atomic |
| 596 | // OR/AND/XOR on `Status.cancelation`, so we don't actually have any other option. |
| 597 | const status = thread.status.load(.monotonic); |
| 598 | switch (status.cancelation) { |
| 599 | .parked => unreachable, |
| 600 | .blocked => unreachable, |
| 601 | .blocked_windows_dns => unreachable, |
| 602 | .blocked_canceling => unreachable, |
| 603 | .none, .canceled => {}, |
| 604 | .canceling => { |
| 605 | thread.status.store(.{ |
| 606 | .cancelation = .canceled, |
| 607 | .awaitable = status.awaitable, |
| 608 | }, .monotonic); |
| 609 | return error.Canceled; |
| 610 | }, |
| 611 | } |
| 254 | 612 | } |
| 255 | 613 | |
| 256 | | fn currentSignalId() SignaleeId { |
| 614 | fn currentSignaleeId() SignaleeId { |
| 257 | 615 | return if (std.Thread.use_pthreads) std.c.pthread_self() else std.Thread.getCurrentId(); |
| 258 | 616 | } |
| 259 | 617 | |
| ... | ... | @@ -262,10 +620,7 @@ const Thread = struct { |
| 262 | 620 | } |
| 263 | 621 | |
| 264 | 622 | fn futexWait(thread: *Thread, ptr: *const u32, expect: u32) Io.Cancelable!void { |
| 265 | | return Thread.futexWaitTimed(thread, ptr, expect, null) catch |err| switch (err) { |
| 266 | | error.Canceled => return error.Canceled, |
| 267 | | error.Timeout => unreachable, |
| 268 | | }; |
| 623 | return Thread.futexWaitTimed(thread, ptr, expect, null); |
| 269 | 624 | } |
| 270 | 625 | |
| 271 | 626 | fn futexWaitTimed(thread: ?*Thread, ptr: *const u32, expect: u32, timeout_ns: ?u64) Io.Cancelable!void { |
| ... | ... | @@ -543,123 +898,200 @@ const Thread = struct { |
| 543 | 898 | }, |
| 544 | 899 | } |
| 545 | 900 | } |
| 546 | | }; |
| 547 | | |
| 548 | | const max_iovecs_len = 8; |
| 549 | | const splat_buffer_size = 64; |
| 550 | | |
| 551 | | comptime { |
| 552 | | if (@TypeOf(posix.IOV_MAX) != void) assert(max_iovecs_len <= posix.IOV_MAX); |
| 553 | | } |
| 554 | | |
| 555 | | const CancelStatus = enum(usize) { |
| 556 | | /// Cancellation has neither been requested, nor checked. The async |
| 557 | | /// operation will check status before entering a blocking syscall. |
| 558 | | /// This is also the status used for uninteruptible tasks. |
| 559 | | none = 0, |
| 560 | | /// Cancellation has been requested and the status will be checked before |
| 561 | | /// entering a blocking syscall. |
| 562 | | requested = std.math.maxInt(usize) - 1, |
| 563 | | /// Cancellation has been acknowledged and is in progress. Signals should |
| 564 | | /// not be sent. |
| 565 | | acknowledged = std.math.maxInt(usize), |
| 566 | | /// Stores a `Thread.SignaleeId` and indicates that sending a signal to this thread |
| 567 | | /// is needed in order to cancel. This state is set before going into |
| 568 | | /// a blocking operation that needs to get unblocked via signal. |
| 569 | | _, |
| 570 | 901 | |
| 571 | | const Unpacked = union(enum) { |
| 572 | | none, |
| 573 | | requested, |
| 574 | | acknowledged, |
| 575 | | signal_id: Thread.SignaleeId, |
| 576 | | }; |
| 902 | /// Cancels `thread` if it is working on `awaitable`. |
| 903 | /// |
| 904 | /// It is possible that `thread` gets canceled by this function, but is blocked in a syscall. In |
| 905 | /// that case, the thread may need to be sent a signal to interrupt the call. This function will |
| 906 | /// return `true` to indicate this, in which case the caller must call `signalCanceledSyscall`. |
| 907 | fn cancelAwaitable(thread: *Thread, awaitable: AwaitableId) bool { |
| 908 | var status = thread.status.load(.monotonic); |
| 909 | while (true) { |
| 910 | if (status.awaitable != awaitable) return false; // thread is working on something else |
| 911 | status = switch (status.cancelation) { |
| 912 | .none => thread.status.cmpxchgWeak( |
| 913 | .{ .cancelation = .none, .awaitable = awaitable }, |
| 914 | .{ .cancelation = .canceling, .awaitable = awaitable }, |
| 915 | .monotonic, |
| 916 | .monotonic, |
| 917 | ) orelse return false, |
| 918 | |
| 919 | .parked => thread.status.cmpxchgWeak( |
| 920 | .{ .cancelation = .parked, .awaitable = awaitable }, |
| 921 | .{ .cancelation = .canceling, .awaitable = awaitable }, |
| 922 | .monotonic, |
| 923 | .monotonic, |
| 924 | ) orelse { |
| 925 | if (true) @panic("MLUGG TODO: unpark thread"); |
| 926 | return false; |
| 927 | }, |
| 577 | 928 | |
| 578 | | fn unpack(cs: CancelStatus) Unpacked { |
| 579 | | return switch (cs) { |
| 580 | | .none => .none, |
| 581 | | .requested => .requested, |
| 582 | | .acknowledged => .acknowledged, |
| 583 | | _ => |signal_id| .{ |
| 584 | | .signal_id = if (std.Thread.use_pthreads) |
| 585 | | @ptrFromInt(@intFromEnum(signal_id)) |
| 586 | | else |
| 587 | | @truncate(@intFromEnum(signal_id)), |
| 588 | | }, |
| 589 | | }; |
| 590 | | } |
| 929 | .blocked => thread.status.cmpxchgWeak( |
| 930 | .{ .cancelation = .blocked, .awaitable = awaitable }, |
| 931 | .{ .cancelation = .blocked_canceling, .awaitable = awaitable }, |
| 932 | .monotonic, |
| 933 | .monotonic, |
| 934 | ) orelse return true, |
| 935 | |
| 936 | .blocked_windows_dns => thread.status.cmpxchgWeak( |
| 937 | .{ .cancelation = .blocked_windows_dns, .awaitable = awaitable }, |
| 938 | .{ .cancelation = .canceling, .awaitable = awaitable }, |
| 939 | .monotonic, |
| 940 | .monotonic, |
| 941 | ) orelse return false, |
| 942 | |
| 943 | .canceling, .canceled => { |
| 944 | // This can happen when the task start raced with the cancelation, so the thread |
| 945 | // saw the cancelation on the future/group *and* we are trying to signal the |
| 946 | // thread here. |
| 947 | return false; |
| 948 | }, |
| 591 | 949 | |
| 592 | | fn fromSignaleeId(signal_id: Thread.SignaleeId) CancelStatus { |
| 593 | | return if (std.Thread.use_pthreads) |
| 594 | | @enumFromInt(@intFromPtr(signal_id)) |
| 595 | | else |
| 596 | | @enumFromInt(signal_id); |
| 950 | .blocked_canceling => unreachable, |
| 951 | }; |
| 952 | } |
| 597 | 953 | } |
| 598 | | }; |
| 599 | 954 | |
| 600 | | const Closure = struct { |
| 601 | | start: Start, |
| 602 | | node: std.SinglyLinkedList.Node = .{}, |
| 603 | | cancel_status: CancelStatus, |
| 604 | | |
| 605 | | const Start = *const fn (*Closure, *Threaded) void; |
| 955 | /// Sends a signal to `thread` if it is still blocked in a syscall (i.e. has not yet observed |
| 956 | /// the cancelation request from `cancelAwaitable`). |
| 957 | /// |
| 958 | /// Unfortunately, the signal could arrive before the syscall actually starts, so the interrupt |
| 959 | /// is missed. To handle this, we may need to send multiple signals. As such, if this function |
| 960 | /// returns `true`, then it should be called again after a short delay to send another signal if |
| 961 | /// the thread is still blocked. For the implementation, `Future.waitForCancelWithSignaling` and |
| 962 | /// `Group.waitForCancelWithSignaling`: they use exponential backoff starting at a 1us delay and |
| 963 | /// doubling each call. In practice, it is rare to send more than one signal. |
| 964 | fn signalCanceledSyscall(thread: *Thread, t: *Threaded, awaitable: AwaitableId) bool { |
| 965 | const bad_status: Status = .{ .cancelation = .blocked_canceling, .awaitable = awaitable }; |
| 966 | if (thread.status.load(.monotonic) != bad_status) return false; |
| 967 | |
| 968 | // The thread ID can be read non-atomically because it never changes and was released by the |
| 969 | // store that made `thread` available to us. |
| 970 | const signalee_id = thread.signalee_id; |
| 971 | |
| 972 | if (std.Thread.use_pthreads) { |
| 973 | if (std.c.pthread_kill(signalee_id, .IO) != 0) return false; |
| 974 | } else if (native_os == .linux) { |
| 975 | const pid: posix.pid_t = pid: { |
| 976 | const cached_pid = @atomicLoad(Pid, &t.pid, .monotonic); |
| 977 | if (cached_pid != .unknown) break :pid @intFromEnum(cached_pid); |
| 978 | const pid = std.os.linux.getpid(); |
| 979 | @atomicStore(Pid, &t.pid, @enumFromInt(pid), .monotonic); |
| 980 | break :pid pid; |
| 981 | }; |
| 982 | if (std.os.linux.tgkill(pid, @bitCast(signalee_id), .IO) != 0) return false; |
| 983 | } else { |
| 984 | @compileError("MLUGG TODO"); |
| 985 | } |
| 606 | 986 | |
| 607 | | fn requestCancel(closure: *Closure, t: *Threaded) void { |
| 608 | | var signal_id = switch (@atomicRmw(CancelStatus, &closure.cancel_status, .Xchg, .requested, .monotonic).unpack()) { |
| 609 | | .none, .acknowledged, .requested => return, |
| 610 | | .signal_id => |signal_id| signal_id, |
| 611 | | }; |
| 612 | | // The task will enter a blocking syscall before checking for cancellation again. |
| 613 | | // We can send a signal to interrupt the syscall, but if it arrives before |
| 614 | | // the syscall instruction, it will be missed. Therefore, this code tries |
| 615 | | // again until the cancellation request is acknowledged. |
| 616 | | |
| 617 | | // 1 << 10 ns is about 1 microsecond, approximately syscall overhead. |
| 618 | | // 1 << 20 ns is about 1 millisecond. |
| 619 | | // 1 << 30 ns is about 1 second. |
| 620 | | // |
| 621 | | // On a heavily loaded Linux 6.17.5, I observed a maximum of 20 |
| 622 | | // attempts not acknowledged before the timeout (including exponential |
| 623 | | // backoff) was sufficient, despite the heavy load. |
| 624 | | const max_attempts = 22; |
| 625 | | |
| 626 | | for (0..max_attempts) |attempt_index| { |
| 627 | | if (std.Thread.use_pthreads) { |
| 628 | | if (std.c.pthread_kill(signal_id, .IO) != 0) return; |
| 629 | | } else if (native_os == .linux) { |
| 630 | | const pid: posix.pid_t = p: { |
| 631 | | const cached_pid = @atomicLoad(Pid, &t.pid, .monotonic); |
| 632 | | if (cached_pid != .unknown) break :p @intFromEnum(cached_pid); |
| 633 | | const pid = std.os.linux.getpid(); |
| 634 | | @atomicStore(Pid, &t.pid, @enumFromInt(pid), .monotonic); |
| 635 | | break :p pid; |
| 636 | | }; |
| 637 | | if (std.os.linux.tgkill(pid, @bitCast(signal_id), .IO) != 0) return; |
| 638 | | } else { |
| 639 | | return; |
| 640 | | } |
| 987 | return true; |
| 988 | } |
| 641 | 989 | |
| 642 | | if (t.robust_cancel != .enabled) return; |
| 990 | /// Like a `*Thread`, but 2 bits smaller than a pointer (because the LSBs are always 0 due to |
| 991 | /// alignment) so that those two bits can be used in a `packed struct`. |
| 992 | const PackedPtr = enum(@Int(.unsigned, @bitSizeOf(usize) - 2)) { |
| 993 | null = 0, |
| 994 | all_ones = std.math.maxInt(@Int(.unsigned, @bitSizeOf(usize) - 2)), |
| 995 | _, |
| 643 | 996 | |
| 644 | | var timespec: posix.timespec = .{ |
| 645 | | .sec = 0, |
| 646 | | .nsec = @as(isize, 1) << @intCast(attempt_index), |
| 647 | | }; |
| 648 | | if (native_os == .linux) { |
| 649 | | _ = std.os.linux.clock_nanosleep(posix.CLOCK.MONOTONIC, .{ .ABSTIME = false }, ×pec, ×pec); |
| 650 | | } else { |
| 651 | | _ = posix.system.nanosleep(×pec, ×pec); |
| 652 | | } |
| 997 | const Split = packed struct(usize) { low: u2, high: PackedPtr }; |
| 998 | fn pack(ptr: *Thread) PackedPtr { |
| 999 | const split: Split = @bitCast(@intFromPtr(ptr)); |
| 1000 | assert(split.low == 0); |
| 1001 | return split.high; |
| 1002 | } |
| 1003 | fn unpack(ptr: PackedPtr) ?*Thread { |
| 1004 | const split: Split = .{ .low = 0, .high = ptr }; |
| 1005 | return @ptrFromInt(@as(usize, @bitCast(split))); |
| 1006 | } |
| 1007 | }; |
| 1008 | }; |
| 653 | 1009 | |
| 654 | | switch (@atomicRmw(CancelStatus, &closure.cancel_status, .Xchg, .requested, .monotonic).unpack()) { |
| 655 | | .requested => continue, // Retry needed in case other thread hasn't yet entered the syscall. |
| 656 | | .none, .acknowledged => return, |
| 657 | | .signal_id => |new_signal_id| signal_id = new_signal_id, |
| 658 | | } |
| 1010 | const Syscall = struct { |
| 1011 | thread: ?*Thread, |
| 1012 | /// Marks entry to a syscall region. This should be tightly scoped around the actual syscall |
| 1013 | /// to minimize races. The syscall must be marked as "finished" by `checkCancel`, `finish`, |
| 1014 | /// or one of the wrappers of `finish`. |
| 1015 | fn start() Io.Cancelable!Syscall { |
| 1016 | const thread = Thread.current orelse return .{ .thread = null }; |
| 1017 | switch (thread.cancel_protection) { |
| 1018 | .blocked => return .{ .thread = null }, |
| 1019 | .unblocked => {}, |
| 1020 | } |
| 1021 | switch (thread.status.fetchOr(.{ |
| 1022 | .cancelation = @enumFromInt(0b011), |
| 1023 | .awaitable = .null, |
| 1024 | }, .monotonic).cancelation) { |
| 1025 | .parked => unreachable, |
| 1026 | .blocked => unreachable, |
| 1027 | .blocked_windows_dns => unreachable, |
| 1028 | .blocked_canceling => unreachable, |
| 1029 | .none => return .{ .thread = thread }, // new status is `.blocked` |
| 1030 | .canceling => return error.Canceled, // new status is `.canceled` |
| 1031 | .canceled => return .{ .thread = null }, // new status is `.canceled` (unchanged) |
| 1032 | } |
| 1033 | } |
| 1034 | /// Checks whether this syscall has been canceled. This should be called when a syscall is |
| 1035 | /// interrupted through a mechanism which may indicate cancelation, or may be spurious. If |
| 1036 | /// the syscall was canceled, it is finished and `error.Canceled` is returned. Otherwise, |
| 1037 | /// the syscall is not marked finished, and the caller should retry. |
| 1038 | fn checkCancel(s: Syscall) Io.Cancelable!void { |
| 1039 | const thread = s.thread orelse return; |
| 1040 | switch (thread.status.fetchOr(.{ |
| 1041 | .cancelation = @enumFromInt(0b010), |
| 1042 | .awaitable = .null, |
| 1043 | }, .monotonic).cancelation) { |
| 1044 | .none => unreachable, |
| 1045 | .parked => unreachable, |
| 1046 | .blocked_windows_dns => unreachable, |
| 1047 | .canceling => unreachable, |
| 1048 | .canceled => unreachable, |
| 1049 | .blocked => {}, // new status is `.blocked` (unchanged) |
| 1050 | .blocked_canceling => return error.Canceled, // new status is `.canceled` |
| 1051 | } |
| 1052 | } |
| 1053 | /// Marks this syscall as finished. |
| 1054 | fn finish(s: Syscall) void { |
| 1055 | const thread = s.thread orelse return; |
| 1056 | switch (thread.status.fetchXor(.{ |
| 1057 | .cancelation = @enumFromInt(0b011), |
| 1058 | .awaitable = .null, |
| 1059 | }, .monotonic).cancelation) { |
| 1060 | .none => unreachable, |
| 1061 | .parked => unreachable, |
| 1062 | .blocked_windows_dns => unreachable, |
| 1063 | .canceling => unreachable, |
| 1064 | .canceled => unreachable, |
| 1065 | .blocked => {}, // new status is `.none` |
| 1066 | .blocked_canceling => {}, // new status is `.canceling` |
| 659 | 1067 | } |
| 660 | 1068 | } |
| 1069 | /// Convenience wrapper which calls `finish`, then returns `err`. |
| 1070 | fn fail(s: Syscall, err: anytype) @TypeOf(err) { |
| 1071 | s.finish(); |
| 1072 | return err; |
| 1073 | } |
| 1074 | /// Convenience wrapper which calls `finish`, then calls `Threaded.errnoBug`. |
| 1075 | fn errnoBug(s: Syscall, err: posix.E) Io.UnexpectedError { |
| 1076 | @branchHint(.cold); |
| 1077 | s.finish(); |
| 1078 | return Threaded.errnoBug(err); |
| 1079 | } |
| 1080 | /// Convenience wrapper which calls `finish`, then calls `posix.unexpectedErrno`. |
| 1081 | fn unexpectedErrno(s: Syscall, err: posix.E) Io.UnexpectedError { |
| 1082 | @branchHint(.cold); |
| 1083 | s.finish(); |
| 1084 | return posix.unexpectedErrno(err); |
| 1085 | } |
| 661 | 1086 | }; |
| 662 | 1087 | |
| 1088 | const max_iovecs_len = 8; |
| 1089 | const splat_buffer_size = 64; |
| 1090 | |
| 1091 | comptime { |
| 1092 | if (@TypeOf(posix.IOV_MAX) != void) assert(max_iovecs_len <= posix.IOV_MAX); |
| 1093 | } |
| 1094 | |
| 663 | 1095 | pub const InitOptions = struct { |
| 664 | 1096 | /// Affects how many bytes are memory-mapped for threads. |
| 665 | 1097 | stack_size: usize = std.Thread.SpawnConfig.default_stack_size, |
| ... | ... | @@ -727,14 +1159,10 @@ pub fn init( |
| 727 | 1159 | .old_sig_io = undefined, |
| 728 | 1160 | .old_sig_pipe = undefined, |
| 729 | 1161 | .have_signal_handler = false, |
| 730 | | .main_thread = .{ |
| 731 | | .signal_id = Thread.currentSignalId(), |
| 732 | | .current_closure = null, |
| 733 | | .cancel_protection = .unblocked, |
| 734 | | }, |
| 735 | 1162 | .argv0 = options.argv0, |
| 736 | 1163 | .environ = options.environ, |
| 737 | 1164 | .robust_cancel = options.robust_cancel, |
| 1165 | .worker_threads = .init(null), |
| 738 | 1166 | }; |
| 739 | 1167 | |
| 740 | 1168 | if (posix.Sigaction != void) { |
| ... | ... | @@ -768,14 +1196,10 @@ pub const init_single_threaded: Threaded = .{ |
| 768 | 1196 | .old_sig_io = undefined, |
| 769 | 1197 | .old_sig_pipe = undefined, |
| 770 | 1198 | .have_signal_handler = false, |
| 771 | | .main_thread = .{ |
| 772 | | .signal_id = undefined, |
| 773 | | .current_closure = null, |
| 774 | | .cancel_protection = .unblocked, |
| 775 | | }, |
| 776 | 1199 | .robust_cancel = .disabled, |
| 777 | 1200 | .argv0 = .{}, |
| 778 | 1201 | .environ = .{}, |
| 1202 | .worker_threads = .init(null), |
| 779 | 1203 | }; |
| 780 | 1204 | |
| 781 | 1205 | var global_single_threaded_instance: Threaded = .init_single_threaded; |
| ... | ... | @@ -822,22 +1246,40 @@ fn join(t: *Threaded) void { |
| 822 | 1246 | |
| 823 | 1247 | fn worker(t: *Threaded) void { |
| 824 | 1248 | var thread: Thread = .{ |
| 825 | | .signal_id = Thread.currentSignalId(), |
| 826 | | .current_closure = null, |
| 1249 | .next = undefined, |
| 1250 | .signalee_id = Thread.currentSignaleeId(), |
| 1251 | .status = .init(.{ |
| 1252 | .cancelation = .none, |
| 1253 | .awaitable = .null, |
| 1254 | }), |
| 827 | 1255 | .cancel_protection = .unblocked, |
| 828 | 1256 | }; |
| 829 | 1257 | Thread.current = &thread; |
| 830 | 1258 | |
| 1259 | { |
| 1260 | var head = t.worker_threads.load(.monotonic); |
| 1261 | while (true) { |
| 1262 | thread.next = head; |
| 1263 | head = t.worker_threads.cmpxchgWeak( |
| 1264 | head, |
| 1265 | &thread, |
| 1266 | .release, |
| 1267 | .monotonic, |
| 1268 | ) orelse break; |
| 1269 | } |
| 1270 | } |
| 1271 | |
| 831 | 1272 | defer t.wait_group.finish(); |
| 832 | 1273 | |
| 833 | 1274 | t.mutex.lock(); |
| 834 | 1275 | defer t.mutex.unlock(); |
| 835 | 1276 | |
| 836 | 1277 | while (true) { |
| 837 | | while (t.run_queue.popFirst()) |closure_node| { |
| 1278 | while (t.run_queue.popFirst()) |runnable_node| { |
| 838 | 1279 | t.mutex.unlock(); |
| 839 | | const closure: *Closure = @fieldParentPtr("node", closure_node); |
| 840 | | closure.start(closure, t); |
| 1280 | thread.cancel_protection = .unblocked; |
| 1281 | const runnable: *Runnable = @fieldParentPtr("node", runnable_node); |
| 1282 | runnable.startFn(runnable, &thread, t); |
| 841 | 1283 | t.mutex.lock(); |
| 842 | 1284 | t.busy_count -= 1; |
| 843 | 1285 | } |
| ... | ... | @@ -1145,103 +1587,6 @@ const linux_copy_file_range_use_c = std.c.versionCheck(if (builtin.abi.isAndroid |
| 1145 | 1587 | }); |
| 1146 | 1588 | const linux_copy_file_range_sys = if (linux_copy_file_range_use_c) std.c else std.os.linux; |
| 1147 | 1589 | |
| 1148 | | /// Trailing data: |
| 1149 | | /// 1. context |
| 1150 | | /// 2. result |
| 1151 | | const AsyncClosure = struct { |
| 1152 | | closure: Closure, |
| 1153 | | func: *const fn (context: *anyopaque, result: *anyopaque) void, |
| 1154 | | event: Io.Event, |
| 1155 | | select_condition: ?*Io.Event, |
| 1156 | | context_alignment: Alignment, |
| 1157 | | result_offset: usize, |
| 1158 | | alloc_len: usize, |
| 1159 | | |
| 1160 | | const done_event: *Io.Event = @ptrFromInt(@alignOf(Io.Event)); |
| 1161 | | |
| 1162 | | fn start(closure: *Closure, t: *Threaded) void { |
| 1163 | | const ac: *AsyncClosure = @alignCast(@fieldParentPtr("closure", closure)); |
| 1164 | | const current_thread = Thread.getCurrent(t); |
| 1165 | | |
| 1166 | | current_thread.current_closure = closure; |
| 1167 | | current_thread.cancel_protection = .unblocked; |
| 1168 | | |
| 1169 | | ac.func(ac.contextPointer(), ac.resultPointer()); |
| 1170 | | |
| 1171 | | current_thread.current_closure = null; |
| 1172 | | current_thread.cancel_protection = undefined; |
| 1173 | | |
| 1174 | | if (@atomicRmw(?*Io.Event, &ac.select_condition, .Xchg, done_event, .release)) |select_event| { |
| 1175 | | assert(select_event != done_event); |
| 1176 | | select_event.set(ioBasic(t)); |
| 1177 | | } |
| 1178 | | ac.event.set(ioBasic(t)); |
| 1179 | | } |
| 1180 | | |
| 1181 | | fn resultPointer(ac: *AsyncClosure) [*]u8 { |
| 1182 | | const base: [*]u8 = @ptrCast(ac); |
| 1183 | | return base + ac.result_offset; |
| 1184 | | } |
| 1185 | | |
| 1186 | | fn contextPointer(ac: *AsyncClosure) [*]u8 { |
| 1187 | | const base: [*]u8 = @ptrCast(ac); |
| 1188 | | const context_offset = ac.context_alignment.forward(@intFromPtr(ac) + @sizeOf(AsyncClosure)) - @intFromPtr(ac); |
| 1189 | | return base + context_offset; |
| 1190 | | } |
| 1191 | | |
| 1192 | | fn init( |
| 1193 | | gpa: Allocator, |
| 1194 | | result_len: usize, |
| 1195 | | result_alignment: Alignment, |
| 1196 | | context: []const u8, |
| 1197 | | context_alignment: Alignment, |
| 1198 | | func: *const fn (context: *const anyopaque, result: *anyopaque) void, |
| 1199 | | ) Allocator.Error!*AsyncClosure { |
| 1200 | | const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(AsyncClosure); |
| 1201 | | const worst_case_context_offset = context_alignment.forward(@sizeOf(AsyncClosure) + max_context_misalignment); |
| 1202 | | const worst_case_result_offset = result_alignment.forward(worst_case_context_offset + context.len); |
| 1203 | | const alloc_len = worst_case_result_offset + result_len; |
| 1204 | | |
| 1205 | | const ac: *AsyncClosure = @ptrCast(@alignCast(try gpa.alignedAlloc(u8, .of(AsyncClosure), alloc_len))); |
| 1206 | | errdefer comptime unreachable; |
| 1207 | | |
| 1208 | | const actual_context_addr = context_alignment.forward(@intFromPtr(ac) + @sizeOf(AsyncClosure)); |
| 1209 | | const actual_result_addr = result_alignment.forward(actual_context_addr + context.len); |
| 1210 | | const actual_result_offset = actual_result_addr - @intFromPtr(ac); |
| 1211 | | ac.* = .{ |
| 1212 | | .closure = .{ |
| 1213 | | .cancel_status = .none, |
| 1214 | | .start = start, |
| 1215 | | }, |
| 1216 | | .func = func, |
| 1217 | | .context_alignment = context_alignment, |
| 1218 | | .result_offset = actual_result_offset, |
| 1219 | | .alloc_len = alloc_len, |
| 1220 | | .event = .unset, |
| 1221 | | .select_condition = null, |
| 1222 | | }; |
| 1223 | | @memcpy(ac.contextPointer()[0..context.len], context); |
| 1224 | | return ac; |
| 1225 | | } |
| 1226 | | |
| 1227 | | fn waitAndDeinit(ac: *AsyncClosure, t: *Threaded, result: []u8) void { |
| 1228 | | ac.event.wait(ioBasic(t)) catch |err| switch (err) { |
| 1229 | | error.Canceled => { |
| 1230 | | ac.closure.requestCancel(t); |
| 1231 | | ac.event.waitUncancelable(ioBasic(t)); |
| 1232 | | recancel(t); |
| 1233 | | }, |
| 1234 | | }; |
| 1235 | | @memcpy(result, ac.resultPointer()[0..result.len]); |
| 1236 | | ac.deinit(t.allocator); |
| 1237 | | } |
| 1238 | | |
| 1239 | | fn deinit(ac: *AsyncClosure, gpa: Allocator) void { |
| 1240 | | const base: [*]align(@alignOf(AsyncClosure)) u8 = @ptrCast(ac); |
| 1241 | | gpa.free(base[0..ac.alloc_len]); |
| 1242 | | } |
| 1243 | | }; |
| 1244 | | |
| 1245 | 1590 | fn async( |
| 1246 | 1591 | userdata: ?*anyopaque, |
| 1247 | 1592 | result: []u8, |
| ... | ... | @@ -1255,10 +1600,13 @@ fn async( |
| 1255 | 1600 | start(context.ptr, result.ptr); |
| 1256 | 1601 | return null; |
| 1257 | 1602 | } |
| 1603 | |
| 1258 | 1604 | const gpa = t.allocator; |
| 1259 | | const ac = AsyncClosure.init(gpa, result.len, result_alignment, context, context_alignment, start) catch { |
| 1260 | | start(context.ptr, result.ptr); |
| 1261 | | return null; |
| 1605 | const future = Future.create(gpa, result.len, result_alignment, context, context_alignment, start) catch |err| switch (err) { |
| 1606 | error.OutOfMemory => { |
| 1607 | start(context.ptr, result.ptr); |
| 1608 | return null; |
| 1609 | }, |
| 1262 | 1610 | }; |
| 1263 | 1611 | |
| 1264 | 1612 | t.mutex.lock(); |
| ... | ... | @@ -1267,7 +1615,7 @@ fn async( |
| 1267 | 1615 | |
| 1268 | 1616 | if (busy_count >= @intFromEnum(t.async_limit)) { |
| 1269 | 1617 | t.mutex.unlock(); |
| 1270 | | ac.deinit(gpa); |
| 1618 | future.destroy(gpa); |
| 1271 | 1619 | start(context.ptr, result.ptr); |
| 1272 | 1620 | return null; |
| 1273 | 1621 | } |
| ... | ... | @@ -1281,17 +1629,18 @@ fn async( |
| 1281 | 1629 | t.wait_group.finish(); |
| 1282 | 1630 | t.busy_count = busy_count; |
| 1283 | 1631 | t.mutex.unlock(); |
| 1284 | | ac.deinit(gpa); |
| 1632 | future.destroy(gpa); |
| 1285 | 1633 | start(context.ptr, result.ptr); |
| 1286 | 1634 | return null; |
| 1287 | 1635 | }; |
| 1288 | 1636 | thread.detach(); |
| 1289 | 1637 | } |
| 1290 | 1638 | |
| 1291 | | t.run_queue.prepend(&ac.closure.node); |
| 1639 | t.run_queue.prepend(&future.runnable.node); |
| 1640 | |
| 1292 | 1641 | t.mutex.unlock(); |
| 1293 | 1642 | t.cond.signal(); |
| 1294 | | return @ptrCast(ac); |
| 1643 | return @ptrCast(future); |
| 1295 | 1644 | } |
| 1296 | 1645 | |
| 1297 | 1646 | fn concurrent( |
| ... | ... | @@ -1307,9 +1656,10 @@ fn concurrent( |
| 1307 | 1656 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1308 | 1657 | |
| 1309 | 1658 | const gpa = t.allocator; |
| 1310 | | const ac = AsyncClosure.init(gpa, result_len, result_alignment, context, context_alignment, start) catch |
| 1311 | | return error.ConcurrencyUnavailable; |
| 1312 | | errdefer ac.deinit(gpa); |
| 1659 | const future = Future.create(gpa, result_len, result_alignment, context, context_alignment, start) catch |err| switch (err) { |
| 1660 | error.OutOfMemory => return error.ConcurrencyUnavailable, |
| 1661 | }; |
| 1662 | errdefer future.destroy(gpa); |
| 1313 | 1663 | |
| 1314 | 1664 | t.mutex.lock(); |
| 1315 | 1665 | defer t.mutex.unlock(); |
| ... | ... | @@ -1329,110 +1679,32 @@ fn concurrent( |
| 1329 | 1679 | |
| 1330 | 1680 | const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch |
| 1331 | 1681 | return error.ConcurrencyUnavailable; |
| 1682 | |
| 1332 | 1683 | thread.detach(); |
| 1333 | 1684 | } |
| 1334 | 1685 | |
| 1335 | | t.run_queue.prepend(&ac.closure.node); |
| 1686 | t.run_queue.prepend(&future.runnable.node); |
| 1687 | |
| 1336 | 1688 | t.cond.signal(); |
| 1337 | | return @ptrCast(ac); |
| 1689 | return @ptrCast(future); |
| 1338 | 1690 | } |
| 1339 | 1691 | |
| 1340 | | const GroupClosure = struct { |
| 1341 | | closure: Closure, |
| 1342 | | group: *Io.Group, |
| 1343 | | /// Points to sibling `GroupClosure`. Used for walking the group to cancel all. |
| 1344 | | node: std.SinglyLinkedList.Node, |
| 1345 | | func: *const fn (*Io.Group, context: *anyopaque) Io.Cancelable!void, |
| 1346 | | context_alignment: Alignment, |
| 1347 | | alloc_len: usize, |
| 1348 | | |
| 1349 | | fn start(closure: *Closure, t: *Threaded) void { |
| 1350 | | const gc: *GroupClosure = @alignCast(@fieldParentPtr("closure", closure)); |
| 1351 | | const current_thread = Thread.getCurrent(t); |
| 1352 | | const group = gc.group; |
| 1353 | | const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state); |
| 1354 | | const event: *Io.Event = @ptrCast(&group.context); |
| 1355 | | current_thread.current_closure = closure; |
| 1356 | | current_thread.cancel_protection = .unblocked; |
| 1357 | | |
| 1358 | | assertResult(closure, gc.func(group, gc.contextPointer())); |
| 1359 | | |
| 1360 | | current_thread.current_closure = null; |
| 1361 | | current_thread.cancel_protection = undefined; |
| 1362 | | |
| 1363 | | const prev_state = group_state.fetchSub(sync_one_pending, .acq_rel); |
| 1364 | | assert((prev_state / sync_one_pending) > 0); |
| 1365 | | if (prev_state == (sync_one_pending | sync_is_waiting)) event.set(ioBasic(t)); |
| 1366 | | } |
| 1367 | | |
| 1368 | | fn assertResult(closure: *Closure, result: Io.Cancelable!void) void { |
| 1369 | | if (result) |_| switch (closure.cancel_status.unpack()) { |
| 1370 | | .none, .requested => {}, |
| 1371 | | .acknowledged => unreachable, // task illegally swallowed error.Canceled |
| 1372 | | .signal_id => unreachable, |
| 1373 | | } else |err| switch (err) { |
| 1374 | | error.Canceled => assert(closure.cancel_status == .acknowledged), |
| 1375 | | } |
| 1376 | | } |
| 1377 | | |
| 1378 | | fn contextPointer(gc: *GroupClosure) [*]u8 { |
| 1379 | | const base: [*]u8 = @ptrCast(gc); |
| 1380 | | const context_offset = gc.context_alignment.forward(@intFromPtr(gc) + @sizeOf(GroupClosure)) - @intFromPtr(gc); |
| 1381 | | return base + context_offset; |
| 1382 | | } |
| 1383 | | |
| 1384 | | /// Does not initialize the `node` field. |
| 1385 | | fn init( |
| 1386 | | gpa: Allocator, |
| 1387 | | group: *Io.Group, |
| 1388 | | context: []const u8, |
| 1389 | | context_alignment: Alignment, |
| 1390 | | func: *const fn (*Io.Group, context: *const anyopaque) Io.Cancelable!void, |
| 1391 | | ) Allocator.Error!*GroupClosure { |
| 1392 | | const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(GroupClosure); |
| 1393 | | const worst_case_context_offset = context_alignment.forward(@sizeOf(GroupClosure) + max_context_misalignment); |
| 1394 | | const alloc_len = worst_case_context_offset + context.len; |
| 1395 | | |
| 1396 | | const gc: *GroupClosure = @ptrCast(@alignCast(try gpa.alignedAlloc(u8, .of(GroupClosure), alloc_len))); |
| 1397 | | errdefer comptime unreachable; |
| 1398 | | |
| 1399 | | gc.* = .{ |
| 1400 | | .closure = .{ |
| 1401 | | .cancel_status = .none, |
| 1402 | | .start = start, |
| 1403 | | }, |
| 1404 | | .group = group, |
| 1405 | | .node = undefined, |
| 1406 | | .func = func, |
| 1407 | | .context_alignment = context_alignment, |
| 1408 | | .alloc_len = alloc_len, |
| 1409 | | }; |
| 1410 | | @memcpy(gc.contextPointer()[0..context.len], context); |
| 1411 | | return gc; |
| 1412 | | } |
| 1413 | | |
| 1414 | | fn deinit(gc: *GroupClosure, gpa: Allocator) void { |
| 1415 | | const base: [*]align(@alignOf(GroupClosure)) u8 = @ptrCast(gc); |
| 1416 | | gpa.free(base[0..gc.alloc_len]); |
| 1417 | | } |
| 1418 | | |
| 1419 | | const sync_is_waiting: usize = 1 << 0; |
| 1420 | | const sync_one_pending: usize = 1 << 1; |
| 1421 | | }; |
| 1422 | | |
| 1423 | 1692 | fn groupAsync( |
| 1424 | 1693 | userdata: ?*anyopaque, |
| 1425 | | group: *Io.Group, |
| 1694 | type_erased: *Io.Group, |
| 1426 | 1695 | context: []const u8, |
| 1427 | 1696 | context_alignment: Alignment, |
| 1428 | 1697 | start: *const fn (*Io.Group, context: *const anyopaque) Io.Cancelable!void, |
| 1429 | 1698 | ) void { |
| 1430 | 1699 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1431 | | if (builtin.single_threaded) return start(group, context.ptr) catch unreachable; |
| 1700 | const g: Group = .{ .ptr = type_erased }; |
| 1701 | |
| 1702 | if (builtin.single_threaded) return start(g.ptr, context.ptr) catch unreachable; |
| 1432 | 1703 | |
| 1433 | 1704 | const gpa = t.allocator; |
| 1434 | | const gc = GroupClosure.init(gpa, group, context, context_alignment, start) catch |
| 1435 | | return t.assertGroupResult(start(group, context.ptr)); |
| 1705 | const task = Group.Task.create(gpa, g, context, context_alignment, start) catch |err| switch (err) { |
| 1706 | error.OutOfMemory => return t.assertGroupResult(start(g.ptr, context.ptr)), |
| 1707 | }; |
| 1436 | 1708 | |
| 1437 | 1709 | t.mutex.lock(); |
| 1438 | 1710 | |
| ... | ... | @@ -1440,8 +1712,8 @@ fn groupAsync( |
| 1440 | 1712 | |
| 1441 | 1713 | if (busy_count >= @intFromEnum(t.async_limit)) { |
| 1442 | 1714 | t.mutex.unlock(); |
| 1443 | | gc.deinit(gpa); |
| 1444 | | return t.assertGroupResult(start(group, context.ptr)); |
| 1715 | task.destroy(gpa); |
| 1716 | return t.assertGroupResult(start(g.ptr, context.ptr)); |
| 1445 | 1717 | } |
| 1446 | 1718 | |
| 1447 | 1719 | t.busy_count = busy_count + 1; |
| ... | ... | @@ -1453,37 +1725,48 @@ fn groupAsync( |
| 1453 | 1725 | t.wait_group.finish(); |
| 1454 | 1726 | t.busy_count = busy_count; |
| 1455 | 1727 | t.mutex.unlock(); |
| 1456 | | gc.deinit(gpa); |
| 1457 | | return t.assertGroupResult(start(group, context.ptr)); |
| 1728 | task.destroy(gpa); |
| 1729 | return t.assertGroupResult(start(g.ptr, context.ptr)); |
| 1458 | 1730 | }; |
| 1459 | 1731 | thread.detach(); |
| 1460 | 1732 | } |
| 1461 | 1733 | |
| 1462 | | // Append to the group linked list inside the mutex to make `Io.Group.async` thread-safe. |
| 1463 | | gc.node = .{ .next = @ptrCast(@alignCast(group.token.load(.monotonic))) }; |
| 1464 | | group.token.store(&gc.node, .monotonic); |
| 1465 | | |
| 1466 | | t.run_queue.prepend(&gc.closure.node); |
| 1467 | | |
| 1468 | | // This needs to be done before unlocking the mutex to avoid a race with |
| 1469 | | // the associated task finishing. |
| 1470 | | const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state); |
| 1471 | | const prev_state = group_state.fetchAdd(GroupClosure.sync_one_pending, .monotonic); |
| 1472 | | assert((prev_state / GroupClosure.sync_one_pending) < (std.math.maxInt(usize) / GroupClosure.sync_one_pending)); |
| 1734 | // TODO: if this logic is changed to be lock-free, this `fetchAdd` must be released by the queue |
| 1735 | // prepend so that the task doesn't finish without observing this and try to decrement the count |
| 1736 | // below zero. |
| 1737 | _ = g.status().fetchAdd(.{ |
| 1738 | .num_running = 1, |
| 1739 | .have_awaiter = false, |
| 1740 | .canceled = false, |
| 1741 | }, .monotonic); |
| 1742 | t.run_queue.prepend(&task.runnable.node); |
| 1473 | 1743 | |
| 1474 | 1744 | t.mutex.unlock(); |
| 1475 | 1745 | t.cond.signal(); |
| 1476 | 1746 | } |
| 1477 | 1747 | |
| 1478 | | fn assertGroupResult(t: *Threaded, result: Io.Cancelable!void) void { |
| 1479 | | const current_thread: *Thread = .getCurrent(t); |
| 1480 | | const current_closure = current_thread.current_closure orelse return; |
| 1481 | | GroupClosure.assertResult(current_closure, result); |
| 1748 | fn assertGroupResult(result: Io.Cancelable!void) void { |
| 1749 | const cancel_acknowledged = if (Thread.current) |thread| |
| 1750 | switch (thread.status.load(.monotonic).cancelation) { |
| 1751 | .none, .canceling => false, |
| 1752 | .canceled => true, |
| 1753 | .parked => unreachable, |
| 1754 | .blocked => unreachable, |
| 1755 | .blocked_windows_dns => unreachable, |
| 1756 | .blocked_canceling => unreachable, |
| 1757 | } |
| 1758 | else |
| 1759 | false; |
| 1760 | if (result) { |
| 1761 | assert(!cancel_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled` |
| 1762 | } else |err| switch (err) { |
| 1763 | error.Canceled => assert(cancel_acknowledged), // group task returned `error.Canceled` but was never canceled |
| 1764 | } |
| 1482 | 1765 | } |
| 1483 | 1766 | |
| 1484 | 1767 | fn groupConcurrent( |
| 1485 | 1768 | userdata: ?*anyopaque, |
| 1486 | | group: *Io.Group, |
| 1769 | type_erased: *Io.Group, |
| 1487 | 1770 | context: []const u8, |
| 1488 | 1771 | context_alignment: Alignment, |
| 1489 | 1772 | start: *const fn (*Io.Group, context: *const anyopaque) Io.Cancelable!void, |
| ... | ... | @@ -1491,10 +1774,13 @@ fn groupConcurrent( |
| 1491 | 1774 | if (builtin.single_threaded) return error.ConcurrencyUnavailable; |
| 1492 | 1775 | |
| 1493 | 1776 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1777 | const g: Group = .{ .ptr = type_erased }; |
| 1494 | 1778 | |
| 1495 | 1779 | const gpa = t.allocator; |
| 1496 | | const gc = GroupClosure.init(gpa, group, context, context_alignment, start) catch |
| 1497 | | return error.ConcurrencyUnavailable; |
| 1780 | const task = Group.Task.create(gpa, g, context, context_alignment, start) catch |err| switch (err) { |
| 1781 | error.OutOfMemory => return error.ConcurrencyUnavailable, |
| 1782 | }; |
| 1783 | errdefer task.destroy(gpa); |
| 1498 | 1784 | |
| 1499 | 1785 | t.mutex.lock(); |
| 1500 | 1786 | defer t.mutex.unlock(); |
| ... | ... | @@ -1514,102 +1800,126 @@ fn groupConcurrent( |
| 1514 | 1800 | |
| 1515 | 1801 | const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch |
| 1516 | 1802 | return error.ConcurrencyUnavailable; |
| 1803 | |
| 1517 | 1804 | thread.detach(); |
| 1518 | 1805 | } |
| 1519 | 1806 | |
| 1520 | | // Append to the group linked list inside the mutex to make `Io.Group.concurrent` thread-safe. |
| 1521 | | gc.node = .{ .next = @ptrCast(@alignCast(group.token.load(.monotonic))) }; |
| 1522 | | group.token.store(&gc.node, .monotonic); |
| 1523 | | |
| 1524 | | t.run_queue.prepend(&gc.closure.node); |
| 1525 | | |
| 1526 | | // This needs to be done before unlocking the mutex to avoid a race with |
| 1527 | | // the associated task finishing. |
| 1528 | | const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state); |
| 1529 | | const prev_state = group_state.fetchAdd(GroupClosure.sync_one_pending, .monotonic); |
| 1530 | | assert((prev_state / GroupClosure.sync_one_pending) < (std.math.maxInt(usize) / GroupClosure.sync_one_pending)); |
| 1807 | // TODO: if this logic is changed to be lock-free, this `fetchAdd` must be released by the queue |
| 1808 | // prepend so that the task doesn't finish without observing this and try to decrement the count |
| 1809 | // below zero. |
| 1810 | _ = g.status().fetchAdd(.{ |
| 1811 | .num_running = 1, |
| 1812 | .have_awaiter = false, |
| 1813 | .canceled = false, |
| 1814 | }, .monotonic); |
| 1815 | t.run_queue.prepend(&task.runnable.node); |
| 1531 | 1816 | |
| 1532 | 1817 | t.cond.signal(); |
| 1533 | 1818 | } |
| 1534 | 1819 | |
| 1535 | | fn groupAwait(userdata: ?*anyopaque, group: *Io.Group, initial_token: *anyopaque) Io.Cancelable!void { |
| 1820 | fn groupAwait(userdata: ?*anyopaque, type_erased: *Io.Group, initial_token: *anyopaque) Io.Cancelable!void { |
| 1821 | _ = initial_token; // we need to load `token` *after* the group finishes |
| 1536 | 1822 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1537 | | const gpa = t.allocator; |
| 1823 | const g: Group = .{ .ptr = type_erased }; |
| 1824 | const thread: *Thread = .getCurrent(t); |
| 1538 | 1825 | |
| 1539 | | _ = initial_token; // we need to load `token` *after* the group finishes |
| 1826 | var num_completed: std.atomic.Value(u32) = .init(0); |
| 1827 | g.awaiter().* = &num_completed; |
| 1540 | 1828 | |
| 1541 | | if (builtin.single_threaded) unreachable; // we never set `group.token` to non-`null` |
| 1829 | const pre_await_status = g.status().fetchOr(.{ |
| 1830 | .num_running = 0, |
| 1831 | .have_awaiter = true, |
| 1832 | .canceled = false, |
| 1833 | }, .acq_rel); // acquire results if complete; release `g.awaiter()` |
| 1542 | 1834 | |
| 1543 | | const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state); |
| 1544 | | const event: *Io.Event = @ptrCast(&group.context); |
| 1545 | | const prev_state = group_state.fetchAdd(GroupClosure.sync_is_waiting, .acquire); |
| 1546 | | assert(prev_state & GroupClosure.sync_is_waiting == 0); |
| 1547 | | { |
| 1548 | | errdefer _ = group_state.fetchSub(GroupClosure.sync_is_waiting, .monotonic); |
| 1549 | | // This event.wait can return error.Canceled, in which case this logic does |
| 1550 | | // *not* propagate cancel requests to each group member. Instead, the user |
| 1551 | | // code will likely do this with a defered call to groupCancel, or, |
| 1552 | | // intentionally not do this. |
| 1553 | | if ((prev_state / GroupClosure.sync_one_pending) > 0) try event.wait(ioBasic(t)); |
| 1835 | assert(!pre_await_status.have_awaiter); |
| 1836 | assert(!pre_await_status.canceled); |
| 1837 | if (pre_await_status.num_running == 0) { |
| 1838 | // Already done. Since the group is finished, it's illegal to spawn more tasks in it |
| 1839 | // until we return, so we can access `g.status()` non-atomically. |
| 1840 | g.status().raw.have_awaiter = false; |
| 1841 | return; |
| 1554 | 1842 | } |
| 1555 | 1843 | |
| 1556 | | // Since the group has now finished, it's illegal to add more tasks to it until we return. It's |
| 1557 | | // also illegal for us to race with another `await` or `cancel`. Therefore, we must be the only |
| 1558 | | // thread who can access `group` right now. |
| 1559 | | var it: ?*std.SinglyLinkedList.Node = @ptrCast(@alignCast(group.token.raw)); |
| 1560 | | group.token.raw = null; |
| 1561 | | while (it) |node| { |
| 1562 | | it = node.next; // update `it` now, because `deinit` will invalidate `node` |
| 1563 | | const gc: *GroupClosure = @fieldParentPtr("node", node); |
| 1564 | | gc.deinit(gpa); |
| 1844 | while (thread.futexWait(&num_completed.raw, 0)) { |
| 1845 | switch (num_completed.load(.acquire)) { // acquire task results |
| 1846 | 0 => continue, |
| 1847 | 1 => break, |
| 1848 | else => unreachable, // group was reused before `await` returned |
| 1849 | } |
| 1850 | } else |err| switch (err) { |
| 1851 | error.Canceled => { |
| 1852 | const pre_cancel_status = g.status().fetchOr(.{ |
| 1853 | .num_running = 0, |
| 1854 | .have_awaiter = false, |
| 1855 | .canceled = true, |
| 1856 | }, .acq_rel); // acquire results if complete; release `g.awaiter()` |
| 1857 | assert(pre_cancel_status.have_awaiter); |
| 1858 | assert(!pre_cancel_status.canceled); |
| 1859 | |
| 1860 | // Even if `pre_cancel_status.num_running == 0`, we still need to wait for the signal, |
| 1861 | // because in that case the last member of the group is already trying to modify it. |
| 1862 | // However, if we know everything is done, we *can* skip signaling blocked threads. |
| 1863 | const skip_signals = pre_cancel_status.num_running == 0; |
| 1864 | g.waitForCancelWithSignaling(t, &num_completed, skip_signals); |
| 1865 | |
| 1866 | // The group is finished, so it's illegal to spawn more tasks in it until we return, so |
| 1867 | // we can access `g.status()` non-atomically. |
| 1868 | g.status().raw.canceled = false; |
| 1869 | g.status().raw.have_awaiter = false; |
| 1870 | return error.Canceled; |
| 1871 | }, |
| 1565 | 1872 | } |
| 1873 | |
| 1874 | // The group is finished, so it's illegal to spawn more tasks in it until we return, so |
| 1875 | // we can access `g.status()` non-atomically. |
| 1876 | g.status().raw.have_awaiter = false; |
| 1566 | 1877 | } |
| 1567 | 1878 | |
| 1568 | | fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, initial_token: *anyopaque) void { |
| 1879 | fn groupCancel(userdata: ?*anyopaque, type_erased: *Io.Group, initial_token: *anyopaque) void { |
| 1880 | _ = initial_token; |
| 1569 | 1881 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1570 | | const gpa = t.allocator; |
| 1882 | const g: Group = .{ .ptr = type_erased }; |
| 1571 | 1883 | |
| 1572 | | _ = initial_token; // we need to load `token` *after* the group finishes |
| 1884 | var num_completed: std.atomic.Value(u32) = .init(0); |
| 1885 | g.awaiter().* = &num_completed; |
| 1573 | 1886 | |
| 1574 | | if (builtin.single_threaded) unreachable; // we never set `group.token` to non-`null` |
| 1887 | const pre_cancel_status = g.status().fetchOr(.{ |
| 1888 | .num_running = 0, |
| 1889 | .have_awaiter = true, |
| 1890 | .canceled = true, |
| 1891 | }, .acq_rel); // acquire results if complete; release `g.awaiter()` |
| 1575 | 1892 | |
| 1576 | | { |
| 1577 | | var it: ?*std.SinglyLinkedList.Node = @ptrCast(@alignCast(group.token.load(.monotonic))); |
| 1578 | | while (it) |node| : (it = node.next) { |
| 1579 | | const gc: *GroupClosure = @fieldParentPtr("node", node); |
| 1580 | | gc.closure.requestCancel(t); |
| 1581 | | } |
| 1893 | assert(!pre_cancel_status.have_awaiter); |
| 1894 | assert(!pre_cancel_status.canceled); |
| 1895 | if (pre_cancel_status.num_running == 0) { |
| 1896 | // Already done. Since the group is finished, it's illegal to spawn more tasks in it |
| 1897 | // until we return, so we can access `g.status()` non-atomically. |
| 1898 | g.status().raw.have_awaiter = false; |
| 1899 | g.status().raw.canceled = false; |
| 1900 | return; |
| 1582 | 1901 | } |
| 1583 | 1902 | |
| 1584 | | const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state); |
| 1585 | | const event: *Io.Event = @ptrCast(&group.context); |
| 1586 | | const prev_state = group_state.fetchAdd(GroupClosure.sync_is_waiting, .acquire); |
| 1587 | | assert(prev_state & GroupClosure.sync_is_waiting == 0); |
| 1588 | | if ((prev_state / GroupClosure.sync_one_pending) > 0) event.waitUncancelable(ioBasic(t)); |
| 1903 | g.waitForCancelWithSignaling(t, &num_completed, false); |
| 1589 | 1904 | |
| 1590 | | // Since the group has now finished, it's illegal to add more tasks to it until we return. It's |
| 1591 | | // also illegal for us to race with another `await` or `cancel`. Therefore, we must be the only |
| 1592 | | // thread who can access `group` right now. |
| 1593 | | var it: ?*std.SinglyLinkedList.Node = @ptrCast(@alignCast(group.token.raw)); |
| 1594 | | group.token.raw = null; |
| 1595 | | while (it) |node| { |
| 1596 | | it = node.next; // update `it` now, because `deinit` will invalidate `node` |
| 1597 | | const gc: *GroupClosure = @fieldParentPtr("node", node); |
| 1598 | | gc.deinit(gpa); |
| 1599 | | } |
| 1905 | g.status().raw = .{ .num_running = 0, .have_awaiter = false, .canceled = false }; |
| 1600 | 1906 | } |
| 1601 | 1907 | |
| 1602 | 1908 | fn recancel(userdata: ?*anyopaque) void { |
| 1603 | 1909 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1604 | 1910 | const current_thread: *Thread = .getCurrent(t); |
| 1605 | | const cancel_status = ¤t_thread.current_closure.?.cancel_status; |
| 1606 | | switch (@atomicLoad(CancelStatus, cancel_status, .monotonic)) { |
| 1607 | | .none => unreachable, // called `recancel` when not canceled |
| 1608 | | .requested => unreachable, // called `recancel` when cancelation was already outstanding |
| 1609 | | .acknowledged => {}, |
| 1610 | | _ => unreachable, // invalid state: not in a syscall |
| 1911 | switch (current_thread.status.fetchXor(.{ |
| 1912 | .cancelation = @enumFromInt(0b001), |
| 1913 | .awaitable = .null, |
| 1914 | }, .monotonic).cancelation) { |
| 1915 | .canceled => {}, |
| 1916 | .none => unreachable, // called `recancel` but was not canceled |
| 1917 | .canceling => unreachable, // called `recancel` but cancelation was already pending |
| 1918 | .parked => unreachable, |
| 1919 | .blocked => unreachable, |
| 1920 | .blocked_windows_dns => unreachable, |
| 1921 | .blocked_canceling => unreachable, |
| 1611 | 1922 | } |
| 1612 | | @atomicStore(CancelStatus, cancel_status, .requested, .monotonic); |
| 1613 | 1923 | } |
| 1614 | 1924 | |
| 1615 | 1925 | fn swapCancelProtection(userdata: ?*anyopaque, new: Io.CancelProtection) Io.CancelProtection { |
| ... | ... | @@ -1622,7 +1932,8 @@ fn swapCancelProtection(userdata: ?*anyopaque, new: Io.CancelProtection) Io.Canc |
| 1622 | 1932 | |
| 1623 | 1933 | fn checkCancel(userdata: ?*anyopaque) Io.Cancelable!void { |
| 1624 | 1934 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1625 | | return Thread.getCurrent(t).checkCancel(); |
| 1935 | _ = t; |
| 1936 | return Thread.checkCancel(); |
| 1626 | 1937 | } |
| 1627 | 1938 | |
| 1628 | 1939 | fn await( |
| ... | ... | @@ -1633,8 +1944,51 @@ fn await( |
| 1633 | 1944 | ) void { |
| 1634 | 1945 | _ = result_alignment; |
| 1635 | 1946 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1636 | | const closure: *AsyncClosure = @ptrCast(@alignCast(any_future)); |
| 1637 | | closure.waitAndDeinit(t, result); |
| 1947 | const future: *Future = @ptrCast(@alignCast(any_future)); |
| 1948 | const thread: *Thread = .getCurrent(t); |
| 1949 | |
| 1950 | var num_completed: std.atomic.Value(u32) = .init(0); |
| 1951 | future.awaiter = &num_completed; |
| 1952 | |
| 1953 | const pre_await_status = future.status.fetchOr(.{ |
| 1954 | .tag = .pending_awaited, |
| 1955 | .thread = .null, |
| 1956 | }, .acq_rel); // acquire results if complete; release `future.awaiter` |
| 1957 | switch (pre_await_status.tag) { |
| 1958 | .pending => while (thread.futexWait(&num_completed.raw, 0)) { |
| 1959 | switch (num_completed.load(.acquire)) { // acquire task results |
| 1960 | 0 => continue, |
| 1961 | 1 => break, |
| 1962 | else => unreachable, // group was reused before `await` returned |
| 1963 | } |
| 1964 | } else |err| switch (err) { |
| 1965 | error.Canceled => { |
| 1966 | const pre_cancel_status = future.status.fetchOr(.{ |
| 1967 | .tag = .pending_canceled, |
| 1968 | .thread = .null, |
| 1969 | }, .acq_rel); // acquire results if complete; release `future.awaiter` |
| 1970 | switch (pre_cancel_status.tag) { |
| 1971 | .pending => unreachable, // invalid state: we already awaited |
| 1972 | .pending_awaited => { |
| 1973 | const working_thread = pre_cancel_status.thread.unpack(); |
| 1974 | future.waitForCancelWithSignaling(t, &num_completed, @alignCast(working_thread)); |
| 1975 | }, |
| 1976 | .pending_canceled => unreachable, // `await` raced with `cancel` |
| 1977 | .done => { |
| 1978 | // The task just finished, but we still need to wait for the signal, because the |
| 1979 | // task thread already figured out that they need to update `future.awaiter`. |
| 1980 | future.waitForCancelWithSignaling(t, &num_completed, null); |
| 1981 | }, |
| 1982 | } |
| 1983 | recancel(t); |
| 1984 | }, |
| 1985 | }, |
| 1986 | .pending_awaited => unreachable, // `await` raced with `await` |
| 1987 | .pending_canceled => unreachable, // `await` raced with `cancel` |
| 1988 | .done => {}, |
| 1989 | } |
| 1990 | @memcpy(result, future.resultPointer()); |
| 1991 | future.destroy(t.allocator); |
| 1638 | 1992 | } |
| 1639 | 1993 | |
| 1640 | 1994 | fn cancel( |
| ... | ... | @@ -1645,9 +1999,26 @@ fn cancel( |
| 1645 | 1999 | ) void { |
| 1646 | 2000 | _ = result_alignment; |
| 1647 | 2001 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 1648 | | const ac: *AsyncClosure = @ptrCast(@alignCast(any_future)); |
| 1649 | | ac.closure.requestCancel(t); |
| 1650 | | ac.waitAndDeinit(t, result); |
| 2002 | const future: *Future = @ptrCast(@alignCast(any_future)); |
| 2003 | |
| 2004 | var num_completed: std.atomic.Value(u32) = .init(0); |
| 2005 | future.awaiter = &num_completed; |
| 2006 | |
| 2007 | const pre_cancel_status = future.status.fetchOr(.{ |
| 2008 | .tag = .pending_canceled, |
| 2009 | .thread = .null, |
| 2010 | }, .acq_rel); // acquire results if complete; release `future.awaiter` |
| 2011 | switch (pre_cancel_status.tag) { |
| 2012 | .pending => { |
| 2013 | const working_thread = pre_cancel_status.thread.unpack(); |
| 2014 | future.waitForCancelWithSignaling(t, &num_completed, @alignCast(working_thread)); |
| 2015 | }, |
| 2016 | .pending_awaited => unreachable, // `await` raced with `await` |
| 2017 | .pending_canceled => unreachable, // `await` raced with `cancel` |
| 2018 | .done => {}, |
| 2019 | } |
| 2020 | @memcpy(result, future.resultPointer()); |
| 2021 | future.destroy(t.allocator); |
| 1651 | 2022 | } |
| 1652 | 2023 | |
| 1653 | 2024 | fn futexWait(userdata: ?*anyopaque, ptr: *const u32, expected: u32, timeout: Io.Timeout) Io.Cancelable!void { |
| ... | ... | @@ -8555,32 +8926,69 @@ fn sleepPosix(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void { |
| 8555 | 8926 | fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) Io.Cancelable!usize { |
| 8556 | 8927 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 8557 | 8928 | |
| 8558 | | var event: Io.Event = .unset; |
| 8929 | var num_completed: std.atomic.Value(u32) = .init(0); |
| 8559 | 8930 | |
| 8560 | | for (futures, 0..) |future, i| { |
| 8561 | | const closure: *AsyncClosure = @ptrCast(@alignCast(future)); |
| 8562 | | if (@atomicRmw(?*Io.Event, &closure.select_condition, .Xchg, &event, .seq_cst) == AsyncClosure.done_event) { |
| 8563 | | for (futures[0..i]) |cleanup_future| { |
| 8564 | | const cleanup_closure: *AsyncClosure = @ptrCast(@alignCast(cleanup_future)); |
| 8565 | | if (@atomicRmw(?*Io.Event, &cleanup_closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_event) { |
| 8566 | | cleanup_closure.event.waitUncancelable(ioBasic(t)); // Ensure no reference to our stack-allocated event. |
| 8567 | | } |
| 8568 | | } |
| 8569 | | return i; |
| 8931 | for (futures, 0..) |any_future, i| { |
| 8932 | const future: *Future = @ptrCast(@alignCast(any_future)); |
| 8933 | future.awaiter = &num_completed; |
| 8934 | const old_status = future.status.fetchOr( |
| 8935 | .{ .tag = .pending_awaited, .thread = .null }, |
| 8936 | .release, // release `future.awaiter` |
| 8937 | ); |
| 8938 | switch (old_status.tag) { |
| 8939 | .pending => {}, |
| 8940 | .pending_awaited => unreachable, // `await` raced with `select` |
| 8941 | .pending_canceled => unreachable, // `cancel` raced with `select` |
| 8942 | .done => { |
| 8943 | future.status.store(old_status, .monotonic); |
| 8944 | _ = finishSelect(&num_completed, futures[0..i]); |
| 8945 | return i; |
| 8946 | }, |
| 8570 | 8947 | } |
| 8571 | 8948 | } |
| 8572 | 8949 | |
| 8573 | | try event.wait(ioBasic(t)); |
| 8950 | errdefer _ = finishSelect(&num_completed, futures); |
| 8951 | const thread: *Thread = .getCurrent(t); |
| 8574 | 8952 | |
| 8575 | | var result: ?usize = null; |
| 8576 | | for (futures, 0..) |future, i| { |
| 8577 | | const closure: *AsyncClosure = @ptrCast(@alignCast(future)); |
| 8578 | | if (@atomicRmw(?*Io.Event, &closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_event) { |
| 8579 | | closure.event.waitUncancelable(ioBasic(t)); // Ensure no reference to our stack-allocated event. |
| 8580 | | if (result == null) result = i; // In case multiple are ready, return first. |
| 8581 | | } |
| 8953 | while (true) { |
| 8954 | const n = num_completed.load(.acquire); |
| 8955 | if (n > 0) break; |
| 8956 | assert(n < futures.len); |
| 8957 | try thread.futexWait(&num_completed.raw, n); |
| 8958 | } |
| 8959 | return finishSelect(&num_completed, futures).?; |
| 8960 | } |
| 8961 | fn finishSelect( |
| 8962 | num_completed: *std.atomic.Value(u32), |
| 8963 | futures: []const *Io.AnyFuture, |
| 8964 | ) ?usize { |
| 8965 | var completed_index: ?usize = null; |
| 8966 | var expect_completed: u32 = 0; |
| 8967 | for (futures, 0..) |any_future, i| { |
| 8968 | const future: *Future = @ptrCast(@alignCast(any_future)); |
| 8969 | // This operation will convert `.pending_awaited` to `.pending`, or leave `.done` untouched. |
| 8970 | switch (future.status.fetchAnd( |
| 8971 | .{ .tag = @enumFromInt(0b10), .thread = .all_ones }, |
| 8972 | .monotonic, |
| 8973 | ).tag) { |
| 8974 | .pending_awaited => {}, |
| 8975 | .pending => unreachable, |
| 8976 | .pending_canceled => unreachable, |
| 8977 | .done => { |
| 8978 | expect_completed += 1; |
| 8979 | completed_index = i; |
| 8980 | }, |
| 8981 | } |
| 8982 | } |
| 8983 | // If any future has just finished, wait for it to signal `num_completed` to avoid dangling |
| 8984 | // references to stack memory. |
| 8985 | while (true) { |
| 8986 | const n = num_completed.load(.acquire); |
| 8987 | if (n == expect_completed) break; |
| 8988 | assert(n < expect_completed); |
| 8989 | Thread.futexWaitUncancelable(&num_completed.raw, n); |
| 8582 | 8990 | } |
| 8583 | | return result.?; |
| 8991 | return completed_index; |
| 8584 | 8992 | } |
| 8585 | 8993 | |
| 8586 | 8994 | fn netListenIpPosix( |
Matthew Lugg <mlugg@mlugg.co.uk> 2025-12-23 20:54:50+00:00
Commit is signed but in an unrecognized format.
lib