tiny.profiling.ingest.inspection

1const std = @import("std");2const sys = @import("sys");3const ingest = @import("root.zig");45const model = ingest.model;67const LineInspection = struct {8    bytes: usize = 0,9    first_significant: ?u8 = null,10    last_significant: ?u8 = null,11    json_depth: usize = 0,12    max_json_depth: usize = 0,13    key_count: usize = 0,14    in_string: bool = false,15    escaped: bool = false,1617    fn observe(self: *LineInspection, byte: u8) error{CapacityOverflow}!void {18        const previous_bytes = self.bytes;19        self.bytes = std.math.add(usize, self.bytes, 1) catch return error.CapacityOverflow;20        std.debug.assert(self.bytes == previous_bytes + 1);21        std.debug.assert(self.json_depth <= self.max_json_depth);22        if (!isTrimByte(byte)) {23            if (self.first_significant == null) self.first_significant = byte;24            self.last_significant = byte;25        }26        if (self.in_string) {27            if (self.escaped) {28                self.escaped = false;29            } else if (byte == '\\') {30                self.escaped = true;31            } else if (byte == '"') {32                self.in_string = false;33            }34            std.debug.assert(self.json_depth <= self.max_json_depth);35            return;36        }37        if (byte == '"') {38            self.in_string = true;39        } else if (byte == ':') {40            self.key_count = std.math.add(usize, self.key_count, 1) catch41                return error.CapacityOverflow;42        } else if (byte == '{' or byte == '[') {43            self.json_depth = std.math.add(usize, self.json_depth, 1) catch44                return error.CapacityOverflow;45            self.max_json_depth = @max(self.max_json_depth, self.json_depth);46            std.debug.assert(self.json_depth <= self.max_json_depth);47        } else if ((byte == '}' or byte == ']') and self.json_depth > 0) {48            self.json_depth -= 1;49        }50        std.debug.assert(self.key_count <= self.bytes);51    }5253    fn isCandidate(self: LineInspection) bool {54        const result = self.first_significant == '{' and self.last_significant == '}';55        if (result) {56            std.debug.assert(self.first_significant == '{');57            std.debug.assert(self.last_significant == '}');58            std.debug.assert(self.bytes >= 2);59        }60        return result;61    }62};6364const ScanState = struct {65    facts: model.SourceFacts,66    hasher: std.crypto.hash.sha2.Sha256,67    line: LineInspection = .{},68    line_number: usize = 0,69};7071pub fn inspect(paths: model.Paths) !model.Limits {72    const sources = paths.sources();73    std.debug.assert(sources.len == model.source_count);74    var facts: model.Facts = undefined;75    for (sources, 0..) |source, index| {76        std.debug.assert(index < model.source_count);77        facts.sources[index] = try inspectSource(source);78    }79    std.debug.assert(facts.sources.len == model.source_count);80    return .{ .paths = paths, .facts = facts };81}8283pub fn inspectSource(source: model.Source) !model.SourceFacts {84    std.debug.assert(source.kind.name().len > 0);85    return scanSource(source, null, {}, ignoreCandidate);86}8788pub fn visitSource(89    source: model.Source,90    line_storage: []u8,91    context: anytype,92    comptime visitCandidate: anytype,93) !model.SourceFacts {94    std.debug.assert(source.kind.name().len > 0);95    return scanSource(source, line_storage, context, visitCandidate);96}9798pub fn metadataMatches(paths: model.Paths, facts: model.Facts) !bool {99    const sources = paths.sources();100    std.debug.assert(sources.len == facts.sources.len);101    for (sources, facts.sources) |source, expected| {102        var file = sys.fs.cwd().openFile(103            sys.fs.debugIo(),104            source.path,105            .{},106        ) catch |err| switch (err) {107            error.FileNotFound => {108                if (expected.present) return false;109                continue;110            },111            else => return err,112        };113        defer file.close(sys.fs.debugIo());114        if (!expected.present) return false;115        std.debug.assert(expected.present);116        const actual = model.Identity.fromStat(try file.stat(sys.fs.debugIo()));117        if (!std.meta.eql(expected.identity, actual)) return false;118    }119    return true;120}121122pub fn admitSourceBytes(bytes: usize) error{StreamTooLong}!void {123    std.debug.assert(model.max_source_bytes > 0);124    if (bytes >= model.max_source_bytes) return error.StreamTooLong;125    std.debug.assert(bytes < model.max_source_bytes);126}127128pub fn admitAllocationLine(bytes: usize, terminated: bool) error{StreamTooLong}!void {129    std.debug.assert(model.allocation_reader_bytes > 0);130    if (bytes > model.allocation_reader_bytes) return error.StreamTooLong;131    if (terminated and bytes == model.allocation_reader_bytes) return error.StreamTooLong;132    std.debug.assert(bytes <= model.allocation_reader_bytes);133}134135fn scanSource(136    source: model.Source,137    line_storage: ?[]u8,138    context: anytype,139    comptime visitCandidate: anytype,140) !model.SourceFacts {141    std.debug.assert(source.kind.name().len > 0);142    var file = sys.fs.cwd().openFile(sys.fs.debugIo(), source.path, .{}) catch |err| switch (err) {143        error.FileNotFound => return .{},144        else => return err,145    };146    defer file.close(sys.fs.debugIo());147    const identity = model.Identity.fromStat(try file.stat(sys.fs.debugIo()));148    std.debug.assert(identity.size <= std.math.maxInt(u64));149    const expected_bytes = std.math.cast(usize, identity.size) orelse150        return error.CapacityOverflow;151    if (!source.kind.isAllocation()) {152        try admitSourceBytes(expected_bytes);153    }154    return scanContents(155        &file,156        source,157        identity,158        expected_bytes,159        line_storage,160        context,161        visitCandidate,162    );163}164165fn scanContents(166    file: *std.Io.File,167    source: model.Source,168    identity: model.Identity,169    expected_bytes: usize,170    line_storage: ?[]u8,171    context: anytype,172    comptime visitCandidate: anytype,173) !model.SourceFacts {174    std.debug.assert(source.kind.name().len > 0);175    var state = ScanState{176        .facts = .{ .present = true, .identity = identity },177        .hasher = std.crypto.hash.sha2.Sha256.init(.{}),178    };179    std.debug.assert(state.facts.present);180    try scanChunks(181        file,182        source,183        expected_bytes,184        line_storage,185        context,186        visitCandidate,187        &state,188    );189    if (state.facts.bytes != expected_bytes) return error.InputChanged;190    if (state.line.bytes > 0) {191        std.debug.assert(state.line_number <= state.facts.lines);192        if (source.kind.isAllocation()) try admitAllocationLine(state.line.bytes, false);193        state.line_number = std.math.add(usize, state.line_number, 1) catch194            return error.CapacityOverflow;195        try finishLine(196            &state.facts,197            &state.line,198            line_storage,199            state.line_number,200            context,201            visitCandidate,202        );203    }204    state.hasher.final(&state.facts.digest);205    const final_identity = model.Identity.fromStat(try file.stat(sys.fs.debugIo()));206    if (!std.meta.eql(identity, final_identity)) return error.InputChanged;207    if (identity.size != state.facts.bytes) return error.InputChanged;208    std.debug.assert(state.facts.present);209    std.debug.assert(state.facts.candidates <= state.facts.lines);210    std.debug.assert(state.facts.max_json_depth <= state.facts.max_line_bytes);211    std.debug.assert(state.facts.max_key_count <= state.facts.max_line_bytes);212    return state.facts;213}214215fn scanChunks(216    file: *std.Io.File,217    source: model.Source,218    expected_bytes: usize,219    line_storage: ?[]u8,220    context: anytype,221    comptime visitCandidate: anytype,222    state: *ScanState,223) !void {224    var reader_buffer: [16 * 1024]u8 = undefined;225    var chunk: [16 * 1024]u8 = undefined;226    var reader = file.reader(sys.fs.debugIo(), &reader_buffer);227    const read_bound = std.math.add(usize, expected_bytes, 1) catch228        return error.CapacityOverflow;229    for (0..read_bound) |_| {230        std.debug.assert(state.facts.bytes <= expected_bytes);231        const count = reader.interface.readSliceShort(&chunk) catch |err| switch (err) {232            error.ReadFailed => return reader.err orelse error.ReadFailed,233        };234        std.debug.assert(count <= chunk.len);235        if (count == 0) return;236        if (count > expected_bytes - state.facts.bytes) return error.InputChanged;237        const bytes = chunk[0..count];238        std.debug.assert(bytes.len > 0);239        state.hasher.update(bytes);240        state.facts.bytes = std.math.add(usize, state.facts.bytes, count) catch241            return error.CapacityOverflow;242        if (!source.kind.isAllocation()) try admitSourceBytes(state.facts.bytes);243        for (bytes) |byte| {244            std.debug.assert(state.line.bytes <= state.facts.bytes);245            if (byte == '\n') {246                if (source.kind.isAllocation()) try admitAllocationLine(state.line.bytes, true);247                state.line_number = std.math.add(usize, state.line_number, 1) catch248                    return error.CapacityOverflow;249                try finishLine(250                    &state.facts,251                    &state.line,252                    line_storage,253                    state.line_number,254                    context,255                    visitCandidate,256                );257                state.line = .{};258                std.debug.assert(state.line.bytes == 0);259                continue;260            }261            if (line_storage) |storage| {262                if (state.line.bytes >= storage.len) return error.InputChanged;263                storage[state.line.bytes] = byte;264            }265            try state.line.observe(byte);266            std.debug.assert(state.line.bytes <= state.facts.bytes);267            if (source.kind.isAllocation() and268                state.line.bytes > model.allocation_reader_bytes)269            {270                return error.StreamTooLong;271            }272        }273    }274    return error.InputChanged;275}276277fn finishLine(278    facts: *model.SourceFacts,279    line: *const LineInspection,280    line_storage: ?[]u8,281    line_number: usize,282    context: anytype,283    comptime visitCandidate: anytype,284) !void {285    std.debug.assert(line.bytes <= facts.bytes);286    std.debug.assert(facts.candidates <= facts.lines);287    facts.lines = std.math.add(usize, facts.lines, 1) catch return error.CapacityOverflow;288    facts.max_line_bytes = @max(facts.max_line_bytes, line.bytes);289    if (!line.isCandidate()) return;290    facts.candidates = std.math.add(usize, facts.candidates, 1) catch291        return error.CapacityOverflow;292    facts.max_json_depth = @max(facts.max_json_depth, line.max_json_depth);293    facts.max_key_count = @max(facts.max_key_count, line.key_count);294    std.debug.assert(facts.candidates <= facts.lines);295    std.debug.assert(facts.max_json_depth <= facts.max_line_bytes);296    std.debug.assert(facts.max_key_count <= facts.max_line_bytes);297    if (line_storage) |storage| {298        std.debug.assert(line.bytes <= storage.len);299        const trimmed = std.mem.trim(u8, storage[0..line.bytes], " \t\r");300        std.debug.assert(trimmed.len >= 2);301        try visitCandidate(context, trimmed, line_number);302    }303}304305fn ignoreCandidate(_: void, _: []const u8, _: usize) error{}!void {}306307fn isTrimByte(byte: u8) bool {308    const result = byte == ' ' or byte == '\t' or byte == '\r';309    if (result) std.debug.assert(byte != '\n');310    return result;311}312313test "ingestion source and allocation line boundaries match the legacy readers" {314    try admitSourceBytes(model.max_source_bytes - 1);315    try std.testing.expectError(error.StreamTooLong, admitSourceBytes(model.max_source_bytes));316    try admitAllocationLine(model.allocation_reader_bytes - 1, true);317    try admitAllocationLine(model.allocation_reader_bytes, false);318    try std.testing.expectError(319        error.StreamTooLong,320        admitAllocationLine(model.allocation_reader_bytes, true),321    );322    try std.testing.expectError(323        error.StreamTooLong,324        admitAllocationLine(model.allocation_reader_bytes + 1, false),325    );326}

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