audio/out/pull: remove race conditions

There were subtle and minor race conditions in the pull.c code, and AOs
using it (jack, portaudio, sdl, wasapi). Attempt to remove these.

There was at least a race condition in the ao_reset() implementation:
mp_ring_reset() was called concurrently to the audio callback. While the
ringbuffer uses atomics to allow concurrent access, the reset function
wasn't concurrency-safe (and can't easily be made to).

Fix this by stopping the audio callback before doing a reset. After
that, we can do anything without needing synchronization. The callback
is resumed when resuming playback at a later point.

Don't call driver->pause, and make driver->resume and driver->reset
start/stop the audio callback. In the initial state, the audio callback
must be disabled.

JackAudio of course is different. Maybe there is no way to suspend the
audio callback without "disconnecting" it (what jack_deactivate() would
do), so I'm not trying my luck, and implemented a really bad hack doing
active waiting until we get the audio callback into a state where it
won't interfere. Once the callback goes from AO_STATE_WAIT to NONE, we
can be sure that the callback doesn't access the ringbuffer or anything
else anymore. Since both sched_yield() and pthread_yield() apparently
are not always available, use mp_sleep_us(1) to avoid burning CPU during
active waiting.

The ao_jack.c change also removes a race condition: apparently we didn't
initialize _all_ ao fields before starting the audio callback.

In ao_wasapi.c, I'm not sure whether reset really waits for the audio
callback to return. Kovensky says it's not guaranteed, so disable the
reset callback - for now the behavior of ao_wasapi.c is like with
ao_jack.c, and active waiting is used to deal with the audio callback.

6 files changed, 68 insertions, 57 deletions

diff --git a/audio/out/ao_jack.c b/audio/out/ao_jack.c
index 6e9b652f2d..727a5f5050 100644
--- a/audio/out/ao_jack.c
+++ b/audio/out/ao_jack.c
@@ -51,6 +51,8 @@ struct priv {
 
     int num_ports;
     jack_port_t *ports[MP_NUM_CHANNELS];
+
+    int activated;
 };
 
 static int process(jack_nframes_t nframes, void *arg)
@@ -134,6 +136,20 @@ err_port_register:
     return -1;
 }
 
