| ... | ... | @@ -1,7 +1,8 @@ |
| 1 | | const std = @import("std"); |
| 2 | 1 | const builtin = @import("builtin"); |
| 3 | | const Pool = @This(); |
| 2 | const std = @import("std"); |
| 3 | const assert = std.debug.assert; |
| 4 | 4 | const WaitGroup = @import("WaitGroup.zig"); |
| 5 | const Pool = @This(); |
| 5 | 6 | |
| 6 | 7 | mutex: std.Thread.Mutex = .{}, |
| 7 | 8 | cond: std.Thread.Condition = .{}, |
| ... | ... | @@ -307,3 +308,60 @@ pub fn waitAndWork(pool: *Pool, wait_group: *WaitGroup) void { |
| 307 | 308 | pub fn getIdCount(pool: *Pool) usize { |
| 308 | 309 | return @intCast(1 + pool.threads.len); |
| 309 | 310 | } |
| 311 | |
| 312 | const AsyncClosure = struct { |
| 313 | func: *const fn (context: ?*anyopaque, result: *anyopaque) void, |
| 314 | context: ?*anyopaque, |
| 315 | run_node: std.Thread.Pool.RunQueue.Node = .{ .data = .{ .runFn = runFn } }, |
| 316 | reset_event: std.Thread.ResetEvent, |
| 317 | |
| 318 | fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void { |
| 319 | const run_node: *std.Thread.Pool.RunQueue.Node = @fieldParentPtr("data", runnable); |
| 320 | const closure: *@This() = @alignCast(@fieldParentPtr("run_node", run_node)); |
| 321 | closure.func(closure.context, closure.resultPointer()); |
| 322 | closure.reset_event.set(); |
| 323 | } |
| 324 | |
| 325 | fn resultPointer(closure: *@This()) [*]u8 { |
| 326 | const base: [*]u8 = @ptrCast(closure); |
| 327 | return base + @sizeOf(@This()); |
| 328 | } |
| 329 | }; |
| 330 | |
| 331 | pub fn @"async"( |
| 332 | userdata: ?*anyopaque, |
| 333 | eager_result: []u8, |
| 334 | context: ?*anyopaque, |
| 335 | start: *const fn (context: ?*anyopaque, result: *anyopaque) void, |
| 336 | ) ?*std.Io.AnyFuture { |
| 337 | const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata)); |
| 338 | pool.mutex.lock(); |
| 339 | |
| 340 | const gpa = pool.allocator; |
| 341 | const n = @sizeOf(AsyncClosure) + eager_result.len; |
| 342 | const closure: *AsyncClosure = @alignCast(@ptrCast(gpa.alignedAlloc(u8, @alignOf(AsyncClosure), n) catch { |
| 343 | pool.mutex.unlock(); |
| 344 | start(context, eager_result.ptr); |
| 345 | return null; |
| 346 | })); |
| 347 | closure.* = .{ |
| 348 | .func = start, |
| 349 | .context = context, |
| 350 | .reset_event = .{}, |
| 351 | }; |
| 352 | pool.run_queue.prepend(&closure.run_node); |
| 353 | pool.mutex.unlock(); |
| 354 | |
| 355 | pool.cond.signal(); |
| 356 | |
| 357 | return @ptrCast(closure); |
| 358 | } |
| 359 | |
| 360 | pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void { |
| 361 | const thread_pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata)); |
| 362 | const closure: *AsyncClosure = @ptrCast(@alignCast(any_future)); |
| 363 | closure.reset_event.wait(); |
| 364 | const base: [*]align(@alignOf(AsyncClosure)) u8 = @ptrCast(closure); |
| 365 | @memcpy(result, (base + @sizeOf(AsyncClosure))[0..result.len]); |
| 366 | thread_pool.allocator.free(base[0 .. @sizeOf(AsyncClosure) + result.len]); |
| 367 | } |
Andrew Kelley <andrew@ziglang.org> 2025-03-24 18:49:03-07:00
lib