lib/memtrace/src/causal.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const pretty_json = @import("pretty").json;
3 const sys = @import("sys");
4 const coverage_mod = @import("coverage.zig");
5 const event_mod = @import("event.zig");
6 const stack = @import("stack/root.zig");
7
8 const Allocator = std.mem.Allocator;
9
10 pub const Format = enum {
11 text,
12 jsonl,
13 };
14
15 pub const Options = struct {
16 format: Format = .text,
17 binary_path: ?[]const u8 = null,
18 frame_limit: usize = stack.capture.max_frames_limit,
19 };
20
21 const Relation = enum {
22 ancestor,
23 target,
24 descendant,
25 };
26
27 const Record = struct {
28 event: event_mod.ReplayEvent,
29 first_child: u32 = 0,
30 next_sibling: u32 = 0,
31 };
32
33 const Related = struct {
34 record_index: u32,
35 depth: u32,
36 relation: Relation,
37 };
38
39 const Work = struct {
40 record_index: u32,
41 depth: u32,
42 };
43
44 const Graph = struct {
45 allocator: Allocator,
46 stack: stack.analyze.Analyzer,
47 records: std.ArrayListUnmanaged(Record) = .empty,
48 operation_indices: std.AutoHashMapUnmanaged(u64, u32) = .{},
49 linked: bool = false,
50
51 fn init(allocator: Allocator) Graph {
52 return .{
53 .allocator = allocator,
54 .stack = stack.analyze.Analyzer.init(allocator, .{}),
55 };
56 }
57
58 fn deinit(self: *Graph) void {
59 self.stack.deinit();
60 self.records.deinit(self.allocator);
61 self.operation_indices.deinit(self.allocator);
62 self.* = undefined;
63 }
64
65 fn ingestJsonLine(self: *Graph, line: []const u8) !void {
66 if (self.linked) return error.CausalGraphAlreadyLinked;
67 const text = std.mem.trim(u8, line, " \t\r\n");
68 if (text.len == 0) return;
69 try self.stack.ingestJsonLine(text);
70 if (stack.identity.isMetadataLine(text) or
71 coverage_mod.isMetadataLine(text) or
72 stack.capture.isMetadataLine(text))
73 {
74 return;
75 }
76 const event = try event_mod.parseReplayFast(text);
77 if (!event.kind.isMemoryOperation() or event.operation_id == 0) return;
78 const record_index = std.math.cast(
79 u32,
80 self.records.items.len,
81 ) orelse return error.CausalGraphTooLarge;
82 try self.records.append(self.allocator, .{ .event = event });
83 errdefer _ = self.records.pop();
84 try self.operation_indices.putNoClobber(
85 self.allocator,
86 event.operation_id,
87 record_index,
88 );
89 }
90
91 fn link(self: *Graph) !void {
92 if (self.linked) return;
93 for (0..self.records.items.len) |record_index| {
94 const parent_operation_id =
95 self.records.items[record_index].event.parent_operation_id;
96 if (parent_operation_id == 0) continue;
97 const parent_index = self.operation_indices.get(
98 parent_operation_id,
99 ) orelse return error.DanglingParentOperation;
100 self.records.items[record_index].next_sibling =
101 self.records.items[parent_index].first_child;
102 self.records.items[parent_index].first_child =
103 @intCast(record_index + 1);
104 }
105 self.linked = true;
106 }
107
108 fn related(
109 self: *Graph,
110 operation_id: u64,
111 ) !std.ArrayListUnmanaged(Related) {
112 try self.link();
113 const target_index = self.operation_indices.get(operation_id) orelse
114 return error.OperationNotFound;
115 var ancestors = std.ArrayListUnmanaged(u32).empty;
116 defer ancestors.deinit(self.allocator);
117 var current_index = target_index;
118 var traversed: usize = 0;
119 while (self.records.items[current_index].event.parent_operation_id != 0) {
120 traversed += 1;
121 if (traversed > self.records.items.len) {
122 return error.CausalOperationCycle;
123 }
124 const parent_index = self.operation_indices.get(
125 self.records.items[current_index].event.parent_operation_id,
126 ) orelse return error.DanglingParentOperation;
127 try ancestors.append(self.allocator, parent_index);
128 current_index = parent_index;
129 }
130
131 var result = std.ArrayListUnmanaged(Related).empty;
132 errdefer result.deinit(self.allocator);
133 var ancestor_offset = ancestors.items.len;
134 while (ancestor_offset > 0) {
135 ancestor_offset -= 1;
136 try result.append(self.allocator, .{
137 .record_index = ancestors.items[ancestor_offset],
138 .depth = @intCast(ancestors.items.len - ancestor_offset - 1),
139 .relation = .ancestor,
140 });
141 }
142
143 var pending = std.ArrayListUnmanaged(Work).empty;
144 defer pending.deinit(self.allocator);
145 try pending.append(self.allocator, .{
146 .record_index = target_index,
147 .depth = @intCast(ancestors.items.len),
148 });
149 while (pending.pop()) |work| {
150 try result.append(self.allocator, .{
151 .record_index = work.record_index,
152 .depth = work.depth,
153 .relation = if (work.record_index == target_index)
154 .target
155 else
156 .descendant,
157 });
158 var child = self.records.items[work.record_index].first_child;
159 while (child != 0) {
160 const child_index = child - 1;
161 try pending.append(self.allocator, .{
162 .record_index = child_index,
163 .depth = work.depth + 1,
164 });
165 child = self.records.items[child_index].next_sibling;
166 }
167 }
168 std.mem.sort(Related, result.items, self, relatedLessThan);
169 return result;
170 }
171 };
172
173 pub fn writeFromPath(
174 allocator: Allocator,
175 events_path: []const u8,
176 operation_id: u64,
177 writer: *std.Io.Writer,
178 options: Options,
179 ) !void {
180 if (operation_id == 0 or
181 options.frame_limit == 0 or
182 options.frame_limit > stack.capture.max_frames_limit)
183 {
184 return error.InvalidCausalQuery;
185 }
186 var graph = Graph.init(allocator);
187 defer graph.deinit();
188 try ingestPath(&graph, events_path);
189 try graph.stack.validate();
190 var related = try graph.related(operation_id);
191 defer related.deinit(allocator);
192
193 var inferred_binary: ?[]u8 = null;
194 defer if (inferred_binary) |path| allocator.free(path);
195 const binary_path = options.binary_path orelse inferred: {
196 inferred_binary = try stack.identity.artifactPathAlloc(
197 allocator,
198 events_path,
199 );
200 break :inferred inferred_binary.?;
201 };
202 const actual_digest = stack.identity.fileDigest(
203 allocator,
204 binary_path,
205 ) catch |err| switch (err) {
206 error.FileNotFound => return error.MissingExecutableArtifact,
207 else => return err,
208 };
209 const expected_digest = graph.stack.executable_digest.?;
210 if (!std.mem.eql(u8, &actual_digest, &expected_digest)) {
211 return error.ExecutableIdentityMismatch;
212 }
213
214 var addresses = try collectAddresses(
215 allocator,
216 &graph,
217 related.items,
218 options.frame_limit,
219 );
220 defer addresses.deinit(allocator);
221 var symbols = try stack.symbolize.resolveAlloc(
222 allocator,
223 binary_path,
224 addresses.items,
225 );
226 defer symbols.deinit(allocator);
227 switch (options.format) {
228 .text => try writeText(
229 writer,
230 &graph,
231 related.items,
232 operation_id,
233 options.frame_limit,
234 &symbols,
235 expected_digest,
236 ),
237 .jsonl => try writeJsonl(
238 writer,
239 &graph,
240 related.items,
241 operation_id,
242 options.frame_limit,
243 &symbols,
244 expected_digest,
245 ),
246 }
247 }
248
249 fn ingestPath(graph: *Graph, path: []const u8) !void {
250 var file = try sys.fs.cwd().openFile(sys.fs.debugIo(), path, .{});
251 defer file.close(sys.fs.debugIo());
252 var buffer: [64 * 1024]u8 = undefined;
253 var reader = file.reader(sys.fs.debugIo(), &buffer);
254 while (true) {
255 const line = reader.interface.takeDelimiter('\n') catch |err| switch (err) {
256 error.ReadFailed => return reader.err.?,
257 else => return err,
258 };
259 const actual = line orelse break;
260 try graph.ingestJsonLine(actual);
261 }
262 }
263
264 fn collectAddresses(
265 allocator: Allocator,
266 graph: *const Graph,
267 related: []const Related,
268 frame_limit: usize,
269 ) !std.ArrayListUnmanaged(u64) {
270 var seen = std.AutoHashMapUnmanaged(u64, void){};
271 defer seen.deinit(allocator);
272 var addresses = std.ArrayListUnmanaged(u64).empty;
273 errdefer addresses.deinit(allocator);
274 for (related) |item| {
275 const event = graph.records.items[item.record_index].event;
276 const definition = graph.stack.stackDefinition(event.stack_id).?;
277 const limit = @min(frame_limit, definition.call_addresses.len);
278 for (definition.call_addresses[0..limit]) |address| {
279 const entry = try seen.getOrPut(allocator, address);
280 if (entry.found_existing) continue;
281 try addresses.append(allocator, address);
282 }
283 }
284 std.mem.sort(u64, addresses.items, {}, u64LessThan);
285 return addresses;
286 }
287
288 fn writeText(
289 writer: *std.Io.Writer,
290 graph: *const Graph,
291 related: []const Related,
292 operation_id: u64,
293 frame_limit: usize,
294 symbols: *const stack.symbolize.Symbols,
295 digest: stack.identity.Digest,
296 ) !void {
297 const digest_hex = std.fmt.bytesToHex(digest, .lower);
298 const root_event = graph.records.items[related[0].record_index].event;
299 const coverage = graph.stack.coverage.?;
300 try writer.print(
301 "causal_operations status={s} universe={s} query_operation_id={d} " ++
302 "root_operation_id={d} operations={d} process_complete={} " ++
303 "binary_sha256={s}\n",
304 .{
305 coverage.statusTag(),
306 coverage.universe.tag(),
307 operation_id,
308 root_event.operation_id,
309 related.len,
310 coverage.processComplete(),
311 digest_hex,
312 },
313 );
314 for (related) |item| {
315 const event = graph.records.items[item.record_index].event;
316 try writer.print(
317 "operation relation={s} depth={d} seq={d} operation_id={d} " ++
318 "parent_operation_id={d} layer={s} producer={s} kind={s} " ++
319 "producer_id={d} allocator_id={d} allocation_id={d} " ++
320 "scope_id={d} alignment={d} succeeded={} requested_bytes={d} " ++
321 "address=0x{x} old_address=0x{x}\n",
322 .{
323 @tagName(item.relation),
324 item.depth,
325 event.seq.?,
326 event.operation_id,
327 event.parent_operation_id,
328 event.layer.tag(),
329 @tagName(event.producer),
330 event.kind.tag(),
331 event.producer_id,
332 event.allocator_id,
333 event.allocation_id,
334 event.scope_id,
335 event.alignment,
336 event.succeeded,
337 requestBytes(event),
338 event.address,
339 event.old_address,
340 },
341 );
342 try writeTextFrames(
343 writer,
344 graph.stack.stackDefinition(event.stack_id).?,
345 frame_limit,
346 symbols,
347 );
348 }
349 }
350
351 fn writeTextFrames(
352 writer: *std.Io.Writer,
353 definition: stack.analyze.Definition,
354 frame_limit: usize,
355 symbols: *const stack.symbolize.Symbols,
356 ) !void {
357 const limit = @min(frame_limit, definition.call_addresses.len);
358 for (definition.call_addresses[0..limit], 0..) |address, frame_index| {
359 const resolved = symbols.find(address);
360 try writer.print(
361 " frame={d} call_address=0x{x}",
362 .{ frame_index, address },
363 );
364 if (resolved.len != 0) {
365 try writer.writeAll(" function=");
366 try pretty_json.writeString(writer, resolved[0].function);
367 try writer.writeAll(" location=");
368 try pretty_json.writeString(writer, resolved[0].location);
369 }
370 try writer.writeByte('\n');
371 for (resolved[1..], 1..) |inline_frame, inline_index| {
372 try writer.print(" inline={d} function=", .{inline_index});
373 try pretty_json.writeString(writer, inline_frame.function);
374 try writer.writeAll(" location=");
375 try pretty_json.writeString(writer, inline_frame.location);
376 try writer.writeByte('\n');
377 }
378 }
379 }
380
381 fn writeJsonl(
382 writer: *std.Io.Writer,
383 graph: *const Graph,
384 related: []const Related,
385 operation_id: u64,
386 frame_limit: usize,
387 symbols: *const stack.symbolize.Symbols,
388 digest: stack.identity.Digest,
389 ) !void {
390 const root_event = graph.records.items[related[0].record_index].event;
391 const coverage = graph.stack.coverage.?;
392 var summary_stream = pretty_json.Writer.init(writer, .minified);
393 const summary = try summary_stream.object();
394 try summary.field("kind", "causal_operation_summary");
395 try summary.field("status", coverage.statusTag());
396 try summary.field("universe", coverage.universe.tag());
397 try summary.field("query_operation_id", operation_id);
398 try summary.field("root_operation_id", root_event.operation_id);
399 try summary.field("operations", related.len);
400 try summary.field("process_complete", coverage.processComplete());
401 try summary.hexString("binary_sha256", &digest);
402 try summary.endLine();
403 for (related) |item| {
404 const event = graph.records.items[item.record_index].event;
405 var event_stream = pretty_json.Writer.init(writer, .minified);
406 const object = try event_stream.object();
407 try object.field("kind", "causal_operation");
408 try object.field("relation", @tagName(item.relation));
409 try object.field("depth", item.depth);
410 try object.field("seq", event.seq.?);
411 try object.field("operation_id", event.operation_id);
412 try object.field("parent_operation_id", event.parent_operation_id);
413 try object.field("layer", event.layer.tag());
414 try object.field("producer", @tagName(event.producer));
415 try object.field("operation", event.kind.tag());
416 try object.field("producer_id", event.producer_id);
417 try object.field("allocator_id", event.allocator_id);
418 try object.field("allocation_id", event.allocation_id);
419 try object.field("scope_id", event.scope_id);
420 try object.field("alignment", event.alignment);
421 try object.field("succeeded", event.succeeded);
422 try object.field("requested_bytes", requestBytes(event));
423 try object.field("address", event.address);
424 try object.field("old_address", event.old_address);
425 try object.endLine();
426 const definition = graph.stack.stackDefinition(event.stack_id).?;
427 const limit = @min(frame_limit, definition.call_addresses.len);
428 for (definition.call_addresses[0..limit], 0..) |address, frame_index| {
429 const resolved = symbols.find(address);
430 if (resolved.len == 0) {
431 try writeJsonFrame(
432 writer,
433 event.operation_id,
434 frame_index,
435 address,
436 0,
437 "",
438 "",
439 );
440 continue;
441 }
442 for (resolved, 0..) |inline_frame, inline_index| {
443 try writeJsonFrame(
444 writer,
445 event.operation_id,
446 frame_index,
447 address,
448 inline_index,
449 inline_frame.function,
450 inline_frame.location,
451 );
452 }
453 }
454 }
455 }
456
457 fn writeJsonFrame(
458 writer: *std.Io.Writer,
459 operation_id: u64,
460 frame_index: usize,
461 address: u64,
462 inline_index: usize,
463 function: []const u8,
464 location: []const u8,
465 ) !void {
466 var stream = pretty_json.Writer.init(writer, .minified);
467 const object = try stream.object();
468 try object.field("kind", "causal_operation_frame");
469 try object.field("operation_id", operation_id);
470 try object.field("frame", frame_index);
471 try object.field("call_address", address);
472 try object.field("inline", inline_index);
473 try object.field("function", function);
474 try object.field("location", location);
475 try object.endLine();
476 }
477
478 fn requestBytes(event: event_mod.ReplayEvent) usize {
479 return switch (event.kind) {
480 .free, .release, .unmap => event.old_len,
481 .alloc, .resize, .remap, .map, .protect, .discard, .decommit, .advise => event.len,
482 else => unreachable,
483 };
484 }
485
486 fn relatedLessThan(graph: *Graph, left: Related, right: Related) bool {
487 const left_id =
488 graph.records.items[left.record_index].event.operation_id;
489 const right_id =
490 graph.records.items[right.record_index].event.operation_id;
491 return left_id < right_id;
492 }
493
494 fn u64LessThan(_: void, left: u64, right: u64) bool {
495 return left < right;
496 }
497
498 test "causal graph returns ancestors target and descendants" {
499 var graph = Graph.init(std.testing.allocator);
500 defer graph.deinit();
501 try graph.ingestJsonLine(
502 "{\"v\":3,\"seq\":1,\"kind\":\"alloc\",\"operation_id\":12," ++
503 "\"parent_operation_id\":11,\"stack_id\":1," ++
504 "\"layer\":\"logical_allocator\",\"producer\":\"debug\"}",
505 );
506 try graph.ingestJsonLine(
507 "{\"v\":3,\"seq\":2,\"kind\":\"alloc\",\"operation_id\":11," ++
508 "\"parent_operation_id\":10,\"stack_id\":1}",
509 );
510 try graph.ingestJsonLine(
511 "{\"v\":3,\"seq\":3,\"kind\":\"alloc\",\"operation_id\":10," ++
512 "\"stack_id\":1,\"layer\":\"logical_allocator\"," ++
513 "\"producer\":\"arena\"}",
514 );
515 try graph.ingestJsonLine(
516 "{\"v\":3,\"seq\":4,\"kind\":\"alloc\",\"operation_id\":20," ++
517 "\"stack_id\":1,\"layer\":\"logical_allocator\"," ++
518 "\"producer\":\"bump\"}",
519 );
520 var related = try graph.related(11);
521 defer related.deinit(std.testing.allocator);
522 try std.testing.expectEqual(@as(usize, 3), related.items.len);
523 try std.testing.expectEqual(Relation.ancestor, related.items[0].relation);
524 try std.testing.expectEqual(Relation.target, related.items[1].relation);
525 try std.testing.expectEqual(Relation.descendant, related.items[2].relation);
526 try std.testing.expectEqual(
527 @as(u64, 10),
528 graph.records.items[related.items[0].record_index].event.operation_id,
529 );
530 try std.testing.expectEqual(
531 @as(u64, 12),
532 graph.records.items[related.items[2].record_index].event.operation_id,
533 );
534 }