From 692785bc9da6904cf17e986fb034730ed3d78231 Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Sat, 14 Dec 2024 11:13:38 -0500 Subject: Work on pause and seek Signed-off-by: Andrew Opalach --- flake.lock | 18 +-- scripts/run_server.sh | 3 +- scripts/screenrc | 1 + src/buffer/clock.c | 19 +++ src/buffer/clock.h | 5 +- src/liana/client.c | 39 ++++-- src/liana/client.h | 6 +- src/liana/handler.h | 1 + src/liana/list.c | 138 ++++++++++++++------ src/liana/list.h | 2 + src/liana/server.c | 16 ++- src/liana/server.h | 6 +- src/libsink/sink.c | 311 ++++++++++++++++++++++++++------------------- src/libsink/sink.h | 7 +- src/mixer/mixer.c | 3 +- src/server/server.c | 6 +- src/sink/input_simulator.c | 6 +- 17 files changed, 372 insertions(+), 215 deletions(-) diff --git a/flake.lock b/flake.lock index 5111b7d..ec27cac 100644 --- a/flake.lock +++ b/flake.lock @@ -41,11 +41,11 @@ ] }, "locked": { - "lastModified": 1733769654, - "narHash": "sha256-aVvYDt8eitZVF6fdOrSoIzYRkQ5Gh6kfRvqkiaDRLL0=", + "lastModified": 1734093295, + "narHash": "sha256-hSwgGpcZtdDsk1dnzA0xj5cNaHgN9A99hRF/mxMtwS4=", "owner": "nix-community", "repo": "home-manager", - "rev": "e952e94955dcc6fa2120c1430789fc41363f5237", + "rev": "66c5d8b62818ec4c1edb3e941f55ef78df8141a8", "type": "github" }, "original": { @@ -57,11 +57,11 @@ }, "nixos-hardware": { "locked": { - "lastModified": 1733481457, - "narHash": "sha256-IS3bxa4N1VMSh3/P6vhEAHQZecQ3oAlKCDvzCQSO5Is=", + "lastModified": 1733861262, + "narHash": "sha256-+jjPup/ByS0LEVIrBbt7FnGugJgLeG9oc+ivFASYn2U=", "owner": "NixOS", "repo": "nixos-hardware", - "rev": "e563803af3526852b6b1d77107a81908c66a9fcf", + "rev": "cf737e2eba82b603f54f71b10cb8fd09d22ce3f5", "type": "github" }, "original": { @@ -110,11 +110,11 @@ }, "nixpkgs_2": { "locked": { - "lastModified": 1733759999, - "narHash": "sha256-463SNPWmz46iLzJKRzO3Q2b0Aurff3U1n0nYItxq7jU=", + "lastModified": 1733940404, + "narHash": "sha256-Pj39hSoUA86ZePPF/UXiYHHM7hMIkios8TYG29kQT4g=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "a73246e2eef4c6ed172979932bc80e1404ba2d56", + "rev": "5d67ea6b4b63378b9c13be21e2ec9d1afc921713", "type": "github" }, "original": { diff --git a/scripts/run_server.sh b/scripts/run_server.sh index c2f74e5..328577a 100755 --- a/scripts/run_server.sh +++ b/scripts/run_server.sh @@ -4,7 +4,8 @@ export PYTHONDONTWRITEBYTECODE=1 export PYTHONPATH=$HOME/c/camu/src/portal/vendor/vendor # screen will use $SHELL. screen -c ../scripts/screenrc -#./src/fruits/cmsrv/cmsrv $@ +#valgrind --leak-check=no --show-error-list=yes ./src/fruits/cmsrv/cmsrv $@ #valgrind --log-file=./server-valgrind.log ./src/fruits/cmsrv/cmsrv $@ +#./src/fruits/cmsrv/cmsrv $@ #cpulimit -l 1 ./src/fruits/cmsrv/cmsrv $@ #gdb -ex run --args ./src/fruits/cmsrv/cmsrv $@ diff --git a/scripts/screenrc b/scripts/screenrc index 925703d..6381a9e 100644 --- a/scripts/screenrc +++ b/scripts/screenrc @@ -4,6 +4,7 @@ mousetrack on defmousetrack on screen -t 'cmsrv' ./src/fruits/cmsrv/cmsrv split +resize -v 75% focus screen -t 'cmsrv_debug' stuff 'sleep 1 && gdb -p $(pgrep cmsrv) -ex cont ' diff --git a/src/buffer/clock.c b/src/buffer/clock.c index 705a525..624c534 100644 --- a/src/buffer/clock.c +++ b/src/buffer/clock.c @@ -27,6 +27,25 @@ void camu_clock_set(struct camu_clock *clock, f64 base) clock->paused_at = 0.0; } +void camu_clock_seek(struct camu_clock *clock, f64 base, u64 target) +{ + clock->base = base; + clock->offset = 0.0; + f64 tick = aki_get_tick(); + if (clock->paused_at == -1.0) { + // We are safe to directly edit the tick here. + if (target == 0) { + al_atomic_store(f64)(&clock->tick, -1.0, AL_ATOMIC_RELAXED); + } else { + tick = calc_tick_offset(tick, aki_get_timestamp(), target); + al_atomic_store(f64)(&clock->tick, tick, AL_ATOMIC_RELAXED); + } + } else { + al_atomic_store(f64)(&clock->pause, PAUSED, AL_ATOMIC_RELAXED); + clock->paused_at = 0.0; + } +} + void camu_clock_offset(struct camu_clock *clock, f64 amount) { clock->offset += amount; diff --git a/src/buffer/clock.h b/src/buffer/clock.h index aae87cc..86eba5e 100644 --- a/src/buffer/clock.h +++ b/src/buffer/clock.h @@ -39,10 +39,11 @@ struct camu_clock { void camu_clock_init(struct camu_clock *clock, void (*callback)(void *, u8), void *userdata); void camu_clock_set(struct camu_clock *clock, f64 base); +void camu_clock_seek(struct camu_clock *clock, f64 base, u64 target); void camu_clock_offset(struct camu_clock *clock, f64 amount); -void camu_clock_pause(struct camu_clock *clock, u64 ts); -void camu_clock_resume(struct camu_clock *clock, u64 ts); +void camu_clock_pause(struct camu_clock *clock, u64 target); +void camu_clock_resume(struct camu_clock *clock, u64 target); bool camu_clock_is_armed(struct camu_clock *clock); bool camu_clock_is_paused(struct camu_clock *clock); diff --git a/src/liana/client.c b/src/liana/client.c index 1454ba0..a841f7a 100644 --- a/src/liana/client.c +++ b/src/liana/client.c @@ -1,3 +1,5 @@ +#include + #include "../server/common.h" #ifdef CAMU_HAVE_FFMPEG #include "../codec/ffmpeg/packet_ext.h" @@ -121,7 +123,7 @@ static void parse_info_packet(struct lia_client *client, struct aki_packet *pack static void info_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) { struct lia_client *client = (struct lia_client *)userdata; - client->connection_id = aki_packet_read_u16(packet); + client->connection_id = aki_packet_read_u32(packet); parse_info_packet(client, packet); aki_packet_free(packet); if (client->mask == 0 || lia_vcr_is_empty(&client->vcr)) { @@ -144,11 +146,27 @@ 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; + if (client->reconnect) { + // Even if seek() was called before the initial connection_callback(), + // we still want to call RESUME_AT here. + struct lia_timing time = { + .at = client->at, + .seek_pos = client->pos, + .pause = LIANA_PAUSE_NONE + }; + client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time); + if (client->mask == 0) { + al_log_warn("liana", "Handling reconnect on unconfigured client."); + } else { + client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, NULL); + } + client->reconnect = false; + } 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); - aki_packet_write_u16(packet, 0); + aki_packet_write_u32(packet, client->id); + aki_packet_write_u32(packet, 0); aki_packet_write_s32(packet, client->mask); aki_packet_write_u64(packet, client->pos); if (client->mask == 0) { @@ -172,13 +190,6 @@ static void connection_closed_callback(void *userdata, struct aki_packet_stream // any unexpected behavior. client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &client->reconnect); if (client->reconnect) { - client->reconnect = false; - struct lia_timing time = { - .at = client->at, - .seek_pos = client->pos, - .pause = LIANA_PAUSE_NONE - }; - client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time); aki_packet_stream_reconnect(stream, &client->addr, client->port); } else { client->callback(client->userdata, LIANA_CLIENT_CLOSED, NULL, NULL); @@ -186,7 +197,7 @@ static void connection_closed_callback(void *userdata, struct aki_packet_stream } void lia_client_connect(struct lia_client *client, struct aki_event_loop *loop, u8 type, - str *addr, u16 port, u16 id, u64 pos, struct camu_renderer *renderer) + str *addr, u16 port, u32 id, u64 pos, struct camu_renderer *renderer) { client->loop = loop; client->id = id; @@ -211,10 +222,12 @@ void lia_client_set_renderer(struct lia_client *client, struct camu_renderer *re void lia_client_seek(struct lia_client *client, u64 pos, u64 at) { - client->reconnect = true; client->pos = pos; client->at = at; - aki_packet_stream_disconnect(&client->data); + if (!client->reconnect) { + client->reconnect = true; + aki_packet_stream_disconnect(&client->data); + } } void lia_client_reseek(struct lia_client *client) diff --git a/src/liana/client.h b/src/liana/client.h index 24526af..cd1297e 100644 --- a/src/liana/client.h +++ b/src/liana/client.h @@ -8,14 +8,14 @@ struct lia_client { struct aki_event_loop *loop; - u16 id; + u32 id; s32 mask; u64 pos; u64 at; bool reconnect; str addr; u16 port; - u16 connection_id; + u32 connection_id; struct aki_packet_stream data; u64 duration; struct lia_vcr vcr; @@ -25,7 +25,7 @@ struct lia_client { }; void lia_client_connect(struct lia_client *client, struct aki_event_loop *loop, u8 type, - str *addr, u16 port, u16 id, u64 pos, struct camu_renderer *renderer); + str *addr, u16 port, u32 id, u64 pos, struct camu_renderer *renderer); void lia_client_seek(struct lia_client *client, u64 pos, u64 at); void lia_client_reseek(struct lia_client *client); void lia_client_disconnect(struct lia_client *client); diff --git a/src/liana/handler.h b/src/liana/handler.h index 84aa1de..21437c4 100644 --- a/src/liana/handler.h +++ b/src/liana/handler.h @@ -29,6 +29,7 @@ enum { LIANA_CLIENT_SUBTITLE, LIANA_CLIENT_REMOVE_BUFFERS, LIANA_CLIENT_RESUME_AT, + LIANA_CLIENT_RECONNECTED, LIANA_CLIENT_EOF, LIANA_CLIENT_CLOSED }; diff --git a/src/liana/list.c b/src/liana/list.c index 059cd07..23eabc3 100644 --- a/src/liana/list.c +++ b/src/liana/list.c @@ -10,13 +10,19 @@ static void buffer_ahead(struct lia_list *list) { s32 size = (s32)list->entries.size; - if (list->queued >= 0 && list->queued + 1 < size) { - s32 ahead = list->queued + 1; + if (list->current >= 0 && list->current + 1 < size) { + s32 ahead = list->current + 1; for (s32 i = ahead; i < MIN(ahead + LIANA_BUFFER_AHEAD, size); i++) { - struct lia_list_entry *buffered = al_array_at(list->entries, i); + struct lia_list_entry *entry = al_array_at(list->entries, i); + struct lia_timing time = { + .at = LIANA_TIMESTAMP_INVALID, + .seek_pos = entry->offset, + .pause = LIANA_PAUSE_NONE, + .ended = false + }; struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { - sink->callback(sink->userdata, LIANA_SINK_BUFFER, buffered, i, LIANA_TIMESTAMP_INVALID); + sink->callback(sink->userdata, LIANA_SINK_BUFFER, entry, i, &time); } } } @@ -46,6 +52,7 @@ void lia_list_init(struct lia_list *list, str *name) list->previous = -1; list->queued = -1; list->idle = true; + list->increment = 0; al_array_init(list->entries); al_array_init(list->sinks); al_array_init(list->queue); @@ -56,6 +63,7 @@ static void pump_queue(struct lia_list *list); static bool assume_ended(struct lia_list_entry *entry, u64 at) { + if (entry->duration == LIANA_TIMESTAMP_INVALID) return false; 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) { @@ -92,7 +100,13 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) u8 pause; u64 at = LIANA_TIMESTAMP_INVALID; u64 seek_pos = current->offset; - if (current->paused_at == LIANA_TIMESTAMP_INVALID) { + bool ended = assume_ended(current, now); + if (ended && current->duration != LIANA_TIMESTAMP_INVALID) { + seek_pos = current->duration; + } + if (ended || current->paused_at != LIANA_TIMESTAMP_INVALID) { + pause = LIANA_PAUSE_NONE; + } else { at = now + LIANA_BASE_DELAY; if (at > current->start && at - current->start > LIANA_BASE_DELAY) { seek_pos += at - current->start; @@ -100,14 +114,12 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) at = current->start; } pause = LIANA_PAUSE_RESUME; - } else { - pause = LIANA_PAUSE_NONE; } struct lia_timing time = { .at = at, .seek_pos = seek_pos, .pause = pause, - .ended = assume_ended(current, now) + .ended = ended }; sink->callback(sink->userdata, LIANA_SINK_SET, current, list->current, &time); } else { @@ -236,6 +248,7 @@ 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 < 0) return true; 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. @@ -244,7 +257,8 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) 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 && !current->held); + if (!target) return true; al_assert(current != target); if (!entry_load_and_get_duration(list, target)) { return false; @@ -258,23 +272,30 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) if (target->held) { al_assert(target->paused_at != LIANA_TIMESTAMP_INVALID); target->paused_at = LIANA_TIMESTAMP_INVALID; + target->start = at; target->held = false; + } else if (target->start == LIANA_TIMESTAMP_INVALID && target->paused_at == LIANA_TIMESTAMP_INVALID) { + // Target was never started before. + target->start = at; + } else { + // An ended entry may never have been paused, but a non-ended entry that wasn't set + // cannot be unpaused. Checking assume_ended(target) should be safe here as long + // as it can't go from true to false (consideration for seek?). + al_assert(assume_ended(target, now) || target->paused_at != LIANA_TIMESTAMP_INVALID); } - al_assert(!current->held); - // These are not equivalent to current/target->ended. bool current_ended = assume_ended(current, now); bool target_ended = assume_ended(target, now); - // Current is ended or paused. + // If current is ended or paused. if (current_ended || current->paused_at != LIANA_TIMESTAMP_INVALID) { if (target_ended || 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; + al_assert(target->start != LIANA_TIMESTAMP_INVALID); pause = LIANA_PAUSE_RESUME; } } else { // Current is playing. @@ -283,7 +304,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) pause = LIANA_PAUSE_PAUSE; } else { // Pause-swap current and resume target. - target->start = at; + al_assert(target->start != LIANA_TIMESTAMP_INVALID); pause = LIANA_PAUSE_BOTH; } } @@ -292,12 +313,16 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) // if it becomes the target of a skip. if (!current_ended && current->paused_at == LIANA_TIMESTAMP_INVALID) { current->paused_at = at; + if (current->paused_at < current->start) { + current->paused_at = current->start; + } 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)); + al_log_debug("list", "skipto [%u-%u]: pause: %hhu, held: %s, current_ended: %s, target_ended: %s.", + current->id, target->id, pause, BOOLSTR(current->held), BOOLSTR(current_ended), BOOLSTR(target_ended)); list->current = index; list->previous = sequence; @@ -334,30 +359,42 @@ static bool handle_skip(struct lia_list *list, s32 sequence, s32 n) static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) { + // pts should be treated as a hint. if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current; - struct lia_list_entry *current = get_entry_from_sequence(list, sequence); - if (!current) return; + if (sequence < 0) return; + if (sequence != list->current) { + // A non-current entry should always be paused. + return; + } + struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); + al_assert(entry && !entry->held); u64 now = aki_get_timestamp(); - u8 pause = current->paused_at == LIANA_TIMESTAMP_INVALID ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_RESUME; + u8 pause = entry->paused_at == LIANA_TIMESTAMP_INVALID ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_RESUME; u64 at; + bool ended = assume_ended(entry, now); + if (ended) return; switch (pause) { case LIANA_PAUSE_PAUSE: - // This assert should exist but the correct behavior for this is unfinished. - //al_assert(pts != -1.0); - (void)pts; - current->paused_at = now + LIANA_PAUSE_DELAY; - current->offset += current->paused_at - current->start; - current->start = LIANA_TIMESTAMP_INVALID; - at = current->paused_at; + al_assert(entry->start != LIANA_TIMESTAMP_INVALID); + entry->paused_at = now + LIANA_PAUSE_DELAY; + if (entry->paused_at < entry->start) { + entry->paused_at = entry->start; + } + entry->offset += entry->paused_at - entry->start; + entry->start = LIANA_TIMESTAMP_INVALID; + at = entry->paused_at; break; case LIANA_PAUSE_RESUME: - current->paused_at = LIANA_TIMESTAMP_INVALID; - current->start = now + LIANA_PAUSE_DELAY; - at = current->start; + al_assert(entry->start == LIANA_TIMESTAMP_INVALID); + entry->paused_at = LIANA_TIMESTAMP_INVALID; + entry->start = now + LIANA_PAUSE_DELAY; + at = entry->start; break; } + al_log_debug("list", "toggle_pause [%u]: pts: %f, pause: %hhu.", entry->id, pts, pause); + struct lia_timing time = { .at = at, .seek_pos = LIANA_TIMESTAMP_INVALID, @@ -367,30 +404,49 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { - sink->callback(sink->userdata, LIANA_SINK_PAUSE, current, sequence, &time); + sink->callback(sink->userdata, LIANA_SINK_PAUSE, entry, sequence, &time); } } +// TODO: Take entry id instead of sequence for seek() and end()? + static void handle_seek(struct lia_list *list, s32 sequence, f64 percent) { if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current; if (sequence < 0) return; - list->idle = false; - struct lia_list_entry *current = get_entry_from_sequence(list, sequence); - u64 pos = (u64)(current->duration * percent); + if (sequence != list->current) { + return; + } + struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); + al_assert(entry); + if (entry->duration == LIANA_TIMESTAMP_INVALID) { + // Until any kind of live resource buffering. + return; + } + entry->ended = false; + u64 now = aki_get_timestamp(); - current->offset = pos; - current->start = now + LIANA_BASE_DELAY; - current->ended = false; + u64 pos = (u64)(entry->duration * percent); + u64 at = now + LIANA_BASE_DELAY; + entry->offset = pos; + if (entry->paused_at == LIANA_TIMESTAMP_INVALID) { + entry->start = at; + } + + list->idle = false; + + al_log_debug("list", "seek [%u]: pos: %f.", entry->id, pos / 1000000.0); + struct lia_timing time = { - .at = current->start, - .seek_pos = current->offset, + .at = at, + .seek_pos = pos, .pause = LIANA_PAUSE_NONE, .ended = false }; + struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { - sink->callback(sink->userdata, LIANA_SINK_SEEK, current, sequence, &time); + sink->callback(sink->userdata, LIANA_SINK_SEEK, entry, sequence, &time); } } @@ -400,11 +456,13 @@ static bool handle_end(struct lia_list *list, s32 sequence) s32 size = (s32)list->entries.size; struct lia_list_entry *entry = al_array_at(list->entries, sequence); + al_assert(entry); if (entry->ended) { al_log_warn("list", "Got end() from an already ended resource, ignoring."); return true; } entry->ended = true; + // TODO: Calculate duration for live resources? entry->offset = entry->duration; if (sequence == list->current) { @@ -510,11 +568,13 @@ void pump_queue(struct lia_list *list) break; case SKIPTO: if (!handle_skipto(list, cmd->sequence, cmd->i)) { + // Target entry not loaded. return; } break; case SKIP: if (!handle_skip(list, cmd->sequence, cmd->i)) { + // Converted to skipto and entry not loaded. return; } break; @@ -576,6 +636,8 @@ void lia_list_add(struct lia_list *list, void *opaque, u64 duration, wstr *name) { struct lia_list_entry *entry = al_alloc_object(struct lia_list_entry); entry->opaque = opaque; + entry->id = list->increment; + list->increment = al_u32_inc_wrap(list->increment); entry->paused_at = LIANA_TIMESTAMP_INVALID; entry->held = false; entry->offset = 0; diff --git a/src/liana/list.h b/src/liana/list.h index 6443db6..90f8c02 100644 --- a/src/liana/list.h +++ b/src/liana/list.h @@ -51,6 +51,7 @@ struct lia_timing { struct lia_list_entry { void *opaque; + u32 id; u64 start; u64 paused_at; bool held; @@ -84,6 +85,7 @@ struct lia_list { s32 previous; s32 queued; bool idle; + u32 increment; array(struct lia_list_entry *) entries; array(struct lia_list_sink *) sinks; array(struct lia_list_cmd *) queue; diff --git a/src/liana/server.c b/src/liana/server.c index e03b8ef..44b3f12 100644 --- a/src/liana/server.c +++ b/src/liana/server.c @@ -3,6 +3,7 @@ #include "server.h" #include "handler.h" #include "handlers.h" +#include "list.h" bool lia_server_init(struct lia_server *server, struct aki_event_loop *loop) { @@ -124,7 +125,7 @@ static void handle_connection(struct lia_node_connection *conn, struct aki_packe if (mask == 0) { struct aki_packet *rpacket = aki_packet_create(); - aki_packet_write_u16(rpacket, conn->id); + aki_packet_write_u32(rpacket, conn->id); aki_packet_write_str(rpacket, cch_entry_get_liana(conn->node->entry)); conn->handler->write_info(conn->handler, rpacket); stream->packet_callback = subscribe_packet_callback; @@ -164,7 +165,7 @@ static void signal_callback(void *userdata) } if (!conn->errored) { conn->id = server->increment; - server->increment = al_u16_inc_wrap(server->increment); + server->increment = al_u32_inc_wrap(server->increment); aki_packet_pool_init(&conn->pool, 96, server->loop, packet_pool_callback, conn); al_array_push(node->connections, conn); handle_connection(conn, packet); @@ -192,7 +193,7 @@ static aki_thread_result AKI_THREADCALL init_thread(void *userdata) return 0; } -static struct lia_node_connection *get_connection_from_id(struct lia_node *node, u16 id) +static struct lia_node_connection *get_connection_from_id(struct lia_node *node, u32 id) { struct lia_node_connection *conn = NULL; al_array_foreach(node->connections, i, conn) { @@ -201,7 +202,7 @@ static struct lia_node_connection *get_connection_from_id(struct lia_node *node, return NULL; } -static struct lia_node *get_node_from_id(struct lia_server *server, u16 id) +static struct lia_node *get_node_from_id(struct lia_server *server, u32 id) { struct lia_node *node; al_array_foreach(server->nodes, i, node) { @@ -226,8 +227,8 @@ static void packet_callback(void *userdata, struct aki_packet_stream *stream, st // We got a packet, this connection is no longer a zombie. remove_zombie(server, stream); - u16 node_id = aki_packet_read_u16(packet); - u16 connection_id = aki_packet_read_u16(packet); + u32 node_id = aki_packet_read_u32(packet); + u32 connection_id = aki_packet_read_u32(packet); struct lia_node *node = get_node_from_id(server, node_id); struct lia_node_connection *conn = NULL; @@ -283,7 +284,7 @@ struct lia_node *lia_server_create_node(struct lia_server *server, struct cch_en { struct lia_node *node = al_alloc_object(struct lia_node); node->id = server->increment; - server->increment = al_u16_inc_wrap(server->increment); + server->increment = al_u32_inc_wrap(server->increment); node->entry = entry; al_array_init(node->connections); node->server = server; @@ -296,6 +297,7 @@ static aki_thread_result AKI_THREADCALL init_duration_thread(void *userdata) struct lia_node *node = (struct lia_node *)userdata; if (!node->handler->init(node->handler, &node->handle)) { node->errored = true; + node->duration = LIANA_TIMESTAMP_INVALID; } else { node->duration = node->handler->get_duration(node->handler); } diff --git a/src/liana/server.h b/src/liana/server.h index 11fdfa9..c62eb29 100644 --- a/src/liana/server.h +++ b/src/liana/server.h @@ -9,7 +9,7 @@ #include "../cache/entry.h" struct lia_node_connection { - u16 id; + u32 id; struct aki_packet *packet; struct aki_packet_stream *stream; struct lia_server_handler *handler; @@ -26,7 +26,7 @@ enum { }; struct lia_node { - u16 id; + u32 id; struct cch_entry *entry; array(struct lia_node_connection *) connections; struct lia_server *server; @@ -44,7 +44,7 @@ struct lia_node { struct lia_server { struct aki_event_loop *loop; - u16 increment; + u32 increment; array(struct lia_node *) nodes; array(struct aki_packet_stream *) zombies; }; diff --git a/src/libsink/sink.c b/src/libsink/sink.c index d6d54fd..26c144b 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -109,15 +109,6 @@ static inline bool entry_video_buffer_held(struct camu_sink_entry *entry) } #endif -static inline bool entry_buffers_held(struct camu_sink_entry *entry) -{ -#ifndef CAMU_SINK_NO_VIDEO - return entry_audio_buffer_held(entry) || entry_video_buffer_held(entry); -#else - return entry_audio_buffer_held(entry); -#endif -} - static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_entry *entry) { al_assert(!entry->ended && entry->audio.state != BUFFER_ENDED); @@ -177,13 +168,14 @@ static void add_or_queue_entry(struct camu_sink_entry *entry) if (entry->audio.state == BUFFER_INIT) { entry->audio.state = BUFFER_QUEUED; } else { + al_assert(entry->audio.state != BUFFER_QUEUED); add_audio_if_set_and_buffered(entry); } - #ifndef CAMU_SINK_NO_VIDEO if (entry->video.state == BUFFER_INIT) { entry->video.state = BUFFER_QUEUED; } else { + al_assert(entry->video.state != BUFFER_QUEUED); add_video_if_set_and_buffered(entry); } #endif @@ -216,6 +208,17 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent } #endif +static inline s32 get_sequence_for_command(struct camu_sink *sink) +{ + s32 sequence = LIANA_SEQUENCE_ANY; + if (sink->target) { + sequence = sink->target->sequence; + } else if (sink->current) { + sequence = sink->current->sequence; + } + return sequence; +} + static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) { switch (cmd->op) { @@ -266,13 +269,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); aki_packet_write_str(packet, &sink->default_list); aki_packet_write_u8(packet, CAMU_LIST_SKIP); - s32 sequence = LIANA_SEQUENCE_ANY; - if (sink->target) { - sequence = sink->target->sequence; - } else if (sink->current) { - sequence = sink->current->sequence; - } - aki_packet_write_s32(packet, sequence); + aki_packet_write_s32(packet, get_sequence_for_command(sink)); aki_packet_write_s32(packet, cmd->value.i); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); break; @@ -293,8 +290,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); aki_packet_write_str(packet, &sink->default_list); aki_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE); - s32 sequence = sink->current ? sink->current->sequence : LIANA_SEQUENCE_ANY; - aki_packet_write_s32(packet, sequence); + aki_packet_write_s32(packet, get_sequence_for_command(sink)); aki_packet_write_f64(packet, cmd->value.f); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); #endif @@ -305,8 +301,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); aki_packet_write_str(packet, &sink->default_list); aki_packet_write_u8(packet, CAMU_LIST_SEEK); - s32 sequence = sink->current ? sink->current->sequence : LIANA_SEQUENCE_ANY; - aki_packet_write_s32(packet, sequence); + aki_packet_write_s32(packet, get_sequence_for_command(sink)); aki_packet_write_f64(packet, cmd->value.f); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); break; @@ -509,6 +504,10 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) #endif if (VIDEO_ADDED_OR_EMPTY(entry)) { entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = entry->audio.buffer_paused ? STOP : START, + .value.i = CAMU_SINK_AUDIO + }); #ifndef CAMU_SINK_NO_VIDEO bool stop_video = false; // Single frames are unconditionally added in add_video_if_set_and_buffered(). @@ -532,10 +531,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) } #endif } - queue_cmd(entry->sink, (struct camu_sink_cmd){ - .op = START, - .value.i = CAMU_SINK_AUDIO - }); + } } @@ -555,6 +551,10 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) if (AUDIO_ADDED_OR_EMPTY(entry)) { if (entry->audio.state == BUFFER_ADDED) { entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = entry->audio.buffer_paused ? STOP : START, + .value.i = CAMU_SINK_AUDIO + }); } entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); maybe_remove_previous(entry->sink); @@ -600,16 +600,19 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) } sink->current = target; + sink->current->audio.ignore_paused = false; } #ifndef CAMU_SINK_LOCAL static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at) { - if (!sink->current || sink->current->ended) { + struct camu_sink_entry *current = sink->current; + if (!current || current->ended) { switch_to(sink, target); } else { sink->target = target; - camu_clock_pause(&sink->current->clock, at); + current->audio.ignore_paused = true; + camu_clock_pause(¤t->clock, at); } } #endif @@ -647,7 +650,8 @@ static void audio_buffer_callback(void *userdata, u8 op) break; case CAMU_BUFFER_PAUSED: aki_mutex_lock(&sink->mutex); - if (!entry->audio.armed) { + al_log_info("sink", "Audio buffer paused."); + if (!entry->audio.ignore_paused) { queue_cmd(sink, (struct camu_sink_cmd){ .op = STOP, .value.i = CAMU_SINK_AUDIO @@ -808,10 +812,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str if (!camu_audio_buffer_configure(&entry->audio.buf, stream, sink->audio.mixer)) { al_log_warn("sink", "Audio buffer failed to configure."); lia_client_disconnect(&entry->client); + return; } aki_mutex_lock(&sink->mutex); - entry->audio.state = entry->audio.state == BUFFER_QUEUED ? - BUFFER_SET_OR_BUFFERED : BUFFER_CONFIGURED; + if (entry->audio.state == BUFFER_QUEUED) { + entry->audio.state = BUFFER_SET_OR_BUFFERED; + } else if (entry->audio.state == BUFFER_INIT) { + entry->audio.state = BUFFER_CONFIGURED; + } else { + assert(false); + } if (VIDEO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry); aki_mutex_unlock(&sink->mutex); break; @@ -821,10 +831,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str if (!camu_video_buffer_configure(&entry->video.buf, stream, sink->video.renderer)) { al_log_warn("sink", "Video buffer failed to configure."); lia_client_disconnect(&entry->client); + return; } aki_mutex_lock(&sink->mutex); - entry->video.state = entry->video.state == BUFFER_QUEUED ? - BUFFER_SET_OR_BUFFERED : BUFFER_CONFIGURED; + if (entry->video.state == BUFFER_QUEUED) { + entry->video.state = BUFFER_SET_OR_BUFFERED; + } else if (entry->video.state == BUFFER_INIT) { + entry->video.state = BUFFER_CONFIGURED; + } else { + assert(false); + } if (AUDIO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry); aki_mutex_unlock(&sink->mutex); break; @@ -889,37 +905,46 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str bool reconnect = *(bool *)opaque; aki_mutex_lock(&sink->mutex); - al_assert(!(reconnect && entry->ended)); - - // There are 2 reasons this entry might be in previous. - // 1. It's getting cleaned up before any entries set after it are buffered. - // 2. It got added to previous then seeked. + // We need to call this if entry was added to previous then, + // - it's being cleaned up after (ENTRY_MAX_AGE - 1) entries were added but none buffered. + // - it was seeked. maybe_remove_from_previous(sink, entry); - if (entry->audio.state == BUFFER_ENDED) { - entry->audio.state = BUFFER_SET_OR_BUFFERED; - } else if (BUFFER_NOT_EMPTY(&entry->audio)) { - remove_entry_audio_buffer(sink, entry); + if (reconnect) { + // This may not be correct if entry->ended can ever be set anywhere + // besides end_entry_and_advance_queue(). + entry->ended = false; } + bool skip_audio = sink->audio.state == SINK_PAUSED; + if (entry->audio.state == BUFFER_ADDED) { + remove_entry_audio_buffer(sink, entry); + } + // SET_OR_BUFFERED, ADDED, or ENDED. + if (entry->audio.state > BUFFER_CONFIGURED) { + entry->audio.state = BUFFER_CONFIGURED; + } #ifndef CAMU_SINK_NO_VIDEO bool ignore_video = BUFFER_EMPTY(&entry->video); - if (entry->video.state == BUFFER_ENDED) { - entry->video.state = BUFFER_SET_OR_BUFFERED; - } else if (BUFFER_NOT_EMPTY(&entry->video)) { + if (!ignore_video) { ignore_video = reconnect && ENTRY_IS_SINGLE_FRAME(entry); - if (!ignore_video) { + } + if (!ignore_video) { + if (entry->video.state == BUFFER_ADDED) { remove_entry_video_buffer(sink, entry); } + if (entry->video.state > BUFFER_CONFIGURED) { + entry->video.state = BUFFER_CONFIGURED; + } } #endif aki_mutex_unlock(&sink->mutex); while ( // Block until buffers are removed. #ifndef CAMU_SINK_NO_VIDEO - ignore_video ? entry_audio_buffer_held(entry) : entry_buffers_held(entry) + (!skip_audio && entry_audio_buffer_held(entry)) || (!ignore_video && entry_video_buffer_held(entry)) #else - entry_buffers_held(entry) + (!skip_audio && entry_audio_buffer_held(entry)) #endif ) { BLOCKING_SLEEP(AKI_TS_FROM_USEC(2500)); } @@ -938,8 +963,24 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } case LIANA_CLIENT_RESUME_AT: { struct lia_timing *time = (struct lia_timing *)opaque; - camu_clock_set(&entry->clock, time->seek_pos / 1000000.0); - camu_clock_resume(&entry->clock, time->at); + aki_mutex_lock(&sink->mutex); + camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at); + aki_mutex_unlock(&sink->mutex); + break; + } + case LIANA_CLIENT_RECONNECTED: { + aki_mutex_lock(&sink->mutex); + if (entry == sink->current) { + if (entry->audio.state > BUFFER_QUEUED) { + add_audio_if_set_and_buffered(entry); + } +#ifndef CAMU_SINK_NO_VIDEO + if (entry->video.state > BUFFER_QUEUED && !ENTRY_IS_SINGLE_FRAME(entry)) { + add_video_if_set_and_buffered(entry); + } +#endif + } + aki_mutex_unlock(&sink->mutex); break; } case LIANA_CLIENT_EOF: { @@ -1004,25 +1045,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } } -static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id, bool *created) +static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id) { - struct camu_sink_entry *entry; - al_array_foreach(sink->entries, i, entry) { - if (entry->id == id) { - *created = false; - return entry; - } - } - - entry = al_alloc_object(struct camu_sink_entry); + struct camu_sink_entry *entry = al_alloc_object(struct camu_sink_entry); + entry->sink = sink; entry->id = id; entry->ended = false; - entry->sink = sink; camu_clock_init(&entry->clock, clock_callback, entry); entry->audio.state = BUFFER_INIT; - entry->audio.armed = false; + entry->audio.ignore_paused = false; camu_audio_buffer_init(&entry->audio.buf, &entry->clock); entry->audio.buf.callback = audio_buffer_callback; entry->audio.buf.userdata = entry; @@ -1039,11 +1072,20 @@ static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id, al_array_push(sink->entries, entry); - *created = true; - return entry; } +static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u32 id) +{ + struct camu_sink_entry *entry; + al_array_foreach(sink->entries, i, entry) { + if (entry->id == id) { + return entry; + } + } + return NULL; +} + static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn, struct aki_packet *packet, struct aki_packet *rpacket) { @@ -1059,10 +1101,14 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn goto out; } + // Liana node info. str addr; aki_packet_read_str(packet, &addr); u16 port = aki_packet_read_u16(packet); - u16 node_id = aki_packet_read_u16(packet); + u32 node_id = aki_packet_read_u32(packet); + + // List entry info. + u32 id = aki_packet_read_u32(packet); s32 sequence = aki_packet_read_s32(packet); u64 at = aki_packet_read_u64(packet); u64 seek_pos = aki_packet_read_u64(packet); @@ -1070,16 +1116,25 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn bool previous_ended = aki_packet_read_bool(packet); bool ended = aki_packet_read_bool(packet); (void)previous_ended; - (void)ended; - bool created; - struct camu_sink_entry *entry = get_entry_from_id(sink, node_id, &created); + bool created = false; + struct camu_sink_entry *entry = get_entry_from_id(sink, id); + if (!entry) { + entry = create_entry(sink, id); + created = true; + } entry->sequence = sequence; sink->lru = al_u16_inc_wrap(sink->lru); entry->lru = sink->lru; if (created) { camu_clock_set(&entry->clock, seek_pos / 1000000.0); + // ended here does not map to entry->ended, we use it as a hint. + if (!ended) { + entry->audio.buffer_paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE; + } else { + camu_clock_resume(&entry->clock, 0); + } struct camu_renderer *renderer = NULL; #ifndef CAMU_SINK_NO_VIDEO renderer = sink->video.renderer; @@ -1108,81 +1163,66 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn #else switch (pause) { case LIANA_PAUSE_NONE: { - struct camu_sink_entry *target = sink->target; if (sink->target) { - al_log_warn("sink", "Ignoring target on NONE."); - if (!camu_clock_is_armed(&target->clock)) { // TMP - camu_clock_pause(&target->clock, at); - } sink->target = NULL; + } else { + al_assert(sink->current != entry); } if (entry != sink->current) { - if (camu_clock_is_armed(&entry->clock)) { // TMP - camu_clock_resume(&entry->clock, at); - } switch_to(sink, entry); } break; } case LIANA_PAUSE_RESUME: { - struct camu_sink_entry *target = sink->target; if (sink->target) { - al_log_warn("sink", "Ignoring target on RESUME."); - if (!camu_clock_is_armed(&target->clock)) { // TMP - camu_clock_pause(&target->clock, at); - } sink->target = NULL; + } else { + al_assert(sink->current != entry); } - camu_clock_resume(&entry->clock, at); if (entry != sink->current) { switch_to(sink, entry); } else { camu_audio_buffer_unpause(&entry->audio.buf); } + camu_clock_resume(&entry->clock, at); break; } case LIANA_PAUSE_PAUSE: { struct camu_sink_entry *prev_target = sink->target; if (prev_target) { - // For target to be set that must mean that current is set, - // armed to pause, and not ended. + // For target to be set that must mean that current is + // set, armed to pause, and not ended. al_assert(sink->current && !sink->current->ended); - if (prev_target == entry) { - sink->target = NULL; - } else { + al_assert(prev_target != entry); + if (sink->current != entry) { sink->target = entry; + } else { + sink->target = NULL; } - // The targets clock is only guaranteed to be resumed - // if it was set in the PAUSE_BOTH case. I feel like - // this needs to be simpler. - if (!camu_clock_is_armed(&prev_target->clock)) { // TMP - camu_clock_pause(&prev_target->clock, at); - } + camu_clock_pause(&prev_target->clock, at); } else { - if (camu_clock_is_armed(&entry->clock)) { // TMP - camu_clock_resume(&entry->clock, at); - } + al_assert(sink->current != entry); pause_and_swap_to(sink, entry, at); } break; } case LIANA_PAUSE_BOTH: { - camu_clock_resume(&entry->clock, at); struct camu_sink_entry *prev_target = sink->target; if (prev_target) { al_assert(sink->current && !sink->current->ended); - if (prev_target == entry) { - sink->target = NULL; - } else { + al_assert(prev_target != entry); + if (sink->current != entry) { sink->target = entry; + } else { + sink->target = NULL; camu_audio_buffer_unpause(&entry->audio.buf); } - if (!camu_clock_is_armed(&prev_target->clock)) { // TMP - camu_clock_pause(&prev_target->clock, at); - } + camu_clock_pause(&prev_target->clock, at); } else { + al_assert(sink->current != entry); pause_and_swap_to(sink, entry, at); } + camu_clock_resume(&entry->clock, at); break; } } @@ -1198,7 +1238,6 @@ out: return false; } -// TODO: pause and seek shouldn't depend on the entry being current. static bool pause_command_callback(void *userdata, struct aki_rpc_connection *conn, struct aki_packet *packet, struct aki_packet *rpacket) { @@ -1206,33 +1245,38 @@ static bool pause_command_callback(void *userdata, struct aki_rpc_connection *co (void)conn; (void)rpacket; + u32 id = aki_packet_read_u32(packet); s32 sequence = aki_packet_read_s32(packet); u64 at = aki_packet_read_u64(packet); u8 pause = aki_packet_read_u8(packet); aki_mutex_lock(&sink->mutex); - struct camu_sink_entry *current = sink->current; - if (!current) goto out; + struct camu_sink_entry *entry = get_entry_from_id(sink, id); + if (!entry) { + aki_mutex_unlock(&sink->mutex); + goto out; + } + al_assert(entry->sequence == sequence); #ifdef CAMU_SINK_LOCAL - (void)sequence; (void)at; (void)pause; - sink_local_pause(sink, current); + sink_local_pause(sink, entry); #else - if (current->sequence == sequence) { - if (pause == LIANA_PAUSE_PAUSE) { - camu_clock_pause(&sink->current->clock, at); - current->audio.armed = false; - } else { - camu_clock_resume(&sink->current->clock, at); - if (sink->audio.state == SINK_PAUSED) { - sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); - sink->audio.state = SINK_PLAYING; - } else { - current->audio.armed = true; - } - } + switch (pause) { + case LIANA_PAUSE_PAUSE: + entry->audio.buffer_paused = true; + camu_clock_pause(&entry->clock, at); + break; + case LIANA_PAUSE_RESUME: + entry->audio.buffer_paused = false; + camu_clock_resume(&entry->clock, at); + camu_audio_buffer_unpause(&entry->audio.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = START, + .value.i = CAMU_SINK_AUDIO + }); + break; } #endif out: @@ -1249,22 +1293,29 @@ static bool seek_command_callback(void *userdata, struct aki_rpc_connection *con (void)conn; (void)rpacket; + u32 id = aki_packet_read_u32(packet); s32 sequence = aki_packet_read_s32(packet); u64 at = aki_packet_read_u64(packet); u64 pos = aki_packet_read_u64(packet); aki_mutex_lock(&sink->mutex); - struct camu_sink_entry *current = sink->current; + struct camu_sink_entry *entry = get_entry_from_id(sink, id); + if (!entry) { + aki_mutex_unlock(&sink->mutex); + goto out; + } + if (entry == sink->current && sink->target) { + switch_to(sink, sink->target); + sink->target = NULL; + } + al_assert(entry->sequence == sequence); aki_mutex_unlock(&sink->mutex); - if (!current) goto out; - if (current->sequence == sequence) { - current->ended = false; #ifdef CAMU_SINK_LOCAL - at = 0; + at = 0; #endif - lia_client_seek(¤t->client, pos, at); - } + // The rest of the seek is handled in LIANA_CLIENT_REMOVE_BUFFERS. + lia_client_seek(&entry->client, pos, at); out: aki_packet_free(packet); @@ -1364,14 +1415,10 @@ void camu_sink_toggle_pause(struct camu_sink *sink) aki_mutex_lock(&sink->mutex); struct camu_sink_entry *current = sink->current; aki_mutex_unlock(&sink->mutex); - f64 pts = -1.0; - if (!camu_clock_is_paused(¤t->clock)) { - pts = camu_clock_get_pts(¤t->clock, 0.0); - } if (current) { queue_cmd(sink, (struct camu_sink_cmd){ .op = TOGGLE_PAUSE, - .value.f = pts, + .value.f = camu_clock_get_pts(¤t->clock, 0.0), .opaque = current }); } diff --git a/src/libsink/sink.h b/src/libsink/sink.h index f544a36..674b684 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -37,7 +37,7 @@ enum { }; struct camu_sink_entry { - u16 id; + u32 id; s32 sequence; u16 lru; bool ended; @@ -45,7 +45,10 @@ struct camu_sink_entry { struct lia_client client; struct { u8 state; - bool armed; + // Don't start the audio output for this entry. + bool buffer_paused; + // Don't stop the audio output when this entry is paused. + bool ignore_paused; struct camu_audio_buffer buf; struct lia_vcr_track *track; } audio; diff --git a/src/mixer/mixer.c b/src/mixer/mixer.c index 49a04c2..5590563 100644 --- a/src/mixer/mixer.c +++ b/src/mixer/mixer.c @@ -7,7 +7,7 @@ #define MIXER_WANT_INITIAL_SILENCE 1 // Number of silent frames to append before signaling MIXER_EMPTY. -#define MIXER_TRAILING_SILENCE 20 +#define MIXER_TRAILING_SILENCE 12 static s32 data_callback(void *userdata, u8 *data, s32 frame_count, bool *silence) { @@ -256,6 +256,7 @@ void camu_mixer_pause(struct camu_mixer *mixer) mixer->paused = true; } #ifdef CAMU_MIXER_THREADED + run_queue_internal(mixer); aki_mutex_unlock(&mixer->mutex); #endif } diff --git a/src/server/server.c b/src/server/server.c index f53eb52..109be14 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -144,9 +144,11 @@ static void list_sink_callback(void *userdata, u8 op, struct lia_list_entry *ent aki_packet_write_u8(packet, op); aki_packet_write_str(packet, &sink->server->addr); aki_packet_write_u16(packet, CAMU_PORT); - aki_packet_write_u16(packet, resource->node->id); + aki_packet_write_u32(packet, resource->node->id); + aki_packet_write_u32(packet, entry->id); aki_packet_write_s32(packet, sequence); aki_packet_write_u64(packet, timing->at); + al_assert(timing->seek_pos <= INT64_MAX); aki_packet_write_u64(packet, timing->seek_pos); aki_packet_write_u8(packet, timing->pause); aki_packet_write_bool(packet, timing->previous_ended); @@ -162,6 +164,7 @@ static void list_sink_callback(void *userdata, u8 op, struct lia_list_entry *ent } case LIANA_SINK_PAUSE: { struct aki_packet *packet = aki_rpc_get_packet(sink->conn->rpc, CAMU_SINK_PAUSE); + aki_packet_write_u32(packet, entry->id); aki_packet_write_s32(packet, sequence); aki_packet_write_u64(packet, timing->at); aki_packet_write_u8(packet, timing->pause); @@ -170,6 +173,7 @@ static void list_sink_callback(void *userdata, u8 op, struct lia_list_entry *ent } case LIANA_SINK_SEEK: { struct aki_packet *packet = aki_rpc_get_packet(sink->conn->rpc, CAMU_SINK_SEEK); + aki_packet_write_u32(packet, entry->id); aki_packet_write_s32(packet, sequence); aki_packet_write_u64(packet, timing->at); aki_packet_write_u64(packet, timing->seek_pos); diff --git a/src/sink/input_simulator.c b/src/sink/input_simulator.c index 92e3936..af0105d 100644 --- a/src/sink/input_simulator.c +++ b/src/sink/input_simulator.c @@ -10,12 +10,12 @@ static s32 quit = 1; static struct aki_thread thread; enum { - SKIP = 0, + SKIP, BACKSKIP, SHUFFLE, - MARK, // count TOGGLE_PAUSE, SEEK, + MARK, // count }; static aki_thread_result AKI_THREADCALL input_simulation_thread(void *userdata) @@ -37,7 +37,7 @@ static aki_thread_result AKI_THREADCALL input_simulation_thread(void *userdata) camu_sink_toggle_pause(sink); break; case SEEK: - camu_sink_seek(sink, 0.0); + camu_sink_seek(sink, al_rand() / (f64)AL_RAND_MAX); break; } } -- cgit v1.2.3-101-g0448