audio: refactor how data is passed to AO

This replaces the two buffers (ao_chain.ao_buffer in the core, and
buffer_state.buffers in the AO) with a single queue. Instead of having a
byte based buffer, the queue is simply a list of audio frames, as output
by the decoder. This should make dataflow simpler and reduce copying.

It also attempts to simplify fill_audio_out_buffers(), the function I
always hated most, because it's full of subtle and buggy logic.

Unfortunately, I got assaulted by corner cases, dumb features (attempt
at seamless looping, really?), and other crap, so it got pretty
complicated again. fill_audio_out_buffers() is still full of subtle and
buggy logic. Maybe it got worse. On the other hand, maybe there really
is some progress. Who knows.

Originally, the data flow parts was meant to be in f_output_chain, but
due to tricky interactions with the playloop code, it's now in the dummy
filter in audio.c.

At least this improves the way the audio PTS is passed to the encoder in
encoding mode. Now it attempts to pass frames directly, along with the
pts, which should minimize timestamp problems. But to be honest, encoder
mode is one big kludge that shouldn't exist in this way.

This commit should be considered pre-alpha code. There are lots of bugs
still hiding.

4 files changed, 304 insertions, 371 deletions

diff --git a/audio/out/ao.h b/audio/out/ao.h
index 0207e5a8bf..cfcb39790f 100644
--- a/audio/out/ao.h
+++ b/audio/out/ao.h
@@ -45,7 +45,6 @@ enum {
     AO_EVENT_RELOAD = 1,
     AO_EVENT_HOTPLUG = 2,
     AO_EVENT_INITIAL_UNBLOCK = 4,
-    AO_EVENT_UNDERRUN = 8,
 };
 
 enum {
@@ -98,16 +97,16 @@ void ao_get_format(struct ao *ao,
 const char *ao_get_name(struct ao *ao);
 const char *ao_get_description(struct ao *ao);
 bool ao_untimed(struct ao *ao);
-int ao_play(struct ao *ao, void **data, int samples, int flags);
 int ao_control(struct ao *ao, enum aocontrol cmd, void *arg);
 void ao_set_gain(struct ao *ao, float gain);
 double ao_get_delay(struct ao *ao);
-int ao_get_space(struct ao *ao);
 void ao_reset(struct ao *ao);
-void ao_pause(struct ao *ao);
-void ao_resume(struct ao *ao);
+void ao_start(struct ao *ao);
+void ao_set_paused(struct ao *ao, bool paused);
 void ao_drain(struct ao *ao);
-bool ao_eof_reached(struct ao *ao);
+bool ao_is_playing(struct ao *ao);
+struct mp_async_queue;
+struct mp_async_queue *ao_get_queue(struct ao *ao);
 int ao_query_and_reset_events(struct ao *ao, int events);
 int ao_add_events(struct ao *ao, int events);
 void ao_unblock(struct ao *ao);
diff --git a/audio/out/ao_lavc.c b/audio/out/ao_lavc.c
index c524e9e02d..049f8df2cf 100644
--- a/audio/out/ao_lavc.c
+++ b/audio/out/ao_lavc.c
@@ -30,8 +30,11 @@
 #include "config.h"
 #include "options/options.h"
 #include "common/common.h"
+#include "audio/aframe.h"
 #include "audio/format.h"
 #include "audio/fmt-conversion.h"
+#include "filters/filter_internal.h"
+#include "filters/f_utils.h"
 #include "mpv_talloc.h"
 #include "ao.h"
 #include "internal.h"
@@ -44,20 +47,19 @@ struct priv {
 
     int pcmhack;
     int aframesize;
-    int aframecount;
-    int64_t savepts;
     int framecount;
     int64_t lastpts;
     int sample_size;
-    const void *sample_padding;
     double expected_next_pts;
+    struct mp_filter *filter_root;
+    struct mp_filter *fix_frame_size;
 
     AVRational worst_time_base;
 
     bool shutdown;
 };
 
-static void encode(struct ao *ao, double apts, void **data);
+static void read_frames(struct ao *ao);
 
 static bool supports_format(const AVCodec *codec, int format)
 {
@@ -151,7 +153,6 @@ static int init(struct ao *ao)
     // but at least one!
     ac->framecount = MPMAX(ac->framecount, 1);
 
-    ac->savepts = AV_NOPTS_VALUE;
     ac->lastpts = AV_NOPTS_VALUE;
 
     ao->untimed = true;
@@ -159,8 +160,10 @@ static int init(struct ao *ao)
     ao->device_buffer = ac->aframesize * ac->framecount;
     ao->period_size = ao->device_buffer;
 
-    if (ao->channels.num > AV_NUM_DATA_POINTERS)
-        goto fail;
+    ac->filter_root = mp_filter_create_root(ao->global);
+    ac->fix_frame_size = mp_fixed_aframe_size_create(ac->filter_root,
+                                                     ac->aframesize, true);
+    MP_HANDLE_OOM(ac->fix_frame_size);
 
     return 0;
 
@@ -185,103 +188,81 @@ static void uninit(struct ao *ao)
         pthread_mutex_unlock(&ectx->lock);
 
         outpts += encoder_get_offset(ac->enc);
-        encode(ao, outpts, NULL);
+
+        if (!mp_pin_in_write(ac->fix_frame_size->pins[0], MP_EOF_FRAME))
+            MP_WARN(ao, "could not flush last frame\n");
+        read_frames(ao);
+        encoder_encode(ac->enc, NULL);
     }
+
+    talloc_free(ac->filter_root);
 }
 
 // must get exactly ac->aframesize amount of data
-static void encode(struct ao *ao, double apts, void **data)
+static void encode(struct ao *ao, struct mp_aframe *af)
 {
     struct priv *ac = ao->priv;
-    struct encode_lavc_context *ectx = ao->encode_lavc_ctx;
     AVCodecContext *encoder = ac->enc->encoder;
-    double realapts = ac->aframecount * (double) ac->aframesize /
-                      ao->samplerate;
+    double outpts = mp_aframe_get_pts(af);
 
-    ac->aframecount++;
+    AVFrame *frame = mp_aframe_to_avframe(af);
+    if (!frame)
+        abort();
 
-    pthread_mutex_lock(&ectx->lock);
-    if (data)
-        ectx->audio_pts_offset = realapts - apts;
-    pthread_mutex_unlock(&ectx->lock);
+    frame->pts = rint(outpts * av_q2d(av_inv_q(encoder->time_base)));
 
-    if(data) {
-        AVFrame *frame = av_frame_alloc();
-        frame->format = af_to_avformat(ao->format);
-        frame->nb_samples = ac->aframesize;
-        frame->channels = encoder->channels;
-        frame->channel_layout = encoder->channel_layout;
-
-        size_t num_planes = af_fmt_is_planar(ao->format) ? ao->channels.num : 1;
-        assert(num_planes <= AV_NUM_DATA_POINTERS);
-        for (int n = 0; n < num_planes; n++)
-            frame->extended_data[n] = data[n];
-
-        frame->linesize[0] = frame->nb_samples * ao->sstride;
-
-        frame->pts = rint(apts * av_q2d(av_inv_q(encoder->time_base)));
-
-        int64_t frame_pts = av_rescale_q(frame->pts, encoder->time_base,
-                                         ac->worst_time_base);
-        while (ac->lastpts != AV_NOPTS_VALUE && frame_pts <= ac->lastpts) {
-            // whatever the fuck this code does?
-            MP_WARN(ao, "audio frame pts went backwards (%d <- %d), autofixed\n",
-                    (int)frame->pts, (int)ac->lastpts);
-            frame_pts = ac->lastpts + 1;
-            ac->lastpts = frame_pts;
-            frame->pts = av_rescale_q(frame_pts, ac->worst_time_base,
-                                      encoder->time_base);
-            frame_pts = av_rescale_q(frame->pts, encoder->time_base,
+    int64_t frame_pts = av_rescale_q(frame->pts, encoder->time_base,
                                      ac->worst_time_base);
-        }
+    if (ac->lastpts != AV_NOPTS_VALUE && frame_pts <= ac->lastpts) {
+        // whatever the fuck this code does?
+        MP_WARN(ao, "audio frame pts went backwards (%d <- %d), autofixed\n",
+                (int)frame->pts, (int)ac->lastpts);
+        frame_pts = ac->lastpts + 1;
         ac->lastpts = frame_pts;
+        frame->pts = av_rescale_q(frame_pts, ac->worst_time_base,
+                                  encoder->time_base);
+        frame_pts = av_rescale_q(frame->pts, encoder->time_base,
+                                 ac->worst_time_base);
+    }
+    ac->lastpts = frame_pts;
 
-        frame->quality = encoder->global_quality;
-        encoder_encode(ac->enc, frame);
-        av_frame_free(&frame);
-    } else {
-        encoder_encode(ac->enc, NULL);
+    frame->quality = encoder->global_quality;
+    encoder_encode(ac->enc, frame);
+    av_frame_free(&frame);
+}
+
+static void read_frames(struct ao *ao)
+{
+    struct priv *ac = ao->priv;
+
+    while (1) {
+        struct mp_frame fr = mp_pin_out_read(ac->fix_frame_size->pins[1]);
+        if (!fr.type)
+            break;
+        if (fr.type != MP_FRAME_AUDIO)
+            continue;
+        struct mp_aframe *af = fr.data;
+        encode(ao, af);
+        mp_frame_unref(&fr);
     }
 }
 
-// Note: currently relies on samples aligned to period sizes - will not work
-//       in the future.
 static bool audio_write(struct ao *ao, void **data, int samples)
 {
     struct priv *ac = ao->priv;
-    struct encoder_context *enc = ac->enc;
     struct encode_lavc_context *ectx = ao->encode_lavc_ctx;
-    int bufpos = 0;
+
+    // See ao_driver.write_frames.
+    struct mp_aframe *af = mp_aframe_new_ref(*(struct mp_aframe **)data);
+
     double nextpts;
-    int orig_samples = samples;
+    double pts = mp_aframe_get_pts(af);
+    double outpts = pts;
 
     // for ectx PTS fields
     pthread_mutex_lock(&ectx->lock);
 
-    double pts = ectx->last_audio_in_pts;
-    pts += ectx->samples_since_last_pts / (double)ao->samplerate;
-
-    size_t num_planes = af_fmt_is_planar(ao->format) ? ao->channels.num : 1;
-
-    void *tempdata = NULL;
-    void *padded[MP_NUM_CHANNELS];
-
-    if (samples % ac->aframesize) {
-       tempdata = talloc_new(NULL);
-       size_t bytelen = samples * ao->sstride;
-       size_t extralen = (ac->aframesize - 1) * ao->sstride;
-       for (int n = 0; n < num_planes; n++) {
-           padded[n] = talloc_size(tempdata, bytelen + extralen);
-           memcpy(padded[n], data[n], bytelen);
-           af_fill_silence((char *)padded[n] + bytelen, extralen, ao->format);
-       }
-       data = padded;
-       samples = (bytelen + extralen) / ao->sstride;
-       MP_VERBOSE(ao, "padding final frame with silence\n");
-    }
-
-    double outpts = pts;
-    if (!enc->options->rawts) {
+    if (!ectx->options->rawts) {
         // Fix and apply the discontinuity pts offset.
         nextpts = pts;
         if (ectx->discontinuity_pts_offset == MP_NOPTS_VALUE) {
@@ -298,44 +279,36 @@ static bool audio_write(struct ao *ao, void **data, int samples)
         outpts = pts + ectx->discontinuity_pts_offset;
     }
 
-    pthread_mutex_unlock(&ectx->lock);
-
     // Shift pts by the pts offset first.
-    outpts += encoder_get_offset(enc);
-
-    while (samples - bufpos >= ac->aframesize) {
-        void *start[MP_NUM_CHANNELS] = {0};
-        for (int n = 0; n < num_planes; n++)
-            start[n] = (char *)data[n] + bufpos * ao->sstride;
-        encode(ao, outpts + bufpos / (double) ao->samplerate, start);
-        bufpos += ac->aframesize;
-    }
+    outpts += encoder_get_offset(ac->enc);
 
     // Calculate expected pts of next audio frame (input side).
-    ac->expected_next_pts = pts + bufpos / (double) ao->samplerate;
-
-    pthread_mutex_lock(&ectx->lock);
+    ac->expected_next_pts = pts + mp_aframe_get_size(af) / (double) ao->samplerate;
 
     // Set next allowed input pts value (input side).
-    if (!enc->options->rawts) {
+    if (!ectx->options->rawts) {
         nextpts = ac->expected_next_pts + ectx->discontinuity_pts_offset;
         if (nextpts > ectx->next_in_pts)
             ectx->next_in_pts = nextpts;
     }
 
-    talloc_free(tempdata);
+    pthread_mutex_unlock(&ectx->lock);
 
-    int taken = MPMIN(bufpos, orig_samples);
-    ectx->samples_since_last_pts += taken;
+    mp_aframe_set_pts(af, outpts);
 
-    pthread_mutex_unlock(&ectx->lock);
+    // Can't push in frame if it doesn't want it output one.
+    mp_pin_out_request_data(ac->fix_frame_size->pins[1]);
 
+    if (!mp_pin_in_write(ac->fix_frame_size->pins[0],
+                         MAKE_FRAME(MP_FRAME_AUDIO, af)))
+        return false; // shouldn't happen™
+    read_frames(ao);
     return true;
 }
 
 static void get_state(struct ao *ao, struct mp_pcm_state *state)
 {
-    state->free_samples = ao->device_buffer;
+    state->free_samples = 1;
     state->queued_samples = 0;
     state->delay = 0;
 }
@@ -359,6 +332,7 @@ const struct ao_driver audio_out_lavc = {
     .description = "audio encoding using libavcodec",
     .name      = "lavc",
     .initially_blocked = true,
+    .write_frames = true,
     .priv_size = sizeof(struct priv),
     .init      = init,
     .uninit    = uninit,
diff --git a/audio/out/buffer.c b/audio/out/buffer.c
index 2992180854..4c08a26b0b 100644
--- a/audio/out/buffer.c
+++ b/audio/out/buffer.c
@@ -31,13 +31,11 @@
 #include "common/msg.h"
 #include "common/common.h"
 
-#include "input/input.h"
+#include "filters/f_async_queue.h"
+#include "filters/filter_internal.h"
 
-#include "osdep/io.h"
 #include "osdep/timer.h"
 #include "osdep/threads.h"
-#include "osdep/atomic.h"
-#include "misc/ring.h"
 
 struct buffer_state {
     // Buffer and AO
@@ -51,27 +49,26 @@ struct buffer_state {
     // Access from AO driver's thread only.
     char *convert_buffer;
 
-    // --- protected by lock
+    // Immutable.
+    struct mp_async_queue *queue;
 
-    struct mp_ring *buffers[MP_NUM_CHANNELS];
+    // --- protected by lock
 
+    struct mp_filter *filter_root;
+    struct mp_filter *input;    // connected to queue
+    struct mp_aframe *pending;  // last, not fully consumed output
 
     bool streaming;             // AO streaming active
     bool playing;               // logically playing audio from buffer
-    bool paused;                // logically paused; implies playing=true
-    bool final_chunk;           // if buffer contains EOF
+    bool paused;                // logically paused
 
     int64_t end_time_us;        // absolute output time of last played sample
-    int64_t underflow;          // number of samples missing since last check
 
     bool initial_unblocked;
 
     // "Push" AOs only (AOs with driver->write).
-    bool still_playing;
     bool hw_paused;             // driver->set_pause() was used successfully
     bool recover_pause;         // non-hw_paused: needs to recover delay
-    bool draining;
-    bool ao_wait_low_buffer;
     struct mp_pcm_state prepause_state;
     pthread_t thread;           // thread shoveling data to AO
     bool thread_valid;          // thread is running
@@ -98,7 +95,7 @@ static void get_dev_state(struct ao *ao, struct mp_pcm_state *state)
 {
     struct buffer_state *p = ao->buffer_state;
 
-    if (p->paused) {
+    if (p->paused && p->playing) {
         *state = p->prepause_state;
         return;
     }
@@ -111,84 +108,66 @@ static void get_dev_state(struct ao *ao, struct mp_pcm_state *state)
     ao->driver->get_state(ao, state);
 }
 
-static int unlocked_get_space(struct ao *ao)
+struct mp_async_queue *ao_get_queue(struct ao *ao)
 {
     struct buffer_state *p = ao->buffer_state;
-
-    int space = mp_ring_available(p->buffers[0]) / ao->sstride;
-
-    // The following code attempts to keep the total buffered audio at
-    // ao->buffer in order to improve latency.
-    if (ao->driver->write) {
-        struct mp_pcm_state state;
-        get_dev_state(ao, &state);
-        int align = af_format_sample_alignment(ao->format);
-        int device_space = MPMAX(state.free_samples, 0);
-        int device_buffered = ao->device_buffer - device_space;
-        int soft_buffered = mp_ring_size(p->buffers[0]) / ao->sstride - space;
-        // The extra margin helps avoiding too many wakeups if the AO is fully
-        // byte based and doesn't do proper chunked processing.
-        int min_buffer = ao->buffer + 64;
-        int missing = min_buffer - device_buffered - soft_buffered;
-        missing = (missing + align - 1) / align * align;
-        // But always keep the device's buffer filled as much as we can.
-        int device_missing = device_space - soft_buffered;
-        missing = MPMAX(missing, device_missing);
-        space = MPMIN(space, missing);
-        space = MPMAX(0, space);
-    }
-
-    return space;
+    return p->queue;
 }
 
-int ao_get_space(struct ao *ao)
+// Special behavior with data==NULL: caller uses p->pending.
+static int read_buffer(struct ao *ao, void **data, int samples, bool *eof)
 {
     struct buffer_state *p = ao->buffer_state;
-    pthread_mutex_lock(&p->lock);
-    int space = unlocked_get_space(ao);
-    pthread_mutex_unlock(&p->lock);
-    return space;
-}
-
-int ao_play(struct ao *ao, void **data, int samples, int flags)
-{
-    struct buffer_state *p = ao->buffer_state;
-
-    pthread_mutex_lock(&p->lock);
-
-    int write_samples = mp_ring_available(p->buffers[0]) / ao->sstride;
-    write_samples = MPMIN(write_samples, samples);
-
-    int write_bytes = write_samples * ao->sstride;
-    for (int n = 0; n < ao->num_planes; n++) {
-        int r = mp_ring_write(p->buffers[n], data[n], write_bytes);
-        assert(r == write_bytes);
-    }
-
-    p->paused = false;
-    p->final_chunk = write_samples == samples && (flags & PLAYER_FINAL_CHUNK);
-
-    if (p->underflow)
-        MP_DBG(ao, "Audio underrun by %lld samples.\n", (long long)p->underflow);
-    p->underflow = 0;
+    int pos = 0;
+    *eof = false;
+
+    while (p->playing && !p->paused && pos < samples) {
+        if (!p->pending || !mp_aframe_get_size(p->pending)) {
+            TA_FREEP(&p->pending);
+            struct mp_frame frame = mp_pin_out_read(p->input->pins[0]);
+            if (!frame.type)
+                break; // we can't/don't want to block
+            if (frame.type != MP_FRAME_AUDIO) {
+                if (frame.type == MP_FRAME_EOF)
+                    *eof = true;
+                mp_frame_unref(&frame);
+                continue;
+            }
+            p->pending = frame.data;
+        }
 
-    if (write_samples) {
-        p->playing = true;
-        p->still_playing = true;
-        p->draining = false;
+        if (!data)
+            break;
 
-        if (!ao->driver->write && !p->streaming) {
-            p->streaming = true;
-            ao->driver->start(ao);
+        int copy = mp_aframe_get_size(p->pending);
+        uint8_t **fdata = mp_aframe_get_data_ro(p->pending);
+        copy = MPMIN(copy, samples - pos);
+        for (int n = 0; n < ao->num_planes; n++) {
+            memcpy((char *)data[n] + pos * ao->sstride,
+                   fdata[n], copy * ao->sstride);
         }
+        mp_aframe_skip_samples(p->pending, copy);
+        pos += copy;
+        *eof = false;
+    }
 
+    if (!data) {
+        if (!p->pending)
+            return 0;
+        void **pd = (void *)mp_aframe_get_data_rw(p->pending);
+        if (pd)
+            ao_post_process_data(ao, pd, mp_aframe_get_size(p->pending));
+        return 1;
     }
-    pthread_mutex_unlock(&p->lock);
 
-    if (write_samples)
-        ao_wakeup_playthread(ao);
+    // pad with silence (underflow/paused/eof)
+    for (int n = 0; n < ao->num_planes; n++) {
+        af_fill_silence((char *)data[n] + pos, (samples - pos) * ao->sstride,
+                        ao->format);
+    }
 
-    return write_samples;
+    ao_post_process_data(ao, data, pos);
+    return pos;
 }
 
 // Read the given amount of samples in the user-provided data buffer. Returns
@@ -202,47 +181,24 @@ int ao_play(struct ao *ao, void **data, int samples, int flags)
 int ao_read_data(struct ao *ao, void **data, int samples, int64_t out_time_us)
 {
     struct buffer_state *p = ao->buffer_state;
-    int full_bytes = samples * ao->sstride;
-    bool need_wakeup = false;
-    int bytes = 0;
+    assert(!ao->driver->write);
 
     pthread_mutex_lock(&p->lock);
 
-    if (!p->playing || p->paused)
-        goto end;
-
-    int buffered_bytes = mp_ring_buffered(p->buffers[0]);
-    bytes = MPMIN(buffered_bytes, full_bytes);
+    int pos = read_buffer(ao, data, samples, &(bool){0});
 
-    if (full_bytes > bytes && !p->final_chunk) {
-        p->underflow += (full_bytes - bytes) / ao->sstride;
-        ao_add_events(ao, AO_EVENT_UNDERRUN);
-    }
-
-    if (bytes > 0)
+    if (pos > 0)
         p->end_time_us = out_time_us;
 
-    for (int n = 0; n < ao->num_planes; n++)
-        mp_ring_read(p->buffers[n], data[n], bytes);
-
-    // Half of the buffer played -> request more.
-    if (!ao->driver->write)
-        need_wakeup = buffered_bytes - bytes <= mp_ring_size(p->buffers[0]) / 2;
-
-end:
+    if (pos < samples && p->playing && !p->paused) {
+        p->playing = false;
+        // For ao_drain().
+        pthread_cond_broadcast(&p->wakeup);
+    }
 
     pthread_mutex_unlock(&p->lock);
 
-    if (need_wakeup)
-        ao->wakeup_cb(ao->wakeup_ctx);
-
-    // pad with silence (underflow/paused/eof)
-    for (int n = 0; n < ao->num_planes; n++)
-        af_fill_silence((char *)data[n] + bytes, full_bytes - bytes, ao->format);
-
-    ao_post_process_data(ao, data, samples);
-
-    return bytes / ao->sstride;
+    return pos;
 }
 
 // Same as ao_read_data(), but convert data according to *fmt.
@@ -300,11 +256,13 @@ int ao_control(struct ao *ao, enum aocontrol cmd, void *arg)
     return r;
 }
 
-static double unlocked_get_delay(struct ao *ao)
+double ao_get_delay(struct ao *ao)
 {
     struct buffer_state *p = ao->buffer_state;
-    double driver_delay = 0;
 
+    pthread_mutex_lock(&p->lock);
+
+    double driver_delay;
     if (ao->driver->write) {
         struct mp_pcm_state state;
         get_dev_state(ao, &state);
@@ -312,22 +270,18 @@ static double unlocked_get_delay(struct ao *ao)
     } else {
         int64_t end = p->end_time_us;
         int64_t now = mp_time_us();
-        driver_delay += MPMAX(0, (end - now) / (1000.0 * 1000.0));
+        driver_delay = MPMAX(0, (end - now) / (1000.0 * 1000.0));
     }
 
-    return mp_ring_buffered(p->buffers[0]) / (double)ao->bps + driver_delay;
-}
+    int pending = mp_async_queue_get_samples(p->queue);
+    if (p->pending)
+        pending += mp_aframe_get_size(p->pending);
 
-double ao_get_delay(struct ao *ao)
-{
-    struct buffer_state *p = ao->buffer_state;
-
-    pthread_mutex_lock(&p->lock);
-    double delay = unlocked_get_delay(ao);
     pthread_mutex_unlock(&p->lock);
-    return delay;
+    return driver_delay + pending / (double)ao->samplerate;
 }
 
+// Fully stop playback; clear buffers, including queue.
 void ao_reset(struct ao *ao)
 {
     struct buffer_state *p = ao->buffer_state;
@@ -336,8 +290,10 @@ void ao_reset(struct ao *ao)
 
     pthread_mutex_lock(&p->lock);
 
-    for (int n = 0; n < ao->num_planes; n++)
-        mp_ring_reset(p->buffers[n]);
+    TA_FREEP(&p->pending);
+    mp_async_queue_reset(p->queue);
+    mp_filter_reset(p->filter_root);
+    mp_async_queue_resume_reading(p->queue);
 
     if (!ao->stream_silence && ao->driver->reset) {
         if (ao->driver->write) {
@@ -349,17 +305,12 @@ void ao_reset(struct ao *ao)
         }
         p->streaming = false;
     }
-    p->paused = false;
+    wakeup = p->playing;
     p->playing = false;
     p->recover_pause = false;
     p->hw_paused = false;
-    wakeup = p->still_playing || p->draining;
-    p->draining = false;
-    p->still_playing = false;
     p->end_time_us = 0;
 
-    atomic_fetch_and(&ao->events_, ~(unsigned int)AO_EVENT_UNDERRUN);
-
     pthread_mutex_unlock(&p->lock);
 
     if (do_reset)
@@ -369,7 +320,29 @@ void ao_reset(struct ao *ao)
         ao_wakeup_playthread(ao);
 }
 
-void ao_pause(struct ao *ao)
+// Initiate playback. This moves from the stop/underrun state to actually
+// playing (orthogonally taking the paused state into account). Plays all
+// data in the queue, and goes into underrun state if no more data available.
+// No-op if already running.
+void ao_start(struct ao *ao)
+{
+    struct buffer_state *p = ao->buffer_state;
+
+    pthread_mutex_lock(&p->lock);
+
+    p->playing = true;
+
+    if (!ao->driver->write && !p->streaming) {
+        p->streaming = true;
+        ao->driver->start(ao);
+    }
+
+    pthread_mutex_unlock(&p->lock);
+
+    ao_wakeup_playthread(ao);
+}
+
+void ao_set_paused(struct ao *ao, bool paused)
 {
     struct buffer_state *p = ao->buffer_state;
     bool wakeup = false;
@@ -377,7 +350,7 @@ void ao_pause(struct ao *ao)
 
     pthread_mutex_lock(&p->lock);
 
-    if (p->playing && !p->paused) {
+    if (p->playing && !p->paused && paused) {
         if (p->streaming && !ao->stream_silence) {
             if (ao->driver->write) {
                 if (!p->recover_pause)
@@ -387,6 +360,7 @@ void ao_pause(struct ao *ao)
                 } else {
                     ao->driver->reset(ao);
                     p->streaming = false;
+                    p->recover_pause = !ao->untimed;
                 }
             } else if (ao->driver->reset) {
                 // See ao_reset() why this is done outside of the lock.
@@ -394,65 +368,43 @@ void ao_pause(struct ao *ao)
                 p->streaming = false;
             }
         }
-        p->paused = true;
         wakeup = true;
-    }
-
-    pthread_mutex_unlock(&p->lock);
-
-    if (do_reset)
-        ao->driver->reset(ao);
-
-    if (wakeup)
-        ao_wakeup_playthread(ao);
-}
-
-void ao_resume(struct ao *ao)
-{
-    struct buffer_state *p = ao->buffer_state;
-    bool wakeup = false;
-
-    pthread_mutex_lock(&p->lock);
-
-    if (p->playing && p->paused) {
+    } else if (p->playing && p->paused && !paused) {
         if (ao->driver->write) {
-            if (p->streaming && p->hw_paused) {
+            if (p->hw_paused)
                 ao->driver->set_pause(ao, false);
-            } else {
-                p->recover_pause = true;
-            }
             p->hw_paused = false;
         } else {
             if (!p->streaming)
                 ao->driver->start(ao);
             p->streaming = true;
         }
-        p->paused = false;
         wakeup = true;
     }
+    p->paused = paused;
 
     pthread_mutex_unlock(&p->lock);
 
+    if (do_reset)
+        ao->driver->reset(ao);
+
     if (wakeup)
         ao_wakeup_playthread(ao);
 }
 
-bool ao_eof_reached(struct ao *ao)
+// Whether audio is playing. This means that there is still data in the buffers,
+// and ao_start() was called. This returns true even if playback was logically
+// paused. On false, EOF was reached, or an underrun happened, or ao_reset()
+// was called.
+bool ao_is_playing(struct ao *ao)
 {
     struct buffer_state *p = ao->buffer_state;
 
     pthread_mutex_lock(&p->lock);
-    bool eof = !p->playing;
-    if (ao->driver->write) {
-        eof |= !p->still_playing;
-    } else {
-        // For simplicity, ignore the latency. Otherwise, we would have to run
-        // an extra thread to time it.
-        eof |= mp_ring_buffered(p->buffers[0]) == 0;
-    }
+    bool playing = p->playing;
     pthread_mutex_unlock(&p->lock);
 
-    return eof;
+    return playing;
 }
 
 // Block until the current audio buffer has played completely.
@@ -461,35 +413,25 @@ void ao_drain(struct ao *ao)
     struct buffer_state *p = ao->buffer_state;
 
     pthread_mutex_lock(&p->lock);
-    p->final_chunk = true;
-    while (!p->paused && p->still_playing && p->streaming) {
-        if (ao->driver->write) {
-            if (p->draining) {
-                // Wait for EOF signal from AO.
-                pthread_cond_wait(&p->wakeup, &p->lock);
-            } else {
-                p->draining = true;
-                MP_VERBOSE(ao, "waiting for draining...\n");
-                pthread_mutex_unlock(&p->lock);
-                ao_wakeup_playthread(ao);
-                pthread_mutex_lock(&p->lock);
-            }
-        } else {
-            double left = mp_ring_buffered(p->buffers[0]) / (double)ao->bps * 1e6;
-            pthread_mutex_unlock(&p->lock);
+    while (!p->paused && p->playing) {
+        pthread_mutex_unlock(&p->lock);
+        double delay = ao_get_delay(ao);
+        pthread_mutex_lock(&p->lock);
 
-            if (left > 0) {
-                // Wait for lower bound.
-                mp_sleep_us(left);
-                // And then poll for actual end. No other way.
-                // Limit to arbitrary ~250ms max. waiting for robustness.
-                int64_t max = mp_time_us() + 250000;
-                while (mp_time_us() < max && !ao_eof_reached(ao))
-                    mp_sleep_us(1);
-            } else {
-                p->still_playing = false;
-            }
+        // Limit to buffer + arbitrary ~250ms max. waiting for robustness.
+        delay += mp_async_queue_get_samples(p->queue) / (double)ao->samplerate;
+        struct timespec ts = mp_rel_time_to_timespec(MPMAX(delay, 0) + 0.25);
+
+        // Wait for EOF signal from AO.
+        if (pthread_cond_timedwait(&p->wakeup, &p->lock, &ts)) {
+            MP_VERBOSE(ao, "drain timeout\n");
+            break;
+        }
 
+        if (!p->playing && mp_async_queue_get_samples(p->queue)) {
+            MP_WARN(ao, "underrun during draining\n");
+            pthread_mutex_unlock(&p->lock);
+            ao_start(ao);
             pthread_mutex_lock(&p->lock);
         }
     }
@@ -498,6 +440,12 @@ void ao_drain(struct ao *ao)
     ao_reset(ao);
 }
 
+static void wakeup_filters(void *ctx)
+{
+    struct ao *ao = ctx;
+    ao_wakeup_playthread(ao);
+}
+
 void ao_uninit(struct ao *ao)
 {
     struct buffer_state *p = ao->buffer_state;
@@ -515,6 +463,9 @@ void ao_uninit(struct ao *ao)
     if (ao->driver_initialized)
         ao->driver->uninit(ao);
 
+    talloc_free(p->filter_root);
+    talloc_free(p->queue);
+    talloc_free(p->pending);
     talloc_free(p->convert_buffer);
     talloc_free(p->temp_buf);
 
@@ -542,16 +493,28 @@ bool init_buffer_post(struct ao *ao)
         assert(ao->driver->get_state);
     }
 
-    for (int n = 0; n < ao->num_planes; n++)
-        p->buffers[n] = mp_ring_new(ao, ao->buffer * ao->sstride);
-
-    mpthread_mutex_init_recursive(&p->lock);
+    pthread_mutex_init(&p->lock, NULL);
     pthread_cond_init(&p->wakeup, NULL);
 
     pthread_mutex_init(&p->pt_lock, NULL);
     pthread_cond_init(&p->pt_wakeup, NULL);
 
+    p->queue = mp_async_queue_create();
+    p->filter_root = mp_filter_create_root(ao->global);
+    p->input = mp_async_queue_create_filter(p->filter_root, MP_PIN_OUT, p->queue);
+
+    mp_async_queue_resume_reading(p->queue);
+
+    struct mp_async_queue_config cfg = {
+        .sample_unit = AQUEUE_UNIT_SAMPLES,
+        .max_samples = ao->buffer,
+        .max_bytes = INT64_MAX,
+    };
+    mp_async_queue_set_config(p->queue, cfg);
+
     if (ao->driver->write) {
+        mp_filter_graph_set_wakeup_cb(p->filter_root, wakeup_filters, ao);
+
         p->thread_valid = true;
         if (pthread_create(&p->thread, NULL, playthread, ao)) {
             p->thread_valid = false;
@@ -590,89 +553,85 @@ static bool realloc_buf(struct ao *ao, int samples)
 }
 
 // called locked
-static void ao_play_data(struct ao *ao)
+static bool ao_play_data(struct ao *ao)
 {
     struct buffer_state *p = ao->buffer_state;
+
+    if (!(p->playing && (!p->paused || ao->stream_silence)))
+        return false;
+
     struct mp_pcm_state state;
     get_dev_state(ao, &state);
 
-    if (p->streaming && !state.playing && !ao->untimed) {
-        if (p->draining) {
-            MP_VERBOSE(ao, "underrun signaled for audio end\n");
-            p->still_playing = false;
-            pthread_cond_broadcast(&p->wakeup);
-        } else {
-            ao_add_events(ao, AO_EVENT_UNDERRUN);
+    if (p->streaming && !state.playing && !ao->untimed)
+        goto eof;
+
+    void **planes = NULL;
+    int space = state.free_samples;
+    if (!space)
+        return false;
+    assert(space >= 0);
+
+    int samples = 0;
+    bool got_eof = false;
+    if (ao->driver->write_frames) {
+        TA_FREEP(&p->pending);
+        samples = read_buffer(ao, NULL, 1, &got_eof);
+        planes = (void **)&p->pending;
+    } else {
+        if (!realloc_buf(ao, space)) {
+            MP_ERR(ao, "Failed to allocate buffer.\n");
+            return false;
         }
+        planes = (void **)mp_aframe_get_data_rw(p->temp_buf);
+        assert(planes);
 
-        p->streaming = false;
-    }
+        if (p->recover_pause) {
+            samples = MPCLAMP(p->prepause_state.delay * ao->samplerate, 0, space);
+            p->recover_pause = false;
+            mp_aframe_set_silence(p->temp_buf, 0, space);
+        }
 
-    // Round free space to period sizes to reduce number of write() calls.
-    int space = state.free_samples / ao->period_size * ao->period_size;
-    bool play_silence = p->paused || (ao->stream_silence && !p->still_playing);
-    space = MPMAX(space, 0);
-    if (!realloc_buf(ao, space)) {
-        MP_ERR(ao, "Failed to allocate buffer.\n");
-        return;
-    }
-    void **planes = (void **)mp_aframe_get_data_rw(p->temp_buf);
-    assert(planes);
-    int samples = mp_ring_buffered(p->buffers[0]) / ao->sstride;
-    if (samples > space)
-        samples = space;
-    if (play_silence)
-        samples = space;
-    if (p->recover_pause) {
-        samples = MPCLAMP(p->prepause_state.delay * ao->samplerate, 0, space);
-        p->recover_pause = false;
-        mp_aframe_set_silence(p->temp_buf, 0, space);
-    } else {
-        samp