+static void resume(struct ao *ao)
+{
+    struct priv *p = ao->priv;
+    if (!p->activated) {
+        p->activated = true;
+
+        if (jack_activate(p->client))
+            MP_FATAL(ao, "activate failed\n");
+
+        if (p->connect)
+            connect_to_outports(ao);
+    }
+}
+
 static int init(struct ao *ao)
 {
     struct priv *p = ao->priv;
@@ -173,17 +189,8 @@ static int init(struct ao *ao)
 
     jack_set_process_callback(p->client, process, ao);
 
-    if (jack_activate(p->client)) {
-        MP_FATAL(ao, "activate failed\n");
-        goto err_activate;
-    }
-
     ao->samplerate = jack_get_sample_rate(p->client);
 
-    if (p->connect)
-        if (connect_to_outports(ao))
-            goto err_connect;
-
     jack_latency_range_t jack_latency_range;
     jack_port_get_latency_range(p->ports[0], JackPlaybackLatency,
                                 &jack_latency_range);
@@ -196,9 +203,6 @@ static int init(struct ao *ao)
     return 0;
 
 err_chmap_sel_get_def:
-err_connect:
-    jack_deactivate(p->client);
-err_activate:
 err_create_ports:
     jack_client_close(p->client);
 err_client_open:
@@ -221,6 +225,7 @@ const struct ao_driver audio_out_jack = {
     .name        = "jack",
     .init      = init,
     .uninit    = uninit,
+    .resume    = resume,
     .priv_size = sizeof(struct priv),
     .priv_defaults = &(const struct priv) {
         .cfg_client_name = "mpv",
diff --git a/audio/out/ao_portaudio.c b/audio/out/ao_portaudio.c
index 52c67d2a3c..61af83f8de 100644
--- a/audio/out/ao_portaudio.c
+++ b/audio/out/ao_portaudio.c
@@ -245,7 +245,6 @@ const struct ao_driver audio_out_portaudio = {
     .init      = init,
     .uninit    = uninit,
     .reset     = reset,
-    .pause     = reset,
     .resume    = resume,
     .priv_size = sizeof(struct priv),
     .options = (const struct m_option[]) {
diff --git a/audio/out/ao_sdl.c b/audio/out/ao_sdl.c
index 95f1a0d24d..9af5db4021 100644
--- a/audio/out/ao_sdl.c
+++ b/audio/out/ao_sdl.c
@@ -194,7 +194,7 @@ static int init(struct ao *ao)
     return 1;
 }
 
-static void pause(struct ao *ao)
+static void reset(struct ao *ao)
 {
     struct priv *priv = ao->priv;
     if (!priv->paused)
@@ -217,7 +217,7 @@ const struct ao_driver audio_out_sdl = {
     .name      = "sdl",
     .init      = init,
     .uninit    = uninit,
-    .pause     = pause,
+    .reset     = reset,
     .resume    = resume,
     .priv_size = sizeof(struct priv),
     .priv_defaults = &(const struct priv) {
diff --git a/audio/out/ao_wasapi.c b/audio/out/ao_wasapi.c
index 431b24c3a1..aef20fe2c4 100644
--- a/audio/out/ao_wasapi.c
+++ b/audio/out/ao_wasapi.c
@@ -310,7 +310,7 @@ static int control(struct ao *ao, enum aocontrol cmd, void *arg)
 }
 
 
-static void audio_pause(struct ao *ao)
+static void audio_reset(struct ao *ao)
 {
     struct wasapi_state *state = (struct wasapi_state *)ao->priv;
 
@@ -328,11 +328,6 @@ static void audio_resume(struct ao *ao)
     IAudioClient_Start(state->pAudioClientProxy);
 }
 
-static void audio_reset(struct ao *ao)
-{
-    audio_pause(ao);
-}
-
 #define OPT_BASE_STRUCT struct wasapi_state
 
 const struct ao_driver audio_out_wasapi = {
@@ -341,9 +336,8 @@ const struct ao_driver audio_out_wasapi = {
     .init      = init,
     .uninit    = uninit,
     .control   = control,
-    .pause     = audio_pause,
+    //.reset     = audio_reset, <- doesn't wait for audio callback to return
     .resume    = audio_resume,
-    .reset     = audio_reset,
     .priv_size = sizeof(wasapi_state),
     .options   = (const struct m_option[]) {
         OPT_FLAG("exclusive", opt_exclusive, 0),
diff --git a/audio/out/internal.h b/audio/out/internal.h
index cc3abb14f5..2f2fbd307b 100644
--- a/audio/out/internal.h
+++ b/audio/out/internal.h
@@ -86,6 +86,9 @@ extern const struct ao_driver ao_api_pull;
  *          get_delay
  *          pause
  *          resume
+ *     Optional:
+ *          control
+ *          drain
  *  b) ->play must be NULL. The driver can start the audio API in init(). The
  *     audio API in turn will start a thread and call a callback provided by the
  *     driver. That callback calls ao_read_data() to get audio data. Most
@@ -94,6 +97,10 @@ extern const struct ao_driver ao_api_pull;
  *     Mandatory:
  *          init
  *          uninit
+ *          resume      (starts the audio callback)
+ *     Also, the following optional callbacks can be provided:
+ *          reset       (stops the audio callback, resume() restarts it)
+ *          control
  */
 struct ao_driver {
     // If true, use with encoding only.
diff --git a/audio/out/pull.c b/audio/out/pull.c
index 71f14f38dc..7902c9afb6 100644
--- a/audio/out/pull.c
+++ b/audio/out/pull.c
@@ -33,11 +33,16 @@
 #include "compat/atomics.h"
 #include "misc/ring.h"
 
+/*
+ * Note: there is some stupid stuff in this file in order to avoid mutexes.
+ * This requirement is dictated by several audio APIs, at least jackaudio.
+ */
+
 enum {
     AO_STATE_NONE,  // idle (e.g. before playback started, or after playback
                     // finished, but device is open)
+    AO_STATE_WAIT,  // wait for callback to go into AO_STATE_NONE state
     AO_STATE_PLAY,  // play the buffer
-    AO_STATE_PAUSE, // pause playback
 };
 
 struct ao_pull_state {
@@ -47,9 +52,6 @@ struct ao_pull_state {
     // AO_STATE_*
     atomic_int state;
 
-    // Whether buffers[] can be accessed.
-    atomic_bool ready;
-
     // Device delay of the last written sample, in realtime.
     atomic_llong end_time_us;
 };
@@ -78,10 +80,8 @@ static int play(struct ao *ao, void **data, int samples, int flags)
         assert(r == write_bytes);
     }
     if (atomic_load(&p->state) != AO_STATE_PLAY) {
-        atomic_store(&p->end_time_us, 0);
         atomic_store(&p->state, AO_STATE_PLAY);
-        if (ao->driver->resume)
-            ao->driver->resume(ao);
+        ao->driver->resume(ao);
     }
 
     return write_samples;
@@ -101,37 +101,37 @@ int ao_read_data(struct ao *ao, void **data, int samples, int64_t out_time_us)
 
     struct ao_pull_state *p = ao->api_priv;
     int full_bytes = samples * ao->sstride;
+    int state = atomic_load(&p->state);
+    int bytes = 0;
 
-    if (!atomic_load(&p->ready)) {
-        for (int n = 0; n < ao->num_planes; n++)
-            af_fill_silence(data[n], full_bytes, ao->format);
-        return 0;
+    if (state != AO_STATE_PLAY) {
+        if (state == AO_STATE_WAIT)
+            atomic_store(&p->state, AO_STATE_NONE);
+        goto end;
     }
 
     // Since the writer will write the first plane last, its buffered amount
     // of data is the minimum amount across all planes.
     int buffered_bytes = mp_ring_buffered(p->buffers[0]);
-    int bytes = MPMIN(buffered_bytes, full_bytes);
+    bytes = MPMIN(buffered_bytes, full_bytes);
 
     if (bytes > 0)
         atomic_store(&p->end_time_us, out_time_us);
 
-    if (atomic_load(&p->state) == AO_STATE_PAUSE)
-        bytes = 0;
-
     for (int n = 0; n < ao->num_planes; n++) {
         int r = mp_ring_read(p->buffers[n], data[n], bytes);
-        assert(r == bytes);
-        // pad with silence (underflow/paused/eof)
-        int silence = full_bytes - bytes;
-        if (silence)
-            af_fill_silence((char *)data[n] + bytes, silence, ao->format);
+        bytes = MPMIN(bytes, r);
     }
 
     // Half of the buffer played -> request more.
     if (buffered_bytes - bytes <= mp_ring_size(p->buffers[0]) / 2)
         mp_input_wakeup_nolock(ao->input_ctx);
 
+end:
+    // pad with silence (underflow/paused/eof)
+    for (int n = 0; n < ao->num_planes; n++)
+        af_fill_silence(data[n], full_bytes - bytes, ao->format);
+
     return bytes / ao->sstride;
 }
 
@@ -160,32 +160,37 @@ static float get_delay(struct ao *ao)
 static void reset(struct ao *ao)
 {
     struct ao_pull_state *p = ao->api_priv;
-    if (ao->driver->reset)
-        ao->driver->reset(ao);
-    // Not like this is race-condition free. It will work if ->reset
-    // stops the audio callback, though.
-    atomic_store(&p->ready, false);
-    atomic_store(&p->state, AO_STATE_NONE);
+    if (ao->driver->reset) {
+        ao->driver->reset(ao); // assumes the audio callback thread is stopped
+        atomic_store(&p->state, AO_STATE_NONE);
+    } else {
+        // The thread keeps running. Wait until the audio callback gets into
+        // a defined state where it won't touch the ringbuffer. We must do
+        // this, because emptying the ringbuffer is not an atomic operation.
+        if (atomic_load(&p->state) != AO_STATE_NONE) {
+            atomic_store(&p->state, AO_STATE_WAIT);
+            while (atomic_load(&p->state) != AO_STATE_NONE)
+                mp_sleep_us(1);
+        }
+    }
     for (int n = 0; n < ao->num_planes; n++)
         mp_ring_reset(p->buffers[n]);
     atomic_store(&p->end_time_us, 0);
-    atomic_store(&p->ready, true);
 }
 
 static void pause(struct ao *ao)
 {
     struct ao_pull_state *p = ao->api_priv;
-    if (ao->driver->pause)
-        ao->driver->pause(ao);
-    atomic_store(&p->state, AO_STATE_PAUSE);
+    if (ao->driver->reset)
+        ao->driver->reset(ao);
+    atomic_store(&p->state, AO_STATE_NONE);
 }
 
 static void resume(struct ao *ao)
 {
     struct ao_pull_state *p = ao->api_priv;
     atomic_store(&p->state, AO_STATE_PLAY);
-    if (ao->driver->resume)
-        ao->driver->resume(ao);
+    ao->driver->resume(ao);
 }
 
 static void uninit(struct ao *ao)
@@ -197,8 +202,9 @@ static int init(struct ao *ao)
 {
     struct ao_pull_state *p = ao->api_priv;
     for (int n = 0; n < ao->num_planes; n++)
-            p->buffers[n] = mp_ring_new(ao, ao->buffer * ao->sstride);
-    atomic_store(&p->ready, true);
+        p->buffers[n] = mp_ring_new(ao, ao->buffer * ao->sstride);
+    atomic_store(&p->state, AO_STATE_NONE);
+    assert(ao->driver->resume);
     return 0;
 }