Sink-side list sync resilience, vcr fix

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

2 files changed, 17 insertions, 12 deletions

diff --git a/src/liana/list.c b/src/liana/list.c
index 18bfd6c..a8d5f71 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -180,8 +180,12 @@ static bool assume_done(struct lia_list_entry *entry, u64 at)
 
 void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
 {
-    if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
     if (index == list->current) return;
+    if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
+    if (sequence == list->previous) {
+        // This can happen but almost certainly won't be expected behavior.
+        return;
+    }
 
     struct lia_list_entry *current = get_entry_from_sequence(list, sequence);
     struct lia_list_entry *target = get_entry_from_sequence(list, index);
@@ -208,7 +212,6 @@ void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
 
     // This should only happen if `start` has never been set.
     if (target->start == LIANA_TIMESTAMP_INVALID && target->paused_at == LIANA_TIMESTAMP_INVALID) {
-        al_log_info("dd", "Hello");
         target->start = at;
     }
 
@@ -233,7 +236,7 @@ void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
             target->start = at;
             pause = !done ? LIANA_PAUSE_BOTH : LIANA_PAUSE_RESUME;
         }
-        al_log_info("dd", "%d: not paused, done: %d, pause: %d", sequence, done, pause);
+        al_log_debug("list", "%d: not paused, done: %d, pause: %d", sequence, done, pause);
     } else {
         if (assume_done(target, now) || target->paused_at != LIANA_TIMESTAMP_INVALID) {
             pause = LIANA_PAUSE_NONE;
@@ -241,7 +244,7 @@ void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
             target->start = at;
             pause = LIANA_PAUSE_RESUME;
         }
-        al_log_info("dd", "%d: paused, pause: %d", sequence, pause);
+        al_log_debug("list", "%d: paused, pause: %d", sequence, pause);
     }
 
     list->current = index;
@@ -331,11 +334,11 @@ void lia_list_end(struct lia_list *list, s32 sequence)
     if (sequence != list->current) return;
     s32 size = (s32)list->entries.size;
     s32 next = sequence + 1;
-    list->previous = sequence;
     struct lia_list_entry *current = al_array_at(list->entries, list->current);
     current->offset = current->duration;
     if (list->queued >= 0) {
         list->current = list->queued;
+        list->previous = sequence;
         list->queued = -1;
         struct lia_list_sink *sink;
         al_array_foreach(list->sinks, i, sink) {
diff --git a/src/liana/vcr.c b/src/liana/vcr.c
index 39fe0fe..5d7b851 100644
--- a/src/liana/vcr.c
+++ b/src/liana/vcr.c
@@ -5,7 +5,7 @@
 
 #define VCR_BUFFER_INIT 64
 #define VCR_BUFFER_BUFFERED (VCR_BUFFER_INIT - 8)
-#define VCR_BUFFER_LOW (VCR_BUFFER_INIT - 16)
+#define VCR_BUFFER_LOW (VCR_BUFFER_INIT - 32)
 
 static void signal_callback(void *userdata)
 {
@@ -91,7 +91,7 @@ void lia_vcr_add_track(struct lia_vcr *vcr, struct lia_vcr_track *track)
     aki_cond_init(&track->cond);
     aki_mutex_init(&track->mutex);
     al_atomic_store(u8)(&track->buffered, 0, AL_ATOMIC_RELAXED);
-    aki_packet_cache_init(&track->cache, VCR_BUFFER_INIT - 1);
+    aki_packet_cache_init(&track->cache, VCR_BUFFER_INIT);
     al_array_push(vcr->tracks, track);
     al_atomic_store(s32)(&track->state, LIANA_STREAM_RUNNING, AL_ATOMIC_RELAXED);
 }
@@ -112,13 +112,17 @@ static void cork_if_buffered(struct lia_vcr *vcr)
     al_array_foreach(vcr->tracks, i, track) {
         buffered &= al_atomic_load(u8)(&track->buffered, AL_ATOMIC_RELAXED);
     }
-    if (buffered) aki_packet_stream_cork(vcr->data, true);
+    if (buffered) {
+        aki_packet_stream_cork(vcr->data, true);
+        al_array_foreach(vcr->tracks, i, track) {
+            aki_packet_cache_flush(&track->cache);
+        }
+    }
 }
 
 bool lia_vcr_push_packet(struct lia_vcr *vcr, struct aki_packet *packet)
 {
     struct lia_vcr_track *track;
-    s32 count;
     u8 op = aki_packet_read_u8(packet);
     switch (op) {
     case LIANA_PACKET_DATA:
@@ -134,9 +138,7 @@ bool lia_vcr_push_packet(struct lia_vcr *vcr, struct aki_packet *packet)
         }
         if (!aki_packet_cache_send_packet(&track->cache, packet)) {
             aki_packet_free(packet);
-        } else if ((count = al_atomic_add(s32)(&vcr->count, 1, AL_ATOMIC_RELAXED)) > vcr->mark.buffered) {
-            vcr->mark.buffered = count;
-            vcr->mark.low = vcr->mark.buffered - 16;
+        } else if (al_atomic_add(s32)(&vcr->count, 1, AL_ATOMIC_RELAXED) > vcr->mark.buffered) {
             cork_if_buffered(vcr);
         }
         break;