Server-side fixes from DIRECT_MODE testing

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

1 files changed, 102 insertions, 49 deletions

diff --git a/src/liana/list.c b/src/liana/list.c
index 01c2ff3..fc28c39 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -3,10 +3,11 @@
 #include <al/log.h>
 #include <al/lib.h>
 #include <al/random.h>
+#include <al/math.h>
 #include <nnwt/time.h>
+#include <nnwt/sort.h>
 
 #include "list.h"
-#include "list_cmp.h"
 
 enum {
     ADD_SINK = 0,
@@ -44,6 +45,12 @@ void lia_list_init(struct lia_list *list, str *name)
     list->cmd = NULL;
 }
 
+void lia_list_start_at(struct lia_list *list, s32 i)
+{
+    al_assert(list->current < 0);
+    list->current = -(i + 1);
+}
+
 // These small functions may seem excessive but their purpose is an attempt
 // to reduce noise in parts that are harder to understand.
 static inline void list_signal_meta(struct lia_list *list, struct lia_list_entry *entry, u8 meta)
@@ -165,6 +172,11 @@ static inline void sink_set_entry(struct lia_list_sink *sink, struct lia_list_en
     sink->callback(sink->userdata, LIANA_SINK_SET, entry, sequence, time);
 }
 
+static inline void sink_sequence_change(struct lia_list_sink *sink, struct lia_list_entry *entry, s32 sequence)
+{
+    sink->callback(sink->userdata, LIANA_SINK_SEQUENCE, entry, sequence, NULL);
+}
+
 static inline void sink_seek_entry(struct lia_list_sink *sink, struct lia_list_entry *entry, s32 sequence, struct lia_timing *time)
 {
     sink->callback(sink->userdata, LIANA_SINK_SEEK, entry, sequence, time);
@@ -181,14 +193,13 @@ static inline void sink_unset_entry(struct lia_list_sink *sink)
     sink->callback(sink->userdata, LIANA_SINK_UNSET, NULL, -1, NULL);
 }
 
-static bool list_set_current(struct lia_list *list, struct lia_list_entry *entry, s32 sequence)
+static void list_set_current(struct lia_list *list, struct lia_list_entry *entry, s32 sequence)
 {
     list->idle = false;
     entry_ref(list, entry);
     al_assert(list->current != sequence);
     list->current = sequence;
     list_signal_meta(list, entry, LIANA_META_CURRENT_CHANGED);
-    return true;
 }
 
 static void pump_queue(struct lia_list *list);
@@ -201,7 +212,7 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
         u8 pause;
         u64 at = LIANA_TIMESTAMP_INVALID;
         u64 pos = current->offset;
