authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-04-03 17:54:37-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:39-07:00
log0f67ea4fa45ffd62a81d906447ca821488ef0598
tree4f76b16d5d9e54f18e100b0c8b8a7aba512b9e4e
parent7b5886118dabb59967d3e9b17d0502146df2ef92

introduce Io.select and implement it in thread pool

2 files changed, 228 insertions(+), 78 deletions(-)

lib/std/Io.zig+155-60
...@@ -979,17 +979,21 @@ pub const VTable = struct {...@@ -979,17 +979,21 @@ pub const VTable = struct {
979 /// Thread-safe.979 /// Thread-safe.
980 cancelRequested: *const fn (?*anyopaque) bool,980 cancelRequested: *const fn (?*anyopaque) bool,
981981
982 /// Blocks until one of the futures from the list has a result ready, such
983 /// that awaiting it will not block. Returns that index.
984 select: *const fn (?*anyopaque, futures: []const *AnyFuture) usize,
985
982 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,986 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,
983 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,987 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
984988
985 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,989 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,
986 conditionWake: *const fn (?*anyopaque, cond: *Condition, wake: Condition.Wake) void,990 conditionWake: *const fn (?*anyopaque, cond: *Condition, wake: Condition.Wake) void,
987991
988 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,992 createFile: *const fn (?*anyopaque, dir: Dir, sub_path: []const u8, flags: File.CreateFlags) File.OpenError!File,
989 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,993 openFile: *const fn (?*anyopaque, dir: Dir, sub_path: []const u8, flags: File.OpenFlags) File.OpenError!File,
990 closeFile: *const fn (?*anyopaque, fs.File) void,994 closeFile: *const fn (?*anyopaque, File) void,
991 pread: *const fn (?*anyopaque, file: fs.File, buffer: []u8, offset: std.posix.off_t) FilePReadError!usize,995 pread: *const fn (?*anyopaque, file: File, buffer: []u8, offset: std.posix.off_t) File.PReadError!usize,
992 pwrite: *const fn (?*anyopaque, file: fs.File, buffer: []const u8, offset: std.posix.off_t) FilePWriteError!usize,996 pwrite: *const fn (?*anyopaque, file: File, buffer: []const u8, offset: std.posix.off_t) File.PWriteError!usize,
993997
994 now: *const fn (?*anyopaque, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp,998 now: *const fn (?*anyopaque, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp,
995 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,999 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,
...@@ -1000,28 +1004,118 @@ pub const Cancelable = error{...@@ -1000,28 +1004,118 @@ pub const Cancelable = error{
1000 Canceled,1004 Canceled,
1001};1005};
10021006
1003pub const OpenFlags = fs.File.OpenFlags;1007pub const Dir = struct {
1004pub const CreateFlags = fs.File.CreateFlags;1008 handle: Handle,
1009
1010 pub fn cwd() Dir {
1011 return .{ .handle = std.fs.cwd().fd };
1012 }
1013
1014 pub const Handle = std.posix.fd_t;
1015
1016 pub fn openFile(dir: Dir, io: Io, sub_path: []const u8, flags: File.OpenFlags) File.OpenError!File {
1017 return io.vtable.openFile(io.userdata, dir, sub_path, flags);
1018 }
1019
1020 pub fn createFile(dir: Dir, io: Io, sub_path: []const u8, flags: File.CreateFlags) File.OpenError!File {
1021 return io.vtable.createFile(io.userdata, dir, sub_path, flags);
1022 }
1023
1024 pub const WriteFileOptions = struct {
1025 /// On Windows, `sub_path` should be encoded as [WTF-8](https://simonsapin.github.io/wtf-8/).
1026 /// On WASI, `sub_path` should be encoded as valid UTF-8.
1027 /// On other platforms, `sub_path` is an opaque sequence of bytes with no particular encoding.
1028 sub_path: []const u8,
1029 data: []const u8,
1030 flags: File.CreateFlags = .{},
1031 };
1032
1033 pub const WriteFileError = File.WriteError || File.OpenError || Cancelable;
1034
1035 /// Writes content to the file system, using the file creation flags provided.
1036 pub fn writeFile(dir: Dir, io: Io, options: WriteFileOptions) WriteFileError!void {
1037 var file = try dir.createFile(io, options.sub_path, options.flags);
1038 defer file.close(io);
1039 try file.writeAll(io, options.data);
1040 }
1041};
1042
1043pub const File = struct {
1044 handle: Handle,
1045
1046 pub const Handle = std.posix.fd_t;
10051047
1006pub const FileOpenError = fs.File.OpenError || Cancelable;1048 pub const OpenFlags = fs.File.OpenFlags;
1007pub const FileReadError = fs.File.ReadError || Cancelable;1049 pub const CreateFlags = fs.File.CreateFlags;
1008pub const FilePReadError = fs.File.PReadError || Cancelable;1050
1009pub const FileWriteError = fs.File.WriteError || Cancelable;1051 pub const OpenError = fs.File.OpenError || Cancelable;
1010pub const FilePWriteError = fs.File.PWriteError || Cancelable;1052
1053 pub fn close(file: File, io: Io) void {
1054 return io.vtable.closeFile(io.userdata, file);
1055 }
1056
1057 pub const ReadError = fs.File.ReadError || Cancelable;
1058
1059 pub fn read(file: File, io: Io, buffer: []u8) ReadError!usize {
1060 return @errorCast(file.pread(io, buffer, -1));
1061 }
1062
1063 pub const PReadError = fs.File.PReadError || Cancelable;
1064
1065 pub fn pread(file: File, io: Io, buffer: []u8, offset: std.posix.off_t) PReadError!usize {
1066 return io.vtable.pread(io.userdata, file, buffer, offset);
1067 }
1068
1069 pub const WriteError = fs.File.WriteError || Cancelable;
1070
1071 pub fn write(file: File, io: Io, buffer: []const u8) WriteError!usize {
1072 return @errorCast(file.pwrite(io, buffer, -1));
1073 }
1074
1075 pub const PWriteError = fs.File.PWriteError || Cancelable;
1076
1077 pub fn pwrite(file: File, io: Io, buffer: []const u8, offset: std.posix.off_t) PWriteError!usize {
1078 return io.vtable.pwrite(io.userdata, file, buffer, offset);
1079 }
1080
1081 pub fn writeAll(file: File, io: Io, bytes: []const u8) WriteError!void {
1082 var index: usize = 0;
1083 while (index < bytes.len) {
1084 index += try file.write(io, bytes[index..]);
1085 }
1086 }
1087
1088 pub fn readAll(file: File, io: Io, buffer: []u8) ReadError!usize {
1089 var index: usize = 0;
1090 while (index != buffer.len) {
1091 const amt = try file.read(io, buffer[index..]);
1092 if (amt == 0) break;
1093 index += amt;
1094 }
1095 return index;
1096 }
1097};
10111098
1012pub const Timestamp = enum(i96) {1099pub const Timestamp = enum(i96) {
1013 _,1100 _,
10141101
1015 pub fn durationTo(from: Timestamp, to: Timestamp) i96 {1102 pub fn durationTo(from: Timestamp, to: Timestamp) Duration {
1016 return @intFromEnum(to) - @intFromEnum(from);1103 return .{ .nanoseconds = @intFromEnum(to) - @intFromEnum(from) };
1017 }1104 }
10181105
1019 pub fn addDuration(from: Timestamp, duration: i96) Timestamp {1106 pub fn addDuration(from: Timestamp, duration: Duration) Timestamp {
1020 return @enumFromInt(@intFromEnum(from) + duration);1107 return @enumFromInt(@intFromEnum(from) + duration.nanoseconds);
1021 }1108 }
1022};1109};
1023pub const Deadline = union(enum) {1110pub const Duration = struct {
1024 nanoseconds: i96,1111 nanoseconds: i96,
1112
1113 pub fn ms(x: u64) Duration {
1114 return .{ .nanoseconds = @as(i96, x) * std.time.ns_per_ms };
1115 }
1116};
1117pub const Deadline = union(enum) {
1118 duration: Duration,
1025 timestamp: Timestamp,1119 timestamp: Timestamp,
1026};1120};
1027pub const ClockGetTimeError = std.posix.ClockGetTimeError || Cancelable;1121pub const ClockGetTimeError = std.posix.ClockGetTimeError || Cancelable;
...@@ -1408,7 +1502,7 @@ pub fn Queue(Elem: type) type {...@@ -1408,7 +1502,7 @@ pub fn Queue(Elem: type) type {
14081502
1409/// Calls `function` with `args`, such that the return value of the function is1503/// Calls `function` with `args`, such that the return value of the function is
1410/// not guaranteed to be available until `await` is called.1504/// not guaranteed to be available until `await` is called.
1411pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {1505pub fn async(io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {
1412 const Result = @typeInfo(@TypeOf(function)).@"fn".return_type.?;1506 const Result = @typeInfo(@TypeOf(function)).@"fn".return_type.?;
1413 const Args = @TypeOf(args);1507 const Args = @TypeOf(args);
1414 const TypeErased = struct {1508 const TypeErased = struct {
...@@ -1432,7 +1526,7 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(...@@ -1432,7 +1526,7 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(
14321526
1433/// Calls `function` with `args` asynchronously. The resource cleans itself up1527/// Calls `function` with `args` asynchronously. The resource cleans itself up
1434/// when the function returns. Does not support await, cancel, or a return value.1528/// when the function returns. Does not support await, cancel, or a return value.
1435pub fn go(io: Io, function: anytype, args: anytype) void {1529pub fn go(io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {
1436 const Args = @TypeOf(args);1530 const Args = @TypeOf(args);
1437 const TypeErased = struct {1531 const TypeErased = struct {
1438 fn start(context: *const anyopaque) void {1532 fn start(context: *const anyopaque) void {
...@@ -1448,55 +1542,56 @@ pub fn go(io: Io, function: anytype, args: anytype) void {...@@ -1448,55 +1542,56 @@ pub fn go(io: Io, function: anytype, args: anytype) void {
1448 );1542 );
1449}1543}
14501544
1451pub fn openFile(io: Io, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File {1545pub fn now(io: Io, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp {
1452 return io.vtable.openFile(io.userdata, dir, sub_path, flags);1546 return io.vtable.now(io.userdata, clockid);
1453}
1454
1455pub fn createFile(io: Io, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File {
1456 return io.vtable.createFile(io.userdata, dir, sub_path, flags);
1457}
1458
1459pub fn closeFile(io: Io, file: fs.File) void {
1460 return io.vtable.closeFile(io.userdata, file);
1461}
1462
1463pub fn read(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {
1464 return @errorCast(io.pread(file, buffer, -1));
1465}
1466
1467pub fn pread(io: Io, file: fs.File, buffer: []u8, offset: std.posix.off_t) FilePReadError!usize {
1468 return io.vtable.pread(io.userdata, file, buffer, offset);
1469}1547}
14701548
1471pub fn write(io: Io, file: fs.File, buffer: []const u8) FileWriteError!usize {1549pub fn sleep(io: Io, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void {
1472 return @errorCast(io.pwrite(file, buffer, -1));1550 return io.vtable.sleep(io.userdata, clockid, deadline);
1473}1551}
14741552
1475pub fn pwrite(io: Io, file: fs.File, buffer: []const u8, offset: std.posix.off_t) FilePWriteError!usize {1553pub fn sleepDuration(io: Io, duration: Duration) SleepError!void {
1476 return io.vtable.pwrite(io.userdata, file, buffer, offset);1554 return io.vtable.sleep(io.userdata, .MONOTONIC, .{ .duration = duration });
1477}1555}
14781556
1479pub fn writeAll(io: Io, file: fs.File, bytes: []const u8) FileWriteError!void {1557/// Given a struct with each field a `*Future`, returns a union with the same
1480 var index: usize = 0;1558/// fields, each field type the future's result.
1481 while (index < bytes.len) {1559pub fn SelectUnion(S: type) type {
1482 index += try io.write(file, bytes[index..]);1560 const struct_fields = @typeInfo(S).@"struct".fields;
1561 var fields: [struct_fields.len]std.builtin.Type.UnionField = undefined;
1562 for (&fields, struct_fields) |*union_field, struct_field| {
1563 const F = @typeInfo(struct_field.type).pointer.child;
1564 const Result = @TypeOf(@as(F, undefined).result);
1565 union_field.* = .{
1566 .name = struct_field.name,
1567 .type = Result,
1568 .alignment = struct_field.alignment,
1569 };
1483 }1570 }
1571 return @Type(.{ .@"union" = .{
1572 .layout = .auto,
1573 .tag_type = std.meta.FieldEnum(S),
1574 .fields = &fields,
1575 .decls = &.{},
1576 } });
1484}1577}
14851578
1486pub fn readAll(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {1579/// `s` is a struct with every field a `*Future(T)`, where `T` can be any type,
1487 var index: usize = 0;1580/// and can be different for each field.
1488 while (index != buffer.len) {1581pub fn select(io: Io, s: anytype) SelectUnion(@TypeOf(s)) {
1489 const amt = try io.read(file, buffer[index..]);1582 const U = SelectUnion(@TypeOf(s));
1490 if (amt == 0) break;1583 const S = @TypeOf(s);
1491 index += amt;1584 const fields = @typeInfo(S).@"struct".fields;
1585 var futures: [fields.len]*AnyFuture = undefined;
1586 inline for (fields, &futures) |field, *any_future| {
1587 const future = @field(s, field.name);
1588 any_future.* = future.any_future orelse return @unionInit(U, field.name, future.result);
1589 }
1590 switch (io.vtable.select(io.userdata, &futures)) {
1591 inline 0...(fields.len - 1) => |selected_index| {
1592 const field_name = fields[selected_index].name;
1593 return @unionInit(U, field_name, @field(s, field_name).await(io));
1594 },
1595 else => unreachable,
1492 }1596 }
1493 return index;
1494}
1495
1496pub fn now(io: Io, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp {
1497 return io.vtable.now(io.userdata, clockid);
1498}
1499
1500pub fn sleep(io: Io, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void {
1501 return io.vtable.sleep(io.userdata, clockid, deadline);
1502}1597}
lib/std/Thread/Pool.zig+73-18
...@@ -335,6 +335,7 @@ pub fn io(pool: *Pool) Io {...@@ -335,6 +335,7 @@ pub fn io(pool: *Pool) Io {
335 .go = go,335 .go = go,
336 .cancel = cancel,336 .cancel = cancel,
337 .cancelRequested = cancelRequested,337 .cancelRequested = cancelRequested,
338 .select = select,
338339
339 .mutexLock = mutexLock,340 .mutexLock = mutexLock,
340 .mutexUnlock = mutexUnlock,341 .mutexUnlock = mutexUnlock,
...@@ -358,10 +359,13 @@ const AsyncClosure = struct {...@@ -358,10 +359,13 @@ const AsyncClosure = struct {
358 func: *const fn (context: *anyopaque, result: *anyopaque) void,359 func: *const fn (context: *anyopaque, result: *anyopaque) void,
359 runnable: Runnable = .{ .runFn = runFn },360 runnable: Runnable = .{ .runFn = runFn },
360 reset_event: std.Thread.ResetEvent,361 reset_event: std.Thread.ResetEvent,
362 select_condition: ?*std.Thread.ResetEvent,
361 cancel_tid: std.Thread.Id,363 cancel_tid: std.Thread.Id,
362 context_offset: usize,364 context_offset: usize,
363 result_offset: usize,365 result_offset: usize,
364366
367 const done_reset_event: *std.Thread.ResetEvent = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(std.Thread.ResetEvent)));
368
365 const canceling_tid: std.Thread.Id = switch (@typeInfo(std.Thread.Id)) {369 const canceling_tid: std.Thread.Id = switch (@typeInfo(std.Thread.Id)) {
366 .int => |int_info| switch (int_info.signedness) {370 .int => |int_info| switch (int_info.signedness) {
367 .signed => -1,371 .signed => -1,
...@@ -396,6 +400,17 @@ const AsyncClosure = struct {...@@ -396,6 +400,17 @@ const AsyncClosure = struct {
396 .acq_rel,400 .acq_rel,
397 .acquire,401 .acquire,
398 )) |cancel_tid| assert(cancel_tid == canceling_tid);402 )) |cancel_tid| assert(cancel_tid == canceling_tid);
403
404 if (@atomicRmw(
405 ?*std.Thread.ResetEvent,
406 &closure.select_condition,
407 .Xchg,
408 done_reset_event,
409 .release,
410 )) |select_reset| {
411 assert(select_reset != done_reset_event);
412 select_reset.set();
413 }
399 closure.reset_event.set();414 closure.reset_event.set();
400 }415 }
401416
...@@ -455,6 +470,7 @@ fn @"async"(...@@ -455,6 +470,7 @@ fn @"async"(
455 .result_offset = result_offset,470 .result_offset = result_offset,
456 .reset_event = .{},471 .reset_event = .{},
457 .cancel_tid = 0,472 .cancel_tid = 0,
473 .select_condition = null,
458 };474 };
459 @memcpy(closure.contextPointer()[0..context.len], context);475 @memcpy(closure.contextPointer()[0..context.len], context);
460 pool.run_queue.prepend(&closure.runnable.node);476 pool.run_queue.prepend(&closure.runnable.node);
...@@ -720,47 +736,54 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, wake: Io.Condition....@@ -720,47 +736,54 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, wake: Io.Condition.
720736
721fn createFile(737fn createFile(
722 userdata: ?*anyopaque,738 userdata: ?*anyopaque,
723 dir: std.fs.Dir,739 dir: Io.Dir,
724 sub_path: []const u8,740 sub_path: []const u8,
725 flags: std.fs.File.CreateFlags,741 flags: Io.File.CreateFlags,
726) Io.FileOpenError!std.fs.File {742) Io.File.OpenError!Io.File {
727 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));743 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
728 try pool.checkCancel();744 try pool.checkCancel();
729 return dir.createFile(sub_path, flags);745 const fs_dir: std.fs.Dir = .{ .fd = dir.handle };
746 const fs_file = try fs_dir.createFile(sub_path, flags);
747 return .{ .handle = fs_file.handle };
730}748}
731749
732fn openFile(750fn openFile(
733 userdata: ?*anyopaque,751 userdata: ?*anyopaque,
734 dir: std.fs.Dir,752 dir: Io.Dir,
735 sub_path: []const u8,753 sub_path: []const u8,
736 flags: std.fs.File.OpenFlags,754 flags: Io.File.OpenFlags,
737) Io.FileOpenError!std.fs.File {755) Io.File.OpenError!Io.File {
738 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));756 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
739 try pool.checkCancel();757 try pool.checkCancel();
740 return dir.openFile(sub_path, flags);758 const fs_dir: std.fs.Dir = .{ .fd = dir.handle };
759 const fs_file = try fs_dir.openFile(sub_path, flags);
760 return .{ .handle = fs_file.handle };
741}761}
742762
743fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {763fn closeFile(userdata: ?*anyopaque, file: Io.File) void {
744 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));764 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
745 _ = pool;765 _ = pool;
746 return file.close();766 const fs_file: std.fs.File = .{ .handle = file.handle };
767 return fs_file.close();
747}768}
748769
749fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {770fn pread(userdata: ?*anyopaque, file: Io.File, buffer: []u8, offset: std.posix.off_t) Io.File.PReadError!usize {
750 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));771 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
751 try pool.checkCancel();772 try pool.checkCancel();
773 const fs_file: std.fs.File = .{ .handle = file.handle };
752 return switch (offset) {774 return switch (offset) {
753 -1 => file.read(buffer),775 -1 => fs_file.read(buffer),
754 else => file.pread(buffer, @bitCast(offset)),776 else => fs_file.pread(buffer, @bitCast(offset)),
755 };777 };
756}778}
757779
758fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {780fn pwrite(userdata: ?*anyopaque, file: Io.File, buffer: []const u8, offset: std.posix.off_t) Io.File.PWriteError!usize {
759 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));781 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
760 try pool.checkCancel();782 try pool.checkCancel();
783 const fs_file: std.fs.File = .{ .handle = file.handle };
761 return switch (offset) {784 return switch (offset) {
762 -1 => file.write(buffer),785 -1 => fs_file.write(buffer),
763 else => file.pwrite(buffer, @bitCast(offset)),786 else => fs_file.pwrite(buffer, @bitCast(offset)),
764 };787 };
765}788}
766789
...@@ -774,7 +797,7 @@ fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError...@@ -774,7 +797,7 @@ fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError
774fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {797fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
775 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));798 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
776 const deadline_nanoseconds: i96 = switch (deadline) {799 const deadline_nanoseconds: i96 = switch (deadline) {
777 .nanoseconds => |nanoseconds| nanoseconds,800 .duration => |duration| duration.nanoseconds,
778 .timestamp => |timestamp| @intFromEnum(timestamp),801 .timestamp => |timestamp| @intFromEnum(timestamp),
779 };802 };
780 var timespec: std.posix.timespec = .{803 var timespec: std.posix.timespec = .{
...@@ -784,7 +807,7 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl...@@ -784,7 +807,7 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl
784 while (true) {807 while (true) {
785 try pool.checkCancel();808 try pool.checkCancel();
786 switch (std.os.linux.E.init(std.os.linux.clock_nanosleep(clockid, .{ .ABSTIME = switch (deadline) {809 switch (std.os.linux.E.init(std.os.linux.clock_nanosleep(clockid, .{ .ABSTIME = switch (deadline) {
787 .nanoseconds => false,810 .duration => false,
788 .timestamp => true,811 .timestamp => true,
789 } }, &timespec, &timespec))) {812 } }, &timespec, &timespec))) {
790 .SUCCESS => return,813 .SUCCESS => return,
...@@ -795,3 +818,35 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl...@@ -795,3 +818,35 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl
795 }818 }
796 }819 }
797}820}
821
822fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
823 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
824 _ = pool;
825
826 var reset_event: std.Thread.ResetEvent = .{};
827
828 for (futures, 0..) |future, i| {
829 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
830 if (@atomicRmw(?*std.Thread.ResetEvent, &closure.select_condition, .Xchg, &reset_event, .seq_cst) == AsyncClosure.done_reset_event) {
831 for (futures[0..i]) |cleanup_future| {
832 const cleanup_closure: *AsyncClosure = @ptrCast(@alignCast(cleanup_future));
833 if (@atomicRmw(?*std.Thread.ResetEvent, &cleanup_closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
834 cleanup_closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.
835 }
836 }
837 return i;
838 }
839 }
840
841 reset_event.wait();
842
843 var result: ?usize = null;
844 for (futures, 0..) |future, i| {
845 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
846 if (@atomicRmw(?*std.Thread.ResetEvent, &closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
847 closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.
848 if (result == null) result = i; // In case multiple are ready, return first.
849 }
850 }
851 return result.?;
852}