authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-24 18:43:53-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:38-07:00
log5b2f54fc804cf097e673c8ba99cf4fb0a3b969ce
treedf053b75143f61da239f5a671532473ca237216e
parentf657767b600064c520fac0e72da6016508a8b601

introduce std.Io interface

which is planned to have all I/O operations in the interface, but for now has only async and await.

2 files changed, 72 insertions(+), 20 deletions(-)

lib/std/Io.zig+69
...@@ -911,3 +911,72 @@ test {...@@ -911,3 +911,72 @@ test {
911 _ = @import("Io/stream_source.zig");911 _ = @import("Io/stream_source.zig");
912 _ = @import("Io/test.zig");912 _ = @import("Io/test.zig");
913}913}
914
915const Io = @This();
916
917userdata: ?*anyopaque,
918vtable: *const VTable,
919
920pub const VTable = struct {
921 /// If it returns `null` it means `result` has been already populated and
922 /// `await` will be a no-op.
923 async: *const fn (
924 /// Corresponds to `Io.userdata`.
925 userdata: ?*anyopaque,
926 /// The pointer of this slice is an "eager" result value.
927 /// The length is the size in bytes of the result type.
928 eager_result: []u8,
929 /// Passed to `start`.
930 context: ?*anyopaque,
931 start: *const fn (context: ?*anyopaque, result: *anyopaque) void,
932 ) ?*AnyFuture,
933
934 /// This function is only called when `async` returns a non-null value.
935 await: *const fn (
936 /// Corresponds to `Io.userdata`.
937 userdata: ?*anyopaque,
938 /// The same value that was returned from `async`.
939 any_future: *AnyFuture,
940 /// Points to a buffer where the result is written.
941 /// The length is equal to size in bytes of result type.
942 result: []u8,
943 ) void,
944};
945
946pub const AnyFuture = opaque {};
947
948pub fn Future(Result: type) type {
949 return struct {
950 any_future: ?*AnyFuture,
951 result: Result,
952
953 pub fn await(f: *@This(), io: Io) Result {
954 const any_future = f.any_future orelse return f.result;
955 io.vtable.await(io.userdata, any_future, @ptrCast((&f.result)[0..1]));
956 f.any_future = null;
957 return f.result;
958 }
959 };
960}
961
962/// `s` is a struct instance that contains a function like this:
963/// ```
964/// struct {
965/// pub fn start(s: S) Result { ... }
966/// }
967/// ```
968/// where `Result` is any type.
969pub fn async(io: Io, s: anytype) Future(@typeInfo(@TypeOf(@TypeOf(s).start)).@"fn".return_type.?) {
970 const S = @TypeOf(s);
971 const Result = @typeInfo(@TypeOf(S.start)).@"fn".return_type.?;
972 const TypeErased = struct {
973 fn start(context: ?*anyopaque, result: *anyopaque) void {
974 const context_casted: *const S = @alignCast(@ptrCast(context));
975 const result_casted: *Result = @ptrCast(@alignCast(result));
976 result_casted.* = S.start(context_casted.*);
977 }
978 };
979 var future: Future(Result) = undefined;
980 future.any_future = io.vtable.async(io.userdata, @ptrCast((&future.result)[0..1]), @constCast(&s), TypeErased.start);
981 return future;
982}
lib/std/Thread/Pool.zig+3-20
...@@ -7,6 +7,7 @@ mutex: std.Thread.Mutex = .{},...@@ -7,6 +7,7 @@ mutex: std.Thread.Mutex = .{},
7cond: std.Thread.Condition = .{},7cond: std.Thread.Condition = .{},
8run_queue: std.SinglyLinkedList = .{},8run_queue: std.SinglyLinkedList = .{},
9is_running: bool = true,9is_running: bool = true,
10/// Must be a thread-safe allocator.
10allocator: std.mem.Allocator,11allocator: std.mem.Allocator,
11threads: if (builtin.single_threaded) [0]std.Thread else []std.Thread,12threads: if (builtin.single_threaded) [0]std.Thread else []std.Thread,
12ids: if (builtin.single_threaded) struct {13ids: if (builtin.single_threaded) struct {
...@@ -16,12 +17,12 @@ ids: if (builtin.single_threaded) struct {...@@ -16,12 +17,12 @@ ids: if (builtin.single_threaded) struct {
16 }17 }
17} else std.AutoArrayHashMapUnmanaged(std.Thread.Id, void),18} else std.AutoArrayHashMapUnmanaged(std.Thread.Id, void),
1819
19const Runnable = struct {20pub const Runnable = struct {
20 runFn: RunProto,21 runFn: RunProto,
21 node: std.SinglyLinkedList.Node = .{},22 node: std.SinglyLinkedList.Node = .{},
22};23};
2324
24const RunProto = *const fn (*Runnable, id: ?usize) void;25pub const RunProto = *const fn (*Runnable, id: ?usize) void;
2526
26pub const Options = struct {27pub const Options = struct {
27 allocator: std.mem.Allocator,28 allocator: std.mem.Allocator,
...@@ -117,12 +118,6 @@ pub fn spawnWg(pool: *Pool, wait_group: *WaitGroup, comptime func: anytype, args...@@ -117,12 +118,6 @@ pub fn spawnWg(pool: *Pool, wait_group: *WaitGroup, comptime func: anytype, args
117 const closure: *@This() = @alignCast(@fieldParentPtr("runnable", runnable));118 const closure: *@This() = @alignCast(@fieldParentPtr("runnable", runnable));
118 @call(.auto, func, closure.arguments);119 @call(.auto, func, closure.arguments);
119 closure.wait_group.finish();120 closure.wait_group.finish();
120
121 // The thread pool's allocator is protected by the mutex.
122 const mutex = &closure.pool.mutex;
123 mutex.lock();
124 defer mutex.unlock();
125
126 closure.pool.allocator.destroy(closure);121 closure.pool.allocator.destroy(closure);
127 }122 }
128 };123 };
...@@ -179,12 +174,6 @@ pub fn spawnWgId(pool: *Pool, wait_group: *WaitGroup, comptime func: anytype, ar...@@ -179,12 +174,6 @@ pub fn spawnWgId(pool: *Pool, wait_group: *WaitGroup, comptime func: anytype, ar
179 const closure: *@This() = @alignCast(@fieldParentPtr("runnable", runnable));174 const closure: *@This() = @alignCast(@fieldParentPtr("runnable", runnable));
180 @call(.auto, func, .{id.?} ++ closure.arguments);175 @call(.auto, func, .{id.?} ++ closure.arguments);
181 closure.wait_group.finish();176 closure.wait_group.finish();
182
183 // The thread pool's allocator is protected by the mutex.
184 const mutex = &closure.pool.mutex;
185 mutex.lock();
186 defer mutex.unlock();
187
188 closure.pool.allocator.destroy(closure);177 closure.pool.allocator.destroy(closure);
189 }178 }
190 };179 };
...@@ -228,12 +217,6 @@ pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) !void {...@@ -228,12 +217,6 @@ pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) !void {
228 fn runFn(runnable: *Runnable, _: ?usize) void {217 fn runFn(runnable: *Runnable, _: ?usize) void {
229 const closure: *@This() = @alignCast(@fieldParentPtr("runnable", runnable));218 const closure: *@This() = @alignCast(@fieldParentPtr("runnable", runnable));
230 @call(.auto, func, closure.arguments);219 @call(.auto, func, closure.arguments);
231
232 // The thread pool's allocator is protected by the mutex.
233 const mutex = &closure.pool.mutex;
234 mutex.lock();
235 defer mutex.unlock();
236
237 closure.pool.allocator.destroy(closure);220 closure.pool.allocator.destroy(closure);
238 }221 }
239 };222 };