summaryrefslogtreecommitdiff
path: root/src/liana
diff options
context:
space:
mode:
Diffstat (limited to 'src/liana')
-rw-r--r--src/liana/client.c4
-rw-r--r--src/liana/list.c200
-rw-r--r--src/liana/list.h1
-rw-r--r--src/liana/server.c8
-rw-r--r--src/liana/vcr.c10
-rw-r--r--src/liana/vcr.h4
6 files changed, 125 insertions, 102 deletions
diff --git a/src/liana/client.c b/src/liana/client.c
index 0edc262..59de366 100644
--- a/src/liana/client.c
+++ b/src/liana/client.c
@@ -143,7 +143,7 @@ static void packet_sent_callback(void *userdata, struct aki_packet *packet)
static void connection_callback(void *userdata, struct aki_packet_stream *stream)
{
struct lia_client *client = (struct lia_client *)userdata;
- lia_vcr_start(&client->vcr, client->loop);
+ lia_vcr_start(&client->vcr);
stream->packet_sent_callback = packet_sent_callback;
struct aki_packet *packet = aki_packet_create();
aki_packet_write_u16(packet, client->id);
@@ -192,7 +192,7 @@ void lia_client_connect(struct lia_client *client, struct aki_event_loop *loop,
client->pos = pos;
client->mask = 0;
client->reconnect = false;
- lia_vcr_init(&client->vcr, &client->data);
+ lia_vcr_init(&client->vcr, client->loop, &client->data);
al_str_clone(&client->addr, addr);
client->port = port;
if (!aki_packet_stream_init(&client->data, type, connection_callback, connection_closed_callback, client)) {
diff --git a/src/liana/list.c b/src/liana/list.c
index 94e3fa2..a192848 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -56,7 +56,7 @@ static void pump_queue(struct lia_list *list);
static bool assume_ended(struct lia_list_entry *entry, u64 at)
{
- if (entry->duration == 0) return true;
+ if (entry->ended || entry->duration == 0) return true;
if (entry->paused_at == LIANA_TIMESTAMP_INVALID && entry->start != LIANA_TIMESTAMP_INVALID) {
if (entry->offset > entry->duration) {
return true;
@@ -179,35 +179,42 @@ static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
return true;
}
+static void adjust_current(struct lia_list *list, struct lia_list_entry *previous)
+{
+ al_assert(list->current >= 0);
+ struct lia_list_cmd *cmd = list->cmd;
+ struct lia_list_entry *entry;
+ al_array_foreach(list->entries, i, entry) {
+ if (entry->opaque == previous->opaque) {
+ list->previous = -1;
+ if (i == (u32)list->current) return;
+ cmd->op = SKIPTO;
+ cmd->sequence = i;
+ cmd->i = list->current;
+ list->current = i;
+ break;
+ }
+ }
+ al_assert(cmd->op == SKIPTO && !entry->held);
+ pump_queue(list);
+}
-static void unset_all(struct lia_list *list, struct lia_list_entry *previous)
+static void unset_all(struct lia_list *list)
{
+ list->current = -1;
+ list->previous = -1;
+ list->queued = -1;
struct lia_list_sink *sink;
al_array_foreach(list->sinks, i, sink) {
sink->set = -1;
sink->queued = -1;
}
- if (previous) {
- list->cmd->op = SKIPTO;
- struct lia_list_entry *entry;
- al_array_foreach(list->entries, i, entry) {
- if (entry->opaque == previous->opaque) {
- list->cmd->i = list->current;
- list->current = i;
- list->cmd->sequence = i;
- break;
- }
- }
- pump_queue(list);
- }
}
static void handle_unset(struct lia_list *list)
{
- unset_all(list, NULL);
+ 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) {
@@ -223,31 +230,22 @@ static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32
}
// TODO:
-// - Think about what is means for an entry to be done. never put into a pause state?
-// - clock_end()??
-// - Sink needs to handle case where entry gets queued but the list already expects it to be playing
-// - It's possible to know if sink->current is done during a set command, synchronously.
-// So, check that when queueing an entry.
-// - Can clock be ended during a queue command in any other case?
-// - In the simplest case of our only operation being skip, how could client's become desynced?
-// - Then with toggle pause
-// - Is it safe to assert paused state on the client.
-// - Do queued
-// - Do seek
//if (current->start != LIANA_TIMESTAMP_INVALID && current->start > ts - LIANA_BASE_PING) {
static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
{
if (index == list->current) return true;
if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
- if (sequence == list->previous) {
- // This can happen but almost certainly won't be expected behavior.
+ if (sequence != list->current) {
+ // Skipping from an entry other than current is not handled and will cause very
+ // confusing errors. On top of likely resulting in unexpected behavior.
return true;
}
struct lia_list_entry *current = get_entry_from_sequence(list, sequence);
struct lia_list_entry *target = get_entry_from_sequence(list, index);
if (!current || !target) return true;
+ al_assert(current != target);
if (!entry_load_and_get_duration(list, target)) {
return false;
}
@@ -256,43 +254,51 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
u64 at = now + LIANA_BASE_DELAY;
u8 pause;
- // This should only happen if `start` has never been set.
- if (target->start == LIANA_TIMESTAMP_INVALID && target->paused_at == LIANA_TIMESTAMP_INVALID) {
- target->start = at;
- }
-
+ // Resume target if it's held.
if (target->held) {
+ al_assert(target->paused_at != LIANA_TIMESTAMP_INVALID);
target->paused_at = LIANA_TIMESTAMP_INVALID;
target->held = false;
}
al_assert(!current->held);
- bool ended = assume_ended(current, now);
+ // These are not equivalent to current/target->ended.
+ bool current_ended = assume_ended(current, now);
bool target_ended = assume_ended(target, now);
- al_log_info("list", "ended: %d, target_ended: %d", ended, target_ended);
- 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;
- }
+
+ // Current is ended or paused.
+ if (current_ended || current->paused_at != LIANA_TIMESTAMP_INVALID) {
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) {
+ // Swap entries and ignore their clocks.
+ pause = LIANA_PAUSE_NONE;
+ } else {
+ // Swap entries and resume target.
target->start = at;
- pause = !ended ? LIANA_PAUSE_BOTH : LIANA_PAUSE_RESUME;
+ pause = LIANA_PAUSE_RESUME;
}
- } else {
+ } else { // Current is playing.
if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) {
- pause = LIANA_PAUSE_NONE;
- } else { //if (target->paused_at == LIANA_TIMESTAMP_INVALID) {
+ // Pause-swap current, don't touch target's clock.
+ pause = LIANA_PAUSE_PAUSE;
+ } else {
+ // Pause-swap current and resume target.
target->start = at;
- pause = LIANA_PAUSE_RESUME;
+ pause = LIANA_PAUSE_BOTH;
}
}
+ // Pause current and set the held flag indicating it should be resumed
+ // if it becomes the target of a skip.
+ if (!current_ended && current->paused_at == LIANA_TIMESTAMP_INVALID) {
+ current->paused_at = at;
+ current->offset += current->paused_at - current->start;
+ current->start = LIANA_TIMESTAMP_INVALID;
+ current->held = true;
+ }
+
+ al_log_info("list", "pause: %u, held: %s.", pause, BOOLSTR(current->held));
+
list->current = index;
list->previous = sequence;
list->idle = false;
@@ -301,6 +307,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
.at = at,
.seek_pos = target->offset,
.pause = pause,
+ .previous_ended = current_ended,
.ended = target_ended
};
@@ -318,7 +325,11 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
static bool handle_skip(struct lia_list *list, s32 sequence, s32 n)
{
if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
- return handle_skipto(list, sequence, sequence + n);
+ struct lia_list_cmd *cmd = list->cmd;
+ cmd->op = SKIPTO;
+ cmd->sequence = sequence;
+ cmd->i = sequence + n;
+ return handle_skipto(list, cmd->sequence, cmd->i);
}
static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
@@ -383,45 +394,53 @@ static void handle_seek(struct lia_list *list, s32 sequence, f64 percent)
}
}
-static void handle_end(struct lia_list *list, s32 sequence)
+static bool handle_end(struct lia_list *list, s32 sequence)
{
al_assert(sequence != LIANA_SEQUENCE_ANY);
- if (sequence != list->current) return;
+
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) {
+ struct lia_list_entry *entry = al_array_at(list->entries, sequence);
+ if (entry->ended) {
al_log_warn("list", "Got end() from an already ended resource, ignoring.");
- return;
+ return true;
}
- current->offset = current->duration;
- current->ended = true;
- 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) {
- sink->queued = -1;
- }
- al_log_info("list", "Now playing: %ls.", AL_WSTR_PRINTF(&current->name));
- } else if (next < size) {
- struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
- cmd->op = SKIPTO;
- cmd->sequence = sequence;
- cmd->i = next;
- al_array_push(list->queue, cmd);
- } else {
- list->idle = true;
- struct lia_list_sink *sink;
- al_array_foreach(list->sinks, i, sink) {
- sink->set = -1;
+ entry->ended = true;
+ entry->offset = entry->duration;
+
+ if (sequence == list->current) {
+ s32 next = sequence + 1;
+ 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) {
+ sink->queued = -1;
+ }
+ struct lia_list_entry *current = al_array_at(list->entries, list->current);
+ al_log_info("list", "Now playing: %ls.", AL_WSTR_PRINTF(&current->name));
+ } else if (next < size) {
+ struct lia_list_cmd *cmd = list->cmd;
+ cmd->op = SKIPTO;
+ cmd->sequence = sequence;
+ cmd->i = next;
+ pump_queue(list);
+ return false;
+ } else {
+ list->idle = true;
+ struct lia_list_sink *sink;
+ al_array_foreach(list->sinks, i, sink) {
+ sink->set = -1;
+ }
}
}
+
+ return true;
}
static void handle_reverse(struct lia_list *list)
{
+ if (list->current == -1) return;
struct lia_list_entry *previous = al_array_at(list->entries, list->current);
u32 size = list->entries.size;
for (u32 i = 0; i < size; i++) {
@@ -429,18 +448,20 @@ static void handle_reverse(struct lia_list *list)
if (tail <= i) break;
SWAP(al_array_at(list->entries, i), al_array_at(list->entries, tail));
}
- unset_all(list, previous);
+ adjust_current(list, previous);
}
static void handle_sort(struct lia_list *list)
{
+ if (list->current == -1) return;
struct lia_list_entry *previous = al_array_at(list->entries, list->current);
al_array_sort(list->entries, struct lia_list_entry *, camu_db_compare);
- unset_all(list, previous);
+ adjust_current(list, previous);
}
static void handle_shuffle(struct lia_list *list)
{
+ if (list->current == -1) return;
struct lia_list_entry *previous = al_array_at(list->entries, list->current);
u32 size = list->entries.size;
if (size == 0) return;
@@ -448,21 +469,19 @@ static void handle_shuffle(struct lia_list *list)
u32 j = i + al_rand() / (AL_RAND_MAX / (size - i) + 1);
SWAP(al_array_at(list->entries, i), al_array_at(list->entries, j));
}
- unset_all(list, previous);
+ adjust_current(list, previous);
}
static void handle_clear(struct lia_list *list)
{
+ unset_all(list);
+ list->idle = true;
struct lia_list_entry *entry;
al_array_foreach(list->entries, i, entry) {
al_wstr_free(&entry->name);
al_free(entry);
}
list->entries.size = 0;
- unset_all(list, NULL);
- list->current = -1;
- list->previous = -1;
- list->idle = true;
}
void pump_queue(struct lia_list *list)
@@ -506,7 +525,10 @@ void pump_queue(struct lia_list *list)
handle_seek(list, cmd->sequence, cmd->f);
break;
case END:
- handle_end(list, cmd->sequence);
+ if (!handle_end(list, cmd->sequence)) {
+ // End was converted to a skip.
+ return;
+ }
break;
case REVERSE:
handle_reverse(list);
diff --git a/src/liana/list.h b/src/liana/list.h
index 7a24104..6443db6 100644
--- a/src/liana/list.h
+++ b/src/liana/list.h
@@ -45,6 +45,7 @@ struct lia_timing {
u64 at;
u64 seek_pos;
u8 pause;
+ bool previous_ended;
bool ended;
};
diff --git a/src/liana/server.c b/src/liana/server.c
index 93ee643..e03b8ef 100644
--- a/src/liana/server.c
+++ b/src/liana/server.c
@@ -237,8 +237,8 @@ static void packet_callback(void *userdata, struct aki_packet_stream *stream, st
conn->node = node;
conn->stream = stream;
conn->packet = packet;
- aki_signal_init(&conn->signal, signal_callback, conn);
- aki_signal_start(&conn->signal, server->loop);
+ aki_signal_init(&conn->signal, server->loop, signal_callback, conn);
+ aki_signal_start(&conn->signal);
cch_entry_get_handle(node->entry, &conn->handle);
conn->handler = lia_handler_by_name(cch_entry_get_liana(node->entry))->create_server_handler();
conn->errored = false;
@@ -315,8 +315,8 @@ static void duration_signal_callback(void *userdata)
void lia_node_get_duration(struct lia_node *node)
{
- aki_signal_init(&node->signal, duration_signal_callback, node);
- aki_signal_start(&node->signal, node->server->loop);
+ aki_signal_init(&node->signal, node->server->loop, duration_signal_callback, node);
+ aki_signal_start(&node->signal);
cch_entry_get_handle(node->entry, &node->handle);
node->handler = lia_handler_by_name(cch_entry_get_liana(node->entry))->create_server_handler();
aki_thread_create(&node->thread, init_duration_thread, node);
diff --git a/src/liana/vcr.c b/src/liana/vcr.c
index 571fe4e..435fcba 100644
--- a/src/liana/vcr.c
+++ b/src/liana/vcr.c
@@ -3,7 +3,7 @@
#include "vcr.h"
#include "handler.h"
-#define VCR_BUFFER_BUFFERED MB(12)
+#define VCR_BUFFER_BUFFERED MB(8)
enum {
VCR_EXPAND_UNTOUCHED = 0,
@@ -28,7 +28,7 @@ static void reset_metrics(struct lia_vcr *vcr)
vcr->metric.last_report_ts = 0;
}
-void lia_vcr_init(struct lia_vcr *vcr, struct aki_packet_stream *data)
+void lia_vcr_init(struct lia_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data)
{
al_array_init(vcr->tracks);
al_atomic_store(u64)(&vcr->count, 0, AL_ATOMIC_RELAXED);
@@ -36,13 +36,13 @@ void lia_vcr_init(struct lia_vcr *vcr, struct aki_packet_stream *data)
vcr->mark.low = 0;
vcr->expand = VCR_EXPAND_UNTOUCHED;
vcr->data = data;
- aki_signal_init(&vcr->signal, signal_callback, vcr);
+ aki_signal_init(&vcr->signal, loop, signal_callback, vcr);
reset_metrics(vcr);
}
-void lia_vcr_start(struct lia_vcr *vcr, struct aki_event_loop *loop)
+void lia_vcr_start(struct lia_vcr *vcr)
{
- aki_signal_start(&vcr->signal, loop);
+ aki_signal_start(&vcr->signal);
}
static aki_thread_result AKI_THREADCALL vcr_track_thread(void *userdata)
diff --git a/src/liana/vcr.h b/src/liana/vcr.h
index fc7fb32..8041d96 100644
--- a/src/liana/vcr.h
+++ b/src/liana/vcr.h
@@ -40,8 +40,8 @@ struct lia_vcr {
} metric;
};
-void lia_vcr_init(struct lia_vcr *vcr, struct aki_packet_stream *data);
-void lia_vcr_start(struct lia_vcr *vcr, struct aki_event_loop *loop);
+void lia_vcr_init(struct lia_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data);
+void lia_vcr_start(struct lia_vcr *vcr);
void lia_vcr_add_track(struct lia_vcr *vcr, struct lia_vcr_track *track);
bool lia_vcr_is_empty(struct lia_vcr *vcr);
void lia_vcr_push_packet(struct lia_vcr *vcr, struct aki_packet *packet);
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 19:40:24 +0000