authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-08 14:10:31-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-09 20:47:24-08:00
logd776ebc913aa2b4fb7f4ff6783cd5b67d009f1c5
tree4a141e28a0d4aae19a5d9b4ffcb95bff985024f0
parent0deaf9957c34eceecd3cb7b0033e447ef94addc5

std.Io.Threaded.operate: handle cancelation and poll errors


1 files changed, 37 insertions(+), 12 deletions(-)

lib/std/Io/Threaded.zig+37-12
...@@ -1179,6 +1179,7 @@ const Syscall = struct {...@@ -1179,6 +1179,7 @@ const Syscall = struct {
11791179
1180const max_iovecs_len = 8;1180const max_iovecs_len = 8;
1181const splat_buffer_size = 64;1181const splat_buffer_size = 64;
1182const poll_buffer_len = 100;
1182const default_PATH = "/usr/local/bin:/bin/:/usr/bin";1183const default_PATH = "/usr/local/bin:/bin/:/usr/bin";
11831184
1184comptime {1185comptime {
...@@ -2274,15 +2275,13 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {...@@ -2274,15 +2275,13 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
22742275
2275 if (is_windows) @panic("TODO");2276 if (is_windows) @panic("TODO");
22762277
2277 var poll_buffer: [100]posix.pollfd = undefined;2278 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2278 var map_buffer: [poll_buffer.len]u8 = undefined; // poll_buffer index to operations index2279 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
2279 var poll_i: usize = 0;2280 var poll_i: usize = 0;
22802281
2281 // Put all the file reads with nonblocking enabled into the poll set.2282 // Put all the file reads with nonblocking enabled into the poll set.
2282 if (operations.len > poll_buffer.len) @panic("TODO");2283 if (operations.len > poll_buffer.len) @panic("TODO");
22832284
2284 // TODO if any operation is canceled, cancel the rest
2285
2286 for (operations, 0..) |*operation, operation_index| switch (operation.*) {2285 for (operations, 0..) |*operation, operation_index| switch (operation.*) {
2287 .noop => continue,2286 .noop => continue,
2288 .file_read_streaming => |*o| {2287 .file_read_streaming => |*o| {
...@@ -2296,7 +2295,13 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {...@@ -2296,7 +2295,13 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
2296 map_buffer[poll_i] = @intCast(operation_index);2295 map_buffer[poll_i] = @intCast(operation_index);
2297 poll_i += 1;2296 poll_i += 1;
2298 } else {2297 } else {
2299 o.result = fileReadStreaming(o.file, o.data);2298 o.result = fileReadStreaming(o.file, o.data) catch |err| switch (err) {
2299 error.Canceled => {
2300 setOperationsCanceled(operations[operation_index..]);
2301 return;
2302 },
2303 else => err,
2304 };
2300 }2305 }
2301 },2306 },
2302 };2307 };
...@@ -2309,12 +2314,7 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {...@@ -2309,12 +2314,7 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
2309 while (true) {2314 while (true) {
2310 const syscall = Syscall.start() catch |err| switch (err) {2315 const syscall = Syscall.start() catch |err| switch (err) {
2311 error.Canceled => {2316 error.Canceled => {
2312 for (map_buffer[0..poll_i]) |operation_index| {2317 setAllOperationsError(operations, map_buffer[0..poll_i], error.Canceled);
2313 switch (operations[operation_index]) {
2314 .noop => unreachable,
2315 inline else => |*o| o.result = error.Canceled,
2316 }
2317 }
2318 return;2318 return;
2319 },2319 },
2320 };2320 };
...@@ -2329,7 +2329,14 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {...@@ -2329,7 +2329,14 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
2329 break;2329 break;
2330 },2330 },
2331 .INTR => continue,2331 .INTR => continue,
2332 else => @panic("TODO handle unexpected error from poll()"),2332 .NOMEM => {
2333 setAllOperationsError(operations, map_buffer[0..poll_i], error.SystemResources);
2334 return;
2335 },
2336 else => {
2337 setAllOperationsError(operations, map_buffer[0..poll_i], error.Unexpected);
2338 return;
2339 },
2333 }2340 }
2334 }2341 }
23352342
...@@ -2344,6 +2351,24 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {...@@ -2344,6 +2351,24 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
2344 }2351 }
2345}2352}
23462353
2354fn setAllOperationsError(
2355 operations: []Io.Operation,
2356 map: []const u8,
2357 err: error{ Canceled, SystemResources, Unexpected },
2358) void {
2359 for (map) |operation_index| switch (operations[operation_index]) {
2360 .noop => unreachable,
2361 inline else => |*o| o.result = err,
2362 };
2363}
2364
2365fn setOperationsCanceled(operations: []Io.Operation) void {
2366 for (operations) |*op| switch (op.*) {
2367 .noop => unreachable,
2368 inline else => |*o| o.result = error.Canceled,
2369 };
2370}
2371
2347const dirCreateDir = switch (native_os) {2372const dirCreateDir = switch (native_os) {
2348 .windows => dirCreateDirWindows,2373 .windows => dirCreateDirWindows,
2349 .wasi => dirCreateDirWasi,2374 .wasi => dirCreateDirWasi,