atomic rate meter

written against zig 0.16.

a pattern for "what's the current rate of X" when X is happening continuously on a worker thread, and many other threads (HTTP handlers, dashboards) need to read the rate cheaply and concurrently. used in relay-eval to expose live per-relay event rates for sonification.

the shape

three threads cooperate, none of them ever blocks the others:

worker thread (1 per resource) sampler thread (1 shared) reader threads (any number) ───────────────────────── ────────────────────── ──────────────────────── counter.fetchAdd(1, .monotonic) loop { m.ratePerSec() on every event for (meters) |*m| → reads ring under m.snapshot() small mutex sleep(1s) }

• writers only do atomic increments — no locks on the hot path
• a single shared sampler thread snapshots all counters at 1Hz into a per-meter ring buffer
• readers compute rate from the ring under a tiny mutex (microseconds)

why not just events_in_window / window?

the obvious thing — keep a counter and a "last sampled value" — gives you only one window, no history, and races with concurrent rate calls. the ring lets multiple readers compute rate over different sub-windows, and gives you an exact lower-bound (oldest sample) without re-arming a counter.

minimal implementation

const std = @import("std");

pub const window_seconds: u32 = 10;
pub const sample_buckets: usize = window_seconds + 1; // N+1 samples for N intervals

const Snapshot = struct {
    counter: u64,
    ts_ms: i64,
};

pub const RateMeter = struct {
    // hot path: writer thread bumps; readers (sampler, /api) read atomically
    counter: std.atomic.Value(u64),

    // sampler ring (1Hz). `idx` points at the newest entry. only the sampler
    // thread mutates the ring; readers take the lock briefly for consistency.
    snapshots: [sample_buckets]Snapshot,
    idx: usize,
    lock: std.Thread.Mutex,

    pub fn init() RateMeter {
        return .{
            .counter = .init(0),
            .snapshots = [_]Snapshot{.{ .counter = 0, .ts_ms = 0 }} ** sample_buckets,
            .idx = 0,
            .lock = .{},
        };
    }

    /// called by the writer thread on each event
    pub inline fn record(self: *RateMeter) void {
        _ = self.counter.fetchAdd(1, .monotonic);
    }

    /// called by the shared sampler thread, ~1Hz
    pub fn snapshot(self: *RateMeter) void {
        const c = self.counter.load(.monotonic);
        const t = std.time.milliTimestamp();
        self.lock.lock();
        defer self.lock.unlock();
        self.idx = (self.idx + 1) % sample_buckets;
        self.snapshots[self.idx] = .{ .counter = c, .ts_ms = t };
    }

    /// events/sec averaged over the ring window (~10s here).
    /// returns 0.0 if the ring hasn't filled enough to span a real window.
    pub fn ratePerSec(self: *RateMeter) f64 {
        self.lock.lock();
        defer self.lock.unlock();
        const newest = self.snapshots[self.idx];
        const oldest_idx = (self.idx + 1) % sample_buckets;
        const oldest = self.snapshots[oldest_idx];
        if (oldest.ts_ms == 0 or newest.ts_ms <= oldest.ts_ms) return 0.0;
        const dt_ms = newest.ts_ms - oldest.ts_ms;
        const dc = newest.counter -| oldest.counter; // saturating sub
        return @as(f64, @floatFromInt(dc)) * 1000.0 /
               @as(f64, @floatFromInt(dt_ms));
    }
};

pub fn sampleLoop(meters: []RateMeter, shutdown: *std.atomic.Value(bool)) void {
    while (!shutdown.load(.acquire)) {
        for (meters) |*m| m.snapshot();
        std.Thread.sleep(std.time.ns_per_s);
    }
}

subtle bits

memory ordering. writes use .monotonic because we don't care about ordering with other memory operations — each fetchAdd is independent, and we tolerate readers seeing slightly-stale counter values (the next sample will catch up). don't pay for stronger ordering you don't need.

ring size = window+1. to compute rate over N seconds you need N+1 samples (so you have N intervals). off-by-one easy to miss.

-| saturating subtract. the counter is monotonic so newest >= oldest always — but during a wraparound or if the writer thread is briefly preempted between counter.load and the subtraction, defensive saturation prevents a panic.

oldest.ts_ms == 0 check. the ring is initialized to zeros. for the first ~N seconds, oldest_idx is still pointing at an uninitialized slot. checking ts_ms == 0 distinguishes "no data yet" from "rate is genuinely zero".

reset backoff on success. when the writer thread's underlying connection is itself flaky and uses exponential backoff for reconnect, reset the backoff inside the successful connect path (after handshake, not just on init). otherwise a long-running connection that briefly drops will inherit the maxed-out backoff from earlier failures, and waste minutes reconnecting. the bug is invisible until you've been running for hours.

// in the persistent connect-and-stream loop:
fn connectAndStream(self: *Worker) !void {
    var client = try connect();
    defer client.close();
    self.backoff_ms = 1000;  // ← reset HERE, not in init
    while (...) { ... record ... }
}

why one shared sampler instead of per-meter

with N meters, the alternatives are:

• N timer threads (one per meter) — wastes thread stacks, more scheduler load
• a tick from the writer thread — couples sampling cadence to event rate (silent meter never samples)
• the sampler-on-read pattern — pays cost in the HTTP path, racy when multiple readers arrive simultaneously

one shared sampler thread snapshots every meter at exactly the same ts_ms, which makes cross-meter comparisons (e.g. "which relay is fastest right now") naturally synchronized. cost: 1 thread, ~50ns × N meters of work per second.

not for everything

this pattern is for cheap counts of high-frequency events with multiple concurrent readers. for measurements that are themselves expensive to make, or where readers are rare, just compute on read. for distributions (p50/p99 latency), use a histogram structure instead — a single counter doesn't capture that shape.