Audio buffer, mixer and sink improvements

Signed-off-by: Andrew Opalach <andrew@akon.city>

2 files changed, 201 insertions, 86 deletions

diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 1569286..f588cfa 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -38,51 +38,64 @@ enum {
 
 // Command queue commands.
 enum {
-    START,
+    START = 0,
     STOP,
+    ADD_BUFFER,
+    REMOVE_BUFFER,
+    CLEAR,
+    CLOSE,
     TOGGLE_PAUSE,
     SEEK,
     SKIP,
     SHUFFLE,
     RESEEK,
     UNSET,
-    END,
-    CLOSE
+    END
 };
 
 // Number of entries to keep buffered at one time.
 #define ENTRY_MAX_AGE 7
 
-// If a buffer is still INIT or QUEUED after an entry is configured, it's "empty".
+// If a buffer is still INIT or QUEUED after the entry is configured, it's "empty".
 #define BUFFER_EMPTY(buf) ((buf)->state == BUFFER_INIT || (buf)->state == BUFFER_QUEUED)
 #define BUFFER_NOT_EMPTY(buf) (!BUFFER_EMPTY(buf))
 
-#ifdef CAMU_SINK_NO_VIDEO
-#define VIDEO_ADDED_OR_EMPTY(entry) true
-#else
-#define VIDEO_ADDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->video))
+#define AUDIO_EMPTY(entry) BUFFER_EMPTY(&(entry)->audio)
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_EMPTY(entry) BUFFER_EMPTY(&(entry)->video)
 #endif
-#define AUDIO_ADDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->audio))
 
-#ifdef CAMU_SINK_NO_VIDEO
-#define VIDEO_ENDED_OR_EMPTY(entry) true
+#define AUDIO_NOT_EMPTY(entry) BUFFER_NOT_EMPTY(&(entry)->audio)
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_NOT_EMPTY(entry) BUFFER_NOT_EMPTY(&(entry)->video)
+#endif
+
+#define AUDIO_ADDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->audio))
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_ADDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->video))
 #else
-#define VIDEO_ENDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->video))
+#define VIDEO_ADDED_OR_EMPTY(entry) true
 #endif
+
 #define AUDIO_ENDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->audio))
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_ENDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->video))
+#else
+#define VIDEO_ENDED_OR_EMPTY(entry) true
+#endif
 
-#ifdef CAMU_SINK_NO_VIDEO
-#define VIDEO_REMOVED_OR_EMPTY(entry) true
+#define AUDIO_NOT_ADDED(entry) ((entry)->audio.state != BUFFER_ADDED)
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_NOT_ADDED(entry) ((entry)->video.state != BUFFER_ADDED)
 #else
-#define VIDEO_REMOVED_OR_EMPTY(entry) ((entry)->video.state != BUFFER_ADDED)
+#define VIDEO_NOT_ADDED(entry) true
 #endif
-#define AUDIO_REMOVED_OR_EMPTY(entry) ((entry)->audio.state != BUFFER_ADDED)
 
 #ifndef CAMU_SINK_NO_VIDEO
-#define ENTRY_IS_SINGLE_FRAME(entry) camu_video_buffer_is_single_frame(&(entry)->video.buf)
+#define VIDEO_IS_SINGLE_FRAME(entry) camu_video_buffer_is_single_frame(&(entry)->video.buf)
 #endif
 
-#if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED
+#if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED_START_STOP
 #define BLOCKING_SLEEP(delay) nn_thread_sleep(delay)
 #else
 #define BLOCKING_SLEEP(delay) nn_event_loop_sleep(sink->loop, delay)
@@ -110,12 +123,48 @@ static inline bool entry_video_buffer_held(struct camu_sink_entry *entry)
 }
 #endif
 
