Fixes and cleanup around synced gapless

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

3 files changed, 72 insertions, 61 deletions

diff --git a/src/libsink/desktop.c b/src/libsink/desktop.c
index 780818c..8582cd4 100644
--- a/src/libsink/desktop.c
+++ b/src/libsink/desktop.c
@@ -2,6 +2,9 @@
 #include <al/log.h>
 
 #ifdef CAMU_NO_MINIAUDIO_BACKENDS
+// @TODO: audio_null is incomplete in that it never actually reads from the buffers.
+//  This is an issue in CAMU_DIRECT_MODE because eventually the server-side packet_pool
+//  will be starved by pending audio packets.
 #define DESKTOP_NULL_AUDIO
 #endif
 
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 7b173a9..e9b4df4 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -71,7 +71,8 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX);
 #define CONNECTION_NUMBER(id) ((id >> 48) & 0xffff)
 
 // Only sink->target can ever be 0xb00b.
-#define ENTRY_IS_VALID(entry) ((entry) && (entry) != (struct camu_sink_entry *)0xb00b)
+#define OxbOOb ((struct camu_sink_entry *)0xb00b)
+#define ENTRY_IS_VALID(entry) ((entry) && (entry) != OxbOOb)
 
 // printf format for entries.
 #ifdef AL_DEBUG
@@ -601,7 +602,7 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
     al_assert(previous != target);
     al_assert(!previous->ended);
     // If none of the entry's buffers are added, we don't care about adding it to previous.
