tiny.coz.Profiler

1pub const Profiler = struct {2    registry: registry_mod.Registry = .{},3    sources: source_map.Index = .{},4    source_mutex: std.atomic.Mutex = .unlocked,5    delays: delay.Coordinator = .{},6    selected_line_address: std.atomic.Value(usize) = .init(0),7    next_line_address: std.atomic.Value(usize) = .init(0),8    sampling: SamplingCounters = .{},9    experiment_duration_ns: u64 = experiment.experiment_min_time_ns,1011    pub fn deinit(self: *Profiler, allocator: std.mem.Allocator) void {12        self.registry.deinit(allocator);13        self.sources.deinit(allocator);14        self.delays = .{};15        self.selected_line_address.store(0, .release);16        self.next_line_address.store(0, .release);17        self.sampling = .{};18        self.experiment_duration_ns = experiment.experiment_min_time_ns;19    }2021    pub fn getCounter(22        self: *Profiler,23        allocator: std.mem.Allocator,24        kind: abi.CounterKind,25        name: []const u8,26    ) !*abi.Counter {27        return self.registry.getCounter(allocator, kind, name);28    }2930    pub fn preBlock(self: *Profiler, thread: *delay.ThreadState) void {31        self.delays.preBlock(thread);32    }3334    pub fn catchUp(self: *Profiler, thread: *delay.ThreadState, wait: anytype) u64 {35        return self.delays.addDelays(thread, wait);36    }3738    pub fn postBlock(self: *Profiler, thread: *delay.ThreadState, skip_delays: bool) void {39        self.delays.postBlock(thread, skip_delays);40    }4142    pub fn creditSelectedHit(self: *Profiler, thread: *delay.ThreadState) void {43        self.delays.creditSelectedHit(thread);44    }4546    pub fn addSourceRange(47        self: *Profiler,48        allocator: std.mem.Allocator,49        filename: []const u8,50        line_no: u64,51        range: source_map.Interval,52    ) !*source_map.Line {53        lockMutex(&self.source_mutex);54        defer self.source_mutex.unlock();5556        return self.sources.addRange(allocator, filename, line_no, range);57    }5859    pub fn selectLine(self: *Profiler, line: ?*source_map.Line) void {60        self.selected_line_address.store(lineAddress(line), .release);61        self.clearNextLine();62    }6364    pub fn selectedLine(self: *const Profiler) ?*source_map.Line {65        return addressLine(self.selected_line_address.load(.acquire));66    }6768    pub fn clearNextLine(self: *Profiler) void {69        self.next_line_address.store(0, .release);70    }7172    pub fn nextLine(self: *const Profiler) ?*source_map.Line {73        return addressLine(self.next_line_address.load(.acquire));74    }7576    pub fn observeSample(self: *Profiler, thread: *delay.ThreadState, sample: source_map.Sample) source_map.Match {77        if (!self.source_mutex.tryLock()) return .{};78        defer self.source_mutex.unlock();79        return self.observeSampleLocked(thread, sample);80    }8182    fn observeSampleLocked(self: *Profiler, thread: *delay.ThreadState, sample: source_map.Sample) source_map.Match {83        const matched = self.sources.matchSample(sample, self.selectedLine());84        if (matched.line) |line| {85            line.addSample();86            if (self.delays.active()) {87                if (matched.selected_hit) self.creditSelectedHit(thread);88            } else if (!path_filter.isCozHeader(line.file.name)) {89                _ = self.next_line_address.cmpxchgStrong(0, lineAddress(line), .acq_rel, .acquire);90            }91        }92        return matched;93    }9495    pub fn observeResolvedSample(96        self: *Profiler,97        allocator: std.mem.Allocator,98        thread: *delay.ThreadState,99        sample: source_map.Sample,100        scope: debug_info.Scope,101    ) !source_map.Match {102        lockMutex(&self.source_mutex);103        defer self.source_mutex.unlock();104105        try debug_info.resolveSample(allocator, &self.sources, sample, scope);106        return self.observeSampleLocked(thread, sample);107    }108109    pub fn observePerfRecord(110        self: *Profiler,111        thread: *delay.ThreadState,112        record: perf.Record,113        callchain_scratch: []usize,114    ) !source_map.Match {115        try self.observePerfRecordKind(record);116        const sample = try sampleFromPerfRecord(record, callchain_scratch) orelse return .{};117        return self.observeSample(thread, sample);118    }119120    pub fn observeResolvedPerfRecord(121        self: *Profiler,122        allocator: std.mem.Allocator,123        thread: *delay.ThreadState,124        record: perf.Record,125        callchain_scratch: []usize,126        scope: debug_info.Scope,127    ) !source_map.Match {128        try self.observePerfRecordKind(record);129        const sample = try sampleFromPerfRecord(record, callchain_scratch) orelse return .{};130131        lockMutex(&self.source_mutex);132        defer self.source_mutex.unlock();133134        _ = debug_info.resolveSelfAddress(allocator, &self.sources, sample.ip, scope) catch |err| switch (err) {135            error.OutOfMemory => return err,136            else => null,137        };138139        return self.observeSampleLocked(thread, sample);140    }141142    pub fn samplingSnapshot(self: *const Profiler) SamplingSnapshot {143        return .{144            .record_count = self.sampling.record_count.load(.acquire),145            .sample_record_count = self.sampling.sample_record_count.load(.acquire),146            .lost_record_count = self.sampling.lost_record_count.load(.acquire),147            .lost_event_count = self.sampling.lost_event_count.load(.acquire),148            .lost_samples_record_count = self.sampling.lost_samples_record_count.load(.acquire),149            .lost_samples_count = self.sampling.lost_samples_count.load(.acquire),150            .throttle_record_count = self.sampling.throttle_record_count.load(.acquire),151            .unthrottle_record_count = self.sampling.unthrottle_record_count.load(.acquire),152        };153    }154155    fn observePerfRecordKind(self: *Profiler, record: perf.Record) !void {156        addAtomic(&self.sampling.record_count, 1);157        switch (record.recordType()) {158            .sample => addAtomic(&self.sampling.sample_record_count, 1),159            .lost => {160                addAtomic(&self.sampling.lost_record_count, 1);161                addAtomic(&self.sampling.lost_event_count, try record.getLostCount());162            },163            .lost_samples => {164                addAtomic(&self.sampling.lost_samples_record_count, 1);165                addAtomic(&self.sampling.lost_samples_count, try record.getLostCount());166            },167            .throttle => addAtomic(&self.sampling.throttle_record_count, 1),168            .unthrottle => addAtomic(&self.sampling.unthrottle_record_count, 1),169            else => {},170        }171    }172173    pub fn drainPerfRing(174        self: *Profiler,175        thread: *delay.ThreadState,176        reader: *perf.RingReader,177        record_scratch: []u8,178        callchain_scratch: []usize,179    ) !usize {180        var records: usize = 0;181        while (try reader.next(record_scratch)) |record| {182            _ = try self.observePerfRecord(thread, record, callchain_scratch);183            records += 1;184        }185        return records;186    }187188    pub fn drainResolvedPerfRing(189        self: *Profiler,190        allocator: std.mem.Allocator,191        thread: *delay.ThreadState,192        reader: *perf.RingReader,193        record_scratch: []u8,194        callchain_scratch: []usize,195        scope: debug_info.Scope,196    ) !usize {197        var records: usize = 0;198        while (try reader.next(record_scratch)) |record| {199            _ = try self.observeResolvedPerfRecord(allocator, thread, record, callchain_scratch, scope);200            records += 1;201        }202        return records;203    }204205    pub fn processPerfRing(206        self: *Profiler,207        thread: *delay.ThreadState,208        reader: *perf.RingReader,209        record_scratch: []u8,210        callchain_scratch: []usize,211        wait: anytype,212    ) !usize {213        const records = try self.drainPerfRing(thread, reader, record_scratch, callchain_scratch);214        _ = self.catchUp(thread, wait);215        return records;216    }217218    pub fn drainPerfEvent(219        self: *Profiler,220        thread: *delay.ThreadState,221        event: *perf.Event,222        record_scratch: []u8,223        callchain_scratch: []usize,224    ) !usize {225        var reader = event.ringReader() orelse return 0;226        const records = try self.drainPerfRing(thread, &reader, record_scratch, callchain_scratch);227        event.commitReader(reader);228        return records;229    }230231    pub fn processPerfEvent(232        self: *Profiler,233        thread: *delay.ThreadState,234        event: *perf.Event,235        record_scratch: []u8,236        callchain_scratch: []usize,237        wait: anytype,238    ) !usize {239        var reader = event.ringReader() orelse {240            _ = self.catchUp(thread, wait);241            return 0;242        };243        const records = try self.processPerfRing(thread, &reader, record_scratch, callchain_scratch, wait);244        event.commitReader(reader);245        return records;246    }247248    pub fn drainSampler(249        self: *Profiler,250        thread: *delay.ThreadState,251        sampler: *sampler_mod.Sampler,252        record_scratch: []u8,253        callchain_scratch: []usize,254    ) !usize {255        var reader = sampler.ringReader() orelse return 0;256        const records = try self.drainPerfRing(thread, &reader, record_scratch, callchain_scratch);257        sampler.commitReader(reader);258        return records;259    }260261    pub fn drainResolvedSampler(262        self: *Profiler,263        allocator: std.mem.Allocator,264        thread: *delay.ThreadState,265        sampler: *sampler_mod.Sampler,266        record_scratch: []u8,267        callchain_scratch: []usize,268        scope: debug_info.Scope,269    ) !usize {270        var reader = sampler.ringReader() orelse return 0;271        const records = try self.drainResolvedPerfRing(allocator, thread, &reader, record_scratch, callchain_scratch, scope);272        sampler.commitReader(reader);273        return records;274    }275276    pub fn processSampler(277        self: *Profiler,278        thread: *delay.ThreadState,279        sampler: *sampler_mod.Sampler,280        record_scratch: []u8,281        callchain_scratch: []usize,282        wait: sampler_mod.WaitFn,283    ) !usize {284        const paused_wait = sampler.pausingWait(wait);285        var reader = sampler.ringReader() orelse {286            _ = self.catchUp(thread, paused_wait);287            return 0;288        };289        const records = try self.processPerfRing(thread, &reader, record_scratch, callchain_scratch, paused_wait);290        sampler.commitReader(reader);291        return records;292    }293294    pub fn beginExperiment(295        self: *Profiler,296        allocator: std.mem.Allocator,297        selected: profile.Location,298        virtual_speedup: f64,299    ) !Run {300        const selected_file = try allocator.dupe(u8, selected.file);301        errdefer allocator.free(selected_file);302303        const throughput_snapshots = try self.registry.saveThroughputSnapshots(allocator);304        errdefer allocator.free(throughput_snapshots);305306        const latency_snapshots = try self.registry.saveLatencySnapshots(allocator);307        errdefer allocator.free(latency_snapshots);308309        return .{310            .selected = .{311                .file = selected_file,312                .line = selected.line,313            },314            .virtual_speedup = virtual_speedup,315            .throughput_snapshots = throughput_snapshots,316            .latency_snapshots = latency_snapshots,317        };318    }319320    pub fn finishExperiment(321        self: *Profiler,322        writer: *std.Io.Writer,323        run: Run,324        duration_ns: u64,325        selected_samples: u64,326    ) !bool {327        const min_delta = run.minDelta();328        self.experiment_duration_ns = experiment.adjustDuration(self.experiment_duration_ns, min_delta);329        if (min_delta < experiment.experiment_target_delta) return false;330331        try run.writeEvents(writer, duration_ns, selected_samples);332        return true;333    }334335    pub fn runExperiment(336        self: *Profiler,337        allocator: std.mem.Allocator,338        writer: *std.Io.Writer,339        options: ExperimentRunOptions,340        wait_fn: sampler_mod.WaitFn,341    ) !bool {342        var run = try self.beginExperiment(343            allocator,344            options.selected.location(),345            options.plan.virtual_speedup,346        );347        defer run.deinit(allocator);348349        const starting_samples = options.selected.sampleCount();350        const starting_delay_ns = self.delays.globalDelay();351352        self.selectLine(options.selected);353        self.delays.startExperiment(options.plan.delay_size_ns);354        const elapsed_ns = waitForExperiment(options.plan.duration_ns, options.end_to_end, options.running, wait_fn);355        self.delays.finishExperiment();356        self.selectLine(null);357358        const inserted_delay_ns = self.delays.globalDelay() -| starting_delay_ns;359        const selected_samples = options.selected.sampleCount() -| starting_samples;360        const duration_ns = experiment.correctedDurationNs(elapsed_ns, inserted_delay_ns, self.delays.overshoot());361362        return self.finishExperiment(writer, run, duration_ns, selected_samples);363    }364365    pub fn runExperimentStep(366        self: *Profiler,367        allocator: std.mem.Allocator,368        writer: *std.Io.Writer,369        options: ExperimentStepOptions,370        wait_fn: sampler_mod.WaitFn,371    ) !ExperimentStepResult {372        const selected = options.fixed_line orelse self.nextLine() orelse return .{};373        const plan = try self.stepPlan(options);374        const emitted = try self.runExperiment(375            allocator,376            writer,377            .{378                .selected = selected,379                .plan = plan,380                .end_to_end = options.end_to_end,381                .running = options.running,382            },383            wait_fn,384        );385        return .{ .selected = selected, .emitted = emitted };386    }387388    fn stepPlan(self: *const Profiler, options: ExperimentStepOptions) !experiment.Plan {389        const delay_size_ns = if (options.fixed_speedup_percent) |percent|390            experiment.fixedDelaySize(percent) orelse return error.InvalidFixedSpeedup391        else blk: {392            const draw = try experiment.Draw.init(options.draw.value);393            break :blk experiment.delaySizeFromDraw(draw);394        };395396        return .{397            .delay_size_ns = delay_size_ns,398            .virtual_speedup = experiment.virtualSpeedupFromDelay(delay_size_ns),399            .duration_ns = self.experiment_duration_ns,400        };401    }402403    pub fn writeStartup(_: *Profiler, writer: *std.Io.Writer, timestamp_ns: u64) !void {404        try (profile.Event{ .startup = .{ .timestamp_ns = timestamp_ns } }).writeJsonLine(writer);405    }406407    pub fn writeRuntime(_: *Profiler, writer: *std.Io.Writer, duration_ns: u64) !void {408        try (profile.Event{ .runtime = .{ .duration_ns = duration_ns } }).writeJsonLine(writer);409    }410411    pub fn writeSample(_: *Profiler, writer: *std.Io.Writer, location: profile.Location, count: u64) !void {412        try (profile.Event{ .sample = .{ .location = location, .count = count } }).writeJsonLine(writer);413    }414415    pub fn writeRuntimeAndSamples(416        self: *Profiler,417        allocator: std.mem.Allocator,418        writer: *std.Io.Writer,419        duration_ns: u64,420        loss_counter: profile.LossCounter,421        terminal_status: profile.TerminalStatus,422    ) !void {423        const sample_lines = try self.collectSampleLines(allocator);424        defer allocator.free(sample_lines);425426        try self.writeRuntime(writer, duration_ns);427        try (profile.Event{ .sampling = self.profileSampling(428            loss_counter,429            terminal_status,430        ) }).writeJsonLine(writer);431        for (sample_lines) |line| {432            try self.writeSample(writer, line.location(), line.sampleCount());433        }434    }435436    fn profileSampling(437        self: *const Profiler,438        loss_counter: profile.LossCounter,439        terminal_status: profile.TerminalStatus,440    ) profile.Sampling {441        const snapshot = self.samplingSnapshot();442        return .{443            .record_count = snapshot.record_count,444            .sample_record_count = snapshot.sample_record_count,445            .lost_record_count = snapshot.lost_record_count,446            .lost_event_count = snapshot.lost_event_count,447            .lost_samples_record_count = snapshot.lost_samples_record_count,448            .lost_samples_count = snapshot.lost_samples_count,449            .throttle_record_count = snapshot.throttle_record_count,450            .unthrottle_record_count = snapshot.unthrottle_record_count,451            .loss_counter = loss_counter,452            .terminal_status = terminal_status,453        };454    }455456    fn collectSampleLines(self: *Profiler, allocator: std.mem.Allocator) ![]*source_map.Line {457        var sample_lines: std.ArrayListUnmanaged(*source_map.Line) = .empty;458        errdefer sample_lines.deinit(allocator);459460        lockMutex(&self.source_mutex);461        defer self.source_mutex.unlock();462463        var file_iter = self.sources.files.valueIterator();464        while (file_iter.next()) |file| {465            var line_iter = file.*.lines.valueIterator();466            while (line_iter.next()) |line| {467                if (line.*.sampleCount() != 0) try sample_lines.append(allocator, line.*);468            }469        }470471        std.mem.sort(*source_map.Line, sample_lines.items, {}, sampleLineLessThan);472        return try sample_lines.toOwnedSlice(allocator);473    }474};

1pub const Profiler = profiler.Profiler;