1const Kqueue = @This();
2const builtin = @import("builtin");
3
4const std = @import("../std.zig");
5const Io = std.Io;
6const Dir = std.Io.Dir;
7const File = std.Io.File;
8const net = std.Io.net;
9const assert = std.debug.assert;
10const Allocator = std.mem.Allocator;
11const Alignment = std.mem.Alignment;
12const IpAddress = std.Io.net.IpAddress;
13const errnoBug = std.Io.Threaded.errnoBug;
14const closeFd = std.Io.Threaded.closeFd;
15const posix = std.posix;
16const posixSocketModeProtocol = Io.Threaded.posixSocketModeProtocol;
17
18/// Must be a thread-safe allocator.
19gpa: Allocator,
20mutex: Io.Mutex,
21main_fiber_buffer: [@sizeOf(Fiber) + Fiber.max_result_size]u8 align(@alignOf(Fiber)),
22threads: Thread.List,
23
24/// Empirically saw >128KB being used by the self-hosted backend to panic.
25const idle_stack_size = 256 * 1024;
26
27const max_idle_search = 4;
28const max_steal_ready_search = 4;
29const max_iovecs_len = 8;
30
31const changes_buffer_len = 64;
32
33const Thread = struct {
34 thread: std.Thread,
35 idle_context: Io.fiber.Context,
36 current_context: *Io.fiber.Context,
37 ready_queue: ?*Fiber,
38 kq_fd: posix.fd_t,
39 idle_search_index: u32,
40 steal_ready_search_index: u32,
41 /// For ensuring multiple fibers waiting on the same file descriptor and
42 /// filter use the same kevent.
43 wait_queues: std.array_hash_map.Auto(WaitQueueKey, *Fiber),
44
45 const WaitQueueKey = struct {
46 ident: usize,
47 filter: i32,
48 };
49
50 const canceling: ?*Thread = @ptrFromInt(@alignOf(Thread));
51
52 threadlocal var self: *Thread = undefined;
53
54 fn current() *Thread {
55 return self;
56 }
57
58 fn currentFiber(thread: *Thread) *Fiber {
59 return @fieldParentPtr("context", thread.current_context);
60 }
61
62 const List = struct {
63 allocated: []Thread,
64 reserved: u32,
65 active: u32,
66 };
67
68 fn deinit(thread: *Thread, gpa: Allocator) void {
69 closeFd(thread.kq_fd);
70 assert(thread.wait_queues.count() == 0);
71 thread.wait_queues.deinit(gpa);
72 thread.* = undefined;
73 }
74};
75
76const Fiber = struct {
77 required_align: void align(4),
78 context: Io.fiber.Context,
79 awaiter: ?*Fiber,
80 queue_next: ?*Fiber,
81 cancel_thread: ?*Thread,
82 awaiting_completions: std.bit_set.Static(3),
83
84 const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
85
86 const max_result_align: Alignment = .@"16";
87 const max_result_size = max_result_align.forward(64);
88 /// This includes any stack realignments that need to happen, and also the
89 /// initial frame return address slot and argument frame, depending on target.
90 const min_stack_size = 4 * 1024 * 1024;
91 const max_context_align: Alignment = .@"16";
92 const max_context_size = max_context_align.forward(1024);
93 const max_closure_size: usize = @sizeOf(AsyncClosure);
94 const max_closure_align: Alignment = .of(AsyncClosure);
95 const allocation_size = std.mem.alignForward(
96 usize,
97 max_closure_align.max(max_context_align).forward(
98 max_result_align.forward(@sizeOf(Fiber)) + max_result_size + min_stack_size,
99 ) + max_closure_size + max_context_size,
100 std.heap.page_size_max,
101 );
102
103 fn allocate(k: *Kqueue) error{OutOfMemory}!*Fiber {
104 return @ptrCast(try k.gpa.alignedAlloc(u8, .of(Fiber), allocation_size));
105 }
106
107 fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {
108 return @as([*]align(@alignOf(Fiber)) u8, @ptrCast(f))[0..allocation_size];
109 }
110
111 fn allocatedEnd(f: *Fiber) [*]u8 {
112 const allocated_slice = f.allocatedSlice();
113 return allocated_slice[allocated_slice.len..].ptr;
114 }
115
116 fn resultPointer(f: *Fiber, comptime Result: type) *Result {
117 return @ptrCast(@alignCast(f.resultBytes(.of(Result))));
118 }
119
120 fn resultBytes(f: *Fiber, alignment: Alignment) [*]u8 {
121 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
122 }
123
124 fn enterCancelRegion(fiber: *Fiber, thread: *Thread) error{Canceled}!void {
125 if (@cmpxchgStrong(
126 ?*Thread,
127 &fiber.cancel_thread,
128 null,
129 thread,
130 .acq_rel,
131 .acquire,
132 )) |cancel_thread| {
133 assert(cancel_thread == Thread.canceling);
134 return error.Canceled;
135 }
136 }
137
138 fn exitCancelRegion(fiber: *Fiber, thread: *Thread) void {
139 if (@cmpxchgStrong(
140 ?*Thread,
141 &fiber.cancel_thread,
142 thread,
143 null,
144 .acq_rel,
145 .acquire,
146 )) |cancel_thread| assert(cancel_thread == Thread.canceling);
147 }
148
149 const Queue = struct { head: *Fiber, tail: *Fiber };
150};
151
152fn recycle(k: *Kqueue, fiber: *Fiber) void {
153 std.log.debug("recyling {*}", .{fiber});
154 assert(fiber.queue_next == null);
155 k.gpa.free(fiber.allocatedSlice());
156}
157
158pub const InitOptions = struct {
159 n_threads: ?usize = null,
160};
161
162pub const InitError = Allocator.Error || CreateFileDescriptorError;
163
164pub fn init(k: *Kqueue, gpa: Allocator, options: InitOptions) !void {
165 assert(options.n_threads != 0);
166
167 const n_threads = @max(1, options.n_threads orelse std.Thread.getCpuCount() catch 1);
168 const threads_size = n_threads * @sizeOf(Thread);
169 const idle_stack_end_offset = std.mem.alignForward(usize, threads_size + idle_stack_size, std.heap.page_size_max);
170 const allocated_slice = try gpa.alignedAlloc(u8, .of(Thread), idle_stack_end_offset);
171 errdefer gpa.free(allocated_slice);
172 k.* = .{
173 .gpa = gpa,
174 .mutex = .init,
175 .main_fiber_buffer = undefined,
176 .threads = .{
177 .allocated = @ptrCast(allocated_slice[0..threads_size]),
178 .reserved = 1,
179 .active = 1,
180 },
181 };
182 const main_fiber: *Fiber = @ptrCast(&k.main_fiber_buffer);
183 main_fiber.* = .{
184 .required_align = {},
185 .context = undefined,
186 .awaiter = null,
187 .queue_next = null,
188 .cancel_thread = null,
189 .awaiting_completions = .empty,
190 };
191 const main_thread = &k.threads.allocated[0];
192 Thread.self = main_thread;
193 const idle_stack_end: [*]align(16) usize = @ptrCast(@alignCast(allocated_slice[idle_stack_end_offset..].ptr));
194 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(k)};
195 main_thread.* = .{
196 .thread = undefined,
197 .idle_context = switch (builtin.cpu.arch) {
198 .aarch64 => .{
199 .sp = @intFromPtr(idle_stack_end),
200 .fp = 0,
201 .pc = @intFromPtr(&mainIdleEntry),
202 },
203 .x86_64 => .{
204 .rsp = @intFromPtr(idle_stack_end - 1),
205 .rbp = 0,
206 .rip = @intFromPtr(&mainIdleEntry),
207 },
208 else => @compileError("unimplemented architecture"),
209 },
210 .current_context = &main_fiber.context,
211 .ready_queue = null,
212 .kq_fd = try createFileDescriptor(),
213 .idle_search_index = 1,
214 .steal_ready_search_index = 1,
215 .wait_queues = .empty,
216 };
217 errdefer closeFd(main_thread.kq_fd);
218 std.log.debug("created main idle {*}", .{&main_thread.idle_context});
219 std.log.debug("created main {*}", .{main_fiber});
220}
221
222pub fn deinit(k: *Kqueue) void {
223 const active_threads = @atomicLoad(u32, &k.threads.active, .acquire);
224 for (k.threads.allocated[0..active_threads]) |*thread| {
225 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
226 assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async
227 }
228 k.yield(null, .exit);
229 const main_thread = &k.threads.allocated[0];
230 const gpa = k.gpa;
231 main_thread.deinit(gpa);
232 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @ptrCast(@alignCast(k.threads.allocated.ptr));
233 const idle_stack_end_offset = std.mem.alignForward(usize, k.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
234 for (k.threads.allocated[1..active_threads]) |*thread| thread.thread.join();
235 gpa.free(allocated_ptr[0..idle_stack_end_offset]);
236 k.* = undefined;
237}
238
239pub const CreateFileDescriptorError = error{
240 /// The per-process limit on the number of open file descriptors has been reached.
241 ProcessFdQuotaExceeded,
242 /// The system-wide limit on the total number of open files has been reached.
243 SystemFdQuotaExceeded,
244} || Io.UnexpectedError;
245
246pub fn createFileDescriptor() CreateFileDescriptorError!posix.fd_t {
247 const rc = posix.system.kqueue();
248 switch (posix.errno(rc)) {
249 .SUCCESS => return @intCast(rc),
250 .MFILE => return error.ProcessFdQuotaExceeded,
251 .NFILE => return error.SystemFdQuotaExceeded,
252 else => |err| return posix.unexpectedErrno(err),
253 }
254}
255
256fn findReadyFiber(k: *Kqueue, thread: *Thread) ?*Fiber {
257 if (@atomicRmw(?*Fiber, &thread.ready_queue, .Xchg, Fiber.finished, .acquire)) |ready_fiber| {
258 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
259 ready_fiber.queue_next = null;
260 return ready_fiber;
261 }
262 const active_threads = @atomicLoad(u32, &k.threads.active, .acquire);
263 for (0..@min(max_steal_ready_search, active_threads)) |_| {
264 defer thread.steal_ready_search_index += 1;
265 if (thread.steal_ready_search_index == active_threads) thread.steal_ready_search_index = 0;
266 const steal_ready_search_thread = &k.threads.allocated[0..active_threads][thread.steal_ready_search_index];
267 if (steal_ready_search_thread == thread) continue;
268 const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
269 if (ready_fiber == Fiber.finished) continue;
270 if (@cmpxchgWeak(
271 ?*Fiber,
272 &steal_ready_search_thread.ready_queue,
273 ready_fiber,
274 null,
275 .acquire,
276 .monotonic,
277 )) |_| continue;
278 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
279 ready_fiber.queue_next = null;
280 return ready_fiber;
281 }
282 // couldn't find anything to do, so we are now open for business
283 @atomicStore(?*Fiber, &thread.ready_queue, null, .monotonic);
284 return null;
285}
286
287fn yield(k: *Kqueue, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
288 const thread: *Thread = .current();
289 const ready_context = if (maybe_ready_fiber orelse k.findReadyFiber(thread)) |ready_fiber|
290 &ready_fiber.context
291 else
292 &thread.idle_context;
293 const message: SwitchMessage = .{
294 .contexts = .{
295 .old = thread.current_context,
296 .new = ready_context,
297 },
298 .pending_task = pending_task,
299 };
300 std.log.debug("switching from {*} to {*}", .{ message.contexts.old, message.contexts.new });
301 contextSwitch(&message).handle(k);
302}
303
304fn schedule(k: *Kqueue, thread: *Thread, ready_queue: Fiber.Queue) void {
305 {
306 var fiber = ready_queue.head;
307 while (true) {
308 std.log.debug("scheduling {*}", .{fiber});
309 fiber = fiber.queue_next orelse break;
310 }
311 assert(fiber == ready_queue.tail);
312 }
313 // shared fields of previous `Thread` must be initialized before later ones are marked as active
314 const new_thread_index = @atomicLoad(u32, &k.threads.active, .acquire);
315 for (0..@min(max_idle_search, new_thread_index)) |_| {
316 defer thread.idle_search_index += 1;
317 if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;
318 const idle_search_thread = &k.threads.allocated[0..new_thread_index][thread.idle_search_index];
319 if (idle_search_thread == thread) continue;
320 if (@cmpxchgWeak(
321 ?*Fiber,
322 &idle_search_thread.ready_queue,
323 null,
324 ready_queue.head,
325 .release,
326 .monotonic,
327 )) |_| continue;
328 const changes = [_]posix.Kevent{
329 .{
330 .ident = 0,
331 .filter = std.c.EVFILT.USER,
332 .flags = std.c.EV.ADD | std.c.EV.ONESHOT,
333 .fflags = std.c.NOTE.TRIGGER,
334 .data = 0,
335 .udata = @backingInt(Completion.UserData.wakeup),
336 },
337 };
338 // If an error occurs it only pessimises scheduling.
339 _ = kevent(idle_search_thread.kq_fd, &changes, &.{}, null) catch |err| {
340 // TODO handle EINTR for cancellation purposes
341 @panic(@errorName(err)); // TODO
342 };
343 return;
344 }
345 spawn_thread: {
346 // previous failed reservations must have completed before retrying
347 if (new_thread_index == k.threads.allocated.len or @cmpxchgWeak(
348 u32,
349 &k.threads.reserved,
350 new_thread_index,
351 new_thread_index + 1,
352 .acquire,
353 .monotonic,
354 ) != null) break :spawn_thread;
355 const new_thread = &k.threads.allocated[new_thread_index];
356 const next_thread_index = new_thread_index + 1;
357 new_thread.* = .{
358 .thread = undefined,
359 .idle_context = undefined,
360 .current_context = &new_thread.idle_context,
361 .ready_queue = ready_queue.head,
362 .kq_fd = createFileDescriptor() catch |err| {
363 @atomicStore(u32, &k.threads.reserved, new_thread_index, .release);
364 // no more access to `thread` after giving up reservation
365 std.log.warn("unable to create worker thread due to kqueue init failure: {t}", .{err});
366 break :spawn_thread;
367 },
368 .idle_search_index = 0,
369 .steal_ready_search_index = 0,
370 .wait_queues = .empty,
371 };
372 new_thread.thread = std.Thread.spawn(.{
373 .stack_size = idle_stack_size,
374 .allocator = k.gpa,
375 }, threadEntry, .{ k, new_thread_index }) catch |err| {
376 closeFd(new_thread.kq_fd);
377 @atomicStore(u32, &k.threads.reserved, new_thread_index, .release);
378 // no more access to `thread` after giving up reservation
379 std.log.warn("unable to create worker thread due spawn failure: {s}", .{@errorName(err)});
380 break :spawn_thread;
381 };
382 // shared fields of `Thread` must be initialized before being marked active
383 @atomicStore(u32, &k.threads.active, next_thread_index, .release);
384 return;
385 }
386 // nobody wanted it, so just queue it on ourselves
387 while (@cmpxchgWeak(
388 ?*Fiber,
389 &thread.ready_queue,
390 ready_queue.tail.queue_next,
391 ready_queue.head,
392 .acq_rel,
393 .acquire,
394 )) |old_head| ready_queue.tail.queue_next = old_head;
395}
396
397fn mainIdle(k: *Kqueue, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Io.fiber.Context)))) noreturn {
398 message.handle(k);
399 k.idle(&k.threads.allocated[0]);
400 k.yield(@ptrCast(&k.main_fiber_buffer), .nothing);
401 unreachable; // switched to dead fiber
402}
403
404fn threadEntry(k: *Kqueue, index: u32) void {
405 const thread: *Thread = &k.threads.allocated[index];
406 Thread.self = thread;
407 std.log.debug("created thread idle {*}", .{&thread.idle_context});
408 k.idle(thread);
409 thread.deinit(k.gpa);
410}
411
412const Completion = struct {
413 const UserData = enum(usize) {
414 unused,
415 wakeup,
416 cleanup,
417 exit,
418 /// *Fiber
419 _,
420 };
421 /// Corresponds to Kevent field.
422 flags: u16,
423 /// Corresponds to Kevent field.
424 fflags: u32,
425 /// Corresponds to Kevent field.
426 data: isize,
427};
428
429fn idle(k: *Kqueue, thread: *Thread) void {
430 var events_buffer: [changes_buffer_len]posix.Kevent = undefined;
431 var maybe_ready_fiber: ?*Fiber = null;
432 while (true) {
433 while (maybe_ready_fiber orelse k.findReadyFiber(thread)) |ready_fiber| {
434 k.yield(ready_fiber, .nothing);
435 maybe_ready_fiber = null;
436 }
437 const n = kevent(thread.kq_fd, &.{}, &events_buffer, null) catch |err| {
438 // TODO handle EINTR for cancellation purposes
439 @panic(@errorName(err)); // TODO
440 };
441 var maybe_ready_queue: ?Fiber.Queue = null;
442 for (events_buffer[0..n]) |event| switch (@as(Completion.UserData, @fromBackingInt(@intCast(event.udata)))) {
443 .unused => unreachable, // bad submission queued?
444 .wakeup => {},
445 .cleanup => @panic("failed to notify other threads that we are exiting"),
446 .exit => {
447 assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
448 return;
449 },
450 _ => {
451 const event_head_fiber: *Fiber = @ptrFromInt(event.udata);
452 const event_tail_fiber = thread.wait_queues.fetchSwapRemove(.{
453 .ident = event.ident,
454 .filter = event.filter,
455 }).?.value;
456 assert(event_tail_fiber.queue_next == null);
457
458 // TODO reevaluate this logic
459 event_head_fiber.resultPointer(Completion).* = .{
460 .flags = event.flags,
461 .fflags = event.fflags,
462 .data = event.data,
463 };
464
465 queue_ready: {
466 const head: *Fiber = if (maybe_ready_fiber == null) f: {
467 maybe_ready_fiber = event_head_fiber;
468 const next = event_head_fiber.queue_next orelse break :queue_ready;
469 event_head_fiber.queue_next = null;
470 break :f next;
471 } else event_head_fiber;
472
473 if (maybe_ready_queue) |*ready_queue| {
474 ready_queue.tail.queue_next = head;
475 ready_queue.tail = event_tail_fiber;
476 } else {
477 maybe_ready_queue = .{ .head = head, .tail = event_tail_fiber };
478 }
479 }
480 },
481 };
482 if (maybe_ready_queue) |ready_queue| k.schedule(thread, ready_queue);
483 }
484}
485
486const SwitchMessage = struct {
487 contexts: Io.fiber.Switch,
488 pending_task: PendingTask,
489
490 const PendingTask = union(enum) {
491 nothing,
492 reschedule,
493 recycle: *Fiber,
494 register_awaiter: *?*Fiber,
495 exit,
496 };
497
498 fn handle(message: *const SwitchMessage, k: *Kqueue) void {
499 const thread: *Thread = .current();
500 thread.current_context = message.contexts.new;
501 switch (message.pending_task) {
502 .nothing => {},
503 .reschedule => if (message.contexts.old != &thread.idle_context) {
504 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
505 assert(prev_fiber.queue_next == null);
506 k.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
507 },
508 .recycle => |fiber| {
509 k.recycle(fiber);
510 },
511 .register_awaiter => |awaiter| {
512 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
513 assert(prev_fiber.queue_next == null);
514 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished)
515 k.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
516 },
517 .exit => for (k.threads.allocated[0..@atomicLoad(u32, &k.threads.active, .acquire)]) |*each_thread| {
518 const changes = [_]posix.Kevent{
519 .{
520 .ident = 0,
521 .filter = std.c.EVFILT.USER,
522 .flags = std.c.EV.ADD | std.c.EV.ONESHOT,
523 .fflags = std.c.NOTE.TRIGGER,
524 .data = 0,
525 .udata = @backingInt(Completion.UserData.exit),
526 },
527 };
528 _ = kevent(each_thread.kq_fd, &changes, &.{}, null) catch |err| {
529 // TODO handle EINTR for cancellation purposes
530 @panic(@errorName(err)); // TODO
531 };
532 },
533 }
534 }
535};
536
537inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
538 return @fieldParentPtr("contexts", Io.fiber.contextSwitch(&message.contexts));
539}
540
541fn mainIdleEntry() callconv(.naked) void {
542 switch (builtin.cpu.arch) {
543 .x86_64 => asm volatile (
544 \\ movq (%%rsp), %%rdi
545 \\ jmp %[mainIdle:P]
546 :
547 : [mainIdle] "X" (&mainIdle),
548 ),
549 .aarch64 => asm volatile (
550 \\ ldr x0, [sp, #-8]
551 \\ b %[mainIdle]
552 :
553 : [mainIdle] "X" (&mainIdle),
554 ),
555 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
556 }
557}
558
559fn fiberEntry() callconv(.naked) void {
560 switch (builtin.cpu.arch) {
561 .x86_64 => asm volatile (
562 \\ leaq 8(%%rsp), %%rdi
563 \\ jmp %[AsyncClosure_call:P]
564 :
565 : [AsyncClosure_call] "X" (&AsyncClosure.call),
566 ),
567 .aarch64 => asm volatile (
568 \\ mov x0, sp
569 \\ b %[AsyncClosure_call]
570 :
571 : [AsyncClosure_call] "X" (&AsyncClosure.call),
572 ),
573 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
574 }
575}
576
577const AsyncClosure = struct {
578 kqueue: *Kqueue,
579 fiber: *Fiber,
580 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
581 result_align: Alignment,
582 already_awaited: bool,
583
584 fn contextPointer(closure: *AsyncClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
585 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(AsyncClosure));
586 }
587
588 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(AsyncClosure))) noreturn {
589 message.handle(closure.kqueue);
590 const fiber = closure.fiber;
591 std.log.debug("{*} performing async", .{fiber});
592 closure.start(closure.contextPointer(), fiber.resultBytes(closure.result_align));
593 const awaiter = @atomicRmw(?*Fiber, &fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
594 const ready_awaiter = r: {
595 const a = awaiter orelse break :r null;
596 if (@atomicRmw(bool, &closure.already_awaited, .Xchg, true, .acq_rel)) break :r null;
597 break :r a;
598 };
599 closure.kqueue.yield(ready_awaiter, .nothing);
600 unreachable; // switched to dead fiber
601 }
602
603 fn fromFiber(fiber: *Fiber) *AsyncClosure {
604 return @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
605 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
606 ) - @sizeOf(AsyncClosure));
607 }
608};
609
610pub fn io(k: *Kqueue) Io {
611 return .{
612 .userdata = k,
613 .vtable = &.{
614 .async = async,
615 .concurrent = concurrent,
616 .await = await,
617 .cancel = cancel,
618
619 .groupAsync = groupAsync,
620 .groupConcurrent = groupConcurrent,
621 .groupAwait = groupAwait,
622 .groupCancel = groupCancel,
623
624 .dirCreateDir = dirCreateDir,
625 .dirCreateDirPath = dirCreateDirPath,
626 .dirCreateDirPathOpen = dirCreateDirPathOpen,
627 .dirStat = dirStat,
628 .dirStatFile = dirStatFile,
629
630 .fileStat = fileStat,
631 .dirAccess = dirAccess,
632 .dirCreateFile = dirCreateFile,
633 .dirOpenFile = dirOpenFile,
634 .dirOpenDir = dirOpenDir,
635 .dirClose = dirClose,
636 .fileClose = fileClose,
637 .fileWriteStreaming = fileWriteStreaming,
638 .fileWritePositional = fileWritePositional,
639 .fileReadStreaming = fileReadStreaming,
640 .fileReadPositional = fileReadPositional,
641 .fileSeekBy = fileSeekBy,
642 .fileSeekTo = fileSeekTo,
643
644 .now = now,
645 .sleep = sleep,
646
647 .netListenIp = netListenIp,
648 .netListenUnix = netListenUnix,
649 .netAccept = netAccept,
650 .netBindIp = netBindIp,
651 .netConnectIp = netConnectIp,
652 .netConnectUnix = netConnectUnix,
653 .netShutdown = netShutdown,
654 .netRead = netRead,
655 .netSend = netSend,
656 .netReceive = netReceive,
657 .netInterfaceNameResolve = netInterfaceNameResolve,
658 .netInterfaceName = netInterfaceName,
659 .netLookup = netLookup,
660 },
661 };
662}
663
664fn async(
665 userdata: ?*anyopaque,
666 result: []u8,
667 result_alignment: std.mem.Alignment,
668 context: []const u8,
669 context_alignment: std.mem.Alignment,
670 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
671) ?*Io.AnyFuture {
672 return concurrent(userdata, result.len, result_alignment, context, context_alignment, start) catch {
673 start(context.ptr, result.ptr);
674 return null;
675 };
676}
677
678fn concurrent(
679 userdata: ?*anyopaque,
680 result_len: usize,
681 result_alignment: Alignment,
682 context: []const u8,
683 context_alignment: Alignment,
684 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
685) Io.ConcurrentError!*Io.AnyFuture {
686 const k: *Kqueue = @ptrCast(@alignCast(userdata));
687 assert(result_alignment.compare(.lte, Fiber.max_result_align)); // TODO
688 assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO
689 assert(result_len <= Fiber.max_result_size); // TODO
690 assert(context.len <= Fiber.max_context_size); // TODO
691
692 const fiber = Fiber.allocate(k) catch return error.ConcurrencyUnavailable;
693 std.log.debug("allocated {*}", .{fiber});
694
695 const closure: *AsyncClosure = .fromFiber(fiber);
696 fiber.* = .{
697 .required_align = {},
698 .context = switch (builtin.cpu.arch) {
699 .x86_64 => .{
700 .rsp = @intFromPtr(closure) - @sizeOf(usize),
701 .rbp = 0,
702 .rip = @intFromPtr(&fiberEntry),
703 },
704 .aarch64 => .{
705 .sp = @intFromPtr(closure),
706 .fp = 0,
707 .pc = @intFromPtr(&fiberEntry),
708 },
709 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
710 },
711 .awaiter = null,
712 .queue_next = null,
713 .cancel_thread = null,
714 .awaiting_completions = .empty,
715 };
716 closure.* = .{
717 .kqueue = k,
718 .fiber = fiber,
719 .start = start,
720 .result_align = result_alignment,
721 .already_awaited = false,
722 };
723 @memcpy(closure.contextPointer(), context);
724
725 k.schedule(.current(), .{ .head = fiber, .tail = fiber });
726 return @ptrCast(fiber);
727}
728
729fn await(
730 userdata: ?*anyopaque,
731 any_future: *Io.AnyFuture,
732 result: []u8,
733 result_alignment: std.mem.Alignment,
734) void {
735 const k: *Kqueue = @ptrCast(@alignCast(userdata));
736 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
737 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished)
738 k.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
739 @memcpy(result, future_fiber.resultBytes(result_alignment));
740 k.recycle(future_fiber);
741}
742
743fn cancel(
744 userdata: ?*anyopaque,
745 any_future: *Io.AnyFuture,
746 result: []u8,
747 result_alignment: std.mem.Alignment,
748) void {
749 const k: *Kqueue = @ptrCast(@alignCast(userdata));
750 _ = k;
751 _ = any_future;
752 _ = result;
753 _ = result_alignment;
754 @panic("TODO");
755}
756
757fn cancelRequested(userdata: ?*anyopaque) bool {
758 const k: *Kqueue = @ptrCast(@alignCast(userdata));
759 _ = k;
760 return false; // TODO
761}
762
763fn groupAsync(
764 userdata: ?*anyopaque,
765 type_erased: *Io.Group,
766 context: []const u8,
767 context_alignment: Alignment,
768 start: *const fn (context: *const anyopaque) void,
769) void {
770 const k: *Kqueue = @ptrCast(@alignCast(userdata));
771 _ = k;
772 _ = type_erased;
773 _ = context;
774 _ = context_alignment;
775 _ = start;
776 @panic("TODO");
777}
778
779fn groupConcurrent(
780 userdata: ?*anyopaque,
781 type_erased: *Io.Group,
782 context: []const u8,
783 context_alignment: Alignment,
784 start: *const fn (context: *const anyopaque) void,
785) Io.ConcurrentError!void {
786 const k: *Kqueue = @ptrCast(@alignCast(userdata));
787 _ = k;
788 _ = type_erased;
789 _ = context;
790 _ = context_alignment;
791 _ = start;
792 @panic("TODO");
793}
794
795fn groupAwait(userdata: ?*anyopaque, type_erased: *Io.Group, initial_token: *anyopaque) Io.Cancelable!void {
796 const k: *Kqueue = @ptrCast(@alignCast(userdata));
797 _ = k;
798 _ = type_erased;
799 _ = initial_token;
800 @panic("TODO");
801}
802
803fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
804 const k: *Kqueue = @ptrCast(@alignCast(userdata));
805 _ = k;
806 _ = group;
807 _ = token;
808 @panic("TODO");
809}
810
811fn dirCreateDir(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, permissions: Dir.Permissions) Dir.CreateDirError!void {
812 const k: *Kqueue = @ptrCast(@alignCast(userdata));
813 _ = k;
814 _ = dir;
815 _ = sub_path;
816 _ = permissions;
817 @panic("TODO");
818}
819
820fn dirCreateDirPath(
821 userdata: ?*anyopaque,
822 dir: Dir,
823 sub_path: []const u8,
824 permissions: Dir.Permissions,
825) Dir.CreateDirPathError!Dir.CreatePathStatus {
826 const k: *Kqueue = @ptrCast(@alignCast(userdata));
827 _ = k;
828 _ = dir;
829 _ = sub_path;
830 _ = permissions;
831 @panic("TODO");
832}
833
834fn dirCreateDirPathOpen(
835 userdata: ?*anyopaque,
836 dir: Dir,
837 sub_path: []const u8,
838 permissions: Dir.Permissions,
839 options: Dir.OpenOptions,
840) Dir.CreateDirPathOpenError!Dir {
841 const k: *Kqueue = @ptrCast(@alignCast(userdata));
842 _ = k;
843 _ = dir;
844 _ = sub_path;
845 _ = permissions;
846 _ = options;
847 @panic("TODO");
848}
849
850fn dirStat(userdata: ?*anyopaque, dir: Dir) Dir.StatError!Dir.Stat {
851 const k: *Kqueue = @ptrCast(@alignCast(userdata));
852 _ = k;
853 _ = dir;
854 @panic("TODO");
855}
856
857fn dirStatFile(
858 userdata: ?*anyopaque,
859 dir: Dir,
860 sub_path: []const u8,
861 options: Dir.StatFileOptions,
862) Dir.StatFileError!File.Stat {
863 const k: *Kqueue = @ptrCast(@alignCast(userdata));
864 _ = k;
865 _ = dir;
866 _ = sub_path;
867 _ = options;
868 @panic("TODO");
869}
870fn dirAccess(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.AccessOptions) Dir.AccessError!void {
871 const k: *Kqueue = @ptrCast(@alignCast(userdata));
872 _ = k;
873 _ = dir;
874 _ = sub_path;
875 _ = options;
876 @panic("TODO");
877}
878fn dirCreateFile(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, flags: Dir.CreateFileOptions) File.OpenError!File {
879 const k: *Kqueue = @ptrCast(@alignCast(userdata));
880 _ = k;
881 _ = dir;
882 _ = sub_path;
883 _ = flags;
884 @panic("TODO");
885}
886fn dirOpenFile(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, flags: Dir.OpenFileOptions) File.OpenError!File {
887 const k: *Kqueue = @ptrCast(@alignCast(userdata));
888 _ = k;
889 _ = dir;
890 _ = sub_path;
891 _ = flags;
892 @panic("TODO");
893}
894fn dirOpenDir(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.OpenOptions) Dir.OpenError!Dir {
895 const k: *Kqueue = @ptrCast(@alignCast(userdata));
896 _ = k;
897 _ = dir;
898 _ = sub_path;
899 _ = options;
900 @panic("TODO");
901}
902fn dirClose(userdata: ?*anyopaque, dirs: []const Dir) void {
903 const k: *Kqueue = @ptrCast(@alignCast(userdata));
904 _ = k;
905 _ = dirs;
906 @panic("TODO");
907}
908fn fileStat(userdata: ?*anyopaque, file: File) File.StatError!File.Stat {
909 const k: *Kqueue = @ptrCast(@alignCast(userdata));
910 _ = k;
911 _ = file;
912 @panic("TODO");
913}
914
915fn fileClose(userdata: ?*anyopaque, files: []const File) void {
916 const k: *Kqueue = @ptrCast(@alignCast(userdata));
917 _ = k;
918 _ = files;
919 @panic("TODO");
920}
921
922fn fileWriteStreaming(
923 userdata: ?*anyopaque,
924 file: File,
925 header: []const u8,
926 data: []const []const u8,
927 splat: usize,
928) File.Writer.Error!usize {
929 const k: *Kqueue = @ptrCast(@alignCast(userdata));
930 _ = k;
931 _ = file;
932 _ = header;
933 _ = data;
934 _ = splat;
935 @panic("TODO");
936}
937
938fn fileWritePositional(
939 userdata: ?*anyopaque,
940 file: File,
941 header: []const u8,
942 data: []const []const u8,
943 splat: usize,
944 offset: u64,
945) File.WritePositionalError!usize {
946 const k: *Kqueue = @ptrCast(@alignCast(userdata));
947 _ = k;
948 _ = file;
949 _ = header;
950 _ = data;
951 _ = splat;
952 _ = offset;
953 @panic("TODO");
954}
955
956fn fileReadStreaming(userdata: ?*anyopaque, file: File, data: []const []u8) File.Reader.Error!usize {
957 const k: *Kqueue = @ptrCast(@alignCast(userdata));
958 _ = k;
959 _ = file;
960 _ = data;
961 @panic("TODO");
962}
963
964fn fileReadPositional(userdata: ?*anyopaque, file: File, data: []const []u8, offset: u64) File.ReadPositionalError!usize {
965 const k: *Kqueue = @ptrCast(@alignCast(userdata));
966 _ = k;
967 _ = file;
968 _ = data;
969 _ = offset;
970 @panic("TODO");
971}
972fn fileSeekBy(userdata: ?*anyopaque, file: File, relative_offset: i64) File.SeekError!void {
973 const k: *Kqueue = @ptrCast(@alignCast(userdata));
974 _ = k;
975 _ = file;
976 _ = relative_offset;
977 @panic("TODO");
978}
979fn fileSeekTo(userdata: ?*anyopaque, file: File, absolute_offset: u64) File.SeekError!void {
980 const k: *Kqueue = @ptrCast(@alignCast(userdata));
981 _ = k;
982 _ = file;
983 _ = absolute_offset;
984 @panic("TODO");
985}
986
987fn now(userdata: ?*anyopaque, clock: Io.Clock) Io.Clock.Error!Io.Timestamp {
988 const k: *Kqueue = @ptrCast(@alignCast(userdata));
989 _ = k;
990 _ = clock;
991 @panic("TODO");
992}
993fn sleep(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void {
994 const k: *Kqueue = @ptrCast(@alignCast(userdata));
995 _ = k;
996 _ = timeout;
997 @panic("TODO");
998}
999
1000fn netListenIp(
1001 userdata: ?*anyopaque,
1002 address: *const net.IpAddress,
1003 options: net.IpAddress.ListenOptions,
1004) net.IpAddress.ListenError!net.Socket {
1005 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1006 _ = k;
1007 _ = address;
1008 _ = options;
1009 @panic("TODO");
1010}
1011fn netAccept(userdata: ?*anyopaque, server: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
1012 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1013 _ = k;
1014 _ = server;
1015 _ = options;
1016 @panic("TODO");
1017}
1018fn netBindIp(
1019 userdata: ?*anyopaque,
1020 address: *const net.IpAddress,
1021 options: net.IpAddress.BindOptions,
1022) net.IpAddress.BindError!net.Socket {
1023 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1024 const family = Io.Threaded.posixAddressFamily(address);
1025 const socket_fd = try openSocketPosix(k, family, options);
1026 errdefer closeFd(socket_fd);
1027 var storage: Io.Threaded.PosixAddress = undefined;
1028 var addr_len = Io.Threaded.addressToPosix(address, &storage);
1029 try posixBind(k, socket_fd, &storage.any, addr_len);
1030 if (options.allow_broadcast) try setSocketOption(k, socket_fd, posix.SOL.SOCKET, posix.SO.BROADCAST, 1);
1031 try posixGetSockName(k, socket_fd, &storage.any, &addr_len);
1032 return .{ .handle = socket_fd, .address = Io.Threaded.addressFromPosix(&storage) };
1033}
1034fn netConnectIp(userdata: ?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ConnectOptions) net.IpAddress.ConnectError!net.Socket {
1035 if (options.timeout != .none) @panic("TODO");
1036 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1037 const family = Io.Threaded.posixAddressFamily(address);
1038 const socket_fd = try openSocketPosix(k, family, .{
1039 .mode = options.mode,
1040 .protocol = options.protocol,
1041 });
1042 errdefer closeFd(socket_fd);
1043 var storage: Io.Threaded.PosixAddress = undefined;
1044 var addr_len = Io.Threaded.addressToPosix(address, &storage);
1045 try posixConnect(k, socket_fd, &storage.any, addr_len);
1046 try posixGetSockName(k, socket_fd, &storage.any, &addr_len);
1047 return .{ .handle = socket_fd, .address = Io.Threaded.addressFromPosix(&storage) };
1048}
1049
1050fn posixConnect(k: *Kqueue, socket_fd: posix.socket_t, addr: *const posix.sockaddr, addr_len: posix.socklen_t) !void {
1051 while (true) {
1052 try k.checkCancel();
1053 switch (posix.errno(posix.system.connect(socket_fd, addr, addr_len))) {
1054 .SUCCESS => return,
1055 .INTR => continue,
1056 .CANCELED => return error.Canceled,
1057 .AGAIN => @panic("TODO"),
1058 .INPROGRESS => return, // Due to TCP fast open, we find out possible error later.
1059
1060 .ADDRNOTAVAIL => return error.AddressUnavailable,
1061 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1062 .ALREADY => return error.ConnectionPending,
1063 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1064 .CONNREFUSED => return error.ConnectionRefused,
1065 .CONNRESET => return error.ConnectionResetByPeer,
1066 .FAULT => |err| return errnoBug(err),
1067 .ISCONN => |err| return errnoBug(err),
1068 .HOSTUNREACH => return error.HostUnreachable,
1069 .NETUNREACH => return error.NetworkUnreachable,
1070 .NOTSOCK => |err| return errnoBug(err),
1071 .PROTOTYPE => |err| return errnoBug(err),
1072 .TIMEDOUT => return error.Timeout,
1073 .CONNABORTED => |err| return errnoBug(err),
1074 .ACCES => return error.AccessDenied,
1075 .PERM => |err| return errnoBug(err),
1076 .NOENT => |err| return errnoBug(err),
1077 .NETDOWN => return error.NetworkDown,
1078 else => |err| return posix.unexpectedErrno(err),
1079 }
1080 }
1081}
1082
1083fn netListenUnix(
1084 userdata: ?*anyopaque,
1085 unix_address: *const net.UnixAddress,
1086 options: net.UnixAddress.ListenOptions,
1087) net.UnixAddress.ListenError!net.Socket.Handle {
1088 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1089 _ = k;
1090 _ = unix_address;
1091 _ = options;
1092 @panic("TODO");
1093}
1094fn netConnectUnix(
1095 userdata: ?*anyopaque,
1096 unix_address: *const net.UnixAddress,
1097) net.UnixAddress.ConnectError!net.Socket.Handle {
1098 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1099 _ = k;
1100 _ = unix_address;
1101 @panic("TODO");
1102}
1103
1104fn netSend(
1105 userdata: ?*anyopaque,
1106 handle: net.Socket.Handle,
1107 outgoing_messages: []net.OutgoingMessage,
1108 flags: net.SendFlags,
1109) struct { ?net.Socket.SendError, usize } {
1110 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1111
1112 const posix_flags: u32 =
1113 @as(u32, if (@hasDecl(posix.MSG, "CONFIRM") and flags.confirm) posix.MSG.CONFIRM else 0) |
1114 @as(u32, if (@hasDecl(posix.MSG, "DONTROUTE") and flags.dont_route) posix.MSG.DONTROUTE else 0) |
1115 @as(u32, if (@hasDecl(posix.MSG, "EOR") and flags.eor) posix.MSG.EOR else 0) |
1116 @as(u32, if (@hasDecl(posix.MSG, "OOB") and flags.oob) posix.MSG.OOB else 0) |
1117 @as(u32, if (@hasDecl(posix.MSG, "FASTOPEN") and flags.fastopen) posix.MSG.FASTOPEN else 0) |
1118 posix.MSG.NOSIGNAL;
1119
1120 for (outgoing_messages, 0..) |*msg, i| {
1121 netSendOne(k, handle, msg, posix_flags) catch |err| return .{ err, i };
1122 }
1123
1124 return .{ null, outgoing_messages.len };
1125}
1126
1127fn netSendOne(
1128 k: *Kqueue,
1129 handle: net.Socket.Handle,
1130 message: *net.OutgoingMessage,
1131 flags: u32,
1132) net.Socket.SendError!void {
1133 var addr: Io.Threaded.PosixAddress = undefined;
1134 var iovec: posix.iovec_const = .{ .base = @constCast(message.data_ptr), .len = message.data_len };
1135 const msg: posix.msghdr_const = .{
1136 .name = &addr.any,
1137 .namelen = Io.Threaded.addressToPosix(message.address, &addr),
1138 .iov = (&iovec)[0..1],
1139 .iovlen = 1,
1140 // OS returns EINVAL if this pointer is invalid even if controllen is zero.
1141 .control = if (message.control.len == 0) null else @constCast(message.control.ptr),
1142 .controllen = @intCast(message.control.len),
1143 .flags = 0,
1144 };
1145 while (true) {
1146 try k.checkCancel();
1147 const rc = posix.system.sendmsg(handle, &msg, flags);
1148 switch (posix.errno(rc)) {
1149 .SUCCESS => {
1150 message.data_len = @intCast(rc);
1151 return;
1152 },
1153 .INTR => continue,
1154 .CANCELED => return error.Canceled,
1155 .AGAIN => @panic("TODO register kevent"),
1156
1157 .ACCES => return error.AccessDenied,
1158 .ALREADY => return error.FastOpenAlreadyInProgress,
1159 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1160 .CONNRESET => return error.ConnectionResetByPeer,
1161 .DESTADDRREQ => |err| return errnoBug(err),
1162 .FAULT => |err| return errnoBug(err),
1163 .INVAL => |err| return errnoBug(err),
1164 .ISCONN => |err| return errnoBug(err),
1165 .MSGSIZE => return error.MessageOversize,
1166 .NOBUFS => return error.SystemResources,
1167 .NOMEM => return error.SystemResources,
1168 .NOTSOCK => |err| return errnoBug(err),
1169 .OPNOTSUPP => |err| return errnoBug(err),
1170 .PIPE => return error.SocketUnconnected,
1171 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1172 .HOSTUNREACH => return error.HostUnreachable,
1173 .NETUNREACH => return error.NetworkUnreachable,
1174 .NOTCONN => return error.SocketUnconnected,
1175 .NETDOWN => return error.NetworkDown,
1176 else => |err| return posix.unexpectedErrno(err),
1177 }
1178 }
1179}
1180
1181fn netReceive(
1182 userdata: ?*anyopaque,
1183 handle: net.Socket.Handle,
1184 message_buffer: []net.IncomingMessage,
1185 data_buffer: []u8,
1186 flags: net.ReceiveFlags,
1187 timeout: Io.Timeout,
1188) struct { ?net.Socket.ReceiveTimeoutError, usize } {
1189 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1190 _ = k;
1191 _ = handle;
1192 _ = message_buffer;
1193 _ = data_buffer;
1194 _ = flags;
1195 _ = timeout;
1196 @panic("TODO");
1197}
1198
1199fn netRead(userdata: ?*anyopaque, fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize {
1200 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1201
1202 var iovecs_buffer: [max_iovecs_len]posix.iovec = undefined;
1203 var i: usize = 0;
1204 for (data) |buf| {
1205 if (iovecs_buffer.len - i == 0) break;
1206 if (buf.len != 0) {
1207 iovecs_buffer[i] = .{ .base = buf.ptr, .len = buf.len };
1208 i += 1;
1209 }
1210 }
1211 const dest = iovecs_buffer[0..i];
1212 assert(dest[0].len > 0);
1213
1214 while (true) {
1215 try k.checkCancel();
1216 const rc = posix.system.readv(fd, dest.ptr, @intCast(dest.len));
1217 switch (posix.errno(rc)) {
1218 .SUCCESS => return @intCast(rc),
1219 .INTR => continue,
1220 .CANCELED => return error.Canceled,
1221 .AGAIN => {
1222 const thread: *Thread = .current();
1223 const fiber = thread.currentFiber();
1224 const ident: u32 = @bitCast(fd);
1225 const filter = std.c.EVFILT.READ;
1226 const gop = thread.wait_queues.getOrPut(k.gpa, .{
1227 .ident = ident,
1228 .filter = filter,
1229 }) catch return error.SystemResources;
1230 if (gop.found_existing) {
1231 const tail_fiber = gop.value_ptr.*;
1232 assert(tail_fiber.queue_next == null);
1233 tail_fiber.queue_next = fiber;
1234 gop.value_ptr.* = fiber;
1235 } else {
1236 gop.value_ptr.* = fiber;
1237 const changes = [_]posix.Kevent{
1238 .{
1239 .ident = ident,
1240 .filter = filter,
1241 .flags = std.c.EV.ADD | std.c.EV.ONESHOT,
1242 .fflags = 0,
1243 .data = 0,
1244 .udata = @intFromPtr(fiber),
1245 },
1246 };
1247 assert(0 == (kevent(thread.kq_fd, &changes, &.{}, null) catch |err| {
1248 // TODO handle EINTR for cancellation purposes
1249 @panic(@errorName(err)); // TODO
1250 }));
1251 }
1252 yield(k, null, .nothing);
1253 continue;
1254 },
1255
1256 .INVAL => |err| return errnoBug(err),
1257 .FAULT => |err| return errnoBug(err),
1258 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1259 .NOBUFS => return error.SystemResources,
1260 .NOMEM => return error.SystemResources,
1261 .NOTCONN => return error.SocketUnconnected,
1262 .CONNRESET => return error.ConnectionResetByPeer,
1263 .TIMEDOUT => return error.Timeout,
1264 .PIPE => return error.SocketUnconnected,
1265 .NETDOWN => return error.NetworkDown,
1266 else => |err| return posix.unexpectedErrno(err),
1267 }
1268 }
1269}
1270
1271fn netShutdown(userdata: ?*anyopaque, handle: net.Socket.Handle, how: net.ShutdownHow) net.ShutdownError!void {
1272 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1273 _ = k;
1274 _ = handle;
1275 _ = how;
1276 @panic("TODO");
1277}
1278
1279fn netInterfaceNameResolve(
1280 userdata: ?*anyopaque,
1281 name: *const net.Interface.Name,
1282) net.Interface.Name.ResolveError!net.Interface {
1283 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1284 _ = k;
1285 _ = name;
1286 @panic("TODO");
1287}
1288
1289fn netInterfaceName(userdata: ?*anyopaque, interface: net.Interface) net.Interface.NameError!net.Interface.Name {
1290 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1291 _ = k;
1292 _ = interface;
1293 @panic("TODO");
1294}
1295
1296fn netLookup(
1297 userdata: ?*anyopaque,
1298 host_name: net.HostName,
1299 resolved: *Io.Queue(net.HostName.LookupResult),
1300 options: net.HostName.LookupOptions,
1301) net.HostName.LookupError!void {
1302 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1303 _ = k;
1304 _ = host_name;
1305 _ = resolved;
1306 _ = options;
1307 @panic("TODO");
1308}
1309
1310fn openSocketPosix(
1311 k: *Kqueue,
1312 family: posix.sa_family_t,
1313 options: IpAddress.BindOptions,
1314) error{
1315 AddressFamilyUnsupported,
1316 ProtocolUnsupportedBySystem,
1317 ProcessFdQuotaExceeded,
1318 SystemFdQuotaExceeded,
1319 SystemResources,
1320 ProtocolUnsupportedByAddressFamily,
1321 SocketModeUnsupported,
1322 OptionUnsupported,
1323 Unexpected,
1324 Canceled,
1325}!posix.socket_t {
1326 const mode, const protocol = try posixSocketModeProtocol(family, options.mode, options.protocol);
1327 const socket_fd = while (true) {
1328 try k.checkCancel();
1329 const flags: u32 = mode | if (Io.Threaded.socket_flags_unsupported) 0 else posix.SOCK.CLOEXEC;
1330 const socket_rc = posix.system.socket(family, flags, protocol);
1331 switch (posix.errno(socket_rc)) {
1332 .SUCCESS => {
1333 const fd: posix.fd_t = @intCast(socket_rc);
1334 errdefer closeFd(fd);
1335 if (Io.Threaded.socket_flags_unsupported) {
1336 while (true) {
1337 try k.checkCancel();
1338 switch (posix.errno(posix.system.fcntl(fd, posix.F.SETFD, @as(usize, posix.FD_CLOEXEC)))) {
1339 .SUCCESS => break,
1340 .INTR => continue,
1341 .CANCELED => return error.Canceled,
1342 else => |err| return posix.unexpectedErrno(err),
1343 }
1344 }
1345
1346 var fl_flags: usize = while (true) {
1347 try k.checkCancel();
1348 const rc = posix.system.fcntl(fd, posix.F.GETFL, @as(usize, 0));
1349 switch (posix.errno(rc)) {
1350 .SUCCESS => break @intCast(rc),
1351 .INTR => continue,
1352 .CANCELED => return error.Canceled,
1353 else => |err| return posix.unexpectedErrno(err),
1354 }
1355 };
1356 fl_flags |= @as(usize, 1 << @bitOffsetOf(posix.O, "NONBLOCK"));
1357 while (true) {
1358 try k.checkCancel();
1359 switch (posix.errno(posix.system.fcntl(fd, posix.F.SETFL, fl_flags))) {
1360 .SUCCESS => break,
1361 .INTR => continue,
1362 .CANCELED => return error.Canceled,
1363 else => |err| return posix.unexpectedErrno(err),
1364 }
1365 }
1366 }
1367 break fd;
1368 },
1369 .INTR => continue,
1370 .CANCELED => return error.Canceled,
1371
1372 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1373 .INVAL => return error.ProtocolUnsupportedBySystem,
1374 .MFILE => return error.ProcessFdQuotaExceeded,
1375 .NFILE => return error.SystemFdQuotaExceeded,
1376 .NOBUFS => return error.SystemResources,
1377 .NOMEM => return error.SystemResources,
1378 .PROTONOSUPPORT => return error.ProtocolUnsupportedByAddressFamily,
1379 .PROTOTYPE => return error.SocketModeUnsupported,
1380 else => |err| return posix.unexpectedErrno(err),
1381 }
1382 };
1383 errdefer closeFd(socket_fd);
1384
1385 if (options.ip6_only) |ip6_only| {
1386 if (posix.IPV6 == void) return error.OptionUnsupported;
1387 try setSocketOption(k, socket_fd, posix.IPPROTO.IPV6, posix.IPV6.V6ONLY, @intFromBool(ip6_only));
1388 }
1389
1390 return socket_fd;
1391}
1392
1393fn posixBind(
1394 k: *Kqueue,
1395 socket_fd: posix.socket_t,
1396 addr: *const posix.sockaddr,
1397 addr_len: posix.socklen_t,
1398) !void {
1399 while (true) {
1400 try k.checkCancel();
1401 switch (posix.errno(posix.system.bind(socket_fd, addr, addr_len))) {
1402 .SUCCESS => break,
1403 .INTR => continue,
1404 .CANCELED => return error.Canceled,
1405
1406 .ACCES => return error.AccessDenied,
1407 .ADDRINUSE => return error.AddressInUse,
1408 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1409 .INVAL => |err| return errnoBug(err), // invalid parameters
1410 .NOTSOCK => |err| return errnoBug(err), // invalid `sockfd`
1411 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1412 .ADDRNOTAVAIL => return error.AddressUnavailable,
1413 .FAULT => |err| return errnoBug(err), // invalid `addr` pointer
1414 .NOMEM => return error.SystemResources,
1415 else => |err| return posix.unexpectedErrno(err),
1416 }
1417 }
1418}
1419
1420fn posixGetSockName(k: *Kqueue, socket_fd: posix.fd_t, addr: *posix.sockaddr, addr_len: *posix.socklen_t) !void {
1421 while (true) {
1422 try k.checkCancel();
1423 switch (posix.errno(posix.system.getsockname(socket_fd, addr, addr_len))) {
1424 .SUCCESS => break,
1425 .INTR => continue,
1426 .CANCELED => return error.Canceled,
1427
1428 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1429 .FAULT => |err| return errnoBug(err),
1430 .INVAL => |err| return errnoBug(err), // invalid parameters
1431 .NOTSOCK => |err| return errnoBug(err), // always a race condition
1432 .NOBUFS => return error.SystemResources,
1433 else => |err| return posix.unexpectedErrno(err),
1434 }
1435 }
1436}
1437
1438fn setSocketOption(k: *Kqueue, fd: posix.fd_t, level: i32, opt_name: u32, option: u32) !void {
1439 const o: []const u8 = @ptrCast(&option);
1440 while (true) {
1441 try k.checkCancel();
1442 switch (posix.errno(posix.system.setsockopt(fd, level, opt_name, o.ptr, @intCast(o.len)))) {
1443 .SUCCESS => return,
1444 .INTR => continue,
1445 .CANCELED => return error.Canceled,
1446
1447 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1448 .NOTSOCK => |err| return errnoBug(err),
1449 .INVAL => |err| return errnoBug(err),
1450 .FAULT => |err| return errnoBug(err),
1451 else => |err| return posix.unexpectedErrno(err),
1452 }
1453 }
1454}
1455
1456fn checkCancel(k: *Kqueue) error{Canceled}!void {
1457 if (cancelRequested(k)) return error.Canceled;
1458}
1459
1460pub const KEventError = error{
1461 /// The process does not have permission to register a filter.
1462 AccessDenied,
1463 /// The event could not be found to be modified or deleted.
1464 EventNotFound,
1465 /// No memory was available to register the event.
1466 SystemResources,
1467 /// The specified process to attach to does not exist.
1468 ProcessNotFound,
1469 /// changelist or eventlist had too many items on it.
1470 /// TODO remove this possibility
1471 Overflow,
1472};
1473
1474pub fn kevent(
1475 kq: i32,
1476 changelist: []const posix.Kevent,
1477 eventlist: []posix.Kevent,
1478 timeout: ?*const posix.timespec,
1479) KEventError!usize {
1480 while (true) {
1481 const rc = posix.system.kevent(
1482 kq,
1483 changelist.ptr,
1484 std.math.cast(c_int, changelist.len) orelse return error.Overflow,
1485 eventlist.ptr,
1486 std.math.cast(c_int, eventlist.len) orelse return error.Overflow,
1487 timeout,
1488 );
1489 switch (posix.errno(rc)) {
1490 .SUCCESS => return @intCast(rc),
1491 .ACCES => return error.AccessDenied,
1492 .FAULT => unreachable, // TODO use error.Unexpected for these
1493 .BADF => unreachable, // Always a race condition.
1494 .INTR => continue, // TODO handle cancelation
1495 .INVAL => unreachable,
1496 .NOENT => return error.EventNotFound,
1497 .NOMEM => return error.SystemResources,
1498 .SRCH => return error.ProcessNotFound,
1499 else => unreachable,
1500 }
1501 }
1502}