+static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd)
+{
+    camu_queue_push(sink->queue, cmd);
+    nn_signal_send(&sink->queue_signal);
+}
+
+static inline void add_entry_audio_buffer(struct camu_sink_entry *entry)
+{
+    struct camu_sink *sink = entry->sink;
+#ifdef CAMU_MIXER_THREADED_START_STOP
+    sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+#else
+    queue_cmd(sink, (struct camu_sink_cmd){
+        .op = ADD_BUFFER,
+        .value.i = CAMU_SINK_AUDIO,
+        .opaque = entry
+    });
+#endif
+}
+
+#ifndef CAMU_SINK_NO_VIDEO
+static inline void add_entry_video_buffer(struct camu_sink_entry *entry)
+{
+    struct camu_sink *sink = entry->sink;
+    sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+}
+#endif
+
 static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
     al_assert(!entry->ended && entry->audio.state != BUFFER_ENDED);
     al_assert(entry->audio.state != BUFFER_INIT);
     if (entry->audio.state == BUFFER_ADDED) {
+#ifdef CAMU_MIXER_THREADED_START_STOP
         sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+#else
+        queue_cmd(sink, (struct camu_sink_cmd){
+            .op = REMOVE_BUFFER,
+            .value.i = CAMU_SINK_AUDIO,
+            .opaque = entry
+        });
+#endif
         entry->audio.state = BUFFER_SET_OR_BUFFERED;
     } else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) {
         entry->audio.state = BUFFER_CONFIGURED;
@@ -187,12 +236,12 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent
 {
     if (camu_clock_is_paused(&entry->clock)) {
         camu_clock_resume(&entry->clock, 0);
-        if (!BUFFER_EMPTY(&entry->audio) && sink->audio.state == SINK_PAUSED) {
+        if (AUDIO_NOT_EMPTY(entry) && sink->audio.state == SINK_PAUSED) {
             sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL);
             sink->audio.state = SINK_PLAYING;
         }
 #ifndef CAMU_SINK_NO_VIDEO
-        if (!ENTRY_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PAUSED) {
+        if (!VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PAUSED) {
             sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL);
             sink->video.state = SINK_PLAYING;
         }
@@ -200,7 +249,7 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent
     } else {
         camu_clock_pause(&entry->clock, 0);
 #ifndef CAMU_SINK_NO_VIDEO
-        if (!ENTRY_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) {
+        if (!VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) {
             sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL);
             sink->video.state = SINK_PAUSED;
         }
@@ -224,7 +273,6 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
 {
     switch (cmd->op) {
     case START: {
-        nn_mutex_lock(&sink->mutex);
         switch (cmd->value.i) {
         case CAMU_SINK_AUDIO:
             if (sink->audio.state == SINK_PAUSED) {
@@ -241,11 +289,9 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
             break;
 #endif
         }
-        nn_mutex_unlock(&sink->mutex);
         break;
     }
     case STOP: {
-        nn_mutex_lock(&sink->mutex);
         switch (cmd->value.i) {
         case CAMU_SINK_AUDIO:
             if (sink->audio.state == SINK_PLAYING) {
@@ -262,9 +308,54 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
             break;
 #endif
         }
-        nn_mutex_unlock(&sink->mutex);
         break;
     }
+    case ADD_BUFFER: {
+        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+        switch (cmd->value.i) {
+        case CAMU_SINK_AUDIO:
+            sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+            break;
+#ifndef CAMU_SINK_NO_VIDEO
+        case CAMU_SINK_VIDEO:
+            sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+            break;
+#endif
+        }
+        break;
+    }
+    case REMOVE_BUFFER: {
+        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+        switch (cmd->value.i) {
+        case CAMU_SINK_AUDIO:
+            sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+            break;
+#ifndef CAMU_SINK_NO_VIDEO
+        case CAMU_SINK_VIDEO:
+            sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+            break;
+#endif
+        }
+        break;
+    }
+    case CLEAR: {
+        switch (cmd->value.i) {
+        case CAMU_SINK_AUDIO:
+            sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_AUDIO, NULL);
+            break;
+#ifndef CAMU_SINK_NO_VIDEO
+        case CAMU_SINK_VIDEO:
+            sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_VIDEO, NULL);
+            break;
+#endif
+        }
+        break;
+    }
+    case CLOSE: {
+        nn_signal_stop(&sink->queue_signal);
+        sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
+        return;
+    }
     case SKIP: {
         if (!sink->conn) return;
         struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
@@ -338,11 +429,6 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
         nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
         break;
     }
-    case CLOSE: {
-        nn_signal_stop(&sink->queue_signal);
-        sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
-        return;
-    }
     }
 }
 
@@ -358,17 +444,11 @@ static void queue_signal_callback(void *userdata)
     }
 }
 
-static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd)
-{
-    camu_queue_push(sink->queue, cmd);
-    nn_signal_send(&sink->queue_signal);
-}
-
 static void mixer_callback(void *userdata, u8 op)
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
     if (op == CAMU_MIXER_EMPTY) {
-        al_log_debug("sink", "Mixer empty.");
+        al_log_info("sink", "Mixer empty.");
         queue_cmd(sink, (struct camu_sink_cmd){
             .op = STOP,
             .value.i = CAMU_SINK_AUDIO
@@ -510,31 +590,34 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
         entry->audio.state = BUFFER_SET_OR_BUFFERED;
     } else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) {
         entry->audio.state = BUFFER_ADDED;
-#ifndef CAMU_SINK_LOCAL
+        bool ignore_video = true;
+#ifndef CAMU_SINK_NO_VIDEO
+        ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
+#endif
+        camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
+#ifdef CAMU_SINK_LOCAL
+        // If we're local we don't have to worry about syncing audio-only entries.
+        camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video);
+#else
+        // Force resync.
         camu_audio_buffer_unpause(&entry->audio.buf);
 #endif
         if (VIDEO_ADDED_OR_EMPTY(entry)) {
-            entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+            add_entry_audio_buffer(entry);
             queue_cmd(entry->sink, (struct camu_sink_cmd){
                 .op = entry->audio.buffer_paused ? STOP : START,
                 .value.i = CAMU_SINK_AUDIO
             });
 #ifndef CAMU_SINK_NO_VIDEO
-            bool stop_video = false;
-            // Single frames are unconditionally added in add_video_if_set_and_buffered().
-            if (!ENTRY_IS_SINGLE_FRAME(entry)) {
-                if (entry->video.state == BUFFER_ADDED) {
-                    entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
-                } else {
-                    // Entry has no video.
-                    stop_video = true;
-                }
+            if (!ignore_video) {
+                // Single frames are unconditionally added in add_video_if_set_and_buffered().
+                add_entry_video_buffer(entry);
             }
 #endif
             maybe_remove_previous(entry->sink);
 #ifndef CAMU_SINK_NO_VIDEO
-            // This must come after maybe_remove_previous().
-            if (stop_video) {
+            if (VIDEO_EMPTY(entry)) {
+                // This must come after maybe_remove_previous().
                 queue_cmd(entry->sink, (struct camu_sink_cmd){
                     .op = STOP,
                     .value.i = CAMU_SINK_VIDEO
@@ -558,19 +641,19 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
         entry->video.state = BUFFER_SET_OR_BUFFERED;
     } else if (entry->video.state == BUFFER_SET_OR_BUFFERED) {
         entry->video.state = BUFFER_ADDED;
-        bool single_frame = ENTRY_IS_SINGLE_FRAME(entry);
+        bool single_frame = VIDEO_IS_SINGLE_FRAME(entry);
         if (AUDIO_ADDED_OR_EMPTY(entry)) {
             if (entry->audio.state == BUFFER_ADDED) {
-                entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+                add_entry_audio_buffer(entry);
                 queue_cmd(entry->sink, (struct camu_sink_cmd){
                     .op = entry->audio.buffer_paused ? STOP : START,
                     .value.i = CAMU_SINK_AUDIO
                 });
             }
-            entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+            add_entry_video_buffer(entry);
             maybe_remove_previous(entry->sink);
         } else if (single_frame) {
-            entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+            add_entry_video_buffer(entry);
         }
         // Stopping video here is needed if skipping from a video to an image.
         queue_cmd(entry->sink, (struct camu_sink_cmd){
@@ -591,11 +674,11 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
             maybe_add_to_previous(sink, current, target);
         } else {
 #ifndef CAMU_SINK_NO_VIDEO
-            if (ENTRY_IS_SINGLE_FRAME(current)) {
+            if (VIDEO_IS_SINGLE_FRAME(current)) {
                 remove_entry_video_buffer(sink, current);
             }
 #endif
-            al_assert(AUDIO_REMOVED_OR_EMPTY(current) && VIDEO_REMOVED_OR_EMPTY(current));
+            al_assert(AUDIO_NOT_ADDED(current) && VIDEO_NOT_ADDED(current));
         }
 
         if (sink->reconnecting) {
@@ -607,7 +690,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
     if (!target->ended) {
         add_or_queue_entry(target);
 #ifndef CAMU_SINK_NO_VIDEO
-    } else if (ENTRY_IS_SINGLE_FRAME(target)) {
+    } else if (VIDEO_IS_SINGLE_FRAME(target)) {
         add_video_if_set_and_buffered(target);
 #endif
     }
@@ -620,6 +703,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
 static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at)
 {
     struct camu_sink_entry *current = sink->current;
+    al_assert(target != current);
     if (!current || current->ended) {
         switch_to(sink, target);
     } else {
@@ -693,11 +777,11 @@ static void audio_buffer_callback(void *userdata, u8 op)
         // This assert likely doesn't matter due to the handling of the ENDED state.
         al_assert(entry->audio.state == BUFFER_SET_OR_BUFFERED);
         entry->audio.state = BUFFER_ENDED;
-        bool run_queue = VIDEO_ENDED_OR_EMPTY(entry);
+        bool end_entry = VIDEO_ENDED_OR_EMPTY(entry);
 #ifndef CAMU_SINK_NO_VIDEO
-        run_queue = run_queue || ENTRY_IS_SINGLE_FRAME(entry);
+        end_entry = end_entry || VIDEO_IS_SINGLE_FRAME(entry);
 #endif
-        if (run_queue) {
+        if (end_entry) {
             end_entry_and_advance_queue(sink, entry);
         }
         nn_mutex_unlock(&sink->mutex);
@@ -731,7 +815,10 @@ static void video_buffer_callback(void *userdata, u8 op)
         bool swapped = false;
         nn_mutex_lock(&sink->mutex);
         al_log_info("sink", "Video EOF.");
-        if (!ENTRY_IS_SINGLE_FRAME(entry)) {
+        if (AUDIO_NOT_EMPTY(entry)) {
+            camu_audio_buffer_set_no_video(&entry->audio.buf, true);
+        }
+        if (!VIDEO_IS_SINGLE_FRAME(entry)) {
             if (entry->video.state == BUFFER_ADDED) {
                 remove_entry_video_buffer(sink, entry);
             }
@@ -785,7 +872,7 @@ static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *ent
 {
 #ifdef CAMU_SINK_LOCAL
 #ifndef CAMU_SINK_NO_VIDEO
-    if (BUFFER_NOT_EMPTY(&entry->audio) && BUFFER_NOT_EMPTY(&entry->video)) {
+    if (AUDIO_NOT_EMPTY(entry) && VIDEO_NOT_EMPTY(entry)) {
         f64 audio = camu_mixer_get_latency(sink->audio.mixer);
         s32 frames = audio / entry->video.buf.avg_frame_duration;
         frames -= sink->video.renderer->get_latency(sink->video.renderer);
@@ -800,7 +887,7 @@ static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *ent
     // latency directly into the audio buffer.
     f64 audio = camu_mixer_get_latency(sink->audio.mixer);
 #ifndef CAMU_SINK_NO_VIDEO
-    if (BUFFER_NOT_EMPTY(&entry->video)) {
+    if (VIDEO_NOT_EMPTY(entry)) {
         s32 frames = audio / entry->video.buf.avg_frame_duration;
         frames += sink->video.renderer->get_latency(sink->video.renderer);
         camu_video_buffer_set_latency(&entry->video.buf, frames);
@@ -810,6 +897,19 @@ static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *ent
 #endif
 }
 
+static void run_queue_by_opaque(struct camu_sink *sink, void *opaque)
+{
+    camu_queue_lock(sink->queue);
+    struct camu_sink_cmd *cmd;
+    al_array_foreach_ptr(sink->queue.a, i, cmd) {
+        if (cmd->opaque == opaque) {
+            handle_sink_cmd(sink, cmd);
+            al_array_remove_at_iter(sink->queue.a, i);
+        }
+    }
+    camu_queue_unlock(sink->queue);
+}
+
 static void client_callback(void *userdata, u8 op, struct camu_codec_stream *stream, void *opaque)
 {
     struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
@@ -833,7 +933,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             } else {
                 assert(false);
             }
-            if (VIDEO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry);
+            if (VIDEO_ADDED_OR_EMPTY(entry)) {
+                evaluate_latency(sink, entry);
+            }
             nn_mutex_unlock(&sink->mutex);
             break;
 #ifndef CAMU_SINK_NO_VIDEO
@@ -852,7 +954,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             } else {
                 assert(false);
             }
-            if (AUDIO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry);
+            if (AUDIO_ADDED_OR_EMPTY(entry)) {
+                evaluate_latency(sink, entry);
+            }
             nn_mutex_unlock(&sink->mutex);
             break;
         case CAMU_STREAM_SUBTITLE:
@@ -879,14 +983,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         struct camu_codec_frame *frame = (struct camu_codec_frame *)opaque;
         switch (stream->type) {
         case CAMU_STREAM_AUDIO:
-            if (BUFFER_NOT_EMPTY(&entry->audio)) {
+            if (AUDIO_NOT_EMPTY(entry)) {
                 camu_audio_buffer_push(&entry->audio.buf, frame);
                 return;
             }
             break;
 #ifndef CAMU_SINK_NO_VIDEO
         case CAMU_STREAM_VIDEO:
-            if (BUFFER_NOT_EMPTY(&entry->video)) {
+            if (VIDEO_NOT_EMPTY(entry)) {
                 camu_video_buffer_push(&entry->video.buf, frame);
                 return;
             }
@@ -917,11 +1021,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 
         nn_mutex_lock(&sink->mutex);
 
-        // We need to call this if entry was added to previous then,
-        //  - it's being cleaned up after (ENTRY_MAX_AGE - 1) entries were added but none buffered.
-        //  - it was seeked.
-        maybe_remove_from_previous(sink, entry);
-
         if (reconnect) {
             entry->ended = false;
             if (entry == sink->current) {
@@ -933,6 +1032,11 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             }
         }
 
+        // We need to call this if entry was added to previous then,
+        //  - it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered.
+        //  - it was seeked.
+        maybe_remove_from_previous(sink, entry);
+
         bool skip_audio = sink->audio.state == SINK_PAUSED;
         if (entry->audio.state == BUFFER_ADDED) {
             remove_entry_audio_buffer(sink, entry);
@@ -941,11 +1045,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         if (entry->audio.state > BUFFER_CONFIGURED) {
             entry->audio.state = BUFFER_CONFIGURED;
         }
+
 #ifndef CAMU_SINK_NO_VIDEO
-        bool ignore_video = BUFFER_EMPTY(&entry->video);
-        if (!ignore_video) {
-            ignore_video = reconnect && ENTRY_IS_SINGLE_FRAME(entry);
-        }
+        bool ignore_video = VIDEO_EMPTY(entry) || (reconnect && VIDEO_IS_SINGLE_FRAME(entry));
         if (!ignore_video) {
             if (entry->video.state == BUFFER_ADDED) {
                 remove_entry_video_buffer(sink, entry);
@@ -956,6 +1058,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         }
 #endif
 
+        // Resolve any queued REMOVE_BUFFER requests before waiting.
+        run_queue_by_opaque(sink, entry);
+
         nn_mutex_unlock(&sink->mutex);
 
         while ( // Block until buffers are removed.
@@ -967,11 +1072,11 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         ) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); }
 
         if (reconnect) {
-            if (BUFFER_NOT_EMPTY(&entry->audio)) {
+            if (AUDIO_NOT_EMPTY(entry)) {
                 camu_audio_buffer_reset(&entry->audio.buf);
             }
 #ifndef CAMU_SINK_NO_VIDEO
-            if (BUFFER_NOT_EMPTY(&entry->video) && !ignore_video) {
+            if (VIDEO_NOT_EMPTY(entry) && !ignore_video) {
                 camu_video_buffer_reset(&entry->video.buf);
             }
 #endif
@@ -998,14 +1103,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     case LIANA_CLIENT_RECONNECTED: {
         nn_mutex_lock(&sink->mutex);
         if (entry == sink->reconnecting) {
+            al_assert(entry == sink->current);
             if (entry->audio.state > BUFFER_QUEUED) {
                 add_audio_if_set_and_buffered(entry);
             }
 #ifndef CAMU_SINK_NO_VIDEO
-            if (entry->video.state > BUFFER_QUEUED && !ENTRY_IS_SINGLE_FRAME(entry)) {
+            if (entry->video.state > BUFFER_QUEUED && !VIDEO_IS_SINGLE_FRAME(entry)) {
                 add_video_if_set_and_buffered(entry);
             }
 #endif
+            sink->reconnecting = NULL;
         }
         nn_mutex_unlock(&sink->mutex);
         break;
@@ -1013,7 +1120,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     case LIANA_CLIENT_EOF: {
         switch (stream->type) {
         case CAMU_STREAM_AUDIO: {
-            if (BUFFER_NOT_EMPTY(&entry->audio)) {
+            if (AUDIO_NOT_EMPTY(entry)) {
                 camu_audio_buffer_flush(&entry->audio.buf);
             }
             break;
@@ -1021,7 +1128,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 #ifndef CAMU_SINK_NO_VIDEO
         case CAMU_STREAM_VIDEO: {
             // Single frames are immediately flushed inside the buffer.
-            if (BUFFER_NOT_EMPTY(&entry->video) && !ENTRY_IS_SINGLE_FRAME(entry)) {
+            if (VIDEO_NOT_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
                 camu_video_buffer_flush(&entry->video.buf);
             }
             break;
@@ -1053,16 +1160,20 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             sink->queued = NULL;
         }
 
+        bool removed;
+        al_array_check_remove(sink->entries, entry, removed);
+
+        nn_mutex_unlock(&sink->mutex);
+
         lia_client_free(&entry->client);
         camu_audio_buffer_free(&entry->audio.buf);
 #ifndef CAMU_SINK_NO_VIDEO
         camu_video_buffer_free(&entry->video.buf);
 #endif
-        bool removed;
-        al_array_check_remove(sink->entries, entry, removed);
-        nn_mutex_unlock(&sink->mutex);
         al_free(entry);
+
         al_log_info("sink", "Entry closed by %s.", removed ? "disconnect" : "cleanup");
+
         break;
     }
     }
@@ -1440,7 +1551,7 @@ void camu_sink_toggle_pause(struct camu_sink *sink)
     if (current) {
         queue_cmd(sink, (struct camu_sink_cmd){
             .op = TOGGLE_PAUSE,
-            .value.f = camu_clock_get_pts(&current->clock, 0.0, true),
+            .value.f = camu_clock_get_pts(&current->clock, 0.0, false),
             .opaque = current
         });
     }
@@ -1488,6 +1599,10 @@ void camu_sink_stop(struct camu_sink *sink)
         .value.i = CAMU_SINK_AUDIO
     });
     queue_cmd(sink, (struct camu_sink_cmd){
+        .op = CLEAR,
+        .value.i = CAMU_SINK_AUDIO
+    });
+    queue_cmd(sink, (struct camu_sink_cmd){
         .op = CLOSE
     });
 }
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index c08f763..10fc182 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -26,9 +26,9 @@ enum {
 enum {
     CAMU_SINK_ADD_BUFFER = 0,
     CAMU_SINK_REMOVE_BUFFER,
-    CAMU_SINK_SWAP_BUFFER,
     CAMU_SINK_START,
     CAMU_SINK_STOP,
+    CAMU_SINK_CLEAR,
     CAMU_SINK_EXIT
 };