| ... | @@ -85,6 +85,7 @@ pub const Node = struct { | ... | @@ -85,6 +85,7 @@ pub const Node = struct { |
| 85 | estimated_total_count: u32, | 85 | estimated_total_count: u32, |
| 86 | name: [max_name_len]u8, | 86 | name: [max_name_len]u8, |
| 87 | | 87 | |
| | 88 | /// Not thread-safe. |
| 88 | fn getIpcFd(s: Storage) ?posix.fd_t { | 89 | fn getIpcFd(s: Storage) ?posix.fd_t { |
| 89 | return if (s.estimated_total_count == std.math.maxInt(u32)) switch (@typeInfo(posix.fd_t)) { | 90 | return if (s.estimated_total_count == std.math.maxInt(u32)) switch (@typeInfo(posix.fd_t)) { |
| 90 | .Int => @bitCast(s.completed_count), | 91 | .Int => @bitCast(s.completed_count), |
| ... | @@ -93,15 +94,21 @@ pub const Node = struct { | ... | @@ -93,15 +94,21 @@ pub const Node = struct { |
| 93 | } else null; | 94 | } else null; |
| 94 | } | 95 | } |
| 95 | | 96 | |
| | 97 | /// Thread-safe. |
| 96 | fn setIpcFd(s: *Storage, fd: posix.fd_t) void { | 98 | fn setIpcFd(s: *Storage, fd: posix.fd_t) void { |
| 97 | s.estimated_total_count = std.math.maxInt(u32); | 99 | const integer: u32 = switch (@typeInfo(posix.fd_t)) { |
| 98 | s.completed_count = switch (@typeInfo(posix.fd_t)) { | | |
| 99 | .Int => @bitCast(fd), | 100 | .Int => @bitCast(fd), |
| 100 | .Pointer => @intFromPtr(fd), | 101 | .Pointer => @intFromPtr(fd), |
| 101 | else => @compileError("unsupported fd_t of " ++ @typeName(posix.fd_t)), | 102 | else => @compileError("unsupported fd_t of " ++ @typeName(posix.fd_t)), |
| 102 | }; | 103 | }; |
| | 104 | // `estimated_total_count` max int indicates the special state that |
| | 105 | // causes `completed_count` to be treated as a file descriptor, so |
| | 106 | // the order here matters. |
| | 107 | @atomicStore(u32, &s.completed_count, integer, .seq_cst); |
| | 108 | @atomicStore(u32, &s.estimated_total_count, std.math.maxInt(u32), .seq_cst); |
| 103 | } | 109 | } |
| 104 | | 110 | |
| | 111 | /// Not thread-safe. |
| 105 | fn byteSwap(s: *Storage) void { | 112 | fn byteSwap(s: *Storage) void { |
| 106 | s.completed_count = @byteSwap(s.completed_count); | 113 | s.completed_count = @byteSwap(s.completed_count); |
| 107 | s.estimated_total_count = @byteSwap(s.estimated_total_count); | 114 | s.estimated_total_count = @byteSwap(s.estimated_total_count); |
| ... | @@ -208,7 +215,9 @@ pub const Node = struct { | ... | @@ -208,7 +215,9 @@ pub const Node = struct { |
| 208 | pub fn setEstimatedTotalItems(n: Node, count: usize) void { | 215 | pub fn setEstimatedTotalItems(n: Node, count: usize) void { |
| 209 | const index = n.index.unwrap() orelse return; | 216 | const index = n.index.unwrap() orelse return; |
| 210 | const storage = storageByIndex(index); | 217 | const storage = storageByIndex(index); |
| 211 | @atomicStore(u32, &storage.estimated_total_count, std.math.lossyCast(u32, count), .monotonic); | 218 | // Avoid u32 max int which is used to indicate a special state. |
| | 219 | const saturated = @min(std.math.maxInt(u32) - 1, count); |
| | 220 | @atomicStore(u32, &storage.estimated_total_count, saturated, .monotonic); |
| 212 | } | 221 | } |
| 213 | | 222 | |
| 214 | /// Thread-safe. | 223 | /// Thread-safe. |
| ... | @@ -243,10 +252,13 @@ pub const Node = struct { | ... | @@ -243,10 +252,13 @@ pub const Node = struct { |
| 243 | } | 252 | } |
| 244 | } | 253 | } |
| 245 | | 254 | |
| 246 | /// Posix-only. Used by `std.process.Child`. | 255 | /// Posix-only. Used by `std.process.Child`. Thread-safe. |
| 247 | pub fn setIpcFd(node: Node, fd: posix.fd_t) void { | 256 | pub fn setIpcFd(node: Node, fd: posix.fd_t) void { |
| 248 | const index = node.index.unwrap() orelse return; | 257 | const index = node.index.unwrap() orelse return; |
| 249 | assert(fd != -1); | 258 | assert(fd >= 0); |
| | 259 | assert(fd != posix.STDOUT_FILENO); |
| | 260 | assert(fd != posix.STDIN_FILENO); |
| | 261 | assert(fd != posix.STDERR_FILENO); |
| 250 | storageByIndex(index).setIpcFd(fd); | 262 | storageByIndex(index).setIpcFd(fd); |
| 251 | } | 263 | } |
| 252 | | 264 | |
| ... | @@ -582,8 +594,8 @@ fn serialize(serialized_buffer: *Serialized.Buffer) Serialized { | ... | @@ -582,8 +594,8 @@ fn serialize(serialized_buffer: *Serialized.Buffer) Serialized { |
| 582 | while (begin_parent != .unused) { | 594 | while (begin_parent != .unused) { |
| 583 | const dest_storage = &serialized_buffer.storage[serialized_len]; | 595 | const dest_storage = &serialized_buffer.storage[serialized_len]; |
| 584 | @memcpy(&dest_storage.name, &storage_ptr.name); | 596 | @memcpy(&dest_storage.name, &storage_ptr.name); |
| 585 | dest_storage.completed_count = @atomicLoad(u32, &storage_ptr.completed_count, .monotonic); | 597 | dest_storage.completed_count = @atomicLoad(u32, &storage_ptr.completed_count, .seq_cst); |
| 586 | dest_storage.estimated_total_count = @atomicLoad(u32, &storage_ptr.estimated_total_count, .monotonic); | 598 | dest_storage.estimated_total_count = @atomicLoad(u32, &storage_ptr.estimated_total_count, .seq_cst); |
| 587 | const end_parent = @atomicLoad(Node.Parent, parent_ptr, .seq_cst); | 599 | const end_parent = @atomicLoad(Node.Parent, parent_ptr, .seq_cst); |
| 588 | if (begin_parent == end_parent) { | 600 | if (begin_parent == end_parent) { |
| 589 | any_ipc = any_ipc or (dest_storage.getIpcFd() != null); | 601 | any_ipc = any_ipc or (dest_storage.getIpcFd() != null); |