-    bool dangling_target = target == (struct camu_sink_entry *)0xb00b;
+    bool dangling_target = target == OxbOOb;
     if (dangling_target || target->ended || (AUDIO_STATE(previous) != BUFFER_ADDED && VIDEO_STATE(previous) != BUFFER_ADDED)) {
         remove_entry_buffers(previous);
         return;
@@ -713,8 +714,10 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
 
     // @TODO: Cleanup stop_video? and log_trace lengths.
     bool stop_video = false;
-    bool dangling_target = target == (struct camu_sink_entry *)0xb00b;
-    if (!dangling_target) {
+    bool dangling_target = target == OxbOOb;
+    if (dangling_target) {
+        log_trace("Ignored dangling target.");
+    } else {
         if (!target->ended) {
             remove_previous_if_contains(sink, target);
             add_or_queue_entry(target);
@@ -725,8 +728,6 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
             }
             stop_video = true;
         }
-    } else {
-        log_trace("Ignored dangling target.");
     }
 
     if (ensure_removed) {
@@ -737,15 +738,15 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
         al_assert(VIDEO_STATE(current) != BUFFER_ADDED);
     }
 
-    if (!dangling_target) {
+    if (dangling_target) {
+        stop_video = true;
+        sink->current = NULL;
+    } else {
         target->audio.ignore_paused = false;
         if (!target->paused) {
             camu_audio_buffer_resync(&target->audio.buf);
         }
         sink->current = target;
-    } else {
-        stop_video = true;
-        sink->current = NULL;
     }
 
     if (stop_video) {
@@ -775,14 +776,17 @@ 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;
-    log_trace("pause_and_swap_to("ENTRY_FMT"), current: "ENTRY_FMT".", ENTRY_ARG(target), ENTRY_ARG(current));
+    log_trace("pause_and_swap_to("ENTRY_FMT", %.2f), current: "ENTRY_FMT".",
+        ENTRY_ARG(target), at / 1000000.0, ENTRY_ARG(current));
     al_assert(target != current);
     al_assert(!sink->target);
     // This is extra verbose because the order is important.
     //  1. sink->target has to be set before calling clock_pause().
     //  2. current must still be paused even if it's ended.
     //  3. In the immediate case, switch_to() has to come last.
-    if (current && !current->ended) sink->target = target;
+    if (current && !current->ended) {
+        sink->target = target;
+    }
     if (current) {
         current->audio.ignore_paused = true;
         camu_clock_pause(&current->clock, at);
@@ -803,7 +807,7 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink
     entry->ended = true;
     queue_cmd(sink, (struct camu_sink_cmd){
         .op = END,
-        .value.u = entry->reset_id,
+        .value.u = entry->reset_token,
         .opaque = entry
     });
 #ifdef LIANA_LIST_SCUFFED_LOOP
@@ -811,14 +815,9 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink
     return true;
 #endif
     if (sink->target) {
-        al_assert(!sink->queued);
         switch_to(sink, sink->target);
         sink->target = NULL;
         return true;
-    } else if (sink->queued) {
-        switch_to(sink, sink->queued);
-        sink->queued = NULL;
-        return true;
     }
     return false;
 }
@@ -950,20 +949,23 @@ static void clock_callback(void *userdata, u8 op)
 
 static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
-    // @TODO: Video can't be ENDED here right?
     bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
     f64 avg_frame_duration = entry->video.buf.avg_frame_duration;
 #ifdef CAMU_SINK_LOCAL
     if (!AUDIO_EMPTY(entry) && !ignore_video) {
+        // Delay either the audio or video so we can start the clock as
+        // soon as possible while keeping A/V sync.
         f64 audio = camu_mixer_get_latency(sink->audio.mixer);
         struct camu_renderer *renderer = sink->video.renderer;
         f64 video = renderer->get_latency(renderer) * avg_frame_duration;
-        // Start either the audio or video early so we can start the clock as
-        // soon as possible while keeping A/V sync.
+        // If the audio buffer is delayed by less than the audio latency, data will be skipped.
+        f64 base = -audio;
         if (audio > video) {
-            camu_video_buffer_set_latency(&entry->video.buf, video - audio);
+            camu_audio_buffer_set_latency(&entry->audio.buf, base);
+            camu_video_buffer_set_latency(&entry->video.buf, base - (audio - video));
         } else if (video > audio) {
-            camu_audio_buffer_set_latency(&entry->audio.buf, audio - video);
+            camu_audio_buffer_set_latency(&entry->audio.buf, base);
+            camu_video_buffer_set_latency(&entry->video.buf, base + (video - audio));
         }
     }
     // If we're local we don't have to worry about syncing audio-only entries.
@@ -971,7 +973,7 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s
     camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
 #else
     // When trying to sync clients with different audio/video latencies (common case),
-    // our only option is to factor the latency directly into the buffers.
+    // our only option is to shift each buffer forward directly by their latency.
     f64 audio = camu_mixer_get_latency(sink->audio.mixer);
     if (!ignore_video) {
         struct camu_renderer *renderer = sink->video.renderer;
@@ -1064,6 +1066,8 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     case LIANA_CLIENT_CONFIGURE_COMPLETE: {
         // All present buffers were configured.
         nn_mutex_lock(&sink->lock);
+        // This is mainly to assert DETACHED handling.
+        al_assert(!VIDEO_ENDED(entry) && !AUDIO_ENDED(entry));
         evaluate_and_set_buffer_params(sink, entry);
         nn_mutex_unlock(&sink->lock);
         break;
@@ -1215,7 +1219,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         // List has no concept of a local sink, ignore it's request.
         time->at = 0;
 #endif
-        log_trace("resume_at("ENTRY_FMT"), seek_pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->seek_pos / 1000000.0, entry->clock.paused_at);
+        log_trace("resume_at("ENTRY_FMT"), pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->pos / 1000000.0, entry->clock.paused_at);
         nn_mutex_lock(&sink->lock);
         bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
         if (!AUDIO_EMPTY(entry)) {
@@ -1224,14 +1228,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
         }
         if (!ignore_video) {
-            camu_video_buffer_reset(&entry->video.buf, time->seek_pos);
+            camu_video_buffer_reset(&entry->video.buf, time->pos);
         }
 #ifdef LIANA_LIST_SCUFFED_LOOP
-        if (time->seek_pos == 0) {
+        if (time->pos == 0) {
             camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock));
         } else {
 #endif
-            camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
+            camu_clock_seek(&entry->clock, time->pos / 1000000.0, time->at);
 #ifdef LIANA_LIST_SCUFFED_LOOP
         }
 #endif
@@ -1305,7 +1309,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         remove_from_queue_by_opaque(sink, entry);
 
         if (entry == sink->target) {
-            sink->target = (struct camu_sink_entry *)0xb00b;
+            sink->target = OxbOOb;
             log_warn("Attempting to handle a disconnected target.");
         } else if (entry == sink->current) {
             if (sink->suspended) {
@@ -1326,8 +1330,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             }
             // If current was never fully added we need to call this here.
             maybe_remove_previous(sink);
-        } else if (entry == sink->queued) {
-            sink->queued = NULL;
         }
 
         nn_mutex_unlock(&sink->lock);
@@ -1415,11 +1417,9 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
     s32 sequence = nn_packet_read_s32(packet);
     u64 at = nn_packet_read_u64(packet);
-    u64 seek_pos = nn_packet_read_u64(packet);
+    u64 pos = nn_packet_read_u64(packet);
     u8 pause = nn_packet_read_u8(packet);
-    bool ended = nn_packet_read_bool(packet);
-    (void)ended;
-    u32 reset_id = nn_packet_read_u32(packet);
+    u32 reset_token = nn_packet_read_u32(packet);
 
     struct camu_sink_entry *entry = get_entry_from_id(sink, id);
     bool create = !entry;
@@ -1427,24 +1427,26 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     entry->sequence = sequence;
     entry->lru = sink->lru;
     sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
-    entry->reset_id = reset_id;
+    entry->reset_token = reset_token;
     if (create) {
         entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE;
-        camu_clock_set(&entry->clock, seek_pos / 1000000.0);
-        lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, seek_pos);
+        camu_clock_set(&entry->clock, pos / 1000000.0);
+        lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, pos);
     }
 
     // Don't lock before client_connect() or we could deadlock in CLIENT_CLOSED on a failed socket_connect().
     nn_mutex_lock(&sink->lock);
 
+    struct camu_sink_entry *current = sink->current;
+
     if (op == LIANA_SINK_BUFFER) {
+        log_trace("buffered("ENTRY_FMT"), created: %s.", ENTRY_ARG(entry), BOOLSTR(create));
         goto out;
     } else if (op == LIANA_SINK_BUFFER_AND_QUEUE) {
-        sink->queued = entry;
+        log_trace("queued("ENTRY_FMT"), %s, created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create));
         goto out;
     }
 
-    struct camu_sink_entry *current = sink->current;
 #ifdef CAMU_SINK_LOCAL
     (void)at;
     al_assert(entry != current);
@@ -1466,7 +1468,8 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
 #else
     // For target to be set that must mean current is set, armed to pause, and not ended.
     struct camu_sink_entry *prev_target = sink->target;
-    log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create), ENTRY_ARG(prev_target));
+    log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry),
+        lia_pause_op_name(pause), BOOLSTR(create), ENTRY_ARG(prev_target));
     switch (pause) {
     case LIANA_PAUSE_NONE:
         if (prev_target) {
@@ -1481,13 +1484,15 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
     case LIANA_PAUSE_RESUME:
         if (prev_target) {
             sink->target = NULL;
+            if (entry == current) {
+                // Switched back to current before pause_and_swap_to() completed.
+                camu_audio_buffer_resync(&entry->audio.buf);
+            }
         } else {
             al_assert(entry != current);
         }
         if (entry != current) {
             switch_to(sink, entry);
-        } else {
-            camu_audio_buffer_resync(&entry->audio.buf);
         }
         camu_clock_resume(&entry->clock, at);
         break;
@@ -1496,16 +1501,17 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
             al_assert(current);
             al_assert(!current->ended);
             al_assert(prev_target != entry);
-            if (current != entry) {
-                sink->target = entry;
-            } else {
+            if (entry == current) {
+                // pause_and_swap_to() negated.
                 sink->target = NULL;
+            } else {
+                sink->target = entry;
             }
-            if (prev_target != (struct camu_sink_entry *)0xb00b) {
+            if (prev_target != OxbOOb) {
                 camu_clock_pause(&prev_target->clock, at);
             }
         } else {
-            al_assert(current != entry);
+            al_assert(entry != current);
             pause_and_swap_to(sink, entry, at);
         }
         break;
@@ -1514,17 +1520,17 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
             al_assert(current);
             al_assert(!current->ended);
             al_assert(prev_target != entry);
-            if (current != entry) {
-                sink->target = entry;
-            } else {
+            if (entry == current) {
                 sink->target = NULL;
                 camu_audio_buffer_resync(&entry->audio.buf);
+            } else {
+                sink->target = entry;
             }
-            if (prev_target != (struct camu_sink_entry *)0xb00b) {
+            if (prev_target != OxbOOb) {
                 camu_clock_pause(&prev_target->clock, at);
             }
         } else {
-            al_assert(current != entry);
+            al_assert(entry != current);
             pause_and_swap_to(sink, entry, at);
         }
         camu_clock_resume(&entry->clock, at);
@@ -1564,14 +1570,15 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
     if (!entry->held) local_pause(sink, entry);
 #else
     switch (pause) {
-    case LIANA_PAUSE_PAUSE:
+    case LIANA_PAUSE_PAUSE: {
         entry->paused = true;
         camu_clock_pause(&entry->clock, at);
         log_info("Clock paused.");
         // Audio will be stopped in a BUFFER_PAUSED callback.
         // Video will be stopped in a CLOCK_PAUSED callback.
         break;
-    case LIANA_PAUSE_RESUME:
+    }
+    case LIANA_PAUSE_RESUME: {
         entry->paused = false;
         camu_clock_resume(&entry->clock, at);
         log_info("Clock resumed.");
@@ -1590,6 +1597,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
         }
         break;
     }
+    }
 #endif
     nn_mutex_unlock(&sink->lock);
 
@@ -1610,12 +1618,14 @@ static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn
     (void)sequence;
     u64 at = nn_packet_read_u64(packet);
     u64 pos = nn_packet_read_u64(packet);
-    u32 reset_id = nn_packet_read_u32(packet);
+    u32 reset_token = nn_packet_read_u32(packet);
 
     struct camu_sink_entry *entry = get_entry_from_id(sink, id);
     if (!entry) goto out;
-    log_trace("seek("ENTRY_FMT"), reset_id: %u.", ENTRY_ARG(entry), reset_id);
-    entry->reset_id = reset_id;
+    nn_mutex_lock(&sink->lock);
+    log_trace("seek("ENTRY_FMT"), reset_token: %u.", ENTRY_ARG(entry), reset_token);
+    entry->reset_token = reset_token;
+    nn_mutex_unlock(&sink->lock);
     // The rest of the seek is handled in CLIENT_REMOVE_BUFFERS/RESUME_AT/RECONNECTED.
     lia_client_seek(&entry->client, pos, at);
 
@@ -1705,7 +1715,6 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
     nn_signal_start(&sink->queue_signal);
     camu_queue_init(sink->queue);
     sink->current = NULL;
-    sink->queued = NULL;
     sink->target = NULL;
     sink->suspended = NULL;
     al_array_init(sink->previous);
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 2c96313..5958829 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -50,7 +50,7 @@ struct camu_sink_entry {
     u16 lru;
     bool disconnected;
     bool ended;
-    u32 reset_id;
+    u32 reset_token;
     struct camu_clock clock;
     bool held; // Only used in SINK_LOCAL.
     bool paused;
@@ -89,7 +89,6 @@ struct camu_sink {
     struct nn_signal queue_signal;
     queue(struct camu_sink_cmd) queue;
     struct camu_sink_entry *current;
-    struct camu_sink_entry *queued;
     struct camu_sink_entry *target;
     // If an entry that was current was removed for a reconnect, it's "suspended".
     struct camu_sink_entry *suspended;