From 8f208c26b6fa1a9f3372679c047cab559c06e26b Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Mon, 14 Sep 2026 08:57:42 -0400 Subject: Server-side fixes from DIRECT_MODE testing Signed-off-by: Andrew Opalach --- src/liana/list.c | 151 +++++++++++++++++++++++++++++++++++++------------------ 1 file changed, 102 insertions(+), 49 deletions(-) (limited to 'src/liana/list.c') 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 #include #include +#include #include +#include #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) -- cgit v1.2.3-101-g0448