-        if (current->paused_at != LIANA_TIMESTAMP_INVALID) {
+        if (current->duration == 0 || current->paused_at != LIANA_TIMESTAMP_INVALID) {
             pause = LIANA_PAUSE_NONE;
         } else {
             at = nn_get_timestamp() + LIANA_BASE_DELAY;
@@ -210,11 +221,7 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
             } else {
                 at = current->start;
             }
-            // This sink could have an entry set from a connection we no longer
-            // know about. In that case skipping to this entry with a pause_and_swap_to()
-            // would be better. If this sink is empty that's still okay because
-            // PAUSE_BOTH is required to handle that case sink-side.
-            pause = LIANA_PAUSE_BOTH;
+            pause = LIANA_PAUSE_RESUME;
         }
         struct lia_timing time = {
             .at = at,
@@ -264,8 +271,8 @@ static void handle_remove_sink(struct lia_list *list, void *userdata)
 
 static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32 sequence)
 {
-    s32 size = (s32)list->entries.count;
-    if (sequence < 0 || sequence >= size) {
+    s32 count = (s32)list->entries.count;
+    if (sequence < 0 || sequence >= count) {
         return NULL;
     }
     return al_array_at(list->entries, sequence);
@@ -321,7 +328,7 @@ static u8 skipto_entry(struct lia_list_entry *current, struct lia_list_entry *ta
         }
     }
 
-    // Pause current to be resumed if it becomes the target of a skip (hold).
+    // Pause current to be resumed if it becomes the target of a skip (aka "hold" it).
     if (current->duration != 0 && current->paused_at == LIANA_TIMESTAMP_INVALID) {
         al_assert(current->start != LIANA_TIMESTAMP_INVALID);
         current->paused_at = at;
@@ -355,11 +362,28 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
     bool error;
     if (!entry_load_and_get_duration(list, target, index, &error)) {
         if (error) {
-            // On an error, sequence will address the same entry but index might not.
-            // entry_load_and_get_duration() may also adjust the current cmd's sequence,
-            // so use cmd->sequence here.
+            // On an error, sequence will always address the same entry but index might not.
+            // Keep index the same until sequence = index (search forward). Then, start
+            // decreasing index (search backward). entry_load_and_get_duration() may also
+            // adjust the current cmd's sequence, so use cmd->sequence here.
             struct lia_list_cmd *cmd = list->cmd;
-            return handle_skipto(list, cmd->sequence, index);
+            sequence = cmd->sequence;
+            if (sequence == LIANA_SEQUENCE_ANY) {
+                sequence = list->current;
+            }
+            if (sequence == index) {
+                if (index > 0) {
+                    index--;
+                } else {
+                    al_assert(sequence == 0);
+                    struct lia_list_sink *sink;
+                    al_array_foreach(list->sinks, i, sink) {
+                        sink_sequence_change(sink, current, sequence);
+                    }
+                    return true; // Current is now sequence 0.
+                }
+            }
+            return handle_skipto(list, sequence, index);
         }
         return false;
     }
@@ -377,6 +401,8 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
         .pause = pause
     };
 
+    list_set_current(list, target, index);
+
     struct lia_list_sink *sink;
     al_array_foreach(list->sinks, i, sink) {
         al_assert(sink->set != index);
@@ -384,19 +410,26 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
         sink_set_entry(sink, target, index, &time);
     }
 
-    return list_set_current(list, target, index);
+    return true;
 }
 
 static bool handle_skip(struct lia_list *list, s32 sequence, s32 n)
 {
-    if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
+    if (sequence == LIANA_SEQUENCE_ANY) {
+        sequence = list->current;
+    }
     if (sequence < 0) return true; // list->current = -1
-    return handle_skipto(list, sequence, sequence + n);
+    s32 i = sequence + n;
+    s32 max = (s32)list->entries.count - 1;
+    return handle_skipto(list, sequence, CLAMP(i, 0, max));
 }
 
 static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
 {
-    if (list->idle) {
+    s32 count = (s32)list->entries.count;
+    s32 start_at = abs(list->current) - 1;
+    bool on_start_index = list->current < 0 && count == start_at;
+    if (list->idle && on_start_index) {
         bool error;
         if (!entry_load_and_get_duration(list, entry, -1, &error)) {
             // We gave a sequence of -1 so, on error, don't add to list->entries.
@@ -405,21 +438,22 @@ static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
     }
     list_add_entry(list, entry);
     if (list->idle) {
-        if (list->current == -1) { // Start the list.
+        if (on_start_index) { // Start the list.
             entry->start = nn_get_timestamp() + LIANA_BASE_DELAY;
             struct lia_timing time = {
                 .at = entry->start,
                 .pos = entry->offset,
-                .pause = LIANA_PAUSE_RESUME
+                .pause = (entry->duration == 0) ? LIANA_PAUSE_NONE : LIANA_PAUSE_RESUME
             };
+            list_set_current(list, entry, start_at);
             struct lia_list_sink *sink;
             al_array_foreach(list->sinks, i, sink) {
                 al_assert(sink->set == -1);
                 sink->set = 0;
                 sink_set_entry(sink, entry, 0, &time);
             }
-            return list_set_current(list, entry, 0);
-        } else { // Skip to the added entry.
+            return true;
+        } else if (list->current >= 0) { // Skip to the added entry.
             // This is done via SKIPTO for consistency. Ended entries must still be held.
             struct lia_list_cmd *cmd = list->cmd;
             cmd->op = SKIPTO;
@@ -428,7 +462,8 @@ static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
             return handle_skipto(list, cmd->sequence, cmd->arg0.i);
         }
     }
-    al_assert(list->current != -1);
+    // count is the count before this entry was added.
+    al_assert(list->current >= 0 || count < start_at);
     return true;
 }
 
@@ -480,7 +515,8 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
     }
     }
 
-    log_trace("toggle_pause(#%u): pts: %f, pause: %hhu.", entry->id, pts, pause);
+    log_trace("toggle_pause(#%u): pts: %.4f, pause: %hhu.", entry->id,
+        (pause == LIANA_PAUSE_PAUSE) ? pts : NAN, pause);
 
     struct lia_timing time = {
         .at = at,
@@ -488,6 +524,8 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
         .pause = pause
     };
 
+    list_signal_meta(list, entry, LIANA_META_ENTRY_PAUSED);
+
     struct lia_list_sink *sink;
     al_array_foreach(list->sinks, i, sink) {
         if (sequence == list->current && sink->set != sequence) {
@@ -497,8 +535,6 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
             sink_pause_entry(sink, entry, sequence, &time);
         }
     }
-
-    list_signal_meta(list, entry, LIANA_META_ENTRY_PAUSED);
 }
 
 static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos)
@@ -542,6 +578,8 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos)
         .pause = pause
     };
 
+    list_signal_meta(list, entry, LIANA_META_ENTRY_SEEKED);
+
     struct lia_list_sink *sink;
     al_array_foreach(list->sinks, i, sink) {
         if (sequence == list->current && sink->set != sequence) {
@@ -551,8 +589,6 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos)
             sink_seek_entry(sink, entry, sequence, &time);
         }
     }
-
-    list_signal_meta(list, entry, LIANA_META_ENTRY_SEEKED);
 }
 
 static bool handle_end(struct lia_list *list, u32 id, u32 reset_token)
@@ -587,10 +623,10 @@ static bool handle_end(struct lia_list *list, u32 id, u32 reset_token)
     return true;
 #endif
 
