lib/coz/src/profiler.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sys = @import("sys");
3
4 const abi = @import("abi.zig");
5 const debug_info = @import("debug.zig");
6 const delay = @import("delay.zig");
7 const experiment = @import("experiment.zig");
8 const path_filter = @import("filter.zig");
9 const perf = sys.perf;
10 const profile = @import("profile.zig");
11 const progress_point = @import("point.zig");
12 const registry_mod = @import("registry.zig");
13 const sampler_mod = @import("sampler.zig");
14 const source_map = @import("map.zig");
15
16 pub const ExperimentRunOptions = struct {
17 selected: *source_map.Line,
18 plan: experiment.Plan,
19 end_to_end: bool = false,
20 running: ?*std.atomic.Value(bool) = null,
21 };
22
23 pub const ExperimentStepOptions = struct {
24 fixed_line: ?*source_map.Line = null,
25 fixed_speedup_percent: ?i32 = null,
26 draw: experiment.Draw = .{ .value = 0 },
27 end_to_end: bool = false,
28 running: ?*std.atomic.Value(bool) = null,
29 };
30
31 pub const ExperimentStepResult = struct {
32 selected: ?*source_map.Line = null,
33 emitted: bool = false,
34 };
35
36 pub const SamplingSnapshot = struct {
37 record_count: u64,
38 sample_record_count: u64,
39 lost_record_count: u64,
40 lost_event_count: u64,
41 lost_samples_record_count: u64,
42 lost_samples_count: u64,
43 throttle_record_count: u64,
44 unthrottle_record_count: u64,
45 };
46
47 const SamplingCounters = struct {
48 record_count: std.atomic.Value(u64) = .init(0),
49 sample_record_count: std.atomic.Value(u64) = .init(0),
50 lost_record_count: std.atomic.Value(u64) = .init(0),
51 lost_event_count: std.atomic.Value(u64) = .init(0),
52 lost_samples_record_count: std.atomic.Value(u64) = .init(0),
53 lost_samples_count: std.atomic.Value(u64) = .init(0),
54 throttle_record_count: std.atomic.Value(u64) = .init(0),
55 unthrottle_record_count: std.atomic.Value(u64) = .init(0),
56 };
57
58 pub const Profiler = struct {
59 registry: registry_mod.Registry = .{},
60 sources: source_map.Index = .{},
61 source_mutex: std.atomic.Mutex = .unlocked,
62 delays: delay.Coordinator = .{},
63 selected_line_address: std.atomic.Value(usize) = .init(0),
64 next_line_address: std.atomic.Value(usize) = .init(0),
65 sampling: SamplingCounters = .{},
66 experiment_duration_ns: u64 = experiment.experiment_min_time_ns,
67
68 pub fn deinit(self: *Profiler, allocator: std.mem.Allocator) void {
69 self.registry.deinit(allocator);
70 self.sources.deinit(allocator);
71 self.delays = .{};
72 self.selected_line_address.store(0, .release);
73 self.next_line_address.store(0, .release);
74 self.sampling = .{};
75 self.experiment_duration_ns = experiment.experiment_min_time_ns;
76 }
77
78 pub fn getCounter(
79 self: *Profiler,
80 allocator: std.mem.Allocator,
81 kind: abi.CounterKind,
82 name: []const u8,
83 ) !*abi.Counter {
84 return self.registry.getCounter(allocator, kind, name);
85 }
86
87 pub fn preBlock(self: *Profiler, thread: *delay.ThreadState) void {
88 self.delays.preBlock(thread);
89 }
90
91 pub fn catchUp(self: *Profiler, thread: *delay.ThreadState, wait: anytype) u64 {
92 return self.delays.addDelays(thread, wait);
93 }
94
95 pub fn postBlock(self: *Profiler, thread: *delay.ThreadState, skip_delays: bool) void {
96 self.delays.postBlock(thread, skip_delays);
97 }
98
99 pub fn creditSelectedHit(self: *Profiler, thread: *delay.ThreadState) void {
100 self.delays.creditSelectedHit(thread);
101 }
102
103 pub fn addSourceRange(
104 self: *Profiler,
105 allocator: std.mem.Allocator,
106 filename: []const u8,
107 line_no: u64,
108 range: source_map.Interval,
109 ) !*source_map.Line {
110 lockMutex(&self.source_mutex);
111 defer self.source_mutex.unlock();
112
113 return self.sources.addRange(allocator, filename, line_no, range);
114 }
115
116 pub fn selectLine(self: *Profiler, line: ?*source_map.Line) void {
117 self.selected_line_address.store(lineAddress(line), .release);
118 self.clearNextLine();
119 }
120
121 pub fn selectedLine(self: *const Profiler) ?*source_map.Line {
122 return addressLine(self.selected_line_address.load(.acquire));
123 }
124
125 pub fn clearNextLine(self: *Profiler) void {
126 self.next_line_address.store(0, .release);
127 }
128
129 pub fn nextLine(self: *const Profiler) ?*source_map.Line {
130 return addressLine(self.next_line_address.load(.acquire));
131 }
132
133 pub fn observeSample(self: *Profiler, thread: *delay.ThreadState, sample: source_map.Sample) source_map.Match {
134 if (!self.source_mutex.tryLock()) return .{};
135 defer self.source_mutex.unlock();
136 return self.observeSampleLocked(thread, sample);
137 }
138
139 fn observeSampleLocked(self: *Profiler, thread: *delay.ThreadState, sample: source_map.Sample) source_map.Match {
140 const matched = self.sources.matchSample(sample, self.selectedLine());
141 if (matched.line) |line| {
142 line.addSample();
143 if (self.delays.active()) {
144 if (matched.selected_hit) self.creditSelectedHit(thread);
145 } else if (!path_filter.isCozHeader(line.file.name)) {
146 _ = self.next_line_address.cmpxchgStrong(0, lineAddress(line), .acq_rel, .acquire);
147 }
148 }
149 return matched;
150 }
151
152 pub fn observeResolvedSample(
153 self: *Profiler,
154 allocator: std.mem.Allocator,
155 thread: *delay.ThreadState,
156 sample: source_map.Sample,
157 scope: debug_info.Scope,
158 ) !source_map.Match {
159 lockMutex(&self.source_mutex);
160 defer self.source_mutex.unlock();
161
162 try debug_info.resolveSample(allocator, &self.sources, sample, scope);
163 return self.observeSampleLocked(thread, sample);
164 }
165
166 pub fn observePerfRecord(
167 self: *Profiler,
168 thread: *delay.ThreadState,
169 record: perf.Record,
170 callchain_scratch: []usize,
171 ) !source_map.Match {
172 try self.observePerfRecordKind(record);
173 const sample = try sampleFromPerfRecord(record, callchain_scratch) orelse return .{};
174 return self.observeSample(thread, sample);
175 }
176
177 pub fn observeResolvedPerfRecord(
178 self: *Profiler,
179 allocator: std.mem.Allocator,
180 thread: *delay.ThreadState,
181 record: perf.Record,
182 callchain_scratch: []usize,
183 scope: debug_info.Scope,
184 ) !source_map.Match {
185 try self.observePerfRecordKind(record);
186 const sample = try sampleFromPerfRecord(record, callchain_scratch) orelse return .{};
187
188 lockMutex(&self.source_mutex);
189 defer self.source_mutex.unlock();
190
191 _ = debug_info.resolveSelfAddress(allocator, &self.sources, sample.ip, scope) catch |err| switch (err) {
192 error.OutOfMemory => return err,
193 else => null,
194 };
195
196 return self.observeSampleLocked(thread, sample);
197 }
198
199 pub fn samplingSnapshot(self: *const Profiler) SamplingSnapshot {
200 return .{
201 .record_count = self.sampling.record_count.load(.acquire),
202 .sample_record_count = self.sampling.sample_record_count.load(.acquire),
203 .lost_record_count = self.sampling.lost_record_count.load(.acquire),
204 .lost_event_count = self.sampling.lost_event_count.load(.acquire),
205 .lost_samples_record_count = self.sampling.lost_samples_record_count.load(.acquire),
206 .lost_samples_count = self.sampling.lost_samples_count.load(.acquire),
207 .throttle_record_count = self.sampling.throttle_record_count.load(.acquire),
208 .unthrottle_record_count = self.sampling.unthrottle_record_count.load(.acquire),
209 };
210 }
211
212 fn observePerfRecordKind(self: *Profiler, record: perf.Record) !void {
213 addAtomic(&self.sampling.record_count, 1);
214 switch (record.recordType()) {
215 .sample => addAtomic(&self.sampling.sample_record_count, 1),
216 .lost => {
217 addAtomic(&self.sampling.lost_record_count, 1);
218 addAtomic(&self.sampling.lost_event_count, try record.getLostCount());
219 },
220 .lost_samples => {
221 addAtomic(&self.sampling.lost_samples_record_count, 1);
222 addAtomic(&self.sampling.lost_samples_count, try record.getLostCount());
223 },
224 .throttle => addAtomic(&self.sampling.throttle_record_count, 1),
225 .unthrottle => addAtomic(&self.sampling.unthrottle_record_count, 1),
226 else => {},
227 }
228 }
229
230 pub fn drainPerfRing(
231 self: *Profiler,
232 thread: *delay.ThreadState,
233 reader: *perf.RingReader,
234 record_scratch: []u8,
235 callchain_scratch: []usize,
236 ) !usize {
237 var records: usize = 0;
238 while (try reader.next(record_scratch)) |record| {
239 _ = try self.observePerfRecord(thread, record, callchain_scratch);
240 records += 1;
241 }
242 return records;
243 }
244
245 pub fn drainResolvedPerfRing(
246 self: *Profiler,
247 allocator: std.mem.Allocator,
248 thread: *delay.ThreadState,
249 reader: *perf.RingReader,
250 record_scratch: []u8,
251 callchain_scratch: []usize,
252 scope: debug_info.Scope,
253 ) !usize {
254 var records: usize = 0;
255 while (try reader.next(record_scratch)) |record| {
256 _ = try self.observeResolvedPerfRecord(allocator, thread, record, callchain_scratch, scope);
257 records += 1;
258 }
259 return records;
260 }
261
262 pub fn processPerfRing(
263 self: *Profiler,
264 thread: *delay.ThreadState,
265 reader: *perf.RingReader,
266 record_scratch: []u8,
267 callchain_scratch: []usize,
268 wait: anytype,
269 ) !usize {
270 const records = try self.drainPerfRing(thread, reader, record_scratch, callchain_scratch);
271 _ = self.catchUp(thread, wait);
272 return records;
273 }
274
275 pub fn drainPerfEvent(
276 self: *Profiler,
277 thread: *delay.ThreadState,
278 event: *perf.Event,
279 record_scratch: []u8,
280 callchain_scratch: []usize,
281 ) !usize {
282 var reader = event.ringReader() orelse return 0;
283 const records = try self.drainPerfRing(thread, &reader, record_scratch, callchain_scratch);
284 event.commitReader(reader);
285 return records;
286 }
287
288 pub fn processPerfEvent(
289 self: *Profiler,
290 thread: *delay.ThreadState,
291 event: *perf.Event,
292 record_scratch: []u8,
293 callchain_scratch: []usize,
294 wait: anytype,
295 ) !usize {
296 var reader = event.ringReader() orelse {
297 _ = self.catchUp(thread, wait);
298 return 0;
299 };
300 const records = try self.processPerfRing(thread, &reader, record_scratch, callchain_scratch, wait);
301 event.commitReader(reader);
302 return records;
303 }
304
305 pub fn drainSampler(
306 self: *Profiler,
307 thread: *delay.ThreadState,
308 sampler: *sampler_mod.Sampler,
309 record_scratch: []u8,
310 callchain_scratch: []usize,
311 ) !usize {
312 var reader = sampler.ringReader() orelse return 0;
313 const records = try self.drainPerfRing(thread, &reader, record_scratch, callchain_scratch);
314 sampler.commitReader(reader);
315 return records;
316 }
317
318 pub fn drainResolvedSampler(
319 self: *Profiler,
320 allocator: std.mem.Allocator,
321 thread: *delay.ThreadState,
322 sampler: *sampler_mod.Sampler,
323 record_scratch: []u8,
324 callchain_scratch: []usize,
325 scope: debug_info.Scope,
326 ) !usize {
327 var reader = sampler.ringReader() orelse return 0;
328 const records = try self.drainResolvedPerfRing(allocator, thread, &reader, record_scratch, callchain_scratch, scope);
329 sampler.commitReader(reader);
330 return records;
331 }
332
333 pub fn processSampler(
334 self: *Profiler,
335 thread: *delay.ThreadState,
336 sampler: *sampler_mod.Sampler,
337 record_scratch: []u8,
338 callchain_scratch: []usize,
339 wait: sampler_mod.WaitFn,
340 ) !usize {
341 const paused_wait = sampler.pausingWait(wait);
342 var reader = sampler.ringReader() orelse {
343 _ = self.catchUp(thread, paused_wait);
344 return 0;
345 };
346 const records = try self.processPerfRing(thread, &reader, record_scratch, callchain_scratch, paused_wait);
347 sampler.commitReader(reader);
348 return records;
349 }
350
351 pub fn beginExperiment(
352 self: *Profiler,
353 allocator: std.mem.Allocator,
354 selected: profile.Location,
355 virtual_speedup: f64,
356 ) !Run {
357 const selected_file = try allocator.dupe(u8, selected.file);
358 errdefer allocator.free(selected_file);
359
360 const throughput_snapshots = try self.registry.saveThroughputSnapshots(allocator);
361 errdefer allocator.free(throughput_snapshots);
362
363 const latency_snapshots = try self.registry.saveLatencySnapshots(allocator);
364 errdefer allocator.free(latency_snapshots);
365
366 return .{
367 .selected = .{
368 .file = selected_file,
369 .line = selected.line,
370 },
371 .virtual_speedup = virtual_speedup,
372 .throughput_snapshots = throughput_snapshots,
373 .latency_snapshots = latency_snapshots,
374 };
375 }
376
377 pub fn finishExperiment(
378 self: *Profiler,
379 writer: *std.Io.Writer,
380 run: Run,
381 duration_ns: u64,
382 selected_samples: u64,
383 ) !bool {
384 const min_delta = run.minDelta();
385 self.experiment_duration_ns = experiment.adjustDuration(self.experiment_duration_ns, min_delta);
386 if (min_delta < experiment.experiment_target_delta) return false;
387
388 try run.writeEvents(writer, duration_ns, selected_samples);
389 return true;
390 }
391
392 pub fn runExperiment(
393 self: *Profiler,
394 allocator: std.mem.Allocator,
395 writer: *std.Io.Writer,
396 options: ExperimentRunOptions,
397 wait_fn: sampler_mod.WaitFn,
398 ) !bool {
399 var run = try self.beginExperiment(
400 allocator,
401 options.selected.location(),
402 options.plan.virtual_speedup,
403 );
404 defer run.deinit(allocator);
405
406 const starting_samples = options.selected.sampleCount();
407 const starting_delay_ns = self.delays.globalDelay();
408
409 self.selectLine(options.selected);
410 self.delays.startExperiment(options.plan.delay_size_ns);
411 const elapsed_ns = waitForExperiment(options.plan.duration_ns, options.end_to_end, options.running, wait_fn);
412 self.delays.finishExperiment();
413 self.selectLine(null);
414
415 const inserted_delay_ns = self.delays.globalDelay() -| starting_delay_ns;
416 const selected_samples = options.selected.sampleCount() -| starting_samples;
417 const duration_ns = experiment.correctedDurationNs(elapsed_ns, inserted_delay_ns, self.delays.overshoot());
418
419 return self.finishExperiment(writer, run, duration_ns, selected_samples);
420 }
421
422 pub fn runExperimentStep(
423 self: *Profiler,
424 allocator: std.mem.Allocator,
425 writer: *std.Io.Writer,
426 options: ExperimentStepOptions,
427 wait_fn: sampler_mod.WaitFn,
428 ) !ExperimentStepResult {
429 const selected = options.fixed_line orelse self.nextLine() orelse return .{};
430 const plan = try self.stepPlan(options);
431 const emitted = try self.runExperiment(
432 allocator,
433 writer,
434 .{
435 .selected = selected,
436 .plan = plan,
437 .end_to_end = options.end_to_end,
438 .running = options.running,
439 },
440 wait_fn,
441 );
442 return .{ .selected = selected, .emitted = emitted };
443 }
444
445 fn stepPlan(self: *const Profiler, options: ExperimentStepOptions) !experiment.Plan {
446 const delay_size_ns = if (options.fixed_speedup_percent) |percent|
447 experiment.fixedDelaySize(percent) orelse return error.InvalidFixedSpeedup
448 else blk: {
449 const draw = try experiment.Draw.init(options.draw.value);
450 break :blk experiment.delaySizeFromDraw(draw);
451 };
452
453 return .{
454 .delay_size_ns = delay_size_ns,
455 .virtual_speedup = experiment.virtualSpeedupFromDelay(delay_size_ns),
456 .duration_ns = self.experiment_duration_ns,
457 };
458 }
459
460 pub fn writeStartup(_: *Profiler, writer: *std.Io.Writer, timestamp_ns: u64) !void {
461 try (profile.Event{ .startup = .{ .timestamp_ns = timestamp_ns } }).writeJsonLine(writer);
462 }
463
464 pub fn writeRuntime(_: *Profiler, writer: *std.Io.Writer, duration_ns: u64) !void {
465 try (profile.Event{ .runtime = .{ .duration_ns = duration_ns } }).writeJsonLine(writer);
466 }
467
468 pub fn writeSample(_: *Profiler, writer: *std.Io.Writer, location: profile.Location, count: u64) !void {
469 try (profile.Event{ .sample = .{ .location = location, .count = count } }).writeJsonLine(writer);
470 }
471
472 pub fn writeRuntimeAndSamples(
473 self: *Profiler,
474 allocator: std.mem.Allocator,
475 writer: *std.Io.Writer,
476 duration_ns: u64,
477 loss_counter: profile.LossCounter,
478 terminal_status: profile.TerminalStatus,
479 ) !void {
480 const sample_lines = try self.collectSampleLines(allocator);
481 defer allocator.free(sample_lines);
482
483 try self.writeRuntime(writer, duration_ns);
484 try (profile.Event{ .sampling = self.profileSampling(
485 loss_counter,
486 terminal_status,
487 ) }).writeJsonLine(writer);
488 for (sample_lines) |line| {
489 try self.writeSample(writer, line.location(), line.sampleCount());
490 }
491 }
492
493 fn profileSampling(
494 self: *const Profiler,
495 loss_counter: profile.LossCounter,
496 terminal_status: profile.TerminalStatus,
497 ) profile.Sampling {
498 const snapshot = self.samplingSnapshot();
499 return .{
500 .record_count = snapshot.record_count,
501 .sample_record_count = snapshot.sample_record_count,
502 .lost_record_count = snapshot.lost_record_count,
503 .lost_event_count = snapshot.lost_event_count,
504 .lost_samples_record_count = snapshot.lost_samples_record_count,
505 .lost_samples_count = snapshot.lost_samples_count,
506 .throttle_record_count = snapshot.throttle_record_count,
507 .unthrottle_record_count = snapshot.unthrottle_record_count,
508 .loss_counter = loss_counter,
509 .terminal_status = terminal_status,
510 };
511 }
512
513 fn collectSampleLines(self: *Profiler, allocator: std.mem.Allocator) ![]*source_map.Line {
514 var sample_lines: std.ArrayListUnmanaged(*source_map.Line) = .empty;
515 errdefer sample_lines.deinit(allocator);
516
517 lockMutex(&self.source_mutex);
518 defer self.source_mutex.unlock();
519
520 var file_iter = self.sources.files.valueIterator();
521 while (file_iter.next()) |file| {
522 var line_iter = file.*.lines.valueIterator();
523 while (line_iter.next()) |line| {
524 if (line.*.sampleCount() != 0) try sample_lines.append(allocator, line.*);
525 }
526 }
527
528 std.mem.sort(*source_map.Line, sample_lines.items, {}, sampleLineLessThan);
529 return try sample_lines.toOwnedSlice(allocator);
530 }
531 };
532
533 pub const Run = struct {
534 selected: profile.Location,
535 virtual_speedup: f64,
536 throughput_snapshots: []progress_point.ThroughputSnapshot,
537 latency_snapshots: []progress_point.LatencySnapshot,
538
539 pub fn deinit(self: *Run, allocator: std.mem.Allocator) void {
540 allocator.free(self.selected.file);
541 allocator.free(self.throughput_snapshots);
542 allocator.free(self.latency_snapshots);
543 self.* = undefined;
544 }
545
546 pub fn minDelta(self: Run) u64 {
547 var result: u64 = std.math.maxInt(u64);
548 var found = false;
549
550 for (self.throughput_snapshots) |snapshot| {
551 applyDelta(countToU64(snapshot.getDelta()), &found, &result);
552 }
553 for (self.latency_snapshots) |snapshot| {
554 const arrivals = countToU64(snapshot.getBeginDelta());
555 const departures = countToU64(snapshot.getEndDelta());
556 if (arrivals != 0 and departures != 0) {
557 applyDelta(@min(arrivals, departures), &found, &result);
558 }
559 }
560
561 return if (found) result else 0;
562 }
563
564 pub fn writeEvents(
565 self: Run,
566 writer: *std.Io.Writer,
567 duration_ns: u64,
568 selected_samples: u64,
569 ) !void {
570 try (profile.Event{ .experiment = .{
571 .selected = self.selected,
572 .virtual_speedup = self.virtual_speedup,
573 .duration_ns = duration_ns,
574 .selected_samples = selected_samples,
575 } }).writeJsonLine(writer);
576
577 for (self.throughput_snapshots) |snapshot| {
578 const delta = countToU64(snapshot.getDelta());
579 if (delta == 0) continue;
580 try (profile.Event{ .throughput = .{
581 .name = snapshot.getName(),
582 .delta = delta,
583 } }).writeJsonLine(writer);
584 }
585
586 for (self.latency_snapshots) |snapshot| {
587 const arrivals = countToU64(snapshot.getBeginDelta());
588 const departures = countToU64(snapshot.getEndDelta());
589 if (arrivals == 0 or departures == 0) continue;
590 try (profile.Event{ .latency = .{
591 .name = snapshot.getName(),
592 .arrivals = arrivals,
593 .departures = departures,
594 .outstanding = countToU64(snapshot.getDifference()),
595 } }).writeJsonLine(writer);
596 }
597 }
598 };
599
600 fn applyDelta(delta: u64, found: *bool, result: *u64) void {
601 if (delta == 0) return;
602 found.* = true;
603 result.* = @min(result.*, delta);
604 }
605
606 fn countToU64(count: usize) u64 {
607 return @intCast(count);
608 }
609
610 fn addAtomic(counter: *std.atomic.Value(u64), delta: u64) void {
611 var current = counter.load(.monotonic);
612 while (true) {
613 const next = current +| delta;
614 if (counter.cmpxchgWeak(current, next, .monotonic, .monotonic)) |observed| {
615 current = observed;
616 } else {
617 return;
618 }
619 }
620 }
621
622 fn lockMutex(mutex: *std.atomic.Mutex) void {
623 while (!mutex.tryLock()) std.atomic.spinLoopHint();
624 }
625
626 fn lineAddress(line: ?*source_map.Line) usize {
627 return if (line) |selected| @intFromPtr(selected) else 0;
628 }
629
630 fn addressLine(address: usize) ?*source_map.Line {
631 if (address == 0) return null;
632 return @ptrFromInt(address);
633 }
634
635 fn sampleFromPerfRecord(record: perf.Record, callchain_scratch: []usize) !?source_map.Sample {
636 if (!record.isSample()) return null;
637
638 const ip = try addressToUsize(try record.getIp());
639 var callchain_len: usize = 0;
640 if (record.config.isSampling(.callchain)) {
641 const callchain = try record.getCallchain();
642 if (callchain.len() > callchain_scratch.len) return error.CallchainScratchTooSmall;
643 while (callchain_len < callchain.len()) : (callchain_len += 1) {
644 callchain_scratch[callchain_len] = try addressToUsize(callchain.at(callchain_len));
645 }
646 }
647
648 return .{ .ip = ip, .callchain = callchain_scratch[0..callchain_len] };
649 }
650
651 fn addressToUsize(address: u64) !usize {
652 return std.math.cast(usize, address) orelse error.AddressTooLarge;
653 }
654
655 fn waitForExperiment(
656 duration_ns: u64,
657 end_to_end: bool,
658 running: ?*std.atomic.Value(bool),
659 wait_fn: sampler_mod.WaitFn,
660 ) u64 {
661 const running_flag = running orelse return wait_fn(duration_ns);
662
663 if (end_to_end) {
664 var elapsed_ns: u64 = 0;
665 const interval_ns = experiment.experiment_cool_off_time_ns;
666 while (running_flag.load(.acquire)) {
667 elapsed_ns +|= wait_fn(interval_ns);
668 }
669 return elapsed_ns;
670 }
671
672 return waitDurationWhileRunning(duration_ns, running_flag, wait_fn);
673 }
674
675 fn waitDurationWhileRunning(
676 duration_ns: u64,
677 running: *std.atomic.Value(bool),
678 wait_fn: sampler_mod.WaitFn,
679 ) u64 {
680 var elapsed_ns: u64 = 0;
681 while (elapsed_ns < duration_ns and running.load(.acquire)) {
682 const remaining_ns = duration_ns - elapsed_ns;
683 const interval_ns = @min(remaining_ns, experiment.experiment_cool_off_time_ns);
684 const waited_ns = wait_fn(interval_ns);
685 elapsed_ns +|= if (waited_ns == 0 and interval_ns != 0) interval_ns else waited_ns;
686 }
687 return elapsed_ns;
688 }
689
690 fn sampleLineLessThan(_: void, lhs: *source_map.Line, rhs: *source_map.Line) bool {
691 const file_order = std.mem.order(u8, lhs.file.name, rhs.file.name);
692 if (file_order != .eq) return file_order == .lt;
693 return lhs.number < rhs.number;
694 }
695
696 fn exactWait(ns: u64) u64 {
697 return ns;
698 }
699
700 var experiment_test_counter: ?*abi.Counter = null;
701 var experiment_test_running: ?*std.atomic.Value(bool) = null;
702 var experiment_test_waits: std.atomic.Value(u32) = .init(0);
703
704 fn progressWait(ns: u64) u64 {
705 if (experiment_test_counter) |counter| {
706 _ = @atomicRmw(usize, &counter.count, .Add, 6, .monotonic);
707 }
708 return ns;
709 }
710
711 fn stoppingWait(ns: u64) u64 {
712 const waits = experiment_test_waits.fetchAdd(1, .monotonic);
713 if (waits == 1) {
714 if (experiment_test_running) |running| running.store(false, .release);
715 }
716 return ns;
717 }
718
719 fn writePerfU64(bytes: []u8, offset: *usize, value: u64) void {
720 std.mem.writeInt(u64, bytes[offset.*..][0..@sizeOf(u64)], value, .native);
721 offset.* += @sizeOf(u64);
722 }
723
724 fn finishPerfRecord(bytes: []u8, len: usize, record_type: perf.RecordType) []const u8 {
725 std.mem.writeInt(u32, bytes[0..4], @backingInt(record_type), .native);
726 std.mem.writeInt(u16, bytes[4..6], 0, .native);
727 std.mem.writeInt(u16, bytes[6..8], @intCast(len), .native);
728 return bytes[0..len];
729 }
730
731 fn finishPerfSample(bytes: []u8, len: usize) []const u8 {
732 return finishPerfRecord(bytes, len, .sample);
733 }
734
735 fn writeRingBytes(data: []u8, index: u64, bytes: []const u8) void {
736 const ring_len: u64 = @intCast(data.len);
737 for (bytes, 0..) |byte, offset| {
738 const ring_index: usize = @intCast((index + @as(u64, @intCast(offset))) % ring_len);
739 data[ring_index] = byte;
740 }
741 }
742
743 test "profiler writes structured experiment events from registry snapshots" {
744 var profiler: Profiler = .{};
745 defer profiler.deinit(std.testing.allocator);
746
747 const throughput_counter = try profiler.getCounter(std.testing.allocator, .throughput, "items");
748 const begin_counter = try profiler.getCounter(std.testing.allocator, .begin, "request");
749 const end_counter = try profiler.getCounter(std.testing.allocator, .end, "request");
750
751 var run = try profiler.beginExperiment(
752 std.testing.allocator,
753 .{ .file = "src/main.zig", .line = 12 },
754 0.25,
755 );
756 defer run.deinit(std.testing.allocator);
757
758 _ = @atomicRmw(usize, &throughput_counter.count, .Add, 6, .monotonic);
759 _ = @atomicRmw(usize, &begin_counter.count, .Add, 7, .monotonic);
760 _ = @atomicRmw(usize, &end_counter.count, .Add, 6, .monotonic);
761
762 var buffer: [1024]u8 = undefined;
763 var writer = std.Io.Writer.fixed(&buffer);
764
765 try std.testing.expect(try profiler.finishExperiment(&writer, run, 500_000_000, 17));
766 try std.testing.expectEqualStrings(
767 "{\"schema\":\"coz.profile/v1\",\"event\":\"experiment\",\"selected\":{\"file\":\"src/main.zig\",\"line\":12},\"virtual_speedup\":0.25,\"duration_ns\":500000000,\"selected_samples\":17}\n" ++
768 "{\"schema\":\"coz.profile/v1\",\"event\":\"throughput\",\"name\":\"items\",\"delta\":6}\n" ++
769 "{\"schema\":\"coz.profile/v1\",\"event\":\"latency\",\"name\":\"request\",\"arrivals\":7,\"departures\":6,\"outstanding\":1}\n",
770 writer.buffered(),
771 );
772 }
773
774 test "profiler suppresses low-delta experiment output and lengthens duration" {
775 var profiler: Profiler = .{};
776 defer profiler.deinit(std.testing.allocator);
777
778 const throughput_counter = try profiler.getCounter(std.testing.allocator, .throughput, "items");
779
780 var run = try profiler.beginExperiment(
781 std.testing.allocator,
782 .{ .file = "src/main.zig", .line = 12 },
783 0,
784 );
785 defer run.deinit(std.testing.allocator);
786
787 _ = @atomicRmw(usize, &throughput_counter.count, .Add, 1, .monotonic);
788
789 var buffer: [128]u8 = undefined;
790 var writer = std.Io.Writer.fixed(&buffer);
791
792 try std.testing.expect(!try profiler.finishExperiment(&writer, run, 500_000_000, 1));
793 try std.testing.expectEqual(@as(usize, 0), writer.buffered().len);
794 try std.testing.expectEqual(experiment.experiment_min_time_ns * 2, profiler.experiment_duration_ns);
795 }
796
797 test "profiler ignores inactive progress points when gating experiment output" {
798 var profiler: Profiler = .{};
799 defer profiler.deinit(std.testing.allocator);
800
801 const active_counter = try profiler.getCounter(std.testing.allocator, .throughput, "active");
802 _ = try profiler.getCounter(std.testing.allocator, .throughput, "idle-throughput");
803 _ = try profiler.getCounter(std.testing.allocator, .begin, "idle-latency");
804 _ = try profiler.getCounter(std.testing.allocator, .end, "idle-latency");
805
806 var run = try profiler.beginExperiment(
807 std.testing.allocator,
808 .{ .file = "src/main.zig", .line = 12 },
809 0,
810 );
811 defer run.deinit(std.testing.allocator);
812
813 _ = @atomicRmw(usize, &active_counter.count, .Add, 7, .monotonic);
814
815 var buffer: [512]u8 = undefined;
816 var writer = std.Io.Writer.fixed(&buffer);
817
818 try std.testing.expectEqual(@as(u64, 7), run.minDelta());
819 try std.testing.expect(try profiler.finishExperiment(&writer, run, 500_000_000, 1));
820 try std.testing.expectEqualStrings(
821 "{\"schema\":\"coz.profile/v1\",\"event\":\"experiment\",\"selected\":{\"file\":\"src/main.zig\",\"line\":12},\"virtual_speedup\":0,\"duration_ns\":500000000,\"selected_samples\":1}\n" ++
822 "{\"schema\":\"coz.profile/v1\",\"event\":\"throughput\",\"name\":\"active\",\"delta\":7}\n",
823 writer.buffered(),
824 );
825 }
826
827 test "profiler suppresses experiment output when no progress points exist" {
828 var profiler: Profiler = .{};
829 defer profiler.deinit(std.testing.allocator);
830
831 var run = try profiler.beginExperiment(
832 std.testing.allocator,
833 .{ .file = "src/main.zig", .line = 12 },
834 0,
835 );
836 defer run.deinit(std.testing.allocator);
837
838 var buffer: [128]u8 = undefined;
839 var writer = std.Io.Writer.fixed(&buffer);
840
841 try std.testing.expectEqual(@as(u64, 0), run.minDelta());
842 try std.testing.expect(!try profiler.finishExperiment(&writer, run, 500_000_000, 0));
843 try std.testing.expectEqual(@as(usize, 0), writer.buffered().len);
844 }
845
846 test "profiler runs an experiment cycle and clears selected state" {
847 var profiler: Profiler = .{};
848 defer profiler.deinit(std.testing.allocator);
849
850 const counter = try profiler.getCounter(std.testing.allocator, .throughput, "items");
851 const selected = try profiler.addSourceRange(std.testing.allocator, "/tmp/main.zig", 10, try source_map.Interval.init(100, 110));
852 experiment_test_counter = counter;
853 defer experiment_test_counter = null;
854
855 var buffer: [1024]u8 = undefined;
856 var writer = std.Io.Writer.fixed(&buffer);
857
858 try std.testing.expect(try profiler.runExperiment(
859 std.testing.allocator,
860 &writer,
861 .{
862 .selected = selected,
863 .plan = .{
864 .delay_size_ns = 0,
865 .virtual_speedup = 0,
866 .duration_ns = 500_000_000,
867 },
868 },
869 progressWait,
870 ));
871
872 try std.testing.expect(profiler.selectedLine() == null);
873 try std.testing.expect(!profiler.delays.active());
874 try std.testing.expectEqualStrings(
875 "{\"schema\":\"coz.profile/v1\",\"event\":\"experiment\",\"selected\":{\"file\":\"/tmp/main.zig\",\"line\":10},\"virtual_speedup\":0,\"duration_ns\":500000000,\"selected_samples\":0}\n" ++
876 "{\"schema\":\"coz.profile/v1\",\"event\":\"throughput\",\"name\":\"items\",\"delta\":6}\n",
877 writer.buffered(),
878 );
879 }
880
881 test "profiler experiment step uses the sampled next line" {
882 var profiler: Profiler = .{};
883 defer profiler.deinit(std.testing.allocator);
884 var thread: delay.ThreadState = .{};
885
886 const counter = try profiler.getCounter(std.testing.allocator, .throughput, "items");
887 const selected = try profiler.addSourceRange(std.testing.allocator, "/tmp/next.zig", 20, try source_map.Interval.init(100, 110));
888 _ = profiler.observeSample(&thread, .{ .ip = 105 });
889 experiment_test_counter = counter;
890 defer experiment_test_counter = null;
891
892 var buffer: [1024]u8 = undefined;
893 var writer = std.Io.Writer.fixed(&buffer);
894
895 const result = try profiler.runExperimentStep(
896 std.testing.allocator,
897 &writer,
898 .{ .draw = try experiment.Draw.init(12) },
899 progressWait,
900 );
901
902 try std.testing.expectEqual(selected, result.selected.?);
903 try std.testing.expect(result.emitted);
904 try std.testing.expect(profiler.selectedLine() == null);
905 try std.testing.expect(profiler.nextLine() == null);
906 try std.testing.expect(!profiler.delays.active());
907 try std.testing.expectEqualStrings(
908 "{\"schema\":\"coz.profile/v1\",\"event\":\"experiment\",\"selected\":{\"file\":\"/tmp/next.zig\",\"line\":20},\"virtual_speedup\":0.25,\"duration_ns\":500000000,\"selected_samples\":0}\n" ++
909 "{\"schema\":\"coz.profile/v1\",\"event\":\"throughput\",\"name\":\"items\",\"delta\":6}\n",
910 writer.buffered(),
911 );
912 }
913
914 test "profiler experiment step can use a fixed line and fixed speedup" {
915 var profiler: Profiler = .{};
916 defer profiler.deinit(std.testing.allocator);
917
918 const counter = try profiler.getCounter(std.testing.allocator, .throughput, "items");
919 const selected = try profiler.addSourceRange(std.testing.allocator, "/tmp/fixed.zig", 30, try source_map.Interval.init(200, 210));
920 experiment_test_counter = counter;
921 defer experiment_test_counter = null;
922
923 var buffer: [1024]u8 = undefined;
924 var writer = std.Io.Writer.fixed(&buffer);
925
926 const result = try profiler.runExperimentStep(
927 std.testing.allocator,
928 &writer,
929 .{
930 .fixed_line = selected,
931 .fixed_speedup_percent = 50,
932 },
933 progressWait,
934 );
935
936 try std.testing.expectEqual(selected, result.selected.?);
937 try std.testing.expect(result.emitted);
938 try std.testing.expectEqualStrings(
939 "{\"schema\":\"coz.profile/v1\",\"event\":\"experiment\",\"selected\":{\"file\":\"/tmp/fixed.zig\",\"line\":30},\"virtual_speedup\":0.5,\"duration_ns\":500000000,\"selected_samples\":0}\n" ++
940 "{\"schema\":\"coz.profile/v1\",\"event\":\"throughput\",\"name\":\"items\",\"delta\":6}\n",
941 writer.buffered(),
942 );
943 }
944
945 test "profiler experiment step is inert without a selectable line" {
946 var profiler: Profiler = .{};
947 defer profiler.deinit(std.testing.allocator);
948
949 var buffer: [128]u8 = undefined;
950 var writer = std.Io.Writer.fixed(&buffer);
951
952 const result = try profiler.runExperimentStep(
953 std.testing.allocator,
954 &writer,
955 .{},
956 exactWait,
957 );
958
959 try std.testing.expect(result.selected == null);
960 try std.testing.expect(!result.emitted);
961 try std.testing.expectEqual(@as(usize, 0), writer.buffered().len);
962 }
963
964 test "profiler experiment step rejects invalid fixed speedups" {
965 var profiler: Profiler = .{};
966 defer profiler.deinit(std.testing.allocator);
967
968 const selected = try profiler.addSourceRange(std.testing.allocator, "/tmp/fixed.zig", 30, try source_map.Interval.init(200, 210));
969
970 var buffer: [128]u8 = undefined;
971 var writer = std.Io.Writer.fixed(&buffer);
972
973 try std.testing.expectError(error.InvalidFixedSpeedup, profiler.runExperimentStep(
974 std.testing.allocator,
975 &writer,
976 .{
977 .fixed_line = selected,
978 .fixed_speedup_percent = 101,
979 },
980 exactWait,
981 ));
982 try std.testing.expectEqual(@as(usize, 0), writer.buffered().len);
983 try std.testing.expect(profiler.selectedLine() == null);
984 try std.testing.expect(!profiler.delays.active());
985 }
986
987 test "end-to-end experiment wait stops on running flag" {
988 var running: std.atomic.Value(bool) = .init(true);
989 experiment_test_running = &running;
990 experiment_test_waits.store(0, .monotonic);
991 defer experiment_test_running = null;
992
993 const elapsed_ns = waitForExperiment(123, true, &running, stoppingWait);
994
995 try std.testing.expectEqual(@as(u64, experiment.experiment_cool_off_time_ns * 2), elapsed_ns);
996 try std.testing.expectEqual(@as(u32, 2), experiment_test_waits.load(.monotonic));
997 }
998
999 test "duration experiment wait stops on running flag" {
1000 var running: std.atomic.Value(bool) = .init(true);
1001 experiment_test_running = &running;
1002 experiment_test_waits.store(0, .monotonic);
1003 defer experiment_test_running = null;
1004
1005 const elapsed_ns = waitForExperiment(experiment.experiment_cool_off_time_ns * 5, false, &running, stoppingWait);
1006
1007 try std.testing.expectEqual(@as(u64, experiment.experiment_cool_off_time_ns * 2), elapsed_ns);
1008 try std.testing.expectEqual(@as(u32, 2), experiment_test_waits.load(.monotonic));
1009 }
1010
1011 test "profiler writes startup runtime and sample events" {
1012 var profiler: Profiler = .{};
1013 defer profiler.deinit(std.testing.allocator);
1014
1015 var buffer: [512]u8 = undefined;
1016 var writer = std.Io.Writer.fixed(&buffer);
1017
1018 try profiler.writeStartup(&writer, 100);
1019 try profiler.writeRuntime(&writer, 200);
1020 try profiler.writeSample(&writer, .{ .file = "src/hot.zig", .line = 9 }, 3);
1021
1022 try std.testing.expectEqualStrings(
1023 "{\"schema\":\"coz.profile/v1\",\"event\":\"startup\",\"timestamp_ns\":100}\n" ++
1024 "{\"schema\":\"coz.profile/v1\",\"event\":\"runtime\",\"duration_ns\":200}\n" ++
1025 "{\"schema\":\"coz.profile/v1\",\"event\":\"sample\",\"location\":{\"file\":\"src/hot.zig\",\"line\":9},\"count\":3}\n",
1026 writer.buffered(),
1027 );
1028 }
1029
1030 test "profiler attributes a sample after debug symbol resolution" {
1031 var profiler: Profiler = .{};
1032 defer profiler.deinit(std.testing.allocator);
1033 var thread: delay.ThreadState = .{};
1034
1035 const symbols = [_]std.debug.Symbol{.{
1036 .name = "hot",
1037 .compile_unit_name = "unit",
1038 .source_location = .{
1039 .file_name = "/tmp/hot.zig",
1040 .line = 14,
1041 .column = 1,
1042 },
1043 }};
1044 _ = try debug_info.resolveSymbols(std.testing.allocator, &profiler.sources, 0x1000, &symbols, .{});
1045 const matched = profiler.observeSample(&thread, .{ .ip = 0x1000 });
1046
1047 const line = matched.line.?;
1048 try std.testing.expectEqualStrings("/tmp/hot.zig", line.file.name);
1049 try std.testing.expectEqual(@as(u64, 14), line.number);
1050 try std.testing.expectEqual(@as(u64, 1), line.sampleCount());
1051 try std.testing.expectEqual(line, profiler.nextLine().?);
1052 }
1053
1054 test "profiler writes runtime and sorted nonzero source samples" {
1055 var profiler: Profiler = .{};
1056 defer profiler.deinit(std.testing.allocator);
1057 var thread: delay.ThreadState = .{};
1058
1059 _ = try profiler.addSourceRange(std.testing.allocator, "/tmp/b.zig", 5, try source_map.Interval.init(100, 110));
1060 _ = try profiler.addSourceRange(std.testing.allocator, "/tmp/a.zig", 9, try source_map.Interval.init(200, 210));
1061 _ = try profiler.addSourceRange(std.testing.allocator, "/tmp/b.zig", 7, try source_map.Interval.init(300, 310));
1062
1063 _ = profiler.observeSample(&thread, .{ .ip = 105 });
1064 _ = profiler.observeSample(&thread, .{ .ip = 205 });
1065 _ = profiler.observeSample(&thread, .{ .ip = 106 });
1066
1067 var buffer: [1024]u8 = undefined;
1068 var writer = std.Io.Writer.fixed(&buffer);
1069
1070 try profiler.writeRuntimeAndSamples(
1071 std.testing.allocator,
1072 &writer,
1073 900,
1074 .unsupported,
1075 .not_started,
1076 );
1077
1078 try std.testing.expectEqualStrings(
1079 "{\"schema\":\"coz.profile/v1\",\"event\":\"runtime\",\"duration_ns\":900}\n" ++
1080 "{\"schema\":\"coz.profile/v1\",\"event\":\"sampling\"," ++
1081 "\"record_count\":0,\"sample_record_count\":0,\"lost_record_count\":0," ++
1082 "\"lost_event_count\":0,\"lost_samples_record_count\":0," ++
1083 "\"lost_samples_count\":0,\"throttle_record_count\":0," ++
1084 "\"unthrottle_record_count\":0," ++
1085 "\"loss_counter_status\":\"unsupported\",\"loss_counter_value\":null," ++
1086 "\"terminal_status\":\"not_started\"}\n" ++
1087 "{\"schema\":\"coz.profile/v1\",\"event\":\"sample\",\"location\":{\"file\":\"/tmp/a.zig\",\"line\":9},\"count\":1}\n" ++
1088 "{\"schema\":\"coz.profile/v1\",\"event\":\"sample\",\"location\":{\"file\":\"/tmp/b.zig\",\"line\":5},\"count\":2}\n",
1089 writer.buffered(),
1090 );
1091 }
1092
1093 test "profiler observes decoded perf sample records" {
1094 var profiler: Profiler = .{};
1095 defer profiler.deinit(std.testing.allocator);
1096 var thread: delay.ThreadState = .{};
1097
1098 _ = try profiler.addSourceRange(std.testing.allocator, "/tmp/ip.zig", 1, try source_map.Interval.init(100, 110));
1099 const selected = try profiler.addSourceRange(std.testing.allocator, "/tmp/selected.zig", 2, try source_map.Interval.init(200, 210));
1100 profiler.selectLine(selected);
1101 profiler.delays.startExperiment(9);
1102
1103 var record_bytes: [128]u8 = undefined;
1104 var offset: usize = perf.header_size;
1105 writePerfU64(&record_bytes, &offset, 105);
1106 writePerfU64(&record_bytes, &offset, 1);
1107 writePerfU64(&record_bytes, &offset, 206);
1108
1109 const config: perf.Config = .{ .sample_type = perf.sampleMask(&.{ .ip, .callchain }) };
1110 const record = try perf.Record.init(config, finishPerfSample(&record_bytes, offset));
1111 var callchain_scratch: [8]usize = undefined;
1112
1113 const matched = try profiler.observePerfRecord(&thread, record, &callchain_scratch);
1114
1115 try std.testing.expectEqual(selected, matched.line.?);
1116 try std.testing.expect(matched.selected_hit);
1117 try std.testing.expectEqual(@as(u64, 1), selected.sampleCount());
1118 try std.testing.expectEqual(@as(u64, 9), thread.localDelay());
1119 try std.testing.expectEqual(@as(u64, 1), profiler.samplingSnapshot().sample_record_count);
1120 }
1121
1122 test "profiler audits perf loss and throttle records" {
1123 var profiler: Profiler = .{};
1124 defer profiler.deinit(std.testing.allocator);
1125 var thread: delay.ThreadState = .{};
1126 var callchain_scratch: [0]usize = .{};
1127
1128 var lost_bytes: [32]u8 = undefined;
1129 var lost_offset: usize = perf.header_size;
1130 writePerfU64(&lost_bytes, &lost_offset, 1);
1131 writePerfU64(&lost_bytes, &lost_offset, 5);
1132 const lost = try perf.Record.init(.{}, finishPerfRecord(&lost_bytes, lost_offset, .lost));
1133
1134 var samples_bytes: [24]u8 = undefined;
1135 var samples_offset: usize = perf.header_size;
1136 writePerfU64(&samples_bytes, &samples_offset, 7);
1137 const lost_samples = try perf.Record.init(
1138 .{},
1139 finishPerfRecord(&samples_bytes, samples_offset, .lost_samples),
1140 );
1141 var throttle_bytes: [perf.header_size]u8 = undefined;
1142 const throttle = try perf.Record.init(
1143 .{},
1144 finishPerfRecord(&throttle_bytes, perf.header_size, .throttle),
1145 );
1146 const unthrottle = try perf.Record.init(
1147 .{},
1148 finishPerfRecord(&throttle_bytes, perf.header_size, .unthrottle),
1149 );
1150
1151 _ = try profiler.observePerfRecord(&thread, lost, &callchain_scratch);
1152 _ = try profiler.observePerfRecord(&thread, lost_samples, &callchain_scratch);
1153 _ = try profiler.observePerfRecord(&thread, throttle, &callchain_scratch);
1154 _ = try profiler.observePerfRecord(&thread, unthrottle, &callchain_scratch);
1155 const snapshot = profiler.samplingSnapshot();
1156
1157 try std.testing.expectEqual(@as(u64, 4), snapshot.record_count);
1158 try std.testing.expectEqual(@as(u64, 1), snapshot.lost_record_count);
1159 try std.testing.expectEqual(@as(u64, 5), snapshot.lost_event_count);
1160 try std.testing.expectEqual(@as(u64, 1), snapshot.lost_samples_record_count);
1161 try std.testing.expectEqual(@as(u64, 7), snapshot.lost_samples_count);
1162 try std.testing.expectEqual(@as(u64, 1), snapshot.throttle_record_count);
1163 try std.testing.expectEqual(@as(u64, 1), snapshot.unthrottle_record_count);
1164 }
1165
1166 test "profiler rejects perf callchains that exceed scratch space" {
1167 var profiler: Profiler = .{};
1168 defer profiler.deinit(std.testing.allocator);
1169 var thread: delay.ThreadState = .{};
1170
1171 var record_bytes: [128]u8 = undefined;
1172 var offset: usize = perf.header_size;
1173 writePerfU64(&record_bytes, &offset, 0);
1174 writePerfU64(&record_bytes, &offset, 2);
1175 writePerfU64(&record_bytes, &offset, 100);
1176 writePerfU64(&record_bytes, &offset, 200);
1177
1178 const config: perf.Config = .{ .sample_type = perf.sampleMask(&.{ .ip, .callchain }) };
1179 const record = try perf.Record.init(config, finishPerfSample(&record_bytes, offset));
1180 var callchain_scratch: [1]usize = undefined;
1181
1182 try std.testing.expectEqual(error.CallchainScratchTooSmall, profiler.observePerfRecord(&thread, record, &callchain_scratch));
1183 }
1184
1185 test "profiler drains perf ring records into source samples" {
1186 var profiler: Profiler = .{};
1187 defer profiler.deinit(std.testing.allocator);
1188 var thread: delay.ThreadState = .{};
1189
1190 const line = try profiler.addSourceRange(std.testing.allocator, "/tmp/hot.zig", 4, try source_map.Interval.init(100, 110));
1191 const config: perf.Config = .{ .sample_type = perf.sampleMask(&.{.ip}) };
1192
1193 var first_record: [32]u8 = undefined;
1194 var first_offset: usize = perf.header_size;
1195 writePerfU64(&first_record, &first_offset, 105);
1196 const first = finishPerfSample(&first_record, first_offset);
1197
1198 var second_record: [32]u8 = undefined;
1199 var second_offset: usize = perf.header_size;
1200 writePerfU64(&second_record, &second_offset, 106);
1201 const second = finishPerfSample(&second_record, second_offset);
1202
1203 var ring_data: [96]u8 = undefined;
1204 @memset(&ring_data, 0);
1205 const tail: u64 = 88;
1206 writeRingBytes(&ring_data, tail, first);
1207 writeRingBytes(&ring_data, tail + first.len, second);
1208
1209 var reader = perf.RingReader.init(config, &ring_data, tail, tail + first.len + second.len);
1210 var record_scratch: [64]u8 = undefined;
1211 var callchain_scratch: [0]usize = .{};
1212
1213 const drained = try profiler.drainPerfRing(&thread, &reader, &record_scratch, &callchain_scratch);
1214
1215 try std.testing.expectEqual(@as(usize, 2), drained);
1216 try std.testing.expectEqual(@as(u64, 2), line.sampleCount());
1217 try std.testing.expectEqual(tail + first.len + second.len, reader.index);
1218 }
1219
1220 test "profiler processes perf ring samples before applying delays" {
1221 var profiler: Profiler = .{};
1222 defer profiler.deinit(std.testing.allocator);
1223 var thread: delay.ThreadState = .{};
1224
1225 const line = try profiler.addSourceRange(std.testing.allocator, "/tmp/selected.zig", 6, try source_map.Interval.init(100, 110));
1226 profiler.selectLine(line);
1227 profiler.delays.startExperiment(5);
1228
1229 const config: perf.Config = .{ .sample_type = perf.sampleMask(&.{.ip}) };
1230 var record: [32]u8 = undefined;
1231 var offset: usize = perf.header_size;
1232 writePerfU64(&record, &offset, 105);
1233 const sample = finishPerfSample(&record, offset);
1234
1235 var ring_data: [64]u8 = undefined;
1236 @memset(&ring_data, 0);
1237 writeRingBytes(&ring_data, 0, sample);
1238
1239 var reader = perf.RingReader.init(config, &ring_data, 0, sample.len);
1240 var record_scratch: [64]u8 = undefined;
1241 var callchain_scratch: [0]usize = .{};
1242
1243 const processed = try profiler.processPerfRing(&thread, &reader, &record_scratch, &callchain_scratch, exactWait);
1244
1245 try std.testing.expectEqual(@as(usize, 1), processed);
1246 try std.testing.expectEqual(@as(u64, 1), line.sampleCount());
1247 try std.testing.expectEqual(@as(u64, 5), thread.localDelay());
1248 try std.testing.expectEqual(@as(u64, 5), profiler.delays.globalDelay());
1249 }
1250
1251 test "profiler drains zero records from unmapped perf event" {
1252 var profiler: Profiler = .{};
1253 defer profiler.deinit(std.testing.allocator);
1254 var thread: delay.ThreadState = .{};
1255 var event: perf.Event = .{};
1256 var record_scratch: [64]u8 = undefined;
1257 var callchain_scratch: [8]usize = undefined;
1258
1259 const drained = try profiler.drainPerfEvent(&thread, &event, &record_scratch, &callchain_scratch);
1260
1261 try std.testing.expectEqual(@as(usize, 0), drained);
1262 }
1263
1264 test "profiler drains zero records from unopened sampler" {
1265 var profiler: Profiler = .{};
1266 defer profiler.deinit(std.testing.allocator);
1267 var thread: delay.ThreadState = .{};
1268 var sample: sampler_mod.Sampler = .{};
1269 var record_scratch: [64]u8 = undefined;
1270 var callchain_scratch: [8]usize = undefined;
1271
1272 const drained = try profiler.drainSampler(&thread, &sample, &record_scratch, &callchain_scratch);
1273
1274 try std.testing.expectEqual(@as(usize, 0), drained);
1275 }
1276
1277 test "profiler processes unopened sampler by applying pending delay" {
1278 var profiler: Profiler = .{};
1279 defer profiler.deinit(std.testing.allocator);
1280 var thread: delay.ThreadState = .{};
1281 var sample: sampler_mod.Sampler = .{};
1282 var record_scratch: [64]u8 = undefined;
1283 var callchain_scratch: [8]usize = undefined;
1284
1285 profiler.delays.startExperiment(0);
1286 profiler.delays.global_delay_ns.store(9, .monotonic);
1287
1288 const processed = try profiler.processSampler(&thread, &sample, &record_scratch, &callchain_scratch, exactWait);
1289
1290 try std.testing.expectEqual(@as(usize, 0), processed);
1291 try std.testing.expectEqual(@as(u64, 9), thread.localDelay());
1292 }
1293
1294 test "profiler exposes delay accounting for blocking hooks" {
1295 var profiler: Profiler = .{};
1296 defer profiler.deinit(std.testing.allocator);
1297 var thread: delay.ThreadState = .{};
1298
1299 profiler.delays.startExperiment(0);
1300
1301 profiler.preBlock(&thread);
1302 profiler.delays.global_delay_ns.store(30, .monotonic);
1303 try std.testing.expectEqual(@as(u64, 0), profiler.catchUp(&thread, exactWait));
1304
1305 profiler.postBlock(&thread, true);
1306
1307 try std.testing.expectEqual(@as(u64, 30), thread.localDelay());
1308 }
1309
1310 test "profiler observes idle samples and selects the next non-header line" {
1311 var profiler: Profiler = .{};
1312 defer profiler.deinit(std.testing.allocator);
1313 var thread: delay.ThreadState = .{};
1314
1315 const line = try profiler.addSourceRange(std.testing.allocator, "/tmp/main.zig", 10, try source_map.Interval.init(100, 110));
1316
1317 const matched = profiler.observeSample(&thread, .{ .ip = 105 });
1318
1319 try std.testing.expectEqual(line, matched.line.?);
1320 try std.testing.expect(!matched.selected_hit);
1321 try std.testing.expectEqual(@as(u64, 1), line.sampleCount());
1322 try std.testing.expectEqual(line, profiler.nextLine().?);
1323 try std.testing.expectEqual(@as(u64, 0), thread.localDelay());
1324 }
1325
1326 test "profiler observes active selected samples and credits delay" {
1327 var profiler: Profiler = .{};
1328 defer profiler.deinit(std.testing.allocator);
1329 var thread: delay.ThreadState = .{};
1330
1331 const line = try profiler.addSourceRange(std.testing.allocator, "/tmp/main.zig", 10, try source_map.Interval.init(100, 110));
1332 profiler.selectLine(line);
1333 profiler.delays.startExperiment(7);
1334
1335 const matched = profiler.observeSample(&thread, .{ .ip = 0, .callchain = &.{106} });
1336
1337 try std.testing.expectEqual(line, matched.line.?);
1338 try std.testing.expect(matched.selected_hit);
1339 try std.testing.expectEqual(@as(u64, 1), line.sampleCount());
1340 try std.testing.expectEqual(@as(u64, 7), thread.localDelay());
1341 try std.testing.expect(profiler.nextLine() == null);
1342 }
1343
1344 test "profiler does not select coz header samples as the next experiment line" {
1345 var profiler: Profiler = .{};
1346 defer profiler.deinit(std.testing.allocator);
1347 var thread: delay.ThreadState = .{};
1348
1349 const header = try profiler.addSourceRange(std.testing.allocator, "/tmp/include/coz.h", 4, try source_map.Interval.init(10, 20));
1350
1351 const matched = profiler.observeSample(&thread, .{ .ip = 12 });
1352
1353 try std.testing.expectEqual(header, matched.line.?);
1354 try std.testing.expectEqual(@as(u64, 1), header.sampleCount());
1355 try std.testing.expect(profiler.nextLine() == null);
1356 }