| ... | ... | @@ -2474,6 +2474,7 @@ fn operate(userdata: ?*anyopaque, op: *Io.Operation) Io.Cancelable!void { |
| 2474 | 2474 | |
| 2475 | 2475 | fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.WaitError!void { |
| 2476 | 2476 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 2477 | if (is_windows) return batchWaitWindows(t, b, timeout); |
| 2477 | 2478 | const operations = b.operations; |
| 2478 | 2479 | const len: u31 = @intCast(operations.len); |
| 2479 | 2480 | const ring = b.ring[0..len]; |
| ... | ... | @@ -2492,13 +2493,12 @@ fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch. |
| 2492 | 2493 | b.impl.complete_tail = complete_tail; |
| 2493 | 2494 | b.user.complete_tail = complete_tail; |
| 2494 | 2495 | } |
| 2495 | | if (is_windows) @panic("TODO"); |
| 2496 | 2496 | var poll_buffer: [poll_buffer_len]posix.pollfd = undefined; |
| 2497 | 2497 | while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) { |
| 2498 | 2498 | const op = ring[submit_head.index(len)]; |
| 2499 | 2499 | const operation = &operations[op]; |
| 2500 | 2500 | switch (operation.*) { |
| 2501 | | else => { |
| 2501 | .noop => { |
| 2502 | 2502 | try operate(t, operation); |
| 2503 | 2503 | ring[complete_tail.index(len)] = op; |
| 2504 | 2504 | complete_tail = complete_tail.next(len); |
| ... | ... | @@ -2597,147 +2597,98 @@ fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void { |
| 2597 | 2597 | b.user.complete_tail = complete_tail; |
| 2598 | 2598 | } |
| 2599 | 2599 | |
| 2600 | | fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!void { |
| 2601 | | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 2602 | | |
| 2603 | | if (operations.len == 1) { |
| 2604 | | @branchHint(.likely); |
| 2605 | | return operate(&operations[0]); |
| 2606 | | } |
| 2607 | | |
| 2608 | | if (is_windows) return batchWindows(t, operations); |
| 2609 | | |
| 2610 | | var poll_buffer: [poll_buffer_len]posix.pollfd = undefined; |
| 2611 | | var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index |
| 2612 | | var poll_i: usize = 0; |
| 2600 | fn batchWaitWindows(t: *Threaded, b: *Io.Batch, timeout: Io.Timeout) Io.ConcurrentError!void { |
| 2601 | const operations = b.operations; |
| 2602 | const len: u31 = @intCast(operations.len); |
| 2603 | const ring = b.ring[0..len]; |
| 2604 | var submit_head = b.impl.submit_head; |
| 2605 | const submit_tail = b.user.submit_tail; |
| 2606 | b.impl.submit_tail = submit_tail; |
| 2607 | var complete_tail = b.impl.complete_tail; |
| 2613 | 2608 | |
| 2614 | | for (operations, 0..) |*op, operation_index| switch (op.*) { |
| 2615 | | .noop => continue, |
| 2616 | | .file_read_streaming => |*o| { |
| 2617 | | if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable; |
| 2618 | | poll_buffer[poll_i] = .{ |
| 2619 | | .fd = o.file.handle, |
| 2620 | | .events = posix.POLL.IN, |
| 2621 | | .revents = 0, |
| 2622 | | }; |
| 2623 | | map_buffer[poll_i] = @intCast(operation_index); |
| 2624 | | poll_i += 1; |
| 2625 | | }, |
| 2626 | | }; |
| 2609 | var overlapped_buffer: [poll_buffer_len]windows.OVERLAPPED = undefined; |
| 2610 | var handles_buffer: [poll_buffer_len]windows.HANDLE = undefined; |
| 2611 | var map_buffer: [poll_buffer_len]u32 = undefined; // handles_buffer index to operations index |
| 2612 | var buffer_i: usize = 0; |
| 2627 | 2613 | |
| 2628 | | const polls = poll_buffer[0..poll_i]; |
| 2629 | | const map = map_buffer[0..poll_i]; |
| 2614 | defer { |
| 2615 | for (map_buffer[0..buffer_i]) |op| { |
| 2616 | submit_head = submit_head.prev(len); |
| 2617 | ring[submit_head.index(len)] = op; |
| 2618 | } |
| 2619 | b.impl.submit_head = submit_head; |
| 2620 | b.impl.complete_tail = complete_tail; |
| 2621 | b.user.complete_tail = complete_tail; |
| 2622 | } |
| 2630 | 2623 | |
| 2631 | | var pending = poll_i; |
| 2632 | | while (pending > 0) { |
| 2633 | | const syscall = Syscall.start() catch |err| switch (err) { |
| 2634 | | error.Canceled => { |
| 2635 | | if (!setOperationsError(operations, polls, map, error.Canceled)) |
| 2636 | | recancelInner(); |
| 2637 | | return; |
| 2638 | | }, |
| 2639 | | }; |
| 2640 | | const rc = posix.system.poll(polls.ptr, polls.len, -1); |
| 2641 | | syscall.finish(); |
| 2642 | | switch (posix.errno(rc)) { |
| 2643 | | .SUCCESS => for (polls, map) |*poll_fd, i| { |
| 2644 | | if (poll_fd.revents == 0) continue; |
| 2645 | | poll_fd.fd = -1; |
| 2646 | | pending -= 1; |
| 2647 | | operate(&operations[i]); |
| 2648 | | }, |
| 2649 | | .INTR => continue, |
| 2650 | | .NOMEM => { |
| 2651 | | assert(setOperationsError(operations, polls, map, error.SystemResources)); |
| 2652 | | return; |
| 2624 | while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) { |
| 2625 | const op = ring[submit_head.index(len)]; |
| 2626 | const operation = &operations[op]; |
| 2627 | switch (operation.*) { |
| 2628 | .noop => { |
| 2629 | try operate(t, operation); |
| 2630 | ring[complete_tail.index(len)] = op; |
| 2631 | complete_tail = complete_tail.next(len); |
| 2653 | 2632 | }, |
| 2654 | | else => { |
| 2655 | | assert(setOperationsError(operations, polls, map, error.Unexpected)); |
| 2656 | | return; |
| 2633 | .file_read_streaming => |*o| { |
| 2634 | _ = o.status.unstarted; |
| 2635 | if (handles_buffer.len - buffer_i == 0) return error.ConcurrencyUnavailable; |
| 2636 | const overlapped = &overlapped_buffer[buffer_i]; |
| 2637 | overlapped.* = .{ |
| 2638 | .Internal = 0, |
| 2639 | .InternalHigh = 0, |
| 2640 | .DUMMYUNIONNAME = .{ |
| 2641 | .DUMMYSTRUCTNAME = .{ |
| 2642 | .Offset = 0, |
| 2643 | .OffsetHigh = 0, |
| 2644 | }, |
| 2645 | .Pointer = null, |
| 2646 | }, |
| 2647 | .hEvent = null, |
| 2648 | }; |
| 2649 | var n: windows.DWORD = undefined; |
| 2650 | const buf = o.data[0]; |
| 2651 | if (windows.kernel32.ReadFile(o.file.handle, buf.ptr, buf.len, &n, overlapped) == 0) { |
| 2652 | @panic("TODO"); |
| 2653 | } |
| 2654 | handles_buffer[buffer_i] = o.file.handle; |
| 2655 | map_buffer[buffer_i] = op; |
| 2656 | buffer_i += 1; |
| 2657 | 2657 | }, |
| 2658 | 2658 | } |
| 2659 | 2659 | } |
| 2660 | | } |
| 2661 | | |
| 2662 | | fn batchWindows(t: *Threaded, operations: []Io.Operation) Io.ConcurrentError!void { |
| 2663 | | _ = t; |
| 2664 | | var overlapped_buffer: [poll_buffer_len]windows.OVERLAPPED = undefined; |
| 2665 | | var handles_buffer: [poll_buffer_len]windows.HANDLE = undefined; |
| 2666 | | var map_buffer: [poll_buffer_len]u8 = undefined; // handles_buffer index to operations index |
| 2667 | | var buffer_i: usize = 0; |
| 2668 | 2660 | |
| 2669 | | for (operations, 0..) |*op, operation_index| switch (op.*) { |
| 2670 | | .noop => continue, |
| 2671 | | .file_read_streaming => |*o| { |
| 2672 | | if (handles_buffer.len - buffer_i == 0) return error.ConcurrencyUnavailable; |
| 2673 | | |
| 2674 | | const overlapped = &overlapped_buffer[buffer_i]; |
| 2675 | | overlapped.* = .{ |
| 2676 | | .Internal = 0, |
| 2677 | | .InternalHigh = 0, |
| 2678 | | .DUMMYUNIONNAME = .{ |
| 2679 | | .DUMMYSTRUCTNAME = .{ |
| 2680 | | .Offset = 0, |
| 2681 | | .OffsetHigh = 0, |
| 2682 | | }, |
| 2683 | | .Pointer = null, |
| 2684 | | }, |
| 2685 | | .hEvent = null, |
| 2686 | | }; |
| 2687 | | var n: windows.DWORD = undefined; |
| 2688 | | const buf = o.data[0]; |
| 2689 | | if (windows.kernel32.ReadFile(o.file.handle, buf.ptr, buf.len, &n, overlapped) == 0) { |
| 2690 | | @panic("TODO"); |
| 2691 | | } |
| 2692 | | handles_buffer[buffer_i] = o.file.handle; |
| 2693 | | map_buffer[buffer_i] = @intCast(operation_index); |
| 2694 | | buffer_i += 1; |
| 2661 | switch (buffer_i) { |
| 2662 | 0 => return, |
| 2663 | 1 => if (timeout == .none) { |
| 2664 | const op = map_buffer[0]; |
| 2665 | try operate(t, &operations[op]); |
| 2666 | ring[complete_tail.index(len)] = op; |
| 2667 | complete_tail = complete_tail.next(len); |
| 2668 | return; |
| 2695 | 2669 | }, |
| 2696 | | }; |
| 2670 | else => {}, |
| 2671 | } |
| 2697 | 2672 | |
| 2698 | 2673 | const handles = handles_buffer[0..buffer_i]; |
| 2699 | 2674 | const map = map_buffer[0..buffer_i]; |
| 2700 | | var pending = buffer_i; |
| 2701 | 2675 | |
| 2702 | | while (pending > 0) { |
| 2703 | | const syscall: Syscall = try .start(); |
| 2704 | | const index = windows.WaitForMultipleObjectsEx(handles, false, windows.INFINITE, true); |
| 2705 | | syscall.finish(); |
| 2706 | | var n: windows.DWORD = undefined; |
| 2707 | | if (0 == windows.kernel32.GetOverlappedResult(handles[index], overlapped_buffer[index], &n, 0)) { |
| 2708 | | switch (windows.GetLastError()) { |
| 2709 | | .BROKEN_PIPE => @panic("TODO"), |
| 2710 | | .OPERATION_ABORTED => @panic("TODO"), |
| 2711 | | else => @panic("TODO"), |
| 2712 | | } |
| 2713 | | } else switch (operations[map[index]]) { |
| 2714 | | .noop => unreachable, |
| 2715 | | .file_read_streaming => |*o| { |
| 2716 | | o.status = .{ .result = n }; |
| 2717 | | pending -= 1; |
| 2718 | | }, |
| 2719 | | } |
| 2720 | | } |
| 2721 | | } |
| 2722 | | |
| 2723 | | fn setOperationsError( |
| 2724 | | operations: []Io.Operation, |
| 2725 | | polls: []const posix.pollfd, |
| 2726 | | map: []const u8, |
| 2727 | | err: error{ Canceled, SystemResources, Unexpected }, |
| 2728 | | ) bool { |
| 2729 | | var marked = false; |
| 2730 | | for (polls, map) |*poll_fd, i| { |
| 2731 | | if (poll_fd.fd == -1) continue; |
| 2732 | | switch (operations[i]) { |
| 2733 | | .noop => unreachable, |
| 2734 | | inline else => |*o| { |
| 2735 | | o.status = .{ .result = err }; |
| 2736 | | marked = true; |
| 2737 | | }, |
| 2676 | const syscall: Syscall = try .start(); |
| 2677 | const index = windows.WaitForMultipleObjectsEx(handles, false, windows.INFINITE, true); |
| 2678 | syscall.finish(); |
| 2679 | var n: windows.DWORD = undefined; |
| 2680 | if (0 == windows.kernel32.GetOverlappedResult(handles[index], overlapped_buffer[index], &n, 0)) { |
| 2681 | switch (windows.GetLastError()) { |
| 2682 | .BROKEN_PIPE => @panic("TODO"), |
| 2683 | .OPERATION_ABORTED => @panic("TODO"), |
| 2684 | else => @panic("TODO"), |
| 2738 | 2685 | } |
| 2686 | } else switch (operations[map[index]]) { |
| 2687 | .noop => unreachable, |
| 2688 | .file_read_streaming => |*o| { |
| 2689 | o.status = .{ .result = n }; |
| 2690 | }, |
| 2739 | 2691 | } |
| 2740 | | return marked; |
| 2741 | 2692 | } |
| 2742 | 2693 | |
| 2743 | 2694 | const dirCreateDir = switch (native_os) { |
Andrew Kelley <andrew@ziglang.org> 2026-01-12 18:04:56-08:00
lib