tiny.profiling.ingest.owner
Defined in ingest.
API (9)
Actions
Public operations.
ArtifactIngestion.activateArtifactIngestion.deinitArtifactIngestion.initArtifactIngestion.summaryArtifactIngestion.write
Types and contracts
Public types and contracts.
Values and defaults
Public values and defaults.
Source
Source: src/profiling/ingest/owner.zig
zig
const std = @import("std");const alloc_phase = @import("alloc_phase");const profiling = @import("../root.zig");const ingest = @import("root.zig");const Allocator = std.mem.Allocator;const OrdinaryContext = struct { row_bits: []u8, row_capacity: usize, scratch: ingest.classify.Scratch, source_kind: ingest.SourceKind = .stdout, candidate_index: usize = 0, valid_rows: usize = 0, parse_errors: usize = 0, max_benchmark_allocated_bytes: ?u64 = null,};const AllocationContext = struct { scratch: ingest.classify.Scratch, totals: profiling.allocation.Totals = .{}, parse_errors: usize = 0,};const Storage = struct { line: []u8, row_bits: []u8, scratch_bytes: []align(@alignOf(usize)) u8, scratch: ingest.classify.Scratch, fn deinit(self: Storage, allocator: Allocator) void { allocator.free(self.scratch_bytes); allocator.free(self.row_bits); allocator.free(self.line); }};const Prepared = struct { ordinary: OrdinaryContext, allocations: AllocationContext,};pub const ArtifactIngestion = struct { pub const claim: alloc_phase.capacity.Declaration = .{ .source = .{ .id = "profiling.artifact_ingestion", .kind = .phase_static, .limit_source = .caller, .storage = .{ .covered = &.{ .{ .id = "maximum_admitted_source_line_byte_buffer", .lifetime = .steady, .detail = "maximum admitted source-line byte buffer", }, .{ .id = "exact_ordinary_source_candidate_validity_bitmap", .lifetime = .steady, .detail = "exact ordinary-source candidate validity bitmap", }, .{ .id = "exact_json_stack_decoded_key_bytes_key_entries_and_object_frames", .lifetime = .steady, .detail = "exact JSON stack, decoded key bytes, key entries, and object frames", }, .{ .id = "inline_allocation_totals_and_ingestion_summary", .lifetime = .steady, .detail = "inline allocation totals and ingestion summary", }, }, .excluded = &.{ "caller-owned source paths, workload metadata, and run records", "source and structured-output file contents and filesystem resources", "stack-owned read, write, digest, and pending-path buffers", "the per-workload releasable arena outside the ingestion owner", }, }, .capacity = .{ .inputs = &.{}, .type_selectors = &.{}, .nodes = &.{ .{ .constant = 0 }, }, .assertions = &.{.{ .scope = .closure_total, .measure = .retained, .relation = .exact, .expression = 0, }}, }, .overload = .{ .kind = .reject_before_seal, .detail = "oversized or arithmetically unrepresentable input and initialization OOM fail before activation; changed input after activation fails before the pending output replaces the prior artifact", }, .risks = .{ .transitive = .{ .status = .open, .detail = "the helper chain and standard JSON scanner are source-reviewed and sealed, but no machine call-graph certificate excludes future policy allocator reacquisition", }, .foreign = .{ .status = .excluded, .detail = "filesystem handles, reads, writes, metadata, rename, and output storage cross the owner-local heap-storage claim and are explicitly excluded", }, }, .obligations = &.{ .{ .key = "profiling_ingestion_capacity_capacity_model", .role = .capacity_model }, .{ .key = "profiling_ingestion_capacity_overload", .role = .overload }, .{ .key = "profiling_ingestion_mutation", .role = .overload }, .{ .key = "profiling_ingestion_max_plus_one", .role = .overload }, .{ .key = "profiling_ingestion_sealed_transitive_risk", .role = .transitive_risk }, .{ .key = "profiling_ingestion_sealed_foreign_risk", .role = .foreign_risk }, .{ .key = "profiling_ingestion_oom_retry", .role = .overload }, .{ .key = "profiling_ingestion_differential", .role = .transitive_risk }, }, }, .bindings = .{ .owner = @This(), .seal = .{ .family = alloc_phase.capacity.selector(@This().activate), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, .teardown = .{ .family = alloc_phase.capacity.selector(@This().deinit), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, }, }; phase: alloc_phase.capacity.Phase, capacity: ingest.Capacity, paths: ingest.Paths, row_bits: []u8, line_storage: []u8, scratch_storage: []align(@alignOf(usize)) u8, scratch: ingest.classify.Scratch, totals: profiling.allocation.Totals, result: ingest.Summary, pub const Limits: type = ingest.Limits; pub const Capacity: type = ingest.Capacity; const Self = @This(); pub fn init(allocator: Allocator, limits: Limits) !Self { std.debug.assert(limits.facts.sources.len == ingest.model.source_count); const capacity = try Capacity.derive(limits); const storage = try acquireStorage(allocator, capacity); errdefer storage.deinit(allocator); const prepared = try prepare(limits, capacity, storage); const ordinary = prepared.ordinary; const allocations = prepared.allocations; if (ordinary.candidate_index != capacity.row_bits) return error.InputChanged; std.debug.assert(ordinary.valid_rows <= ordinary.candidate_index); std.debug.assert(ordinary.parse_errors <= ordinary.candidate_index); var source_count: usize = 0; for (capacity.facts.sources) |source| { if (source.present) source_count = try add(source_count, 1); } std.debug.assert(source_count <= ingest.model.source_count); const metric_rows = if (allocations.totals.saw) allocations.totals.metrics().len else 0; return .{ .phase = .initialization, .capacity = capacity, .paths = limits.paths, .row_bits = storage.row_bits, .line_storage = storage.line, .scratch_storage = storage.scratch_bytes, .scratch = storage.scratch, .totals = allocations.totals, .result = .{ .rows = try add(ordinary.valid_rows, metric_rows), .parse_errors = try add(ordinary.parse_errors, allocations.parse_errors), .sources = source_count, .max_benchmark_allocated_bytes = ordinary.max_benchmark_allocated_bytes, }, }; } pub fn activate(self: *Self) !void { if (self.phase != .initialization) return error.AlreadyActive; std.debug.assert(self.phase == .initialization); std.debug.assert(self.line_storage.len == self.capacity.line_bytes); std.debug.assert(self.row_bits.len == self.capacity.row_bit_bytes); if (!try ingest.inspection.metadataMatches(self.paths, self.capacity.facts)) { return error.InputChanged; } self.phase = .steady; std.debug.assert(self.phase == .steady); } pub fn write(self: *Self, workload: profiling.catalog.Workload) !ingest.Summary { if (self.phase != .steady) return error.NotActive; std.debug.assert(self.row_bits.len == self.capacity.row_bit_bytes); std.debug.assert(self.line_storage.len == self.capacity.line_bytes); std.debug.assert(self.scratch_storage.len == self.capacity.scratch_bytes); return ingest.output.write( self.paths, self.capacity.facts, workload, self.row_bits, self.capacity.row_bits, self.line_storage, self.totals, self.result, ); } pub fn summary(self: *const Self) ingest.Summary { if (self.phase != .steady) { @panic("artifact ingestion summary requested outside steady phase"); } std.debug.assert(self.result.sources <= ingest.model.source_count); return self.result; } pub fn deinit(self: *Self, allocator: Allocator) void { if (self.phase == .teardown) @panic("artifact ingestion teardown is terminal"); switch (self.phase) { .initialization => std.debug.assert(self.phase == .initialization), .steady => std.debug.assert(self.phase == .steady), .teardown => unreachable, } std.debug.assert(self.line_storage.len == self.capacity.line_bytes); std.debug.assert(self.row_bits.len == self.capacity.row_bit_bytes); std.debug.assert(self.scratch_storage.len == self.capacity.scratch_bytes); self.phase = .teardown; allocator.free(self.scratch_storage); allocator.free(self.row_bits); allocator.free(self.line_storage); self.scratch_storage = undefined; self.scratch = undefined; self.row_bits = undefined; self.line_storage = undefined; }};comptime { alloc_phase.capacity.requireAllocatorExactOwnerShape(ArtifactIngestion);}fn acquireStorage(allocator: Allocator, capacity: ingest.Capacity) !Storage { const line = try allocator.alloc(u8, capacity.line_bytes); errdefer allocator.free(line); const row_bits = try allocator.alloc(u8, capacity.row_bit_bytes); errdefer allocator.free(row_bits); const scratch_bytes = try allocator.alignedAlloc(u8, .of(usize), capacity.scratch_bytes); errdefer allocator.free(scratch_bytes); std.debug.assert(line.len == capacity.line_bytes); std.debug.assert(row_bits.len == capacity.row_bit_bytes); std.debug.assert(scratch_bytes.len == capacity.scratch_bytes); @memset(row_bits, 0); @memset(scratch_bytes, 0); const scratch = capacity.scratch(scratch_bytes); std.debug.assert(scratch.json_stack.len == capacity.json_stack_bytes); std.debug.assert(scratch.keys.len == capacity.key_entries); std.debug.assert(scratch.objects.len == capacity.object_frames); return .{ .line = line, .row_bits = row_bits, .scratch_bytes = scratch_bytes, .scratch = scratch, };}fn prepare(limits: ingest.Limits, capacity: ingest.Capacity, storage: Storage) !Prepared { var result = Prepared{ .ordinary = .{ .row_bits = storage.row_bits, .row_capacity = capacity.row_bits, .scratch = storage.scratch, }, .allocations = .{ .scratch = storage.scratch }, }; const sources = limits.paths.sources(); for (sources, 0..) |source, index| { std.debug.assert(index < ingest.model.source_count); result.ordinary.source_kind = source.kind; const actual = if (source.kind.isAllocation()) try ingest.inspection.visitSource( source, storage.line, &result.allocations, classifyAllocation, ) else try ingest.inspection.visitSource( source, storage.line, &result.ordinary, classifyOrdinary, ); if (!capacity.facts.sources[index].eql(actual)) return error.InputChanged; } return result;}fn classifyOrdinary(context: *OrdinaryContext, raw: []const u8, _: usize) !void { std.debug.assert(raw.len >= 2); std.debug.assert(context.candidate_index <= context.row_capacity); const index = context.candidate_index; context.candidate_index = try add(index, 1); if (index >= context.row_capacity) return error.CapacityExceeded; const parsed = ingest.classify.line(raw, context.scratch) catch return error.CapacityExceeded; if (!parsed.valid) { context.parse_errors = try add(context.parse_errors, 1); return; } if (context.source_kind == .bench_jsonl) { if (parsed.event.benchmarkAllocatedBytes()) |bytes| { if (bytes > 0) { context.max_benchmark_allocated_bytes = @max( context.max_benchmark_allocated_bytes orelse 0, bytes, ); } } } const byte_index = index / 8; const bit_index: u3 = @intCast(index % 8); std.debug.assert(byte_index < context.row_bits.len); context.row_bits[byte_index] |= @as(u8, 1) << bit_index; context.valid_rows = try add(context.valid_rows, 1);}fn classifyAllocation(context: *AllocationContext, raw: []const u8, _: usize) !void { std.debug.assert(raw.len >= 2); const parsed = ingest.classify.line(raw, context.scratch) catch return error.CapacityExceeded; if (!parsed.valid) { context.parse_errors = try add(context.parse_errors, 1); return; } ingest.classify.record(&context.totals, parsed.event);}fn add(left: usize, right: usize) error{CapacityOverflow}!usize { return std.math.add(usize, left, right) catch error.CapacityOverflow;}Source: src/profiling/ingest/root.zig:12
zig
pub const owner = @import("owner.zig");Audit
| Definitions | 10 |
|---|---|
| Public names | 19 |
| Members | 9 |
| Version | 26.7.0 |
| Revision | daab053ee433 |