Something went wrong. Try again.
Play and record audio on Windows from Zig through WASAPI and COM, without linking a C library.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! What the graph cycles have cost, and how the render thread reports it without//! ever waiting for a reader.//!//! ```zig//! const s = stream.stats();//! std.debug.print("busy {d}% of the period, {d} underrun frames\n", .{//! s.process.fractionOfPeriod(s.period_ns) * 100,//! s.underrun_frames,//! });//! ```//!//! ## Min, mean and max rather than a histogram//!//! The number that decides whether a stream is safe is the *worst* cycle, not the//! typical one. Audio has a deadline: a cycle that takes longer than the period//! is a dropout, and it does not matter that the other nine hundred were quick.//! So each metric is kept as its minimum, its mean and its maximum, which is what//! `pw-top` shows and what a person actually looks at.//!//! ## The reader never blocks the render thread//!//! `Stats` is produced by a real-time thread and consumed by whatever thread asks.//! A mutex would be wrong in the only direction that matters -- a reader//! descheduled while holding it would stall a graph cycle -- so the handoff is a//! sequence lock: the writer publishes between two increments of a counter, and a//! reader that sees the counter change under it simply reads again.//!//! That makes the writer wait-free: it never spins, never blocks, and costs two//! atomic stores per cycle. It makes the reader retry occasionally, which is free//! because no reader is in the audio path. This is the opposite trade to//! zig-pipewire's `volume()`, which takes the lock the data thread holds for a//! whole cycle; doing it this way means `stats()` is safe to call from anywhere,//! including from inside a `Process` callback.
const std = @import("std");
/// One metric's spread over the cycles since the last reset.pub const Summary = struct { min_ns: u64 = 0, mean_ns: u64 = 0, max_ns: u64 = 0,
/// This metric as a fraction of the cycle period. /// /// The form `pw-top` reports and the one that says whether there is headroom: /// a `process` maximum at 0.9 of the period is a stream about to glitch, /// whatever its mean says. Zero when the period is not yet known. pub fn fractionOfPeriod(self: Summary, period_ns: u64) f64 { if (period_ns == 0) return 0; return @as(f64, @floatFromInt(self.max_ns)) / @as(f64, @floatFromInt(period_ns)); }
pub fn format(self: Summary, w: *std.Io.Writer) std.Io.Writer.Error!void { try w.print("{d}/{d}/{d} us", .{ self.min_ns / std.time.ns_per_us, self.mean_ns / std.time.ns_per_us, self.max_ns / std.time.ns_per_us, }); }};
/// What the cycles since the last `resetStats` have cost.pub const Stats = struct { /// Cycles processed since the last reset. cycles: u64 = 0,
/// How far the render thread's wakeup drifted from an ideal cycle boundary. /// /// **Not** the same quantity as PipeWire's WAIT, and the difference is worth /// knowing. PipeWire gets a timestamp from the driver saying when the node was /// marked ready, and subtracts. WASAPI provides no such timestamp -- there is /// only the event handle becoming signalled -- so this is measured against a /// line that advances by one period per cycle and is rebased whenever the /// stream starts. It is a good measure of jitter and a poor measure of /// absolute scheduling latency, and it should be read as the former. wake: Summary = .{},
/// Time spent inside the cycle producing audio: `pw-top` calls this BUSY. /// /// A true analogue of PipeWire's figure, from the same two timestamps around /// the same work, so the two can be compared directly. process: Summary = .{},
/// How long a cycle lasts at the current rate and period. Both figures above /// have to fit inside this, together with every other client's. period_ns: u64 = 0,
/// Cycles the render thread believes it missed. /// /// **Inferred**, not reported: WASAPI has no glitch counter, so this counts /// the times the wait timed out and the times the wake was more than a full /// period late. A non-zero value means the thread was not scheduled in time; /// the exact number is a symptom rather than a measurement. missed_cycles: u64 = 0,
/// Frames of silence the render thread had to invent because the ring was /// short. /// /// Exact, and the number that usually matters: it is the audible consequence /// of everything above. Counted from the first write. underrun_frames: u64 = 0,
/// Times the stream moved itself to another endpoint and carried on. /// /// Has no PipeWire counterpart, because PipeWire moves a node without the /// client knowing. A count that climbs while nobody is unplugging anything is /// worth investigating. See `Stream.Options.follow_default`. reopens: u64 = 0,
pub fn format(self: Stats, w: *std.Io.Writer) std.Io.Writer.Error!void { try w.print("{d} cycles, period {d} us, wake {f}, busy {f}", .{ self.cycles, self.period_ns / std.time.ns_per_us, self.wake, self.process, }); if (self.missed_cycles != 0) try w.print(", {d} missed", .{self.missed_cycles}); if (self.underrun_frames != 0) { try w.print(", {d} underrun frames", .{self.underrun_frames}); } if (self.reopens != 0) try w.print(", {d} reopens", .{self.reopens}); }};
/// The render thread's private running totals.////// Not shared: the render thread owns this outright and folds each cycle into it,/// then publishes a snapshot through a `Cell`. Keeping the accumulation private is/// what lets it be plain arithmetic with no atomics in the hot path.pub const Accumulator = struct { cycles: u64 = 0, wake: Running = .{}, process: Running = .{}, period_ns: u64 = 0, missed_cycles: u64 = 0, underrun_frames: u64 = 0, reopens: u64 = 0,
/// Sum and extremes for one metric. The mean is computed on the way out /// rather than maintained, so accumulating is two comparisons and an add. pub const Running = struct { min_ns: u64 = std.math.maxInt(u64), max_ns: u64 = 0, total_ns: u64 = 0, count: u64 = 0,
pub fn add(self: *Running, ns: u64) void { self.min_ns = @min(self.min_ns, ns); self.max_ns = @max(self.max_ns, ns); // Saturating, because a stream left running for weeks should report a // slightly wrong mean rather than wrap to nonsense -- or, in a safe // build, trap on a real-time thread. self.total_ns +|= ns; self.count +|= 1; }
pub fn summary(self: Running) Summary { if (self.count == 0) return .{}; return .{ .min_ns = self.min_ns, .mean_ns = self.total_ns / self.count, .max_ns = self.max_ns, }; } };
/// Fold one completed cycle in. pub fn record(self: *Accumulator, wake_ns: u64, process_ns: u64) void { self.cycles +|= 1; self.wake.add(wake_ns); self.process.add(process_ns); }
/// What a reader should see. pub fn snapshot(self: *const Accumulator) Stats { return .{ .cycles = self.cycles, .wake = self.wake.summary(), .process = self.process.summary(), .period_ns = self.period_ns, .missed_cycles = self.missed_cycles, .underrun_frames = self.underrun_frames, .reopens = self.reopens, }; }
/// Start the measurement window again, keeping the counters that describe the /// stream rather than the window. /// /// `period_ns` and `reopens` survive: the first is a property of the stream as /// it is now, and the second is a count of things that happened to it that a /// caller resetting a benchmark window did not mean to forget. Everything else /// goes, which is the point -- a benchmark wants to exclude the cycles spent /// starting up. pub fn reset(self: *Accumulator) void { self.* = .{ .period_ns = self.period_ns, .reopens = self.reopens }; }};
/// A `Stats` that one thread publishes and any number read, without a lock.////// ## How it works, and why it is not a sequence lock////// The obvious construction is a sequence lock: mark the value in flux, write it,/// mark it settled. That needs a barrier *between* the odd marker and the payload/// write -- a release store keeps earlier writes from moving after it, but not/// later writes from moving before it -- and Zig 0.16 removed `@fence`, which was/// the way to ask for one.////// So the value is published into a small ring of slots instead. The writer fills/// the next slot and then advertises it with one release store; that single store/// is enough, because a release store is exactly the guarantee that everything/// written before it is visible to a reader whose acquire load sees it. A reader/// takes the counter, copies that slot, and takes the counter again to check it was/// not overtaken.////// The slot count is what makes the check sufficient. With `slots` places, the/// writer does not touch the slot a reader is copying until it has published/// `slots` more times -- four whole graph cycles, tens of milliseconds -- so a/// reader copying seventy-odd bytes cannot be overtaken mid-copy in any realistic/// scheduling. The recheck then catches the unrealistic case and retries.////// The writer is wait-free either way: two atomic operations and a struct copy per/// cycle, no spinning, no blocking, and nothing a reader can do to delay it. That/// is the requirement -- the writer is a real-time thread -- and it is why/// `Stream.stats` is safe to call from anywhere, including from inside a `Process`/// callback.pub fn Published(comptime T: type) type { return struct { const Self = @This();
/// Publications since the beginning. The low bits pick the slot. seq: std.atomic.Value(u32) = .init(0), slots: [slot_count]T = @splat(.{}),
/// Enough that a reader copying one slot cannot be overtaken by a writer /// coming round again: at one publication per graph cycle that is four /// cycles, tens of milliseconds. A power of two so the index is a mask. pub const slot_count = 4;
/// Publish. For the render thread only, and only one thread ever. pub fn store(self: *Self, value: T) void { // Only this thread writes `seq`, so reading our own counter needs no // ordering. const seq = self.seq.load(.monotonic); const next = seq +% 1;
// Written before it is advertised, and into a slot no reader can be // looking at yet, because nothing has pointed them here. self.slots[next & (slot_count - 1)] = value;
// The one barrier that matters: everything above is visible to any // reader whose acquire load sees this value. self.seq.store(next, .release); }
/// Read the most recent published value. pub fn load(self: *const Self) T { while (true) { const before = self.seq.load(.acquire); const copy = self.slots[before & (slot_count - 1)];
// Unchanged means the writer has not come round to this slot // again, so the copy is of one consistent publication. if (self.seq.load(.acquire) == before) return copy;
std.atomic.spinLoopHint(); } } };}
/// The cell a `Stream` publishes its statistics through.pub const Cell = Published(Stats);
test "an empty summary is zeroes rather than the sentinel minimum" { // `Running` starts its minimum at the largest `u64` so the first sample wins. // If that leaked out, a stream that had not run a cycle yet would report a // minimum of eighteen quintillion nanoseconds. const running: Accumulator.Running = .{}; const summary = running.summary(); try std.testing.expectEqual(@as(u64, 0), summary.min_ns); try std.testing.expectEqual(@as(u64, 0), summary.mean_ns); try std.testing.expectEqual(@as(u64, 0), summary.max_ns);}
test "a summary reports the extremes and the mean" { var running: Accumulator.Running = .{}; for ([_]u64{ 300, 100, 200, 400 }) |ns| running.add(ns);
const summary = running.summary(); try std.testing.expectEqual(@as(u64, 100), summary.min_ns); try std.testing.expectEqual(@as(u64, 250), summary.mean_ns); try std.testing.expectEqual(@as(u64, 400), summary.max_ns);}
test "the fraction of the period is taken from the worst cycle, not the mean" { // The whole point of reporting a maximum: a stream whose mean is comfortable // and whose worst cycle is over the deadline is a stream that glitches, and // the fraction has to say so. var running: Accumulator.Running = .{}; for ([_]u64{ 100, 100, 100, 9000 }) |ns| running.add(ns);
const summary = running.summary(); try std.testing.expectApproxEqAbs( @as(f64, 0.9), summary.fractionOfPeriod(10_000), 0.0001, );
// And an unknown period is zero rather than a division by it. try std.testing.expectEqual(@as(f64, 0), summary.fractionOfPeriod(0));}
test "recording a cycle advances both metrics and the count" { var acc: Accumulator = .{ .period_ns = 10_000_000 }; acc.record(1_000, 500_000); acc.record(3_000, 700_000);
const stats = acc.snapshot(); try std.testing.expectEqual(@as(u64, 2), stats.cycles); try std.testing.expectEqual(@as(u64, 1_000), stats.wake.min_ns); try std.testing.expectEqual(@as(u64, 3_000), stats.wake.max_ns); try std.testing.expectEqual(@as(u64, 600_000), stats.process.mean_ns); try std.testing.expectEqual(@as(u64, 10_000_000), stats.period_ns);}
test "a reset clears the window but keeps what describes the stream" { // A benchmark resetting to exclude start-up cycles did not mean to forget that // the device was swapped underneath it, nor to forget how long a cycle is. var acc: Accumulator = .{ .period_ns = 10_000_000 }; acc.record(1_000, 500_000); acc.underrun_frames = 480; acc.missed_cycles = 2; acc.reopens = 1;
acc.reset(); const stats = acc.snapshot();
try std.testing.expectEqual(@as(u64, 0), stats.cycles); try std.testing.expectEqual(@as(u64, 0), stats.underrun_frames); try std.testing.expectEqual(@as(u64, 0), stats.missed_cycles); try std.testing.expectEqual(@as(u64, 0), stats.wake.max_ns);
try std.testing.expectEqual(@as(u64, 10_000_000), stats.period_ns); try std.testing.expectEqual(@as(u64, 1), stats.reopens);}
test "a cell hands back what was put in it" { var cell: Cell = .{}; try std.testing.expectEqual(@as(u64, 0), cell.load().cycles);
cell.store(.{ .cycles = 42, .period_ns = 10_000_000 }); const got = cell.load(); try std.testing.expectEqual(@as(u64, 42), got.cycles); try std.testing.expectEqual(@as(u64, 10_000_000), got.period_ns);}
test "a reader never observes a torn snapshot" { // The property the sequence lock exists for. The writer publishes snapshots // whose fields are all derived from one counter, so any reader that saw a // half-written value would see fields that disagree with each other -- which // is exactly what is asserted here, across a hundred thousand publications // from a genuinely concurrent thread. const rounds = 100_000;
var cell: Cell = .{}; var stop: std.atomic.Value(bool) = .init(false);
const Writer = struct { fn run(c: *Cell, done: *std.atomic.Value(bool)) void { var i: u64 = 1; while (i <= rounds) : (i += 1) { // Every field a fixed multiple of `i`, so a torn read is visible // as an inconsistency rather than merely as a stale value. c.store(.{ .cycles = i, .period_ns = i * 2, .missed_cycles = i * 3, .underrun_frames = i * 4, .reopens = i * 5, .wake = .{ .min_ns = i, .mean_ns = i, .max_ns = i }, .process = .{ .min_ns = i * 6, .mean_ns = i * 6, .max_ns = i * 6 }, }); } done.store(true, .release); } };
const writer = try std.Thread.spawn(.{}, Writer.run, .{ &cell, &stop });
var reads: u64 = 0; while (!stop.load(.acquire)) { const s = cell.load(); reads += 1; if (s.cycles == 0) continue; // nothing published yet try std.testing.expectEqual(s.cycles * 2, s.period_ns); try std.testing.expectEqual(s.cycles * 3, s.missed_cycles); try std.testing.expectEqual(s.cycles * 4, s.underrun_frames); try std.testing.expectEqual(s.cycles * 5, s.reopens); try std.testing.expectEqual(s.cycles, s.wake.max_ns); try std.testing.expectEqual(s.cycles * 6, s.process.max_ns); }
writer.join(); try std.testing.expect(reads > 0);
// And the last value published is the one that is there afterwards. try std.testing.expectEqual(@as(u64, rounds), cell.load().cycles);}
test "stats print as a line worth logging" { var buf: [256]u8 = undefined; const quiet: Stats = .{ .cycles = 1000, .period_ns = 10_000_000, .wake = .{ .min_ns = 1_000, .mean_ns = 2_000, .max_ns = 5_000 }, .process = .{ .min_ns = 100_000, .mean_ns = 150_000, .max_ns = 300_000 }, }; try std.testing.expectEqualStrings( "1000 cycles, period 10000 us, wake 1/2/5 us, busy 100/150/300 us", try std.fmt.bufPrint(&buf, "{f}", .{quiet}), );
// The troubled ones only appear when they are non-zero, so a healthy stream's // line stays short enough to read. const troubled: Stats = .{ .cycles = 1, .underrun_frames = 480, .reopens = 2 }; const line = try std.fmt.bufPrint(&buf, "{f}", .{troubled}); try std.testing.expect(std.mem.find(u8, line, "480 underrun frames") != null); try std.testing.expect(std.mem.find(u8, line, "2 reopens") != null);}
test { std.testing.refAllDecls(@This());}