authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2026-01-30 01:44:07-05:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:05-08:00
log3d3f22a14d7639f1a7c607da98926da6e60c3b01
treed7144601a752d50da1c9cea235db650ffb76f8ea
parent10bec043f52c08eef73b758042139635e535c0d3

Io.Batch: implement alternate API

4 files changed, 508 insertions(+), 367 deletions(-)

lib/std/Io.zig+138-109
......@@ -149,8 +149,9 @@ pub const VTable = struct {
149149 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,
150150 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,
151151
152 operate: *const fn (?*anyopaque, *Operation) Cancelable!void,
153 batchWait: *const fn (?*anyopaque, *Batch, Timeout) Batch.WaitError!void,
152 operate: *const fn (?*anyopaque, Operation) Cancelable!Operation.Result,
153 batchAwaitAsync: *const fn (?*anyopaque, *Batch) Batch.AwaitAsyncError!void,
154 batchAwaitConcurrent: *const fn (?*anyopaque, *Batch, Timeout) Batch.AwaitConcurrentError!void,
154155 batchCancel: *const fn (?*anyopaque, *Batch) void,
155156
156157 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
......@@ -255,19 +256,14 @@ pub const VTable = struct {
255256};
256257
257258pub const Operation = union(enum) {
258 noop: Noop,
259259 file_read_streaming: FileReadStreaming,
260260
261 pub const Noop = struct {
262 reserved: [2]usize = .{ 0, 0 },
263 status: Status(void) = .{ .unstarted = {} },
264 };
261 pub const Tag = @typeInfo(Operation).@"union".tag_type.?;
265262
266263 /// May return 0 reads which is different than `error.EndOfStream`.
267264 pub const FileReadStreaming = struct {
268265 file: File,
269266 data: []const []u8,
270 status: Status(Error!usize) = .{ .unstarted = {} },
271267
272268 pub const Error = UnendingError || error{EndOfStream};
273269 pub const UnendingError = error{
......@@ -290,19 +286,72 @@ pub const Operation = union(enum) {
290286 /// lock.
291287 LockViolation,
292288 } || Io.UnexpectedError;
289
290 pub const Result = usize;
291 };
292
293 pub const Result = Result: {
294 const operation_fields = @typeInfo(Operation).@"union".fields;
295 var field_names: [operation_fields.len][]const u8 = undefined;
296 var field_types: [operation_fields.len]type = undefined;
297 for (operation_fields, &field_names, &field_types) |field, *field_name, *field_type| {
298 field_name.* = field.name;
299 field_type.* = field.type.Error!field.type.Result;
300 }
301 break :Result @Union(.auto, Tag, &field_names, &field_types, &@splat(.{}));
293302 };
294303
295 pub fn Status(Result: type) type {
296 return union {
297 unstarted: void,
298 pending: *Batch,
304 pub const Storage = union {
305 unused: List.DoubleNode,
306 submission: Submission,
307 pending: Pending,
308 completion: Completion,
309
310 pub const Submission = struct {
311 node: List.SingleNode,
312 operation: Operation,
313 };
314
315 pub const Pending = struct {
316 node: List.DoubleNode,
317 tag: Tag,
318 context: [3]usize,
319 };
320
321 pub const Completion = struct {
322 node: List.SingleNode,
299323 result: Result,
300324 };
301 }
325 };
326
327 pub const OptionalIndex = enum(u32) {
328 none = std.math.maxInt(u32),
329 _,
330
331 pub fn fromIndex(i: usize) OptionalIndex {
332 const oi: OptionalIndex = @enumFromInt(i);
333 assert(oi != .none);
334 return oi;
335 }
336
337 pub fn toIndex(oi: OptionalIndex) u32 {
338 assert(oi != .none);
339 return @intFromEnum(oi);
340 }
341 };
342 pub const List = struct {
343 head: OptionalIndex,
344 tail: OptionalIndex,
345
346 pub const empty: List = .{ .head = .none, .tail = .none };
347
348 pub const SingleNode = struct { next: OptionalIndex };
349 pub const DoubleNode = struct { prev: OptionalIndex, next: OptionalIndex };
350 };
302351};
303352
304353/// Performs one `Operation`.
305pub fn operate(io: Io, operation: *Operation) Cancelable!void {
354pub fn operate(io: Io, operation: Operation) Cancelable!Operation.Result {
306355 return io.vtable.operate(io.userdata, operation);
307356}
308357
......@@ -312,116 +361,96 @@ pub fn operate(io: Io, operation: *Operation) Cancelable!void {
312361/// This is a low-level abstraction based on `Operation`. For a higher
313362/// level API that operates on `Future`, see `Select`.
314363pub const Batch = struct {
315 operations: []Operation,
316 ring: [*]u32,
317 user: struct {
318 submit_tail: RingIndex,
319 complete_head: RingIndex,
320 complete_tail: RingIndex,
321 },
322 impl: struct {
323 submit_head: RingIndex,
324 submit_tail: RingIndex,
325 complete_tail: RingIndex,
326 reserved: ?*anyopaque,
327 },
328
329 pub const RingIndex = enum(u32) {
330 _,
331
332 pub fn index(ri: RingIndex, len: u31) u31 {
333 const i = @intFromEnum(ri);
334 assert(i < @as(u32, len) * 2);
335 return @intCast(if (i < len) i else i - len);
336 }
337
338 pub fn prev(ri: RingIndex, len: u31) RingIndex {
339 const i = @intFromEnum(ri);
340 const double_len = @as(u32, len) * 2;
341 assert(i <= double_len);
342 return @enumFromInt((if (i > 0) i else double_len) - 1);
343 }
344
345 pub fn next(ri: RingIndex, len: u31) RingIndex {
346 const i = @intFromEnum(ri) + 1;
347 const double_len = @as(u32, len) * 2;
348 assert(i <= double_len);
349 return @enumFromInt(if (i < double_len) i else 0);
350 }
351 };
364 storage: []Operation.Storage,
365 unused: Operation.List,
366 submissions: Operation.List,
367 pending: Operation.List,
368 completions: Operation.List,
369 context: ?*anyopaque,
352370
353371 /// After calling this, it is safe to unconditionally defer a call to
354372 /// `cancel`.
355 pub fn init(operations: []Operation, ring: []u32) Batch {
356 const len: u31 = @intCast(operations.len);
357 assert(ring.len == len);
373 pub fn init(storage: []Operation.Storage) Batch {
374 var prev: Operation.OptionalIndex = .none;
375 for (storage, 0..) |*operation, index| {
376 operation.* = .{ .unused = .{ .prev = prev, .next = .fromIndex(index + 1) } };
377 prev = .fromIndex(index);
378 }
379 storage[storage.len - 1].unused.next = .none;
358380 return .{
359 .operations = operations,
360 .ring = ring.ptr,
361 .user = .{
362 .submit_tail = @enumFromInt(0),
363 .complete_head = @enumFromInt(0),
364 .complete_tail = @enumFromInt(0),
365 },
366 .impl = .{
367 .submit_head = @enumFromInt(0),
368 .submit_tail = @enumFromInt(0),
369 .complete_tail = @enumFromInt(0),
370 .reserved = null,
381 .storage = storage,
382 .unused = .{
383 .head = .fromIndex(0),
384 .tail = .fromIndex(storage.len - 1),
371385 },
386 .submissions = .empty,
387 .pending = .empty,
388 .completions = .empty,
389 .context = null,
372390 };
373391 }
374392
375 /// Adds `b.operations[operation]` to the list of submitted operations
376 /// that will be performed when `wait` is called.
377 pub fn add(b: *Batch, operation: usize) void {
378 const tail = b.user.submit_tail;
379 const len: u31 = @intCast(b.operations.len);
380 b.user.submit_tail = tail.next(len);
381 b.ring[0..len][tail.index(len)] = @intCast(operation);
382 }
383
384 fn flush(b: *Batch) void {
385 @atomicStore(RingIndex, &b.impl.submit_tail, b.user.submit_tail, .release);
386 }
393 /// Adds an operation to be performed at the next await call.
394 /// Returns the index that will be returned by `next` after the operation completes.
395 /// Asserts that no more than `storage.len` operations are active at a time.
396 pub fn add(b: *Batch, operation: Operation) u32 {
397 const index = b.unused.next;
398 b.addAt(index.toIndex(), operation);
399 return index;
400 }
401
402 /// Adds an operation to be performed at the next await call.
403 /// After the operation completes, `next` will return `index`.
404 /// Asserts that the operation at `index` is not active.
405 pub fn addAt(b: *Batch, index: u32, operation: Operation) void {
406 const storage = &b.storage[index];
407 const unused = storage.unused;
408 switch (unused.prev) {
409 .none => b.unused.head = .none,
410 else => |prev_index| b.storage[prev_index.toIndex()].unused.next = unused.next,
411 }
412 switch (unused.next) {
413 .none => b.unused.tail = .none,
414 else => |next_index| b.storage[next_index.toIndex()].unused.prev = unused.prev,
415 }
387416
388 /// Returns `operation` such that `b.operations[operation]` has completed.
389 /// Returns `null` when `wait` should be called.
390 pub fn next(b: *Batch) ?u32 {
391 const head = b.user.complete_head;
392 if (head == b.user.complete_tail) {
393 @branchHint(.unlikely);
394 b.flush();
395 const tail = @atomicLoad(RingIndex, &b.impl.complete_tail, .acquire);
396 if (head == tail) {
397 @branchHint(.unlikely);
398 return null;
399 }
400 assert(head != tail);
401 b.user.complete_tail = tail;
417 switch (b.submissions.tail) {
418 .none => b.submissions.head = .fromIndex(index),
419 else => |tail_index| b.storage[tail_index.toIndex()].submission.node.next = .fromIndex(index),
420 }
421 storage.* = .{ .submission = .{ .node = .{ .next = .none }, .operation = operation } };
422 b.submissions.tail = .fromIndex(index);
423 }
424
425 pub fn next(b: *Batch) ?struct { index: u32, result: Operation.Result } {
426 const index = b.completions.head;
427 if (index == .none) return null;
428 const storage = &b.storage[index.toIndex()];
429 const completion = storage.completion;
430 const next_index = completion.node.next;
431 b.completions.head = next_index;
432 if (next_index == .none) b.completions.tail = .none;
433
434 const tail_index = b.unused.tail;
435 switch (tail_index) {
436 .none => b.unused.head = index,
437 else => b.storage[tail_index.toIndex()].unused.next = index,
402438 }
403 const len: u31 = @intCast(b.operations.len);
404 b.user.complete_head = head.next(len);
405 return b.ring[0..len][head.index(len)];
439 storage.* = .{ .unused = .{ .prev = tail_index, .next = .none } };
440 b.unused.tail = index;
441 return .{ .index = index.toIndex(), .result = completion.result };
406442 }
407443
408 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
444 pub const AwaitAsyncError = Cancelable;
445 pub fn awaitAsync(b: *Batch, io: Io) AwaitAsyncError!void {
446 return io.vtable.batchAwaitAsync(io.userdata, b);
447 }
409448
410 /// Starts work on any submitted operations and returns when at least one has completeed.
411 ///
412 /// Returns `error.Timeout` if `timeout` expires first.
413 ///
414 /// Depending on the `Io` implementation, may allocate resources that are
415 /// freed with `cancel`, even if an error is returned.
416 pub fn wait(b: *Batch, io: Io, timeout: Timeout) WaitError!void {
417 return io.vtable.batchWait(io.userdata, b, timeout);
449 pub const AwaitConcurrentError = ConcurrentError || Cancelable || Timeout.Error;
450 pub fn awaitConcurrent(b: *Batch, io: Io, timeout: Timeout) AwaitConcurrentError!void {
451 return io.vtable.batchAwaitConcurrent(io.userdata, b, timeout);
418452 }
419453
420 /// Returns after all `operations` have completed. Operations which have not completed
421 /// after this function returns were successfully dropped and had no side effects.
422 ///
423 /// This function is idempotent with respect to itself and `wait`. It is
424 /// safe to unconditionally `defer` a call to this function after `init`.
425454 pub fn cancel(b: *Batch, io: Io) void {
426455 return io.vtable.batchCancel(io.userdata, b);
427456 }
lib/std/Io/File.zig+3-4
......@@ -559,12 +559,11 @@ pub const ReadStreamingError = error{EndOfStream} || Reader.Error;
559559/// See also:
560560/// * `reader`
561561pub fn readStreaming(file: File, io: Io, buffer: []const []u8) ReadStreamingError!usize {
562 var operation: Io.Operation = .{ .file_read_streaming = .{
562 const result = try io.operate(.{ .file_read_streaming = .{
563563 .file = file,
564564 .data = buffer,
565 } };
566 try io.operate(&operation);
567 return operation.file_read_streaming.status.result;
565 } });
566 return result.file_read_streaming;
568567}
569568
570569pub const ReadPositionalError = error{
lib/std/Io/File/MultiReader.zig+20-32
......@@ -22,8 +22,7 @@ pub const UnendingError = Allocator.Error || File.Reader.Error || Io.ConcurrentE
2222
2323/// Trailing:
2424/// * `contexts: [len]Context`
25/// * `ring: [len]u32`
26/// * `operations: [len]Io.Operation`
25/// * `storage: [len]Io.Operation.Storage`
2726pub const Streams = extern struct {
2827 len: u32,
2928
......@@ -33,17 +32,10 @@ pub const Streams = extern struct {
3332 return ptr[0..s.len];
3433 }
3534
36 pub fn ring(s: *Streams) []u32 {
35 pub fn storage(s: *Streams) []Io.Operation.Storage {
3736 const prev = contexts(s);
3837 const end = prev.ptr + prev.len;
39 const ptr: [*]u32 = @ptrFromInt(std.mem.alignForward(usize, @intFromPtr(end), @alignOf(u32)));
40 return ptr[0..s.len];
41 }
42
43 pub fn operations(s: *Streams) []Io.Operation {
44 const prev = ring(s);
45 const end = prev.ptr + prev.len;
46 const ptr: [*]Io.Operation = @ptrFromInt(std.mem.alignForward(usize, @intFromPtr(end), @alignOf(Io.Operation)));
38 const ptr: [*]Io.Operation.Storage = @ptrFromInt(std.mem.alignForward(usize, @intFromPtr(end), @alignOf(Io.Operation.Storage)));
4739 return ptr[0..s.len];
4840 }
4941};
......@@ -52,8 +44,7 @@ pub fn Buffer(comptime n: usize) type {
5244 return extern struct {
5345 len: u32,
5446 contexts: [n][@sizeOf(Context)]u8 align(@alignOf(Context)),
55 ring: [n]u32,
56 operations: [n][@sizeOf(Io.Operation)]u8 align(@alignOf(Io.Operation)),
47 storage: [n][@sizeOf(Io.Operation.Storage)]u8 align(@alignOf(Io.Operation.Storage)),
5748
5849 pub fn toStreams(b: *@This()) *Streams {
5950 b.len = n;
......@@ -86,25 +77,22 @@ pub fn init(mr: *MultiReader, gpa: Allocator, io: Io, streams: *Streams, files:
8677 .vec = .{&.{}},
8778 .err = null,
8879 };
89 const operations = streams.operations();
90 const ring = streams.ring();
9180 mr.* = .{
9281 .gpa = gpa,
9382 .streams = streams,
94 .batch = .init(operations, ring),
83 .batch = .init(streams.storage()),
9584 };
96 for (operations, contexts, files, 0..) |*op, *context, file, i| {
85 for (contexts, 0..) |*context, i| {
9786 const r = &context.fr.interface;
98 op.* = .{ .file_read_streaming = .{
99 .file = file,
100 .data = &context.vec,
101 } };
10287 rebaseGrowing(mr, context, 1) catch |err| {
10388 context.err = err;
10489 continue;
10590 };
10691 context.vec[0] = r.buffer;
107 mr.batch.add(i);
92 mr.batch.addAt(@intCast(i), .{ .file_read_streaming = .{
93 .file = context.fr.file,
94 .data = &context.vec,
95 } });
10896 }
10997}
11098
......@@ -204,7 +192,7 @@ fn fillUntimed(context: *Context, capacity: usize) Io.Reader.Error!void {
204192 };
205193}
206194
207pub const FillError = Io.Batch.WaitError || error{
195pub const FillError = Io.Batch.AwaitConcurrentError || error{
208196 /// `fill` was called when all streams already have failed or reached the
209197 /// end.
210198 EndOfStream,
......@@ -213,17 +201,15 @@ pub const FillError = Io.Batch.WaitError || error{
213201/// Wait until at least one stream receives more data.
214202pub fn fill(mr: *MultiReader, unused_capacity: usize, timeout: Io.Timeout) FillError!void {
215203 const contexts = mr.streams.contexts();
216 const operations = mr.streams.operations();
217204 const io = contexts[0].fr.io;
218205 var any_completed = false;
219206
220 try mr.batch.wait(io, timeout);
207 try mr.batch.awaitConcurrent(io, timeout);
221208
222 while (mr.batch.next()) |i| {
209 while (mr.batch.next()) |operation| {
223210 any_completed = true;
224 const context = &contexts[i];
225 const operation = &operations[i];
226 const n = operation.file_read_streaming.status.result catch |err| {
211 const context = &contexts[operation.index];
212 const n = operation.result.file_read_streaming catch |err| {
227213 context.err = err;
228214 continue;
229215 };
......@@ -237,15 +223,17 @@ pub fn fill(mr: *MultiReader, unused_capacity: usize, timeout: Io.Timeout) FillE
237223 assert(r.seek == 0);
238224 }
239225 context.vec[0] = r.buffer[r.end..];
240 operation.file_read_streaming.status = .{ .unstarted = {} };
241 mr.batch.add(i);
226 mr.batch.addAt(operation.index, .{ .file_read_streaming = .{
227 .file = context.fr.file,
228 .data = &context.vec,
229 } });
242230 }
243231
244232 if (!any_completed) return error.EndOfStream;
245233}
246234
247235/// Wait until all streams fail or reach the end.
248pub fn fillRemaining(mr: *MultiReader, timeout: Io.Timeout) Io.Batch.WaitError!void {
236pub fn fillRemaining(mr: *MultiReader, timeout: Io.Timeout) Io.Batch.AwaitConcurrentError!void {
249237 while (fill(mr, 1, timeout)) |_| {} else |err| switch (err) {
250238 error.EndOfStream => return,
251239 else => |e| return e,
lib/std/Io/Threaded.zig+347-222
......@@ -1617,7 +1617,8 @@ pub fn io(t: *Threaded) Io {
16171617 .futexWake = futexWake,
16181618
16191619 .operate = operate,
1620 .batchWait = batchWait,
1620 .batchAwaitAsync = batchAwaitAsync,
1621 .batchAwaitConcurrent = batchAwaitConcurrent,
16211622 .batchCancel = batchCancel,
16221623
16231624 .dirCreateDir = dirCreateDir,
......@@ -1780,7 +1781,8 @@ pub fn ioBasic(t: *Threaded) Io {
17801781 .futexWake = futexWake,
17811782
17821783 .operate = operate,
1783 .batchWait = batchWait,
1784 .batchAwaitAsync = batchAwaitAsync,
1785 .batchAwaitConcurrent = batchAwaitConcurrent,
17841786 .batchCancel = batchCancel,
17851787
17861788 .dirCreateDir = dirCreateDir,
......@@ -2483,85 +2485,227 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
24832485 Thread.futexWake(ptr, max_waiters);
24842486}
24852487
2486fn operate(userdata: ?*anyopaque, op: *Io.Operation) Io.Cancelable!void {
2488fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Operation.Result {
24872489 const t: *Threaded = @ptrCast(@alignCast(userdata));
2488 switch (op.*) {
2489 .noop => |*o| {
2490 _ = o.status.unstarted;
2491 o.status = .{ .result = {} };
2492 },
2493 .file_read_streaming => |*o| {
2494 _ = o.status.unstarted;
2495 o.status = .{ .result = fileReadStreaming(t, o.file, o.data) catch |err| switch (err) {
2490 switch (operation) {
2491 .file_read_streaming => |o| return .{
2492 .file_read_streaming = fileReadStreaming(t, o.file, o.data) catch |err| switch (err) {
24962493 error.Canceled => |e| return e,
24972494 else => |e| e,
2498 } };
2495 },
24992496 },
25002497 }
25012498}
25022499
2503fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.WaitError!void {
2500fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Batch.AwaitAsyncError!void {
25042501 const t: *Threaded = @ptrCast(@alignCast(userdata));
2505 if (is_windows) return batchWaitWindows(t, b, timeout);
2502 if (is_windows) {
2503 try batchAwaitWindows(b);
2504 const alertable_syscall = try AlertableSyscall.start();
2505 while (b.pending.head != .none and b.completions.head == .none) waitForApcOrAlert();
2506 alertable_syscall.finish();
2507 return;
2508 }
25062509 if (native_os == .wasi and !builtin.link_libc) @panic("TODO");
2507 const operations = b.operations;
2508 const len: u31 = @intCast(operations.len);
2509 const ring = b.ring[0..len];
2510 var submit_head = b.impl.submit_head;
2511 const submit_tail = b.user.submit_tail;
2512 b.impl.submit_tail = submit_tail;
2513 var complete_tail = b.impl.complete_tail;
2514 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
2515 var poll_i: u8 = 0;
2516 defer {
2517 for (map_buffer[0..poll_i]) |op| {
2518 submit_head = submit_head.prev(len);
2519 ring[submit_head.index(len)] = op;
2520 }
2521 b.impl.submit_head = submit_head;
2522 b.impl.complete_tail = complete_tail;
2523 b.user.complete_tail = complete_tail;
2510 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2511 var poll_len: u32 = 0;
2512 {
2513 var index = b.submissions.head;
2514 while (index != .none and poll_len < poll_buffer_len) {
2515 const submission = &b.storage[index.toIndex()].submission;
2516 switch (submission.operation) {
2517 .file_read_streaming => |o| {
2518 poll_buffer[poll_len] = .{ .fd = o.file.handle, .events = posix.POLL.IN, .revents = 0 };
2519 poll_len += 1;
2520 },
2521 }
2522 index = submission.node.next;
2523 }
2524 }
2525 switch (poll_len) {
2526 0 => return,
2527 1 => {},
2528 else => while (true) {
2529 const timeout_ms: i32 = t: {
2530 if (b.completions.head != .none) {
2531 // It is legal to call batchWait with already completed
2532 // operations in the ring. In such case, we need to avoid
2533 // blocking in the poll syscall, but we can still take this
2534 // opportunity to find additional ready operations.
2535 break :t 0;
2536 }
2537 const max_poll_ms = std.math.maxInt(i32);
2538 break :t max_poll_ms;
2539 };
2540 const syscall = try Syscall.start();
2541 const rc = posix.system.poll(&poll_buffer, poll_len, timeout_ms);
2542 syscall.finish();
2543 switch (posix.errno(rc)) {
2544 .SUCCESS => {
2545 if (rc == 0) {
2546 if (b.completions.head != .none) {
2547 // Since there are already completions available in the
2548 // queue, this is neither a timeout nor a case for
2549 // retrying.
2550 return;
2551 }
2552 continue;
2553 }
2554 var prev_index: Io.Operation.OptionalIndex = .none;
2555 var index = b.submissions.head;
2556 for (poll_buffer[0..poll_len]) |poll_entry| {
2557 const storage = &b.storage[index.toIndex()];
2558 const submission = &storage.submission;
2559 const next_index = submission.node.next;
2560 if (poll_entry.revents != 0) {
2561 const result = try operate(t, submission.operation);
2562
2563 switch (prev_index) {
2564 .none => b.submissions.head = next_index,
2565 else => b.storage[prev_index.toIndex()].submission.node.next = next_index,
2566 }
2567 if (next_index == .none) b.submissions.tail = prev_index;
2568
2569 switch (b.completions.tail) {
2570 .none => b.completions.head = index,
2571 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
2572 }
2573 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2574 b.completions.tail = index;
2575 } else prev_index = index;
2576 index = next_index;
2577 }
2578 assert(index == .none);
2579 return;
2580 },
2581 .INTR => continue,
2582 else => break,
2583 }
2584 },
2585 }
2586 {
2587 var tail_index = b.completions.tail;
2588 defer b.completions.tail = tail_index;
2589 var index = b.submissions.head;
2590 errdefer b.submissions.head = index;
2591 while (index != .none) {
2592 const storage = &b.storage[index.toIndex()];
2593 const submission = &storage.submission;
2594 const next_index = submission.node.next;
2595 const result = try operate(t, submission.operation);
2596
2597 switch (tail_index) {
2598 .none => b.completions.head = index,
2599 else => b.storage[tail_index.toIndex()].completion.node.next = index,
2600 }
2601 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2602 tail_index = index;
2603 index = next_index;
2604 }
2605 b.submissions = .{ .head = .none, .tail = .none };
25242606 }
2607}
2608
2609fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.AwaitConcurrentError!void {
2610 const t: *Threaded = @ptrCast(@alignCast(userdata));
2611 if (is_windows) {
2612 const deadline: ?Io.Clock.Timestamp = timeout.toDeadline(ioBasic(t)) catch |err| switch (err) {
2613 error.Unexpected => deadline: {
2614 recoverableOsBugDetected();
2615 break :deadline .{ .raw = .{ .nanoseconds = 0 }, .clock = .awake };
2616 },
2617 error.UnsupportedClock => |e| return e,
2618 };
2619 try batchAwaitWindows(b);
2620 while (b.pending.head != .none and b.completions.head == .none) {
2621 var delay_interval: windows.LARGE_INTEGER = interval: {
2622 const d = deadline orelse break :interval std.math.minInt(windows.LARGE_INTEGER);
2623 break :interval t.deadlineToWindowsInterval(d) catch |err| switch (err) {
2624 error.UnsupportedClock => |e| return e,
2625 error.Unexpected => {
2626 recoverableOsBugDetected();
2627 break :interval -1;
2628 },
2629 };
2630 };
2631 const alertable_syscall = try AlertableSyscall.start();
2632 const delay_rc = windows.ntdll.NtDelayExecution(windows.TRUE, &delay_interval);
2633 alertable_syscall.finish();
2634 switch (delay_rc) {
2635 .SUCCESS, .TIMEOUT => {
2636 // The thread woke due to the timeout. Although spurious
2637 // timeouts are OK, when no deadline is passed we must not
2638 // return `error.Timeout`.
2639 if (timeout != .none and b.completions.head == .none) return error.Timeout;
2640 },
2641 else => {},
2642 }
2643 }
2644 return;
2645 }
2646 if (native_os == .wasi and !builtin.link_libc) @panic("TODO");
25252647 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2526 while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) {
2527 const op = ring[submit_head.index(len)];
2528 const operation = &operations[op];
2529 switch (operation.*) {
2530 .noop => |*o| {
2531 _ = o.status.unstarted;
2532 o.status = .{ .result = {} };
2533 submitComplete(ring, &complete_tail, op);
2534 },
2535 .file_read_streaming => |*o| {
2536 _ = o.status.unstarted;
2537 if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable;
2538 poll_buffer[poll_i] = .{
2539 .fd = o.file.handle,
2540 .events = posix.POLL.IN,
2541 .revents = 0,
2648 var poll_storage: struct {
2649 gpa: std.mem.Allocator,
2650 b: *Io.Batch,
2651 slice: []posix.pollfd,
2652 len: u32,
2653
2654 fn add(storage: *@This(), file: Io.File, events: @FieldType(posix.pollfd, "events")) Io.ConcurrentError!void {
2655 const len = storage.len;
2656 if (len == poll_buffer_len) {
2657 const slice: []posix.pollfd = if (storage.b.context) |context|
2658 @as([*]posix.pollfd, @ptrCast(@alignCast(context)))[0..storage.b.storage.len]
2659 else allocation: {
2660 const allocation = storage.gpa.alloc(posix.pollfd, storage.b.storage.len) catch
2661 return error.ConcurrencyUnavailable;
2662 storage.b.context = allocation.ptr;
2663 break :allocation allocation;
25422664 };
2543 map_buffer[poll_i] = @intCast(op);
2544 poll_i += 1;
2545 },
2665 @memcpy(slice[0..poll_buffer_len], storage.slice);
2666 }
2667 storage.slice[len] = .{
2668 .fd = file.handle,
2669 .events = events,
2670 .revents = 0,
2671 };
2672 storage.len = len + 1;
2673 }
2674 } = .{ .gpa = t.allocator, .b = b, .slice = &poll_buffer, .len = 0 };
2675 {
2676 var index = b.submissions.head;
2677 while (index != .none) {
2678 const submission = &b.storage[index.toIndex()].submission;
2679 switch (submission.operation) {
2680 .file_read_streaming => |o| try poll_storage.add(o.file, posix.POLL.IN),
2681 }
2682 index = submission.node.next;
25462683 }
25472684 }
2548 switch (poll_i) {
2685 switch (poll_storage.len) {
25492686 0 => return,
25502687 1 => if (timeout == .none) {
2551 const op = map_buffer[0];
2552 try operate(t, &operations[op]);
2553 submitComplete(ring, &complete_tail, op);
2554 poll_i = 0;
2688 const index = b.submissions.head;
2689 const storage = &b.storage[index.toIndex()];
2690 const result = try operate(t, storage.submission.operation);
2691
2692 b.submissions = .{ .head = .none, .tail = .none };
2693
2694 switch (b.completions.tail) {
2695 .none => b.completions.head = index,
2696 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
2697 }
2698 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2699 b.completions.tail = index;
25552700 return;
25562701 },
25572702 else => {},
25582703 }
25592704 const t_io = ioBasic(t);
25602705 const deadline = timeout.toDeadline(t_io) catch return error.UnsupportedClock;
2561 const max_poll_ms = std.math.maxInt(i32);
25622706 while (true) {
25632707 const timeout_ms: i32 = t: {
2564 if (b.user.complete_head != complete_tail) {
2708 if (b.completions.head != .none) {
25652709 // It is legal to call batchWait with already completed
25662710 // operations in the ring. In such case, we need to avoid
25672711 // blocking in the poll syscall, but we can still take this
......@@ -2571,15 +2715,16 @@ fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.
25712715 const d = deadline orelse break :t -1;
25722716 const duration = d.durationFromNow(t_io) catch return error.UnsupportedClock;
25732717 if (duration.raw.nanoseconds <= 0) return error.Timeout;
2718 const max_poll_ms = std.math.maxInt(i32);
25742719 break :t @intCast(@min(max_poll_ms, duration.raw.toMilliseconds()));
25752720 };
25762721 const syscall = try Syscall.start();
2577 const rc = posix.system.poll(&poll_buffer, poll_i, timeout_ms);
2722 const rc = posix.system.poll(&poll_buffer, poll_storage.len, timeout_ms);
25782723 syscall.finish();
25792724 switch (posix.errno(rc)) {
25802725 .SUCCESS => {
25812726 if (rc == 0) {
2582 if (b.user.complete_head != complete_tail) {
2727 if (b.completions.head != .none) {
25832728 // Since there are already completions available in the
25842729 // queue, this is neither a timeout nor a case for
25852730 // retrying.
......@@ -2590,18 +2735,30 @@ fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.
25902735 if (deadline == null) continue;
25912736 return error.Timeout;
25922737 }
2593 while (poll_i != 0) {
2594 poll_i -= 1;
2595 const poll_fd = &poll_buffer[poll_i];
2596 const op = map_buffer[poll_i];
2597 if (poll_fd.revents == 0) {
2598 submit_head = submit_head.prev(len);
2599 ring[submit_head.index(len)] = op;
2600 } else {
2601 try operate(t, &operations[op]);
2602 submitComplete(ring, &complete_tail, op);
2603 }
2738 var prev_index: Io.Operation.OptionalIndex = .none;
2739 var index = b.submissions.head;
2740 for (poll_storage.slice[0..poll_storage.len]) |poll_entry| {
2741 const submission = &b.storage[index.toIndex()].submission;
2742 const next_index = submission.node.next;
2743 if (poll_entry.revents != 0) {
2744 const result = try operate(t, submission.operation);
2745
2746 switch (prev_index) {
2747 .none => b.submissions.head = next_index,
2748 else => b.storage[prev_index.toIndex()].submission.node.next = next_index,
2749 }
2750 if (next_index == .none) b.submissions.tail = prev_index;
2751
2752 switch (b.completions.tail) {
2753 .none => b.completions.head = index,
2754 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
2755 }
2756 b.completions.tail = index;
2757 b.storage[index.toIndex()] = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2758 } else prev_index = index;
2759 index = next_index;
26042760 }
2761 assert(index == .none);
26052762 return;
26062763 },
26072764 .INTR => continue,
......@@ -2610,166 +2767,126 @@ fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.
26102767 }
26112768}
26122769
2770const WindowsBatchPendingOperationContext = extern struct {
2771 file: windows.HANDLE,
2772 iosb: windows.IO_STATUS_BLOCK,
2773
2774 const Erased = [3]usize;
2775
2776 comptime {
2777 assert(@sizeOf(Erased) <= @sizeOf(WindowsBatchPendingOperationContext));
2778 }
2779
2780 fn toErased(context: *WindowsBatchPendingOperationContext) *Erased {
2781 return @ptrCast(context);
2782 }
2783
2784 fn fromErased(erased: *Erased) *WindowsBatchPendingOperationContext {
2785 return @ptrCast(erased);
2786 }
2787};
2788
26132789fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void {
26142790 const t: *Threaded = @ptrCast(@alignCast(userdata));
2615 const operations = b.operations;
2616 const len: u31 = @intCast(operations.len);
2617 const ring = b.ring[0..len];
2618 var submit_head = b.impl.submit_head;
2619 const submit_tail = b.user.submit_tail;
2620 b.impl.submit_tail = submit_tail;
2621 var complete_tail = b.impl.complete_tail;
2622 while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) {
2623 const op = ring[submit_head.index(len)];
2624 switch (operations[op]) {
2625 .noop => |*o| {
2626 _ = o.status.unstarted;
2627 o.status = .{ .result = {} };
2628 submitComplete(ring, &complete_tail, op);
2629 },
2630 .file_read_streaming => |*o| _ = o.status.unstarted,
2791 {
2792 var tail_index = b.unused.tail;
2793 defer b.unused.tail = tail_index;
2794 var index = b.submissions.head;
2795 errdefer b.submissions.head = index;
2796 while (index != .none) {
2797 const next_index = b.storage[index.toIndex()].submission.node.next;
2798 switch (tail_index) {
2799 .none => b.unused.head = index,
2800 else => b.storage[tail_index.toIndex()].unused.next = index,
2801 }
2802 b.storage[index.toIndex()] = .{ .unused = .{ .prev = tail_index, .next = .none } };
2803 tail_index = index;
2804 index = next_index;
26312805 }
2806 b.submissions = .{ .head = .none, .tail = .none };
26322807 }
26332808 if (is_windows) {
2634 // Iterate over pending and issue cancelations, then free the allocation for IO_STATUS_BLOCK
2635 if (b.impl.reserved) |reserved| {
2636 const gpa = t.allocator;
2637 const metadatas_ptr: [*]WinOpMetadata = @ptrCast(@alignCast(reserved));
2638 const metadatas = metadatas_ptr[0..b.operations.len];
2639 for (metadatas, 0..) |*metadata, op| {
2640 if (!metadata.pending) continue;
2641 const done = @atomicLoad(windows.NTSTATUS, &metadata.iosb.u.Status, .acquire) != .PENDING;
2642 if (done) continue;
2643 switch (operations[op]) {
2644 .noop => unreachable,
2645 .file_read_streaming => |*o| {
2646 _ = windows.ntdll.NtCancelIoFile(o.file.handle, &metadata.iosb);
2647 },
2648 }
2809 var index = b.pending.head;
2810 while (index != .none) {
2811 const pending = &b.storage[index.toIndex()].pending;
2812 const context: *WindowsBatchPendingOperationContext = .fromErased(&pending.context);
2813 _ = windows.ntdll.NtCancelIoFile(context.file, &context.iosb);
2814 index = pending.node.next;
2815 }
2816 while (b.pending.head != .none) waitForApcOrAlert();
2817 } else if (b.context) |context| {
2818 t.allocator.free(@as([*]posix.pollfd, @ptrCast(@alignCast(context)))[0..b.storage.len]);
2819 b.context = null;
2820 }
2821 assert(b.pending.head == .none);
2822}
2823
2824fn batchApc(apc_context: ?*anyopaque, iosb: *windows.IO_STATUS_BLOCK, _: windows.ULONG) callconv(.winapi) void {
2825 const b: *Io.Batch = @ptrCast(@alignCast(apc_context));
2826 const context: *WindowsBatchPendingOperationContext = @fieldParentPtr("iosb", iosb);
2827 const erased_context = context.toErased();
2828 const pending: *Io.Operation.Storage.Pending = @fieldParentPtr("context", erased_context);
2829 switch (pending.node.prev) {
2830 .none => b.pending.head = pending.node.next,
2831 else => |prev_index| b.storage[prev_index.toIndex()].pending.node.next = pending.node.next,
2832 }
2833 switch (pending.node.next) {
2834 .none => b.pending.tail = pending.node.prev,
2835 else => |next_index| b.storage[next_index.toIndex()].pending.node.prev = pending.node.prev,
2836 }
2837 const storage: *Io.Operation.Storage = @fieldParentPtr("pending", pending);
2838 const index = storage - b.storage.ptr;
2839 switch (iosb.u.Status) {
2840 .CANCELLED => {
2841 const tail_index = b.unused.tail;
2842 switch (tail_index) {
2843 .none => b.unused.head = .fromIndex(index),
2844 else => b.storage[tail_index.toIndex()].unused.next = .fromIndex(index),
26492845 }
2650 for (metadatas) |*metadata| {
2651 if (!metadata.pending) continue;
2652 while (@atomicLoad(windows.NTSTATUS, &metadata.iosb.u.Status, .acquire) == .PENDING) {
2653 waitForApcOrAlert();
2654 }
2846 storage.* = .{ .unused = .{ .prev = tail_index, .next = .none } };
2847 b.unused.tail = .fromIndex(index);
2848 },
2849 else => {
2850 switch (b.completions.tail) {
2851 .none => b.completions.head = .fromIndex(index),
2852 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = .fromIndex(index),
26552853 }
2656 gpa.free(metadatas);
2657 b.impl.reserved = null;
2658 }
2854 b.completions.tail = .fromIndex(index);
2855 const result: Io.Operation.Result = switch (pending.tag) {
2856 .file_read_streaming => .{ .file_read_streaming = ntReadFileResult(iosb) },
2857 };
2858 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2859 },
26592860 }
2660 b.impl.submit_head = submit_tail;
2661 b.impl.complete_tail = complete_tail;
2662 b.user.complete_tail = complete_tail;
26632861}
26642862
2665const WinOpMetadata = struct {
2666 iosb: windows.IO_STATUS_BLOCK,
2667 pending: bool,
2668};
2669
2670fn batchWaitWindows(t: *Threaded, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.WaitError!void {
2671 const operations = b.operations;
2672 const len: u31 = @intCast(operations.len);
2673 const ring = b.ring[0..len];
2674 var submit_head = b.impl.submit_head;
2675 const submit_tail = b.user.submit_tail;
2676 b.impl.submit_tail = submit_tail;
2677 var complete_tail = b.impl.complete_tail;
2678
2679 const metadatas_ptr: [*]WinOpMetadata = if (b.impl.reserved) |reserved| @ptrCast(@alignCast(reserved)) else a: {
2680 const gpa = t.allocator;
2681 const metadatas = gpa.alloc(WinOpMetadata, operations.len) catch return error.ConcurrencyUnavailable;
2682 b.impl.reserved = metadatas.ptr;
2683 @memset(metadatas, .{ .iosb = undefined, .pending = false });
2684 break :a metadatas.ptr;
2685 };
2686 const metadatas = metadatas_ptr[0..operations.len];
2687
2688 defer {
2689 b.impl.submit_head = submit_head;
2690 b.impl.complete_tail = complete_tail;
2691 b.user.complete_tail = complete_tail;
2692 }
2693
2694 while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) {
2695 const op = ring[submit_head.index(len)];
2696 const operation = &operations[op];
2697 const metadata = &metadatas[op];
2698 metadata.* = .{ .iosb = .{
2699 .u = .{ .Status = .PENDING },
2700 .Information = 0,
2701 }, .pending = false };
2702 switch (operation.*) {
2703 .noop => |*o| {
2704 _ = o.status.unstarted;
2705 o.status = .{ .result = {} };
2706 submitComplete(ring, &complete_tail, op);
2707 },
2708 .file_read_streaming => |*o| {
2709 _ = o.status.unstarted;
2710 try ntReadFile(o.file.handle, o.data, &metadata.iosb);
2711 if (@atomicLoad(windows.NTSTATUS, &metadata.iosb.u.Status, .acquire) == .PENDING) {
2712 o.status = .{ .pending = b };
2713 metadata.pending = true;
2714 } else {
2715 o.status = .{ .result = ntReadFileResult(&metadata.iosb) };
2716 submitComplete(ring, &complete_tail, op);
2717 }
2863fn batchAwaitWindows(b: *Io.Batch) Io.Cancelable!void {
2864 var index = b.submissions.head;
2865 errdefer b.submissions.head = index;
2866 while (index != .none) {
2867 const storage = &b.storage[index.toIndex()];
2868 const submission = storage.submission;
2869 errdefer storage.* = .{ .submission = submission };
2870 storage.* = .{ .pending = .{
2871 .node = .{ .prev = b.pending.tail, .next = .none },
2872 .tag = submission.operation,
2873 .context = undefined,
2874 } };
2875 const context: *WindowsBatchPendingOperationContext = .fromErased(&storage.pending.context);
2876 switch (submission.operation) {
2877 .file_read_streaming => |o| {
2878 context.file = o.file.handle;
2879 try ntReadFile(o.file.handle, o.data, &batchApc, b, &context.iosb);
27182880 },
27192881 }
2720 }
2721
2722 const deadline: ?Io.Clock.Timestamp = timeout.toDeadline(ioBasic(t)) catch |err| switch (err) {
2723 error.Unexpected => deadline: {
2724 recoverableOsBugDetected();
2725 break :deadline .{ .raw = .{ .nanoseconds = 0 }, .clock = .awake };
2726 },
2727 error.UnsupportedClock => |e| return e,
2728 };
2729
2730 while (true) {
2731 var any_pending = false;
2732 for (metadatas, 0..) |*metadata, op_usize| {
2733 if (!metadata.pending) continue;
2734 any_pending = true;
2735 const op: u31 = @intCast(op_usize);
2736 const done = @atomicLoad(windows.NTSTATUS, &metadata.iosb.u.Status, .acquire) != .PENDING;
2737 switch (operations[op]) {
2738 .noop => unreachable,
2739 .file_read_streaming => |*o| {
2740 assert(o.status.pending == b);
2741 if (!done) continue;
2742 o.status = .{ .result = ntReadFileResult(&metadata.iosb) };
2743 },
2744 }
2745 metadata.pending = false;
2746 submitComplete(ring, &complete_tail, op);
2747 }
2748 if (b.user.complete_head != complete_tail) return;
2749 if (!any_pending) return;
2750 var delay_interval: windows.LARGE_INTEGER = interval: {
2751 const d = deadline orelse break :interval std.math.minInt(windows.LARGE_INTEGER);
2752 break :interval t.deadlineToWindowsInterval(d) catch |err| switch (err) {
2753 error.UnsupportedClock => |e| return e,
2754 error.Unexpected => {
2755 recoverableOsBugDetected();
2756 break :interval -1;
2757 },
2758 };
2759 };
2760 const alertable_syscall = try AlertableSyscall.start();
2761 const delay_rc = windows.ntdll.NtDelayExecution(windows.TRUE, &delay_interval);
2762 alertable_syscall.finish();
2763 switch (delay_rc) {
2764 .SUCCESS, .TIMEOUT => {
2765 // The thread woke due to the timeout. Although spurious
2766 // timeouts are OK, when no deadline is passed we must not
2767 // return `error.Timeout`.
2768 if (timeout != .none) return error.Timeout;
2769 },
2770 else => {},
2882 switch (b.pending.tail) {
2883 .none => b.pending.head = index,
2884 else => |tail_index| b.storage[tail_index.toIndex()].pending.node.next = index,
27712885 }
2886 b.pending.tail = index;
2887 index = submission.node.next;
27722888 }
2889 b.submissions = .{ .head = .none, .tail = .none };
27732890}
27742891
27752892fn submitComplete(ring: []u32, complete_tail: *Io.Batch.RingIndex, op: u32) void {
......@@ -8701,7 +8818,7 @@ fn fileReadStreamingWindows(file: File, data: []const []u8) File.ReadStreamingEr
87018818 .u = .{ .Status = .PENDING },
87028819 .Information = 0,
87038820 };
8704 try ntReadFile(file.handle, data, &io_status_block);
8821 try ntReadFile(file.handle, data, &noopApc, null, &io_status_block);
87058822 while (@atomicLoad(windows.NTSTATUS, &io_status_block.u.Status, .acquire) == .PENDING) {
87068823 // Once we get here we must not return from the function until the
87078824 // operation completes, thereby releasing reference to io_status_block.
......@@ -8736,12 +8853,20 @@ fn ntReadFileResult(io_status_block: *const windows.IO_STATUS_BLOCK) !usize {
87368853 }
87378854}
87388855
8739fn ntReadFile(handle: windows.HANDLE, data: []const []u8, iosb: *windows.IO_STATUS_BLOCK) Io.Cancelable!void {
8856fn ntReadFile(
8857 handle: windows.HANDLE,
8858 data: []const []u8,
8859 apcRoutine: ?*const windows.IO_APC_ROUTINE,
8860 apc_context: ?*anyopaque,
8861 iosb: *windows.IO_STATUS_BLOCK,
8862) Io.Cancelable!void {
87408863 var index: usize = 0;
87418864 while (index < data.len and data[index].len == 0) index += 1;
87428865 if (index == data.len) {
8743 iosb.u.Status = .SUCCESS;
8744 iosb.Information = 0;
8866 iosb.* = .{ .u = .{ .Status = .SUCCESS }, .Information = 0 };
8867 if (apcRoutine) |routine| if (routine != &noopApc) {
8868 _ = windows.ntdll.NtQueueApcThread(windows.current_process, routine, apc_context, iosb, null);
8869 };
87458870 return;
87468871 }
87478872 const buffer = data[index];
......@@ -8750,8 +8875,8 @@ fn ntReadFile(handle: windows.HANDLE, data: []const []u8, iosb: *windows.IO_STAT
87508875 while (true) switch (windows.ntdll.NtReadFile(
87518876 handle,
87528877 null, // event
8753 noopApc, // apc callback
8754 null, // apc context
8878 apcRoutine,
8879 apc_context,
87558880 iosb,
87568881 buffer.ptr,
87578882 @min(std.math.maxInt(u32), buffer.len),