If concurrency is false, error.ConcurrencyUnavailable is unreachable.
fn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io.ConcurrentError || Io.Cancelable)!voidfn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io.ConcurrentError || Io.Cancelable)!void {
var index = b.submitted.head;
errdefer b.submitted.head = index;
while (index != .none) {
const storage = &b.storage[index.toIndex()];
const submission = storage.submission;
storage.* = .{ .pending = .{
.node = .{ .prev = b.pending.tail, .next = .none },
.tag = submission.operation,
.userdata = undefined,
} };
switch (b.pending.tail) {
.none => b.pending.head = index,
else => |tail_index| b.storage[tail_index.toIndex()].pending.node.next = index,
}
b.pending.tail = index;
const operation_userdata: *WindowsBatchOperationUserdata = .fromErased(&storage.pending.userdata);
errdefer {
operation_userdata.iosb = .{ .u = .{ .Status = .CANCELLED }, .Information = undefined };
batchApc(b, &operation_userdata.iosb, 0);
}
switch (submission.operation) {
.file_read_streaming => |o| o: {
var data_index: usize = 0;
while (o.data.len - data_index != 0 and o.data[data_index].len == 0) data_index += 1;
if (o.data.len - data_index == 0) {
operation_userdata.iosb = .{ .u = .{ .Status = .SUCCESS }, .Information = 0 };
batchApc(b, &operation_userdata.iosb, 0);
break :o;
}
const buffer = o.data[data_index];
const short_buffer_len = std.math.lossyCast(u32, buffer.len);
if (o.file.flags.nonblocking) {
operation_userdata.file = o.file.handle;
switch (windows.ntdll.NtReadFile(
o.file.handle,
null, // event
&batchApc,
b,
&operation_userdata.iosb,
buffer.ptr,
short_buffer_len,
null, // byte offset
null, // key
)) {
.PENDING, .SUCCESS => {},
.CANCELLED => unreachable,
else => |status| {
operation_userdata.iosb.u.Status = status;
batchApc(b, &operation_userdata.iosb, 0);
},
}
} else {
if (concurrency) return error.ConcurrencyUnavailable;
const syscall: Syscall = try .start();
while (true) switch (windows.ntdll.NtReadFile(
o.file.handle,
null, // event
null, // APC routine
null, // APC context
&operation_userdata.iosb,
buffer.ptr,
short_buffer_len,
null, // byte offset
null, // key
)) {
.PENDING => unreachable, // unrecoverable: wrong File nonblocking flag
.CANCELLED => {
try syscall.checkCancel();
continue;
},
else => |status| {
syscall.finish();
operation_userdata.iosb.u.Status = status;
batchApc(b, &operation_userdata.iosb, 0);
break;
},
};
}
},
.file_write_streaming => |o| o: {
const buffer = windowsWriteBuffer(o.header, o.data, o.splat);
if (buffer.len == 0) {
operation_userdata.iosb = .{ .u = .{ .Status = .SUCCESS }, .Information = 0 };
batchApc(b, &operation_userdata.iosb, 0);
break :o;
}
if (o.file.flags.nonblocking) {
operation_userdata.file = o.file.handle;
switch (windows.ntdll.NtWriteFile(
o.file.handle,
null, // event
&batchApc,
b,
&operation_userdata.iosb,
buffer.ptr,
@intCast(buffer.len),
null, // byte offset
null, // key
)) {
.PENDING, .SUCCESS => {},
.CANCELLED => unreachable,
else => |status| {
operation_userdata.iosb.u.Status = status;
batchApc(b, &operation_userdata.iosb, 0);
},
}
} else {
if (concurrency) return error.ConcurrencyUnavailable;
const syscall: Syscall = try .start();
while (true) switch (windows.ntdll.NtWriteFile(
o.file.handle,
null, // event
null, // APC routine
null, // APC context
&operation_userdata.iosb,
buffer.ptr,
@intCast(buffer.len),
null, // byte offset
null, // key
)) {
.PENDING => unreachable, // unrecoverable: wrong File nonblocking flag
.CANCELLED => {
try syscall.checkCancel();
continue;
},
else => |status| {
syscall.finish();
operation_userdata.iosb.u.Status = status;
batchApc(b, &operation_userdata.iosb, 0);
break;
},
};
}
},
.device_io_control => |o| {
const NtControlFile = switch (o.code.DeviceType) {
.FILE_SYSTEM, .NAMED_PIPE => &windows.ntdll.NtFsControlFile,
else => &windows.ntdll.NtDeviceIoControlFile,
};
if (o.file.flags.nonblocking) {
operation_userdata.file = o.file.handle;
switch (NtControlFile(
o.file.handle,
null, // event
&batchApc,
b,
&operation_userdata.iosb,
o.code,
if (o.in.len > 0) o.in.ptr else null,
@intCast(o.in.len),
if (o.out.len > 0) o.out.ptr else null,
@intCast(o.out.len),
)) {
.PENDING, .SUCCESS => {},
.CANCELLED => unreachable,
else => |status| {
operation_userdata.iosb.u.Status = status;
batchApc(b, &operation_userdata.iosb, 0);
},
}
} else {
if (concurrency) return error.ConcurrencyUnavailable;
const syscall: Syscall = try .start();
while (true) switch (NtControlFile(
o.file.handle,
null, // event
null, // APC routine
null, // APC context
&operation_userdata.iosb,
o.code,
if (o.in.len > 0) o.in.ptr else null,
@intCast(o.in.len),
if (o.out.len > 0) o.out.ptr else null,
@intCast(o.out.len),
)) {
.PENDING => unreachable, // unrecoverable: wrong File nonblocking flag
.CANCELLED => {
try syscall.checkCancel();
continue;
},
else => |status| {
syscall.finish();
operation_userdata.iosb.u.Status = status;
batchApc(b, &operation_userdata.iosb, 0);
break;
},
};
}
},
.net_receive => |*o| {
// TODO integrate with overlapped I/O or equivalent to avoid this error
if (concurrency) return error.ConcurrencyUnavailable;
batchCompleteBlockingWindows(b, operation_userdata, .{
.net_receive = netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.flags),
});
},
.net_read => |*o| {
// TODO integrate with overlapped I/O or equivalent to avoid this error
if (concurrency) return error.ConcurrencyUnavailable;
batchCompleteBlockingWindows(b, operation_userdata, .{
.net_read = netRead(o.socket_handle, o.data) catch |err| switch (err) {
error.Canceled => |e| return e,
else => |e| e,
},
});
},
}
index = submission.node.next;
}
b.submitted = .{ .head = .none, .tail = .none };
}