| ... | @@ -2277,81 +2277,84 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void { | ... | @@ -2277,81 +2277,84 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void { |
| 2277 | | 2277 | |
| 2278 | var poll_buffer: [poll_buffer_len]posix.pollfd = undefined; | 2278 | var poll_buffer: [poll_buffer_len]posix.pollfd = undefined; |
| 2279 | var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index | 2279 | var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index |
| 2280 | var poll_i: usize = 0; | 2280 | var operation_index: usize = 0; |
| 2281 | | 2281 | |
| 2282 | // Put all the file reads with nonblocking enabled into the poll set. | 2282 | while (operation_index < operations.len) { |
| 2283 | if (operations.len > poll_buffer.len) @panic("TODO"); | 2283 | var poll_i: usize = 0; |
| 2284 | | 2284 | while (operation_index < operations.len) : (operation_index += 1) { |
| 2285 | for (operations, 0..) |*operation, operation_index| switch (operation.*) { | 2285 | switch (operations[operation_index]) { |
| 2286 | .noop => continue, | 2286 | .noop => continue, |
| 2287 | .file_read_streaming => |*o| { | 2287 | .file_read_streaming => |*o| { |
| 2288 | if (o.nonblocking) { | 2288 | if (o.nonblocking) { |
| 2289 | o.result = error.WouldBlock; | 2289 | o.result = error.WouldBlock; |
| 2290 | poll_buffer[poll_i] = .{ | 2290 | poll_buffer[poll_i] = .{ |
| 2291 | .fd = o.file.handle, | 2291 | .fd = o.file.handle, |
| 2292 | .events = posix.POLL.IN, | 2292 | .events = posix.POLL.IN, |
| 2293 | .revents = undefined, | 2293 | .revents = 0, |
| 2294 | }; | 2294 | }; |
| 2295 | map_buffer[poll_i] = @intCast(operation_index); | 2295 | if (map_buffer.len - poll_i == 0) break; |
| 2296 | poll_i += 1; | 2296 | map_buffer[poll_i] = @intCast(operation_index); |
| 2297 | } else { | 2297 | poll_i += 1; |
| 2298 | o.result = fileReadStreaming(o.file, o.data) catch |err| switch (err) { | 2298 | } else { |
| 2299 | error.Canceled => { | 2299 | o.result = fileReadStreaming(o.file, o.data) catch |err| switch (err) { |
| 2300 | setOperationsCanceled(operations[operation_index..]); | 2300 | error.Canceled => { |
| 2301 | return; | 2301 | setOperationsError(operations[operation_index..], error.Canceled); |
| 2302 | }, | 2302 | return; |
| 2303 | else => err, | 2303 | }, |
| 2304 | }; | 2304 | else => err, |
| | 2305 | }; |
| | 2306 | } |
| | 2307 | }, |
| 2305 | } | 2308 | } |
| 2306 | }, | 2309 | } |
| 2307 | }; | | |
| 2308 | | | |
| 2309 | if (poll_i == 0) { | | |
| 2310 | @branchHint(.likely); | | |
| 2311 | return; | | |
| 2312 | } | | |
| 2313 | | 2310 | |
| 2314 | while (true) { | 2311 | if (poll_i == 0) { |
| 2315 | const syscall = Syscall.start() catch |err| switch (err) { | 2312 | @branchHint(.likely); |
| 2316 | error.Canceled => { | 2313 | return; |
| 2317 | setAllOperationsError(operations, map_buffer[0..poll_i], error.Canceled); | | |
| 2318 | return; | | |
| 2319 | }, | | |
| 2320 | }; | | |
| 2321 | const poll_rc = posix.system.poll(&poll_buffer, poll_i, -1); | | |
| 2322 | syscall.finish(); | | |
| 2323 | switch (posix.errno(poll_rc)) { | | |
| 2324 | .SUCCESS => { | | |
| 2325 | if (poll_rc == 0) { | | |
| 2326 | // Spurious timeout; handle same as INTR. | | |
| 2327 | continue; | | |
| 2328 | } | | |
| 2329 | break; | | |
| 2330 | }, | | |
| 2331 | .INTR => continue, | | |
| 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 | }, | | |
| 2340 | } | 2314 | } |
| 2341 | } | | |
| 2342 | | 2315 | |
| 2343 | for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, operation_index| { | 2316 | while (true) { |
| 2344 | if (poll_fd.revents == 0) continue; | 2317 | const syscall = Syscall.start() catch |err| switch (err) { |
| 2345 | switch (operations[operation_index]) { | 2318 | error.Canceled => { |
| 2346 | .noop => unreachable, | 2319 | setPollOperationsError(operations, map_buffer[0..poll_i], error.Canceled); |
| 2347 | .file_read_streaming => |*o| { | 2320 | setOperationsError(operations[operation_index..], error.Canceled); |
| 2348 | o.result = fileReadStreaming(o.file, o.data); | 2321 | return; |
| 2349 | }, | 2322 | }, |
| | 2323 | }; |
| | 2324 | const poll_rc = posix.system.poll(&poll_buffer, poll_i, -1); |
| | 2325 | syscall.finish(); |
| | 2326 | switch (posix.errno(poll_rc)) { |
| | 2327 | .SUCCESS => { |
| | 2328 | if (poll_rc == 0) { |
| | 2329 | // Spurious timeout; handle same as INTR. |
| | 2330 | continue; |
| | 2331 | } |
| | 2332 | for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| { |
| | 2333 | if (poll_fd.revents == 0) continue; |
| | 2334 | switch (operations[i]) { |
| | 2335 | .noop => unreachable, |
| | 2336 | .file_read_streaming => |*o| { |
| | 2337 | o.result = fileReadStreaming(o.file, o.data); |
| | 2338 | }, |
| | 2339 | } |
| | 2340 | } |
| | 2341 | break; |
| | 2342 | }, |
| | 2343 | .INTR => continue, |
| | 2344 | .NOMEM => { |
| | 2345 | setPollOperationsError(operations, map_buffer[0..poll_i], error.SystemResources); |
| | 2346 | break; |
| | 2347 | }, |
| | 2348 | else => { |
| | 2349 | setPollOperationsError(operations, map_buffer[0..poll_i], error.Unexpected); |
| | 2350 | break; |
| | 2351 | }, |
| | 2352 | } |
| 2350 | } | 2353 | } |
| 2351 | } | 2354 | } |
| 2352 | } | 2355 | } |
| 2353 | | 2356 | |
| 2354 | fn setAllOperationsError( | 2357 | fn setPollOperationsError( |
| 2355 | operations: []Io.Operation, | 2358 | operations: []Io.Operation, |
| 2356 | map: []const u8, | 2359 | map: []const u8, |
| 2357 | err: error{ Canceled, SystemResources, Unexpected }, | 2360 | err: error{ Canceled, SystemResources, Unexpected }, |
| ... | @@ -2362,10 +2365,10 @@ fn setAllOperationsError( | ... | @@ -2362,10 +2365,10 @@ fn setAllOperationsError( |
| 2362 | }; | 2365 | }; |
| 2363 | } | 2366 | } |
| 2364 | | 2367 | |
| 2365 | fn setOperationsCanceled(operations: []Io.Operation) void { | 2368 | fn setOperationsError(operations: []Io.Operation, err: error{ Canceled, SystemResources, Unexpected }) void { |
| 2366 | for (operations) |*op| switch (op.*) { | 2369 | for (operations) |*op| switch (op.*) { |
| 2367 | .noop => unreachable, | 2370 | .noop => unreachable, |
| 2368 | inline else => |*o| o.result = error.Canceled, | 2371 | inline else => |*o| o.result = err, |
| 2369 | }; | 2372 | }; |
| 2370 | } | 2373 | } |
| 2371 | | 2374 | |
Andrew Kelley <andrew@ziglang.org> 2026-01-08 15:07:03-08:00
lib