fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.AwaitConcurrentError!void {
const t: *Threaded = @ptrCast(@alignCast(userdata));
if (is_windows) {
const deadline: ?Io.Clock.Timestamp = timeout.toTimestamp(io(t));
try batchDrainSubmittedWindows(t, b, true);
while (b.pending.head != .none and b.completed.head == .none) {
var delay_interval: windows.LARGE_INTEGER = interval: {
const d = deadline orelse break :interval std.math.minInt(windows.LARGE_INTEGER);
break :interval timeoutToWindowsInterval(.{ .deadline = d }).?;
};
const alertable_syscall = try AlertableSyscall.start();
const delay_rc = windows.ntdll.NtDelayExecution(.TRUE, &delay_interval);
alertable_syscall.finish();
switch (delay_rc) {
.SUCCESS, .TIMEOUT => {
// The thread woke due to the timeout. Although spurious
// timeouts are OK, when no deadline is passed we must not
// return `error.Timeout`.
if (timeout != .none and b.completed.head == .none) return error.Timeout;
},
else => {},
}
}
return;
}
if (native_os == .wasi) {
// TODO call poll_oneoff
return error.ConcurrencyUnavailable;
}
if (!have_poll) return error.ConcurrencyUnavailable;
var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
var poll_storage: struct {
gpa: Allocator,
batch: *Io.Batch,
slice: []posix.pollfd,
len: u32,
fn add(storage: *@This(), fd: File.Handle, events: @FieldType(posix.pollfd, "events")) Io.ConcurrentError!void {
const len = storage.len;
if (len == poll_buffer_len) {
const slice: []posix.pollfd = if (storage.batch.userdata) |batch_userdata|
@as([*]posix.pollfd, @ptrCast(@alignCast(batch_userdata)))[0..storage.batch.storage.len]
else allocation: {
const allocation = storage.gpa.alloc(posix.pollfd, storage.batch.storage.len) catch
return error.ConcurrencyUnavailable;
storage.batch.userdata = allocation.ptr;
break :allocation allocation;
};
@memcpy(slice[0..poll_buffer_len], storage.slice);
storage.slice = slice;
}
storage.slice[len] = .{
.fd = fd,
.events = events,
};
storage.len = len + 1;
}
} = .{ .gpa = t.allocator, .batch = b, .slice = &poll_buffer, .len = 0 };
{
var index = b.submitted.head;
while (index != .none) {
const storage = &b.storage[index.toIndex()];
const submission = storage.submission;
switch (submission.operation) {
.file_read_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.ERR),
.file_write_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.OUT | posix.POLL.ERR),
.device_io_control => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.OUT | posix.POLL.ERR),
.net_receive => |*o| nb: {
var data_i: usize = 0;
const result: Io.Operation.Result = .{ .net_receive = for (o.message_buffer, 0..) |*msg, msg_i| {
const remaining_data_buffer = o.data_buffer[data_i..];
netReceivePosix(o.socket_handle, msg, remaining_data_buffer, o.flags, true) catch |err| switch (err) {
error.Canceled => |e| return e,
error.WouldBlock => {
if (msg_i != 0) break .{ null, msg_i };
try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR);
break :nb;
},
else => |e| break .{ e, 0 },
};
data_i += msg.data.len;
} else .{ null, o.message_buffer.len } };
switch (b.completed.tail) {
.none => b.completed.head = index,
else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
}
storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
b.completed.tail = index;
},
.net_send => |*o| nb: {
const result: Io.Operation.Result = .{
.net_send = o: {
const send_err, const sent = netSendPosix(t, o.socket_handle, o.messages, o.flags, true);
if (send_err) |err| switch (err) {
error.Canceled => |e| if (sent == 0) {
return e;
} else {
// Leave the `error.Canceled` for later, but don't try to send any more messages.
recancelInner();
break :o .{ null, sent };
},
error.WouldBlock => {
if (sent != 0) break :o .{ null, sent };
try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR);
break :nb;
},
else => |e| break :o .{ e, sent },
};
break :o .{ null, sent };
},
};
switch (b.completed.tail) {
.none => b.completed.head = index,
else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
}
storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
b.completed.tail = index;
},
.net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),
.net_write => |o| try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR),
}
index = submission.node.next;
}
}
switch (poll_storage.len) {
0 => return,
1 => if (timeout == .none and b.completed.head == .none) {
const index = b.submitted.head;
const storage = &b.storage[index.toIndex()];
const result = try operate(t, storage.submission.operation);
b.submitted = .{ .head = .none, .tail = .none };
switch (b.completed.tail) {
.none => b.completed.head = index,
else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
}
storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
b.completed.tail = index;
return;
},
else => {},
}
const t_io = io(t);
const deadline = timeout.toTimestamp(t_io);
while (true) {
const timeout_ms: i32 = t: {
if (b.completed.head != .none) {
// It is legal to call batchWait with already completed
// operations in the ring. In such case, we need to avoid
// blocking in the poll syscall, but we can still take this
// opportunity to find additional ready operations.
break :t 0;
}
const d = deadline orelse break :t -1;
const duration = d.durationFromNow(t_io);
break :t @min(@max(0, duration.raw.toMilliseconds()), std.math.maxInt(i32));
};
const syscall = try Syscall.start();
const rc = posix.system.poll(poll_storage.slice.ptr, poll_storage.len, timeout_ms);
syscall.finish();
switch (posix.errno(rc)) {
.SUCCESS => {
if (rc == 0) {
if (b.completed.head != .none) {
// Since there are already completions available in the
// queue, this is neither a timeout nor a case for
// retrying.
return;
}
// Although spurious timeouts are OK, when no deadline is
// passed we must not return `error.Timeout`.
if (deadline == null) continue;
return error.Timeout;
}
var prev_index: Io.Operation.OptionalIndex = .none;
var index = b.submitted.head;
for (poll_storage.slice[0..poll_storage.len]) |poll_entry| {
const submission = &b.storage[index.toIndex()].submission;
const next_index = submission.node.next;
if (poll_entry.revents != 0) {
const result = try operate(t, submission.operation);
switch (prev_index) {
.none => b.submitted.head = next_index,
else => b.storage[prev_index.toIndex()].submission.node.next = next_index,
}
if (next_index == .none) b.submitted.tail = prev_index;
switch (b.completed.tail) {
.none => b.completed.head = index,
else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
}
b.completed.tail = index;
b.storage[index.toIndex()] = .{ .completion = .{
.node = .{ .next = .none },
.result = result,
} };
} else prev_index = index;
index = next_index;
}
assert(index == .none);
return;
},
.INTR => continue,
else => return error.ConcurrencyUnavailable,
}
}
}
