authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-27 20:53:14-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:38-07:00
logad3c5f02925fe939cc54619c11f687c975dccb89
treeddfac9a93c2e9fd36ce77434440fb7682d351fba
parent0c1f5dbd64877d9e002648605ba2eec8ddbfcf16

fix context passing in threaded Io impl

3 files changed, 73 insertions(+), 20 deletions(-)

lib/std/Io.zig+16-8
...@@ -928,10 +928,12 @@ pub const VTable = struct {...@@ -928,10 +928,12 @@ pub const VTable = struct {
928 /// The pointer of this slice is an "eager" result value.928 /// The pointer of this slice is an "eager" result value.
929 /// The length is the size in bytes of the result type.929 /// The length is the size in bytes of the result type.
930 /// This pointer's lifetime expires directly after the call to this function.930 /// This pointer's lifetime expires directly after the call to this function.
931 eager_result: []u8,931 result: []u8,
932 /// Passed to `start`.932 result_alignment: std.mem.Alignment,
933 context: ?*anyopaque,933 /// Copied and then passed to `start`.
934 start: *const fn (context: ?*anyopaque, result: *anyopaque) void,934 context: []const u8,
935 context_alignment: std.mem.Alignment,
936 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
935 ) ?*AnyFuture,937 ) ?*AnyFuture,
936938
937 /// This function is only called when `async` returns a non-null value.939 /// This function is only called when `async` returns a non-null value.
...@@ -969,17 +971,23 @@ pub fn Future(Result: type) type {...@@ -969,17 +971,23 @@ pub fn Future(Result: type) type {
969/// }971/// }
970/// ```972/// ```
971/// where `Result` is any type.973/// where `Result` is any type.
972pub fn async(io: Io, s: anytype) Future(@typeInfo(@TypeOf(@TypeOf(s).start)).@"fn".return_type.?) {974pub fn async(io: Io, S: type, s: S) Future(@typeInfo(@TypeOf(S.start)).@"fn".return_type.?) {
973 const S = @TypeOf(s);
974 const Result = @typeInfo(@TypeOf(S.start)).@"fn".return_type.?;975 const Result = @typeInfo(@TypeOf(S.start)).@"fn".return_type.?;
975 const TypeErased = struct {976 const TypeErased = struct {
976 fn start(context: ?*anyopaque, result: *anyopaque) void {977 fn start(context: *const anyopaque, result: *anyopaque) void {
977 const context_casted: *const S = @alignCast(@ptrCast(context));978 const context_casted: *const S = @alignCast(@ptrCast(context));
978 const result_casted: *Result = @ptrCast(@alignCast(result));979 const result_casted: *Result = @ptrCast(@alignCast(result));
979 result_casted.* = S.start(context_casted.*);980 result_casted.* = S.start(context_casted.*);
980 }981 }
981 };982 };
982 var future: Future(Result) = undefined;983 var future: Future(Result) = undefined;
983 future.any_future = io.vtable.async(io.userdata, @ptrCast((&future.result)[0..1]), @constCast(&s), TypeErased.start);984 future.any_future = io.vtable.async(
985 io.userdata,
986 @ptrCast((&future.result)[0..1]),
987 .fromByteUnits(@alignOf(Result)),
988 if (@sizeOf(S) == 0) &.{} else @ptrCast((&s)[0..1]), // work around compiler bug
989 .fromByteUnits(@alignOf(S)),
990 TypeErased.start,
991 );
984 return future;992 return future;
985}993}
lib/std/Io/EventLoop.zig+11
...@@ -55,6 +55,17 @@ const Fiber = struct {...@@ -55,6 +55,17 @@ const Fiber = struct {
55 }55 }
56};56};
5757
58pub fn io(el: *EventLoop) Io {
59 return .{
60 .userdata = el,
61 .vtable = &.{
62 .@"async" = @"async",
63 .@"await" = @"await",
64 },
65 };
66}
67
68
58pub fn init(el: *EventLoop, gpa: Allocator) error{OutOfMemory}!void {69pub fn init(el: *EventLoop, gpa: Allocator) error{OutOfMemory}!void {
59 const threads_bytes = ((std.Thread.getCpuCount() catch 1) -| 1) * @sizeOf(Thread);70 const threads_bytes = ((std.Thread.getCpuCount() catch 1) -| 1) * @sizeOf(Thread);
60 const idle_context_offset = std.mem.alignForward(usize, threads_bytes, @alignOf(Context));71 const idle_context_offset = std.mem.alignForward(usize, threads_bytes, @alignOf(Context));
lib/std/Thread/Pool.zig+46-12
...@@ -309,46 +309,80 @@ pub fn getIdCount(pool: *Pool) usize {...@@ -309,46 +309,80 @@ pub fn getIdCount(pool: *Pool) usize {
309 return @intCast(1 + pool.threads.len);309 return @intCast(1 + pool.threads.len);
310}310}
311311
312pub fn io(pool: *Pool) std.Io {
313 return .{
314 .userdata = pool,
315 .vtable = &.{
316 .@"async" = @"async",
317 .@"await" = @"await",
318 },
319 };
320}
321
312const AsyncClosure = struct {322const AsyncClosure = struct {
313 func: *const fn (context: ?*anyopaque, result: *anyopaque) void,323 func: *const fn (context: *anyopaque, result: *anyopaque) void,
314 context: ?*anyopaque,
315 run_node: std.Thread.Pool.RunQueue.Node = .{ .data = .{ .runFn = runFn } },324 run_node: std.Thread.Pool.RunQueue.Node = .{ .data = .{ .runFn = runFn } },
316 reset_event: std.Thread.ResetEvent,325 reset_event: std.Thread.ResetEvent,
326 context_offset: usize,
327 result_offset: usize,
317328
318 fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void {329 fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void {
319 const run_node: *std.Thread.Pool.RunQueue.Node = @fieldParentPtr("data", runnable);330 const run_node: *std.Thread.Pool.RunQueue.Node = @fieldParentPtr("data", runnable);
320 const closure: *@This() = @alignCast(@fieldParentPtr("run_node", run_node));331 const closure: *AsyncClosure = @alignCast(@fieldParentPtr("run_node", run_node));
321 closure.func(closure.context, closure.resultPointer());332 closure.func(closure.contextPointer(), closure.resultPointer());
322 closure.reset_event.set();333 closure.reset_event.set();
323 }334 }
324335
325 fn resultPointer(closure: *@This()) [*]u8 {336 fn contextOffset(context_alignment: std.mem.Alignment) usize {
337 return context_alignment.forward(@sizeOf(AsyncClosure));
338 }
339
340 fn resultOffset(
341 context_alignment: std.mem.Alignment,
342 context_len: usize,
343 result_alignment: std.mem.Alignment,
344 ) usize {
345 return result_alignment.forward(contextOffset(context_alignment) + context_len);
346 }
347
348 fn resultPointer(closure: *AsyncClosure) [*]u8 {
349 const base: [*]u8 = @ptrCast(closure);
350 return base + closure.result_offset;
351 }
352
353 fn contextPointer(closure: *AsyncClosure) [*]u8 {
326 const base: [*]u8 = @ptrCast(closure);354 const base: [*]u8 = @ptrCast(closure);
327 return base + @sizeOf(@This());355 return base + closure.context_offset;
328 }356 }
329};357};
330358
331pub fn @"async"(359pub fn @"async"(
332 userdata: ?*anyopaque,360 userdata: ?*anyopaque,
333 eager_result: []u8,361 result: []u8,
334 context: ?*anyopaque,362 result_alignment: std.mem.Alignment,
335 start: *const fn (context: ?*anyopaque, result: *anyopaque) void,363 context: []const u8,
364 context_alignment: std.mem.Alignment,
365 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
336) ?*std.Io.AnyFuture {366) ?*std.Io.AnyFuture {
337 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));367 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
338 pool.mutex.lock();368 pool.mutex.lock();
339369
340 const gpa = pool.allocator;370 const gpa = pool.allocator;
341 const n = @sizeOf(AsyncClosure) + eager_result.len;371 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));
372 const result_offset = result_alignment.forward(context_offset + context.len);
373 const n = result_offset + result.len;
342 const closure: *AsyncClosure = @alignCast(@ptrCast(gpa.alignedAlloc(u8, @alignOf(AsyncClosure), n) catch {374 const closure: *AsyncClosure = @alignCast(@ptrCast(gpa.alignedAlloc(u8, @alignOf(AsyncClosure), n) catch {
343 pool.mutex.unlock();375 pool.mutex.unlock();
344 start(context, eager_result.ptr);376 start(context.ptr, result.ptr);
345 return null;377 return null;
346 }));378 }));
347 closure.* = .{379 closure.* = .{
348 .func = start,380 .func = start,
349 .context = context,381 .context_offset = context_offset,
382 .result_offset = result_offset,
350 .reset_event = .{},383 .reset_event = .{},
351 };384 };
385 @memcpy(closure.contextPointer()[0..context.len], context);
352 pool.run_queue.prepend(&closure.run_node);386 pool.run_queue.prepend(&closure.run_node);
353 pool.mutex.unlock();387 pool.mutex.unlock();
354388