authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-29 20:58:07-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
logc278830592792ed724f68b4abbad9d253bed5404
treebaed506c94a589928a9a66c2aa1a431d33f3947b
parent79e278f6a26467ea7788ae14267000517d91f1ac

std.Io: introduce cancellation

3 files changed, 239 insertions(+), 117 deletions(-)

lib/std/Io.zig+53-10
......@@ -564,6 +564,8 @@ vtable: *const VTable,
564564pub const VTable = struct {
565565 /// If it returns `null` it means `result` has been already populated and
566566 /// `await` will be a no-op.
567 ///
568 /// Thread-safe.
567569 async: *const fn (
568570 /// Corresponds to `Io.userdata`.
569571 userdata: ?*anyopaque,
......@@ -579,6 +581,8 @@ pub const VTable = struct {
579581 ) ?*AnyFuture,
580582
581583 /// This function is only called when `async` returns a non-null value.
584 ///
585 /// Thread-safe.
582586 await: *const fn (
583587 /// Corresponds to `Io.userdata`.
584588 userdata: ?*anyopaque,
......@@ -589,13 +593,41 @@ pub const VTable = struct {
589593 result: []u8,
590594 ) void,
591595
592 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) fs.File.OpenError!fs.File,
593 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) fs.File.OpenError!fs.File,
596 /// Equivalent to `await` but initiates cancel request.
597 ///
598 /// This function is only called when `async` returns a non-null value.
599 ///
600 /// Thread-safe.
601 cancel: *const fn (
602 /// Corresponds to `Io.userdata`.
603 userdata: ?*anyopaque,
604 /// The same value that was returned from `async`.
605 any_future: *AnyFuture,
606 /// Points to a buffer where the result is written.
607 /// The length is equal to size in bytes of result type.
608 result: []u8,
609 ) void,
610
611 /// Returns whether the current thread of execution is known to have
612 /// been requested to cancel.
613 ///
614 /// Thread-safe.
615 cancelRequested: *const fn (?*anyopaque) bool,
616
617 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
618 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
594619 closeFile: *const fn (?*anyopaque, fs.File) void,
595 read: *const fn (?*anyopaque, file: fs.File, buffer: []u8) fs.File.ReadError!usize,
596 write: *const fn (?*anyopaque, file: fs.File, buffer: []const u8) fs.File.WriteError!usize,
620 read: *const fn (?*anyopaque, file: fs.File, buffer: []u8) FileReadError!usize,
621 write: *const fn (?*anyopaque, file: fs.File, buffer: []const u8) FileWriteError!usize,
597622};
598623
624pub const OpenFlags = fs.File.OpenFlags;
625pub const CreateFlags = fs.File.CreateFlags;
626
627pub const FileOpenError = fs.File.OpenError || error{AsyncCancel};
628pub const FileReadError = fs.File.ReadError || error{AsyncCancel};
629pub const FileWriteError = fs.File.WriteError || error{AsyncCancel};
630
599631pub const AnyFuture = opaque {};
600632
601633pub fn Future(Result: type) type {
......@@ -603,6 +635,17 @@ pub fn Future(Result: type) type {
603635 any_future: ?*AnyFuture,
604636 result: Result,
605637
638 /// Equivalent to `await` but sets a flag observable to application
639 /// code that cancellation has been requested.
640 ///
641 /// Idempotent.
642 pub fn cancel(f: *@This(), io: Io) Result {
643 const any_future = f.any_future orelse return f.result;
644 io.vtable.cancel(io.userdata, any_future, @ptrCast((&f.result)[0..1]));
645 f.any_future = null;
646 return f.result;
647 }
648
606649 pub fn await(f: *@This(), io: Io) Result {
607650 const any_future = f.any_future orelse return f.result;
608651 io.vtable.await(io.userdata, any_future, @ptrCast((&f.result)[0..1]));
......@@ -636,11 +679,11 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(
636679 return future;
637680}
638681
639pub fn openFile(io: Io, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) fs.File.OpenError!fs.File {
682pub fn openFile(io: Io, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File {
640683 return io.vtable.openFile(io.userdata, dir, sub_path, flags);
641684}
642685
643pub fn createFile(io: Io, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) fs.File.OpenError!fs.File {
686pub fn createFile(io: Io, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File {
644687 return io.vtable.createFile(io.userdata, dir, sub_path, flags);
645688}
646689
......@@ -648,22 +691,22 @@ pub fn closeFile(io: Io, file: fs.File) void {
648691 return io.vtable.closeFile(io.userdata, file);
649692}
650693
651pub fn read(io: Io, file: fs.File, buffer: []u8) fs.File.ReadError!usize {
694pub fn read(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {
652695 return io.vtable.read(io.userdata, file, buffer);
653696}
654697
655pub fn write(io: Io, file: fs.File, buffer: []const u8) fs.File.WriteError!usize {
698pub fn write(io: Io, file: fs.File, buffer: []const u8) FileWriteError!usize {
656699 return io.vtable.write(io.userdata, file, buffer);
657700}
658701
659pub fn writeAll(io: Io, file: fs.File, bytes: []const u8) fs.File.WriteError!void {
702pub fn writeAll(io: Io, file: fs.File, bytes: []const u8) FileWriteError!void {
660703 var index: usize = 0;
661704 while (index < bytes.len) {
662705 index += try io.write(file, bytes[index..]);
663706 }
664707}
665708
666pub fn readAll(io: Io, file: fs.File, buffer: []u8) fs.File.ReadError!usize {
709pub fn readAll(io: Io, file: fs.File, buffer: []u8) FileReadError!usize {
667710 var index: usize = 0;
668711 while (index != buffer.len) {
669712 const amt = try io.read(file, buffer[index..]);
lib/std/Io/EventLoop.zig+13-3
......@@ -7,12 +7,13 @@ const EventLoop = @This();
77const Alignment = std.mem.Alignment;
88const IoUring = std.os.linux.IoUring;
99
10/// Must be a thread-safe allocator.
1011gpa: Allocator,
1112mutex: std.Thread.Mutex,
12queue: std.DoublyLinkedList(void),
13queue: std.DoublyLinkedList,
1314/// Atomic copy of queue.len
1415queue_len: u32,
15free: std.DoublyLinkedList(void),
16free: std.DoublyLinkedList,
1617main_fiber: Fiber,
1718idle_count: usize,
1819threads: std.ArrayListUnmanaged(Thread),
......@@ -39,7 +40,7 @@ const Thread = struct {
3940const Fiber = struct {
4041 context: Context,
4142 awaiter: ?*Fiber,
42 queue_node: std.DoublyLinkedList(void).Node,
43 queue_node: std.DoublyLinkedList.Node,
4344 result_align: Alignment,
4445
4546 const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));
......@@ -447,6 +448,15 @@ pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []
447448 event_loop.recycle(future_fiber);
448449}
449450
451pub fn cancel(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
452 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
453 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
454 // TODO set a flag that makes all IO operations for this fiber return error.Canceled
455 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
456 @memcpy(result, future_fiber.resultPointer());
457 event_loop.recycle(future_fiber);
458}
459
450460pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.CreateFlags) std.fs.File.OpenError!std.fs.File {
451461 const el: *EventLoop = @ptrCast(@alignCast(userdata));
452462
lib/std/Thread/Pool.zig+173-104
......@@ -1,22 +1,27 @@
11const builtin = @import("builtin");
22const std = @import("std");
3const Allocator = std.mem.Allocator;
34const assert = std.debug.assert;
45const WaitGroup = @import("WaitGroup.zig");
6const Io = std.Io;
57const Pool = @This();
68
9/// Must be a thread-safe allocator.
10allocator: std.mem.Allocator,
711mutex: std.Thread.Mutex = .{},
812cond: std.Thread.Condition = .{},
913run_queue: std.SinglyLinkedList = .{},
1014is_running: bool = true,
11/// Must be a thread-safe allocator.
12allocator: std.mem.Allocator,
13threads: if (builtin.single_threaded) [0]std.Thread else []std.Thread,
15threads: std.ArrayListUnmanaged(std.Thread),
1416ids: if (builtin.single_threaded) struct {
1517 inline fn deinit(_: @This(), _: std.mem.Allocator) void {}
1618 fn getIndex(_: @This(), _: std.Thread.Id) usize {
1719 return 0;
1820 }
1921} else std.AutoArrayHashMapUnmanaged(std.Thread.Id, void),
22stack_size: usize,
23
24threadlocal var current_closure: ?*AsyncClosure = null;
2025
2126pub const Runnable = struct {
2227 runFn: RunProto,
......@@ -33,48 +38,36 @@ pub const Options = struct {
3338};
3439
3540pub fn init(pool: *Pool, options: Options) !void {
36 const allocator = options.allocator;
41 const gpa = options.allocator;
42 const thread_count = options.n_jobs orelse @max(1, std.Thread.getCpuCount() catch 1);
43 const threads = try gpa.alloc(std.Thread, thread_count);
44 errdefer gpa.free(threads);
3745
3846 pool.* = .{
39 .allocator = allocator,
40 .threads = if (builtin.single_threaded) .{} else &.{},
47 .allocator = gpa,
48 .threads = .initBuffer(threads),
4149 .ids = .{},
50 .stack_size = options.stack_size,
4251 };
4352
44 if (builtin.single_threaded) {
45 return;
46 }
53 if (builtin.single_threaded) return;
4754
48 const thread_count = options.n_jobs orelse @max(1, std.Thread.getCpuCount() catch 1);
4955 if (options.track_ids) {
50 try pool.ids.ensureTotalCapacity(allocator, 1 + thread_count);
56 try pool.ids.ensureTotalCapacity(gpa, 1 + thread_count);
5157 pool.ids.putAssumeCapacityNoClobber(std.Thread.getCurrentId(), {});
5258 }
53
54 // kill and join any threads we spawned and free memory on error.
55 pool.threads = try allocator.alloc(std.Thread, thread_count);
56 var spawned: usize = 0;
57 errdefer pool.join(spawned);
58
59 for (pool.threads) |*thread| {
60 thread.* = try std.Thread.spawn(.{
61 .stack_size = options.stack_size,
62 .allocator = allocator,
63 }, worker, .{pool});
64 spawned += 1;
65 }
6659}
6760
6861pub fn deinit(pool: *Pool) void {
69 pool.join(pool.threads.len); // kill and join all threads.
70 pool.ids.deinit(pool.allocator);
62 const gpa = pool.allocator;
63 pool.join();
64 pool.threads.deinit(gpa);
65 pool.ids.deinit(gpa);
7166 pool.* = undefined;
7267}
7368
74fn join(pool: *Pool, spawned: usize) void {
75 if (builtin.single_threaded) {
76 return;
77 }
69fn join(pool: *Pool) void {
70 if (builtin.single_threaded) return;
7871
7972 {
8073 pool.mutex.lock();
......@@ -87,11 +80,7 @@ fn join(pool: *Pool, spawned: usize) void {
8780 // wake up any sleeping threads (this can be done outside the mutex)
8881 // then wait for all the threads we know are spawned to complete.
8982 pool.cond.broadcast();
90 for (pool.threads[0..spawned]) |thread| {
91 thread.join();
92 }
93
94 pool.allocator.free(pool.threads);
83 for (pool.threads.items) |thread| thread.join();
9584}
9685
9786/// Runs `func` in the thread pool, calling `WaitGroup.start` beforehand, and
......@@ -123,26 +112,34 @@ pub fn spawnWg(pool: *Pool, wait_group: *WaitGroup, comptime func: anytype, args
123112 }
124113 };
125114
126 {
127 pool.mutex.lock();
128
129 const closure = pool.allocator.create(Closure) catch {
130 pool.mutex.unlock();
131 @call(.auto, func, args);
132 wait_group.finish();
133 return;
134 };
135 closure.* = .{
136 .arguments = args,
137 .pool = pool,
138 .wait_group = wait_group,
139 };
115 pool.mutex.lock();
140116
141 pool.run_queue.prepend(&closure.runnable.node);
117 const gpa = pool.allocator;
118 const closure = gpa.create(Closure) catch {
142119 pool.mutex.unlock();
120 @call(.auto, func, args);
121 wait_group.finish();
122 return;
123 };
124 closure.* = .{
125 .arguments = args,
126 .pool = pool,
127 .wait_group = wait_group,
128 };
129
130 pool.run_queue.prepend(&closure.runnable.node);
131
132 if (pool.threads.items.len < pool.threads.capacity) {
133 pool.threads.addOneAssumeCapacity().* = std.Thread.spawn(.{
134 .stack_size = pool.stack_size,
135 .allocator = gpa,
136 }, worker, .{pool}) catch t: {
137 pool.threads.items.len -= 1;
138 break :t undefined;
139 };
143140 }
144141
145 // Notify waiting threads outside the lock to try and keep the critical section small.
142 pool.mutex.unlock();
146143 pool.cond.signal();
147144}
148145
......@@ -179,31 +176,39 @@ pub fn spawnWgId(pool: *Pool, wait_group: *WaitGroup, comptime func: anytype, ar
179176 }
180177 };
181178
182 {
183 pool.mutex.lock();
184
185 const closure = pool.allocator.create(Closure) catch {
186 const id: ?usize = pool.ids.getIndex(std.Thread.getCurrentId());
187 pool.mutex.unlock();
188 @call(.auto, func, .{id.?} ++ args);
189 wait_group.finish();
190 return;
191 };
192 closure.* = .{
193 .arguments = args,
194 .pool = pool,
195 .wait_group = wait_group,
196 };
179 pool.mutex.lock();
197180
198 pool.run_queue.prepend(&closure.runnable.node);
181 const gpa = pool.allocator;
182 const closure = gpa.create(Closure) catch {
183 const id: ?usize = pool.ids.getIndex(std.Thread.getCurrentId());
199184 pool.mutex.unlock();
185 @call(.auto, func, .{id.?} ++ args);
186 wait_group.finish();
187 return;
188 };
189 closure.* = .{
190 .arguments = args,
191 .pool = pool,
192 .wait_group = wait_group,
193 };
194
195 pool.run_queue.prepend(&closure.runnable.node);
196
197 if (pool.threads.items.len < pool.threads.capacity) {
198 pool.threads.addOneAssumeCapacity().* = std.Thread.spawn(.{
199 .stack_size = pool.stack_size,
200 .allocator = gpa,
201 }, worker, .{pool}) catch t: {
202 pool.threads.items.len -= 1;
203 break :t undefined;
204 };
200205 }
201206
202 // Notify waiting threads outside the lock to try and keep the critical section small.
207 pool.mutex.unlock();
203208 pool.cond.signal();
204209}
205210
206pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) !void {
211pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) void {
207212 if (builtin.single_threaded) {
208213 @call(.auto, func, args);
209214 return;
......@@ -222,20 +227,32 @@ pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) !void {
222227 }
223228 };
224229
225 {
226 pool.mutex.lock();
227 defer pool.mutex.unlock();
230 pool.mutex.lock();
228231
229 const closure = try pool.allocator.create(Closure);
230 closure.* = .{
231 .arguments = args,
232 .pool = pool,
233 };
232 const gpa = pool.allocator;
233 const closure = gpa.create(Closure) catch {
234 pool.mutex.unlock();
235 @call(.auto, func, args);
236 return;
237 };
238 closure.* = .{
239 .arguments = args,
240 .pool = pool,
241 };
242
243 pool.run_queue.prepend(&closure.runnable.node);
234244
235 pool.run_queue.prepend(&closure.runnable.node);
245 if (pool.threads.items.len < pool.threads.capacity) {
246 pool.threads.addOneAssumeCapacity().* = std.Thread.spawn(.{
247 .stack_size = pool.stack_size,
248 .allocator = gpa,
249 }, worker, .{pool}) catch t: {
250 pool.threads.items.len -= 1;
251 break :t undefined;
252 };
236253 }
237254
238 // Notify waiting threads outside the lock to try and keep the critical section small.
255 pool.mutex.unlock();
239256 pool.cond.signal();
240257}
241258
......@@ -254,7 +271,7 @@ test spawn {
254271 .allocator = std.testing.allocator,
255272 });
256273 defer pool.deinit();
257 try pool.spawn(TestFn.checkRun, .{&completed});
274 pool.spawn(TestFn.checkRun, .{&completed});
258275 }
259276
260277 try std.testing.expectEqual(true, completed);
......@@ -306,15 +323,17 @@ pub fn waitAndWork(pool: *Pool, wait_group: *WaitGroup) void {
306323}
307324
308325pub fn getIdCount(pool: *Pool) usize {
309 return @intCast(1 + pool.threads.len);
326 return @intCast(1 + pool.threads.items.len);
310327}
311328
312pub fn io(pool: *Pool) std.Io {
329pub fn io(pool: *Pool) Io {
313330 return .{
314331 .userdata = pool,
315332 .vtable = &.{
316333 .@"async" = @"async",
317334 .@"await" = @"await",
335 .cancel = cancel,
336 .cancelRequested = cancelRequested,
318337 .createFile = createFile,
319338 .openFile = openFile,
320339 .closeFile = closeFile,
......@@ -326,15 +345,17 @@ pub fn io(pool: *Pool) std.Io {
326345
327346const AsyncClosure = struct {
328347 func: *const fn (context: *anyopaque, result: *anyopaque) void,
329 run_node: std.Thread.Pool.RunQueue.Node = .{ .data = .{ .runFn = runFn } },
348 runnable: Runnable = .{ .runFn = runFn },
330349 reset_event: std.Thread.ResetEvent,
350 cancel_flag: bool,
331351 context_offset: usize,
332352 result_offset: usize,
333353
334354 fn runFn(runnable: *std.Thread.Pool.Runnable, _: ?usize) void {
335 const run_node: *std.Thread.Pool.RunQueue.Node = @fieldParentPtr("data", runnable);
336 const closure: *AsyncClosure = @alignCast(@fieldParentPtr("run_node", run_node));
355 const closure: *AsyncClosure = @alignCast(@fieldParentPtr("runnable", runnable));
356 current_closure = closure;
337357 closure.func(closure.contextPointer(), closure.resultPointer());
358 current_closure = null;
338359 closure.reset_event.set();
339360 }
340361
......@@ -359,16 +380,23 @@ const AsyncClosure = struct {
359380 const base: [*]u8 = @ptrCast(closure);
360381 return base + closure.context_offset;
361382 }
383
384 fn waitAndFree(closure: *AsyncClosure, gpa: Allocator, result: []u8) void {
385 closure.reset_event.wait();
386 const base: [*]align(@alignOf(AsyncClosure)) u8 = @ptrCast(closure);
387 @memcpy(result, closure.resultPointer()[0..result.len]);
388 gpa.free(base[0 .. closure.result_offset + result.len]);
389 }
362390};
363391
364pub fn @"async"(
392fn @"async"(
365393 userdata: ?*anyopaque,
366394 result: []u8,
367395 result_alignment: std.mem.Alignment,
368396 context: []const u8,
369397 context_alignment: std.mem.Alignment,
370398 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
371) ?*std.Io.AnyFuture {
399) ?*Io.AnyFuture {
372400 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
373401 pool.mutex.lock();
374402
......@@ -386,46 +414,87 @@ pub fn @"async"(
386414 .context_offset = context_offset,
387415 .result_offset = result_offset,
388416 .reset_event = .{},
417 .cancel_flag = false,
389418 };
390419 @memcpy(closure.contextPointer()[0..context.len], context);
391 pool.run_queue.prepend(&closure.run_node);
392 pool.mutex.unlock();
420 pool.run_queue.prepend(&closure.runnable.node);
421
422 if (pool.threads.items.len < pool.threads.capacity) {
423 pool.threads.addOneAssumeCapacity().* = std.Thread.spawn(.{
424 .stack_size = pool.stack_size,
425 .allocator = gpa,
426 }, worker, .{pool}) catch t: {
427 pool.threads.items.len -= 1;
428 break :t undefined;
429 };
430 }
393431
432 pool.mutex.unlock();
394433 pool.cond.signal();
395434
396435 return @ptrCast(closure);
397436}
398437
399pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
400 const thread_pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
438fn @"await"(userdata: ?*anyopaque, any_future: *Io.AnyFuture, result: []u8) void {
439 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
401440 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));
402 closure.reset_event.wait();
403 const base: [*]align(@alignOf(AsyncClosure)) u8 = @ptrCast(closure);
404 @memcpy(result, closure.resultPointer()[0..result.len]);
405 thread_pool.allocator.free(base[0 .. closure.result_offset + result.len]);
441 closure.waitAndFree(pool.allocator, result);
442}
443
444fn cancel(userdata: ?*anyopaque, any_future: *Io.AnyFuture, result: []u8) void {
445 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
446 const closure: *AsyncClosure = @ptrCast(@alignCast(any_future));
447 @atomicStore(bool, &closure.cancel_flag, true, .seq_cst);
448 closure.waitAndFree(pool.allocator, result);
449}
450
451fn cancelRequested(userdata: ?*anyopaque) bool {
452 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
453 _ = pool;
454 const closure = current_closure orelse return false;
455 return @atomicLoad(bool, &closure.cancel_flag, .unordered);
456}
457
458fn checkCancel(pool: *Pool) error{AsyncCancel}!void {
459 if (cancelRequested(pool)) return error.AsyncCancel;
406460}
407461
408pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.CreateFlags) std.fs.File.OpenError!std.fs.File {
409 _ = userdata;
462pub fn createFile(
463 userdata: ?*anyopaque,
464 dir: std.fs.Dir,
465 sub_path: []const u8,
466 flags: std.fs.File.CreateFlags,
467) Io.FileOpenError!std.fs.File {
468 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
469 try pool.checkCancel();
410470 return dir.createFile(sub_path, flags);
411471}
412472
413pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.OpenFlags) std.fs.File.OpenError!std.fs.File {
414 _ = userdata;
473pub fn openFile(
474 userdata: ?*anyopaque,
475 dir: std.fs.Dir,
476 sub_path: []const u8,
477 flags: std.fs.File.OpenFlags,
478) Io.FileOpenError!std.fs.File {
479 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
480 try pool.checkCancel();
415481 return dir.openFile(sub_path, flags);
416482}
417483
418484pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
419 _ = userdata;
485 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
486 _ = pool;
420487 return file.close();
421488}
422489
423pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File.ReadError!usize {
424 _ = userdata;
490pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) Io.FileReadError!usize {
491 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
492 try pool.checkCancel();
425493 return file.read(buffer);
426494}
427495
428pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) std.fs.File.WriteError!usize {
429 _ = userdata;
496pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) Io.FileWriteError!usize {
497 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
498 try pool.checkCancel();
430499 return file.write(buffer);
431500}