More synced list, another vcr fix

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

4 files changed, 85 insertions, 65 deletions

diff --git a/src/liana/list.c b/src/liana/list.c
index a8d5f71..9698497 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -28,7 +28,7 @@ static void buffer_ahead(struct lia_list *list)
             struct lia_list_entry *buffered = al_array_at(list->entries, i);
             struct lia_list_sink *sink;
             al_array_foreach(list->sinks, i, sink) {
-                sink->callback(sink->userdata, CAMU_SINK_BUFFER, buffered, i, LIANA_TIMESTAMP_INVALID);
+                sink->callback(sink->userdata, LIANA_SINK_BUFFER, buffered, i, LIANA_TIMESTAMP_INVALID);
             }
         }
     }
@@ -44,6 +44,21 @@ static void unset_all(struct lia_list *list)
     }
 }
 
+static bool assume_ended(struct lia_list_entry *entry, u64 at)
+{
+    if (entry->duration == 0) return true;
+    if (entry->paused_at == LIANA_TIMESTAMP_INVALID && entry->start != LIANA_TIMESTAMP_INVALID) {
+        if (entry->offset > entry->duration) {
+            return true;
+        }
+        if (at < entry->start) return false;
+        if (at - entry->start >= entry->duration - entry->offset) {
+            return true;
+        }
+    }
+    return false;
+}
+
 void lia_list_add_sink(struct lia_list *list, void (*callback)(void *, u8, struct lia_list_entry *, s32, struct lia_timing *), void *userdata)
 {
     struct lia_list_sink *sink = al_alloc_object(struct lia_list_sink);
@@ -58,10 +73,9 @@ void lia_list_add_sink(struct lia_list *list, void (*callback)(void *, u8, struc
         u64 at = LIANA_TIMESTAMP_INVALID;
         u64 seek_pos = current->offset;
         if (current->paused_at == LIANA_TIMESTAMP_INVALID) {
-            now += LIANA_BASE_DELAY;
-            if (now > current->start && now - current->start > LIANA_BASE_DELAY) {
-                at = now;
-                seek_pos += now - current->start;
+            at = now + LIANA_BASE_DELAY;
+            if (at > current->start && at - current->start > LIANA_BASE_DELAY) {
+                seek_pos += at - current->start;
             } else {
                 at = current->start;
             }
@@ -72,28 +86,16 @@ void lia_list_add_sink(struct lia_list *list, void (*callback)(void *, u8, struc
         struct lia_timing time = {
             .at = at,
             .seek_pos = seek_pos,
-            .pause = pause
+            .pause = pause,
+            .ended = assume_ended(current, now)
         };
-        sink->callback(sink->userdata, CAMU_SINK_SET, current, list->current, &time);
+        sink->callback(sink->userdata, LIANA_SINK_SET, current, list->current, &time);
     } else {
         sink->set = -1;
     }
     sink->queued = -1;
 }
 
-void lia_list_unset(struct lia_list *list)
-{
-    unset_all(list);
-    list->current = list->entries.size - 1;
-    list->previous = -1;
-    list->queued = -1;
-    list->idle = true;
-    struct lia_list_sink *sink;
-    al_array_foreach(list->sinks, i, sink) {
-        sink->callback(sink->userdata, CAMU_SINK_CLEAR, NULL, -1, NULL);
-    }
-}
-
 void lia_list_remove_sink(struct lia_list *list, void *userdata)
 {
     struct lia_list_sink *sink;
@@ -113,6 +115,7 @@ void lia_list_add(struct lia_list *list, void *opaque, u64 duration, str *name)
     entry->paused_at = LIANA_TIMESTAMP_INVALID;
     entry->held = false;
     entry->offset = 0;
+    entry->ended = false;
     entry->duration = duration;
     al_str_clone(&entry->name, name);
     al_array_push(list->entries, entry);
@@ -123,13 +126,14 @@ void lia_list_add(struct lia_list *list, void *opaque, u64 duration, str *name)
         struct lia_timing time = {
             .at = entry->start,
             .seek_pos = entry->offset,
-            .pause = LIANA_PAUSE_RESUME
+            .pause = LIANA_PAUSE_RESUME,
+            .ended = false
         };
         struct lia_list_sink *sink;
         al_array_foreach(list->sinks, i, sink) {
             al_assert(sink->set == -1);
             sink->set = list->current;
-            sink->callback(sink->userdata, CAMU_SINK_SET, entry, list->current, &time);
+            sink->callback(sink->userdata, LIANA_SINK_SET, entry, list->current, &time);
         }
         if (list->callback) list->callback(list->userdata, LIANA_META_PLAYING, entry);
     } else {
@@ -146,7 +150,7 @@ void lia_list_add(struct lia_list *list, void *opaque, u64 duration, str *name)
             struct lia_list_sink *sink;
             al_array_foreach(list->sinks, i, sink) {
                 sink->queued = list->queued;
-                sink->callback(sink->userdata, CAMU_SINK_BUFFER_AND_QUEUE, entry, list->queued, &time);
+                sink->callback(sink->userdata, LIANA_SINK_BUFFER_AND_QUEUE, entry, list->queued, &time);
             }
         } else {
         */
@@ -156,6 +160,19 @@ void lia_list_add(struct lia_list *list, void *opaque, u64 duration, str *name)
     }
 }
 
+void lia_list_unset(struct lia_list *list)
+{
+    unset_all(list);
+    list->current = list->entries.size - 1;
+    list->previous = -1;
+    list->queued = -1;
+    list->idle = true;
+    struct lia_list_sink *sink;
+    al_array_foreach(list->sinks, i, sink) {
+        sink->callback(sink->userdata, LIANA_SINK_UNSET, NULL, -1, NULL);
+    }
+}
+
 static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32 sequence)
 {
     s32 size = (s32)list->entries.size;
@@ -163,20 +180,6 @@ static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32
     return al_array_at(list->entries, sequence);
 }
 
-static bool assume_done(struct lia_list_entry *entry, u64 at)
-{
-    if (entry->duration == 0) return true;
-    if (entry->paused_at == LIANA_TIMESTAMP_INVALID && entry->start != LIANA_TIMESTAMP_INVALID) {
-        if (entry->offset > entry->duration) {
-            return true;
-        }
-        if (at < entry->start) return false;
-        if (at - entry->start >= entry->duration - entry->offset) {
-            return true;
-        }
-    }
-    return false;
-}
 
 void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
 {
@@ -222,29 +225,28 @@ void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
 
     al_assert(!current->held);
 
-    bool done = assume_done(current, now);
-    if (done || current->paused_at == LIANA_TIMESTAMP_INVALID) {
-        if (!done) {
+    bool ended = assume_ended(current, now);
+    bool target_ended = assume_ended(target, now);
+    if (ended || current->paused_at == LIANA_TIMESTAMP_INVALID) {
+        if (!ended) {
             current->paused_at = at;
             current->offset += current->paused_at - current->start;
             current->start = LIANA_TIMESTAMP_INVALID;
             current->held = true;
         }
-        if (assume_done(target, now) || target->paused_at != LIANA_TIMESTAMP_INVALID) {
-            pause = !done ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_NONE;
+        if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) {
+            pause = !ended ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_NONE;
         } else { //if (target->paused_at == LIANA_TIMESTAMP_INVALID) {
             target->start = at;
-            pause = !done ? LIANA_PAUSE_BOTH : LIANA_PAUSE_RESUME;
+            pause = !ended ? LIANA_PAUSE_BOTH : LIANA_PAUSE_RESUME;
         }
-        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) {
+        if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) {
             pause = LIANA_PAUSE_NONE;
         } else { //if (target->paused_at == LIANA_TIMESTAMP_INVALID) {
             target->start = at;
             pause = LIANA_PAUSE_RESUME;
         }
-        al_log_debug("list", "%d: paused, pause: %d", sequence, pause);
     }
 
     list->current = index;
@@ -254,13 +256,14 @@ void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
     struct lia_timing time = {
         .at = at,
         .seek_pos = target->offset,
-        .pause = pause
+        .pause = pause,
+        .ended = target_ended
     };
 
     struct lia_list_sink *sink;
     al_array_foreach(list->sinks, i, sink) {
         sink->set = index;
-        sink->callback(sink->userdata, CAMU_SINK_SET, target, index, &time);
+        sink->callback(sink->userdata, LIANA_SINK_SET, target, index, &time);
     }
 
     list->callback(list->userdata, LIANA_META_PLAYING, target);
@@ -299,12 +302,13 @@ void lia_list_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
     struct lia_timing time = {
         .at = at,
         .seek_pos = LIANA_TIMESTAMP_INVALID,
-        .pause = pause
+        .pause = pause,
+        .ended = false
     };
 
     struct lia_list_sink *sink;
     al_array_foreach(list->sinks, i, sink) {
-        sink->callback(sink->userdata, CAMU_SINK_PAUSE, current, sequence, &time);
+        sink->callback(sink->userdata, LIANA_SINK_PAUSE, current, sequence, &time);
     }
 }
 
@@ -317,14 +321,16 @@ void lia_list_seek(struct lia_list *list, s32 sequence, f64 percent)
     u64 now = aki_get_timestamp();
     current->offset = pos;
     current->start = now + LIANA_BASE_DELAY;
+    current->ended = false;
     struct lia_timing time = {
         .at = current->start,
         .seek_pos = current->offset,
-        .pause = LIANA_PAUSE_NONE
+        .pause = LIANA_PAUSE_NONE,
+        .ended = false
     };
     struct lia_list_sink *sink;
     al_array_foreach(list->sinks, i, sink) {
-        sink->callback(sink->userdata, CAMU_SINK_SEEK, current, sequence, &time);
+        sink->callback(sink->userdata, LIANA_SINK_SEEK, current, sequence, &time);
     }
 }
 
@@ -335,7 +341,12 @@ void lia_list_end(struct lia_list *list, s32 sequence)
     s32 size = (s32)list->entries.size;
     s32 next = sequence + 1;
     struct lia_list_entry *current = al_array_at(list->entries, list->current);
+    if (current->ended) {
+        al_log_warn("list", "Got end() from an already ended resource, ignoring.");
+        return;
+    }
     current->offset = current->duration;
+    current->ended = true;
     if (list->queued >= 0) {
         list->current = list->queued;
         list->previous = sequence;
diff --git a/src/liana/list.h b/src/liana/list.h
index eb9a3cb..bbb2ce4 100644
--- a/src/liana/list.h
+++ b/src/liana/list.h
@@ -7,17 +7,25 @@
 #define LIANA_SEQUENCE_ANY -1
 #define LIANA_TIMESTAMP_INVALID UINT64_MAX
 
-#define LIANA_BASE_PING 200000Lu // 100ms
-#define LIANA_BASE_DELAY 450000Lu // 250ms
+#define LIANA_BASE_DELAY 475000Lu // 475ms
+#define LIANA_BASE_PING 200000Lu // 200ms
 #define LIANA_PAUSE_DELAY LIANA_BASE_PING
 #define LIANA_DELAY_IGNORE 0Lu
 
 #define LIANA_BUFFER_AHEAD 2
 
 enum {
-    LIANA_ENTRY_DURATION = 0,
-    LIANA_ENTRY_ID,
-    LIANA_ENTRY_SKIP
+    LIANA_SINK_SET = 0,
+    LIANA_SINK_UNSET,
+    LIANA_SINK_BUFFER,
+    LIANA_SINK_BUFFER_AND_QUEUE,
+    LIANA_SINK_PAUSE,
+    LIANA_SINK_SEEK
+};
+
+enum {
+    LIANA_LOAD_ENTRY = 0,
+    LIANA_UNLOAD_ENTRY
 };
 
 enum {
@@ -41,6 +49,7 @@ struct lia_timing {
     u64 at;
     u64 seek_pos;
     u8 pause;
+    bool ended;
 };
 
 struct lia_list_entry {
@@ -49,6 +58,7 @@ struct lia_list_entry {
     u64 paused_at;
     bool held;
     u64 offset;
+    bool ended;
     u64 duration;
     str name;
 };
diff --git a/src/liana/server.c b/src/liana/server.c
index 1fc96fb..3eb303e 100644
--- a/src/liana/server.c
+++ b/src/liana/server.c
@@ -164,7 +164,7 @@ static void signal_callback(void *userdata)
     }
     if (!conn->errored) {
         conn->id = al_inc_u16();
-        aki_packet_pool_init(&conn->pool, 48, server->loop, packet_pool_callback, conn);
+        aki_packet_pool_init(&conn->pool, 96, server->loop, packet_pool_callback, conn);
         al_array_push(node->connections, conn);
         handle_connection(conn, packet);
     } else {
diff --git a/src/liana/vcr.c b/src/liana/vcr.c
index 5d7b851..192b4f1 100644
--- a/src/liana/vcr.c
+++ b/src/liana/vcr.c
@@ -3,9 +3,9 @@
 #include "vcr.h"
 #include "handler.h"
 
-#define VCR_BUFFER_INIT 64
-#define VCR_BUFFER_BUFFERED (VCR_BUFFER_INIT - 8)
-#define VCR_BUFFER_LOW (VCR_BUFFER_INIT - 32)
+#define VCR_BUFFER_INIT 256
+#define VCR_BUFFER_BUFFERED (VCR_BUFFER_INIT - 16)
+#define VCR_BUFFER_LOW (VCR_BUFFER_INIT - 64)
 
 static void signal_callback(void *userdata)
 {
@@ -47,7 +47,7 @@ static aki_thread_result AKI_THREADCALL vcr_track_thread(void *userdata)
                 }
                 u8 buffered = al_atomic_load(u8)(&track->buffered, AL_ATOMIC_RELAXED);
                 s32 count = al_atomic_sub(s32)(&vcr->count, 1, AL_ATOMIC_RELAXED);
-                if (count < vcr->mark.low && buffered) {
+                if (count <= vcr->mark.low && buffered) {
                     aki_signal_send(&vcr->signal);
                 }
             }
@@ -167,9 +167,8 @@ void lia_vcr_cork(struct lia_vcr_track *track)
 void lia_vcr_uncork(struct lia_vcr_track *track)
 {
     if (al_atomic_load(s32)(&track->state, AL_ATOMIC_RELAXED) != LIANA_STREAM_STOPPED) {
-        // This can be reached during normal operation. Whether or not that should
-        // be allowed is up for consideration.
-        return;
+        // This can be reached during normal operation. Whether or not that makes
+        // sense is up for consideration.
     }
     al_atomic_store(s32)(&track->state, LIANA_STREAM_RUNNING, AL_ATOMIC_RELAXED);
     aki_mutex_lock(&track->mutex);