Commit 7369008d8ca94fe48d9408d03851a4a17f7e1b73 | zig

author	Jacob Young <jacobly@ziglang.org> 2026-02-10 11:05:35-05:00
committer	Andrew Kelley <andrew@ziglang.org> 2026-02-10 23:45:48+01:00
log	7369008d8ca94fe48d9408d03851a4a17f7e1b73
tree	b64ea36b859f86b0120fe9d19d011deb57ce62bd
parent	e314dadb015a913a2cdc12298986e9181305d8db

Io.IoUring: implement some thread pool options and other cleanup

2 files changed, 666 insertions(+), 478 deletions(-)

lib
- std
  - Io
    IoUring.zig
    Threaded.zig

lib/std/Io/IoUring.zig+663-474

...	...	@@ -59,7 +59,9 @@ backing_allocator: Allocator,
59	59	main_fiber_buffer: [
60	60	std.mem.alignForward(usize, @sizeOf(Fiber), @alignOf(Completion)) + @sizeOf(Completion)
61	61	]u8 align(@max(@alignOf(Fiber), @alignOf(Completion))),
	62	log2_ring_entries: u4,
62	63	threads: Thread.List,
	64	sync_limit: ?Io.Semaphore,
63	65
64	66	stderr_mutex: Io.Mutex,
65	67	stderr_writer: File.Writer = .{
...	...	@@ -84,10 +86,9 @@ csprng: Csprng,
84	86	/// Empirically saw glibc complain about 256KB.
85	87	const idle_stack_size = 512 * 1024;
86	88
87		const max_idle_search = 4;
88		const max_steal_ready_search = 4;
89
90		const io_uring_entries = 64;
	89	const max_idle_search = 1;
	90	const max_steal_ready_search = 2;
	91	const max_steal_free_search = 4;
91	92
92	93	const Thread = struct {
93	94	required_align: void align(4),
...	...	@@ -95,9 +96,11 @@ const Thread = struct {
95	96	idle_context: Context,
96	97	current_context: *Context,
97	98	ready_queue: ?*Fiber,
	99	free_queue: ?*Fiber,
98	100	io_uring: IoUring,
99	101	idle_search_index: u32,
100	102	steal_ready_search_index: u32,
	103	steal_free_search_index: u32,
101	104	name_arena: if (tracy.enable) std.heap.ArenaAllocator.State else struct {},
102	105	csprng: Csprng,
103	106
...	...	@@ -107,6 +110,15 @@ const Thread = struct {
107	110	return self.?;
108	111	}
109	112
	113	fn deinit(thread: *Thread, gpa: Allocator) void {
	114	var next_fiber = thread.free_queue;
	115	while (next_fiber) \|free_fiber\| {
	116	next_fiber = free_fiber.status.free_next;
	117	gpa.free(free_fiber.allocatedSlice());
	118	}
	119	thread.io_uring.deinit();
	120	}
	121
110	122	fn currentFiber(thread: Thread) Fiber {
111	123	assert(thread.current_context != &thread.idle_context);
112	124	return @fieldParentPtr("context", thread.current_context);
...	...	@@ -144,6 +156,7 @@ const Fiber = struct {
144	156	status: union(enum) {
145	157	queue_next: ?*Fiber,
146	158	awaiting_group: Group,
	159	free_next: ?*Fiber,
147	160	},
148	161	cancel_status: CancelStatus,
149	162	cancel_protection: CancelProtection,
...	...	@@ -235,7 +248,7 @@ const Fiber = struct {
235	248	}
236	249	};
237	250
238		const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
	251	const finished: ?*Fiber = @ptrFromInt(@alignOf(Fiber));
239	252
240	253	const max_result_align: Alignment = .@"16";
241	254	const max_result_size = max_result_align.forward(512);
...	...	@@ -259,13 +272,49 @@ const Fiber = struct {
259	272	}
260	273
261	274	fn create(ev: Evented) error{OutOfMemory}!Fiber {
	275	const thread: *Thread = .current();
	276	if (@atomicRmw(?*Fiber, &thread.free_queue, .Xchg, finished, .acquire)) \|free_fiber\| {
	277	assert(free_fiber != finished);
	278	@atomicStore(?*Fiber, &thread.free_queue, free_fiber.status.free_next, .release);
	279	return free_fiber;
	280	}
	281	const active_threads = @atomicLoad(u32, &ev.threads.active, .acquire);
	282	for (0..@min(max_steal_free_search, active_threads)) \|_\| {
	283	defer thread.steal_free_search_index += 1;
	284	if (thread.steal_free_search_index == active_threads) thread.steal_free_search_index = 0;
	285	const steal_free_search_thread =
	286	&ev.threads.allocated[0..active_threads][thread.steal_free_search_index];
	287	if (steal_free_search_thread == thread) continue;
	288	const free_fiber =
	289	@atomicLoad(?*Fiber, &steal_free_search_thread.free_queue, .monotonic) orelse continue;
	290	if (free_fiber == finished) continue;
	291	if (@cmpxchgWeak(
	292	?*Fiber,
	293	&steal_free_search_thread.free_queue,
	294	free_fiber,
	295	null,
	296	.acquire,
	297	.monotonic,
	298	)) \|_\| continue;
	299	@atomicStore(?*Fiber, &thread.free_queue, free_fiber.status.free_next, .release);
	300	return free_fiber;
	301	}
	302	@atomicStore(?*Fiber, &thread.free_queue, null, .monotonic);
262	303	return @ptrCast(try ev.allocator().alignedAlloc(u8, .of(Fiber), allocation_size));
263	304	}
264	305
265		fn destroy(fiber: *Fiber, gpa: std.mem.Allocator) void {
266		log.debug("destroying {*}", .{fiber});
	306	fn destroy(fiber: *Fiber) void {
	307	const thread: *Thread = .current();
267	308	assert(fiber.status.queue_next == null);
268		gpa.free(fiber.allocatedSlice());
	309	fiber.status = .{ .free_next = @atomicLoad(?*Fiber, &thread.free_queue, .acquire) };
	310	while (true) fiber.status.free_next = @cmpxchgWeak(
	311	?*Fiber,
	312	&thread.free_queue,
	313	fiber.status.free_next,
	314	fiber,
	315	.acq_rel,
	316	.acquire,
	317	) orelse break;
269	318	}
270	319
271	320	fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {
...	...	@@ -416,6 +465,62 @@ const CancelRegion = struct {
416	465	fn errno(cancel_region: *const CancelRegion) linux.E {
417	466	return cancel_region.completion().errno();
418	467	}
	468
	469	const Sync = struct {
	470	cancel_region: CancelRegion,
	471	fn init(ev: *Evented) Io.Cancelable!Sync {
	472	if (ev.sync_limit) \|*sync_limit\| try sync_limit.wait(ev.io());
	473	return .{ .cancel_region = .init() };
	474	}
	475	fn initBlocked(ev: *Evented) Sync {
	476	if (ev.sync_limit) \|*sync_limit\| sync_limit.waitUncancelable(ev.io());
	477	return .{ .cancel_region = .initBlocked() };
	478	}
	479	fn deinit(sync: Sync, ev: Evented) void {
	480	sync.cancel_region.deinit();
	481	if (ev.sync_limit) \|*sync_limit\| sync_limit.post(ev.io());
	482	}
	483
	484	const Maybe = union(enum) {
	485	cancel_region: CancelRegion,
	486	sync: Sync,
	487
	488	fn deinit(maybe: Maybe, ev: Evented) void {
	489	switch (maybe.*) {
	490	.cancel_region => \|*cancel_region\| cancel_region.deinit(),
	491	.sync => \|*sync\| sync.deinit(ev),
	492	}
	493	}
	494
	495	fn enterSync(maybe: Maybe, ev: Evented) Io.Cancelable!*Sync {
	496	switch (maybe.*) {
	497	.cancel_region => \|cancel_region\| {
	498	if (ev.sync_limit) \|*sync_limit\| try sync_limit.wait(ev.io());
	499	maybe.* = .{ .sync = .{ .cancel_region = cancel_region } };
	500	},
	501	.sync => {},
	502	}
	503	return &maybe.sync;
	504	}
	505
	506	fn leaveSync(maybe: Maybe, ev: Evented) void {
	507	switch (maybe.*) {
	508	.cancel_region => {},
	509	.sync => \|sync\| {
	510	if (ev.sync_limit) \|*sync_limit\| sync_limit.post(ev.io());
	511	maybe.* = .{ .cancel_region = sync.cancel_region };
	512	},
	513	}
	514	}
	515
	516	fn cancelRegion(maybe: Maybe) CancelRegion {
	517	return switch (maybe.*) {
	518	.cancel_region => \|*cancel_region\| cancel_region,
	519	.sync => \|*sync\| &sync.cancel_region,
	520	};
	521	}
	522	};
	523	};
419	524	};
420	525
421	526	const CachedFd = struct {
...	...	@@ -685,7 +790,7 @@ fn fileMemoryMapSetLength(
685	790	new_len: usize,
686	791	) File.MemoryMap.SetLengthError!void {
687	792	const ev: *Evented = @ptrCast(@alignCast(userdata));
688		_ = ev;
	793
689	794	const page_size = std.heap.pageSize();
690	795	const alignment: Alignment = .fromByteUnits(page_size);
691	796	const page_align = std.heap.page_size_min;
...	...	@@ -695,12 +800,12 @@ fn fileMemoryMapSetLength(
695	800	mm.memory.len = new_len;
696	801	return;
697	802	}
698		var cancel_region: CancelRegion = .init();
699		defer cancel_region.deinit();
700	803	const flags: linux.MREMAP = .{ .MAYMOVE = true };
701	804	const addr_hint: ?[*]const u8 = null;
	805	var sync: CancelRegion.Sync = try .init(ev);
	806	defer sync.deinit(ev);
702	807	const new_memory = while (true) {
703		try cancel_region.await(.nothing);
	808	try sync.cancel_region.await(.nothing);
704	809	const rc = linux.mremap(old_memory.ptr, old_memory.len, new_len, flags, addr_hint);
705	810	switch (linux.errno(rc)) {
706	811	.SUCCESS => break @as([*]align(page_align) u8, @ptrFromInt(rc))[0..new_len],
...	...	@@ -730,17 +835,28 @@ fn fileMemoryMapWrite(userdata: ?anyopaque, mm: File.MemoryMap) File.WritePosi
730	835	pub const InitOptions = struct {
731	836	backing_allocator_needs_mutex: bool = true,
732	837
	838	/// Maximum thread pool size (excluding the main thread).
	839	/// Defaults to one less than the number of logical CPU cores.
	840	thread_limit: ?usize = null,
	841	/// Maximum number of threads that may perform synchronous syscalls.
	842	sync_limit: Io.Limit = .unlimited,
	843
	844	log2_ring_entries: u4 = 3,
	845
733	846	/// Affects the following operations:
734	847	/// * `processExecutablePath` on OpenBSD and Haiku.
735	848	argv0: Argv0 = .empty,
736	849	/// Affects the following operations:
737	850	/// * `fileIsTty`
738	851	/// * `processSpawn`, `processSpawnPath`, `processReplace`, `processReplacePath`
739		environ: process.Environ,
	852	environ: process.Environ = .empty,
740	853	};
741	854
742	855	pub fn init(ev: *Evented, backing_allocator: Allocator, options: InitOptions) !void {
743		const threads_size = @max(std.Thread.getCpuCount() catch 1, 1) * @sizeOf(Thread);
	856	const threads_size = @sizeOf(Thread) * if (options.thread_limit) \|thread_limit\|
	857	1 + thread_limit
	858	else
	859	@max(std.Thread.getCpuCount() catch 1, 1);
744	860	const idle_stack_end_offset =
745	861	std.mem.alignForward(usize, threads_size + idle_stack_size, std.heap.page_size_max);
746	862	const allocated_slice = try backing_allocator.alignedAlloc(u8, .of(Thread), idle_stack_end_offset);
...	...	@@ -750,11 +866,13 @@ pub fn init(ev: *Evented, backing_allocator: Allocator, options: InitOptions) !v
750	866	.backing_allocator_mutex = .init,
751	867	.backing_allocator = backing_allocator,
752	868	.main_fiber_buffer = undefined,
	869	.log2_ring_entries = options.log2_ring_entries,
753	870	.threads = .{
754	871	.allocated = @ptrCast(allocated_slice[0..threads_size]),
755	872	.reserved = 1,
756	873	.active = 1,
757	874	},
	875	.sync_limit = if (options.sync_limit.toInt()) \|sync_limit\| .{ .permits = sync_limit } else null,
758	876
759	877	.stderr_mutex = .init,
760	878	.stderr_writer = .{
...	...	@@ -809,18 +927,18 @@ pub fn init(ev: *Evented, backing_allocator: Allocator, options: InitOptions) !v
809	927	},
810	928	.current_context = &main_fiber.context,
811	929	.ready_queue = null,
	930	.free_queue = null,
812	931	.io_uring = try .init(
813		io_uring_entries,
	932	@as(u16, 1) << ev.log2_ring_entries,
814	933	linux.IORING_SETUP_COOP_TASKRUN \| linux.IORING_SETUP_SINGLE_ISSUER,
815	934	),
816	935	.idle_search_index = 1,
817	936	.steal_ready_search_index = 1,
	937	.steal_free_search_index = 1,
818	938	.name_arena = .{},
819	939	.csprng = .uninitialized,
820	940	};
821	941	errdefer main_thread.io_uring.deinit();
822		log.debug("created main idle {*}", .{&main_thread.idle_context});
823		log.debug("created main {*}", .{main_fiber});
824	942	if (tracy.enable) tracy.fiberEnter(main_fiber.name);
825	943	}
826	944
...	...	@@ -831,7 +949,7 @@ pub fn deinit(ev: *Evented) void {
831	949	assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async
832	950	}
833	951	ev.yield(null, .exit);
834		ev.threads.allocated[0].io_uring.deinit();
	952	ev.threads.allocated[0].deinit(ev.allocator());
835	953	ev.null_fd.close();
836	954	ev.random_fd.close();
837	955	const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @ptrCast(@alignCast(ev.threads.allocated.ptr));
...	...	@@ -848,6 +966,7 @@ pub fn deinit(ev: *Evented) void {
848	966
849	967	fn findReadyFiber(ev: Evented, thread: Thread) ?*Fiber {
850	968	if (@atomicRmw(?*Fiber, &thread.ready_queue, .Xchg, Fiber.finished, .acquire)) \|ready_fiber\| {
	969	assert(ready_fiber != Fiber.finished);
851	970	@atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.status.queue_next, .release);
852	971	ready_fiber.status.queue_next = null;
853	972	return ready_fiber;
...	...	@@ -860,7 +979,7 @@ fn findReadyFiber(ev: Evented, thread: Thread) ?*Fiber {
860	979	&ev.threads.allocated[0..active_threads][thread.steal_ready_search_index];
861	980	if (steal_ready_search_thread == thread) continue;
862	981	const ready_fiber =
863		@atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
	982	@atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .monotonic) orelse continue;
864	983	if (ready_fiber == Fiber.finished) continue;
865	984	if (@cmpxchgWeak(
866	985	?*Fiber,
...	...	@@ -892,19 +1011,10 @@ fn yield(ev: Evented, maybe_ready_fiber: ?Fiber, pending_task: SwitchMessage.P
892	1011	},
893	1012	.pending_task = pending_task,
894	1013	};
895		log.debug("switching from {} to {}", .{ message.contexts.prev, message.contexts.ready });
896	1014	contextSwitch(&message).handle(ev);
897	1015	}
898	1016
899	1017	fn schedule(ev: Evented, thread: Thread, ready_queue: Fiber.Queue) bool {
900		{
901		var fiber = ready_queue.head;
902		while (true) {
903		log.debug("scheduling {*}", .{fiber});
904		fiber = fiber.status.queue_next orelse break;
905		}
906		assert(fiber == ready_queue.tail);
907		}
908	1018	// shared fields of previous `Thread` must be initialized before later ones are marked as active
909	1019	const new_thread_index = @atomicLoad(u32, &ev.threads.active, .acquire);
910	1020	for (0..@min(max_idle_search, new_thread_index)) \|_\| {
...	...	@@ -963,7 +1073,8 @@ fn schedule(ev: Evented, thread: Thread, ready_queue: Fiber.Queue) bool {
963	1073	.idle_context = undefined,
964	1074	.current_context = &new_thread.idle_context,
965	1075	.ready_queue = ready_queue.head,
966		.io_uring = IoUring.init_params(io_uring_entries, &params) catch \|err\| {
	1076	.free_queue = null,
	1077	.io_uring = IoUring.init_params(@as(u16, 1) << ev.log2_ring_entries, &params) catch \|err\| {
967	1078	@atomicStore(u32, &ev.threads.reserved, new_thread_index, .release);
968	1079	// no more access to `thread` after giving up reservation
969	1080	log.warn("unable to create worker thread due to io_uring init failure: {s}", .{
...	...	@@ -973,6 +1084,7 @@ fn schedule(ev: Evented, thread: Thread, ready_queue: Fiber.Queue) bool {
973	1084	},
974	1085	.idle_search_index = 0,
975	1086	.steal_ready_search_index = 0,
	1087	.steal_free_search_index = 0,
976	1088	.name_arena = .{},
977	1089	.csprng = .uninitialized,
978	1090	};
...	...	@@ -991,14 +1103,14 @@ fn schedule(ev: Evented, thread: Thread, ready_queue: Fiber.Queue) bool {
991	1103	return false;
992	1104	}
993	1105	// nobody wanted it, so just queue it on ourselves
994		while (@cmpxchgWeak(
	1106	while (true) ready_queue.tail.status.queue_next = @cmpxchgWeak(
995	1107	?*Fiber,
996	1108	&thread.ready_queue,
997	1109	ready_queue.tail.status.queue_next,
998	1110	ready_queue.head,
999	1111	.acq_rel,
1000	1112	.acquire,
1001		)) \|old_head\| ready_queue.tail.status.queue_next = old_head;
	1113	) orelse break;
1002	1114	return false;
1003	1115	}
1004	1116
...	...	@@ -1015,8 +1127,7 @@ fn mainIdle(
1015	1127	fn threadEntry(ev: *Evented, index: u32) void {
1016	1128	const thread: *Thread = &ev.threads.allocated[index];
1017	1129	Thread.self = thread;
1018		defer thread.io_uring.deinit();
1019		log.debug("created thread idle {*}", .{&thread.idle_context});
	1130	defer thread.deinit(ev.allocator());
1020	1131	switch (linux.errno(linux.io_uring_register(thread.io_uring.fd, .REGISTER_ENABLE_RINGS, null, 0))) {
1021	1132	.SUCCESS => ev.idle(thread),
1022	1133	else => \|err\| @panic(@tagName(err)),
...	...	@@ -1054,103 +1165,107 @@ fn idle(ev: Evented, thread: Thread) void {
1054	1165	error.SignalInterrupt => {},
1055	1166	else => \|e\| @panic(@errorName(e)),
1056	1167	};
1057		var cqes_buffer: [io_uring_entries]linux.io_uring_cqe = undefined;
1058	1168	var maybe_ready_queue: ?Fiber.Queue = null;
1059		for (cqes_buffer[0 .. thread.io_uring.copy_cqes(&cqes_buffer, 0) catch \|err\| switch (err) {
1060		error.SignalInterrupt => 0,
1061		else => \|e\| @panic(@errorName(e)),
1062		}]) \|cqe\| if (cqe.flags & linux.IORING_CQE_F_SKIP == 0) switch (@as(
1063		Completion.UserData,
1064		@enumFromInt(cqe.user_data),
1065		)) {
1066		.unused => unreachable, // bad submission queued?
1067		.wakeup => {},
1068		.futex_wake => switch (Completion.errno(.{ .result = cqe.res, .flags = cqe.flags })) {
1069		.SUCCESS => recoverableOsBugDetected(), // success is skipped
1070		.INVAL => {}, // invalid futex_wait() on ptr done elsewhere
1071		.INTR, .CANCELED => recoverableOsBugDetected(), // `Completion.UserData.futex_wake` is not cancelable
1072		.FAULT => {}, // pointer became invalid while doing the wake
1073		else => recoverableOsBugDetected(), // deadlock due to operating system bug
1074		},
1075		.cleanup => @panic("failed to notify other threads that we are exiting"),
1076		.exit => {
1077		assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
1078		return;
1079		},
1080		_ => if (@as(?*Fiber, ready_fiber: switch (@as(u2, @truncate(cqe.user_data))) {
1081		0b00 => {
1082		const ready_fiber: *Fiber = @ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
1083		ready_fiber.resultPointer(Completion).* = .{
1084		.result = cqe.res,
1085		.flags = cqe.flags,
1086		};
1087		break :ready_fiber ready_fiber;
1088		},
1089		0b01 => {
1090		thread.enqueue().* = .{
1091		.opcode = .ASYNC_CANCEL,
1092		.flags = linux.IOSQE_CQE_SKIP_SUCCESS,
1093		.ioprio = 0,
1094		.fd = 0,
1095		.off = 0,
1096		.addr = cqe.user_data & ~@as(usize, 0b11),
1097		.len = 0,
1098		.rw_flags = 0,
1099		.user_data = @intFromEnum(Completion.UserData.wakeup),
1100		.buf_index = 0,
1101		.personality = 0,
1102		.splice_fd_in = 0,
1103		.addr3 = 0,
1104		.resv = 0,
1105		};
1106		break :ready_fiber null;
	1169	while (true) {
	1170	var cqes_buffer: [1 << 8]linux.io_uring_cqe = undefined;
	1171	const cqes = cqes_buffer[0 .. thread.io_uring.copy_cqes(&cqes_buffer, 0) catch \|err\| switch (err) {
	1172	error.SignalInterrupt => 0,
	1173	else => \|e\| @panic(@errorName(e)),
	1174	}];
	1175	if (cqes.len == 0) break;
	1176	for (cqes) \|cqe\| if (cqe.flags & linux.IORING_CQE_F_SKIP == 0) switch (@as(
	1177	Completion.UserData,
	1178	@enumFromInt(cqe.user_data),
	1179	)) {
	1180	.unused => unreachable, // bad submission queued?
	1181	.wakeup => {},
	1182	.futex_wake => switch (Completion.errno(.{ .result = cqe.res, .flags = cqe.flags })) {
	1183	.SUCCESS => recoverableOsBugDetected(), // success is skipped
	1184	.INVAL => {}, // invalid futex_wait() on ptr done elsewhere
	1185	.INTR, .CANCELED => recoverableOsBugDetected(), // `Completion.UserData.futex_wake` is not cancelable
	1186	.FAULT => {}, // pointer became invalid while doing the wake
	1187	else => recoverableOsBugDetected(), // deadlock due to operating system bug
1107	1188	},
1108		0b10 => {
1109		const context: *Io.Operation.Storage.Pending.Context =
1110		@ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
1111		const batch: *Io.Batch = @ptrFromInt(context[0]);
1112		var next: usize = 0b00;
1113		context[0..3].* = .{ next, @as(u32, @bitCast(cqe.res)), cqe.flags };
1114		while (true) {
1115		next = @cmpxchgWeak(
1116		usize,
1117		@as(*usize, @ptrCast(&batch.context)),
1118		next,
1119		cqe.user_data,
1120		.release,
1121		.acquire,
1122		) orelse break;
1123		context[0] = next;
1124		}
1125		break :ready_fiber switch (@as(u2, @truncate(next))) {
1126		0b00, 0b01 => @ptrFromInt(next & ~@as(usize, 0b11)),
1127		0b10, 0b11 => null,
1128		};
	1189	.cleanup => @panic("failed to notify other threads that we are exiting"),
	1190	.exit => {
	1191	assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
	1192	return;
1129	1193	},
1130		0b11 => switch (Completion.errno(.{ .result = cqe.res, .flags = cqe.flags })) {
1131		.SUCCESS => unreachable, // no event count specified
1132		.TIME => {
1133		const context: *usize = @ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
1134		const fiber = @atomicRmw(usize, context, .Add, 0b01, .acquire);
1135		break :ready_fiber switch (@as(u2, @truncate(fiber))) {
1136		else => unreachable, // timeout completed multiple times
1137		0b00 => @ptrFromInt(fiber & ~@as(usize, 0b11)),
1138		0b10 => null,
	1194	_ => if (@as(?*Fiber, ready_fiber: switch (@as(u2, @truncate(cqe.user_data))) {
	1195	0b00 => {
	1196	const ready_fiber: *Fiber = @ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
	1197	ready_fiber.resultPointer(Completion).* = .{
	1198	.result = cqe.res,
	1199	.flags = cqe.flags,
	1200	};
	1201	break :ready_fiber ready_fiber;
	1202	},
	1203	0b01 => {
	1204	thread.enqueue().* = .{
	1205	.opcode = .ASYNC_CANCEL,
	1206	.flags = linux.IOSQE_CQE_SKIP_SUCCESS,
	1207	.ioprio = 0,
	1208	.fd = 0,
	1209	.off = 0,
	1210	.addr = cqe.user_data & ~@as(usize, 0b11),
	1211	.len = 0,
	1212	.rw_flags = 0,
	1213	.user_data = @intFromEnum(Completion.UserData.wakeup),
	1214	.buf_index = 0,
	1215	.personality = 0,
	1216	.splice_fd_in = 0,
	1217	.addr3 = 0,
	1218	.resv = 0,
1139	1219	};
	1220	break :ready_fiber null;
1140	1221	},
1141		.CANCELED => null, // user data may have been invalidated
1142		else => \|err\| unexpectedErrno(err) catch null,
	1222	0b10 => {
	1223	const context: *Io.Operation.Storage.Pending.Context =
	1224	@ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
	1225	const batch: *Io.Batch = @ptrFromInt(context[0]);
	1226	var next: usize = 0b00;
	1227	context[0..3].* = .{ next, @as(u32, @bitCast(cqe.res)), cqe.flags };
	1228	while (true) {
	1229	next = @cmpxchgWeak(
	1230	usize,
	1231	@as(*usize, @ptrCast(&batch.context)),
	1232	next,
	1233	cqe.user_data,
	1234	.release,
	1235	.acquire,
	1236	) orelse break;
	1237	context[0] = next;
	1238	}
	1239	break :ready_fiber switch (@as(u2, @truncate(next))) {
	1240	0b00, 0b01 => @ptrFromInt(next & ~@as(usize, 0b11)),
	1241	0b10, 0b11 => null,
	1242	};
	1243	},
	1244	0b11 => switch (Completion.errno(.{ .result = cqe.res, .flags = cqe.flags })) {
	1245	.SUCCESS => unreachable, // no event count specified
	1246	.TIME => {
	1247	const context: *usize = @ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
	1248	const fiber = @atomicRmw(usize, context, .Add, 0b01, .acquire);
	1249	break :ready_fiber switch (@as(u2, @truncate(fiber))) {
	1250	else => unreachable, // timeout completed multiple times
	1251	0b00 => @ptrFromInt(fiber & ~@as(usize, 0b11)),
	1252	0b10 => null,
	1253	};
	1254	},
	1255	.CANCELED => null, // user data may have been invalidated
	1256	else => \|err\| unexpectedErrno(err) catch null,
	1257	},
	1258	})) \|ready_fiber\| {
	1259	assert(ready_fiber.status.queue_next == null);
	1260	if (maybe_ready_fiber == null) {
	1261	maybe_ready_fiber = ready_fiber;
	1262	} else if (maybe_ready_queue) \|*ready_queue\| {
	1263	ready_queue.tail.status.queue_next = ready_fiber;
	1264	ready_queue.tail = ready_fiber;
	1265	} else maybe_ready_queue = .{ .head = ready_fiber, .tail = ready_fiber };
1143	1266	},
1144		})) \|ready_fiber\| {
1145		assert(ready_fiber.status.queue_next == null);
1146		if (maybe_ready_fiber == null) {
1147		maybe_ready_fiber = ready_fiber;
1148		} else if (maybe_ready_queue) \|*ready_queue\| {
1149		ready_queue.tail.status.queue_next = ready_fiber;
1150		ready_queue.tail = ready_fiber;
1151		} else maybe_ready_queue = .{ .head = ready_fiber, .tail = ready_fiber };
1152		},
1153		};
	1267	};
	1268	}
1154	1269	if (maybe_ready_queue) \|ready_queue\| _ = ev.schedule(thread, ready_queue);
1155	1270	}
1156	1271	}
...	...	@@ -1220,8 +1335,7 @@ const SwitchMessage = struct {
1220	1335	},
1221	1336	.destroy => {
1222	1337	const fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
1223		fiber.destroy(ev.backing_allocator);
1224		ev.backing_allocator_mutex.unlock(ev.io());
	1338	fiber.destroy();
1225	1339	},
1226	1340	.exit => for (
1227	1341	ev.threads.allocated[0..@atomicLoad(u32, &ev.threads.active, .acquire)],
...	...	@@ -1295,7 +1409,6 @@ inline fn contextSwitch(message: const SwitchMessage) const SwitchMessage {
1295	1409	.x15 = true,
1296	1410	.x16 = true,
1297	1411	.x17 = true,
1298		.x18 = true,
1299	1412	.x19 = true,
1300	1413	.x20 = true,
1301	1414	.x21 = true,
...	...	@@ -1497,7 +1610,6 @@ const AsyncClosure = struct {
1497	1610	) callconv(.withStackAlign(.c, @alignOf(AsyncClosure))) noreturn {
1498	1611	message.handle(closure.ev);
1499	1612	const fiber = closure.fiber;
1500		log.debug("{*} performing async", .{fiber});
1501	1613	closure.start(closure.contextPointer(), fiber.resultBytes(closure.result_align));
1502	1614	closure.ev.yield(
1503	1615	if (@atomicRmw(?*Fiber, &fiber.link.awaiter, .Xchg, Fiber.finished, .acq_rel)) \|awaiter\|
...	...	@@ -1542,7 +1654,6 @@ fn concurrent(
1542	1654	const fiber = Fiber.create(ev) catch \|err\| switch (err) {
1543	1655	error.OutOfMemory => return error.ConcurrencyUnavailable,
1544	1656	};
1545		log.debug("allocated {*}", .{fiber});
1546	1657
1547	1658	const closure: *AsyncClosure = .fromFiber(fiber);
1548	1659	fiber.* = .{
...	...	@@ -1608,7 +1719,7 @@ fn await(
1608	1719	assert(awaiter == fiber); // spurious wakeup
1609	1720	}
1610	1721	@memcpy(result, future_fiber.resultBytes(result_alignment));
1611		future_fiber.destroy(ev.allocator());
	1722	future_fiber.destroy();
1612	1723	}
1613	1724
1614	1725	fn cancel(
...	...	@@ -1866,7 +1977,6 @@ const Group = struct {
1866	1977	) callconv(.withStackAlign(.c, @alignOf(Group.AsyncClosure))) noreturn {
1867	1978	message.handle(closure.ev);
1868	1979	assert(closure.fiber.status.queue_next == null);
1869		log.debug("{*} performing group async", .{closure.fiber});
1870	1980	const result = closure.start(closure.contextPointer());
1871	1981	const ev = closure.ev;
1872	1982	const group = closure.group;
...	...	@@ -1877,9 +1987,7 @@ const Group = struct {
1877	1987	} else \|err\| switch (err) {
1878	1988	error.Canceled => assert(cancel_acknowledged), // group task returned `error.Canceled` but was never canceled
1879	1989	}
1880		const awaiter = group.removeFiber(ev, fiber);
1881		ev.backing_allocator_mutex.lockUncancelable(ev.io());
1882		ev.yield(awaiter, .destroy);
	1990	ev.yield(group.removeFiber(ev, fiber), .destroy);
1883	1991	unreachable; // switched to dead fiber
1884	1992	}
1885	1993	};
...	...	@@ -1930,7 +2038,6 @@ fn groupConcurrent(
1930	2038	const fiber = Fiber.create(ev) catch \|err\| switch (err) {
1931	2039	error.OutOfMemory => return error.ConcurrencyUnavailable,
1932	2040	};
1933		log.debug("allocated {*}", .{fiber});
1934	2041
1935	2042	const closure: *Group.AsyncClosure = .fromFiber(fiber);
1936	2043	fiber.* = .{
...	...	@@ -2080,7 +2187,6 @@ fn futexWait(
2080	2187	timeout: Io.Timeout,
2081	2188	) Io.Cancelable!void {
2082	2189	const ev: *Evented = @ptrCast(@alignCast(userdata));
2083		if (builtin.single_threaded) unreachable; // Deadlock.
2084	2190	const timespec: ?linux.kernel_timespec, const clock: Io.Clock, const timeout_flags: u32 = timespec: switch (timeout) {
2085	2191	.none => .{
2086	2192	null,
...	...	@@ -2163,7 +2269,6 @@ fn futexWait(
2163	2269
2164	2270	fn futexWaitUncancelable(userdata: ?anyopaque, ptr: const u32, expected: u32) void {
2165	2271	const ev: *Evented = @ptrCast(@alignCast(userdata));
2166		if (builtin.single_threaded) unreachable; // Deadlock.
2167	2272	var cancel_region: CancelRegion = .initBlocked();
2168	2273	defer cancel_region.deinit();
2169	2274	const thread = cancel_region.awaitIoUring() catch \|err\| switch (err) {
...	...	@@ -2199,7 +2304,6 @@ fn futexWaitUncancelable(userdata: ?anyopaque, ptr: const u32, expected: u32)
2199	2304	fn futexWake(userdata: ?anyopaque, ptr: const u32, max_waiters: u32) void {
2200	2305	const ev: *Evented = @ptrCast(@alignCast(userdata));
2201	2306	_ = ev;
2202		if (builtin.single_threaded) unreachable; // Nothing to wake up.
2203	2307	const thread: *Thread = .current();
2204	2308	thread.enqueue().* = .{
2205	2309	.opcode = .FUTEX_WAKE,
...	...	@@ -2222,24 +2326,43 @@ fn futexWake(userdata: ?anyopaque, ptr: const u32, max_waiters: u32) void {
2222	2326
2223	2327	fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Operation.Result {
2224	2328	const ev: *Evented = @ptrCast(@alignCast(userdata));
2225		switch (operation) {
2226		.file_read_streaming => \|o\| return .{
2227		.file_read_streaming = ev.fileReadStreaming(o.file, o.data) catch \|err\| switch (err) {
	2329	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	2330	defer maybe_sync.deinit(ev);
	2331	return switch (operation) {
	2332	.file_read_streaming => \|o\| .{
	2333	.file_read_streaming = ev.fileReadStreaming(
	2334	&maybe_sync.cancel_region,
	2335	o.file,
	2336	o.data,
	2337	) catch \|err\| switch (err) {
2228	2338	error.Canceled => \|e\| return e,
2229	2339	else => \|e\| e,
2230	2340	},
2231	2341	},
2232		.file_write_streaming => \|o\| return .{
2233		.file_write_streaming = ev.fileWriteStreaming(o.file, o.header, o.data, o.splat) catch \|err\| switch (err) {
	2342	.file_write_streaming => \|o\| .{
	2343	.file_write_streaming = ev.fileWriteStreaming(
	2344	&maybe_sync.cancel_region,
	2345	o.file,
	2346	o.header,
	2347	o.data,
	2348	o.splat,
	2349	) catch \|err\| switch (err) {
2234	2350	error.Canceled => \|e\| return e,
2235	2351	else => \|e\| e,
2236	2352	},
2237	2353	},
2238		.device_io_control => \|*o\| return .{ .device_io_control = try deviceIoControl(o) },
2239		}
	2354	.device_io_control => \|o\| .{
	2355	.device_io_control = try ev.deviceIoControl(try maybe_sync.enterSync(ev), o),
	2356	},
	2357	};
2240	2358	}
2241	2359
2242		fn fileReadStreaming(ev: *Evented, file: File, data: []const []u8) File.Reader.Error!usize {
	2360	fn fileReadStreaming(
	2361	ev: *Evented,
	2362	cancel_region: *CancelRegion,
	2363	file: File,
	2364	data: []const []u8,
	2365	) File.ReadStreamingError!usize {
2243	2366	var iovecs_buffer: [max_iovecs_len]iovec = undefined;
2244	2367	var i: usize = 0;
2245	2368	for (data) \|buf\| {
...	...	@@ -2252,13 +2375,13 @@ fn fileReadStreaming(ev: *Evented, file: File, data: []const []u8) File.Reader.E
2252	2375	const dest = iovecs_buffer[0..i];
2253	2376	assert(dest[0].len > 0);
2254	2377
2255		var cancel_region: CancelRegion = .init();
2256		defer cancel_region.deinit();
2257		return ev.preadv(&cancel_region, file.handle, dest, null);
	2378	const n = try ev.preadv(cancel_region, file.handle, dest, null);
	2379	return if (n == 0) error.EndOfStream else n;
2258	2380	}
2259	2381
2260	2382	fn fileWriteStreaming(
2261	2383	ev: *Evented,
	2384	cancel_region: *CancelRegion,
2262	2385	file: File,
2263	2386	header: []const u8,
2264	2387	data: []const []const u8,
...	...	@@ -2294,17 +2417,17 @@ fn fileWriteStreaming(
2294	2417	},
2295	2418	},
2296	2419	};
2297
2298		var cancel_region: CancelRegion = .init();
2299		defer cancel_region.deinit();
2300		return ev.pwritev(&cancel_region, file.handle, iovecs[0..iovlen], null);
	2420	return ev.pwritev(cancel_region, file.handle, iovecs[0..iovlen], null);
2301	2421	}
2302	2422
2303		fn deviceIoControl(o: *const Io.Operation.DeviceIoControl) Io.Cancelable!i32 {
2304		var cancel_region: CancelRegion = .init();
2305		defer cancel_region.deinit();
	2423	fn deviceIoControl(
	2424	ev: *Evented,
	2425	sync: *CancelRegion.Sync,
	2426	o: Io.Operation.DeviceIoControl,
	2427	) Io.Cancelable!i32 {
	2428	_ = ev;
2306	2429	while (true) {
2307		try cancel_region.await(.nothing);
	2430	try sync.cancel_region.await(.nothing);
2308	2431	const rc = linux.ioctl(o.file.handle, @bitCast(o.code), @intFromPtr(o.arg));
2309	2432	switch (linux.errno(rc)) {
2310	2433	.SUCCESS => return @bitCast(@as(u32, @truncate(rc))),
...	...	@@ -2316,12 +2439,13 @@ fn deviceIoControl(o: *const Io.Operation.DeviceIoControl) Io.Cancelable!i32 {
2316	2439
2317	2440	fn batchAwaitAsync(userdata: ?anyopaque, batch: Io.Batch) Io.Cancelable!void {
2318	2441	const ev: *Evented = @ptrCast(@alignCast(userdata));
2319		var cancel_region: CancelRegion = .init();
2320		defer cancel_region.deinit();
2321		batchDrainSubmitted(batch, &cancel_region, false) catch \|err\| switch (err) {
	2442	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	2443	defer maybe_sync.deinit(ev);
	2444	ev.batchDrainSubmitted(&maybe_sync, batch, false) catch \|err\| switch (err) {
2322	2445	error.ConcurrencyUnavailable => unreachable, // passed concurrency=false
2323	2446	else => \|e\| return e,
2324	2447	};
	2448	maybe_sync.leaveSync(ev);
2325	2449	while (true) {
2326	2450	batchDrainReady(batch) catch \|err\| switch (err) {
2327	2451	error.Timeout => unreachable, // no timeout
...	...	@@ -2337,9 +2461,10 @@ fn batchAwaitConcurrent(
2337	2461	timeout: Io.Timeout,
2338	2462	) Io.Batch.AwaitConcurrentError!void {
2339	2463	const ev: *Evented = @ptrCast(@alignCast(userdata));
2340		var cancel_region: CancelRegion = .init();
2341		defer cancel_region.deinit();
2342		try batchDrainSubmitted(batch, &cancel_region, true);
	2464	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	2465	defer maybe_sync.deinit(ev);
	2466	try ev.batchDrainSubmitted(&maybe_sync, batch, true);
	2467	maybe_sync.leaveSync(ev);
2343	2468	const timespec: linux.kernel_timespec, const clock: Io.Clock, const timeout_flags: u32 = while (true) {
2344	2469	batchDrainReady(batch) catch \|err\| switch (err) {
2345	2470	error.Timeout => unreachable, // no timeout
...	...	@@ -2372,7 +2497,7 @@ fn batchAwaitConcurrent(
2372	2497	}
2373	2498	};
2374	2499	{
2375		const thread = try cancel_region.awaitIoUring();
	2500	const thread = try maybe_sync.cancel_region.awaitIoUring();
2376	2501	thread.enqueue().* = .{
2377	2502	.opcode = .TIMEOUT,
2378	2503	.flags = 0,
...	...	@@ -2401,7 +2526,7 @@ fn batchAwaitConcurrent(
2401	2526	};
2402	2527	if (batch.completed.head == .none) continue;
2403	2528	}
2404		const thread = try cancel_region.awaitIoUring();
	2529	const thread = try maybe_sync.cancel_region.awaitIoUring();
2405	2530	thread.enqueue().* = .{
2406	2531	.opcode = .TIMEOUT_REMOVE,
2407	2532	.flags = 0,
...	...	@@ -2411,7 +2536,7 @@ fn batchAwaitConcurrent(
2411	2536	.addr = @intFromPtr(&batch.context) \| 0b11,
2412	2537	.len = 0,
2413	2538	.rw_flags = 0,
2414		.user_data = @intFromPtr(cancel_region.fiber),
	2539	.user_data = @intFromPtr(maybe_sync.cancel_region.fiber),
2415	2540	.buf_index = 0,
2416	2541	.personality = 0,
2417	2542	.splice_fd_in = 0,
...	...	@@ -2419,7 +2544,7 @@ fn batchAwaitConcurrent(
2419	2544	.resv = 0,
2420	2545	};
2421	2546	ev.yield(null, .nothing);
2422		switch (cancel_region.errno()) {
	2547	switch (maybe_sync.cancel_region.errno()) {
2423	2548	.SUCCESS => return,
2424	2549	.BUSY, .NOENT => {},
2425	2550	else => \|err\| unexpectedErrno(err) catch {},
...	...	@@ -2434,22 +2559,23 @@ fn batchAwaitConcurrent(
2434	2559
2435	2560	/// If `concurrency` is false, `error.ConcurrencyUnavailable` is unreachable.
2436	2561	fn batchDrainSubmitted(
	2562	ev: *Evented,
	2563	maybe_sync: *CancelRegion.Sync.Maybe,
2437	2564	batch: *Io.Batch,
2438		cancel_region: *CancelRegion,
2439	2565	concurrency: bool,
2440	2566	) (Io.ConcurrentError \|\| Io.Cancelable)!void {
2441	2567	var index = batch.submitted.head;
2442	2568	if (index == .none) return;
2443	2569	errdefer batch.submitted.head = index;
2444		const thread = try cancel_region.awaitIoUring();
	2570	const thread = try maybe_sync.cancelRegion().awaitIoUring();
2445	2571	while (index != .none) {
2446	2572	const storage = &batch.storage[index.toIndex()];
2447	2573	const next_index = storage.submission.node.next;
2448		if (@as(?Io.Operation.Result, operation: switch (storage.submission.operation) {
	2574	if (@as(?Io.Operation.Result, result: switch (storage.submission.operation) {
2449	2575	.file_read_streaming => \|o\| {
2450	2576	const buffer = for (o.data) \|buffer\| {
2451	2577	if (buffer.len != 0) break buffer;
2452		} else break :operation .{ .file_read_streaming = 0 };
	2578	} else break :result .{ .file_read_streaming = 0 };
2453	2579	const fd = o.file.handle;
2454	2580	storage.* = .{ .pending = .{
2455	2581	.node = .{ .prev = batch.pending.tail, .next = .none },
...	...	@@ -2472,7 +2598,7 @@ fn batchDrainSubmitted(
2472	2598	.addr3 = 0,
2473	2599	.resv = 0,
2474	2600	};
2475		break :operation null;
	2601	break :result null;
2476	2602	},
2477	2603	.file_write_streaming => \|o\| {
2478	2604	const buffer = buffer: {
...	...	@@ -2481,7 +2607,7 @@ fn batchDrainSubmitted(
2481	2607	if (buffer.len != 0) break :buffer buffer;
2482	2608	}
2483	2609	if (o.splat > 0) break :buffer o.data[o.data.len - 1];
2484		break :operation .{ .file_write_streaming = 0 };
	2610	break :result .{ .file_write_streaming = 0 };
2485	2611	};
2486	2612	const fd = o.file.handle;
2487	2613	storage.* = .{ .pending = .{
...	...	@@ -2505,12 +2631,12 @@ fn batchDrainSubmitted(
2505	2631	.addr3 = 0,
2506	2632	.resv = 0,
2507	2633	};
2508		break :operation null;
	2634	break :result null;
2509	2635	},
2510	2636	.device_io_control => \|o\| if (concurrency)
2511	2637	return error.ConcurrencyUnavailable
2512	2638	else
2513		.{ .device_io_control = try deviceIoControl(&o) },
	2639	.{ .device_io_control = try ev.deviceIoControl(try maybe_sync.enterSync(ev), o) },
2514	2640	})) \|result\| {
2515	2641	switch (batch.completed.tail) {
2516	2642	.none => batch.completed.head = index,
...	...	@@ -2838,7 +2964,6 @@ fn dirAccess(
2838	2964	options: Dir.AccessOptions,
2839	2965	) Dir.AccessError!void {
2840	2966	const ev: *Evented = @ptrCast(@alignCast(userdata));
2841		_ = ev;
2842	2967
2843	2968	var path_buffer: [PATH_MAX]u8 = undefined;
2844	2969	const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
...	...	@@ -2849,10 +2974,10 @@ fn dirAccess(
2849	2974	@as(u32, if (options.execute) linux.X_OK else 0);
2850	2975	const flags: u32 = if (options.follow_symlinks) 0 else linux.AT.SYMLINK_NOFOLLOW;
2851	2976
2852		var cancel_region: CancelRegion = .init();
2853		defer cancel_region.deinit();
	2977	var sync: CancelRegion.Sync = try .init(ev);
	2978	defer sync.deinit(ev);
2854	2979	while (true) {
2855		try cancel_region.await(.nothing);
	2980	try sync.cancel_region.await(.nothing);
2856	2981	switch (linux.errno(linux.faccessat(dir.handle, sub_path_posix, mode, flags))) {
2857	2982	.SUCCESS => return,
2858	2983	.INTR => continue,
...	...	@@ -2885,21 +3010,21 @@ fn dirCreateFile(
2885	3010	var path_buffer: [PATH_MAX]u8 = undefined;
2886	3011	const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
2887	3012
2888		var cancel_region: CancelRegion = .init();
2889		defer cancel_region.deinit();
2890		const fd = try ev.openat(&cancel_region, dir.handle, sub_path_posix, .{
	3013	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	3014	defer maybe_sync.deinit(ev);
	3015	const fd = try ev.openat(&maybe_sync.cancel_region, dir.handle, sub_path_posix, .{
2891	3016	.ACCMODE = if (flags.read) .RDWR else .WRONLY,
2892	3017	.CREAT = true,
2893	3018	.TRUNC = flags.truncate,
2894	3019	.EXCL = flags.exclusive,
2895	3020	.CLOEXEC = true,
2896	3021	}, flags.permissions.toMode());
2897		errdefer ev.close(fd);
	3022	errdefer ev.close(maybe_sync.cancelRegion(), fd);
2898	3023
2899	3024	switch (flags.lock) {
2900	3025	.none => {},
2901	3026	.shared, .exclusive => try ev.flock(
2902		&cancel_region,
	3027	try maybe_sync.enterSync(ev),
2903	3028	fd,
2904	3029	flags.lock,
2905	3030	if (flags.lock_nonblocking) .nonblocking else .blocking,
...	...	@@ -3069,9 +3194,9 @@ fn dirOpenFile(
3069	3194	var path_buffer: [PATH_MAX]u8 = undefined;
3070	3195	const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
3071	3196
3072		var cancel_region: CancelRegion = .init();
3073		defer cancel_region.deinit();
3074		const fd = try ev.openat(&cancel_region, dir.handle, sub_path_posix, .{
	3197	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	3198	defer maybe_sync.deinit(ev);
	3199	const fd = try ev.openat(&maybe_sync.cancel_region, dir.handle, sub_path_posix, .{
3075	3200	.ACCMODE = switch (flags.mode) {
3076	3201	.read_only => .RDONLY,
3077	3202	.write_only => .WRONLY,
...	...	@@ -3082,11 +3207,11 @@ fn dirOpenFile(
3082	3207	.CLOEXEC = true,
3083	3208	.PATH = flags.path_only,
3084	3209	}, 0);
3085		errdefer ev.close(fd);
	3210	errdefer ev.close(maybe_sync.cancelRegion(), fd);
3086	3211
3087	3212	if (!flags.allow_directory) {
3088	3213	const is_dir = is_dir: {
3089		const s = ev.stat(&cancel_region, fd) catch \|err\| switch (err) {
	3214	const s = ev.stat(&maybe_sync.cancel_region, fd) catch \|err\| switch (err) {
3090	3215	// The directory-ness is either unknown or unknowable
3091	3216	error.Streaming => break :is_dir false,
3092	3217	else => \|e\| return e,
...	...	@@ -3099,7 +3224,7 @@ fn dirOpenFile(
3099	3224	switch (flags.lock) {
3100	3225	.none => {},
3101	3226	.shared, .exclusive => try ev.flock(
3102		&cancel_region,
	3227	try maybe_sync.enterSync(ev),
3103	3228	fd,
3104	3229	flags.lock,
3105	3230	if (flags.lock_nonblocking) .nonblocking else .blocking,
...	...	@@ -3111,7 +3236,9 @@ fn dirOpenFile(
3111	3236
3112	3237	fn dirClose(userdata: ?*anyopaque, dirs: []const Dir) void {
3113	3238	const ev: *Evented = @ptrCast(@alignCast(userdata));
3114		for (dirs) \|dir\| ev.close(dir.handle);
	3239	var cancel_region: CancelRegion = .init();
	3240	defer cancel_region.deinit();
	3241	for (dirs) \|dir\| ev.close(&cancel_region, dir.handle);
3115	3242	}
3116	3243
3117	3244	fn dirRead(userdata: ?anyopaque, dr: Dir.Reader, buffer: []Dir.Entry) Dir.Reader.Error!usize {
...	...	@@ -3122,17 +3249,17 @@ fn dirRead(userdata: ?anyopaque, dr: Dir.Reader, buffer: []Dir.Entry) Dir.Read
3122	3249	// Refill the buffer, unless we've already created references to
3123	3250	// buffered data.
3124	3251	if (buffer_index != 0) break;
3125		var cancel_region: CancelRegion = .init();
3126		defer cancel_region.deinit();
	3252	var sync: CancelRegion.Sync = try .init(ev);
	3253	defer sync.deinit(ev);
3127	3254	if (dr.state == .reset) {
3128		ev.lseek(&cancel_region, dr.dir.handle, 0, linux.SEEK.SET) catch \|err\| switch (err) {
	3255	ev.lseek(&sync, dr.dir.handle, 0, linux.SEEK.SET) catch \|err\| switch (err) {
3129	3256	error.Unseekable => return error.Unexpected,
3130	3257	else => \|e\| return e,
3131	3258	};
3132	3259	dr.state = .reading;
3133	3260	}
3134	3261	const n = while (true) {
3135		try cancel_region.await(.nothing);
	3262	try sync.cancel_region.await(.nothing);
3136	3263	const rc = linux.getdents64(dr.dir.handle, dr.buffer.ptr, dr.buffer.len);
3137	3264	switch (linux.errno(rc)) {
3138	3265	.SUCCESS => break rc,
...	...	@@ -3203,9 +3330,9 @@ fn dirRead(userdata: ?anyopaque, dr: Dir.Reader, buffer: []Dir.Entry) Dir.Read
3203	3330
3204	3331	fn dirRealPath(userdata: ?*anyopaque, dir: Dir, out_buffer: []u8) Dir.RealPathError!usize {
3205	3332	const ev: *Evented = @ptrCast(@alignCast(userdata));
3206		var cancel_region: CancelRegion = .init();
3207		defer cancel_region.deinit();
3208		return ev.realPath(&cancel_region, dir.handle, out_buffer);
	3333	var sync: CancelRegion.Sync = try .init(ev);
	3334	defer sync.deinit(ev);
	3335	return ev.realPath(&sync, dir.handle, out_buffer);
3209	3336	}
3210	3337
3211	3338	fn dirRealPathFile(
...	...	@@ -3219,9 +3346,9 @@ fn dirRealPathFile(
3219	3346	var path_buffer: [PATH_MAX]u8 = undefined;
3220	3347	const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
3221	3348
3222		var cancel_region: CancelRegion = .init();
3223		defer cancel_region.deinit();
3224		const fd = ev.openat(&cancel_region, dir.handle, sub_path_posix, .{
	3349	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	3350	defer maybe_sync.deinit(ev);
	3351	const fd = ev.openat(&maybe_sync.cancel_region, dir.handle, sub_path_posix, .{
3225	3352	.CLOEXEC = true,
3226	3353	.PATH = true,
3227	3354	}, 0) catch \|err\| switch (err) {
...	...	@@ -3229,8 +3356,8 @@ fn dirRealPathFile(
3229	3356	error.FileLocksUnsupported => return errnoBug(.OPNOTSUPP), // Not asking for locks.
3230	3357	else => \|e\| return e,
3231	3358	};
3232		defer ev.close(fd);
3233		return ev.realPath(&cancel_region, fd, out_buffer);
	3359	defer ev.close(maybe_sync.cancelRegion(), fd);
	3360	return ev.realPath(try maybe_sync.enterSync(ev), fd, out_buffer);
3234	3361	}
3235	3362
3236	3363	fn dirDeleteFile(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8) Dir.DeleteFileError!void {
...	...	@@ -3458,15 +3585,14 @@ fn dirReadLink(
3458	3585	buffer: []u8,
3459	3586	) Dir.ReadLinkError!usize {
3460	3587	const ev: *Evented = @ptrCast(@alignCast(userdata));
3461		_ = ev;
3462	3588
3463	3589	var sub_path_buffer: [PATH_MAX]u8 = undefined;
3464	3590	const sub_path_posix = try pathToPosix(sub_path, &sub_path_buffer);
3465	3591
3466		var cancel_region: CancelRegion = .init();
3467		defer cancel_region.deinit();
	3592	var sync: CancelRegion.Sync = try .init(ev);
	3593	defer sync.deinit(ev);
3468	3594	while (true) {
3469		try cancel_region.await(.nothing);
	3595	try sync.cancel_region.await(.nothing);
3470	3596	const rc = linux.readlinkat(dir.handle, sub_path_posix, buffer.ptr, buffer.len);
3471	3597	switch (linux.errno(rc)) {
3472	3598	.SUCCESS => {
...	...	@@ -3496,10 +3622,10 @@ fn dirSetOwner(
3496	3622	group: ?File.Gid,
3497	3623	) Dir.SetOwnerError!void {
3498	3624	const ev: *Evented = @ptrCast(@alignCast(userdata));
3499		var cancel_region: CancelRegion = .init();
3500		defer cancel_region.deinit();
	3625	var sync: CancelRegion.Sync = try .init(ev);
	3626	defer sync.deinit(ev);
3501	3627	try ev.fchownat(
3502		&cancel_region,
	3628	&sync,
3503	3629	dir.handle,
3504	3630	"",
3505	3631	owner orelse std.math.maxInt(linux.uid_t),
...	...	@@ -3519,10 +3645,10 @@ fn dirSetFileOwner(
3519	3645	const ev: *Evented = @ptrCast(@alignCast(userdata));
3520	3646	var path_buffer: [PATH_MAX]u8 = undefined;
3521	3647	const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
3522		var cancel_region: CancelRegion = .init();
3523		defer cancel_region.deinit();
	3648	var sync: CancelRegion.Sync = try .init(ev);
	3649	defer sync.deinit(ev);
3524	3650	try ev.fchownat(
3525		&cancel_region,
	3651	&sync,
3526	3652	dir.handle,
3527	3653	sub_path_posix,
3528	3654	owner orelse std.math.maxInt(linux.uid_t),
...	...	@@ -3537,10 +3663,10 @@ fn dirSetPermissions(
3537	3663	permissions: Dir.Permissions,
3538	3664	) Dir.SetPermissionsError!void {
3539	3665	const ev: *Evented = @ptrCast(@alignCast(userdata));
3540		var cancel_region: CancelRegion = .init();
3541		defer cancel_region.deinit();
	3666	var sync: CancelRegion.Sync = try .init(ev);
	3667	defer sync.deinit(ev);
3542	3668	ev.fchmodat(
3543		&cancel_region,
	3669	&sync,
3544	3670	dir.handle,
3545	3671	"",
3546	3672	permissions.toMode(),
...	...	@@ -3565,10 +3691,10 @@ fn dirSetFilePermissions(
3565	3691	const ev: *Evented = @ptrCast(@alignCast(userdata));
3566	3692	var path_buffer: [PATH_MAX]u8 = undefined;
3567	3693	const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
3568		var cancel_region: CancelRegion = .init();
3569		defer cancel_region.deinit();
	3694	var sync: CancelRegion.Sync = try .init(ev);
	3695	defer sync.deinit(ev);
3570	3696	try ev.fchmodat(
3571		&cancel_region,
	3697	&sync,
3572	3698	dir.handle,
3573	3699	sub_path_posix,
3574	3700	permissions.toMode(),
...	...	@@ -3585,8 +3711,8 @@ fn dirSetTimestamps(
3585	3711	const ev: *Evented = @ptrCast(@alignCast(userdata));
3586	3712	var path_buffer: [PATH_MAX]u8 = undefined;
3587	3713	const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
3588		var cancel_region: CancelRegion = .init();
3589		defer cancel_region.deinit();
	3714	var cancel_region: CancelRegion.Sync = try .init(ev);
	3715	defer cancel_region.deinit(ev);
3590	3716	try ev.utimensat(
3591	3717	&cancel_region,
3592	3718	dir.handle,
...	...	@@ -3680,7 +3806,9 @@ fn fileLength(userdata: ?*anyopaque, file: File) File.LengthError!u64 {
3680	3806
3681	3807	fn fileClose(userdata: ?*anyopaque, files: []const File) void {
3682	3808	const ev: *Evented = @ptrCast(@alignCast(userdata));
3683		for (files) \|file\| ev.close(file.handle);
	3809	var cancel_region: CancelRegion = .init();
	3810	defer cancel_region.deinit();
	3811	for (files) \|file\| ev.close(&cancel_region, file.handle);
3684	3812	}
3685	3813
3686	3814	fn fileWritePositional(
...	...	@@ -3807,16 +3935,16 @@ fn fileReadPositional(
3807	3935
3808	3936	fn fileSeekBy(userdata: ?*anyopaque, file: File, offset: i64) File.SeekError!void {
3809	3937	const ev: *Evented = @ptrCast(@alignCast(userdata));
3810		var cancel_region: CancelRegion = .init();
3811		defer cancel_region.deinit();
3812		try ev.lseek(&cancel_region, file.handle, @bitCast(offset), linux.SEEK.CUR);
	3938	var sync: CancelRegion.Sync = try .init(ev);
	3939	defer sync.deinit(ev);
	3940	try ev.lseek(&sync, file.handle, @bitCast(offset), linux.SEEK.CUR);
3813	3941	}
3814	3942
3815	3943	fn fileSeekTo(userdata: ?*anyopaque, file: File, offset: u64) File.SeekError!void {
3816	3944	const ev: *Evented = @ptrCast(@alignCast(userdata));
3817		var cancel_region: CancelRegion = .init();
3818		defer cancel_region.deinit();
3819		try ev.lseek(&cancel_region, file.handle, offset, linux.SEEK.SET);
	3945	var sync: CancelRegion.Sync = try .init(ev);
	3946	defer sync.deinit(ev);
	3947	try ev.lseek(&sync, file.handle, offset, linux.SEEK.SET);
3820	3948	}
3821	3949
3822	3950	fn fileSync(userdata: ?*anyopaque, file: File) File.SyncError!void {
...	...	@@ -3858,14 +3986,12 @@ fn fileSync(userdata: ?*anyopaque, file: File) File.SyncError!void {
3858	3986
3859	3987	fn fileIsTty(userdata: ?*anyopaque, file: File) Io.Cancelable!bool {
3860	3988	const ev: *Evented = @ptrCast(@alignCast(userdata));
3861		_ = ev;
3862		var cancel_region: CancelRegion = .init();
3863		defer cancel_region.deinit();
	3989	var sync: CancelRegion.Sync = try .init(ev);
	3990	defer sync.deinit(ev);
3864	3991	while (true) {
3865		try cancel_region.await(.nothing);
	3992	try sync.cancel_region.await(.nothing);
3866	3993	var wsz: winsize = undefined;
3867		const fd: usize = @bitCast(@as(isize, file.handle));
3868		const rc = linux.syscall3(.ioctl, fd, linux.T.IOCGWINSZ, @intFromPtr(&wsz));
	3994	const rc = linux.ioctl(file.handle, linux.T.IOCGWINSZ, @intFromPtr(&wsz));
3869	3995	switch (linux.errno(rc)) {
3870	3996	.SUCCESS => return true,
3871	3997	.INTR => continue,
...	...	@@ -3923,10 +4049,10 @@ fn fileSetOwner(
3923	4049	group: ?File.Gid,
3924	4050	) File.SetOwnerError!void {
3925	4051	const ev: *Evented = @ptrCast(@alignCast(userdata));
3926		var cancel_region: CancelRegion = .init();
3927		defer cancel_region.deinit();
	4052	var sync: CancelRegion.Sync = try .init(ev);
	4053	defer sync.deinit(ev);
3928	4054	try ev.fchownat(
3929		&cancel_region,
	4055	&sync,
3930	4056	file.handle,
3931	4057	"",
3932	4058	owner orelse std.math.maxInt(linux.uid_t),
...	...	@@ -3941,10 +4067,10 @@ fn fileSetPermissions(
3941	4067	permissions: File.Permissions,
3942	4068	) File.SetPermissionsError!void {
3943	4069	const ev: *Evented = @ptrCast(@alignCast(userdata));
3944		var cancel_region: CancelRegion = .init();
3945		defer cancel_region.deinit();
	4070	var sync: CancelRegion.Sync = try .init(ev);
	4071	defer sync.deinit(ev);
3946	4072	ev.fchmodat(
3947		&cancel_region,
	4073	&sync,
3948	4074	file.handle,
3949	4075	"",
3950	4076	permissions.toMode(),
...	...	@@ -3965,10 +4091,10 @@ fn fileSetTimestamps(
3965	4091	options: File.SetTimestampsOptions,
3966	4092	) File.SetTimestampsError!void {
3967	4093	const ev: *Evented = @ptrCast(@alignCast(userdata));
3968		var cancel_region: CancelRegion = .init();
3969		defer cancel_region.deinit();
	4094	var sync: CancelRegion.Sync = try .init(ev);
	4095	defer sync.deinit(ev);
3970	4096	try ev.utimensat(
3971		&cancel_region,
	4097	&sync,
3972	4098	file.handle,
3973	4099	"",
3974	4100	if (options.modify_timestamp != .now or options.access_timestamp != .now) &.{
...	...	@@ -3981,9 +4107,9 @@ fn fileSetTimestamps(
3981	4107
3982	4108	fn fileLock(userdata: ?*anyopaque, file: File, lock: File.Lock) File.LockError!void {
3983	4109	const ev: *Evented = @ptrCast(@alignCast(userdata));
3984		var cancel_region: CancelRegion = .init();
3985		defer cancel_region.deinit();
3986		ev.flock(&cancel_region, file.handle, lock, .blocking) catch \|err\| switch (err) {
	4110	var sync: CancelRegion.Sync = try .init(ev);
	4111	defer sync.deinit(ev);
	4112	ev.flock(&sync, file.handle, lock, .blocking) catch \|err\| switch (err) {
3987	4113	error.WouldBlock => unreachable, // blocking
3988	4114	else => \|e\| return e,
3989	4115	};
...	...	@@ -3991,9 +4117,9 @@ fn fileLock(userdata: ?*anyopaque, file: File, lock: File.Lock) File.LockError!v
3991	4117
3992	4118	fn fileTryLock(userdata: ?*anyopaque, file: File, lock: File.Lock) File.LockError!bool {
3993	4119	const ev: *Evented = @ptrCast(@alignCast(userdata));
3994		var cancel_region: CancelRegion = .init();
3995		defer cancel_region.deinit();
3996		ev.flock(&cancel_region, file.handle, lock, switch (lock) {
	4120	var sync: CancelRegion.Sync = try .init(ev);
	4121	defer sync.deinit(ev);
	4122	ev.flock(&sync, file.handle, lock, switch (lock) {
3997	4123	.none => .blocking,
3998	4124	.shared, .exclusive => .nonblocking,
3999	4125	}) catch \|err\| switch (err) {
...	...	@@ -4005,9 +4131,9 @@ fn fileTryLock(userdata: ?*anyopaque, file: File, lock: File.Lock) File.LockErro
4005	4131
4006	4132	fn fileUnlock(userdata: ?*anyopaque, file: File) void {
4007	4133	const ev: *Evented = @ptrCast(@alignCast(userdata));
4008		var cancel_region: CancelRegion = .initBlocked();
4009		defer cancel_region.deinit();
4010		ev.flock(&cancel_region, file.handle, .none, .blocking) catch \|err\| switch (err) {
	4134	var sync: CancelRegion.Sync = .initBlocked(ev);
	4135	defer sync.deinit(ev);
	4136	ev.flock(&sync, file.handle, .none, .blocking) catch \|err\| switch (err) {
4011	4137	error.Canceled => unreachable, // blocked
4012	4138	error.WouldBlock => unreachable, // blocking
4013	4139	error.SystemResources => return recoverableOsBugDetected(), // Resource deallocation.
...	...	@@ -4018,9 +4144,9 @@ fn fileUnlock(userdata: ?*anyopaque, file: File) void {
4018	4144
4019	4145	fn fileDowngradeLock(userdata: ?*anyopaque, file: File) File.DowngradeLockError!void {
4020	4146	const ev: *Evented = @ptrCast(@alignCast(userdata));
4021		var cancel_region: CancelRegion = .init();
4022		defer cancel_region.deinit();
4023		ev.flock(&cancel_region, file.handle, .shared, .nonblocking) catch \|err\| switch (err) {
	4147	var sync: CancelRegion.Sync = try .init(ev);
	4148	defer sync.deinit(ev);
	4149	ev.flock(&sync, file.handle, .shared, .nonblocking) catch \|err\| switch (err) {
4024	4150	error.WouldBlock => return errnoBug(.AGAIN), // File was not locked in exclusive mode.
4025	4151	error.SystemResources => return errnoBug(.NOLCK), // Lock already obtained.
4026	4152	error.FileLocksUnsupported => return errnoBug(.OPNOTSUPP), // Lock already obtained.
...	...	@@ -4030,9 +4156,9 @@ fn fileDowngradeLock(userdata: ?*anyopaque, file: File) File.DowngradeLockError!
4030	4156
4031	4157	fn fileRealPath(userdata: ?*anyopaque, file: File, out_buffer: []u8) File.RealPathError!usize {
4032	4158	const ev: *Evented = @ptrCast(@alignCast(userdata));
4033		var cancel_region: CancelRegion = .init();
4034		defer cancel_region.deinit();
4035		return ev.realPath(&cancel_region, file.handle, out_buffer);
	4159	var sync: CancelRegion.Sync = try .init(ev);
	4160	defer sync.deinit(ev);
	4161	return ev.realPath(&sync, file.handle, out_buffer);
4036	4162	}
4037	4163
4038	4164	fn fileHardLink(
...	...	@@ -4065,7 +4191,7 @@ fn fileMemoryMapCreate(
4065	4191	options: File.MemoryMap.CreateOptions,
4066	4192	) File.MemoryMap.CreateError!File.MemoryMap {
4067	4193	const ev: *Evented = @ptrCast(@alignCast(userdata));
4068		_ = ev;
	4194
4069	4195	const prot: linux.PROT = .{
4070	4196	.READ = options.protection.read,
4071	4197	.WRITE = options.protection.write,
...	...	@@ -4078,10 +4204,10 @@ fn fileMemoryMapCreate(
4078	4204
4079	4205	const page_align = std.heap.page_size_min;
4080	4206
4081		var cancel_region: CancelRegion = .init();
4082		defer cancel_region.deinit();
	4207	var sync: CancelRegion.Sync = try .init(ev);
	4208	defer sync.deinit(ev);
4083	4209	const contents = while (true) {
4084		try cancel_region.await(.nothing);
	4210	try sync.cancel_region.await(.nothing);
4085	4211	const casted_offset = std.math.cast(i64, options.offset) orelse return error.Unseekable;
4086	4212	const rc = linux.mmap(null, options.len, prot, flags, file.handle, casted_offset);
4087	4213	switch (linux.errno(rc)) {
...	...	@@ -4189,11 +4315,10 @@ fn unlockStderr(userdata: ?*anyopaque) void {
4189	4315
4190	4316	fn processCurrentPath(userdata: ?*anyopaque, buffer: []u8) process.CurrentPathError!usize {
4191	4317	const ev: *Evented = @ptrCast(@alignCast(userdata));
4192		_ = ev;
4193		var cancel_region: CancelRegion = .init();
4194		defer cancel_region.deinit();
	4318	var sync: CancelRegion.Sync = try .init(ev);
	4319	defer sync.deinit(ev);
4195	4320	while (true) {
4196		try cancel_region.await(.nothing);
	4321	try sync.cancel_region.await(.nothing);
4197	4322	switch (linux.errno(linux.getcwd(buffer.ptr, buffer.len))) {
4198	4323	.SUCCESS => return std.mem.findScalar(u8, buffer, 0).?,
4199	4324	.INTR => continue,
...	...	@@ -4208,48 +4333,19 @@ fn processCurrentPath(userdata: ?*anyopaque, buffer: []u8) process.CurrentPathEr
4208	4333
4209	4334	fn processSetCurrentDir(userdata: ?*anyopaque, dir: Dir) process.SetCurrentDirError!void {
4210	4335	const ev: *Evented = @ptrCast(@alignCast(userdata));
4211		_ = ev;
4212	4336	if (dir.handle == linux.AT.FDCWD) return;
4213		var cancel_region: CancelRegion = .init();
4214		defer cancel_region.deinit();
4215		while (true) {
4216		try cancel_region.await(.nothing);
4217		switch (linux.errno(linux.fchdir(dir.handle))) {
4218		.SUCCESS => return,
4219		.INTR => continue,
4220		.ACCES => return error.AccessDenied,
4221		.NOTDIR => return error.NotDir,
4222		.IO => return error.FileSystem,
4223		.BADF => \|err\| return errnoBug(err),
4224		else => \|err\| return unexpectedErrno(err),
4225		}
4226		}
	4337	var sync: CancelRegion.Sync = try .init(ev);
	4338	defer sync.deinit(ev);
	4339	return ev.fchdir(&sync, dir.handle);
4227	4340	}
4228	4341
4229	4342	fn processSetCurrentPath(userdata: ?*anyopaque, dir_path: []const u8) ChdirError!void {
4230	4343	const ev: *Evented = @ptrCast(@alignCast(userdata));
4231		_ = ev;
4232	4344	var path_buffer: [PATH_MAX]u8 = undefined;
4233	4345	const dir_path_posix = try pathToPosix(dir_path, &path_buffer);
4234		var cancel_region: CancelRegion = .init();
4235		defer cancel_region.deinit();
4236		while (true) {
4237		try cancel_region.await(.nothing);
4238		switch (linux.errno(linux.chdir(dir_path_posix))) {
4239		.SUCCESS => return,
4240		.INTR => continue,
4241		.ACCES => return error.AccessDenied,
4242		.IO => return error.FileSystem,
4243		.LOOP => return error.SymLinkLoop,
4244		.NAMETOOLONG => return error.NameTooLong,
4245		.NOENT => return error.FileNotFound,
4246		.NOMEM => return error.SystemResources,
4247		.NOTDIR => return error.NotDir,
4248		.ILSEQ => return error.BadPathName,
4249		.FAULT => \|err\| return errnoBug(err),
4250		else => \|err\| return unexpectedErrno(err),
4251		}
4252		}
	4346	var sync: CancelRegion.Sync = try .init(ev);
	4347	defer sync.deinit(ev);
	4348	return ev.chdir(&sync, dir_path_posix);
4253	4349	}
4254	4350
4255	4351	fn processReplace(userdata: ?*anyopaque, options: process.ReplaceOptions) process.ReplaceError {
...	...	@@ -4275,7 +4371,9 @@ fn processReplace(userdata: ?*anyopaque, options: process.ReplaceOptions) proces
4275	4371	});
4276	4372	};
4277	4373
4278		return execv(options.expand_arg0, argv_buf.ptr[0].?, argv_buf.ptr, env_block, PATH);
	4374	var sync: CancelRegion.Sync = try .init(ev);
	4375	defer sync.deinit(ev);
	4376	return ev.execv(&sync, options.expand_arg0, argv_buf.ptr[0].?, argv_buf.ptr, env_block, PATH);
4279	4377	}
4280	4378
4281	4379	fn processReplacePath(
...	...	@@ -4293,12 +4391,12 @@ fn processReplacePath(
4293	4391	fn processSpawn(userdata: ?*anyopaque, options: process.SpawnOptions) process.SpawnError!process.Child {
4294	4392	const ev: *Evented = @ptrCast(@alignCast(userdata));
4295	4393	const spawned = try ev.spawn(options);
4296		defer ev.close(spawned.err_fd);
	4394	var cancel_region: CancelRegion = .initBlocked();
	4395	defer cancel_region.deinit();
	4396	defer ev.close(&cancel_region, spawned.err_fd);
4297	4397
4298	4398	// Wait for the child to report any errors in or before `execvpe`.
4299	4399	var child_err: ForkBailError = undefined;
4300		var cancel_region: CancelRegion = .initBlocked();
4301		defer cancel_region.deinit();
4302	4400	ev.readAll(&cancel_region, spawned.err_fd, @ptrCast(&child_err)) catch \|read_err\| {
4303	4401	switch (read_err) {
4304	4402	error.Canceled => unreachable, // blocked
...	...	@@ -4336,6 +4434,8 @@ fn processSpawnPath(
4336	4434	@panic("TODO processSpawnPath");
4337	4435	}
4338	4436
	4437	const prog_fileno = 3;
	4438
4339	4439	const Spawned = struct {
4340	4440	pid: pid_t,
4341	4441	err_fd: fd_t,
...	...	@@ -4344,6 +4444,9 @@ const Spawned = struct {
4344	4444	stderr: ?File,
4345	4445	};
4346	4446	fn spawn(ev: *Evented, options: process.SpawnOptions) process.SpawnError!Spawned {
	4447	var cancel_region: CancelRegion = .init();
	4448	defer cancel_region.deinit();
	4449
4347	4450	// The child process does need to access (one end of) these pipes. However,
4348	4451	// we must initially set CLOEXEC to avoid a race condition. If another thread
4349	4452	// is racing to spawn a different child process, we don't want it to inherit
...	...	@@ -4358,28 +4461,24 @@ fn spawn(ev: *Evented, options: process.SpawnOptions) process.SpawnError!Spawned
4358	4461
4359	4462	const stdin_pipe = if (options.stdin == .pipe) try pipe2(pipe_flags) else undefined;
4360	4463	errdefer if (options.stdin == .pipe) {
4361		ev.destroyPipe(stdin_pipe);
	4464	ev.destroyPipe(&cancel_region, stdin_pipe);
4362	4465	};
4363	4466
4364	4467	const stdout_pipe = if (options.stdout == .pipe) try pipe2(pipe_flags) else undefined;
4365	4468	errdefer if (options.stdout == .pipe) {
4366		ev.destroyPipe(stdout_pipe);
	4469	ev.destroyPipe(&cancel_region, stdout_pipe);
4367	4470	};
4368	4471
4369	4472	const stderr_pipe = if (options.stderr == .pipe) try pipe2(pipe_flags) else undefined;
4370	4473	errdefer if (options.stderr == .pipe) {
4371		ev.destroyPipe(stderr_pipe);
	4474	ev.destroyPipe(&cancel_region, stderr_pipe);
4372	4475	};
4373	4476
4374	4477	const any_ignore =
4375	4478	options.stdin == .ignore or options.stdout == .ignore or options.stderr == .ignore;
4376		const dev_null_fd = if (any_ignore) dev_null_fd: {
4377		var cancel_region: CancelRegion = .init();
4378		defer cancel_region.deinit();
4379		break :dev_null_fd try ev.null_fd.open(ev, &cancel_region, "/dev/null", .{
4380		.ACCMODE = .RDWR,
4381		});
4382		} else undefined;
	4479	const dev_null_fd = if (any_ignore) try ev.null_fd.open(ev, &cancel_region, "/dev/null", .{
	4480	.ACCMODE = .RDWR,
	4481	}) else undefined;
4383	4482
4384	4483	const prog_pipe: [2]fd_t = if (options.progress_node.index != .none) pipe: {
4385	4484	// We use CLOEXEC for the same reason as in `pipe_flags`.
...	...	@@ -4387,7 +4486,7 @@ fn spawn(ev: *Evented, options: process.SpawnOptions) process.SpawnError!Spawned
4387	4486	_ = linux.fcntl(pipe[0], linux.F.SETPIPE_SZ, @as(u32, std.Progress.max_packet_len * 2));
4388	4487	break :pipe pipe;
4389	4488	} else .{ -1, -1 };
4390		errdefer ev.destroyPipe(prog_pipe);
	4489	errdefer ev.destroyPipe(&cancel_region, prog_pipe);
4391	4490
4392	4491	var arena_allocator = std.heap.ArenaAllocator.init(ev.allocator());
4393	4492	defer arena_allocator.deinit();
...	...	@@ -4405,7 +4504,6 @@ fn spawn(ev: *Evented, options: process.SpawnOptions) process.SpawnError!Spawned
4405	4504	const argv_buf = try arena.allocSentinel(?[*:0]const u8, options.argv.len, null);
4406	4505	for (options.argv, 0..) \|arg, i\| argv_buf[i] = (try arena.dupeZ(u8, arg)).ptr;
4407	4506
4408		const prog_fileno = 3;
4409	4507	comptime assert(@max(linux.STDIN_FILENO, linux.STDOUT_FILENO, linux.STDERR_FILENO) + 1 == prog_fileno);
4410	4508
4411	4509	const env_block = env_block: {
...	...	@@ -4421,7 +4519,7 @@ fn spawn(ev: *Evented, options: process.SpawnOptions) process.SpawnError!Spawned
4421	4519	// This pipe communicates to the parent errors in the child between `fork` and `execvpe`.
4422	4520	// It is closed by the child (via CLOEXEC) without writing if `execvpe` succeeds.
4423	4521	const err_pipe: [2]fd_t = try pipe2(.{ .CLOEXEC = true });
4424		errdefer ev.destroyPipe(err_pipe);
	4522	errdefer ev.destroyPipe(&cancel_region, err_pipe);
4425	4523
4426	4524	try ev.scanEnviron(); // for PATH
4427	4525	const PATH = ev.environ.string.PATH orelse default_PATH;
...	...	@@ -4439,78 +4537,33 @@ fn spawn(ev: *Evented, options: process.SpawnOptions) process.SpawnError!Spawned
4439	4537
4440	4538	if (pid_result == 0) {
4441	4539	defer comptime unreachable; // We are the child.
4442		_ = swapCancelProtection(ev, .blocked);
4443		const ep1 = err_pipe[1];
4444
4445		ev.setUpChildIo(options.stdin, stdin_pipe[0], linux.STDIN_FILENO, dev_null_fd) catch \|err\|
4446		ev.forkBail(ep1, err);
4447		ev.setUpChildIo(options.stdout, stdout_pipe[1], linux.STDOUT_FILENO, dev_null_fd) catch \|err\|
4448		ev.forkBail(ep1, err);
4449		ev.setUpChildIo(options.stderr, stderr_pipe[1], linux.STDERR_FILENO, dev_null_fd) catch \|err\|
4450		ev.forkBail(ep1, err);
4451
4452		switch (options.cwd) {
4453		.inherit => {},
4454		.dir => \|cwd\| processSetCurrentDir(ev, cwd) catch \|err\| ev.forkBail(ep1, err),
4455		.path => \|cwd\| processSetCurrentPath(ev, cwd) catch \|err\| ev.forkBail(ep1, err),
4456		}
4457
4458		// Must happen after fchdir above, the cwd file descriptor might be
4459		// equal to prog_fileno and be clobbered by this dup2 call.
4460		if (prog_pipe[1] != -1) dup2(prog_pipe[1], prog_fileno) catch \|err\| ev.forkBail(ep1, err);
4461
4462		if (options.gid) \|gid\| {
4463		switch (linux.errno(linux.setregid(gid, gid))) {
4464		.SUCCESS => {},
4465		.AGAIN => ev.forkBail(ep1, error.ResourceLimitReached),
4466		.INVAL => ev.forkBail(ep1, error.InvalidUserId),
4467		.PERM => ev.forkBail(ep1, error.PermissionDenied),
4468		else => ev.forkBail(ep1, error.Unexpected),
4469		}
4470		}
4471
4472		if (options.uid) \|uid\| {
4473		switch (linux.errno(linux.setreuid(uid, uid))) {
4474		.SUCCESS => {},
4475		.AGAIN => ev.forkBail(ep1, error.ResourceLimitReached),
4476		.INVAL => ev.forkBail(ep1, error.InvalidUserId),
4477		.PERM => ev.forkBail(ep1, error.PermissionDenied),
4478		else => ev.forkBail(ep1, error.Unexpected),
4479		}
4480		}
4481
4482		if (options.pgid) \|pid\| {
4483		switch (linux.errno(linux.setpgid(0, pid))) {
4484		.SUCCESS => {},
4485		.ACCES => ev.forkBail(ep1, error.ProcessAlreadyExec),
4486		.INVAL => ev.forkBail(ep1, error.InvalidProcessGroupId),
4487		.PERM => ev.forkBail(ep1, error.PermissionDenied),
4488		else => ev.forkBail(ep1, error.Unexpected),
4489		}
4490		}
4491
4492		if (options.start_suspended) {
4493		switch (linux.errno(linux.kill(linux.getpid(), .STOP))) {
4494		.SUCCESS => {},
4495		.PERM => ev.forkBail(ep1, error.PermissionDenied),
4496		else => ev.forkBail(ep1, error.Unexpected),
4497		}
4498		}
4499
4500		const err = execv(options.expand_arg0, argv_buf.ptr[0].?, argv_buf.ptr, env_block, PATH);
4501		ev.forkBail(ep1, err);
	4540	var sync: CancelRegion.Sync = .{ .cancel_region = .initBlocked() };
	4541	const err = ev.setUpChild(&sync, .{
	4542	.stdin_pipe = stdin_pipe[0],
	4543	.stdout_pipe = stdout_pipe[1],
	4544	.stderr_pipe = stderr_pipe[1],
	4545	.dev_null_fd = dev_null_fd,
	4546	.prog_pipe = prog_pipe[1],
	4547	.argv_buf = argv_buf,
	4548	.env_block = env_block,
	4549	.PATH = PATH,
	4550	.spawn = options,
	4551	});
	4552	ev.writeAll(&sync.cancel_region, err_pipe[1], @ptrCast(&err)) catch {};
	4553	const exit = if (builtin.single_threaded) linux.exit else linux.exit_group;
	4554	exit(1);
4502	4555	}
4503	4556
4504	4557	const pid: pid_t = @intCast(pid_result); // We are the parent.
4505	4558	errdefer comptime unreachable; // The child is forked; we must not error from now on
4506	4559
4507		ev.close(err_pipe[1]); // make sure only the child holds the write end open
	4560	ev.close(&cancel_region, err_pipe[1]); // make sure only the child holds the write end open
4508	4561
4509		if (options.stdin == .pipe) ev.close(stdin_pipe[0]);
4510		if (options.stdout == .pipe) ev.close(stdout_pipe[1]);
4511		if (options.stderr == .pipe) ev.close(stderr_pipe[1]);
	4562	if (options.stdin == .pipe) ev.close(&cancel_region, stdin_pipe[0]);
	4563	if (options.stdout == .pipe) ev.close(&cancel_region, stdout_pipe[1]);
	4564	if (options.stderr == .pipe) ev.close(&cancel_region, stderr_pipe[1]);
4512	4565
4513		if (prog_pipe[1] != -1) ev.close(prog_pipe[1]);
	4566	if (prog_pipe[1] != -1) ev.close(&cancel_region, prog_pipe[1]);
4514	4567
4515	4568	options.progress_node.setIpcFile(ev, .{ .handle = prog_pipe[0], .flags = .{ .nonblocking = true } });
4516	4569
...	...	@@ -4546,24 +4599,127 @@ pub fn pipe2(flags: linux.O) PipeError![2]fd_t {
4546	4599	else => \|err\| return unexpectedErrno(err),
4547	4600	}
4548	4601	}
4549		fn destroyPipe(ev: *Evented, pipe: [2]fd_t) void {
4550		if (pipe[0] != -1) ev.close(pipe[0]);
4551		if (pipe[0] != pipe[1]) ev.close(pipe[1]);
	4602	fn destroyPipe(ev: Evented, cancel_region: CancelRegion, pipe: [2]fd_t) void {
	4603	if (pipe[0] != -1) ev.close(cancel_region, pipe[0]);
	4604	if (pipe[0] != pipe[1]) ev.close(cancel_region, pipe[1]);
	4605	}
	4606
	4607	/// Errors that can occur between fork() and execv()
	4608	const ForkBailError = process.SetCurrentDirError \|\| ChdirError \|\|
	4609	process.SpawnError \|\| process.ReplaceError;
	4610	fn setUpChild(
	4611	ev: *Evented,
	4612	sync: *CancelRegion.Sync,
	4613	options: struct {
	4614	stdin_pipe: fd_t,
	4615	stdout_pipe: fd_t,
	4616	stderr_pipe: fd_t,
	4617	dev_null_fd: fd_t,
	4618	prog_pipe: fd_t,
	4619	argv_buf: [:null]?[*:0]const u8,
	4620	env_block: process.Environ.Block,
	4621	PATH: []const u8,
	4622	spawn: process.SpawnOptions,
	4623	},
	4624	) ForkBailError {
	4625	try ev.setUpChildIo(
	4626	sync,
	4627	options.spawn.stdin,
	4628	options.stdin_pipe,
	4629	linux.STDIN_FILENO,
	4630	options.dev_null_fd,
	4631	);
	4632	try ev.setUpChildIo(
	4633	sync,
	4634	options.spawn.stdout,
	4635	options.stdout_pipe,
	4636	linux.STDOUT_FILENO,
	4637	options.dev_null_fd,
	4638	);
	4639	try ev.setUpChildIo(
	4640	sync,
	4641	options.spawn.stderr,
	4642	options.stderr_pipe,
	4643	linux.STDERR_FILENO,
	4644	options.dev_null_fd,
	4645	);
	4646
	4647	switch (options.spawn.cwd) {
	4648	.inherit => {},
	4649	.dir => \|cwd_dir\| try ev.fchdir(sync, cwd_dir.handle),
	4650	.path => \|cwd_path\| {
	4651	var cwd_path_buffer: [PATH_MAX]u8 = undefined;
	4652	const cwd_path_posix = try pathToPosix(cwd_path, &cwd_path_buffer);
	4653	try ev.chdir(sync, cwd_path_posix);
	4654	},
	4655	}
	4656
	4657	// Must happen after fchdir above, the cwd file descriptor might be
	4658	// equal to prog_fileno and be clobbered by this dup2 call.
	4659	if (options.prog_pipe != -1) try ev.dup2(sync, options.prog_pipe, prog_fileno);
	4660
	4661	if (options.spawn.gid) \|gid\| {
	4662	switch (linux.errno(linux.setregid(gid, gid))) {
	4663	.SUCCESS => {},
	4664	.AGAIN => return error.ResourceLimitReached,
	4665	.INVAL => return error.InvalidUserId,
	4666	.PERM => return error.PermissionDenied,
	4667	else => return error.Unexpected,
	4668	}
	4669	}
	4670
	4671	if (options.spawn.uid) \|uid\| {
	4672	switch (linux.errno(linux.setreuid(uid, uid))) {
	4673	.SUCCESS => {},
	4674	.AGAIN => return error.ResourceLimitReached,
	4675	.INVAL => return error.InvalidUserId,
	4676	.PERM => return error.PermissionDenied,
	4677	else => return error.Unexpected,
	4678	}
	4679	}
	4680
	4681	if (options.spawn.pgid) \|pid\| {
	4682	switch (linux.errno(linux.setpgid(0, pid))) {
	4683	.SUCCESS => {},
	4684	.ACCES => return error.ProcessAlreadyExec,
	4685	.INVAL => return error.InvalidProcessGroupId,
	4686	.PERM => return error.PermissionDenied,
	4687	else => return error.Unexpected,
	4688	}
	4689	}
	4690
	4691	if (options.spawn.start_suspended) {
	4692	switch (linux.errno(linux.kill(linux.getpid(), .STOP))) {
	4693	.SUCCESS => {},
	4694	.PERM => return error.PermissionDenied,
	4695	else => return error.Unexpected,
	4696	}
	4697	}
	4698
	4699	return ev.execv(
	4700	sync,
	4701	options.spawn.expand_arg0,
	4702	options.argv_buf.ptr[0].?,
	4703	options.argv_buf.ptr,
	4704	options.env_block,
	4705	options.PATH,
	4706	);
4552	4707	}
4553	4708
4554	4709	fn setUpChildIo(
4555	4710	ev: *Evented,
	4711	sync: *CancelRegion.Sync,
4556	4712	stdio: process.SpawnOptions.StdIo,
4557	4713	pipe_fd: fd_t,
4558	4714	std_fileno: i32,
4559	4715	dev_null_fd: fd_t,
4560	4716	) !void {
4561	4717	switch (stdio) {
4562		.pipe => try dup2(pipe_fd, std_fileno),
4563		.close => ev.close(std_fileno),
	4718	.pipe => try ev.dup2(sync, pipe_fd, std_fileno),
	4719	.close => ev.close(&sync.cancel_region, std_fileno),
4564	4720	.inherit => {},
4565		.ignore => try dup2(dev_null_fd, std_fileno),
4566		.file => \|file\| try dup2(file.handle, std_fileno),
	4721	.ignore => try ev.dup2(sync, dev_null_fd, std_fileno),
	4722	.file => \|file\| try ev.dup2(sync, file.handle, std_fileno),
4567	4723	}
4568	4724	}
4569	4725
...	...	@@ -4571,11 +4727,10 @@ pub const DupError = error{
4571	4727	ProcessFdQuotaExceeded,
4572	4728	SystemResources,
4573	4729	} \|\| Io.UnexpectedError \|\| Io.Cancelable;
4574		pub fn dup2(old_fd: fd_t, new_fd: fd_t) DupError!void {
4575		var cancel_region: CancelRegion = .init();
4576		defer cancel_region.deinit();
	4730	pub fn dup2(ev: Evented, sync: CancelRegion.Sync, old_fd: fd_t, new_fd: fd_t) DupError!void {
	4731	_ = ev;
4577	4732	while (true) {
4578		try cancel_region.await(.nothing);
	4733	try sync.cancel_region.await(.nothing);
4579	4734	switch (linux.errno(linux.dup2(old_fd, new_fd))) {
4580	4735	.SUCCESS => {},
4581	4736	.BUSY, .INTR => continue,
...	...	@@ -4588,20 +4743,9 @@ pub fn dup2(old_fd: fd_t, new_fd: fd_t) DupError!void {
4588	4743	}
4589	4744	}
4590	4745
4591		/// Errors that can occur between fork() and execv()
4592		const ForkBailError = process.SetCurrentDirError \|\| ChdirError \|\|
4593		process.SpawnError \|\| process.ReplaceError;
4594		/// Child of fork calls this to report an error to the fork parent. Then the
4595		/// child exits.
4596		fn forkBail(ev: *Evented, fd: fd_t, err: ForkBailError) noreturn {
4597		var cancel_region: CancelRegion = .initBlocked();
4598		defer cancel_region.deinit();
4599		ev.writeAll(&cancel_region, fd, @ptrCast(&err)) catch {};
4600		const exit = if (builtin.single_threaded) linux.exit else linux.exit_group;
4601		exit(1);
4602		}
4603
4604	4746	fn execv(
	4747	ev: *Evented,
	4748	sync: *CancelRegion.Sync,
4605	4749	arg0_expand: process.ArgExpansion,
4606	4750	file: [*:0]const u8,
4607	4751	child_argv: [:null]?[:0]const u8,
...	...	@@ -4609,7 +4753,7 @@ fn execv(
4609	4753	PATH: []const u8,
4610	4754	) process.ReplaceError {
4611	4755	const file_slice = std.mem.sliceTo(file, 0);
4612		if (std.mem.findScalar(u8, file_slice, '/') != null) return execvPath(file, child_argv, env_block);
	4756	if (std.mem.findScalar(u8, file_slice, '/') != null) return ev.execvPath(sync, file, child_argv, env_block);
4613	4757
4614	4758	// Use of PATH_MAX here is valid as the path_buf will be passed
4615	4759	// directly to the operating system in posixExecvPath.
...	...	@@ -4637,7 +4781,7 @@ fn execv(
4637	4781	.expand => child_argv[0] = full_path,
4638	4782	.no_expand => {},
4639	4783	}
4640		err = execvPath(full_path, child_argv, env_block);
	4784	err = ev.execvPath(sync, full_path, child_argv, env_block);
4641	4785	switch (err) {
4642	4786	error.AccessDenied => seen_eacces = true,
4643	4787	error.FileNotFound, error.NotDir => {},
...	...	@@ -4649,13 +4793,14 @@ fn execv(
4649	4793	}
4650	4794	/// This function ignores PATH environment variable.
4651	4795	pub fn execvPath(
	4796	ev: *Evented,
	4797	sync: *CancelRegion.Sync,
4652	4798	path: [*:0]const u8,
4653	4799	child_argv: [:null]const ?[:0]const u8,
4654	4800	env_block: process.Environ.PosixBlock,
4655	4801	) process.ReplaceError {
4656		var cancel_region: CancelRegion = .init();
4657		defer cancel_region.deinit();
4658		try cancel_region.await(.nothing);
	4802	_ = ev;
	4803	try sync.cancel_region.await(.nothing);
4659	4804	switch (linux.errno(linux.execve(path, child_argv, env_block.slice.ptr))) {
4660	4805	.FAULT => \|err\| return errnoBug(err), // Bad pointer parameter.
4661	4806	.@"2BIG" => return error.SystemResources,
...	...	@@ -4680,14 +4825,15 @@ pub fn execvPath(
4680	4825
4681	4826	fn childWait(userdata: ?anyopaque, child: process.Child) process.Child.WaitError!process.Child.Term {
4682	4827	const ev: *Evented = @ptrCast(@alignCast(userdata));
4683		defer ev.childCleanup(child);
	4828
	4829	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	4830	defer maybe_sync.deinit(ev);
	4831	defer ev.childCleanup(maybe_sync.cancelRegion(), child);
4684	4832
4685	4833	const pid = child.id.?;
4686	4834	var info: linux.siginfo_t = undefined;
4687		var cancel_region: CancelRegion = .init();
4688		defer cancel_region.deinit();
4689	4835	while (true) {
4690		const thread = try cancel_region.awaitIoUring();
	4836	const thread = try maybe_sync.cancel_region.awaitIoUring();
4691	4837	thread.enqueue().* = .{
4692	4838	.opcode = .WAITID,
4693	4839	.flags = 0,
...	...	@@ -4697,7 +4843,7 @@ fn childWait(userdata: ?anyopaque, child: process.Child) process.Child.WaitErr
4697	4843	.addr = 0,
4698	4844	.len = @intFromEnum(linux.P.PID),
4699	4845	.rw_flags = 0,
4700		.user_data = @intFromPtr(cancel_region.fiber),
	4846	.user_data = @intFromPtr(maybe_sync.cancel_region.fiber),
4701	4847	.buf_index = 0,
4702	4848	.personality = 0,
4703	4849	.splice_fd_in = linux.W.EXITED \|
...	...	@@ -4706,27 +4852,30 @@ fn childWait(userdata: ?anyopaque, child: process.Child) process.Child.WaitErr
4706	4852	.resv = 0,
4707	4853	};
4708	4854	ev.yield(null, .nothing);
4709		switch (cancel_region.errno()) {
	4855	switch (maybe_sync.cancel_region.errno()) {
4710	4856	.SUCCESS => {
4711		if (child.request_resource_usage_statistics) while (true) {
4712		try cancel_region.await(.nothing);
4713		var rusage: linux.rusage = undefined;
4714		switch (linux.errno(linux.waitid(
4715		.PID,
4716		pid,
4717		&info,
4718		linux.W.EXITED \| linux.W.NOHANG,
4719		&rusage,
4720		))) {
4721		.SUCCESS => {
4722		child.resource_usage_statistics.rusage = rusage;
4723		break;
4724		},
4725		.INTR, .CANCELED => continue,
4726		.CHILD => \|err\| return errnoBug(err), // Double-free.
4727		else => \|err\| return unexpectedErrno(err),
	4857	if (child.request_resource_usage_statistics) {
	4858	const sync = try maybe_sync.enterSync(ev);
	4859	while (true) {
	4860	try sync.cancel_region.await(.nothing);
	4861	var rusage: linux.rusage = undefined;
	4862	switch (linux.errno(linux.waitid(
	4863	.PID,
	4864	pid,
	4865	&info,
	4866	linux.W.EXITED \| linux.W.NOHANG,
	4867	&rusage,
	4868	))) {
	4869	.SUCCESS => {
	4870	child.resource_usage_statistics.rusage = rusage;
	4871	break;
	4872	},
	4873	.INTR, .CANCELED => continue,
	4874	.CHILD => \|err\| return errnoBug(err), // Double-free.
	4875	else => \|err\| return unexpectedErrno(err),
	4876	}
4728	4877	}
4729		};
	4878	}
4730	4879	const status: u32 = @bitCast(info.fields.common.second.sigchld.status);
4731	4880	const code: linux.CLD = @enumFromInt(info.code);
4732	4881	return switch (code) {
...	...	@@ -4745,11 +4894,12 @@ fn childWait(userdata: ?anyopaque, child: process.Child) process.Child.WaitErr
4745	4894
4746	4895	fn childKill(userdata: ?anyopaque, child: process.Child) void {
4747	4896	const ev: *Evented = @ptrCast(@alignCast(userdata));
4748		defer ev.childCleanup(child);
	4897
	4898	var maybe_sync: CancelRegion.Sync.Maybe = .{ .sync = .initBlocked(ev) };
	4899	defer maybe_sync.deinit(ev);
	4900	defer ev.childCleanup(maybe_sync.cancelRegion(), child);
4749	4901
4750	4902	const pid = child.id.?;
4751		var cancel_region: CancelRegion = .initBlocked();
4752		defer cancel_region.deinit();
4753	4903	while (true) switch (linux.errno(linux.kill(pid, .TERM))) {
4754	4904	.SUCCESS => break,
4755	4905	.INTR => continue,
...	...	@@ -4758,10 +4908,11 @@ fn childKill(userdata: ?anyopaque, child: process.Child) void {
4758	4908	.SRCH => \|err\| return errnoBug(err) catch {},
4759	4909	else => \|err\| return unexpectedErrno(err) catch {},
4760	4910	};
	4911	maybe_sync.leaveSync(ev);
4761	4912
4762	4913	var info: linux.siginfo_t = undefined;
4763	4914	while (true) {
4764		const thread = cancel_region.awaitIoUring() catch \|err\| switch (err) {
	4915	const thread = maybe_sync.cancel_region.awaitIoUring() catch \|err\| switch (err) {
4765	4916	error.Canceled => unreachable, // blocked
4766	4917	};
4767	4918	thread.enqueue().* = .{
...	...	@@ -4773,7 +4924,7 @@ fn childKill(userdata: ?anyopaque, child: process.Child) void {
4773	4924	.addr = 0,
4774	4925	.len = @intFromEnum(linux.P.PID),
4775	4926	.rw_flags = 0,
4776		.user_data = @intFromPtr(cancel_region.fiber),
	4927	.user_data = @intFromPtr(maybe_sync.cancel_region.fiber),
4777	4928	.buf_index = 0,
4778	4929	.personality = 0,
4779	4930	.splice_fd_in = linux.W.EXITED,
...	...	@@ -4781,7 +4932,7 @@ fn childKill(userdata: ?anyopaque, child: process.Child) void {
4781	4932	.resv = 0,
4782	4933	};
4783	4934	ev.yield(null, .nothing);
4784		switch (cancel_region.errno()) {
	4935	switch (maybe_sync.cancel_region.errno()) {
4785	4936	.SUCCESS => return,
4786	4937	.INTR, .CANCELED => continue,
4787	4938	.CHILD => \|err\| return errnoBug(err) catch {}, // Double-free.
...	...	@@ -4790,17 +4941,17 @@ fn childKill(userdata: ?anyopaque, child: process.Child) void {
4790	4941	}
4791	4942	}
4792	4943
4793		fn childCleanup(ev: Evented, child: process.Child) void {
	4944	fn childCleanup(ev: Evented, cancel_region: CancelRegion, child: *process.Child) void {
4794	4945	if (child.stdin) \|*stdin\| {
4795		ev.close(stdin.handle);
	4946	ev.close(cancel_region, stdin.handle);
4796	4947	child.stdin = null;
4797	4948	}
4798	4949	if (child.stdout) \|*stdout\| {
4799		ev.close(stdout.handle);
	4950	ev.close(cancel_region, stdout.handle);
4800	4951	child.stdout = null;
4801	4952	}
4802	4953	if (child.stderr) \|*stderr\| {
4803		ev.close(stderr.handle);
	4954	ev.close(cancel_region, stderr.handle);
4804	4955	child.stderr = null;
4805	4956	}
4806	4957	child.id = null;
...	...	@@ -4985,14 +5136,14 @@ fn netBindIp(
4985	5136	) net.IpAddress.BindError!net.Socket {
4986	5137	const ev: *Evented = @ptrCast(@alignCast(userdata));
4987	5138	const family = posixAddressFamily(address);
4988		var cancel_region: CancelRegion = .init();
4989		defer cancel_region.deinit();
4990		const socket_fd = try ev.socket(&cancel_region, family, options);
4991		errdefer ev.close(socket_fd);
	5139	var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
	5140	defer maybe_sync.deinit(ev);
	5141	const socket_fd = try ev.socket(&maybe_sync.cancel_region, family, options);
	5142	errdefer ev.close(maybe_sync.cancelRegion(), socket_fd);
4992	5143	var storage: PosixAddress = undefined;
4993	5144	var addr_len = addressToPosix(address, &storage);
4994		try ev.bind(&cancel_region, socket_fd, &storage.any, addr_len);
4995		try ev.getsockname(&cancel_region, socket_fd, &storage.any, &addr_len);
	5145	try ev.bind(&maybe_sync.cancel_region, socket_fd, &storage.any, addr_len);
	5146	try ev.getsockname(try maybe_sync.enterSync(ev), socket_fd, &storage.any, &addr_len);
4996	5147	return .{
4997	5148	.handle = socket_fd,
4998	5149	.address = addressFromPosix(&storage),
...	...	@@ -5268,7 +5419,9 @@ fn netWriteFileUnavailable(
5268	5419
5269	5420	fn netClose(userdata: ?*anyopaque, handles: []const net.Socket.Handle) void {
5270	5421	const ev: *Evented = @ptrCast(@alignCast(userdata));
5271		for (handles) \|handle\| ev.close(handle);
	5422	var cancel_region: CancelRegion = .init();
	5423	defer cancel_region.deinit();
	5424	for (handles) \|handle\| ev.close(&cancel_region, handle);
5272	5425	}
5273	5426
5274	5427	fn netCloseUnavailable(userdata: ?*anyopaque, handles: []const net.Socket.Handle) void {
...	...	@@ -5407,9 +5560,28 @@ fn bind(
5407	5560	}
5408	5561	}
5409	5562
5410		fn close(ev: *Evented, fd: fd_t) void {
5411		var cancel_region: CancelRegion = .initBlocked();
5412		defer cancel_region.deinit();
	5563	fn chdir(ev: Evented, sync: CancelRegion.Sync, path: [*:0]const u8) ChdirError!void {
	5564	_ = ev;
	5565	while (true) {
	5566	try sync.cancel_region.await(.nothing);
	5567	switch (linux.errno(linux.chdir(path))) {
	5568	.SUCCESS => return,
	5569	.INTR => continue,
	5570	.ACCES => return error.AccessDenied,
	5571	.IO => return error.FileSystem,
	5572	.LOOP => return error.SymLinkLoop,
	5573	.NAMETOOLONG => return error.NameTooLong,
	5574	.NOENT => return error.FileNotFound,
	5575	.NOMEM => return error.SystemResources,
	5576	.NOTDIR => return error.NotDir,
	5577	.ILSEQ => return error.BadPathName,
	5578	.FAULT => \|err\| return errnoBug(err),
	5579	else => \|err\| return unexpectedErrno(err),
	5580	}
	5581	}
	5582	}
	5583
	5584	fn close(ev: Evented, cancel_region: CancelRegion, fd: fd_t) void {
5413	5585	while (true) {
5414	5586	const thread = cancel_region.awaitIoUring() catch \|err\| switch (err) {
5415	5587	error.Canceled => unreachable, // blocked
...	...	@@ -5440,9 +5612,26 @@ fn close(ev: *Evented, fd: fd_t) void {
5440	5612	}
5441	5613	}
5442	5614
	5615	fn fchdir(ev: Evented, sync: CancelRegion.Sync, dir: fd_t) process.SetCurrentDirError!void {
	5616	_ = ev;
	5617	if (dir == linux.AT.FDCWD) return;
	5618	while (true) {
	5619	try sync.cancel_region.await(.nothing);
	5620	switch (linux.errno(linux.fchdir(dir))) {
	5621	.SUCCESS => return,
	5622	.INTR => continue,
	5623	.ACCES => return error.AccessDenied,
	5624	.NOTDIR => return error.NotDir,
	5625	.IO => return error.FileSystem,
	5626	.BADF => \|err\| return errnoBug(err),
	5627	else => \|err\| return unexpectedErrno(err),
	5628	}
	5629	}
	5630	}
	5631
5443	5632	fn fchmodat(
5444	5633	ev: *Evented,
5445		cancel_region: *CancelRegion,
	5634	sync: *CancelRegion.Sync,
5446	5635	dir: fd_t,
5447	5636	path: [*:0]const u8,
5448	5637	mode: linux.mode_t,
...	...	@@ -5450,7 +5639,7 @@ fn fchmodat(
5450	5639	) Dir.SetFilePermissionsError!void {
5451	5640	_ = ev;
5452	5641	while (true) {
5453		try cancel_region.await(.nothing);
	5642	try sync.cancel_region.await(.nothing);
5454	5643	switch (linux.errno(linux.fchmodat2(dir, path, mode, flags))) {
5455	5644	.SUCCESS => return,
5456	5645	.INTR => continue,
...	...	@@ -5473,7 +5662,7 @@ fn fchmodat(
5473	5662
5474	5663	fn fchownat(
5475	5664	ev: *Evented,
5476		cancel_region: *CancelRegion,
	5665	sync: *CancelRegion.Sync,
5477	5666	dir: fd_t,
5478	5667	path: [*:0]const u8,
5479	5668	owner: linux.uid_t,
...	...	@@ -5482,7 +5671,7 @@ fn fchownat(
5482	5671	) File.SetOwnerError!void {
5483	5672	_ = ev;
5484	5673	while (true) {
5485		try cancel_region.await(.nothing);
	5674	try sync.cancel_region.await(.nothing);
5486	5675	switch (linux.errno(linux.fchownat(dir, path, owner, group, flags))) {
5487	5676	.SUCCESS => return,
5488	5677	.INTR => continue,
...	...	@@ -5504,13 +5693,13 @@ fn fchownat(
5504	5693
5505	5694	fn flock(
5506	5695	ev: *Evented,
5507		cancel_region: *CancelRegion,
	5696	sync: *CancelRegion.Sync,
5508	5697	fd: fd_t,
5509	5698	op: File.Lock,
5510	5699	blocking: enum { blocking, nonblocking },
5511	5700	) (File.LockError \|\| error{WouldBlock})!void {
5512	5701	while (true) {
5513		try cancel_region.await(.nothing);
	5702	try sync.cancel_region.await(.nothing);
5514	5703	switch (linux.errno(linux.flock(fd, LOCK.NB \| @as(i32, switch (op) {
5515	5704	.none => LOCK.UN,
5516	5705	.shared => LOCK.SH,
...	...	@@ -5522,7 +5711,7 @@ fn flock(
5522	5711	.INVAL => \|err\| return errnoBug(err), // invalid parameters
5523	5712	.NOLCK => return error.SystemResources,
5524	5713	.AGAIN => {
5525		const thread = try cancel_region.awaitIoUring();
	5714	const thread = try sync.cancel_region.awaitIoUring();
5526	5715	thread.enqueue().* = .{
5527	5716	.opcode = .NOP,
5528	5717	.flags = 0,
...	...	@@ -5532,7 +5721,7 @@ fn flock(
5532	5721	.addr = 0,
5533	5722	.len = 0,
5534	5723	.rw_flags = 0,
5535		.user_data = @intFromPtr(cancel_region.fiber),
	5724	.user_data = @intFromPtr(sync.cancel_region.fiber),
5536	5725	.buf_index = 0,
5537	5726	.personality = 0,
5538	5727	.splice_fd_in = 0,
...	...	@@ -5540,7 +5729,7 @@ fn flock(
5540	5729	.resv = 0,
5541	5730	};
5542	5731	ev.yield(null, .nothing);
5543		switch (cancel_region.errno()) {
	5732	switch (sync.cancel_region.errno()) {
5544	5733	.SUCCESS, .INTR, .CANCELED => {},
5545	5734	else => unreachable,
5546	5735	}
...	...	@@ -5557,14 +5746,14 @@ fn flock(
5557	5746
5558	5747	fn getsockname(
5559	5748	ev: *Evented,
5560		cancel_region: *CancelRegion,
	5749	sync: *CancelRegion.Sync,
5561	5750	socket_fd: fd_t,
5562	5751	addr: *linux.sockaddr,
5563	5752	addr_len: *linux.socklen_t,
5564	5753	) !void {
5565	5754	_ = ev;
5566	5755	while (true) {
5567		try cancel_region.await(.nothing);
	5756	try sync.cancel_region.await(.nothing);
5568	5757	switch (linux.errno(linux.getsockname(socket_fd, addr, addr_len))) {
5569	5758	.SUCCESS => return,
5570	5759	.INTR => continue,
...	...	@@ -5633,14 +5822,14 @@ fn linkat(
5633	5822
5634	5823	fn lseek(
5635	5824	ev: *Evented,
5636		cancel_region: *CancelRegion,
	5825	sync: *CancelRegion.Sync,
5637	5826	fd: fd_t,
5638	5827	offset: u64,
5639	5828	whence: u32,
5640	5829	) File.SeekError!void {
5641	5830	_ = ev;
5642	5831	while (true) {
5643		try cancel_region.await(.nothing);
	5832	try sync.cancel_region.await(.nothing);
5644	5833	var result: u64 = undefined;
5645	5834	switch (linux.errno(switch (@sizeOf(usize)) {
5646	5835	else => comptime unreachable,
...	...	@@ -5837,7 +6026,7 @@ fn readAll(
5837	6026
5838	6027	fn realPath(
5839	6028	ev: *Evented,
5840		cancel_region: *CancelRegion,
	6029	sync: *CancelRegion.Sync,
5841	6030	fd: fd_t,
5842	6031	out_buffer: []u8,
5843	6032	) File.RealPathError!usize {
...	...	@@ -5846,7 +6035,7 @@ fn realPath(
5846	6035	const proc_path = std.fmt.bufPrintSentinel(&procfs_buf, "/proc/self/fd/{d}", .{fd}, 0) catch
5847	6036	unreachable;
5848	6037	while (true) {
5849		try cancel_region.await(.nothing);
	6038	try sync.cancel_region.await(.nothing);
5850	6039	const rc = linux.readlink(proc_path, out_buffer.ptr, out_buffer.len);
5851	6040	switch (linux.errno(rc)) {
5852	6041	.SUCCESS => return rc,
...	...	@@ -6025,7 +6214,7 @@ fn socket(
6025	6214	else => \|err\| return unexpectedErrno(err),
6026	6215	}
6027	6216	};
6028		errdefer ev.close(socket_fd);
	6217	errdefer ev.close(cancel_region, socket_fd);
6029	6218
6030	6219	if (options.ip6_only) {
6031	6220	if (linux.IPV6 == void) return error.OptionUnsupported;
...	...	@@ -6103,7 +6292,7 @@ fn urandomReadAll(
6103	6292
6104	6293	fn utimensat(
6105	6294	ev: *Evented,
6106		cancel_region: *CancelRegion,
	6295	sync: *CancelRegion.Sync,
6107	6296	dir: fd_t,
6108	6297	path: [*:0]const u8,
6109	6298	times: ?*const [2]linux.timespec,
...	...	@@ -6111,7 +6300,7 @@ fn utimensat(
6111	6300	) File.SetTimestampsError!void {
6112	6301	_ = ev;
6113	6302	while (true) {
6114		try cancel_region.await(.nothing);
	6303	try sync.cancel_region.await(.nothing);
6115	6304	switch (linux.errno(linux.utimensat(dir, path, times, flags))) {
6116	6305	.SUCCESS => return,
6117	6306	.INTR => continue,

lib/std/Io/Threaded.zig+3-4

...	...	@@ -1543,7 +1543,7 @@ pub const InitOptions = struct {
1543	1543	/// this limit, calls to `Io.async` when all threads are busy run the task
1544	1544	/// immediately.
1545	1545	///
1546		/// Defaults to a number equal to logical CPU cores.
	1546	/// Defaults to one less than the number of logical CPU cores.
1547	1547	///
1548	1548	/// Protected by `Threaded.mutex` once the I/O instance is already in use. See
1549	1549	/// `setAsyncLimit`.
...	...	@@ -1552,8 +1552,7 @@ pub const InitOptions = struct {
1552	1552	/// tasks. Until this limit, calls to `Io.concurrent` will increase the thread
1553	1553	/// pool size.
1554	1554	///
1555		/// concurrent tasks. After this number, calls to `Io.concurrent` return
1556		/// `error.ConcurrencyUnavailable`.
	1555	/// After this number, calls to `Io.concurrent` return `error.ConcurrencyUnavailable`.
1557	1556	concurrent_limit: Io.Limit = .unlimited,
1558	1557	/// Affects the following operations:
1559	1558	/// * `processExecutablePath` on OpenBSD and Haiku.
...	...	@@ -1562,7 +1561,7 @@ pub const InitOptions = struct {
1562	1561	/// * `fileIsTty`
1563	1562	/// * `processExecutablePath` on OpenBSD and Haiku (observes "PATH").
1564	1563	/// * `processSpawn`, `processSpawnPath`, `processReplace`, `processReplacePath`
1565		environ: process.Environ,
	1564	environ: process.Environ = .empty,
1566	1565	/// If set to `true`, `File.MemoryMap` APIs will always take the fallback path.
1567	1566	disable_memory_mapping: bool = false,
1568	1567	};