authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-03-30 15:13:41-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:38-07:00
log012ef81b8ba63f757366384a7721647481665762
treeffd4c1c92fe44354481086c020dfa25a78757e35
parent1fbc251ccd0c2c362cb793263285997f6c900d25

Io: implement sleep and fix cancel bugs

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

lib/std/Io.zig+43-6
...@@ -937,7 +937,6 @@ pub const VTable = struct {...@@ -937,7 +937,6 @@ pub const VTable = struct {
937 context_alignment: std.mem.Alignment,937 context_alignment: std.mem.Alignment,
938 start: *const fn (context: *const anyopaque, result: *anyopaque) void,938 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
939 ) ?*AnyFuture,939 ) ?*AnyFuture,
940
941 /// This function is only called when `async` returns a non-null value.940 /// This function is only called when `async` returns a non-null value.
942 ///941 ///
943 /// Thread-safe.942 /// Thread-safe.
...@@ -967,7 +966,6 @@ pub const VTable = struct {...@@ -967,7 +966,6 @@ pub const VTable = struct {
967 result: []u8,966 result: []u8,
968 result_alignment: std.mem.Alignment,967 result_alignment: std.mem.Alignment,
969 ) void,968 ) void,
970
971 /// Returns whether the current thread of execution is known to have969 /// Returns whether the current thread of execution is known to have
972 /// been requested to cancel.970 /// been requested to cancel.
973 ///971 ///
...@@ -977,8 +975,11 @@ pub const VTable = struct {...@@ -977,8 +975,11 @@ pub const VTable = struct {
977 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,975 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
978 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,976 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
979 closeFile: *const fn (?*anyopaque, fs.File) void,977 closeFile: *const fn (?*anyopaque, fs.File) void,
980 read: *const fn (?*anyopaque, file: fs.File, buffer: []u8) FileReadError!usize,978 pread: *const fn (?*anyopaque, file: fs.File, buffer: []u8, offset: std.posix.off_t) FilePReadError!usize,
981 write: *const fn (?*anyopaque, file: fs.File, buffer: []const u8) FileWriteError!usize,979 pwrite: *const fn (?*anyopaque, file: fs.File, buffer: []const u8, offset: std.posix.off_t) FilePWriteError!usize,
980
981 now: *const fn (?*anyopaque, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp,
982 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,
982};983};
983984
984pub const OpenFlags = fs.File.OpenFlags;985pub const OpenFlags = fs.File.OpenFlags;
...@@ -986,7 +987,27 @@ pub const CreateFlags = fs.File.CreateFlags;...@@ -986,7 +987,27 @@ pub const CreateFlags = fs.File.CreateFlags;
986987
987pub const FileOpenError = fs.File.OpenError || error{AsyncCancel};988pub const FileOpenError = fs.File.OpenError || error{AsyncCancel};
988pub const FileReadError = fs.File.ReadError || error{AsyncCancel};989pub const FileReadError = fs.File.ReadError || error{AsyncCancel};
990pub const FilePReadError = fs.File.PReadError || error{AsyncCancel};
989pub const FileWriteError = fs.File.WriteError || error{AsyncCancel};991pub const FileWriteError = fs.File.WriteError || error{AsyncCancel};
992pub const FilePWriteError = fs.File.PWriteError || error{AsyncCancel};
993
994pub const Timestamp = enum(i96) {
995 _,
996
997 pub fn durationTo(from: Timestamp, to: Timestamp) i96 {
998 return @intFromEnum(to) - @intFromEnum(from);
999 }
1000
1001 pub fn addDuration(from: Timestamp, duration: i96) Timestamp {
1002 return @enumFromInt(@intFromEnum(from) + duration);
1003 }
1004};
1005pub const Deadline = union(enum) {
1006 nanoseconds: i96,
1007 timestamp: Timestamp,
1008};
1009pub const ClockGetTimeError = std.posix.ClockGetTimeError || error{AsyncCancel};
1010pub const SleepError = error{ UnsupportedClock, Unexpected, AsyncCancel };
9901011
991pub const AnyFuture = opaque {};1012pub const AnyFuture = opaque {};
9921013
...@@ -1052,11 +1073,19 @@ pub fn closeFile(io: Io, file: fs.File) void {...@@ -1052,11 +1073,19 @@ pub fn closeFile(io: Io, file: fs.File) void {
1052}1073}
10531074
1054pub fn read(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {1075pub fn read(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {
1055 return io.vtable.read(io.userdata, file, buffer);1076 return @errorCast(io.pread(file, buffer, -1));
1077}
1078
1079pub fn pread(io: Io, file: fs.File, buffer: []u8, offset: std.posix.off_t) FilePReadError!usize {
1080 return io.vtable.pread(io.userdata, file, buffer, offset);
1056}1081}
10571082
1058pub fn write(io: Io, file: fs.File, buffer: []const u8) FileWriteError!usize {1083pub fn write(io: Io, file: fs.File, buffer: []const u8) FileWriteError!usize {
1059 return io.vtable.write(io.userdata, file, buffer);1084 return @errorCast(io.pwrite(file, buffer, -1));
1085}
1086
1087pub fn pwrite(io: Io, file: fs.File, buffer: []const u8, offset: std.posix.off_t) FilePWriteError!usize {
1088 return io.vtable.pwrite(io.userdata, file, buffer, offset);
1060}1089}
10611090
1062pub fn writeAll(io: Io, file: fs.File, bytes: []const u8) FileWriteError!void {1091pub fn writeAll(io: Io, file: fs.File, bytes: []const u8) FileWriteError!void {
...@@ -1075,3 +1104,11 @@ pub fn readAll(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {...@@ -1075,3 +1104,11 @@ pub fn readAll(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {
1075 }1104 }
1076 return index;1105 return index;
1077}1106}
1107
1108pub fn now(io: Io, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp {
1109 return io.vtable.now(io.userdata, clockid);
1110}
1111
1112pub fn sleep(io: Io, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void {
1113 return io.vtable.sleep(io.userdata, clockid, deadline);
1114}
lib/std/Io/EventLoop.zig+291-121
...@@ -31,10 +31,12 @@ const Thread = struct {...@@ -31,10 +31,12 @@ const Thread = struct {
31 idle_search_index: u32,31 idle_search_index: u32,
32 steal_ready_search_index: u32,32 steal_ready_search_index: u32,
3333
34 threadlocal var index: u32 = undefined;34 const canceling: ?*Thread = @ptrFromInt(@alignOf(Thread));
3535
36 fn current(el: *EventLoop) *Thread {36 threadlocal var self: *Thread = undefined;
37 return &el.threads.allocated[index];37
38 fn current() *Thread {
39 return self;
38 }40 }
3941
40 fn currentFiber(thread: *Thread) *Fiber {42 fn currentFiber(thread: *Thread) *Fiber {
...@@ -52,10 +54,9 @@ const Fiber = struct {...@@ -52,10 +54,9 @@ const Fiber = struct {
52 context: Context,54 context: Context,
53 awaiter: ?*Fiber,55 awaiter: ?*Fiber,
54 queue_next: ?*Fiber,56 queue_next: ?*Fiber,
55 can_cancel: bool,57 cancel_thread: ?*Thread,
56 canceled: bool,
5758
58 const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));59 const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
5960
60 const max_result_align: Alignment = .@"16";61 const max_result_align: Alignment = .@"16";
61 const max_result_size = max_result_align.forward(64);62 const max_result_size = max_result_align.forward(64);
...@@ -75,7 +76,7 @@ const Fiber = struct {...@@ -75,7 +76,7 @@ const Fiber = struct {
75 );76 );
7677
77 fn allocate(el: *EventLoop) error{OutOfMemory}!*Fiber {78 fn allocate(el: *EventLoop) error{OutOfMemory}!*Fiber {
78 const thread: *Thread = .current(el);79 const thread: *Thread = .current();
79 if (thread.free_queue) |free_fiber| {80 if (thread.free_queue) |free_fiber| {
80 thread.free_queue = free_fiber.queue_next;81 thread.free_queue = free_fiber.queue_next;
81 free_fiber.queue_next = null;82 free_fiber.queue_next = null;
...@@ -101,6 +102,40 @@ const Fiber = struct {...@@ -101,6 +102,40 @@ const Fiber = struct {
101 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));102 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
102 }103 }
103104
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 const Queue = struct { head: *Fiber, tail: *Fiber };139 const Queue = struct { head: *Fiber, tail: *Fiber };
105};140};
106141
...@@ -110,13 +145,18 @@ pub fn io(el: *EventLoop) Io {...@@ -110,13 +145,18 @@ pub fn io(el: *EventLoop) Io {
110 .vtable = &.{145 .vtable = &.{
111 .@"async" = @"async",146 .@"async" = @"async",
112 .@"await" = @"await",147 .@"await" = @"await",
148
113 .cancel = cancel,149 .cancel = cancel,
114 .cancelRequested = cancelRequested,150 .cancelRequested = cancelRequested,
151
115 .createFile = createFile,152 .createFile = createFile,
116 .openFile = openFile,153 .openFile = openFile,
117 .closeFile = closeFile,154 .closeFile = closeFile,
118 .read = read,155 .pread = pread,
119 .write = write,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,8 +173,7 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
133 .context = undefined,173 .context = undefined,
134 .awaiter = null,174 .awaiter = null,
135 .queue_next = null,175 .queue_next = null,
136 .can_cancel = false,176 .cancel_thread = null,
137 .canceled = false,
138 },177 },
139 .threads = .{178 .threads = .{
140 .allocated = @ptrCast(allocated_slice[0..threads_size]),179 .allocated = @ptrCast(allocated_slice[0..threads_size]),
...@@ -142,8 +181,8 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -142,8 +181,8 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
142 .active = 1,181 .active = 1,
143 },182 },
144 };183 };
145 Thread.index = 0;
146 const main_thread = &el.threads.allocated[0];184 const main_thread = &el.threads.allocated[0];
185 Thread.self = main_thread;
147 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));186 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
148 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};187 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
149 main_thread.* = .{188 main_thread.* = .{
...@@ -168,24 +207,22 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -168,24 +207,22 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
168pub fn deinit(el: *EventLoop) void {207pub fn deinit(el: *EventLoop) void {
169 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);208 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
170 for (el.threads.allocated[0..active_threads]) |*thread|209 for (el.threads.allocated[0..active_threads]) |*thread|
171 assert(@atomicLoad(?*Fiber, &thread.ready_queue, .unordered) == null); // pending async210 assert(@atomicLoad(?*Fiber, &thread.ready_queue, .acquire) == null); // pending async
172 el.yield(null, .exit);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 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.allocated.ptr));217 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.allocated.ptr));
174 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);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| {219 for (el.threads.allocated[1..active_threads]) |thread| thread.thread.join();
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 }
183 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);220 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
184 el.* = undefined;221 el.* = undefined;
185}222}
186223
187fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {224fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
188 const thread: *Thread = .current(el);225 const thread: *Thread = .current();
189 const ready_context: *Context = if (maybe_ready_fiber) |ready_fiber|226 const ready_context: *Context = if (maybe_ready_fiber) |ready_fiber|
190 &ready_fiber.context227 &ready_fiber.context
191 else if (thread.ready_queue) |ready_fiber| ready_context: {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,6 +235,7 @@ fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage
198 defer thread.steal_ready_search_index += 1;235 defer thread.steal_ready_search_index += 1;
199 if (thread.steal_ready_search_index == ready_threads) thread.steal_ready_search_index = 0;236 if (thread.steal_ready_search_index == ready_threads) thread.steal_ready_search_index = 0;
200 const steal_ready_search_thread = &el.threads.allocated[thread.steal_ready_search_index];237 const steal_ready_search_thread = &el.threads.allocated[thread.steal_ready_search_index];
238 if (steal_ready_search_thread == thread) continue;
201 const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;239 const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
202 if (@cmpxchgWeak(240 if (@cmpxchgWeak(
203 ?*Fiber,241 ?*Fiber,
...@@ -236,6 +274,7 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {...@@ -236,6 +274,7 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
236 defer thread.idle_search_index += 1;274 defer thread.idle_search_index += 1;
237 if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;275 if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;
238 const idle_search_thread = &el.threads.allocated[thread.idle_search_index];276 const idle_search_thread = &el.threads.allocated[thread.idle_search_index];
277 if (idle_search_thread == thread) continue;
239 if (@cmpxchgWeak(278 if (@cmpxchgWeak(
240 ?*Fiber,279 ?*Fiber,
241 &idle_search_thread.ready_queue,280 &idle_search_thread.ready_queue,
...@@ -249,11 +288,11 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {...@@ -249,11 +288,11 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
249 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,288 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
250 .ioprio = 0,289 .ioprio = 0,
251 .fd = idle_search_thread.io_uring.fd,290 .fd = idle_search_thread.io_uring.fd,
252 .off = @intFromEnum(Completion.Key.wakeup),291 .off = @intFromEnum(Completion.UserData.wakeup),
253 .addr = 0,292 .addr = 0,
254 .len = 0,293 .len = 0,
255 .rw_flags = 0,294 .rw_flags = 0,
256 .user_data = @intFromEnum(Completion.Key.wakeup),295 .user_data = @intFromEnum(Completion.UserData.wakeup),
257 .buf_index = 0,296 .buf_index = 0,
258 .personality = 0,297 .personality = 0,
259 .splice_fd_in = 0,298 .splice_fd_in = 0,
...@@ -314,15 +353,6 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {...@@ -314,15 +353,6 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
314 )) |old_head| ready_queue.tail.queue_next = old_head;353 )) |old_head| ready_queue.tail.queue_next = old_head;
315}354}
316355
317fn 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
326fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {356fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {
327 message.handle(el);357 message.handle(el);
328 const thread: *Thread = &el.threads.allocated[0];358 const thread: *Thread = &el.threads.allocated[0];
...@@ -332,17 +362,16 @@ fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAl...@@ -332,17 +362,16 @@ fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAl
332}362}
333363
334fn threadEntry(el: *EventLoop, index: u32) void {364fn threadEntry(el: *EventLoop, index: u32) void {
335 Thread.index = index;
336 const thread: *Thread = &el.threads.allocated[index];365 const thread: *Thread = &el.threads.allocated[index];
366 Thread.self = thread;
337 std.log.debug("created thread idle {*}", .{&thread.idle_context});367 std.log.debug("created thread idle {*}", .{&thread.idle_context});
338 el.idle(thread);368 el.idle(thread);
339}369}
340370
341const Completion = struct {371const Completion = struct {
342 const Key = enum(usize) {372 const UserData = enum(usize) {
343 unused,373 unused,
344 wakeup,374 wakeup,
345 cancel,
346 cleanup,375 cleanup,
347 exit,376 exit,
348 /// *Fiber377 /// *Fiber
...@@ -369,26 +398,43 @@ fn idle(el: *EventLoop, thread: *Thread) void {...@@ -369,26 +398,43 @@ fn idle(el: *EventLoop, thread: *Thread) void {
369 break :cqes_len 0;398 break :cqes_len 0;
370 },399 },
371 else => |e| @panic(@errorName(e)),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 .unused => unreachable, // bad submission queued?402 .unused => unreachable, // bad submission queued?
374 .wakeup => {},403 .wakeup => {},
375 .cancel => {},
376 .cleanup => @panic("failed to notify other threads that we are exiting"),404 .cleanup => @panic("failed to notify other threads that we are exiting"),
377 .exit => {405 .exit => {
378 assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async406 assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
379 return;407 return;
380 },408 },
381 _ => {409 _ => switch (errno(cqe.res)) {
382 const fiber: *Fiber = @ptrFromInt(cqe.user_data);410 .INTR => getSqe(&thread.io_uring).* = .{
383 assert(fiber.queue_next == null);411 .opcode = .ASYNC_CANCEL,
384 fiber.resultPointer(Completion).* = .{412 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
385 .result = cqe.res,413 .ioprio = 0,
386 .flags = cqe.flags,414 .fd = 0,
387 };415 .off = 0,
388 if (maybe_ready_fiber == null) maybe_ready_fiber = fiber else if (maybe_ready_queue) |*ready_queue| {416 .addr = cqe.user_data,
389 ready_queue.tail.queue_next = fiber;417 .len = 0,
390 ready_queue.tail = fiber;418 .rw_flags = 0,
391 } else maybe_ready_queue = .{ .head = fiber, .tail = fiber };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 if (maybe_ready_queue) |ready_queue| el.schedule(thread, ready_queue);440 if (maybe_ready_queue) |ready_queue| el.schedule(thread, ready_queue);
...@@ -409,7 +455,7 @@ const SwitchMessage = struct {...@@ -409,7 +455,7 @@ const SwitchMessage = struct {
409 };455 };
410456
411 fn handle(message: *const SwitchMessage, el: *EventLoop) void {457 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
412 const thread: *Thread = .current(el);458 const thread: *Thread = .current();
413 thread.current_context = message.contexts.ready;459 thread.current_context = message.contexts.ready;
414 switch (message.pending_task) {460 switch (message.pending_task) {
415 .nothing => {},461 .nothing => {},
...@@ -429,11 +475,11 @@ const SwitchMessage = struct {...@@ -429,11 +475,11 @@ const SwitchMessage = struct {
429 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,475 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
430 .ioprio = 0,476 .ioprio = 0,
431 .fd = each_thread.io_uring.fd,477 .fd = each_thread.io_uring.fd,
432 .off = @intFromEnum(Completion.Key.exit),478 .off = @intFromEnum(Completion.UserData.exit),
433 .addr = 0,479 .addr = 0,
434 .len = 0,480 .len = 0,
435 .rw_flags = 0,481 .rw_flags = 0,
436 .user_data = @intFromEnum(Completion.Key.cleanup),482 .user_data = @intFromEnum(Completion.UserData.cleanup),
437 .buf_index = 0,483 .buf_index = 0,
438 .personality = 0,484 .personality = 0,
439 .splice_fd_in = 0,485 .splice_fd_in = 0,
...@@ -544,6 +590,7 @@ fn @"async"(...@@ -544,6 +590,7 @@ fn @"async"(
544 start(context.ptr, result.ptr);590 start(context.ptr, result.ptr);
545 return null;591 return null;
546 };592 };
593 errdefer fiber.recycle();
547 std.log.debug("allocated {*}", .{fiber});594 std.log.debug("allocated {*}", .{fiber});
548595
549 const closure: *AsyncClosure = @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(596 const closure: *AsyncClosure = @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
...@@ -560,8 +607,7 @@ fn @"async"(...@@ -560,8 +607,7 @@ fn @"async"(
560 },607 },
561 .awaiter = null,608 .awaiter = null,
562 .queue_next = null,609 .queue_next = null,
563 .can_cancel = false,610 .cancel_thread = null,
564 .canceled = false,
565 };611 };
566 closure.* = .{612 closure.* = .{
567 .event_loop = event_loop,613 .event_loop = event_loop,
...@@ -571,7 +617,7 @@ fn @"async"(...@@ -571,7 +617,7 @@ fn @"async"(
571 };617 };
572 @memcpy(closure.contextPointer(), context);618 @memcpy(closure.contextPointer(), context);
573619
574 event_loop.schedule(.current(event_loop), .{ .head = fiber, .tail = fiber });620 event_loop.schedule(.current(), .{ .head = fiber, .tail = fiber });
575 return @ptrCast(fiber);621 return @ptrCast(fiber);
576}622}
577623
...@@ -585,7 +631,7 @@ fn @"await"(...@@ -585,7 +631,7 @@ fn @"await"(
585 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));631 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
586 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });632 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
587 @memcpy(result, future_fiber.resultBytes(result_alignment));633 @memcpy(result, future_fiber.resultBytes(result_alignment));
588 event_loop.recycle(future_fiber);634 future_fiber.recycle();
589}635}
590636
591fn cancel(637fn cancel(
...@@ -594,35 +640,37 @@ fn cancel(...@@ -594,35 +640,37 @@ fn cancel(
594 result: []u8,640 result: []u8,
595 result_alignment: Alignment,641 result_alignment: Alignment,
596) void {642) void {
597 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
598 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));643 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
599 @atomicStore(bool, &future_fiber.canceled, true, .release);644 if (@atomicRmw(
600 if (@atomicLoad(bool, &future_fiber.can_cancel, .acquire)) {645 ?*Thread,
601 const thread: *Thread = .current(event_loop);646 &future_fiber.cancel_thread,
602 getSqe(&thread.io_uring).* = .{647 .Xchg,
603 .opcode = .ASYNC_CANCEL,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 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,653 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
605 .ioprio = 0,654 .ioprio = 0,
606 .fd = 0,655 .fd = cancel_thread.io_uring.fd,
607 .off = 0,656 .off = @intFromPtr(future_fiber),
608 .addr = @intFromPtr(future_fiber),657 .addr = 0,
609 .len = 0,658 .len = @bitCast(-@as(i32, @intFromEnum(std.os.linux.E.INTR))),
610 .rw_flags = 0,659 .rw_flags = 0,
611 .user_data = @intFromEnum(Completion.Key.cancel),660 .user_data = @intFromEnum(Completion.UserData.cleanup),
612 .buf_index = 0,661 .buf_index = 0,
613 .personality = 0,662 .personality = 0,
614 .splice_fd_in = 0,663 .splice_fd_in = 0,
615 .addr3 = 0,664 .addr3 = 0,
616 .resv = 0,665 .resv = 0,
617 };666 };
618 }667 };
619 @"await"(userdata, any_future, result, result_alignment);668 @"await"(userdata, any_future, result, result_alignment);
620}669}
621670
622fn cancelRequested(userdata: ?*anyopaque) bool {671fn cancelRequested(userdata: ?*anyopaque) bool {
623 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));672 _ = userdata;
624 const thread: *Thread = .current(event_loop);673 return @atomicLoad(?*Thread, &Thread.current().currentFiber().cancel_thread, .acquire) == Thread.canceling;
625 return thread.currentFiber().canceled;
626}674}
627675
628pub fn createFile(676pub fn createFile(
...@@ -632,6 +680,10 @@ pub fn createFile(...@@ -632,6 +680,10 @@ pub fn createFile(
632 flags: Io.CreateFlags,680 flags: Io.CreateFlags,
633) Io.FileOpenError!std.fs.File {681) Io.FileOpenError!std.fs.File {
634 const el: *EventLoop = @alignCast(@ptrCast(userdata));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);
635687
636 const posix = std.posix;688 const posix = std.posix;
637 const sub_path_c = try posix.toPosixPath(sub_path);689 const sub_path_c = try posix.toPosixPath(sub_path);
...@@ -670,23 +722,30 @@ pub fn createFile(...@@ -670,23 +722,30 @@ pub fn createFile(
670 @panic("TODO");722 @panic("TODO");
671 }723 }
672724
673 const thread: *Thread = .current(el);725 getSqe(iou).* = .{
674 const iou = &thread.io_uring;726 .opcode = .OPENAT,
675 const fiber = thread.currentFiber();727 .flags = 0,
676 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;728 .ioprio = 0,
677729 .fd = dir.fd,
678 const sqe = getSqe(iou);730 .off = 0,
679 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, flags.mode);731 .addr = @intFromPtr(&sub_path_c),
680 sqe.user_data = @intFromPtr(fiber);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 };
681741
682 @atomicStore(bool, &fiber.can_cancel, true, .release);
683 el.yield(null, .nothing);742 el.yield(null, .nothing);
684 @atomicStore(bool, &fiber.can_cancel, false, .release);743 fiber.exitCancelRegion(thread);
685744
686 const completion = fiber.resultPointer(Completion);745 const completion = fiber.resultPointer(Completion);
687 switch (errno(completion.result)) {746 switch (errno(completion.result)) {
688 .SUCCESS => return .{ .handle = completion.result },747 .SUCCESS => return .{ .handle = completion.result },
689 .INTR => @panic("TODO is this reachable?"),748 .INTR => unreachable,
690 .CANCELED => return error.AsyncCancel,749 .CANCELED => return error.AsyncCancel,
691750
692 .FAULT => unreachable,751 .FAULT => unreachable,
...@@ -723,10 +782,10 @@ pub fn openFile(...@@ -723,10 +782,10 @@ pub fn openFile(
723 flags: Io.OpenFlags,782 flags: Io.OpenFlags,
724) Io.FileOpenError!std.fs.File {783) Io.FileOpenError!std.fs.File {
725 const el: *EventLoop = @alignCast(@ptrCast(userdata));784 const el: *EventLoop = @alignCast(@ptrCast(userdata));
726 const thread: *Thread = .current(el);785 const thread: *Thread = .current();
727 const iou = &thread.io_uring;786 const iou = &thread.io_uring;
728 const fiber = thread.currentFiber();787 const fiber = thread.currentFiber();
729 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;788 try fiber.enterCancelRegion(thread);
730789
731 const posix = std.posix;790 const posix = std.posix;
732 const sub_path_c = try posix.toPosixPath(sub_path);791 const sub_path_c = try posix.toPosixPath(sub_path);
...@@ -771,18 +830,30 @@ pub fn openFile(...@@ -771,18 +830,30 @@ pub fn openFile(
771 @panic("TODO");830 @panic("TODO");
772 }831 }
773832
774 const sqe = getSqe(iou);833 getSqe(iou).* = .{
775 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);834 .opcode = .OPENAT,
776 sqe.user_data = @intFromPtr(fiber);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 };
777849
778 @atomicStore(bool, &fiber.can_cancel, true, .release);
779 el.yield(null, .nothing);850 el.yield(null, .nothing);
780 @atomicStore(bool, &fiber.can_cancel, false, .release);851 fiber.exitCancelRegion(thread);
781852
782 const completion = fiber.resultPointer(Completion);853 const completion = fiber.resultPointer(Completion);
783 switch (errno(completion.result)) {854 switch (errno(completion.result)) {
784 .SUCCESS => return .{ .handle = completion.result },855 .SUCCESS => return .{ .handle = completion.result },
785 .INTR => @panic("TODO is this reachable?"),856 .INTR => unreachable,
786 .CANCELED => return error.AsyncCancel,857 .CANCELED => return error.AsyncCancel,
787858
788 .FAULT => unreachable,859 .FAULT => unreachable,
...@@ -814,20 +885,33 @@ pub fn openFile(...@@ -814,20 +885,33 @@ pub fn openFile(
814885
815pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {886pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
816 const el: *EventLoop = @alignCast(@ptrCast(userdata));887 const el: *EventLoop = @alignCast(@ptrCast(userdata));
817 const thread: *Thread = .current(el);888 const thread: *Thread = .current();
818 const iou = &thread.io_uring;889 const iou = &thread.io_uring;
819 const fiber = thread.currentFiber();890 const fiber = thread.currentFiber();
820891
821 const sqe = getSqe(iou);892 getSqe(iou).* = .{
822 sqe.prep_close(file.handle);893 .opcode = .CLOSE,
823 sqe.user_data = @intFromPtr(fiber);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 };
824908
825 el.yield(null, .nothing);909 el.yield(null, .nothing);
826910
827 const completion = fiber.resultPointer(Completion);911 const completion = fiber.resultPointer(Completion);
828 switch (errno(completion.result)) {912 switch (errno(completion.result)) {
829 .SUCCESS => return,913 .SUCCESS => return,
830 .INTR => @panic("TODO is this reachable?"),914 .INTR => unreachable,
831 .CANCELED => return,915 .CANCELED => return,
832916
833 .BADF => unreachable, // Always a race condition.917 .BADF => unreachable, // Always a race condition.
...@@ -835,25 +919,37 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {...@@ -835,25 +919,37 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
835 }919 }
836}920}
837921
838pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadError!usize {922pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {
839 const el: *EventLoop = @alignCast(@ptrCast(userdata));923 const el: *EventLoop = @alignCast(@ptrCast(userdata));
840 const thread: *Thread = .current(el);924 const thread: *Thread = .current();
841 const iou = &thread.io_uring;925 const iou = &thread.io_uring;
842 const fiber = thread.currentFiber();926 const fiber = thread.currentFiber();
843 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;927 try fiber.enterCancelRegion(thread);
844928
845 const sqe = getSqe(iou);929 getSqe(iou).* = .{
846 sqe.prep_read(file.handle, buffer, std.math.maxInt(u64));930 .opcode = .READ,
847 sqe.user_data = @intFromPtr(fiber);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 };
848945
849 @atomicStore(bool, &fiber.can_cancel, true, .release);
850 el.yield(null, .nothing);946 el.yield(null, .nothing);
851 @atomicStore(bool, &fiber.can_cancel, false, .release);947 fiber.exitCancelRegion(thread);
852948
853 const completion = fiber.resultPointer(Completion);949 const completion = fiber.resultPointer(Completion);
854 switch (errno(completion.result)) {950 switch (errno(completion.result)) {
855 .SUCCESS => return @as(u32, @bitCast(completion.result)),951 .SUCCESS => return @as(u32, @bitCast(completion.result)),
856 .INTR => @panic("TODO is this reachable?"),952 .INTR => unreachable,
857 .CANCELED => return error.AsyncCancel,953 .CANCELED => return error.AsyncCancel,
858954
859 .INVAL => unreachable,955 .INVAL => unreachable,
...@@ -868,30 +964,44 @@ pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadE...@@ -868,30 +964,44 @@ pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadE
868 .NOTCONN => return error.SocketNotConnected,964 .NOTCONN => return error.SocketNotConnected,
869 .CONNRESET => return error.ConnectionResetByPeer,965 .CONNRESET => return error.ConnectionResetByPeer,
870 .TIMEDOUT => return error.ConnectionTimedOut,966 .TIMEDOUT => return error.ConnectionTimedOut,
967 .NXIO => return error.Unseekable,
968 .SPIPE => return error.Unseekable,
969 .OVERFLOW => return error.Unseekable,
871 else => |err| return std.posix.unexpectedErrno(err),970 else => |err| return std.posix.unexpectedErrno(err),
872 }971 }
873}972}
874973
875pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.FileWriteError!usize {974pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {
876 const el: *EventLoop = @alignCast(@ptrCast(userdata));975 const el: *EventLoop = @alignCast(@ptrCast(userdata));
877976 const thread: *Thread = .current();
878 const thread: *Thread = .current(el);
879 const iou = &thread.io_uring;977 const iou = &thread.io_uring;
880 const fiber = thread.currentFiber();978 const fiber = thread.currentFiber();
881 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;979 try fiber.enterCancelRegion(thread);
882980
883 const sqe = getSqe(iou);981 getSqe(iou).* = .{
884 sqe.prep_write(file.handle, buffer, std.math.maxInt(u64));982 .opcode = .WRITE,
885 sqe.user_data = @intFromPtr(fiber);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 };
886997
887 @atomicStore(bool, &fiber.can_cancel, true, .release);
888 el.yield(null, .nothing);998 el.yield(null, .nothing);
889 @atomicStore(bool, &fiber.can_cancel, false, .release);999 fiber.exitCancelRegion(thread);
8901000
891 const completion = fiber.resultPointer(Completion);1001 const completion = fiber.resultPointer(Completion);
892 switch (errno(completion.result)) {1002 switch (errno(completion.result)) {
893 .SUCCESS => return @as(u32, @bitCast(completion.result)),1003 .SUCCESS => return @as(u32, @bitCast(completion.result)),
894 .INTR => @panic("TODO is this reachable?"),1004 .INTR => unreachable,
895 .CANCELED => return error.AsyncCancel,1005 .CANCELED => return error.AsyncCancel,
8961006
897 .INVAL => return error.InvalidArgument,1007 .INVAL => return error.InvalidArgument,
...@@ -907,17 +1017,77 @@ pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.Fi...@@ -907,17 +1017,77 @@ pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.Fi
907 .ACCES => return error.AccessDenied,1017 .ACCES => return error.AccessDenied,
908 .PERM => return error.PermissionDenied,1018 .PERM => return error.PermissionDenied,
909 .PIPE => return error.BrokenPipe,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 .BUSY => return error.DeviceBusy,1023 .BUSY => return error.DeviceBusy,
912 .NXIO => return error.NoDevice,1024 .CONNRESET => return error.ConnectionResetByPeer,
913 .MSGSIZE => return error.MessageTooBig,1025 .MSGSIZE => return error.MessageTooBig,
914 else => |err| return std.posix.unexpectedErrno(err),1026 else => |err| return std.posix.unexpectedErrno(err),
915 }1027 }
916}1028}
9171029
918fn errno(signed: i32) std.posix.E {1030pub fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
919 const int = if (signed > -4096 and signed < 0) -signed else 0;1031 _ = userdata;
920 return @enumFromInt(int);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
1036pub 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
1089fn errno(signed: i32) std.os.linux.E {
1090 return .init(@bitCast(@as(isize, signed)));
921}1091}
9221092
923fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {1093fn 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,13 +332,18 @@ pub fn io(pool: *Pool) Io {
332 .vtable = &.{332 .vtable = &.{
333 .@"async" = @"async",333 .@"async" = @"async",
334 .@"await" = @"await",334 .@"await" = @"await",
335
335 .cancel = cancel,336 .cancel = cancel,
336 .cancelRequested = cancelRequested,337 .cancelRequested = cancelRequested,
338
337 .createFile = createFile,339 .createFile = createFile,
338 .openFile = openFile,340 .openFile = openFile,
339 .closeFile = closeFile,341 .closeFile = closeFile,
340 .read = read,342 .pread = pread,
341 .write = write,343 .pwrite = pwrite,
344
345 .now = now,
346 .sleep = sleep,
342 },347 },
343 };348 };
344}349}
...@@ -347,15 +352,44 @@ const AsyncClosure = struct {...@@ -347,15 +352,44 @@ const AsyncClosure = struct {
347 func: *const fn (context: *anyopaque, result: *anyopaque) void,352 func: *const fn (context: *anyopaque, result: *anyopaque) void,
348 runnable: Runnable = .{ .runFn = runFn },353 runnable: Runnable = .{ .runFn = runFn },
349 reset_event: std.Thread.ResetEvent,354 reset_event: std.Thread.ResetEvent,
350 cancel_flag: bool,355 cancel_tid: std.Thread.Id,
351 context_offset: usize,356 context_offset: usize,
352 result_offset: usize,357 result_offset: usize,
353358
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 fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void {368 fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void {
355 const closure: *AsyncClosure = @alignCast(@fieldParentPtr("runnable", runnable));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 current_closure = closure;382 current_closure = closure;
357 closure.func(closure.contextPointer(), closure.resultPointer());383 closure.func(closure.contextPointer(), closure.resultPointer());
358 current_closure = null;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 closure.reset_event.set();393 closure.reset_event.set();
360 }394 }
361395
...@@ -414,7 +448,7 @@ fn @"async"(...@@ -414,7 +448,7 @@ fn @"async"(
414 .context_offset = context_offset,448 .context_offset = context_offset,
415 .result_offset = result_offset,449 .result_offset = result_offset,
416 .reset_event = .{},450 .reset_event = .{},
417 .cancel_flag = false,451 .cancel_tid = 0,
418 };452 };
419 @memcpy(closure.contextPointer()[0..context.len], context);453 @memcpy(closure.contextPointer()[0..context.len], context);
420 pool.run_queue.prepend(&closure.runnable.node);454 pool.run_queue.prepend(&closure.runnable.node);
...@@ -456,7 +490,23 @@ fn cancel(...@@ -456,7 +490,23 @@ fn cancel(
456 _ = result_alignment;490 _ = result_alignment;
457 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));491 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
458 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));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 closure.waitAndFree(pool.allocator, result);510 closure.waitAndFree(pool.allocator, result);
461}511}
462512
...@@ -464,7 +514,7 @@ fn cancelRequested(userdata: ?*anyopaque) bool {...@@ -464,7 +514,7 @@ fn cancelRequested(userdata: ?*anyopaque) bool {
464 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));514 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
465 _ = pool;515 _ = pool;
466 const closure = current_closure orelse return false;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}
469519
470fn checkCancel(pool: *Pool) error{AsyncCancel}!void {520fn checkCancel(pool: *Pool) error{AsyncCancel}!void {
...@@ -499,14 +549,52 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {...@@ -499,14 +549,52 @@ pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
499 return file.close();549 return file.close();
500}550}
501551
502pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadError!usize {552pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {
503 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));553 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
504 try pool.checkCancel();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}
507560
508pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.FileWriteError!usize {561pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {
509 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));562 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
510 try pool.checkCancel();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
570pub 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
577pub 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
...@@ -631,8 +631,8 @@ inline fn callMainWithArgs(argc: usize, argv: [*][*:0]u8, envp: [][*:0]u8) u8 {...@@ -631,8 +631,8 @@ inline fn callMainWithArgs(argc: usize, argv: [*][*:0]u8, envp: [][*:0]u8) u8 {
631 std.os.argv = argv[0..argc];631 std.os.argv = argv[0..argc];
632 std.os.environ = envp;632 std.os.environ = envp;
633633
634 maybeIgnoreSignals();
634 std.debug.maybeEnableSegfaultHandler();635 std.debug.maybeEnableSegfaultHandler();
635 maybeIgnoreSigpipe();
636636
637 return callMain();637 return callMain();
638}638}
...@@ -734,8 +734,8 @@ pub fn call_wWinMain() std.os.windows.INT {...@@ -734,8 +734,8 @@ pub fn call_wWinMain() std.os.windows.INT {
734 return root.wWinMain(hInstance, null, lpCmdLine, nCmdShow);734 return root.wWinMain(hInstance, null, lpCmdLine, nCmdShow);
735}735}
736736
737fn maybeIgnoreSigpipe() void {737fn maybeIgnoreSignals() void {
738 const have_sigpipe_support = switch (builtin.os.tag) {738 switch (builtin.os.tag) {
739 .linux,739 .linux,
740 .plan9,740 .plan9,
741 .solaris,741 .solaris,
...@@ -749,22 +749,20 @@ fn maybeIgnoreSigpipe() void {...@@ -749,22 +749,20 @@ fn maybeIgnoreSigpipe() void {
749 .visionos,749 .visionos,
750 .dragonfly,750 .dragonfly,
751 .freebsd,751 .freebsd,
752 => true,752 => {},
753753 else => return,
754 else => false,
755 };
756
757 if (have_sigpipe_support and !std.options.keep_sigpipe) {
758 const posix = std.posix;
759 const act: posix.Sigaction = .{
760 // Set handler to a noop function instead of `SIG.IGN` to prevent
761 // leaking signal disposition to a child process.
762 .handler = .{ .handler = noopSigHandler },
763 .mask = posix.sigemptyset(),
764 .flags = 0,
765 };
766 posix.sigaction(posix.SIG.PIPE, &act, null);
767 }754 }
755 const posix = std.posix;
756 const act: posix.Sigaction = .{
757 // Set handler to a noop function instead of `SIG.IGN` to prevent
758 // leaking signal disposition to a child process.
759 .handler = .{ .handler = noopSigHandler },
760 .mask = posix.sigemptyset(),
761 .flags = 0,
762 };
763 if (!std.options.keep_sigpoll) posix.sigaction(posix.SIG.POLL, &act, null);
764 if (@hasField(posix.SIG, "IO") and posix.SIG.IO != posix.SIG.POLL and !std.options.keep_sigio) posix.sigaction(posix.SIG.IO, &act, null);
765 if (!std.options.keep_sigpipe) posix.sigaction(posix.SIG.PIPE, &act, null);
768}766}
769767
770fn noopSigHandler(_: i32) callconv(.c) void {}768fn noopSigHandler(_: i32) callconv(.c) void {}
lib/std/std.zig+3
...@@ -137,6 +137,9 @@ pub const Options = struct {...@@ -137,6 +137,9 @@ pub const Options = struct {
137137
138 crypto_fork_safety: bool = true,138 crypto_fork_safety: bool = true,
139139
140 keep_sigpoll: bool = false,
141 keep_sigio: bool = false,
142
140 /// By default Zig disables SIGPIPE by setting a "no-op" handler for it. Set this option143 /// By default Zig disables SIGPIPE by setting a "no-op" handler for it. Set this option
141 /// to `true` to prevent that.144 /// to `true` to prevent that.
142 ///145 ///