[client] audio: rework synchronization for low latency

Drive playback timing from audio sample positions instead of packet arrival cadence. Use SPICE multimedia timestamps to anchor streams, detect discontinuities, and estimate the long-term source rate. Ignore their coarse phase corrections when steering the playback buffer. Measure the device clock independently and wait for it to stabilize before enabling a bounded, slew-limited phase controller. Align startup buffering from logical producer and consumer positions so backend startup reserves do not become persistent latency. Size the buffer from the backend period, measured delivery jitter, and resampler delay. Update PipeWire and PulseAudio to report their actual startup requirements and presentation latency. Default to a 10 ms device period with 4 ms of additional buffering, and make detailed synchronization diagnostics opt-in through audio:debug. This reduces startup and steady-state latency while compensating for physical and virtual device clock drift without reacting to transport jitter or SPICE timestamp quantization.
2026-08-02 05:12:02 +00:00 · 2026-07-28 21:07:02 +10:00
parent 4ef6672e64
commit d956dd8fd3
8 changed files with 1123 additions and 383 deletions
--- a/client/audiodevs/PipeWire/pipewire.c
+++ b/client/audiodevs/PipeWire/pipewire.c
@@ -22,8 +22,11 @@

 #include <spa/param/audio/format-utils.h>
 #include <spa/param/props.h>
+#include <spa/utils/result.h>
 #include <pipewire/pipewire.h>
+#include <limits.h>
 #include <math.h>
+#include <stdatomic.h>

 #include "common/debug.h"
 #include "common/stringutils.h"
