| ... | @@ -2366,14 +2366,13 @@ fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void { | ... | @@ -2366,14 +2366,13 @@ fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void { |
| 2366 | | 2366 | |
| 2367 | fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!void { | 2367 | fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!void { |
| 2368 | const t: *Threaded = @ptrCast(@alignCast(userdata)); | 2368 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 2369 | _ = t; | | |
| 2370 | | 2369 | |
| 2371 | if (operations.len == 1) { | 2370 | if (operations.len == 1) { |
| 2372 | @branchHint(.likely); | 2371 | @branchHint(.likely); |
| 2373 | return operate(&operations[0]); | 2372 | return operate(&operations[0]); |
| 2374 | } | 2373 | } |
| 2375 | | 2374 | |
| 2376 | if (is_windows) @panic("TODO"); | 2375 | if (is_windows) return batchWindows(t, operations); |
| 2377 | | 2376 | |
| 2378 | var poll_buffer: [poll_buffer_len]posix.pollfd = undefined; | 2377 | var poll_buffer: [poll_buffer_len]posix.pollfd = undefined; |
| 2379 | var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index | 2378 | var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index |
| ... | @@ -2397,7 +2396,7 @@ fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!v | ... | @@ -2397,7 +2396,7 @@ fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!v |
| 2397 | const map = map_buffer[0..poll_i]; | 2396 | const map = map_buffer[0..poll_i]; |
| 2398 | | 2397 | |
| 2399 | var pending = poll_i; | 2398 | var pending = poll_i; |
| 2400 | while (pending > 1) { | 2399 | while (pending > 0) { |
| 2401 | const syscall = Syscall.start() catch |err| switch (err) { | 2400 | const syscall = Syscall.start() catch |err| switch (err) { |
| 2402 | error.Canceled => { | 2401 | error.Canceled => { |
| 2403 | if (!setOperationsError(operations, polls, map, error.Canceled)) | 2402 | if (!setOperationsError(operations, polls, map, error.Canceled)) |
| ... | @@ -2408,17 +2407,11 @@ fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!v | ... | @@ -2408,17 +2407,11 @@ fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!v |
| 2408 | const rc = posix.system.poll(polls.ptr, polls.len, -1); | 2407 | const rc = posix.system.poll(polls.ptr, polls.len, -1); |
| 2409 | syscall.finish(); | 2408 | syscall.finish(); |
| 2410 | switch (posix.errno(rc)) { | 2409 | switch (posix.errno(rc)) { |
| 2411 | .SUCCESS => { | 2410 | .SUCCESS => for (polls, map) |*poll_fd, i| { |
| 2412 | if (rc == 0) { | 2411 | if (poll_fd.revents == 0) continue; |
| 2413 | // Spurious timeout; handle the same as INTR. | 2412 | poll_fd.fd = -1; |
| 2414 | continue; | 2413 | pending -= 1; |
| 2415 | } | 2414 | operate(&operations[i]); |
| 2416 | for (polls, map) |*poll_fd, i| { | | |
| 2417 | if (poll_fd.revents == 0) continue; | | |
| 2418 | poll_fd.fd = -1; | | |
| 2419 | pending -= 1; | | |
| 2420 | operate(&operations[i]); | | |
| 2421 | } | | |
| 2422 | }, | 2415 | }, |
| 2423 | .INTR => continue, | 2416 | .INTR => continue, |
| 2424 | .NOMEM => { | 2417 | .NOMEM => { |
| ... | @@ -2431,11 +2424,67 @@ fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!v | ... | @@ -2431,11 +2424,67 @@ fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!v |
| 2431 | }, | 2424 | }, |
| 2432 | } | 2425 | } |
| 2433 | } | 2426 | } |
| | 2427 | } |
| 2434 | | 2428 | |
| 2435 | if (pending == 1) for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| { | 2429 | fn batchWindows(t: *Threaded, operations: []Io.Operation) Io.ConcurrentError!void { |
| 2436 | if (poll_fd.fd == -1) continue; | 2430 | _ = t; |
| 2437 | operate(&operations[i]); | 2431 | var overlapped_buffer: [poll_buffer_len]windows.OVERLAPPED = undefined; |
| | 2432 | var handles_buffer: [poll_buffer_len]windows.HANDLE = undefined; |
| | 2433 | var map_buffer: [poll_buffer_len]u8 = undefined; // handles_buffer index to operations index |
| | 2434 | var buffer_i: usize = 0; |
| | 2435 | |
| | 2436 | for (operations, 0..) |*op, operation_index| switch (op.*) { |
| | 2437 | .noop => continue, |
| | 2438 | .file_read_streaming => |*o| { |
| | 2439 | if (handles_buffer.len - buffer_i == 0) return error.ConcurrencyUnavailable; |
| | 2440 | |
| | 2441 | const overlapped = &overlapped_buffer[buffer_i]; |
| | 2442 | overlapped.* = .{ |
| | 2443 | .Internal = 0, |
| | 2444 | .InternalHigh = 0, |
| | 2445 | .DUMMYUNIONNAME = .{ |
| | 2446 | .DUMMYSTRUCTNAME = .{ |
| | 2447 | .Offset = 0, |
| | 2448 | .OffsetHigh = 0, |
| | 2449 | }, |
| | 2450 | .Pointer = null, |
| | 2451 | }, |
| | 2452 | .hEvent = null, |
| | 2453 | }; |
| | 2454 | var n: windows.DWORD = undefined; |
| | 2455 | const buf = o.data[0]; |
| | 2456 | if (windows.kernel32.ReadFile(o.file.handle, buf.ptr, buf.len, &n, overlapped) == 0) { |
| | 2457 | @panic("TODO"); |
| | 2458 | } |
| | 2459 | handles_buffer[buffer_i] = o.file.handle; |
| | 2460 | map_buffer[buffer_i] = @intCast(operation_index); |
| | 2461 | buffer_i += 1; |
| | 2462 | }, |
| 2438 | }; | 2463 | }; |
| | 2464 | |
| | 2465 | const handles = handles_buffer[0..buffer_i]; |
| | 2466 | const map = map_buffer[0..buffer_i]; |
| | 2467 | var pending = buffer_i; |
| | 2468 | |
| | 2469 | while (pending > 0) { |
| | 2470 | const syscall: Syscall = try .start(); |
| | 2471 | const index = windows.WaitForMultipleObjectsEx(handles, false, windows.INFINITE, true); |
| | 2472 | syscall.finish(); |
| | 2473 | var n: windows.DWORD = undefined; |
| | 2474 | if (0 == windows.kernel32.GetOverlappedResult(handles[index], overlapped_buffer[index], &n, 0)) { |
| | 2475 | switch (windows.GetLastError()) { |
| | 2476 | .BROKEN_PIPE => @panic("TODO"), |
| | 2477 | .OPERATION_ABORTED => @panic("TODO"), |
| | 2478 | else => @panic("TODO"), |
| | 2479 | } |
| | 2480 | } else switch (operations[map[index]]) { |
| | 2481 | .noop => unreachable, |
| | 2482 | .file_read_streaming => |*o| { |
| | 2483 | o.status = .{ .result = n }; |
| | 2484 | pending -= 1; |
| | 2485 | }, |
| | 2486 | } |
| | 2487 | } |
| 2439 | } | 2488 | } |
| 2440 | | 2489 | |
| 2441 | fn setOperationsError( | 2490 | fn setOperationsError( |
Andrew Kelley <andrew@ziglang.org> 2026-01-09 20:46:51-08:00
lib