tiny.profiling.ingest.output

1const std = @import("std");2const pretty = @import("pretty");3const sys = @import("sys");4const profiling = @import("../root.zig");5const ingest = @import("root.zig");67const pretty_json = pretty.json;89const RowContext = struct {10    writer: *std.Io.Writer,11    workload: profiling.catalog.Workload,12    source: ingest.Source,13    row_bits: []const u8,14    candidate_capacity: usize,15    candidate_index: usize,16    rows: usize,17};1819pub fn write(20    paths: ingest.Paths,21    facts: ingest.Facts,22    workload: profiling.catalog.Workload,23    row_bits: []const u8,24    candidate_capacity: usize,25    line_storage: []u8,26    totals: profiling.allocation.Totals,27    expected: ingest.Summary,28) !ingest.Summary {29    std.debug.assert(facts.sources.len == ingest.model.source_count);30    std.debug.assert(expected.sources <= ingest.model.source_count);31    std.debug.assert(candidate_capacity / 8 <= row_bits.len);32    if (candidate_capacity % 8 > 0) std.debug.assert(candidate_capacity / 8 < row_bits.len);33    for (facts.sources) |source| {34        std.debug.assert(source.max_line_bytes <= line_storage.len);35    }36    var path_storage: [std.fs.max_path_bytes]u8 = undefined;37    const pending_path = try std.fmt.bufPrint(38        &path_storage,39        "{s}.ingest.pending",40        .{paths.structured},41    );42    std.debug.assert(pending_path.len > paths.structured.len);43    std.debug.assert(pending_path.len <= path_storage.len);44    var file = try sys.fs.createFile(pending_path, .{ .truncate = true });45    var file_open = true;46    defer if (file_open) file.close(sys.fs.debugIo());47    defer sys.fs.deleteFile(pending_path) catch {};48    var buffer: [64 * 1024]u8 = undefined;49    var file_writer = file.writer(sys.fs.debugIo(), &buffer);5051    const sources = paths.sources();52    var context = RowContext{53        .writer = &file_writer.interface,54        .workload = workload,55        .source = undefined,56        .row_bits = row_bits,57        .candidate_capacity = candidate_capacity,58        .candidate_index = 0,59        .rows = 0,60    };61    try writeOrdinarySources(&context, facts, &sources, line_storage);62    if (context.candidate_index != candidate_capacity) return error.InputChanged;63    std.debug.assert(context.candidate_index == candidate_capacity);64    try writeMetrics(&context, totals, sources[ingest.model.ordinary_source_count]);65    const allocation_facts = try ingest.inspection.inspectSource(66        sources[ingest.model.ordinary_source_count],67    );68    if (!facts.sources[ingest.model.ordinary_source_count].eql(allocation_facts)) {69        return error.InputChanged;70    }71    if (context.rows != expected.rows) return error.InputChanged;72    std.debug.assert(context.rows == expected.rows);73    try file_writer.interface.flush();74    file.close(sys.fs.debugIo());75    file_open = false;76    try sys.fs.rename(pending_path, paths.structured);77    return expected;78}7980fn writeOrdinarySources(81    context: *RowContext,82    facts: ingest.Facts,83    sources: *const [ingest.model.source_count]ingest.Source,84    line_storage: []u8,85) !void {86    for (sources[0..ingest.model.ordinary_source_count], 0..) |source, index| {87        std.debug.assert(index < ingest.model.ordinary_source_count);88        context.source = source;89        const actual = try ingest.inspection.visitSource(90            source,91            line_storage,92            context,93            writeCandidate,94        );95        if (!facts.sources[index].eql(actual)) return error.InputChanged;96    }97}9899fn writeMetrics(100    context: *RowContext,101    totals: profiling.allocation.Totals,102    source: ingest.Source,103) !void {104    if (!totals.saw) return;105    for (totals.metrics()) |metric| {106        try writeMemoryMetricRow(context.writer, context.workload, source, metric);107        context.rows += 1;108    }109    std.debug.assert(totals.metrics().len > 0);110}111112fn writeCandidate(context: *RowContext, raw: []const u8, line_number: usize) !void {113    std.debug.assert(raw.len >= 2);114    std.debug.assert(line_number > 0);115    std.debug.assert(context.candidate_index <= context.candidate_capacity);116    const index = context.candidate_index;117    context.candidate_index = std.math.add(usize, index, 1) catch118        return error.InputChanged;119    if (index >= context.candidate_capacity) return error.InputChanged;120    if (!bitIsSet(context.row_bits, index)) return;121    try writeRow(context.writer, context.workload, context.source, line_number, raw);122    context.rows = std.math.add(usize, context.rows, 1) catch return error.InputChanged;123}124125fn bitIsSet(bits: []const u8, index: usize) bool {126    const byte_index = index / 8;127    const bit_index: u3 = @intCast(index % 8);128    std.debug.assert(byte_index < bits.len);129    std.debug.assert(bit_index < 8);130    return bits[byte_index] & (@as(u8, 1) << bit_index) != 0;131}132133fn writeRow(134    writer: *std.Io.Writer,135    workload: profiling.catalog.Workload,136    source: ingest.Source,137    line_number: usize,138    raw: []const u8,139) !void {140    std.debug.assert(raw.len >= 2);141    std.debug.assert(raw[0] == '{');142    std.debug.assert(raw[raw.len - 1] == '}');143    std.debug.assert(line_number > 0);144    var out = pretty_json.Writer.init(writer, .minified);145    try out.beginObject();146    try out.objectField("schema");147    try out.write(ingest.schema);148    try out.objectField("workload");149    try writeWorkload(&out, workload);150    try out.objectField("source");151    try out.beginObject();152    try out.objectField("kind");153    try out.write(source.kind.name());154    try out.objectField("path");155    try out.write(source.path);156    try out.objectField("line");157    try out.write(line_number);158    try out.endObject();159    try out.objectField("row");160    try out.raw(raw);161    try out.endObject();162    try writer.writeByte('\n');163}164165fn writeMemoryMetricRow(166    writer: *std.Io.Writer,167    workload: profiling.catalog.Workload,168    source: ingest.Source,169    metric: profiling.allocation.Metric,170) !void {171    std.debug.assert(source.kind.isAllocation());172    std.debug.assert(metric.name.len > 0);173    std.debug.assert(metric.unit.len > 0);174    var out = pretty_json.Writer.init(writer, .minified);175    try out.beginObject();176    try out.objectField("schema");177    try out.write(ingest.schema);178    try out.objectField("workload");179    try writeWorkload(&out, workload);180    try out.objectField("source");181    try out.beginObject();182    try out.objectField("kind");183    try out.write(source.kind.name());184    try out.objectField("path");185    try out.write(source.path);186    try out.objectField("line");187    try out.write(0);188    try out.endObject();189    try out.objectField("row");190    try out.beginObject();191    try out.objectField("schema");192    try out.write(profiling.schema.metric_schema);193    try out.objectField("family");194    try out.write("memory");195    try out.objectField("metric");196    try out.write(metric.name);197    try out.objectField("name");198    try out.write(metric.name);199    try out.objectField("unit");200    try out.write(metric.unit);201    try out.objectField("value");202    try out.write(metric.value);203    try out.endObject();204    try out.endObject();205    try writer.writeByte('\n');206}207208fn writeWorkload(out: *pretty_json.Writer, workload: profiling.catalog.Workload) !void {209    try out.beginObject();210    try out.objectField("name");211    try out.write(workload.name);212    try out.objectField("package");213    try out.write(workload.package);214    try out.objectField("step");215    try out.write(workload.step);216    try out.endObject();217}

1pub const output = @import("output.zig");