lib/coz/src/registry.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2
3 const abi = @import("abi.zig");
4 const progress_point = @import("point.zig");
5
6 pub const Registry = struct {
7 mutex: std.atomic.Mutex = .unlocked,
8 throughput_points: std.StringHashMapUnmanaged(*progress_point.ThroughputPoint) = .empty,
9 latency_points: std.StringHashMapUnmanaged(*progress_point.LatencyPoint) = .empty,
10
11 pub fn deinit(self: *Registry, allocator: std.mem.Allocator) void {
12 lockMutex(&self.mutex);
13 defer self.mutex.unlock();
14
15 var throughput_iter = self.throughput_points.iterator();
16 while (throughput_iter.next()) |entry| {
17 allocator.free(entry.key_ptr.*);
18 allocator.destroy(entry.value_ptr.*);
19 }
20 self.throughput_points.deinit(allocator);
21
22 var latency_iter = self.latency_points.iterator();
23 while (latency_iter.next()) |entry| {
24 allocator.free(entry.key_ptr.*);
25 allocator.destroy(entry.value_ptr.*);
26 }
27 self.latency_points.deinit(allocator);
28
29 self.throughput_points = .empty;
30 self.latency_points = .empty;
31 }
32
33 pub fn getThroughputPoint(
34 self: *Registry,
35 allocator: std.mem.Allocator,
36 name: []const u8,
37 ) !*progress_point.ThroughputPoint {
38 lockMutex(&self.mutex);
39 defer self.mutex.unlock();
40
41 if (self.throughput_points.get(name)) |point| return point;
42
43 const owned_name = try allocator.dupe(u8, name);
44 errdefer allocator.free(owned_name);
45
46 const point = try allocator.create(progress_point.ThroughputPoint);
47 errdefer allocator.destroy(point);
48 point.* = progress_point.ThroughputPoint.init(owned_name);
49
50 try self.throughput_points.put(allocator, owned_name, point);
51 return point;
52 }
53
54 pub fn getLatencyPoint(
55 self: *Registry,
56 allocator: std.mem.Allocator,
57 name: []const u8,
58 ) !*progress_point.LatencyPoint {
59 lockMutex(&self.mutex);
60 defer self.mutex.unlock();
61
62 if (self.latency_points.get(name)) |point| return point;
63
64 const owned_name = try allocator.dupe(u8, name);
65 errdefer allocator.free(owned_name);
66
67 const point = try allocator.create(progress_point.LatencyPoint);
68 errdefer allocator.destroy(point);
69 point.* = progress_point.LatencyPoint.init(owned_name);
70
71 try self.latency_points.put(allocator, owned_name, point);
72 return point;
73 }
74
75 pub fn getCounter(
76 self: *Registry,
77 allocator: std.mem.Allocator,
78 kind: abi.CounterKind,
79 name: []const u8,
80 ) !*abi.Counter {
81 return switch (kind) {
82 .throughput => (try self.getThroughputPoint(allocator, name)).counterStruct(),
83 .begin => (try self.getLatencyPoint(allocator, name)).beginCounterStruct(),
84 .end => (try self.getLatencyPoint(allocator, name)).endCounterStruct(),
85 };
86 }
87
88 pub fn saveThroughputSnapshots(
89 self: *Registry,
90 allocator: std.mem.Allocator,
91 ) ![]progress_point.ThroughputSnapshot {
92 lockMutex(&self.mutex);
93 defer self.mutex.unlock();
94
95 var snapshots: std.ArrayListUnmanaged(progress_point.ThroughputSnapshot) = .empty;
96 errdefer snapshots.deinit(allocator);
97
98 var iter = self.throughput_points.valueIterator();
99 while (iter.next()) |point| {
100 try snapshots.append(allocator, point.*.save());
101 }
102
103 return snapshots.toOwnedSlice(allocator);
104 }
105
106 pub fn saveLatencySnapshots(
107 self: *Registry,
108 allocator: std.mem.Allocator,
109 ) ![]progress_point.LatencySnapshot {
110 lockMutex(&self.mutex);
111 defer self.mutex.unlock();
112
113 var snapshots: std.ArrayListUnmanaged(progress_point.LatencySnapshot) = .empty;
114 errdefer snapshots.deinit(allocator);
115
116 var iter = self.latency_points.valueIterator();
117 while (iter.next()) |point| {
118 try snapshots.append(allocator, point.*.save());
119 }
120
121 return snapshots.toOwnedSlice(allocator);
122 }
123 };
124
125 fn lockMutex(mutex: *std.atomic.Mutex) void {
126 while (!mutex.tryLock()) std.atomic.spinLoopHint();
127 }
128
129 test "registry returns a stable throughput counter by name" {
130 var registry: Registry = .{};
131 defer registry.deinit(std.testing.allocator);
132
133 const first = try registry.getThroughputPoint(std.testing.allocator, "items");
134 const second = try registry.getThroughputPoint(std.testing.allocator, "items");
135
136 try std.testing.expectEqual(first, second);
137
138 const counter = try registry.getCounter(std.testing.allocator, .throughput, "items");
139 _ = @atomicRmw(usize, &counter.count, .Add, 3, .monotonic);
140
141 try std.testing.expectEqual(@as(usize, 3), first.getCount());
142 }
143
144 test "registry keeps throughput and latency points separate" {
145 var registry: Registry = .{};
146 defer registry.deinit(std.testing.allocator);
147
148 const throughput = try registry.getThroughputPoint(std.testing.allocator, "operation");
149 const latency = try registry.getLatencyPoint(std.testing.allocator, "operation");
150
151 throughput.visit(4);
152 latency.visitBegin(5);
153 latency.visitEnd(2);
154
155 try std.testing.expectEqual(@as(usize, 4), throughput.getCount());
156 try std.testing.expectEqual(@as(usize, 5), latency.getBeginCount());
157 try std.testing.expectEqual(@as(usize, 2), latency.getEndCount());
158 }
159
160 test "registry returns stable latency begin and end counters by name" {
161 var registry: Registry = .{};
162 defer registry.deinit(std.testing.allocator);
163
164 const first_begin = try registry.getCounter(std.testing.allocator, .begin, "request");
165 const second_begin = try registry.getCounter(std.testing.allocator, .begin, "request");
166 const end = try registry.getCounter(std.testing.allocator, .end, "request");
167
168 try std.testing.expectEqual(first_begin, second_begin);
169 try std.testing.expect(first_begin != end);
170
171 _ = @atomicRmw(usize, &first_begin.count, .Add, 7, .monotonic);
172 _ = @atomicRmw(usize, &end.count, .Add, 3, .monotonic);
173
174 const point = try registry.getLatencyPoint(std.testing.allocator, "request");
175 try std.testing.expectEqual(@as(usize, 7), point.getBeginCount());
176 try std.testing.expectEqual(@as(usize, 3), point.getEndCount());
177 }
178
179 test "registry snapshots record deltas from all owned points" {
180 var registry: Registry = .{};
181 defer registry.deinit(std.testing.allocator);
182
183 const throughput = try registry.getThroughputPoint(std.testing.allocator, "items");
184 const latency = try registry.getLatencyPoint(std.testing.allocator, "request");
185
186 throughput.visit(2);
187 latency.visitBegin(1);
188
189 const throughput_snapshots = try registry.saveThroughputSnapshots(std.testing.allocator);
190 defer std.testing.allocator.free(throughput_snapshots);
191
192 const latency_snapshots = try registry.saveLatencySnapshots(std.testing.allocator);
193 defer std.testing.allocator.free(latency_snapshots);
194
195 throughput.visit(5);
196 latency.visitBegin(4);
197 latency.visitEnd(3);
198
199 try std.testing.expectEqual(@as(usize, 1), throughput_snapshots.len);
200 try std.testing.expectEqual(@as(usize, 1), latency_snapshots.len);
201 try std.testing.expectEqualStrings("items", throughput_snapshots[0].getName());
202 try std.testing.expectEqual(@as(usize, 5), throughput_snapshots[0].getDelta());
203 try std.testing.expectEqualStrings("request", latency_snapshots[0].getName());
204 try std.testing.expectEqual(@as(usize, 4), latency_snapshots[0].getBeginDelta());
205 try std.testing.expectEqual(@as(usize, 3), latency_snapshots[0].getEndDelta());
206 }