-    s32 size = (s32)list->entries.count;
+    s32 count = (s32)list->entries.count;
     if (sequence == list->current) {
         s32 next = sequence + 1;
-        if (next < size) {
+        if (next < count) {
             struct lia_list_cmd *cmd = list->cmd;
             cmd->op = SKIPTO;
             cmd->sequence = sequence;
@@ -637,9 +673,9 @@ static bool handle_reverse(struct lia_list *list)
 {
     if (list->current < 0) return true;
     struct lia_list_entry *previous = al_array_at(list->entries, list->current);
-    u32 size = list->entries.count;
-    for (u32 i = 0; i < size; i++) {
-        u32 tail = size - (i + 1);
+    u32 count = list->entries.count;
+    for (u32 i = 0; i < count; i++) {
+        u32 tail = count - (i + 1);
         if (tail <= i) break;
         SWAP(al_array_at(list->entries, i), al_array_at(list->entries, tail));
     }
@@ -647,32 +683,39 @@ static bool handle_reverse(struct lia_list *list)
     return adjust_for_order_change(list, previous);
 }
 
+static s32 list_entry_compare(const void *a, const void *b)
+{
+    struct lia_list_entry *entry1 = *(struct lia_list_entry **)a;
+    struct lia_list_entry *entry2 = *(struct lia_list_entry **)b;
+    return nn_numerical_compare(&entry1->brief, &entry2->brief);
+}
+
 static bool handle_sort(struct lia_list *list)
 {
     if (list->current < 0) return true;
     struct lia_list_entry *previous = al_array_at(list->entries, list->current);
-    al_array_sort(list->entries, struct lia_list_entry *, camu_db_compare);
+    al_array_sort(list->entries, struct lia_list_entry *, list_entry_compare);
     log_trace("sort()");
     return adjust_for_order_change(list, previous);
 }
 
 static bool handle_shuffle(struct lia_list *list)
 {
-    u32 size = list->entries.count;
-    if (list->current < 0 || size < 2) return true;
+    u32 count = list->entries.count;
+    if (list->current < 0 || count < 2) return true;
     struct lia_list_entry *previous = al_array_at(list->entries, list->current);
     /* https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle
      for i from 0 to n−2 do
          j ← random integer such that i ≤ j ≤ n-1
          exchange a[i] and a[j]
     */
-    if (size == 2) {
+    if (count == 2) {
         if (al_random_int(0, 1) == 0) {
             SWAP(al_array_at(list->entries, 0), al_array_at(list->entries, 1));
         }
     } else {
-        for (u32 i = 0; i < size - 2; i++) {
-            u32 j = al_random_int(i, size - 1);
+        for (u32 i = 0; i < count - 2; i++) {
+            u32 j = al_random_int(i, count - 1);
             SWAP(al_array_at(list->entries, i), al_array_at(list->entries, j));
         }
     }
@@ -717,12 +760,14 @@ static void run_queue(struct lia_list *list)
         } else {
             return;
         }
+    } else {
+        return;
     }
     struct lia_list_cmd *cmd = list->cmd;
     switch (cmd->op) {
     case ADD_SINK:
         if (!handle_add_sink(list, cmd->sink)) {
-            return;
+            goto retry;
         }
         break;
     case REMOVE_SINK:
@@ -730,18 +775,18 @@ static void run_queue(struct lia_list *list)
         break;
     case ADD:
         if (!handle_add(list, cmd->entry)) {
-            return;
+            goto retry;
         }
         break;
     // Return on SKIPTO/SKIP: Target entry not loaded.
     case SKIPTO:
         if (!handle_skipto(list, cmd->sequence, cmd->arg0.i)) {
-            return;
+            goto retry;
         }
         break;
     case SKIP:
         if (!handle_skip(list, cmd->sequence, cmd->arg0.i)) {
-            return;
+            goto retry;
         }
         break;
     case TOGGLE_PAUSE:
@@ -753,23 +798,23 @@ static void run_queue(struct lia_list *list)
     case END:
         if (!handle_end(list, cmd->arg0.u, cmd->arg1.u)) {
             // Converted to a SKIP and target entry not loaded.
-            return;
+            goto retry;
         }
         break;
     // Return on order change: Converted to SKIPTO and new current not loaded.
     case REVERSE:
         if (!handle_reverse(list)) {
-            return;
+            goto retry;
         }
         break;
     case SORT:
         if (!handle_sort(list)) {
-            return;
+            goto retry;
         }
         break;
     case SHUFFLE:
         if (!handle_shuffle(list)) {
-            return;
+            goto retry;
         }
         break;
     case UNSET:
@@ -782,6 +827,14 @@ static void run_queue(struct lia_list *list)
     al_free(cmd);
     list->cmd = NULL;
     pump_queue(list);
+    return;
+retry:
+    list->cmd = NULL;
+    al_array_insert(list->command_queue, 0, cmd);
+    // We could have swallowed a pump_queue() meant for a different
+    // command while list->cmd was set. That's ok because it doesn't
+    // change anything about finishing this command, which will call
+    // pump_queue() once complete.
 }
 
 void pump_queue(struct lia_list *list)