serialize

fn serialize(io: Io, serialized_buffer: *Serialized.Buffer) !Serialized

File

Code

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],
    };
}