fn serialize(io: Io, serialized_buffer: *Serialized.Buffer) !Serialized {
var prev_parents: [node_storage_buffer_len]Node.Parent = undefined;
var prev_storage: [node_storage_buffer_len]Node.Storage = undefined;
{
const ipc_start = serialized_buffer.ipc_start;
const ipc_end = serialized_buffer.ipc_end;
@memcpy(prev_parents[ipc_start..ipc_end], serialized_buffer.parents[ipc_start..ipc_end]);
@memcpy(prev_storage[ipc_start..ipc_end], serialized_buffer.storage[ipc_start..ipc_end]);
}
// Iterate all of the nodes and construct a serializable copy of the state that can be examined
// without atomics. The `@min` call is here because `node_end_index` might briefly exceed the
// node count sometimes.
const end_index = @min(
@atomicLoad(u32, &global_progress.node_end_index, .monotonic),
node_storage_buffer_len,
);
var map: [node_storage_buffer_len]Node.OptionalIndex = undefined;
var serialized_len: u8 = 0;
var maybe_ipc_start: ?u8 = null;
for (
global_progress.node_parents[0..end_index],
global_progress.node_storage[0..end_index],
map[0..end_index],
) |*parent_ptr, *storage_ptr, *map_entry| {
const parent = @atomicLoad(Node.Parent, parent_ptr, .monotonic);
if (parent == .unused) {
// We might read "mixed" node data in this loop, due to weird atomic things
// or just a node actually being freed while this loop runs. That could cause
// there to be a parent reference to a nonexistent node. Without this assignment,
// this would lead to the map entry containing stale data. By assigning none, the
// child node with the bad parent pointer will be harmlessly omitted from the tree.
//
// Note that there's no concern of potentially creating "looping" data if we read
// "mixed" node data like this, because if a node is (directly or indirectly) its own
// parent, it will just not be printed at all. The general idea here is that performance
// is more important than 100% correct output every frame, given that this API is likely
// to be used in hot paths!
map_entry.* = .none;
continue;
}
const dest_storage = &serialized_buffer.storage[serialized_len];
copyAtomicLoad(&dest_storage.name, &storage_ptr.name);
dest_storage.estimated_total_count = @atomicLoad(u32, &storage_ptr.estimated_total_count, .acquire); // sychronizes with release in `setIpcIndex`
dest_storage.completed_count = @atomicLoad(u32, &storage_ptr.completed_count, .monotonic);
serialized_buffer.parents[serialized_len] = parent;
map_entry.* = @fromBackingInt(@intCast(serialized_len));
if (maybe_ipc_start == null and dest_storage.getIpcIndex() != null) maybe_ipc_start = serialized_len;
serialized_len += 1;
}
// Remap parents to point inside serialized arrays.
for (serialized_buffer.parents[0..serialized_len]) |*parent| {
parent.* = switch (parent.*) {
.unused => unreachable,
.none => .none,
_ => |p| map[@backingInt(p)].toParent(),
};
}
// Fill pipe buffers.
const batch = &serialized_buffer.batch;
batch.awaitConcurrent(io, .{
.duration = .{ .raw = .zero, .clock = .awake },
}) catch |err| switch (err) {
error.Timeout => {},
else => |e| return e,
};
var ready_len: u8 = 0;
while (batch.next()) |operation| switch (operation.index) {
0...ipc_storage_buffer_len - 1 => {
const ipc_data = &serialized_buffer.ipc_data[operation.index];
ipc_data.bytes_read += @intCast(
operation.result.file_read_streaming catch |err| switch (err) {
error.EndOfStream => {
const file = global_progress.ipc_files[operation.index];
const ipc = @atomicRmw(
Ipc,
&global_progress.ipc[operation.index],
.And,
.{
.locked = false,
.valid = true,
.generation = std.math.maxInt(Ipc.Generation),
},
.release,
);
assert(ipc.locked);
if (!ipc.valid) file.close(io);
ipc_data.* = .unused;
continue;
},
else => |e| return e,
},
);
assert(ipc_data.state == .pending);
ipc_data.state = .ready;
ready_len += 1;
},
else => unreachable,
};
// Find nodes which correspond to child processes.
const ipc_start = maybe_ipc_start orelse serialized_len;
serialized_buffer.ipc_start = ipc_start;
for (
serialized_buffer.parents[ipc_start..serialized_len],
serialized_buffer.storage[ipc_start..serialized_len],
ipc_start..,
) |main_parent, *main_storage, main_index| {
if (main_parent == .unused) continue;
const ipc_index = main_storage.getIpcIndex() orelse continue;
const ipc = &global_progress.ipc[ipc_index.slot];
const ipc_data = &serialized_buffer.ipc_data[ipc_index.slot];
state: switch (ipc_data.state) {
.unused => {
if (@cmpxchgWeak(
Ipc,
ipc,
.{ .locked = false, .valid = true, .generation = ipc_index.generation },
.{ .locked = true, .valid = true, .generation = ipc_index.generation },
.acquire,
.monotonic,
)) |_| continue;
const ipc_vec = &serialized_buffer.ipc_vecs[ipc_index.slot];
ipc_vec.* = .{&serialized_buffer.ipc_buffers[ipc_index.slot]};
batch.addAt(ipc_index.slot, .{ .file_read_streaming = .{
.file = global_progress.ipc_files[ipc_index.slot],
.data = ipc_vec,
} });
ipc_data.* = .{
.state = .pending,
.bytes_read = 0,
.main_index = @intCast(main_index),
.start_index = serialized_len,
.nodes_len = 0,
};
main_storage.completed_count = 0;
main_storage.estimated_total_count = 0;
},
.pending => {
const start_index = ipc_data.start_index;
const nodes_len = @min(ipc_data.nodes_len, node_storage_buffer_len - serialized_len);
main_storage.copyRoot(&prev_storage[ipc_data.main_index]);
@memcpy(
serialized_buffer.storage[serialized_len..][0..nodes_len],
prev_storage[start_index..][0..nodes_len],
);
for (
serialized_buffer.parents[serialized_len..][0..nodes_len],
prev_parents[serialized_len..][0..nodes_len],
) |*parent, prev_parent| parent.* = switch (prev_parent) {
.none, .unused => .none,
_ => if (@backingInt(prev_parent) == ipc_data.main_index)
@fromBackingInt(@intCast(main_index))
else if (@backingInt(prev_parent) >= start_index and
@backingInt(prev_parent) < start_index + nodes_len)
@fromBackingInt(@intCast(@backingInt(prev_parent) - start_index + serialized_len))
else
.none,
};
ipc_data.main_index = @intCast(main_index);
ipc_data.start_index = serialized_len;
ipc_data.nodes_len = nodes_len;
serialized_len += nodes_len;
},
.ready => {
const ipc_buffer = &serialized_buffer.ipc_buffers[ipc_index.slot];
const packet_start, const packet_end = ipc_data.findLastPacket(ipc_buffer);
const packet_is_empty = packet_end - packet_start <= 1;
if (!packet_is_empty) {
const storage, const parents, const nodes_len = packet_contents: {
var packet_index: usize = packet_start;
const nodes_len: u16 = ipc_buffer[packet_index];
packet_index += 1;
const storage_bytes =
ipc_buffer[packet_index..][0 .. nodes_len * @sizeOf(Node.Storage)];
packet_index += storage_bytes.len;
const parents_bytes =
ipc_buffer[packet_index..][0 .. nodes_len * @sizeOf(Node.Parent)];
packet_index += parents_bytes.len;
assert(packet_index == packet_end);
const storage: []align(1) const Node.Storage = @ptrCast(storage_bytes);
const parents: []align(1) const Node.Parent = @ptrCast(parents_bytes);
const children_nodes_len =
@min(nodes_len - 1, node_storage_buffer_len - serialized_len);
break :packet_contents .{ storage, parents, children_nodes_len };
};
// Mount the root here.
main_storage.copyRoot(&storage[0]);
if (is_big_endian) main_storage.byteSwap();
// Copy the rest of the tree to the end.
const serialized_storage =
serialized_buffer.storage[serialized_len..][0..nodes_len];
@memcpy(serialized_storage, storage[1..][0..nodes_len]);
if (is_big_endian) for (serialized_storage) |*s| s.byteSwap();
// Patch up parent pointers taking into account how the subtree is mounted.
for (
serialized_buffer.parents[serialized_len..][0..nodes_len],
parents[1..][0..nodes_len],
) |*parent, prev_parent| parent.* = switch (prev_parent) {
// Fix bad data so the rest of the code does not see `unused`.
.none, .unused => .none,
// Root node is being mounted here.
@as(Node.Parent, @fromBackingInt(@intCast(0))) => @fromBackingInt(@intCast(main_index)),
// Other nodes mounted at the end.
// Don't trust child data; if the data is outside the expected range,
// ignore the data. This also handles the case when data was truncated.
_ => if (@backingInt(prev_parent) <= nodes_len)
@fromBackingInt(@intCast(@backingInt(prev_parent) - 1 + serialized_len))
else
.none,
};
ipc_data.main_index = @intCast(main_index);
ipc_data.start_index = serialized_len;
ipc_data.nodes_len = nodes_len;
serialized_len += nodes_len;
}
const ipc_vec = &serialized_buffer.ipc_vecs[ipc_index.slot];
ipc_data.rebase(ipc_buffer, ipc_vec, batch, ipc_index.slot, packet_end);
ready_len -= 1;
if (packet_is_empty) continue :state .pending;
},
}
}
serialized_buffer.ipc_end = serialized_len;
// Ignore data from unused pipes. This ensures that if a child process exists we will
// eventually see `EndOfStream` and close the pipe.
if (ready_len > 0) for (
&serialized_buffer.ipc_data,
&serialized_buffer.ipc_buffers,
&serialized_buffer.ipc_vecs,
0..,
) |*ipc_data, *ipc_buffer, *ipc_vec, ipc_slot| switch (ipc_data.state) {
.unused, .pending => {},
.ready => {
_, const packet_end = ipc_data.findLastPacket(ipc_buffer);
ipc_data.rebase(ipc_buffer, ipc_vec, batch, @intCast(ipc_slot), packet_end);
ready_len -= 1;
},
};
assert(ready_len == 0);
return .{
.parents = serialized_buffer.parents[0..serialized_len],
.storage = serialized_buffer.storage[0..serialized_len],
};
}