@@ -48,7 +51,7 @@ struct PipeWire
  {
    struct pw_stream * stream;
    struct spa_io_rate_match * rateMatch;
-    struct pw_time time;
+    _Atomic(int64_t) presentationDeadline;

    int            channels;
    int            sampleRate;
@@ -77,6 +80,42 @@ struct PipeWire

 static struct PipeWire pw = {0};

+static int64_t pipewire_monotonicTime(void)
+{
+  struct timespec time;
+  clock_gettime(CLOCK_MONOTONIC, &time);
+  return SPA_TIMESPEC_TO_NSEC(&time);
+}
+
+static void pipewire_updatePlaybackLatency(void)
+{
+  struct pw_time time;
+#if PW_CHECK_VERSION(0, 3, 50)
+  if (pw_stream_get_time_n(pw.playback.stream, &time, sizeof(time)) < 0)
+#else
+  if (pw_stream_get_time(pw.playback.stream, &time) < 0)
+#endif
+  {
+    DEBUG_ERROR("pw_stream_get_time failed");
+    return;
+  }
+
+  if (time.rate.num == 0 || time.rate.denom == 0)
+    return;
+
+#if PW_CHECK_VERSION(0, 3, 50)
+  const double latencyTicks =
+    time.delay + (double)time.queued + time.buffered;
+#else
+  const double latencyTicks =
+    time.delay + (double)time.queued / pw.playback.stride;
+#endif
+  const int64_t latencyNs = llrint(max(0.0, latencyTicks) *
+      time.rate.num * SPA_NSEC_PER_SEC / time.rate.denom);
+  atomic_store_explicit(&pw.playback.presentationDeadline,
+      time.now + latencyNs, memory_order_release);
+}
+
 static void pipewire_onPlaybackIoChanged(void * userdata, uint32_t id,
  void * data, uint32_t size)
 {
@@ -92,14 +131,6 @@ static void pipewire_onPlaybackProcess(void * userdata)
 {
  struct pw_buffer * pbuf;

-#if PW_CHECK_VERSION(0, 3, 50)
-  if (pw_stream_get_time_n(pw.playback.stream, &pw.playback.time,
-        sizeof(pw.playback.time)) < 0)
-#else
-  if (pw_stream_get_time(pw.playback.stream, &pw.playback.time) < 0)
-#endif
-    DEBUG_ERROR("pw_stream_get_time failed");
-
  if (!(pbuf = pw_stream_dequeue_buffer(pw.playback.stream)))
  {
    DEBUG_WARN("out of buffers");
@@ -110,7 +141,10 @@ static void pipewire_onPlaybackProcess(void * userdata)
  uint8_t * dst;

  if (!(dst = sbuf->datas[0].data))
+  {
+    pw_stream_return_buffer(pw.playback.stream, pbuf);
    return;
+  }

  int frames = sbuf->datas[0].maxsize / pw.playback.stride;
  if (pw.playback.rateMatch && pw.playback.rateMatch->size > 0)
@@ -121,6 +155,7 @@ static void pipewire_onPlaybackProcess(void * userdata)
  {
    sbuf->datas[0].chunk->size = 0;
    pw_stream_queue_buffer(pw.playback.stream, pbuf);
+    pipewire_updatePlaybackLatency();
    return;
  }

@@ -130,6 +165,7 @@ static void pipewire_onPlaybackProcess(void * userdata)
  sbuf->datas[0].chunk->size   = frames * pw.playback.stride;

  pw_stream_queue_buffer(pw.playback.stream, pbuf);
+  pipewire_updatePlaybackLatency();
 }

 static void pipewire_onPlaybackDrained(void * userdata)
@@ -221,10 +257,12 @@ static void pipewire_playbackStopStream(void)
  pw_stream_destroy(pw.playback.stream);
  pw.playback.stream    = NULL;
  pw.playback.rateMatch = NULL;
+  atomic_store_explicit(
+      &pw.playback.presentationDeadline, 0, memory_order_release);
  pw_thread_loop_unlock(pw.thread);
 }

-static void pipewire_playbackSetup(int channels, int sampleRate,
+static bool pipewire_playbackSetup(int channels, int sampleRate,
    int requestedPeriodFrames, int * maxPeriodFrames, int * startFrames,
    LG_AudioPullFn pullFn)
 {
@@ -245,7 +283,7 @@ static void pipewire_playbackSetup(int channels, int sampleRate,
  {
    *maxPeriodFrames = pw.playback.maxPeriodFrames;
    *startFrames     = pw.playback.startFrames;
-    return;
+    return true;
  }

  pipewire_playbackStopStream();
@@ -271,6 +309,12 @@ static void pipewire_playbackSetup(int channels, int sampleRate,
      PW_KEY_NODE_LATENCY  , requestedNodeLatency,
      NULL
    );
+  if (!props)
+  {
+    pw_thread_loop_unlock(pw.thread);
+    DEBUG_ERROR("Failed to create playback stream properties");
+    return false;
+  }

  const char * device = option_get_string("pipewire", "outDevice");
  if (device)
@@ -290,23 +334,31 @@ static void pipewire_playbackSetup(int channels, int sampleRate,
    NULL
  );

+  if (!pw.playback.stream)
+  {
+    pw_thread_loop_unlock(pw.thread);
+    DEBUG_ERROR("Failed to create the stream");
+    return false;
+  }
+
  // The user can override the default node latency with the PIPEWIRE_LATENCY
  // environment variable, so get the actual node latency value from the stream.
  // The actual quantum size may be lower than this value depending on what else
  // is using the audio device, but we can treat this value as a maximum
  const struct pw_properties * properties =
    pw_stream_get_properties(pw.playback.stream);
-  const char *actualNodeLatency =
-    pw_properties_get(properties, PW_KEY_NODE_LATENCY);
-  DEBUG_ASSERT(actualNodeLatency != NULL);
+  const char *actualNodeLatency = properties ?
+    pw_properties_get(properties, PW_KEY_NODE_LATENCY) : NULL;

  unsigned num, denom;
-  if (sscanf(actualNodeLatency, "%u/%u", &num, &denom) != 2 ||
-    denom != sampleRate)
+  if (!actualNodeLatency ||
+    sscanf(actualNodeLatency, "%u/%u", &num, &denom) != 2 ||
+    num == 0 || num > INT_MAX || denom != sampleRate)
  {
    DEBUG_WARN(
      "PIPEWIRE_LATENCY value '%s' is invalid or does not match stream sample "
-      "rate; using %d/%d", actualNodeLatency, requestedPeriodFrames,
+      "rate; using %d/%d",
+      actualNodeLatency ? actualNodeLatency : "(unset)", requestedPeriodFrames,
      sampleRate);

    struct spa_dict_item items[] = {
@@ -327,13 +379,6 @@ static void pipewire_playbackSetup(int channels, int sampleRate,
  *maxPeriodFrames = pw.playback.maxPeriodFrames;
  *startFrames     = pw.playback.startFrames;

-  if (!pw.playback.stream)
-  {
-    pw_thread_loop_unlock(pw.thread);
-    DEBUG_ERROR("Failed to create the stream");
-    return;
-  }
-
  params[0] = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat,
      &SPA_AUDIO_INFO_RAW_INIT(
        .format   = SPA_AUDIO_FORMAT_F32,
@@ -341,7 +386,7 @@ static void pipewire_playbackSetup(int channels, int sampleRate,
        .rate     = sampleRate
        ));

-  pw_stream_connect(
+  const int result = pw_stream_connect(
      pw.playback.stream,
      PW_DIRECTION_OUTPUT,
      PW_ID_ANY,
@@ -351,7 +396,21 @@ static void pipewire_playbackSetup(int channels, int sampleRate,
      PW_STREAM_FLAG_INACTIVE,
      params, 1);

+  if (result < 0)
+  {
+    DEBUG_ERROR("Failed to connect playback stream: %s", spa_strerror(result));
+    pw_stream_destroy(pw.playback.stream);
+    pw.playback.stream = NULL;
+    pw.playback.rateMatch = NULL;
+    pw_thread_loop_unlock(pw.thread);
+    return false;
+  }
+
+  pw.playback.state = STREAM_STATE_INACTIVE;
+  atomic_store_explicit(
+      &pw.playback.presentationDeadline, 0, memory_order_release);
  pw_thread_loop_unlock(pw.thread);
+  return true;
 }

 static void pipewire_playbackStart(void)
@@ -431,30 +490,12 @@ static void pipewire_playbackMute(bool mute)

 static uint64_t pipewire_playbackLatency(void)
 {
-#if PW_CHECK_VERSION(0, 3, 50)
-  if (pw.playback.time.rate.num == 0)
+  const int64_t deadline = atomic_load_explicit(
+      &pw.playback.presentationDeadline, memory_order_acquire);
+  if (deadline <= 0)
    return 0;

-  struct timespec ts;
-  clock_gettime(CLOCK_MONOTONIC, &ts);
-
-  // diff in ns
-  int64_t diff = SPA_TIMESPEC_TO_NSEC(&ts) - pw.playback.time.now;
-
-  // elapsed frames
-  int64_t elapsed =
-    (pw.playback.time.rate.denom * diff) /
-    (pw.playback.time.rate.num * SPA_NSEC_PER_SEC);
-
-  const int64_t buffered = pw.playback.time.buffered + pw.playback.time.queued;
-  int64_t latency = (buffered * 1000 / pw.playback.sampleRate) +
-     ((pw.playback.time.delay - elapsed) * 1000 *
-      pw.playback.time.rate.num / pw.playback.time.rate.denom);
-
-  return max(0, -latency);
-#else
-  return pw.playback.time.delay + pw.playback.time.queued / pw.playback.stride;
-#endif
+  return max(INT64_C(0), deadline - pipewire_monotonicTime()) / 1000;
 }

 static void pipewire_recordStopStream(void)
--- a/client/audiodevs/PulseAudio/pulseaudio.c
+++ b/client/audiodevs/PulseAudio/pulseaudio.c
@@ -23,8 +23,10 @@
 #include <pulse/pulseaudio.h>
 #include <string.h>
 #include <math.h>
+#include <stdatomic.h>

 #include "common/debug.h"
+#include "common/time.h"

 struct PulseAudio
 {
@@ -44,10 +46,17 @@ struct PulseAudio
  int                    sinkChannels;
  int                    sinkStride;
  LG_AudioPullFn         sinkPullFn;
+  _Atomic(int64_t)       sinkPresentationDeadline;
 };

 static struct PulseAudio pa = {0};

+static void pulseaudio_unrefOperation(pa_operation * operation)
+{
+  if (operation)
+    pa_operation_unref(operation);
+}
+
 static void pulseaudio_sink_input_cb(pa_context *c, const pa_sink_input_info *i,
    int eol, void *userdata)
 {
@@ -69,7 +78,7 @@ static void pulseaudio_subscribe_cb(pa_context *c,
      {
        pa_operation *o = pa_context_get_sink_input_info(c, index,
            pulseaudio_sink_input_cb, NULL);
-        pa_operation_unref(o);
+        pulseaudio_unrefOperation(o);
      }
      break;
  }
@@ -194,10 +203,15 @@ static void pulseaudio_sink_close_nl(void)
  if (!pa.sink)
    return;

+  pa_stream_set_state_callback(pa.sink, NULL, NULL);
  pa_stream_set_write_callback(pa.sink, NULL, NULL);
-  pa_stream_flush(pa.sink, NULL, NULL);
+  pa_stream_set_underflow_callback(pa.sink, NULL, NULL);
+  pa_stream_set_overflow_callback(pa.sink, NULL, NULL);
+  pulseaudio_unrefOperation(pa_stream_flush(pa.sink, NULL, NULL));
  pa_stream_unref(pa.sink);
  pa.sink = NULL;
+  atomic_store_explicit(
+      &pa.sinkPresentationDeadline, 0, memory_order_release);
 }

 static void pulseaudio_free(void)
@@ -224,10 +238,12 @@ static void pulseaudio_state_cb(pa_stream * p, void * userdata)
 {
  if (pa.sinkStarting && pa_stream_get_state(pa.sink) == PA_STREAM_READY)
  {
-    pa_stream_cork(pa.sink, 0, NULL, NULL);
+    pulseaudio_unrefOperation(pa_stream_cork(pa.sink, 0, NULL, NULL));
    pa.sinkCorked   = false;
    pa.sinkStarting = false;
  }
+
+  pa_threaded_mainloop_signal(pa.loop, 0);
 }

 static void pulseaudio_write_cb(pa_stream * p, size_t nbytes, void * userdata)
@@ -239,12 +255,38 @@ static void pulseaudio_write_cb(pa_stream * p, size_t nbytes, void * userdata)

  uint8_t * dst;

-  pa_stream_begin_write(p, (void **)&dst, &nbytes);
+  if (pa_stream_begin_write(p, (void **)&dst, &nbytes) < 0)
+  {
+    DEBUG_ERROR("pa_stream_begin_write failed: %s",
+        pa_strerror(pa_context_errno(pa.context)));
+    return;
+  }

  int frames = nbytes / pa.sinkStride;
  frames = pa.sinkPullFn(dst, frames);
+  if (frames <= 0)
+  {
+    pa_stream_cancel_write(p);
+    return;
+  }

-  pa_stream_write(p, dst, frames * pa.sinkStride, NULL, 0, PA_SEEK_RELATIVE);
+  if (pa_stream_write(
+        p, dst, frames * pa.sinkStride, NULL, 0, PA_SEEK_RELATIVE) < 0)
+  {
+    DEBUG_ERROR("pa_stream_write failed: %s",
+        pa_strerror(pa_context_errno(pa.context)));
+    pa_stream_cancel_write(p);
+    return;
+  }
+
+  pa_usec_t latency;
+  int negative;
+  if (pa_stream_get_latency(p, &latency, &negative) == 0)
+  {
+    const int64_t latencyNs = negative ? 0 : (int64_t)latency * 1000;
+    atomic_store_explicit(&pa.sinkPresentationDeadline,
+        (int64_t)nanotime() + latencyNs, memory_order_release);
+  }
 }

 static void pulseaudio_underflow_cb(pa_stream * p, void * userdata)
@@ -257,7 +299,7 @@ static void pulseaudio_overflow_cb(pa_stream * p, void * userdata)
  DEBUG_WARN("Overflow");
 }

-static void pulseaudio_setup(int channels, int sampleRate,
+static bool pulseaudio_setup(int channels, int sampleRate,
    int requestedPeriodFrames, int * maxPeriodFrames, int * startFrames,
    LG_AudioPullFn pullFn)
 {
@@ -265,7 +307,7 @@ static void pulseaudio_setup(int channels, int sampleRate,
  {
    *maxPeriodFrames = pa.sinkMaxPeriodFrames;
    *startFrames     = pa.sinkStartFrames;
-    return;
+    return true;
  }

  pa_sample_spec spec = {
@@ -289,30 +331,76 @@ static void pulseaudio_setup(int channels, int sampleRate,

  pa.sinkChannels   = channels;
  pa.sinkSampleRate = sampleRate;
+  pa.sinkStride     = stride;
+  pa.sinkPullFn     = pullFn;
+  pa.sinkCorked     = true;
+  pa.sinkStarting   = false;

  pa.sink = pa_stream_new(pa.context, "Looking Glass", &spec, NULL);
+  if (!pa.sink)
+  {
+    DEBUG_ERROR("Failed to create PulseAudio stream: %s",
+        pa_strerror(pa_context_errno(pa.context)));
+    pa_threaded_mainloop_unlock(pa.loop);
+    return false;
+  }
+
  pa_stream_set_state_callback    (pa.sink, pulseaudio_state_cb    , NULL);
  pa_stream_set_write_callback    (pa.sink, pulseaudio_write_cb    , NULL);
  pa_stream_set_underflow_callback(pa.sink, pulseaudio_underflow_cb, NULL);
  pa_stream_set_overflow_callback (pa.sink, pulseaudio_overflow_cb , NULL);

-  pa_stream_connect_playback(pa.sink, NULL, &attribs, PA_STREAM_START_CORKED,
-    NULL, NULL);
+  const pa_stream_flags_t flags =
+    PA_STREAM_START_CORKED |
+    PA_STREAM_ADJUST_LATENCY |
+    PA_STREAM_INTERPOLATE_TIMING |
+    PA_STREAM_AUTO_TIMING_UPDATE;
+  if (pa_stream_connect_playback(
+        pa.sink, NULL, &attribs, flags, NULL, NULL) < 0)
+  {
+    DEBUG_ERROR("Failed to connect PulseAudio stream: %s",
+        pa_strerror(pa_context_errno(pa.context)));
+    pulseaudio_sink_close_nl();
+    pa_threaded_mainloop_unlock(pa.loop);
+    return false;
+  }

-  pa.sinkStride          = stride;
-  pa.sinkPullFn          = pullFn;
-  pa.sinkMaxPeriodFrames = requestedPeriodFrames;
-  pa.sinkCorked          = true;
-  pa.sinkStarting        = false;
+  while (pa_stream_get_state(pa.sink) == PA_STREAM_CREATING)
+    pa_threaded_mainloop_wait(pa.loop);

-  // If something else is, or was recently using a small latency value,
-  // PulseAudio can request way more data at startup than is reasonable
-  pa.sinkStartFrames = requestedPeriodFrames * 4;
+  if (pa_stream_get_state(pa.sink) != PA_STREAM_READY)
+  {
+    DEBUG_ERROR("PulseAudio stream did not become ready: %s",
+        pa_strerror(pa_context_errno(pa.context)));
+    pulseaudio_sink_close_nl();
+    pa_threaded_mainloop_unlock(pa.loop);
+    return false;
+  }

-  *maxPeriodFrames = requestedPeriodFrames;
+  const pa_buffer_attr * actual = pa_stream_get_buffer_attr(pa.sink);
+  const uint64_t minRequestFrames =
+    actual && actual->minreq != UINT32_MAX ?
+      actual->minreq / (uint64_t)stride : requestedPeriodFrames;
+  const uint64_t targetFrames =
+    actual && actual->tlength != UINT32_MAX ?
+      actual->tlength / (uint64_t)stride :
+      (uint64_t)requestedPeriodFrames * 2;
+  const int actualMinRequest =
+    clamp(minRequestFrames, UINT64_C(1), (uint64_t)INT_MAX);
+  const int actualTarget =
+    clamp(max(targetFrames, minRequestFrames),
+        UINT64_C(1), (uint64_t)INT_MAX);
+
+  pa.sinkMaxPeriodFrames = actualMinRequest;
+  pa.sinkStartFrames     = actualTarget;
+
+  *maxPeriodFrames = pa.sinkMaxPeriodFrames;
  *startFrames     = pa.sinkStartFrames;

+  atomic_store_explicit(
+      &pa.sinkPresentationDeadline, 0, memory_order_release);
  pa_threaded_mainloop_unlock(pa.loop);
+  return true;
 }

 static void pulseaudio_start(void)
@@ -327,7 +415,7 @@ static void pulseaudio_start(void)
    pa.sinkStarting = true;
  else
  {
-    pa_stream_cork(pa.sink, 0, NULL, NULL);
+    pulseaudio_unrefOperation(pa_stream_cork(pa.sink, 0, NULL, NULL));
    pa.sinkCorked = false;
  }

@@ -343,9 +431,11 @@ static void pulseaudio_stop(void)
  if (needLock)
    pa_threaded_mainloop_lock(pa.loop);

-  pa_stream_cork(pa.sink, 1, NULL, NULL);
+  pulseaudio_unrefOperation(pa_stream_cork(pa.sink, 1, NULL, NULL));
  pa.sinkCorked   = true;
  pa.sinkStarting = false;
+  atomic_store_explicit(
+      &pa.sinkPresentationDeadline, 0, memory_order_release);

  if (needLock)
    pa_threaded_mainloop_unlock(pa.loop);
@@ -362,7 +452,8 @@ static void pulseaudio_volume(int channels, const uint16_t volume[])
      9.3234e-7 * pow(1.000211902, volume[i]) - 0.000172787);

  pa_threaded_mainloop_lock(pa.loop);
-  pa_context_set_sink_input_volume(pa.context, pa.sinkIndex, &v, NULL, NULL);
+  pulseaudio_unrefOperation(pa_context_set_sink_input_volume(
+      pa.context, pa.sinkIndex, &v, NULL, NULL));
  pa_threaded_mainloop_unlock(pa.loop);
 }

@@ -373,10 +464,21 @@ static void pulseaudio_mute(bool mute)

  pa.sinkMuted = mute;
  pa_threaded_mainloop_lock(pa.loop);
-  pa_context_set_sink_input_mute(pa.context, pa.sinkIndex, mute, NULL, NULL);
+  pulseaudio_unrefOperation(pa_context_set_sink_input_mute(
+      pa.context, pa.sinkIndex, mute, NULL, NULL));
  pa_threaded_mainloop_unlock(pa.loop);
 }

+static uint64_t pulseaudio_latency(void)
+{
+  const int64_t deadline = atomic_load_explicit(
+      &pa.sinkPresentationDeadline, memory_order_acquire);
+  if (deadline <= 0)
+    return 0;
+
+  return max(INT64_C(0), deadline - (int64_t)nanotime()) / 1000;
+}
+
 struct LG_AudioDevOps LGAD_PulseAudio =
 {
  .name   = "PulseAudio",
@@ -388,6 +490,7 @@ struct LG_AudioDevOps LGAD_PulseAudio =
    .start  = pulseaudio_start,
    .stop   = pulseaudio_stop,
    .volume = pulseaudio_volume,
-    .mute   = pulseaudio_mute
+    .mute   = pulseaudio_mute,
+    .latency = pulseaudio_latency
  }
 };