authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-03-30 01:54:02-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
logb37126bc086059b4f7a0fc29dfa2ce30b2f3458c
tree97b077399aae746997a89f3129266780bb917cd7
parentc278830592792ed724f68b4abbad9d253bed5404

EventLoop: implement thread-local queues and cancellation

3 files changed, 427 insertions(+), 237 deletions(-)

lib/std/Io.zig+6-4
...@@ -591,6 +591,7 @@ pub const VTable = struct {...@@ -591,6 +591,7 @@ pub const VTable = struct {
591 /// Points to a buffer where the result is written.591 /// Points to a buffer where the result is written.
592 /// The length is equal to size in bytes of result type.592 /// The length is equal to size in bytes of result type.
593 result: []u8,593 result: []u8,
594 result_alignment: std.mem.Alignment,
594 ) void,595 ) void,
595596
596 /// Equivalent to `await` but initiates cancel request.597 /// Equivalent to `await` but initiates cancel request.
...@@ -606,6 +607,7 @@ pub const VTable = struct {...@@ -606,6 +607,7 @@ pub const VTable = struct {
606 /// Points to a buffer where the result is written.607 /// Points to a buffer where the result is written.
607 /// The length is equal to size in bytes of result type.608 /// The length is equal to size in bytes of result type.
608 result: []u8,609 result: []u8,
610 result_alignment: std.mem.Alignment,
609 ) void,611 ) void,
610612
611 /// Returns whether the current thread of execution is known to have613 /// Returns whether the current thread of execution is known to have
...@@ -641,14 +643,14 @@ pub fn Future(Result: type) type {...@@ -641,14 +643,14 @@ pub fn Future(Result: type) type {
641 /// Idempotent.643 /// Idempotent.
642 pub fn cancel(f: *@This(), io: Io) Result {644 pub fn cancel(f: *@This(), io: Io) Result {
643 const any_future = f.any_future orelse return f.result;645 const any_future = f.any_future orelse return f.result;
644 io.vtable.cancel(io.userdata, any_future, @ptrCast((&f.result)[0..1]));646 io.vtable.cancel(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));
645 f.any_future = null;647 f.any_future = null;
646 return f.result;648 return f.result;
647 }649 }
648650
649 pub fn await(f: *@This(), io: Io) Result {651 pub fn await(f: *@This(), io: Io) Result {
650 const any_future = f.any_future orelse return f.result;652 const any_future = f.any_future orelse return f.result;
651 io.vtable.await(io.userdata, any_future, @ptrCast((&f.result)[0..1]));653 io.vtable.await(io.userdata, any_future, @ptrCast((&f.result)[0..1]), .of(Result));
652 f.any_future = null;654 f.any_future = null;
653 return f.result;655 return f.result;
654 }656 }
...@@ -671,9 +673,9 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(...@@ -671,9 +673,9 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(
671 future.any_future = io.vtable.async(673 future.any_future = io.vtable.async(
672 io.userdata,674 io.userdata,
673 @ptrCast((&future.result)[0..1]),675 @ptrCast((&future.result)[0..1]),
674 .fromByteUnits(@alignOf(Result)),676 .of(Result),
675 if (@sizeOf(Args) == 0) &.{} else @ptrCast((&args)[0..1]), // work around compiler bug677 if (@sizeOf(Args) == 0) &.{} else @ptrCast((&args)[0..1]), // work around compiler bug
676 .fromByteUnits(@alignOf(Args)),678 .of(Args),
677 TypeErased.start,679 TypeErased.start,
678 );680 );
679 return future;681 return future;
lib/std/Io/EventLoop.zig+407-231
...@@ -10,38 +10,50 @@ const IoUring = std.os.linux.IoUring;...@@ -10,38 +10,50 @@ const IoUring = std.os.linux.IoUring;
10/// Must be a thread-safe allocator.10/// Must be a thread-safe allocator.
11gpa: Allocator,11gpa: Allocator,
12mutex: std.Thread.Mutex,12mutex: std.Thread.Mutex,
13queue: std.DoublyLinkedList,
14/// Atomic copy of queue.len
15queue_len: u32,
16free: std.DoublyLinkedList,
17main_fiber: Fiber,13main_fiber: Fiber,
18idle_count: usize,14threads: Thread.List,
19threads: std.ArrayListUnmanaged(Thread),
20exiting: bool,
21
22threadlocal var thread_index: u32 = undefined;
2315
24/// Empirically saw >128KB being used by the self-hosted backend to panic.16/// Empirically saw >128KB being used by the self-hosted backend to panic.
25const idle_stack_size = 256 * 1024;17const idle_stack_size = 256 * 1024;
2618
19const max_idle_search = 4;
20const max_steal_ready_search = 4;
21
27const io_uring_entries = 64;22const io_uring_entries = 64;
2823
29const Thread = struct {24const Thread = struct {
30 thread: std.Thread,25 thread: std.Thread,
31 idle_context: Context,26 idle_context: Context,
32 current_context: *Context,27 current_context: *Context,
28 ready_queue: ?*Fiber,
29 free_queue: ?*Fiber,
33 io_uring: IoUring,30 io_uring: IoUring,
31 idle_search_index: u32,
32 steal_ready_search_index: u32,
33
34 threadlocal var index: u32 = undefined;
35
36 fn current(el: *EventLoop) *Thread {
37 return &el.threads.allocated[index];
38 }
3439
35 fn currentFiber(thread: *Thread) *Fiber {40 fn currentFiber(thread: *Thread) *Fiber {
36 return @fieldParentPtr("context", thread.current_context);41 return @fieldParentPtr("context", thread.current_context);
37 }42 }
43
44 const List = struct {
45 allocated: []Thread,
46 reserved: u32,
47 active: u32,
48 };
38};49};
3950
40const Fiber = struct {51const Fiber = struct {
41 context: Context,52 context: Context,
42 awaiter: ?*Fiber,53 awaiter: ?*Fiber,
43 queue_node: std.DoublyLinkedList.Node,54 queue_next: ?*Fiber,
44 result_align: Alignment,55 can_cancel: bool,
56 canceled: bool,
4557
46 const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));58 const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));
4759
...@@ -63,14 +75,13 @@ const Fiber = struct {...@@ -63,14 +75,13 @@ const Fiber = struct {
63 );75 );
6476
65 fn allocate(el: *EventLoop) error{OutOfMemory}!*Fiber {77 fn allocate(el: *EventLoop) error{OutOfMemory}!*Fiber {
66 return if (free_node: {78 const thread: *Thread = .current(el);
67 el.mutex.lock();79 if (thread.free_queue) |free_fiber| {
68 defer el.mutex.unlock();80 thread.free_queue = free_fiber.queue_next;
69 break :free_node el.free.pop();81 free_fiber.queue_next = null;
70 }) |free_node|82 return free_fiber;
71 @alignCast(@fieldParentPtr("queue_node", free_node))83 }
72 else84 return @ptrCast(try el.gpa.alignedAlloc(u8, @alignOf(Fiber), allocation_size));
73 @ptrCast(try el.gpa.alignedAlloc(u8, @alignOf(Fiber), allocation_size));
74 }85 }
7586
76 fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {87 fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {
...@@ -82,9 +93,15 @@ const Fiber = struct {...@@ -82,9 +93,15 @@ const Fiber = struct {
82 return allocated_slice[allocated_slice.len..].ptr;93 return allocated_slice[allocated_slice.len..].ptr;
83 }94 }
8495
85 fn resultPointer(f: *Fiber) [*]u8 {96 fn resultPointer(f: *Fiber, comptime Result: type) *Result {
86 return @ptrFromInt(f.result_align.forward(@intFromPtr(f) + @sizeOf(Fiber)));97 return @alignCast(@ptrCast(f.resultBytes(.of(Result))));
98 }
99
100 fn resultBytes(f: *Fiber, alignment: Alignment) [*]u8 {
101 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
87 }102 }
103
104 const Queue = struct { head: *Fiber, tail: *Fiber };
88};105};
89106
90pub fn io(el: *EventLoop) Io {107pub fn io(el: *EventLoop) Io {
...@@ -93,6 +110,8 @@ pub fn io(el: *EventLoop) Io {...@@ -93,6 +110,8 @@ pub fn io(el: *EventLoop) Io {
93 .vtable = &.{110 .vtable = &.{
94 .@"async" = @"async",111 .@"async" = @"async",
95 .@"await" = @"await",112 .@"await" = @"await",
113 .cancel = cancel,
114 .cancelRequested = cancelRequested,
96 .createFile = createFile,115 .createFile = createFile,
97 .openFile = openFile,116 .openFile = openFile,
98 .closeFile = closeFile,117 .closeFile = closeFile,
...@@ -110,58 +129,86 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {...@@ -110,58 +129,86 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
110 el.* = .{129 el.* = .{
111 .gpa = gpa,130 .gpa = gpa,
112 .mutex = .{},131 .mutex = .{},
113 .queue = .{},132 .main_fiber = .{
114 .queue_len = 0,133 .context = undefined,
115 .free = .{},134 .awaiter = null,
116 .main_fiber = undefined,135 .queue_next = null,
117 .idle_count = 0,136 .can_cancel = false,
118 .threads = .initBuffer(@ptrCast(allocated_slice[0..threads_size])),137 .canceled = false,
119 .exiting = false,138 },
139 .threads = .{
140 .allocated = @ptrCast(allocated_slice[0..threads_size]),
141 .reserved = 1,
142 .active = 1,
143 },
120 };144 };
121 thread_index = 0;145 Thread.index = 0;
122 const main_thread = el.threads.addOneAssumeCapacity();146 const main_thread = &el.threads.allocated[0];
123 main_thread.io_uring = try IoUring.init(io_uring_entries, 0);
124 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));147 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
125 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};148 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
126 main_thread.idle_context = .{149 main_thread.* = .{
127 .rsp = @intFromPtr(idle_stack_end - 1),150 .thread = undefined,
128 .rbp = 0,151 .idle_context = .{
129 .rip = @intFromPtr(&mainIdleEntry),152 .rsp = @intFromPtr(idle_stack_end - 1),
153 .rbp = 0,
154 .rip = @intFromPtr(&mainIdleEntry),
155 },
156 .current_context = &el.main_fiber.context,
157 .ready_queue = null,
158 .free_queue = null,
159 .io_uring = try IoUring.init(io_uring_entries, 0),
160 .idle_search_index = 1,
161 .steal_ready_search_index = 1,
130 };162 };
163 errdefer main_thread.io_uring.deinit();
131 std.log.debug("created main idle {*}", .{&main_thread.idle_context});164 std.log.debug("created main idle {*}", .{&main_thread.idle_context});
132 std.log.debug("created main {*}", .{&el.main_fiber});165 std.log.debug("created main {*}", .{&el.main_fiber});
133 main_thread.current_context = &el.main_fiber.context;
134}166}
135167
136pub fn deinit(el: *EventLoop) void {168pub fn deinit(el: *EventLoop) void {
137 assert(el.queue.len == 0); // pending async169 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
170 for (el.threads.allocated[0..active_threads]) |*thread|
171 assert(@atomicLoad(?*Fiber, &thread.ready_queue, .unordered) == null); // pending async
138 el.yield(null, .exit);172 el.yield(null, .exit);
139 while (el.free.pop()) |free_node| {173 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.allocated.ptr));
140 const free_fiber: *Fiber = @alignCast(@fieldParentPtr("queue_node", free_node));174 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
141 el.gpa.free(free_fiber.allocatedSlice());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 }
142 }182 }
143 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.capacity * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
144 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
145 for (el.threads.items[1..]) |*thread| thread.thread.join();
146 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);183 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
147 el.* = undefined;184 el.* = undefined;
148}185}
149186
150fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {187fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
151 const thread: *Thread = &el.threads.items[thread_index];188 const thread: *Thread = .current(el);
152 const ready_context: *Context = ready_context: {189 const ready_context: *Context = if (maybe_ready_fiber) |ready_fiber|
153 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {190 &ready_fiber.context
154 el.mutex.lock();191 else if (thread.ready_queue) |ready_fiber| ready_context: {
155 defer el.mutex.unlock();192 thread.ready_queue = ready_fiber.queue_next;
156 const expected_queue_len = std.math.lossyCast(u32, el.queue.len);193 ready_fiber.queue_next = null;
157 const ready_node = el.queue.pop();
158 _ = @cmpxchgStrong(u32, &el.queue_len, expected_queue_len, std.math.lossyCast(u32, el.queue.len), .monotonic, .monotonic);
159 break :ready_node ready_node;
160 }) |ready_node|
161 @alignCast(@fieldParentPtr("queue_node", ready_node))
162 else
163 break :ready_context &thread.idle_context;
164 break :ready_context &ready_fiber.context;194 break :ready_context &ready_fiber.context;
195 } else ready_context: {
196 const ready_threads = @atomicLoad(u32, &el.threads.active, .acquire);
197 break :ready_context for (0..max_steal_ready_search) |_| {
198 defer thread.steal_ready_search_index += 1;
199 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];
201 const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
202 if (@cmpxchgWeak(
203 ?*Fiber,
204 &steal_ready_search_thread.ready_queue,
205 ready_fiber,
206 @atomicLoad(?*Fiber, &ready_fiber.queue_next, .acquire),
207 .acq_rel,
208 .monotonic,
209 )) |_| continue;
210 break &ready_fiber.context;
211 } else &thread.idle_context;
165 };212 };
166 const message: SwitchMessage = .{213 const message: SwitchMessage = .{
167 .contexts = .{214 .contexts = .{
...@@ -174,111 +221,177 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.Pe...@@ -174,111 +221,177 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.Pe
174 contextSwitch(&message).handle(el);221 contextSwitch(&message).handle(el);
175}222}
176223
177fn schedule(el: *EventLoop, fiber: *Fiber) void {224fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
178 std.log.debug("scheduling {*}", .{fiber});225 {
179 if (idle_count: {226 var fiber = ready_queue.head;
180 el.mutex.lock();227 while (true) {
181 defer el.mutex.unlock();228 std.log.debug("scheduling {*}", .{fiber});
182 const expected_queue_len = std.math.lossyCast(u32, el.queue.len);229 fiber = fiber.queue_next orelse break;
183 el.queue.append(&fiber.queue_node);230 }
184 _ = @cmpxchgStrong(u32, &el.queue_len, expected_queue_len, std.math.lossyCast(u32, el.queue.len), .monotonic, .monotonic);231 assert(fiber == ready_queue.tail);
185 break :idle_count el.idle_count;232 }
186 } > 0) {233 // shared fields of previous `Thread` must be initialized before later ones are marked as active
187 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(u32), 1, std.os.linux.FUTEX2.SIZE_U32 | std.os.linux.FUTEX2.PRIVATE); // TODO: io_uring234 const new_thread_index = @atomicLoad(u32, &el.threads.active, .acquire);
235 for (0..max_idle_search) |_| {
236 defer thread.idle_search_index += 1;
237 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];
239 if (@cmpxchgWeak(
240 ?*Fiber,
241 &idle_search_thread.ready_queue,
242 null,
243 ready_queue.head,
244 .acq_rel,
245 .monotonic,
246 )) |_| continue;
247 getSqe(&thread.io_uring).* = .{
248 .opcode = .MSG_RING,
249 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
250 .ioprio = 0,
251 .fd = idle_search_thread.io_uring.fd,
252 .off = @intFromEnum(Completion.Key.wakeup),
253 .addr = 0,
254 .len = 0,
255 .rw_flags = 0,
256 .user_data = @intFromEnum(Completion.Key.wakeup),
257 .buf_index = 0,
258 .personality = 0,
259 .splice_fd_in = 0,
260 .addr3 = 0,
261 .resv = 0,
262 };
188 return;263 return;
189 }264 }
190 if (el.threads.items.len == el.threads.capacity) return;265 spawn_thread: {
191 const thread = el.threads.addOneAssumeCapacity();266 // previous failed reservations must have completed before retrying
192 thread.thread = std.Thread.spawn(.{267 if (new_thread_index == el.threads.allocated.len or @cmpxchgWeak(
193 .stack_size = idle_stack_size,268 u32,
194 .allocator = el.gpa,269 &el.threads.reserved,
195 }, threadEntry, .{ el, el.threads.items.len - 1 }) catch {270 new_thread_index,
196 el.threads.items.len -= 1;271 new_thread_index + 1,
272 .acquire,
273 .monotonic,
274 ) != null) break :spawn_thread;
275 const new_thread = &el.threads.allocated[new_thread_index];
276 const next_thread_index = new_thread_index + 1;
277 new_thread.* = .{
278 .thread = undefined,
279 .idle_context = undefined,
280 .current_context = &new_thread.idle_context,
281 .ready_queue = ready_queue.head,
282 .free_queue = null,
283 .io_uring = IoUring.init(io_uring_entries, 0) catch |err| {
284 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
285 // no more access to `thread` after giving up reservation
286 std.log.warn("unable to create worker thread due to io_uring init failure: {s}", .{@errorName(err)});
287 break :spawn_thread;
288 },
289 .idle_search_index = next_thread_index,
290 .steal_ready_search_index = next_thread_index,
291 };
292 new_thread.thread = std.Thread.spawn(.{
293 .stack_size = idle_stack_size,
294 .allocator = el.gpa,
295 }, threadEntry, .{ el, new_thread_index }) catch |err| {
296 new_thread.io_uring.deinit();
297 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
298 // no more access to `thread` after giving up reservation
299 std.log.warn("unable to create worker thread due spawn failure: {s}", .{@errorName(err)});
300 break :spawn_thread;
301 };
302 // shared fields of `Thread` must be initialized before being marked active
303 @atomicStore(u32, &el.threads.active, next_thread_index, .release);
197 return;304 return;
198 };305 }
306 // nobody wanted it, so just queue it on ourselves
307 while (@cmpxchgWeak(
308 ?*Fiber,
309 &thread.ready_queue,
310 ready_queue.tail.queue_next,
311 ready_queue.head,
312 .acq_rel,
313 .acquire,
314 )) |old_head| ready_queue.tail.queue_next = old_head;
199}315}
200316
201fn recycle(el: *EventLoop, fiber: *Fiber) void {317fn recycle(el: *EventLoop, fiber: *Fiber) void {
318 const thread: *Thread = .current(el);
202 std.log.debug("recyling {*}", .{fiber});319 std.log.debug("recyling {*}", .{fiber});
320 assert(fiber.queue_next == null);
203 @memset(fiber.allocatedSlice(), undefined);321 @memset(fiber.allocatedSlice(), undefined);
204 el.mutex.lock();322 fiber.queue_next = thread.free_queue;
205 defer el.mutex.unlock();323 thread.free_queue = fiber;
206 el.free.append(&fiber.queue_node);
207}324}
208325
209fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {326fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {
210 message.handle(el);327 message.handle(el);
211 el.idle();328 const thread: *Thread = &el.threads.allocated[0];
329 el.idle(thread);
212 el.yield(&el.main_fiber, .nothing);330 el.yield(&el.main_fiber, .nothing);
213 unreachable; // switched to dead fiber331 unreachable; // switched to dead fiber
214}332}
215333
216fn threadEntry(el: *EventLoop, index: usize) void {334fn threadEntry(el: *EventLoop, index: u32) void {
217 thread_index = @intCast(index);335 Thread.index = index;
218 const thread: *Thread = &el.threads.items[index];336 const thread: *Thread = &el.threads.allocated[index];
219 std.log.debug("created thread idle {*}", .{&thread.idle_context});337 std.log.debug("created thread idle {*}", .{&thread.idle_context});
220 thread.io_uring = IoUring.init(io_uring_entries, 0) catch |err| {338 el.idle(thread);
221 std.log.warn("exiting worker thread during init due to io_uring init failure: {s}", .{@errorName(err)});
222 return;
223 };
224 thread.current_context = &thread.idle_context;
225 el.idle();
226}339}
227340
228const CompletionKey = enum(u64) {341const Completion = struct {
229 queue_len_futex_wait = 1,342 const Key = enum(usize) {
230 _,343 unused,
344 wakeup,
345 cancel,
346 cleanup,
347 exit,
348 /// *Fiber
349 _,
350 };
351 result: i32,
352 flags: u32,
231};353};
232354
233fn idle(el: *EventLoop) void {355fn idle(el: *EventLoop, thread: *Thread) void {
234 const thread: *Thread = &el.threads.items[thread_index];356 var maybe_ready_fiber: ?*Fiber = null;
235 const iou = &thread.io_uring;
236 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
237 var queue_len_futex_is_scheduled: bool = false;
238
239 while (true) {357 while (true) {
240 el.yield(null, .nothing);358 el.yield(maybe_ready_fiber, .nothing);
241 if (@atomicLoad(bool, &el.exiting, .acquire)) return;359 maybe_ready_fiber = null;
242 if (!queue_len_futex_is_scheduled) {360 _ = thread.io_uring.submit_and_wait(1) catch |err| switch (err) {
243 const sqe = getSqe(&thread.io_uring);361 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
244 sqe.prep_rw(.FUTEX_WAIT, std.os.linux.FUTEX2.SIZE_U32 | std.os.linux.FUTEX2.PRIVATE, @intFromPtr(&el.queue_len), 0, 0);362 else => |e| @panic(@errorName(e)),
245 sqe.addr3 = std.math.maxInt(u32);
246 sqe.user_data = @intFromEnum(CompletionKey.queue_len_futex_wait);
247 queue_len_futex_is_scheduled = true;
248 }
249 _ = iou.submit_and_wait(1) catch |err| switch (err) {
250 error.SignalInterrupt => std.log.debug("submit_and_wait: SignalInterrupt", .{}),
251 else => @panic(@errorName(err)),
252 };363 };
253 for (cqes_buffer[0 .. iou.copy_cqes(&cqes_buffer, 1) catch |err| switch (err) {364 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
365 var maybe_ready_queue: ?Fiber.Queue = null;
366 for (cqes_buffer[0 .. thread.io_uring.copy_cqes(&cqes_buffer, 0) catch |err| switch (err) {
254 error.SignalInterrupt => cqes_len: {367 error.SignalInterrupt => cqes_len: {
255 std.log.debug("copy_cqes: SignalInterrupt", .{});368 std.log.warn("copy_cqes failed with SignalInterrupt", .{});
256 break :cqes_len 0;369 break :cqes_len 0;
257 },370 },
258 else => @panic(@errorName(err)),371 else => |e| @panic(@errorName(e)),
259 }]) |cqe| switch (@as(CompletionKey, @enumFromInt(cqe.user_data))) {372 }]) |cqe| switch (@as(Completion.Key, @enumFromInt(cqe.user_data))) {
260 .queue_len_futex_wait => {373 .unused => unreachable, // bad submission queued?
261 switch (errno(cqe.res)) {374 .wakeup => {},
262 .SUCCESS, .AGAIN => {},375 .cancel => {},
263 .INVAL => unreachable,376 .cleanup => @panic("failed to notify other threads that we are exiting"),
264 else => |err| {377 .exit => {
265 std.posix.unexpectedErrno(err) catch {};378 assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
266 @panic("unexpected");379 return;
267 },
268 }
269 std.log.debug("{*} woken up with queue size of {d}", .{
270 &thread.idle_context,
271 @atomicLoad(u32, &el.queue_len, .unordered),
272 });
273 queue_len_futex_is_scheduled = false;
274 },380 },
275 _ => {381 _ => {
276 const fiber: *Fiber = @ptrFromInt(cqe.user_data);382 const fiber: *Fiber = @ptrFromInt(cqe.user_data);
277 const res: *i32 = @ptrCast(@alignCast(fiber.resultPointer()));383 assert(fiber.queue_next == null);
278 res.* = cqe.res;384 fiber.resultPointer(Completion).* = .{
279 el.schedule(fiber);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 };
280 },392 },
281 };393 };
394 if (maybe_ready_queue) |ready_queue| el.schedule(thread, ready_queue);
282 }395 }
283}396}
284397
...@@ -296,18 +409,37 @@ const SwitchMessage = struct {...@@ -296,18 +409,37 @@ const SwitchMessage = struct {
296 };409 };
297410
298 fn handle(message: *const SwitchMessage, el: *EventLoop) void {411 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
299 const thread: *Thread = &el.threads.items[thread_index];412 const thread: *Thread = .current(el);
300 thread.current_context = message.contexts.ready;413 thread.current_context = message.contexts.ready;
301 switch (message.pending_task) {414 switch (message.pending_task) {
302 .nothing => {},415 .nothing => {},
303 .register_awaiter => |awaiter| {416 .register_awaiter => |awaiter| {
304 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));417 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
305 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);418 if (@atomicRmw(
419 ?*Fiber,
420 awaiter,
421 .Xchg,
422 prev_fiber,
423 .acq_rel,
424 ) == Fiber.finished) el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
306 },425 },
307 .exit => {426 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {
308 @atomicStore(bool, &el.exiting, true, .unordered);427 getSqe(&thread.io_uring).* = .{
309 @atomicStore(u32, &el.queue_len, std.math.maxInt(u32), .release);428 .opcode = .MSG_RING,
310 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(u32), std.math.maxInt(i32), std.os.linux.FUTEX2.SIZE_U32 | std.os.linux.FUTEX2.PRIVATE); // TODO: use io_uring429 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
430 .ioprio = 0,
431 .fd = each_thread.io_uring.fd,
432 .off = @intFromEnum(Completion.Key.exit),
433 .addr = 0,
434 .len = 0,
435 .rw_flags = 0,
436 .user_data = @intFromEnum(Completion.Key.cleanup),
437 .buf_index = 0,
438 .personality = 0,
439 .splice_fd_in = 0,
440 .addr3 = 0,
441 .resv = 0,
442 };
311 },443 },
312 }444 }
313 }445 }
...@@ -374,7 +506,27 @@ fn fiberEntry() callconv(.naked) void {...@@ -374,7 +506,27 @@ fn fiberEntry() callconv(.naked) void {
374 }506 }
375}507}
376508
377pub fn @"async"(509const AsyncClosure = struct {
510 event_loop: *EventLoop,
511 fiber: *Fiber,
512 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
513 result_align: Alignment,
514
515 fn contextPointer(closure: *AsyncClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
516 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(AsyncClosure));
517 }
518
519 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(AsyncClosure))) noreturn {
520 message.handle(closure.event_loop);
521 std.log.debug("{*} performing async", .{closure.fiber});
522 closure.start(closure.contextPointer(), closure.fiber.resultBytes(closure.result_align));
523 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
524 closure.event_loop.yield(awaiter, .nothing);
525 unreachable; // switched to dead fiber
526 }
527};
528
529fn @"async"(
378 userdata: ?*anyopaque,530 userdata: ?*anyopaque,
379 result: []u8,531 result: []u8,
380 result_alignment: Alignment,532 result_alignment: Alignment,
...@@ -407,58 +559,79 @@ pub fn @"async"(...@@ -407,58 +559,79 @@ pub fn @"async"(
407 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),559 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
408 },560 },
409 .awaiter = null,561 .awaiter = null,
410 .queue_node = undefined,562 .queue_next = null,
411 .result_align = result_alignment,563 .can_cancel = false,
564 .canceled = false,
412 };565 };
413 closure.* = .{566 closure.* = .{
414 .event_loop = event_loop,567 .event_loop = event_loop,
415 .fiber = fiber,568 .fiber = fiber,
416 .start = start,569 .start = start,
570 .result_align = result_alignment,
417 };571 };
418 @memcpy(closure.contextPointer(), context);572 @memcpy(closure.contextPointer(), context);
419573
420 event_loop.schedule(fiber);574 event_loop.schedule(.current(event_loop), .{ .head = fiber, .tail = fiber });
421 return @ptrCast(fiber);575 return @ptrCast(fiber);
422}576}
423577
424const AsyncClosure = struct {578fn @"await"(
425 event_loop: *EventLoop,579 userdata: ?*anyopaque,
426 fiber: *Fiber,580 any_future: *std.Io.AnyFuture,
427 start: *const fn (context: *const anyopaque, result: *anyopaque) void,581 result: []u8,
428582 result_alignment: Alignment,
429 fn contextPointer(closure: *AsyncClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {583) void {
430 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(AsyncClosure));
431 }
432
433 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(AsyncClosure))) noreturn {
434 message.handle(closure.event_loop);
435 std.log.debug("{*} performing async", .{closure.fiber});
436 closure.start(closure.contextPointer(), closure.fiber.resultPointer());
437 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
438 closure.event_loop.yield(awaiter, .nothing);
439 unreachable; // switched to dead fiber
440 }
441};
442
443pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
444 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));584 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
445 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));585 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
446 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });586 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
447 @memcpy(result, future_fiber.resultPointer());587 @memcpy(result, future_fiber.resultBytes(result_alignment));
448 event_loop.recycle(future_fiber);588 event_loop.recycle(future_fiber);
449}589}
450590
451pub fn cancel(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {591fn cancel(
592 userdata: ?*anyopaque,
593 any_future: *std.Io.AnyFuture,
594 result: []u8,
595 result_alignment: Alignment,
596) void {
452 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));597 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
453 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));598 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
454 // TODO set a flag that makes all IO operations for this fiber return error.Canceled599 @atomicStore(bool, &future_fiber.canceled, true, .release);
455 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });600 if (@atomicLoad(bool, &future_fiber.can_cancel, .acquire)) {
456 @memcpy(result, future_fiber.resultPointer());601 const thread: *Thread = .current(event_loop);
457 event_loop.recycle(future_fiber);602 getSqe(&thread.io_uring).* = .{
603 .opcode = .ASYNC_CANCEL,
604 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
605 .ioprio = 0,
606 .fd = 0,
607 .off = 0,
608 .addr = @intFromPtr(future_fiber),
609 .len = 0,
610 .rw_flags = 0,
611 .user_data = @intFromEnum(Completion.Key.cancel),
612 .buf_index = 0,
613 .personality = 0,
614 .splice_fd_in = 0,
615 .addr3 = 0,
616 .resv = 0,
617 };
618 }
619 @"await"(userdata, any_future, result, result_alignment);
620}
621
622fn cancelRequested(userdata: ?*anyopaque) bool {
623 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
624 const thread: *Thread = .current(event_loop);
625 return thread.currentFiber().canceled;
458}626}
459627
460pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.CreateFlags) std.fs.File.OpenError!std.fs.File {628pub fn createFile(
461 const el: *EventLoop = @ptrCast(@alignCast(userdata));629 userdata: ?*anyopaque,
630 dir: std.fs.Dir,
631 sub_path: []const u8,
632 flags: Io.CreateFlags,
633) Io.FileOpenError!std.fs.File {
634 const el: *EventLoop = @alignCast(@ptrCast(userdata));
462635
463 const posix = std.posix;636 const posix = std.posix;
464 const sub_path_c = try posix.toPosixPath(sub_path);637 const sub_path_c = try posix.toPosixPath(sub_path);
...@@ -497,22 +670,24 @@ pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8,...@@ -497,22 +670,24 @@ pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8,
497 @panic("TODO");670 @panic("TODO");
498 }671 }
499672
500 const thread: *Thread = &el.threads.items[thread_index];673 const thread: *Thread = .current(el);
501 const iou = &thread.io_uring;674 const iou = &thread.io_uring;
502 const sqe = getSqe(iou);
503 const fiber = thread.currentFiber();675 const fiber = thread.currentFiber();
676 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
504677
678 const sqe = getSqe(iou);
505 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, flags.mode);679 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, flags.mode);
506 sqe.user_data = @intFromPtr(fiber);680 sqe.user_data = @intFromPtr(fiber);
507681
682 @atomicStore(bool, &fiber.can_cancel, true, .release);
508 el.yield(null, .nothing);683 el.yield(null, .nothing);
684 @atomicStore(bool, &fiber.can_cancel, false, .release);
509685
510 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));686 const completion = fiber.resultPointer(Completion);
511 const rc = result.*;687 switch (errno(completion.result)) {
512 switch (errno(rc)) {688 .SUCCESS => return .{ .handle = completion.result },
513 .SUCCESS => return .{ .handle = rc },
514 .INTR => @panic("TODO is this reachable?"),689 .INTR => @panic("TODO is this reachable?"),
515 .CANCELED => @panic("TODO figure out how this error code fits into things"),690 .CANCELED => return error.AsyncCancel,
516691
517 .FAULT => unreachable,692 .FAULT => unreachable,
518 .INVAL => return error.BadPathName,693 .INVAL => return error.BadPathName,
...@@ -541,8 +716,17 @@ pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8,...@@ -541,8 +716,17 @@ pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8,
541 }716 }
542}717}
543718
544pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.OpenFlags) std.fs.File.OpenError!std.fs.File {719pub fn openFile(
545 const el: *EventLoop = @ptrCast(@alignCast(userdata));720 userdata: ?*anyopaque,
721 dir: std.fs.Dir,
722 sub_path: []const u8,
723 flags: Io.OpenFlags,
724) Io.FileOpenError!std.fs.File {
725 const el: *EventLoop = @alignCast(@ptrCast(userdata));
726 const thread: *Thread = .current(el);
727 const iou = &thread.io_uring;
728 const fiber = thread.currentFiber();
729 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
546730
547 const posix = std.posix;731 const posix = std.posix;
548 const sub_path_c = try posix.toPosixPath(sub_path);732 const sub_path_c = try posix.toPosixPath(sub_path);
...@@ -587,22 +771,19 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl...@@ -587,22 +771,19 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl
587 @panic("TODO");771 @panic("TODO");
588 }772 }
589773
590 const thread: *Thread = &el.threads.items[thread_index];
591 const iou = &thread.io_uring;
592 const sqe = getSqe(iou);774 const sqe = getSqe(iou);
593 const fiber = thread.currentFiber();
594
595 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);775 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);
596 sqe.user_data = @intFromPtr(fiber);776 sqe.user_data = @intFromPtr(fiber);
597777
778 @atomicStore(bool, &fiber.can_cancel, true, .release);
598 el.yield(null, .nothing);779 el.yield(null, .nothing);
780 @atomicStore(bool, &fiber.can_cancel, false, .release);
599781
600 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));782 const completion = fiber.resultPointer(Completion);
601 const rc = result.*;783 switch (errno(completion.result)) {
602 switch (errno(rc)) {784 .SUCCESS => return .{ .handle = completion.result },
603 .SUCCESS => return .{ .handle = rc },
604 .INTR => @panic("TODO is this reachable?"),785 .INTR => @panic("TODO is this reachable?"),
605 .CANCELED => @panic("TODO figure out how this error code fits into things"),786 .CANCELED => return error.AsyncCancel,
606787
607 .FAULT => unreachable,788 .FAULT => unreachable,
608 .INVAL => return error.BadPathName,789 .INVAL => return error.BadPathName,
...@@ -631,63 +812,49 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl...@@ -631,63 +812,49 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl
631 }812 }
632}813}
633814
634fn errno(signed: i32) std.posix.E {
635 const int = if (signed > -4096 and signed < 0) -signed else 0;
636 return @enumFromInt(int);
637}
638
639fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
640 return iou.get_sqe() catch @panic("TODO: handle submission queue full");
641}
642
643pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {815pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
644 const el: *EventLoop = @ptrCast(@alignCast(userdata));816 const el: *EventLoop = @alignCast(@ptrCast(userdata));
645817 const thread: *Thread = .current(el);
646 const posix = std.posix;
647
648 const thread: *Thread = &el.threads.items[thread_index];
649 const iou = &thread.io_uring;818 const iou = &thread.io_uring;
650 const sqe = getSqe(iou);
651 const fiber = thread.currentFiber();819 const fiber = thread.currentFiber();
652820
821 const sqe = getSqe(iou);
653 sqe.prep_close(file.handle);822 sqe.prep_close(file.handle);
654 sqe.user_data = @intFromPtr(fiber);823 sqe.user_data = @intFromPtr(fiber);
655824
656 el.yield(null, .nothing);825 el.yield(null, .nothing);
657826
658 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));827 const completion = fiber.resultPointer(Completion);
659 const rc = result.*;828 switch (errno(completion.result)) {
660 switch (errno(rc)) {
661 .SUCCESS => return,829 .SUCCESS => return,
662 .INTR => @panic("TODO is this reachable?"),830 .INTR => @panic("TODO is this reachable?"),
663 .CANCELED => @panic("TODO figure out how this error code fits into things"),831 .CANCELED => return,
664832
665 .BADF => unreachable, // Always a race condition.833 .BADF => unreachable, // Always a race condition.
666 else => return,834 else => return,
667 }835 }
668}836}
669837
670pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File.ReadError!usize {838pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadError!usize {
671 const el: *EventLoop = @ptrCast(@alignCast(userdata));839 const el: *EventLoop = @alignCast(@ptrCast(userdata));
672840 const thread: *Thread = .current(el);
673 const posix = std.posix;
674
675 const thread: *Thread = &el.threads.items[thread_index];
676 const iou = &thread.io_uring;841 const iou = &thread.io_uring;
677 const sqe = getSqe(iou);
678 const fiber = thread.currentFiber();842 const fiber = thread.currentFiber();
843 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
679844
845 const sqe = getSqe(iou);
680 sqe.prep_read(file.handle, buffer, std.math.maxInt(u64));846 sqe.prep_read(file.handle, buffer, std.math.maxInt(u64));
681 sqe.user_data = @intFromPtr(fiber);847 sqe.user_data = @intFromPtr(fiber);
682848
849 @atomicStore(bool, &fiber.can_cancel, true, .release);
683 el.yield(null, .nothing);850 el.yield(null, .nothing);
851 @atomicStore(bool, &fiber.can_cancel, false, .release);
684852
685 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));853 const completion = fiber.resultPointer(Completion);
686 const rc = result.*;854 switch (errno(completion.result)) {
687 switch (errno(rc)) {855 .SUCCESS => return @as(u32, @bitCast(completion.result)),
688 .SUCCESS => return @as(u32, @bitCast(rc)),
689 .INTR => @panic("TODO is this reachable?"),856 .INTR => @panic("TODO is this reachable?"),
690 .CANCELED => @panic("TODO figure out how this error code fits into things"),857 .CANCELED => return error.AsyncCancel,
691858
692 .INVAL => unreachable,859 .INVAL => unreachable,
693 .FAULT => unreachable,860 .FAULT => unreachable,
...@@ -701,31 +868,31 @@ pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File....@@ -701,31 +868,31 @@ pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File.
701 .NOTCONN => return error.SocketNotConnected,868 .NOTCONN => return error.SocketNotConnected,
702 .CONNRESET => return error.ConnectionResetByPeer,869 .CONNRESET => return error.ConnectionResetByPeer,
703 .TIMEDOUT => return error.ConnectionTimedOut,870 .TIMEDOUT => return error.ConnectionTimedOut,
704 else => |err| return posix.unexpectedErrno(err),871 else => |err| return std.posix.unexpectedErrno(err),
705 }872 }
706}873}
707874
708pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) std.fs.File.WriteError!usize {875pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.FileWriteError!usize {
709 const el: *EventLoop = @ptrCast(@alignCast(userdata));876 const el: *EventLoop = @alignCast(@ptrCast(userdata));
710
711 const posix = std.posix;
712877
713 const thread: *Thread = &el.threads.items[thread_index];878 const thread: *Thread = .current(el);
714 const iou = &thread.io_uring;879 const iou = &thread.io_uring;
715 const sqe = getSqe(iou);
716 const fiber = thread.currentFiber();880 const fiber = thread.currentFiber();
881 if (@atomicLoad(bool, &fiber.canceled, .acquire)) return error.AsyncCancel;
717882
883 const sqe = getSqe(iou);
718 sqe.prep_write(file.handle, buffer, std.math.maxInt(u64));884 sqe.prep_write(file.handle, buffer, std.math.maxInt(u64));
719 sqe.user_data = @intFromPtr(fiber);885 sqe.user_data = @intFromPtr(fiber);
720886
887 @atomicStore(bool, &fiber.can_cancel, true, .release);
721 el.yield(null, .nothing);888 el.yield(null, .nothing);
889 @atomicStore(bool, &fiber.can_cancel, false, .release);
722890
723 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));891 const completion = fiber.resultPointer(Completion);
724 const rc = result.*;892 switch (errno(completion.result)) {
725 switch (errno(rc)) {893 .SUCCESS => return @as(u32, @bitCast(completion.result)),
726 .SUCCESS => return @as(u32, @bitCast(rc)),
727 .INTR => @panic("TODO is this reachable?"),894 .INTR => @panic("TODO is this reachable?"),
728 .CANCELED => @panic("TODO figure out how this error code fits into things"),895 .CANCELED => return error.AsyncCancel,
729896
730 .INVAL => return error.InvalidArgument,897 .INVAL => return error.InvalidArgument,
731 .FAULT => unreachable,898 .FAULT => unreachable,
...@@ -744,6 +911,15 @@ pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) std.f...@@ -744,6 +911,15 @@ pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) std.f
744 .BUSY => return error.DeviceBusy,911 .BUSY => return error.DeviceBusy,
745 .NXIO => return error.NoDevice,912 .NXIO => return error.NoDevice,
746 .MSGSIZE => return error.MessageTooBig,913 .MSGSIZE => return error.MessageTooBig,
747 else => |err| return posix.unexpectedErrno(err),914 else => |err| return std.posix.unexpectedErrno(err),
748 }915 }
749}916}
917
918fn errno(signed: i32) std.posix.E {
919 const int = if (signed > -4096 and signed < 0) -signed else 0;
920 return @enumFromInt(int);
921}
922
923fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
924 return iou.get_sqe() catch @panic("TODO: handle submission queue full");
925}
lib/std/Thread/Pool.zig+14-2
...@@ -435,13 +435,25 @@ fn @"async"(...@@ -435,13 +435,25 @@ fn @"async"(
435 return @ptrCast(closure);435 return @ptrCast(closure);
436}436}
437437
438fn @"await"(userdata: ?*anyopaque, any_future: *Io.AnyFuture, result: []u8) void {438fn @"await"(
439 userdata: ?*anyopaque,
440 any_future: *std.Io.AnyFuture,
441 result: []u8,
442 result_alignment: std.mem.Alignment,
443) void {
444 _ = result_alignment;
439 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));445 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
440 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));446 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));
441 closure.waitAndFree(pool.allocator, result);447 closure.waitAndFree(pool.allocator, result);
442}448}
443449
444fn cancel(userdata: ?*anyopaque, any_future: *Io.AnyFuture, result: []u8) void {450fn cancel(
451 userdata: ?*anyopaque,
452 any_future: *Io.AnyFuture,
453 result: []u8,
454 result_alignment: std.mem.Alignment,
455) void {
456 _ = result_alignment;
445 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));457 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
446 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));458 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));
447 @atomicStore(bool, &closure.cancel_flag, true, .seq_cst);459 @atomicStore(bool, &closure.cancel_flag, true, .seq_cst);