diff --git a/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift b/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift index 86dbb4cc..cfd751ff 100644 --- a/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift +++ b/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift @@ -132,6 +132,17 @@ final class SessionModel: ObservableObject { /// and under stage-1. @Published var osFloorP50Ms = 0.0 @Published var osFloorValid = false + /// The AUDIO plane's latency, from the playback ring (`SessionAudio.Stats`): how much decoded + /// audio is queued ahead of the speaker, and where that PUTS it relative to the picture + /// (positive = audio behind). `audioValid` is false until playback runs. + /// + /// Both numbers, never just the depth — a deep ring on a jittery link is the adaptive floor + /// doing its job, and only the offset separates that from audio simply being held late. They + /// existed nowhere a surface could render them until now, which is why a field report of "the + /// audio delay seems way too high" was triaged all the way to a conclusion without them. + @Published var audioBufferMs = 0 + @Published var audioAvOffsetMs = 0 + @Published var audioValid = false /// The floor-shaved values every HUD tier displays (raw − floor, never below 0). Identical /// to the raw values whenever no floor is measured. @@ -628,6 +639,7 @@ final class SessionModel: ObservableObject { displayValid = false clientQueueValid = false osFloorValid = false + audioValid = false lostFrames = 0 lostPct = 0 mouseCaptured = false @@ -702,7 +714,14 @@ final class SessionModel: ObservableObject { micUID: settings.micUID, micChannel: settings.micChannel, micEnabled: settings.micEnabled, - echoCancel: settings.echoCancel) + echoCancel: settings.echoCancel, + // The A/V sync reference: `endToEnd` is capture→on-glass, the one figure that says + // where the picture actually IS, and the audio ring steers its depth to land with it. + // The same meter object the presenter writes per presented frame, so audio reads the + // video plane's own measurement rather than a second estimate of it — and under the + // stage-1 fallback presenter, which stamps nothing, it stays empty and the loop + // correctly declines to correct. + videoLatency: endToEnd) self.audio = audio // Gamepads: forward every controller GamepadManager selected — each on its own wire pad // index (a pin forwards only one, Automatic forwards all) — and render the host's feedback @@ -860,6 +879,15 @@ final class SessionModel: ObservableObject { } else { self.clientQueueValid = false } + // The audio plane is a LEVEL, not a window: the ring's depth and the sync loop's + // smoothed offset are both current values, so they are read rather than drained. + if let a = self.audio?.stats { + self.audioBufferMs = a.bufferMS + self.audioAvOffsetMs = a.avOffsetMS + self.audioValid = true + } else { + self.audioValid = false + } // Mirror the window to the unified log (see statsLog) — one line per second, // stages in ms, only while frames actually flowed. `fps` counts RECEIVED AUs; // `presents` counts frames that reached glass (the display meter's sample count) @@ -875,7 +903,12 @@ final class SessionModel: ObservableObject { // the whole line (a cascade error that also mis-blames the float args). format: "fps=%lld presents=%lld e2e_p50=%.1f e2e_p95=%.1f hostnet_p50=%.1f " + "decode_p50=%.1f display_p50=%.1f lost=%lld " - + "floor_p50=%.1f display_adj=%.1f e2e_adj=%.1f queue_p50=%.1f", + + "floor_p50=%.1f display_adj=%.1f e2e_adj=%.1f queue_p50=%.1f " + // Appended LAST, so every existing parser of this line is unaffected. + // In the log as well as on the HUD because the overlay is only up when + // someone thought to turn it on, and the reports that need these + // numbers arrive after the fact. + + "audio_buffer=%lld audio_av_offset=%lld", frames, displayWindow?.count ?? 0, self.endToEndValid ? self.endToEndP50Ms : -1, @@ -887,7 +920,9 @@ final class SessionModel: ObservableObject { self.osFloorValid ? self.osFloorP50Ms : -1, self.displayValid ? self.displayAdjP50Ms : -1, self.endToEndValid ? self.endToEndAdjP50Ms : -1, - self.clientQueueValid ? self.clientQueueP50Ms : -1) + self.clientQueueValid ? self.clientQueueP50Ms : -1, + self.audioValid ? self.audioBufferMs : -1, + self.audioValid ? self.audioAvOffsetMs : 0) statsLog.info("\(line, privacy: .public)") } } diff --git a/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift b/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift index 9e06fdb3..02631eac 100644 --- a/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift +++ b/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift @@ -154,6 +154,28 @@ struct StreamHUDView: View { .foregroundStyle(.secondary) } } + // The AUDIO plane's own latency (detailed tier). Deliberately OUTSIDE the video branch + // above: it is not a term of that equation — audio is steered to MEET the video total, + // never summed into it — and the depth is exactly as worth seeing under the stage-1 + // fallback presenter, which measures no end-to-end at all. + // + // `buffer` is how much decoded audio is queued ahead of the speaker; `a/v` is where + // that puts it relative to the picture (+ = audio behind). Both, not just the depth: a + // deep ring on a jittery link is the adaptive floor doing its job, and only the offset + // distinguishes that from a ring holding audio late. Neither number was renderable + // anywhere before — they lived in a periodic log line — which is how a report of "the + // audio delay seems way too high" got triaged to a conclusion with no instrument. + if verbosity == .detailed && model.audioValid && model.audioBufferMs > 0 { + // String(format:) for the signed offset: `%+d` has no specifier-interpolation + // equivalent, and Swift's Int is 64-bit (%lld, never the 32-bit %d). + Text(model.audioAvOffsetMs == 0 + ? "audio buffer \(model.audioBufferMs) ms" + : String( + format: "audio buffer %lld ms · a/v %+lld ms", + model.audioBufferMs, model.audioAvOffsetMs)) + .font(.system(.caption2, design: .monospaced)) + .foregroundStyle(.tertiary) + } if model.lostFrames > 0 { // Unrecoverable network drops this window; hidden while the link is clean. // String(format:) rather than specifier interpolation: the literal % would diff --git a/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift b/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift index 222d4be0..f96aa4e6 100644 --- a/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift +++ b/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift @@ -21,6 +21,12 @@ import os /// long quiet spell relaxes it back toward the base — so a session on Wi-Fi that bunches arrivals /// deepens until it stops crackling, while a clean LAN keeps the tight base latency. Keep the /// constants here in step with `JitterTuning.COREAUDIO`. +/// +/// **A/V sync.** On top of all that the depth can be STEERED, by `setSyncTarget` from the drain +/// thread's `AvSync` — because a ring that is the right depth for the link is not thereby the +/// right depth for the picture. Continuity still outranks sync: the request is clamped between +/// the underrun-driven floor above and the hard cap, so the loop can never buy alignment with a +/// dropout. `nil` (the default) is exactly the pre-sync behaviour. final class AudioRing: @unchecked Sendable { /// Mirrors `JitterTuning::COREAUDIO` — see that type for the rationale. private static let targetMS = 20 @@ -48,6 +54,13 @@ final class AudioRing: @unchecked Sendable { private static let growWindowMS = 5_000 private static let growStepMS = 10 private static let shrinkQuietMS = 30_000 + /// The same quiet span, while the A/V sync loop is actively asking to run shallower. A grown + /// target normally relaxes only after a long spell because, absent other evidence, the only + /// thing that can justify giving up hard-won slack is time; a sync request IS that evidence — + /// a measurement saying the extra depth is costing alignment right now — so a smaller target + /// gets tested sooner. Wrong guesses are cheap and self-correcting (one underrun and the + /// growth path takes it straight back). Mirrors `SHRINK_QUIET_SYNC_MS`. + private static let shrinkQuietSyncMS = 5_000 private var buf: [Float] private var readIdx = 0 @@ -70,6 +83,14 @@ final class AudioRing: @unchecked Sendable { /// which is a different problem from the depth being wrong. private var underrunCount = 0 private var shedCount = 0 + /// The depth the A/V sync loop would like, in interleaved samples (`AvSync.desiredDepth`). + /// `nil` — the default, and what an un-wired session keeps — reproduces the pre-sync + /// behaviour exactly, so this ring could adopt sync without the other three diverging. + private var syncTarget: Int? + /// The sync loop's smoothed offset in ms, STORED not computed: the ring owns the depth but has + /// no timestamps, so the drain thread (which has both a packet's `pts_ns` and the video leg) + /// hands the number back for reporting. Mirrors `NativeClient::audio_av_offset_ms`. + private var avOffsetMS = 0 private let channels: Int private let perMS: Int private let lock = OSAllocatedUnfairLock() @@ -85,9 +106,64 @@ final class AudioRing: @unchecked Sendable { /// Effective target depth in interleaved samples: the (adaptively grown) live target, lifted /// so it can always serve one device quantum plus a packet (a large-buffer device cannot - /// sustain a target below its own quantum). + /// sustain a target below its own quantum) — then, if the A/V sync loop has asked for a depth, + /// its request CLAMPED into that band. Mirrors `JitterPolicy::effective_target`. + /// + /// The clamp order is the whole safety argument for steering playback depth off a network + /// measurement at all: sync may pull the ring shallower to catch the picture up, or push it + /// deeper when audio runs early, but never below what underrun pressure has proven this link + /// needs, and never past the hard cap that bounds added latency. A link whose jitter genuinely + /// demands more buffer than the picture is away keeps its buffer and the residual is REPORTED + /// (`Stats.avOffsetMS`) rather than taken out of the listener's stream. + /// + /// The ceiling is raised to the floor rather than used as-is: a device whose callback quantum + /// alone exceeds `hardCapMS` makes `floor > cap`, and a plain `min(max(s, floor), cap)` would + /// then return the CAP — i.e. quietly below the continuity floor, inverting the very ordering + /// this exists to guarantee, on exactly the awkward hardware it exists to survive. (Rust's + /// `Ord::clamp` announces the same condition by panicking; Swift would just get it wrong.) private var target: Int { - max(targetLive, renderQuantum + Self.frameMS * perMS) + let floor = max(targetLive, renderQuantum + Self.frameMS * perMS) + guard let want = syncTarget else { return floor } + let cap = max(Self.hardCapMS * perMS, floor) + return min(max(want, floor), cap) + } + + /// The sync loop is asking to run shallower than the adaptive target has grown to — the + /// evidence `noteRead` relaxes a grown target on. Compared against the LIVE target, not the + /// effective one: it is the underrun-driven growth that a sync request is evidence against, + /// not the device-quantum lift, which no amount of measurement can argue with. + private var syncWantsLess: Bool { + guard let want = syncTarget else { return false } + return want < targetLive + } + + /// Hand the ring the depth the A/V sync loop wants (`AvSync.desiredDepth`), in interleaved + /// samples, or `nil` to run unsynchronised. Called from the drain thread. + /// + /// This is a REQUEST, not a command — see `target` for what happens to it. `nil` is the + /// default and reproduces the pre-sync behaviour exactly. + func setSyncTarget(_ samples: Int?) { + lock.lock() + defer { lock.unlock() } + syncTarget = samples + } + + /// Store the sync loop's smoothed A/V offset for reporting (positive = audio behind the + /// picture). The ring cannot compute this — it has no timestamps — but it is where the two + /// numbers a listener's complaint needs, depth and offset, can be read under one lock. + func noteAvOffset(_ ms: Int) { + lock.lock() + defer { lock.unlock() } + avOffsetMS = ms + } + + /// Buffered depth in interleaved samples — what the sync loop measures against (`bufferedMS` + /// is the same quantity rounded for humans). Everything queued here must play before the frame + /// the drain thread is about to write, which is exactly what delays it. + var bufferedSamples: Int { + lock.lock() + defer { lock.unlock() } + return writeIdx - readIdx } func write(_ samples: UnsafePointer, count: Int) { @@ -196,7 +272,12 @@ final class AudioRing: @unchecked Sendable { } else { emptyReads = 0 quietRun += count - if quietRun >= Self.shrinkQuietMS * perMS { + // Without a sync request, time is the only evidence that hard-won slack is no longer + // needed, so a grown target waits out the long window. A request for less IS evidence, + // and without this branch a ring that ratcheted to the ceiling during a transient would + // hold audio a ceiling's worth late for minutes after the cause had gone. + let quietNeeded = syncWantsLess ? Self.shrinkQuietSyncMS : Self.shrinkQuietMS + if quietRun >= quietNeeded * perMS { quietRun = 0 targetLive = max(targetLive - Self.growStepMS * perMS, Self.targetMS * perMS) } @@ -239,6 +320,12 @@ final class AudioRing: @unchecked Sendable { let targetMS: Int let underruns: Int let sheds: Int + /// The A/V sync loop's smoothed offset (ms): **positive = audio playing BEHIND the + /// picture**, negative = ahead of it. `0` before the loop has evidence, or with sync off. + /// + /// Reported next to the depth, never instead of it: a deep ring on a jittery link is + /// CORRECT behaviour, and only the offset separates that from a ring holding audio late. + let avOffsetMS: Int } var stats: Stats { @@ -248,7 +335,154 @@ final class AudioRing: @unchecked Sendable { bufferedMS: (writeIdx - readIdx) / max(perMS, 1), targetMS: target / max(perMS, 1), underruns: underrunCount, - sheds: shedCount) + sheds: shedCount, + avOffsetMS: avOffsetMS) + } +} + +// MARK: - A/V sync + +/// The A/V synchronisation controller: turns "when will this audio actually play" and "when did +/// the picture it belongs with reach the glass" into a ring depth `AudioRing` should aim for. +/// The Swift mirror of `punktfunk_core::audio::AvSync` — keep the two in step. +/// +/// **The defect it exists to fix.** The host stamps `pts_ns` on every audio datagram and the +/// client decoded it into `AudioPCM` — and then never read it. Video's `pts_ns`, by contrast, is +/// used end to end (`LatencyMeter` computes a true glass-to-glass `displayed + clockOffset − pts` +/// per presented frame). So audio free-ran at whatever depth its jitter ring happened to settle +/// at, video was presented on a wholly independent path, and nothing ever compared them: the A/V +/// offset was an accident of buffer depths. It moved whenever the ring ratcheted under underrun +/// pressure, and — the way this surfaced in the field — it got WORSE every time video got faster, +/// because a quicker decoder lowers the video leg while leaving the audio leg exactly where it was. +/// +/// **Video is the master.** In a game streamer the video leg is the input-feel budget and must +/// never be inflated to satisfy the audio clock; audio tolerates small, crossfaded, rate-limited +/// corrections that are inaudible, and `AudioRing.shedOneFrame` already applies them. So audio +/// moves. +/// +/// **Continuity outranks sync.** This type only ever PROPOSES a depth. `AudioRing` clamps the +/// proposal to its own underrun-driven floor (see `AudioRing.target`), so a link whose jitter +/// genuinely needs more buffer than the picture is away keeps its buffer and the residual is +/// reported instead of being taken out of the listener's stream. +/// +/// Not a class and not locked: it is owned outright by the drain thread that observes packets. +struct AvSync { + /// Smoothing time constant for the measured offset, in ms of consumed audio. Long enough that + /// network jitter and a single late datagram do not move it; short enough to track real drift. + private static let ewmaTauMS = 2_000 + /// Offsets inside this band are left alone. Correcting a few ms costs a (crossfaded, but real) + /// discontinuity and buys nothing a listener can perceive — detectability for A/V misalignment + /// sits an order of magnitude above it. The deadband is what keeps the loop from hunting + /// forever around zero, which would be audible in a way the misalignment it chased was not. + private static let deadbandMS = 10 + /// Observations folded before the first correction is offered. The offset is derived from a + /// clock skew estimate and a video figure that both need a moment to settle after connect; + /// acting on the first sample would chase the handshake, not the stream. + private static let minObservations = 100 + /// An offset larger than this is not believed. A wall-clock step, a paused host, or a stale + /// video figure can all produce an enormous apparent misalignment, and steering the ring by it + /// would empty or overfill it outright. Beyond this the loop reports and waits rather than acts. + private static let saneLimitMS = 1_000 + /// The protocol's frame, in ms — the EWMA is weighted by it so the time constant means the + /// same thing however often the caller observes. + private static let frameMS = 5 + + /// Interleaved samples per millisecond at the negotiated layout (48 × channels). + private let perMS: Int + /// EWMA of the measured offset in ns. Positive = audio is scheduled to play LATE relative to + /// the picture it belongs with. + private var offsetAvgNs: Double = 0 + private var observations = 0 + /// Set once an observation lands outside `saneLimitMS`, for reporting. + private(set) var implausible = false + + /// `channels` is the negotiated interleaved channel count (2/6/8). + init(channels: Int) { + perMS = 48 * max(channels, 1) + } + + /// One measurement handed to `observe`. Every field is in the units its source already + /// produces, so no caller has to do clock arithmetic to use it correctly. + struct Observation { + /// The host capture timestamp carried by the audio frame being queued (host clock). + let ptsNs: UInt64 + /// Local `CLOCK_REALTIME` now — the same basis `LatencyMeter` stamps video in. + let nowLocalNs: Int64 + /// Host clock minus client clock, from the skew handshake (`clockOffsetNs`). + /// + /// It very nearly CANCELS: the video figure this is differenced against was computed with + /// the same offset and the same sign, so as long as both terms use one value the skew + /// drops out of the result entirely. That is what makes the connect-time offset good + /// enough here even though the absolute legs would prefer a re-synced one. + let clockOffsetNs: Int64 + /// How much audio is already queued AHEAD of this frame, in interleaved samples — + /// everything that must play before it does. + let bufferedAhead: Int + /// The video plane's current end-to-end figure in ns: `displayed + clockOffset − pts`, as + /// `LatencyMeter` already computes it per presented frame. `nil` while nothing has reached + /// the glass recently — no reference, no correction. + let videoE2eNs: Int64? + } + + /// Fold one measurement. Returns the smoothed offset in ns once there is enough evidence to + /// believe it (positive = audio late), or `nil` while still settling. + /// + /// Rejecting the implausible rather than clamping it is deliberate: a wall-clock step or a + /// stale video figure produces a huge apparent offset, and a clamped-but-wrong value would be + /// acted on as though it were a small real one. + @discardableResult + mutating func observe(_ o: Observation) -> Int64? { + // No frame on the glass yet ⇒ no reference to align against, so nothing to say. + guard let videoE2eNs = o.videoE2eNs else { return nil } + // When this frame's samples will actually reach the speaker, expressed in the host's + // capture clock — the same clock, and the same shape, as the video figure it is compared + // against. + let bufferedNs = Int64(o.bufferedAhead / max(perMS, 1)) * 1_000_000 + // Overflow-reporting arithmetic, NOT the wrapping `&+`/`&-` the meters use. Every term is + // a nanosecond count on the same epoch (~1.8e18), so the DIFFERENCE is tiny while the + // operands sit within a factor of five of `Int64.max` — and a garbage `pts_ns` would wrap + // a nonsense value round into a small, plausible-looking offset. This loop's entire + // defence is that it can tell nonsense from a real misalignment, so an overflow takes the + // same exit the sanity limit does rather than being silently believed. + let (playAtLocal, o1) = o.nowLocalNs.addingReportingOverflow(bufferedNs) + let (playAtHost, o2) = playAtLocal.addingReportingOverflow(o.clockOffsetNs) + let (audioE2eNs, o3) = playAtHost.subtractingReportingOverflow(Int64(bitPattern: o.ptsNs)) + let (offsetNs, o4) = audioE2eNs.subtractingReportingOverflow(videoE2eNs) + guard !o1, !o2, !o3, !o4, abs(offsetNs) <= Int64(Self.saneLimitMS) * 1_000_000 else { + implausible = true + return nil + } + implausible = false + + let alpha = min(1.0, Double(Self.frameMS) / Double(Self.ewmaTauMS)) + if observations == 0 { + offsetAvgNs = Double(offsetNs) + } else { + offsetAvgNs += (Double(offsetNs) - offsetAvgNs) * alpha + } + observations += 1 + return settled ? Int64(offsetAvgNs) : nil + } + + /// Enough evidence folded to act on. + var settled: Bool { observations >= Self.minObservations } + + /// The smoothed offset in ms (positive = audio late), for the HUD. Reported as soon as it is + /// measured, including while still settling — a number the operator can watch converge is more + /// useful than a blank that hides whether the loop is working at all. + var offsetMS: Int { Int(offsetAvgNs / 1_000_000) } + + /// The ring depth that would place audio with the picture, given where the ring is now. + /// `nil` while unsettled or inside the deadband — the caller then leaves the ring alone. + /// + /// Audio late (offset > 0) means there is too much queued: aim shallower. Audio early means + /// aim deeper. + func desiredDepth(currentDepth: Int) -> Int? { + guard settled else { return nil } + let offsetMs = offsetAvgNs / 1_000_000 + guard abs(offsetMs) >= Double(Self.deadbandMS) else { return nil } + let delta = Int(offsetMs * Double(perMS)) + return max(0, currentDepth - delta) } } diff --git a/clients/apple/Sources/PunktfunkKit/Audio/SessionAudio.swift b/clients/apple/Sources/PunktfunkKit/Audio/SessionAudio.swift index 60fd6ef4..82086456 100644 --- a/clients/apple/Sources/PunktfunkKit/Audio/SessionAudio.swift +++ b/clients/apple/Sources/PunktfunkKit/Audio/SessionAudio.swift @@ -62,6 +62,13 @@ public final class SessionAudio { /// not the ring, so the drain thread never has to be re-pointed). Main-thread confined, /// like every start path. private var ring: AudioRing? + /// The video plane's end-to-end meter (capture→on-glass), if the owner wired one — the + /// reference the A/V sync loop steers the ring against. `nil` leaves the loop inert and the + /// ring exactly as it was before sync existed, which is also what the stage-1 fallback + /// presenter gets: it decodes and presents inside the layer with no per-frame stamp, so it can + /// offer no reference, and a loop with no reference must not invent one. Main-thread confined, + /// like `ring`; the meter itself is internally locked and read from the drain thread. + private var videoLatency: LatencyMeter? #if !os(macOS) /// AVAudioSession `setCategory`/`setActive` are synchronous and block on the audio server, so /// they must not run on the main thread (UI stall — AVFoundation warns about it). PROCESS-WIDE @@ -91,9 +98,16 @@ public final class SessionAudio { /// a later main-queue hop (gated by `!flag.isStopped`) — so playback is live shortly after, not /// on return. The mic may start later still if the permission prompt is pending. /// `echoCancel` picks the engine topology — see the header note and `wantsCombined`. + /// + /// `videoLatency` is the session's END-TO-END latency meter (capture→on-glass). Pass it to arm + /// A/V sync: it is the only thing that tells the audio plane where the picture actually is, and + /// without it the ring keeps today's free-running behaviour. Omit it for a playback-only or + /// stage-1 session, where no such figure is measured. public func start( - speakerUID: String, micUID: String, micChannel: Int, micEnabled: Bool, echoCancel: Bool + speakerUID: String, micUID: String, micChannel: Int, micEnabled: Bool, echoCancel: Bool, + videoLatency: LatencyMeter? = nil ) { + self.videoLatency = videoLatency #if os(macOS) // No AVAudioSession on macOS — start the engines directly (caller's thread, as before). startEngines( @@ -305,6 +319,31 @@ public final class SessionAudio { } } + // MARK: - Stats + + /// The playback plane's two latency numbers, for the stats overlay. + /// + /// Both, never just the depth: a deep ring on a jittery link is CORRECT behaviour — the + /// adaptive floor put it there because the link kept starving — and only the offset separates + /// that from a ring that is simply holding audio late. Before this pair existed the plane + /// published nothing any surface could render (depth and target lived in a periodic log line), + /// and a field investigation into "the audio delay seems way too high" ran all the way to its + /// conclusion without either number. + public struct Stats: Sendable { + /// Decoded audio queued ahead of the speaker (ms). + public let bufferMS: Int + /// The A/V sync loop's smoothed offset (ms): positive = audio playing BEHIND the picture. + /// `0` before the loop has evidence, with sync unwired, or genuinely aligned. + public let avOffsetMS: Int + } + + /// A snapshot of `Stats`, or nil before playback starts. Main thread (`ring` is main-confined; + /// the ring's own numbers are taken under its lock, so they describe one instant). + public var stats: Stats? { + guard let s = ring?.stats else { return nil } + return Stats(bufferMS: s.bufferedMS, avOffsetMS: s.avOffsetMS) + } + // MARK: - Playback (host → speaker) /// The playback jitter ring + the source node draining it — shared by the plain playback @@ -401,9 +440,25 @@ public final class SessionAudio { } drainStarted = true stateLock.unlock() + // A/V sync. This thread is the only place that holds all three ingredients at once: the + // packet's host capture `ptsNs`, the ring depth, and the video plane's end-to-end figure. + // `ptsNs` was decoded into `AudioPCM` and then dropped on the floor right here for the + // plane's entire existence, which is why audio ran at whatever depth its jitter ring + // happened to settle at and nothing ever placed it against the picture. + // + // The escape hatch mirrors the Rust clients': a field regression in a loop that steers + // PLAYBACK should be bisectable without a rebuild. macOS honours it from the environment; + // elsewhere it simply never trips, which is the same as today's behaviour. + let syncEnabled = !["1", "true"].contains( + ProcessInfo.processInfo.environment["PUNKTFUNK_NO_AV_SYNC"] ?? "") + // nil disarms the loop entirely — no reference, no correction (see `videoLatency`). + let videoLatency = syncEnabled ? self.videoLatency : nil + if !syncEnabled { log.info("A/V sync disabled by PUNKTFUNK_NO_AV_SYNC") } + let channels = Int(connection.resolvedAudioChannels) let thread = Thread { [connection, flag, drainDone] in defer { drainDone.signal() } var drained = 0 + var av = AvSync(channels: channels) // Decode happens IN-CORE (libopus multistream) — AudioToolbox's Opus path is // stereo-only — and is handed back as interleaved f32 PCM in wire channel order. // Per-iteration autorelease pool: no runloop on this thread (see Stage2Pipeline). @@ -417,6 +472,25 @@ public final class SessionAudio { return false // session closed } guard let pcm, pcm.frameCount > 0 else { return true } + // Place this frame against the picture it belongs with BEFORE queueing it: the + // depth read here is everything that must still play first, which is exactly what + // delays it. Skipped wholesale when no meter was wired, so an un-armed session + // does not even read the ring. + if let videoLatency { + let depth = ring.bufferedSamples + var ts = timespec() + clock_gettime(CLOCK_REALTIME, &ts) + let nowNs = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec) + // Half a second of tolerance on the reference: long enough to ride out a + // stalled or hitching present path, short enough that a backgrounded session + // (video decode dropped, audio still playing) stops steering almost at once. + av.observe(AvSync.Observation( + ptsNs: pcm.ptsNs, nowLocalNs: nowNs, + clockOffsetNs: connection.clockOffsetNs, bufferedAhead: depth, + videoE2eNs: videoLatency.latestSample(asOfNs: nowNs, maxAgeMs: 500))) + ring.setSyncTarget(av.desiredDepth(currentDepth: depth)) + ring.noteAvOffset(av.offsetMS) + } pcm.samples.withUnsafeBufferPointer { p in if let base = p.baseAddress { ring.write(base, count: pcm.frameCount * pcm.channels) @@ -430,7 +504,7 @@ public final class SessionAudio { if drained % 2_000 == 0 { let s = ring.stats log.info( - "audio: buffer_ms=\(s.bufferedMS) target_ms=\(s.targetMS) underruns=\(s.underruns) drift_sheds=\(s.sheds)" + "audio: buffer_ms=\(s.bufferedMS) target_ms=\(s.targetMS) underruns=\(s.underruns) drift_sheds=\(s.sheds) av_offset_ms=\(s.avOffsetMS)" ) } return true diff --git a/clients/apple/Sources/PunktfunkKit/Video/LatencyMeter.swift b/clients/apple/Sources/PunktfunkKit/Video/LatencyMeter.swift index a3162043..919a8ee0 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/LatencyMeter.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/LatencyMeter.swift @@ -24,6 +24,10 @@ public final class LatencyMeter: @unchecked Sendable { private let lock = NSLock() private var samplesUs: [Int64] = [] private var skewCorrected = false + /// The most recent sample and the instant it ended, for `latestSample(asOfNs:maxAgeMs:)` — + /// a LEVEL, not a window, so `drain` deliberately leaves both alone. + private var latestNs: Int64 = 0 + private var latestAtNs: Int64 = 0 public init() {} @@ -49,10 +53,42 @@ public final class LatencyMeter: @unchecked Sendable { guard latNs > 0, latNs < 10_000_000_000 else { return } lock.lock() samplesUs.append(latNs / 1000) + latestNs = latNs + latestAtNs = atNs if offsetNs != 0 { skewCorrected = true } lock.unlock() } + /// The most recent single sample in ns, or `nil` if none has landed or the last one ended more + /// than `maxAgeMs` before `nowNs` (both `CLOCK_REALTIME`). Unlike `drain`, this reports a level + /// rather than a window, and reading it consumes nothing. + /// + /// **What it is for.** Read off the END-TO-END meter, this is the video plane's live + /// glass-to-glass figure — `displayed + clockOffset − pts`, exactly the shape `AvSync` compares + /// audio against — and it is the reference the A/V sync loop needs. It is published from + /// `record`, so BOTH present paths (arrival and deadline) feed it without either knowing that + /// audio exists. + /// + /// **Why staleness is not optional.** The number is a level, so absent an age check it would + /// simply keep its last value forever. This client has a state where that matters: the + /// backgrounded keep-alive keeps audio playing and DROPS video decode entirely, so the loop + /// would go on steering the ring against a reference minutes old and frozen. Expiring it + /// returns `nil`, which is the same "no reference yet" case as session start — the loop holds + /// its last correction and stops chasing. `nowNs` is caller-supplied rather than read fresh so + /// the audio side compares against exactly the instant it timestamped its own frame at. + /// + /// Only the PAST is bounded. A present stamp can legitimately sit a hair ahead of the reader's + /// clock (the deadline presenter stamps at the link's target present time), and discarding the + /// only reference we have over a fraction of a refresh would make it flap in and out; a stamp + /// wildly in the future instead yields a huge offset, which `AvSync` refuses on its own terms. + public func latestSample(asOfNs nowNs: Int64, maxAgeMs: Int) -> Int64? { + lock.lock() + defer { lock.unlock() } + guard latestNs > 0 else { return nil } + guard (nowNs &- latestAtNs) <= Int64(maxAgeMs) * 1_000_000 else { return nil } + return latestNs + } + public struct Stats: Sendable { public let p50Ms: Double public let p95Ms: Double diff --git a/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift b/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift index 7fb85be1..6888230f 100644 --- a/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift +++ b/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift @@ -198,5 +198,315 @@ final class AudioRingDriftTests: XCTestCase { silentTail, 0, "after adapting, the last 3 s must play through the bunching without a dropout") } + + // MARK: - A/V sync (audio latency overhaul, W6) + // + // The second half of the same story. Depth alone is not correctness: a ring can be exactly as + // deep as its link needs and still put audio in the wrong place, because nothing ever compared + // it to the picture. `AvSync` measures that comparison and asks the ring to move; the ring is + // free to refuse. These pin both halves — that the loop DOES act (the previous pass in this + // area shipped a correction that was structurally unreachable and had a green test), and that + // it can never act far enough to starve the callback. + + /// Build an observation whose measured offset is exactly `offsetMS` (positive = audio late). + /// Mirrors the Rust `obs` helper: pin now/skew/pts so the only free term is the buffered depth, + /// then choose the video figure so the difference lands where we want it. + private func obs(offsetMS: Int, depth: Int) -> AvSync.Observation { + let bufferedMS = depth / perMS + let audioE2eMS = bufferedMS + 40 // 40 ms of transport, arbitrary but fixed + let videoE2eMS = audioE2eMS - offsetMS + return AvSync.Observation( + ptsNs: 1_000_000_000, + nowLocalNs: 1_000_000_000 + 40 * 1_000_000, + clockOffsetNs: 0, + bufferedAhead: depth, + videoE2eNs: Int64(max(0, videoE2eMS)) * 1_000_000) + } + + /// Fold `n` identical observations in. + private func settle(_ sync: inout AvSync, offsetMS: Int, depth: Int, count: Int = 100) { + for _ in 0.. Int { + var scratch = [Float](repeating: 0, count: want) + let feed = [Float](repeating: 0.5, count: 5 * perMS) + let start = ring.stats.targetMS + var reads = 0 + while ring.stats.targetMS == start, reads < 200_000 { + feed.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: 5 * perMS) } + scratch.withUnsafeMutableBufferPointer { + ring.read(into: $0.baseAddress!, count: want) + } + reads += 1 + } + return reads + } + + let slow = AudioRing(capacity: 48_000 * channels, channels: channels) + grow(slow) + slow.setSyncTarget(nil) + let slowReads = quietToRelax(slow) + + let fast = AudioRing(capacity: 48_000 * channels, channels: channels) + grow(fast) + fast.setSyncTarget(perMS) // strictly shallower than the grown target + let fastReads = quietToRelax(fast) + + XCTAssertLessThan( + fastReads, slowReads, + "sync pressure should relax sooner: \(fastReads) vs \(slowReads) quiet reads") + } + + /// The four client rings adopt sync one at a time; an un-wired one must behave exactly as it + /// did. `nil` is the default, so this pins the initializer too — and every other test in this + /// file runs without a sync target, which is the real guard that nothing moved underneath them. + func testNoSyncTargetLeavesTheRingExactlyAsItWas() { + let a = AudioRing(capacity: 48_000 * channels, channels: channels) + let b = AudioRing(capacity: 48_000 * channels, channels: channels) + b.setSyncTarget(nil) + let want = 5 * perMS + var sa = [Float](repeating: 0, count: want) + var sb = [Float](repeating: 0, count: want) + let feed = [Float](repeating: 0.5, count: 30 * perMS) + for step in 0..<4_000 { + // Uneven delivery so the depth actually moves around and the two rings have something + // to disagree about. + if step % 7 == 0 { + for r in [a, b] { + feed.withUnsafeBufferPointer { r.write($0.baseAddress!, count: 30 * perMS) } + } + } + sa.withUnsafeMutableBufferPointer { a.read(into: $0.baseAddress!, count: want) } + sb.withUnsafeMutableBufferPointer { b.read(into: $0.baseAddress!, count: want) } + XCTAssertEqual(sa, sb, "step \(step): an explicit nil diverged from the default") + XCTAssertEqual(a.stats.targetMS, b.stats.targetMS, "step \(step)") + } + } + + /// The reporting half of §1.3: the offset must reach the same snapshot the depth does, because + /// a depth on its own cannot distinguish "deep because the link needs it" from "deep and + /// therefore late". This is the number the HUD and the 1 Hz log line read. + func testAvOffsetIsReportedAlongsideTheDepth() { + let ring = AudioRing(capacity: 48_000 * channels, channels: channels) + XCTAssertEqual(ring.stats.avOffsetMS, 0, "no evidence yet reads as zero, not as noise") + let feed = [Float](repeating: 0.5, count: 30 * perMS) + feed.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: 30 * perMS) } + + var s = AvSync(channels: channels) + settle(&s, offsetMS: 37, depth: 30 * perMS, count: 400) + ring.noteAvOffset(s.offsetMS) + let stats = ring.stats + XCTAssertEqual(stats.bufferedMS, 30) + XCTAssertEqual(stats.avOffsetMS, 37, "positive = audio behind the picture") + } } #endif diff --git a/clients/apple/Tests/PunktfunkKitTests/LatencyMeterTests.swift b/clients/apple/Tests/PunktfunkKitTests/LatencyMeterTests.swift index bb647098..72455723 100644 --- a/clients/apple/Tests/PunktfunkKitTests/LatencyMeterTests.swift +++ b/clients/apple/Tests/PunktfunkKitTests/LatencyMeterTests.swift @@ -71,4 +71,65 @@ final class LatencyMeterTests: XCTestCase { m.record(ptsNs: now - 20_000_000_000, offsetNs: 0) XCTAssertNil(m.drain()) } + + // MARK: - latestSample: the A/V sync loop's video reference + + /// The end-to-end meter doubles as the reference the audio ring steers against, so its most + /// recent sample must be readable as a LEVEL — without consuming it, and independently of the + /// 1 Hz percentile window the HUD drains. + func testLatestSampleSurvivesDrainAndIsNotAWindow() { + let m = LatencyMeter() + let atNs: Int64 = 1_000_000_000_000 + m.record(ptsNs: UInt64(atNs - 12_000_000), atNs: atNs, offsetNs: 0) // 12 ms + XCTAssertEqual(m.latestSample(asOfNs: atNs, maxAgeMs: 500), 12_000_000) + _ = m.drain() + XCTAssertEqual( + m.latestSample(asOfNs: atNs, maxAgeMs: 500), 12_000_000, + "the reference is a level — draining the percentile window must not clear it") + // …and it tracks the newest frame. + m.record(ptsNs: UInt64(atNs - 20_000_000), atNs: atNs, offsetNs: 0) + XCTAssertEqual(m.latestSample(asOfNs: atNs, maxAgeMs: 500), 20_000_000) + } + + /// No frame yet ⇒ no reference. This is what keeps the sync loop inert at session start and + /// under the stage-1 presenter, which stamps no present at all. + func testLatestSampleIsNilBeforeAnyFrame() { + XCTAssertNil(LatencyMeter().latestSample(asOfNs: 1_000_000_000_000, maxAgeMs: 500)) + } + + /// THE staleness gate: video can stop while audio keeps playing (the backgrounded keep-alive + /// drops decode entirely). A level with no expiry would go on offering a minutes-old figure as + /// though it were live, and the ring would be steered against a frozen reference. + func testLatestSampleExpires() { + let m = LatencyMeter() + let atNs: Int64 = 1_000_000_000_000 + m.record(ptsNs: UInt64(atNs - 12_000_000), atNs: atNs, offsetNs: 0) + XCTAssertNotNil(m.latestSample(asOfNs: atNs + 499_000_000, maxAgeMs: 500)) + XCTAssertNil( + m.latestSample(asOfNs: atNs + 501_000_000, maxAgeMs: 500), + "a stale reference must read as NO reference, not as a live one") + // A stamp marginally ahead of the reader's clock is normal (the deadline presenter stamps + // at the link's TARGET present time) and must not drop the only reference we have. + XCTAssertNotNil(m.latestSample(asOfNs: atNs - 8_000_000, maxAgeMs: 500)) + } + + /// A sample the meter refused must not become a reference either — the sync loop would then be + /// steered by a value the percentile window itself judged absurd. + /// + /// The ABSURDLY LARGE case is the load-bearing one: a negative interval would also be stopped + /// by `latestSample`'s own `> 0` check, so on its own it proves nothing about where the publish + /// sits relative to the guard. + func testRefusedSampleIsNotPublishedAsAReference() { + let m = LatencyMeter() + let atNs: Int64 = 1_000_000_000_000 + m.record(ptsNs: UInt64(atNs - 20_000_000_000), atNs: atNs, offsetNs: 0) // 20 s → refused + XCTAssertNil( + m.latestSample(asOfNs: atNs, maxAgeMs: 500), + "a sample too absurd for the window is too absurd to steer the ring") + m.record(ptsNs: UInt64(atNs + 1), atNs: atNs, offsetNs: 0) // negative interval + XCTAssertNil(m.latestSample(asOfNs: atNs, maxAgeMs: 500)) + // …and a good sample after them still lands, so the refusals cost nothing. + m.record(ptsNs: UInt64(atNs - 9_000_000), atNs: atNs, offsetNs: 0) + XCTAssertEqual(m.latestSample(asOfNs: atNs, maxAgeMs: 500), 9_000_000) + } }