Commit b01244d225eb35eac9e06f0677e6b3fc212b4b26 | zig

author	Jacob Young <jacobly@ziglang.org> 2025-03-30 15:13:41-04:00
committer	Andrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log	b01244d225eb35eac9e06f0677e6b3fc212b4b26
tree	bb249ab216c29e9b77dd348cc26f8d600635f3f3
parent	b37126bc086059b4f7a0fc29dfa2ce30b2f3458c

Io: implement sleep and fix cancel bugs

5 files changed, 451 insertions(+), 155 deletions(-)

lib
- std
  - Io.zig
  - Io
    EventLoop.zig
  - Thread
    Pool.zig
  - start.zig
  - std.zig

lib/std/Io.zig+43-6

...	...	@@ -579,7 +579,6 @@ pub const VTable = struct {
579	579	context_alignment: std.mem.Alignment,
580	580	start: const fn (context: const anyopaque, result: *anyopaque) void,
581	581	) ?*AnyFuture,
582
583	582	/// This function is only called when `async` returns a non-null value.
584	583	///
585	584	/// Thread-safe.
...	...	@@ -609,7 +608,6 @@ pub const VTable = struct {
609	608	result: []u8,
610	609	result_alignment: std.mem.Alignment,
611	610	) void,
612
613	611	/// Returns whether the current thread of execution is known to have
614	612	/// been requested to cancel.
615	613	///
...	...	@@ -619,8 +617,11 @@ pub const VTable = struct {
619	617	createFile: const fn (?anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
620	618	openFile: const fn (?anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
621	619	closeFile: const fn (?anyopaque, fs.File) void,
622		read: const fn (?anyopaque, file: fs.File, buffer: []u8) FileReadError!usize,
623		write: const fn (?anyopaque, file: fs.File, buffer: []const u8) FileWriteError!usize,
	620	pread: const fn (?anyopaque, file: fs.File, buffer: []u8, offset: std.posix.off_t) FilePReadError!usize,
	621	pwrite: const fn (?anyopaque, file: fs.File, buffer: []const u8, offset: std.posix.off_t) FilePWriteError!usize,
	622
	623	now: const fn (?anyopaque, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp,
	624	sleep: const fn (?anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,
624	625	};
625	626
626	627	pub const OpenFlags = fs.File.OpenFlags;
...	...	@@ -628,7 +629,27 @@ pub const CreateFlags = fs.File.CreateFlags;
628	629
629	630	pub const FileOpenError = fs.File.OpenError \|\| error{AsyncCancel};
630	631	pub const FileReadError = fs.File.ReadError \|\| error{AsyncCancel};
	632	pub const FilePReadError = fs.File.PReadError \|\| error{AsyncCancel};
631	633	pub const FileWriteError = fs.File.WriteError \|\| error{AsyncCancel};
	634	pub const FilePWriteError = fs.File.PWriteError \|\| error{AsyncCancel};
	635
	636	pub const Timestamp = enum(i96) {
	637	_,
	638
	639	pub fn durationTo(from: Timestamp, to: Timestamp) i96 {
	640	return @intFromEnum(to) - @intFromEnum(from);
	641	}
	642
	643	pub fn addDuration(from: Timestamp, duration: i96) Timestamp {
	644	return @enumFromInt(@intFromEnum(from) + duration);
	645	}
	646	};
	647	pub const Deadline = union(enum) {
	648	nanoseconds: i96,
	649	timestamp: Timestamp,
	650	};
	651	pub const ClockGetTimeError = std.posix.ClockGetTimeError \|\| error{AsyncCancel};
	652	pub const SleepError = error{ UnsupportedClock, Unexpected, AsyncCancel };
632	653
633	654	pub const AnyFuture = opaque {};
634	655
...	...	@@ -694,11 +715,19 @@ pub fn closeFile(io: Io, file: fs.File) void {
694	715	}
695	716
696	717	pub fn read(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {
697		return io.vtable.read(io.userdata, file, buffer);
	718	return @errorCast(io.pread(file, buffer, -1));
	719	}
	720
	721	pub fn pread(io: Io, file: fs.File, buffer: []u8, offset: std.posix.off_t) FilePReadError!usize {
	722	return io.vtable.pread(io.userdata, file, buffer, offset);
698	723	}
699	724
700	725	pub fn write(io: Io, file: fs.File, buffer: []const u8) FileWriteError!usize {
701		return io.vtable.write(io.userdata, file, buffer);
	726	return @errorCast(io.pwrite(file, buffer, -1));
	727	}
	728
	729	pub fn pwrite(io: Io, file: fs.File, buffer: []const u8, offset: std.posix.off_t) FilePWriteError!usize {
	730	return io.vtable.pwrite(io.userdata, file, buffer, offset);
702	731	}
703	732
704	733	pub fn writeAll(io: Io, file: fs.File, bytes: []const u8) FileWriteError!void {
...	...	@@ -717,3 +746,11 @@ pub fn readAll(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {
717	746	}
718	747	return index;
719	748	}
	749
	750	pub fn now(io: Io, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp {
	751	return io.vtable.now(io.userdata, clockid);
	752	}
	753
	754	pub fn sleep(io: Io, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void {
	755	return io.vtable.sleep(io.userdata, clockid, deadline);
	756	}

lib/std/Io/EventLoop.zig+291-121

...	...	@@ -31,10 +31,12 @@ const Thread = struct {
31	31	idle_search_index: u32,
32	32	steal_ready_search_index: u32,
33	33
34		threadlocal var index: u32 = undefined;
	34	const canceling: ?*Thread = @ptrFromInt(@alignOf(Thread));
35	35
36		fn current(el: EventLoop) Thread {
37		return &el.threads.allocated[index];
	36	threadlocal var self: *Thread = undefined;
	37
	38	fn current() *Thread {
	39	return self;
38	40	}
39	41
40	42	fn currentFiber(thread: Thread) Fiber {
...	...	@@ -52,10 +54,9 @@ const Fiber = struct {
52	54	context: Context,
53	55	awaiter: ?*Fiber,
54	56	queue_next: ?*Fiber,
55		can_cancel: bool,
56		canceled: bool,
	57	cancel_thread: ?*Thread,
57	58
58		const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));
	59	const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
59	60
60	61	const max_result_align: Alignment = .@"16";
61	62	const max_result_size = max_result_align.forward(64);
...	...	@@ -75,7 +76,7 @@ const Fiber = struct {
75	76	);
76	77
77	78	fn allocate(el: EventLoop) error{OutOfMemory}!Fiber {
78		const thread: *Thread = .current(el);
	79	const thread: *Thread = .current();
79	80	if (thread.free_queue) \|free_fiber\| {
80	81	thread.free_queue = free_fiber.queue_next;
81	82	free_fiber.queue_next = null;
...	...	@@ -101,6 +102,40 @@ const Fiber = struct {
101	102	return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
102	103	}
103	104
	105	fn enterCancelRegion(fiber: Fiber, thread: Thread) error{AsyncCancel}!void {
	106	if (@cmpxchgStrong(
	107	?*Thread,
	108	&fiber.cancel_thread,
	109	null,
	110	thread,
	111	.acq_rel,
	112	.acquire,
	113	)) \|cancel_thread\| {
	114	assert(cancel_thread == Thread.canceling);
	115	return error.AsyncCancel;
	116	}
	117	}
	118
	119	fn exitCancelRegion(fiber: Fiber, thread: Thread) void {
	120	if (@cmpxchgStrong(
	121	?*Thread,
	122	&fiber.cancel_thread,
	123	thread,
	124	null,
	125	.acq_rel,
	126	.acquire,
	127	)) \|cancel_thread\| assert(cancel_thread == Thread.canceling);
	128	}
	129
	130	fn recycle(fiber: *Fiber) void {
	131	const thread: *Thread = .current();
	132	std.log.debug("recyling {*}", .{fiber});
	133	assert(fiber.queue_next == null);
	134	@memset(fiber.allocatedSlice(), undefined);
	135	fiber.queue_next = thread.free_queue;
	136	thread.free_queue = fiber;
	137	}
	138
104	139	const Queue = struct { head: Fiber, tail: Fiber };
105	140	};
106	141
...	...	@@ -110,13 +145,18 @@ pub fn io(el: *EventLoop) Io {
110	145	.vtable = &.{
111	146	.@"async" = @"async",
112	147	.@"await" = @"await",
	148
113	149	.cancel = cancel,
114	150	.cancelRequested = cancelRequested,
	151
115	152	.createFile = createFile,
116	153	.openFile = openFile,
117	154	.closeFile = closeFile,
118		.read = read,
119		.write = write,
	155	.pread = pread,
	156	.pwrite = pwrite,
	157
	158	.now = now,
	159	.sleep = sleep,
120	160	},
121	161	};
122	162	}
...	...	@@ -133,8 +173,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
133	173	.context = undefined,
134	174	.awaiter = null,
135	175	.queue_next = null,
136		.can_cancel = false,
137		.canceled = false,
	176	.cancel_thread = null,
138	177	},
139	178	.threads = .{
140	179	.allocated = @ptrCast(allocated_slice[0..threads_size]),
...	...	@@ -142,8 +181,8 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
142	181	.active = 1,
143	182	},
144	183	};
145		Thread.index = 0;
146	184	const main_thread = &el.threads.allocated[0];
	185	Thread.self = main_thread;
147	186	const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
148	187	(idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
149	188	main_thread.* = .{
...	...	@@ -168,24 +207,22 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
168	207	pub fn deinit(el: *EventLoop) void {
169	208	const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
170	209	for (el.threads.allocated[0..active_threads]) \|*thread\|
171		assert(@atomicLoad(?*Fiber, &thread.ready_queue, .unordered) == null); // pending async
	210	assert(@atomicLoad(?*Fiber, &thread.ready_queue, .acquire) == null); // pending async
172	211	el.yield(null, .exit);
	212	for (el.threads.allocated[0..active_threads]) \|*thread\| while (thread.free_queue) \|free_fiber\| {
	213	thread.free_queue = free_fiber.queue_next;
	214	free_fiber.queue_next = null;
	215	el.gpa.free(free_fiber.allocatedSlice());
	216	};
173	217	const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.allocated.ptr));
174	218	const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
175		for (el.threads.allocated[1..active_threads]) \|*thread\| {
176		thread.thread.join();
177		while (thread.free_queue) \|free_fiber\| {
178		thread.free_queue = free_fiber.queue_next;
179		free_fiber.queue_next = null;
180		el.gpa.free(free_fiber.allocatedSlice());
181		}
182		}
	219	for (el.threads.allocated[1..active_threads]) \|thread\| thread.thread.join();
183	220	el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
184	221	el.* = undefined;
185	222	}
186	223
187	224	fn yield(el: EventLoop, maybe_ready_fiber: ?Fiber, pending_task: SwitchMessage.PendingTask) void {
188		const thread: *Thread = .current(el);
	225	const thread: *Thread = .current();
189	226	const ready_context: *Context = if (maybe_ready_fiber) \|ready_fiber\|
190	227	&ready_fiber.context
191	228	else if (thread.ready_queue) \|ready_fiber\| ready_context: {
...	...	@@ -198,6 +235,7 @@ fn yield(el: EventLoop, maybe_ready_fiber: ?Fiber, pending_task: SwitchMessage
198	235	defer thread.steal_ready_search_index += 1;
199	236	if (thread.steal_ready_search_index == ready_threads) thread.steal_ready_search_index = 0;
200	237	const steal_ready_search_thread = &el.threads.allocated[thread.steal_ready_search_index];
	238	if (steal_ready_search_thread == thread) continue;
201	239	const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
202	240	if (@cmpxchgWeak(
203	241	?*Fiber,
...	...	@@ -236,6 +274,7 @@ fn schedule(el: EventLoop, thread: Thread, ready_queue: Fiber.Queue) void {
236	274	defer thread.idle_search_index += 1;
237	275	if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;
238	276	const idle_search_thread = &el.threads.allocated[thread.idle_search_index];
	277	if (idle_search_thread == thread) continue;
239	278	if (@cmpxchgWeak(
240	279	?*Fiber,
241	280	&idle_search_thread.ready_queue,
...	...	@@ -249,11 +288,11 @@ fn schedule(el: EventLoop, thread: Thread, ready_queue: Fiber.Queue) void {
249	288	.flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
250	289	.ioprio = 0,
251	290	.fd = idle_search_thread.io_uring.fd,
252		.off = @intFromEnum(Completion.Key.wakeup),
	291	.off = @intFromEnum(Completion.UserData.wakeup),
253	292	.addr = 0,
254	293	.len = 0,
255	294	.rw_flags = 0,
256		.user_data = @intFromEnum(Completion.Key.wakeup),
	295	.user_data = @intFromEnum(Completion.UserData.wakeup),
257	296	.buf_index = 0,
258	297	.personality = 0,
259	298	.splice_fd_in = 0,
...	...	@@ -314,15 +353,6 @@ fn schedule(el: EventLoop, thread: Thread, ready_queue: Fiber.Queue) void {
314	353	)) \|old_head\| ready_queue.tail.queue_next = old_head;
315	354	}
316	355
317		fn recycle(el: EventLoop, fiber: Fiber) void {
318		const thread: *Thread = .current(el);
319		std.log.debug("recyling {*}", .{fiber});
320		assert(fiber.queue_next == null);
321		@memset(fiber.allocatedSlice(), undefined);
322		fiber.queue_next = thread.free_queue;
323		thread.free_queue = fiber;
324		}
325
326	356	fn mainIdle(el: EventLoop, message: const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {
327	357	message.handle(el);
328	358	const thread: *Thread = &el.threads.allocated[0];
...	...	@@ -332,17 +362,16 @@ fn mainIdle(el: EventLoop, message: const SwitchMessage) callconv(.withStackAl
332	362	}
333	363
334	364	fn threadEntry(el: *EventLoop, index: u32) void {
335		Thread.index = index;
336	365	const thread: *Thread = &el.threads.allocated[index];
	366	Thread.self = thread;
337	367	std.log.debug("created thread idle {*}", .{&thread.idle_context});
338	368	el.idle(thread);
339	369	}
340	370
341	371	const Completion = struct {
342		const Key = enum(usize) {
	372	const UserData = enum(usize) {
343	373	unused,
344	374	wakeup,
345		cancel,
346	375	cleanup,
347	376	exit,
348	377	/// *Fiber
...	...	@@ -369,26 +398,43 @@ fn idle(el: EventLoop, thread: Thread) void {
369	398	break :cqes_len 0;
370	399	},
371	400	else => \|e\| @panic(@errorName(e)),
372		}]) \|cqe\| switch (@as(Completion.Key, @enumFromInt(cqe.user_data))) {
	401	}]) \|cqe\| switch (@as(Completion.UserData, @enumFromInt(cqe.user_data))) {
373	402	.unused => unreachable, // bad submission queued?
374	403	.wakeup => {},
375		.cancel => {},
376	404	.cleanup => @panic("failed to notify other threads that we are exiting"),
377	405	.exit => {
378	406	assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
379	407	return;
380	408	},
381		_ => {
382		const fiber: *Fiber = @ptrFromInt(cqe.user_data);
383		assert(fiber.queue_next == null);
384		fiber.resultPointer(Completion).* = .{
385		.result = cqe.res,
386		.flags = cqe.flags,
387		};
388		if (maybe_ready_fiber == null) maybe_ready_fiber = fiber else if (maybe_ready_queue) \|*ready_queue\| {
389		ready_queue.tail.queue_next = fiber;
390		ready_queue.tail = fiber;
391		} else maybe_ready_queue = .{ .head = fiber, .tail = fiber };
	409	_ => switch (errno(cqe.res)) {
	410	.INTR => getSqe(&thread.io_uring).* = .{
	411	.opcode = .ASYNC_CANCEL,
	412	.flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
	413	.ioprio = 0,
	414	.fd = 0,
	415	.off = 0,
	416	.addr = cqe.user_data,
	417	.len = 0,
	418	.rw_flags = 0,
	419	.user_data = @intFromEnum(Completion.UserData.wakeup),
	420	.buf_index = 0,
	421	.personality = 0,
	422	.splice_fd_in = 0,
	423	.addr3 = 0,
	424	.resv = 0,
	425	},
	426	else => {
	427	const fiber: *Fiber = @ptrFromInt(cqe.user_data);
	428	assert(fiber.queue_next == null);
	429	fiber.resultPointer(Completion).* = .{
	430	.result = cqe.res,
	431	.flags = cqe.flags,
	432	};
	433	if (maybe_ready_fiber == null) maybe_ready_fiber = fiber else if (maybe_ready_queue) \|*ready_queue\| {
	434	ready_queue.tail.queue_next = fiber;
	435	ready_queue.tail = fiber;
	436	} else maybe_ready_queue = .{ .head = fiber, .tail = fiber };
	437	},
392	438	},
393	439	};
394	440	if (maybe_ready_queue) \|ready_queue\| el.schedule(thread, ready_queue);
...	...	@@ -409,7 +455,7 @@ const SwitchMessage = struct {
409	455	};
410	456
411	457	fn handle(message: const SwitchMessage, el: EventLoop) void {
412		const thread: *Thread = .current(el);
	458	const thread: *Thread = .current();
413	459	thread.current_context = message.contexts.ready;
414	460	switch (message.pending_task) {
415	461	.nothing => {},
...	...	@@ -429,11 +475,11 @@ const SwitchMessage = struct {
429	475	.flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
430	476	.ioprio = 0,
431	477	.fd = each_thread.io_uring.fd,
432		.off = @intFromEnum(Completion.Key.exit),
	478	.off = @intFromEnum(Completion.UserData.exit),
433	479	.addr = 0,
434	480	.len = 0,
435	481	.rw_flags = 0,
436		.user_data = @intFromEnum(Completion.Key.cleanup),
	482	.user_data = @intFromEnum(Completion.UserData.cleanup),
437	483	.buf_index = 0,
438	484	.personality = 0,
439	485	.splice_fd_in = 0,
...	...	@@ -544,6 +590,7 @@ fn @"async"(
544	590	start(context.ptr, result.ptr);
545	591	return null;
546	592	};
	593	errdefer fiber.recycle();
547	594	std.log.debug("allocated {*}", .{fiber});
548	595
549	596	const closure: *AsyncClosure = @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
...	...	@@ -560,8 +607,7 @@ fn @"async"(
560	607	},
561	608	.awaiter = null,
562	609	.queue_next = null,
563		.can_cancel = false,
564		.canceled = false,
	610	.cancel_thread = null,
565	611	};
566	612	closure.* = .{
567	613	.event_loop = event_loop,
...	...	@@ -571,7 +617,7 @@ fn @"async"(
571	617	};
572	618	@memcpy(closure.contextPointer(), context);
573	619
574		event_loop.schedule(.current(event_loop), .{ .head = fiber, .tail = fiber });
	620	event_loop.schedule(.current(), .{ .head = fiber, .tail = fiber });
575	621	return @ptrCast(fiber);
576	622	}
577	623
...	...	@@ -585,7 +631,7 @@ fn @"await"(
585	631	const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
586	632	if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
587	633	@memcpy(result, future_fiber.resultBytes(result_alignment));
588		event_loop.recycle(future_fiber);
	634	future_fiber.recycle();
589	635	}
590	636
591	637	fn cancel(
...	...	@@ -594,35 +640,37 @@ fn cancel(
594	640	result: []u8,
595	641	result_alignment: Alignment,
596	642	) void {
597		const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
598	643	const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
599		@atomicStore(bool, &future_fiber.canceled, true, .release);
600		if (@atomicLoad(bool, &future_fiber.can_cancel, .acquire)) {
601		const thread: *Thread = .current(event_loop);
602		getSqe(&thread.io_uring).* = .{
603		.opcode = .ASYNC_CANCEL,
	644	if (@atomicRmw(
	645	?*Thread,
	646	&future_fiber.cancel_thread,
	647	.Xchg,
	648	Thread.canceling,
	649	.acq_rel,
	650	)) \|cancel_thread\| if (cancel_thread != Thread.canceling) {
	651	getSqe(&Thread.current().io_uring).* = .{
	652	.opcode = .MSG_RING,
604	653	.flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
605	654	.ioprio = 0,
606		.fd = 0,
607		.off = 0,
608		.addr = @intFromPtr(future_fiber),
609		.len = 0,
	655	.fd = cancel_thread.io_uring.fd,
	656	.off = @intFromPtr(future_fiber),
	657	.addr = 0,
	658	.len = @bitCast(-@as(i32, @intFromEnum(std.os.linux.E.INTR))),
610	659	.rw_flags = 0,
611		.user_data = @intFromEnum(Completion.Key.cancel),
	660	.user_data = @intFromEnum(Completion.UserData.cleanup),
612	661	.buf_index = 0,
613	662	.personality = 0,
614	663	.splice_fd_in = 0,
615	664	.addr3 = 0,
616	665	.resv = 0,
617	666	};
618		}
	667	};
619	668	@"await"(userdata, any_future, result, result_alignment);
620	669	}
621	670
622	671	fn cancelRequested(userdata: ?*anyopaque) bool {
623		const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
624		const thread: *Thread = .current(event_loop);
625		return thread.currentFiber().canceled;
	672	_ = userdata;
	673	return @atomicLoad(?*Thread, &Thread.current().currentFiber().cancel_thread, .acquire) == Thread.canceling;
626	674	}
627	675
628	676	pub fn createFile(
...	...	@@ -632,6 +680,10 @@ pub fn createFile(
632	680	flags: Io.CreateFlags,
633	681	) Io.FileOpenError!std.fs.File {
634	682	const el: *EventLoop = @alignCast(@ptrCast(userdata));
	683	const thread: *Thread = .current();
	684	const iou = &thread.io_uring;
	685	const fiber = thread.currentFiber();
	686	try fiber.enterCancelRegion(thread);
635	687
636	688	const posix = std.posix;
637	689	const sub_path_c = try posix.toPosixPath(sub_path);
...	...	@@ -670,23 +722,30 @@ pub fn createFile(
670	722	@panic("TODO");
671	723	}
672	724
673		const thread: *Thread = .current(el);
674		const iou = &thread.io_uring;
675		const fiber = thread.currentFiber();
676		if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
677
678		const sqe = getSqe(iou);
679		sqe.prep_openat(dir.fd, &sub_path_c, os_flags, flags.mode);
680		sqe.user_data = @intFromPtr(fiber);
	725	getSqe(iou).* = .{
	726	.opcode = .OPENAT,
	727	.flags = 0,
	728	.ioprio = 0,
	729	.fd = dir.fd,
	730	.off = 0,
	731	.addr = @intFromPtr(&sub_path_c),
	732	.len = @intCast(flags.mode),
	733	.rw_flags = @bitCast(os_flags),
	734	.user_data = @intFromPtr(fiber),
	735	.buf_index = 0,
	736	.personality = 0,
	737	.splice_fd_in = 0,
	738	.addr3 = 0,
	739	.resv = 0,
	740	};
681	741
682		@atomicStore(bool, &fiber.can_cancel, true, .release);
683	742	el.yield(null, .nothing);
684		@atomicStore(bool, &fiber.can_cancel, false, .release);
	743	fiber.exitCancelRegion(thread);
685	744
686	745	const completion = fiber.resultPointer(Completion);
687	746	switch (errno(completion.result)) {
688	747	.SUCCESS => return .{ .handle = completion.result },
689		.INTR => @panic("TODO is this reachable?"),
	748	.INTR => unreachable,
690	749	.CANCELED => return error.AsyncCancel,
691	750
692	751	.FAULT => unreachable,
...	...	@@ -723,10 +782,10 @@ pub fn openFile(
723	782	flags: Io.OpenFlags,
724	783	) Io.FileOpenError!std.fs.File {
725	784	const el: *EventLoop = @alignCast(@ptrCast(userdata));
726		const thread: *Thread = .current(el);
	785	const thread: *Thread = .current();
727	786	const iou = &thread.io_uring;
728	787	const fiber = thread.currentFiber();
729		if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
	788	try fiber.enterCancelRegion(thread);
730	789
731	790	const posix = std.posix;
732	791	const sub_path_c = try posix.toPosixPath(sub_path);
...	...	@@ -771,18 +830,30 @@ pub fn openFile(
771	830	@panic("TODO");
772	831	}
773	832
774		const sqe = getSqe(iou);
775		sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);
776		sqe.user_data = @intFromPtr(fiber);
	833	getSqe(iou).* = .{
	834	.opcode = .OPENAT,
	835	.flags = 0,
	836	.ioprio = 0,
	837	.fd = dir.fd,
	838	.off = 0,
	839	.addr = @intFromPtr(&sub_path_c),
	840	.len = 0,
	841	.rw_flags = @bitCast(os_flags),
	842	.user_data = @intFromPtr(fiber),
	843	.buf_index = 0,
	844	.personality = 0,
	845	.splice_fd_in = 0,
	846	.addr3 = 0,
	847	.resv = 0,
	848	};
777	849
778		@atomicStore(bool, &fiber.can_cancel, true, .release);
779	850	el.yield(null, .nothing);
780		@atomicStore(bool, &fiber.can_cancel, false, .release);
	851	fiber.exitCancelRegion(thread);
781	852
782	853	const completion = fiber.resultPointer(Completion);
783	854	switch (errno(completion.result)) {
784	855	.SUCCESS => return .{ .handle = completion.result },
785		.INTR => @panic("TODO is this reachable?"),
	856	.INTR => unreachable,
786	857	.CANCELED => return error.AsyncCancel,
787	858
788	859	.FAULT => unreachable,
...	...	@@ -814,20 +885,33 @@ pub fn openFile(
814	885
815	886	pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
816	887	const el: *EventLoop = @alignCast(@ptrCast(userdata));
817		const thread: *Thread = .current(el);
	888	const thread: *Thread = .current();
818	889	const iou = &thread.io_uring;
819	890	const fiber = thread.currentFiber();
820	891
821		const sqe = getSqe(iou);
822		sqe.prep_close(file.handle);
823		sqe.user_data = @intFromPtr(fiber);
	892	getSqe(iou).* = .{
	893	.opcode = .CLOSE,
	894	.flags = 0,
	895	.ioprio = 0,
	896	.fd = file.handle,
	897	.off = 0,
	898	.addr = 0,
	899	.len = 0,
	900	.rw_flags = 0,
	901	.user_data = @intFromPtr(fiber),
	902	.buf_index = 0,
	903	.personality = 0,
	904	.splice_fd_in = 0,
	905	.addr3 = 0,
	906	.resv = 0,
	907	};
824	908
825	909	el.yield(null, .nothing);
826	910
827	911	const completion = fiber.resultPointer(Completion);
828	912	switch (errno(completion.result)) {
829	913	.SUCCESS => return,
830		.INTR => @panic("TODO is this reachable?"),
	914	.INTR => unreachable,
831	915	.CANCELED => return,
832	916
833	917	.BADF => unreachable, // Always a race condition.
...	...	@@ -835,25 +919,37 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
835	919	}
836	920	}
837	921
838		pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadError!usize {
	922	pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {
839	923	const el: *EventLoop = @alignCast(@ptrCast(userdata));
840		const thread: *Thread = .current(el);
	924	const thread: *Thread = .current();
841	925	const iou = &thread.io_uring;
842	926	const fiber = thread.currentFiber();
843		if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
844
845		const sqe = getSqe(iou);
846		sqe.prep_read(file.handle, buffer, std.math.maxInt(u64));
847		sqe.user_data = @intFromPtr(fiber);
	927	try fiber.enterCancelRegion(thread);
	928
	929	getSqe(iou).* = .{
	930	.opcode = .READ,
	931	.flags = 0,
	932	.ioprio = 0,
	933	.fd = file.handle,
	934	.off = @bitCast(offset),
	935	.addr = @intFromPtr(buffer.ptr),
	936	.len = @min(buffer.len, 0x7ffff000),
	937	.rw_flags = 0,
	938	.user_data = @intFromPtr(fiber),
	939	.buf_index = 0,
	940	.personality = 0,
	941	.splice_fd_in = 0,
	942	.addr3 = 0,
	943	.resv = 0,
	944	};
848	945
849		@atomicStore(bool, &fiber.can_cancel, true, .release);
850	946	el.yield(null, .nothing);
851		@atomicStore(bool, &fiber.can_cancel, false, .release);
	947	fiber.exitCancelRegion(thread);
852	948
853	949	const completion = fiber.resultPointer(Completion);
854	950	switch (errno(completion.result)) {
855	951	.SUCCESS => return @as(u32, @bitCast(completion.result)),
856		.INTR => @panic("TODO is this reachable?"),
	952	.INTR => unreachable,
857	953	.CANCELED => return error.AsyncCancel,
858	954
859	955	.INVAL => unreachable,
...	...	@@ -868,30 +964,44 @@ pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadE
868	964	.NOTCONN => return error.SocketNotConnected,
869	965	.CONNRESET => return error.ConnectionResetByPeer,
870	966	.TIMEDOUT => return error.ConnectionTimedOut,
	967	.NXIO => return error.Unseekable,
	968	.SPIPE => return error.Unseekable,
	969	.OVERFLOW => return error.Unseekable,
871	970	else => \|err\| return std.posix.unexpectedErrno(err),
872	971	}
873	972	}
874	973
875		pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.FileWriteError!usize {
	974	pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {
876	975	const el: *EventLoop = @alignCast(@ptrCast(userdata));
877
878		const thread: *Thread = .current(el);
	976	const thread: *Thread = .current();
879	977	const iou = &thread.io_uring;
880	978	const fiber = thread.currentFiber();
881		if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
882
883		const sqe = getSqe(iou);
884		sqe.prep_write(file.handle, buffer, std.math.maxInt(u64));
885		sqe.user_data = @intFromPtr(fiber);
	979	try fiber.enterCancelRegion(thread);
	980
	981	getSqe(iou).* = .{
	982	.opcode = .WRITE,
	983	.flags = 0,
	984	.ioprio = 0,
	985	.fd = file.handle,
	986	.off = @bitCast(offset),
	987	.addr = @intFromPtr(buffer.ptr),
	988	.len = @min(buffer.len, 0x7ffff000),
	989	.rw_flags = 0,
	990	.user_data = @intFromPtr(fiber),
	991	.buf_index = 0,
	992	.personality = 0,
	993	.splice_fd_in = 0,
	994	.addr3 = 0,
	995	.resv = 0,
	996	};
886	997
887		@atomicStore(bool, &fiber.can_cancel, true, .release);
888	998	el.yield(null, .nothing);
889		@atomicStore(bool, &fiber.can_cancel, false, .release);
	999	fiber.exitCancelRegion(thread);
890	1000
891	1001	const completion = fiber.resultPointer(Completion);
892	1002	switch (errno(completion.result)) {
893	1003	.SUCCESS => return @as(u32, @bitCast(completion.result)),
894		.INTR => @panic("TODO is this reachable?"),
	1004	.INTR => unreachable,
895	1005	.CANCELED => return error.AsyncCancel,
896	1006
897	1007	.INVAL => return error.InvalidArgument,
...	...	@@ -907,17 +1017,77 @@ pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.Fi
907	1017	.ACCES => return error.AccessDenied,
908	1018	.PERM => return error.PermissionDenied,
909	1019	.PIPE => return error.BrokenPipe,
910		.CONNRESET => return error.ConnectionResetByPeer,
	1020	.NXIO => return error.Unseekable,
	1021	.SPIPE => return error.Unseekable,
	1022	.OVERFLOW => return error.Unseekable,
911	1023	.BUSY => return error.DeviceBusy,
912		.NXIO => return error.NoDevice,
	1024	.CONNRESET => return error.ConnectionResetByPeer,
913	1025	.MSGSIZE => return error.MessageTooBig,
914	1026	else => \|err\| return std.posix.unexpectedErrno(err),
915	1027	}
916	1028	}
917	1029
918		fn errno(signed: i32) std.posix.E {
919		const int = if (signed > -4096 and signed < 0) -signed else 0;
920		return @enumFromInt(int);
	1030	pub fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
	1031	_ = userdata;
	1032	const timespec = try std.posix.clock_gettime(clockid);
	1033	return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
	1034	}
	1035
	1036	pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
	1037	const el: *EventLoop = @alignCast(@ptrCast(userdata));
	1038	const thread: *Thread = .current();
	1039	const iou = &thread.io_uring;
	1040	const fiber = thread.currentFiber();
	1041	try fiber.enterCancelRegion(thread);
	1042
	1043	const deadline_nanoseconds: i96 = switch (deadline) {
	1044	.nanoseconds => \|nanoseconds\| nanoseconds,
	1045	.timestamp => \|timestamp\| @intFromEnum(timestamp),
	1046	};
	1047	const timespec: std.os.linux.kernel_timespec = .{
	1048	.sec = @intCast(@divFloor(deadline_nanoseconds, std.time.ns_per_s)),
	1049	.nsec = @intCast(@mod(deadline_nanoseconds, std.time.ns_per_s)),
	1050	};
	1051	getSqe(iou).* = .{
	1052	.opcode = .TIMEOUT,
	1053	.flags = 0,
	1054	.ioprio = 0,
	1055	.fd = 0,
	1056	.off = 0,
	1057	.addr = @intFromPtr(&timespec),
	1058	.len = 1,
	1059	.rw_flags = @as(u32, switch (deadline) {
	1060	.nanoseconds => 0,
	1061	.timestamp => std.os.linux.IORING_TIMEOUT_ABS,
	1062	}) \| @as(u32, switch (clockid) {
	1063	.REALTIME => std.os.linux.IORING_TIMEOUT_REALTIME,
	1064	.MONOTONIC => 0,
	1065	.BOOTTIME => std.os.linux.IORING_TIMEOUT_BOOTTIME,
	1066	else => return error.UnsupportedClock,
	1067	}),
	1068	.user_data = @intFromPtr(fiber),
	1069	.buf_index = 0,
	1070	.personality = 0,
	1071	.splice_fd_in = 0,
	1072	.addr3 = 0,
	1073	.resv = 0,
	1074	};
	1075
	1076	el.yield(null, .nothing);
	1077	fiber.exitCancelRegion(thread);
	1078
	1079	const completion = fiber.resultPointer(Completion);
	1080	switch (errno(completion.result)) {
	1081	.SUCCESS, .TIME => return,
	1082	.INTR => unreachable,
	1083	.CANCELED => return error.AsyncCancel,
	1084
	1085	else => \|err\| return std.posix.unexpectedErrno(err),
	1086	}
	1087	}
	1088
	1089	fn errno(signed: i32) std.os.linux.E {
	1090	return .init(@bitCast(@as(isize, signed)));
921	1091	}
922	1092
923	1093	fn getSqe(iou: IoUring) std.os.linux.io_uring_sqe {

lib/std/Thread/Pool.zig+98-10

...	...	@@ -332,13 +332,18 @@ pub fn io(pool: *Pool) Io {
332	332	.vtable = &.{
333	333	.@"async" = @"async",
334	334	.@"await" = @"await",
	335
335	336	.cancel = cancel,
336	337	.cancelRequested = cancelRequested,
	338
337	339	.createFile = createFile,
338	340	.openFile = openFile,
339	341	.closeFile = closeFile,
340		.read = read,
341		.write = write,
	342	.pread = pread,
	343	.pwrite = pwrite,
	344
	345	.now = now,
	346	.sleep = sleep,
342	347	},
343	348	};
344	349	}
...	...	@@ -347,15 +352,44 @@ const AsyncClosure = struct {
347	352	func: const fn (context: anyopaque, result: *anyopaque) void,
348	353	runnable: Runnable = .{ .runFn = runFn },
349	354	reset_event: std.Thread.ResetEvent,
350		cancel_flag: bool,
	355	cancel_tid: std.Thread.Id,
351	356	context_offset: usize,
352	357	result_offset: usize,
353	358
	359	const canceling_tid: std.Thread.Id = switch (@typeInfo(std.Thread.Id)) {
	360	.int => \|int_info\| switch (int_info.signedness) {
	361	.signed => -1,
	362	.unsigned => std.math.maxInt(std.Thread.Id),
	363	},
	364	.pointer => @ptrFromInt(std.math.maxInt(usize)),
	365	else => @compileError("unsupported std.Thread.Id: " ++ @typeName(std.Thread.Id)),
	366	};
	367
354	368	fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void {
355	369	const closure: *AsyncClosure = @alignCast(@fieldParentPtr("runnable", runnable));
	370	const tid = std.Thread.getCurrentId();
	371	if (@cmpxchgStrong(
	372	std.Thread.Id,
	373	&closure.cancel_tid,
	374	0,
	375	tid,
	376	.acq_rel,
	377	.acquire,
	378	)) \|cancel_tid\| {
	379	assert(cancel_tid == canceling_tid);
	380	return;
	381	}
356	382	current_closure = closure;
357	383	closure.func(closure.contextPointer(), closure.resultPointer());
358	384	current_closure = null;
	385	if (@cmpxchgStrong(
	386	std.Thread.Id,
	387	&closure.cancel_tid,
	388	tid,
	389	0,
	390	.acq_rel,
	391	.acquire,
	392	)) \|cancel_tid\| assert(cancel_tid == canceling_tid);
359	393	closure.reset_event.set();
360	394	}
361	395
...	...	@@ -414,7 +448,7 @@ fn @"async"(
414	448	.context_offset = context_offset,
415	449	.result_offset = result_offset,
416	450	.reset_event = .{},
417		.cancel_flag = false,
	451	.cancel_tid = 0,
418	452	};
419	453	@memcpy(closure.contextPointer()[0..context.len], context);
420	454	pool.run_queue.prepend(&closure.runnable.node);
...	...	@@ -456,7 +490,23 @@ fn cancel(
456	490	_ = result_alignment;
457	491	const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
458	492	const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));
459		@atomicStore(bool, &closure.cancel_flag, true, .seq_cst);
	493	switch (@atomicRmw(
	494	std.Thread.Id,
	495	&closure.cancel_tid,
	496	.Xchg,
	497	AsyncClosure.canceling_tid,
	498	.acq_rel,
	499	)) {
	500	0, AsyncClosure.canceling_tid => {},
	501	else => \|cancel_tid\| switch (builtin.os.tag) {
	502	.linux => _ = std.os.linux.tgkill(
	503	std.os.linux.getpid(),
	504	@bitCast(cancel_tid),
	505	std.posix.SIG.IO,
	506	),
	507	else => {},
	508	},
	509	}
460	510	closure.waitAndFree(pool.allocator, result);
461	511	}
462	512
...	...	@@ -464,7 +514,7 @@ fn cancelRequested(userdata: ?*anyopaque) bool {
464	514	const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
465	515	_ = pool;
466	516	const closure = current_closure orelse return false;
467		return @atomicLoad(bool, &closure.cancel_flag, .unordered);
	517	return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == AsyncClosure.canceling_tid;
468	518	}
469	519
470	520	fn checkCancel(pool: *Pool) error{AsyncCancel}!void {
...	...	@@ -499,14 +549,52 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
499	549	return file.close();
500	550	}
501	551
502		pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadError!usize {
	552	pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {
503	553	const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
504	554	try pool.checkCancel();
505		return file.read(buffer);
	555	return switch (offset) {
	556	-1 => file.read(buffer),
	557	else => file.pread(buffer, @bitCast(offset)),
	558	};
506	559	}
507	560
508		pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.FileWriteError!usize {
	561	pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {
509	562	const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
510	563	try pool.checkCancel();
511		return file.write(buffer);
	564	return switch (offset) {
	565	-1 => file.write(buffer),
	566	else => file.pwrite(buffer, @bitCast(offset)),
	567	};
	568	}
	569
	570	pub fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
	571	const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
	572	try pool.checkCancel();
	573	const timespec = try std.posix.clock_gettime(clockid);
	574	return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
	575	}
	576
	577	pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
	578	const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
	579	const deadline_nanoseconds: i96 = switch (deadline) {
	580	.nanoseconds => \|nanoseconds\| nanoseconds,
	581	.timestamp => \|timestamp\| @intFromEnum(timestamp),
	582	};
	583	var timespec: std.posix.timespec = .{
	584	.sec = @intCast(@divFloor(deadline_nanoseconds, std.time.ns_per_s)),
	585	.nsec = @intCast(@mod(deadline_nanoseconds, std.time.ns_per_s)),
	586	};
	587	while (true) {
	588	try pool.checkCancel();
	589	switch (std.os.linux.E.init(std.os.linux.clock_nanosleep(clockid, .{ .ABSTIME = switch (deadline) {
	590	.nanoseconds => false,
	591	.timestamp => true,
	592	} }, &timespec, &timespec))) {
	593	.SUCCESS => return,
	594	.FAULT => unreachable,
	595	.INTR => {},
	596	.INVAL => return error.UnsupportedClock,
	597	else => \|err\| return std.posix.unexpectedErrno(err),
	598	}
	599	}
512	600	}

lib/std/start.zig+16-18

...	...	@@ -581,8 +581,8 @@ inline fn callMainWithArgs(argc: usize, argv: [][:0]u8, envp: [][*:0]u8) u8 {
581	581	std.os.argv = argv[0..argc];
582	582	std.os.environ = envp;
583	583
	584	maybeIgnoreSignals();
584	585	std.debug.maybeEnableSegfaultHandler();
585		maybeIgnoreSigpipe();
586	586
587	587	return callMain();
588	588	}
...	...	@@ -687,8 +687,8 @@ pub fn call_wWinMain() std.os.windows.INT {
687	687	return root.wWinMain(hInstance, null, lpCmdLine, nCmdShow);
688	688	}
689	689
690		fn maybeIgnoreSigpipe() void {
691		const have_sigpipe_support = switch (builtin.os.tag) {
	690	fn maybeIgnoreSignals() void {
	691	switch (builtin.os.tag) {
692	692	.linux,
693	693	.plan9,
694	694	.solaris,
...	...	@@ -703,22 +703,20 @@ fn maybeIgnoreSigpipe() void {
703	703	.dragonfly,
704	704	.freebsd,
705	705	.serenity,
706		=> true,
707
708		else => false,
709		};
710
711		if (have_sigpipe_support and !std.options.keep_sigpipe) {
712		const posix = std.posix;
713		const act: posix.Sigaction = .{
714		// Set handler to a noop function instead of `SIG.IGN` to prevent
715		// leaking signal disposition to a child process.
716		.handler = .{ .handler = noopSigHandler },
717		.mask = posix.sigemptyset(),
718		.flags = 0,
719		};
720		posix.sigaction(posix.SIG.PIPE, &act, null);
	706	=> {},
	707	else => return,
721	708	}
	709	const posix = std.posix;
	710	const act: posix.Sigaction = .{
	711	// Set handler to a noop function instead of `SIG.IGN` to prevent
	712	// leaking signal disposition to a child process.
	713	.handler = .{ .handler = noopSigHandler },
	714	.mask = posix.sigemptyset(),
	715	.flags = 0,
	716	};
	717	if (!std.options.keep_sigpoll) posix.sigaction(posix.SIG.POLL, &act, null);
	718	if (@hasField(posix.SIG, "IO") and posix.SIG.IO != posix.SIG.POLL and !std.options.keep_sigio) posix.sigaction(posix.SIG.IO, &act, null);
	719	if (!std.options.keep_sigpipe) posix.sigaction(posix.SIG.PIPE, &act, null);
722	720	}
723	721
724	722	fn noopSigHandler(_: i32) callconv(.c) void {}

lib/std/std.zig+3

...	...	@@ -145,6 +145,9 @@ pub const Options = struct {
145	145
146	146	crypto_fork_safety: bool = true,
147	147
	148	keep_sigpoll: bool = false,
	149	keep_sigio: bool = false,
	150
148	151	/// By default Zig disables SIGPIPE by setting a "no-op" handler for it. Set this option
149	152	/// to `true` to prevent that.
150	153	///