From 846bd3d1b7bcec3653aa9d662f559235c4689e1b Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Sat, 15 Mar 2025 20:40:38 -0400 Subject: Sink refactor wip Dangling target won't stay around. I have a couple of ideas about how to massively simplify the handling of ended entries. Signed-off-by: Andrew Opalach --- src/liana/client.c | 79 ++++++++---------- src/liana/client.h | 2 +- src/liana/handlers/codec_client.c | 2 +- src/liana/list.c | 166 ++++++++++++++++++-------------------- src/liana/list.h | 2 +- src/liana/server.c | 6 +- src/liana/server.h | 3 +- src/liana/vcr.c | 4 +- 8 files changed, 125 insertions(+), 139 deletions(-) (limited to 'src/liana') diff --git a/src/liana/client.c b/src/liana/client.c index b53fb7e..726f8ff 100644 --- a/src/liana/client.c +++ b/src/liana/client.c @@ -29,12 +29,8 @@ static s32 stream_compare(const void *a, const void *b) return (s32)((struct camu_codec_stream *)a)->type - (s32)((struct camu_codec_stream *)b)->type; } -static void parse_info_packet(struct lia_client *client, struct nn_packet *packet) +static void collect_streams(struct lia_client *client, struct nn_packet *packet) { - str liana; - nn_packet_read_str(packet, &liana); - client->duration = nn_packet_read_u64(packet); - u32 count = nn_packet_read_u32(packet); for (u32 i = 0; i < count; i++) { // Very important zero-initialization. @@ -77,13 +73,23 @@ static void parse_info_packet(struct lia_client *client, struct nn_packet *packe continue; } stream.av.format_context = format_context; - if (type == CAMU_STREAM_AUDIO) { + AVCodecParameters *codecpar = stream.av.stream->codecpar; + switch (type) { + case CAMU_STREAM_AUDIO: { struct camu_audio_format *fmt = &stream.audio.fmt; - AVCodecParameters *codecpar = stream.av.stream->codecpar; fmt->format = codecpar->format; fmt->sample_rate = codecpar->sample_rate; av_channel_layout_copy(&fmt->channel_layout, &codecpar->ch_layout); fmt->channel_count = codecpar->ch_layout.nb_channels; + break; + } + case CAMU_STREAM_VIDEO: { + struct camu_video_format *fmt = &stream.video.fmt; + fmt->width = (u32)codecpar->width; + fmt->height = (u32)codecpar->height; + fmt->format = codecpar->format; + break; + } } break; } @@ -95,56 +101,44 @@ static void parse_info_packet(struct lia_client *client, struct nn_packet *packe stream.index = index; al_array_push(client->streams, stream); } - // Video streams have to come before subtitle streams. al_array_sort(client->streams, struct camu_codec_stream, stream_compare); +} - bool have_audio = false; - bool have_video = false; - bool have_subs = false; +static u8 type_to_mask[] = { + [CAMU_STREAM_AUDIO] = CAMU_MASK_AUDIO, + [CAMU_STREAM_VIDEO] = CAMU_MASK_VIDEO, + [CAMU_STREAM_SUBTITLE] = CAMU_MASK_SUBTITLE +}; +static void parse_info_packet(struct lia_client *client, struct nn_packet *packet) +{ + str liana; + nn_packet_read_str(packet, &liana); + client->duration = nn_packet_read_u64(packet); + collect_streams(client, packet); + u8 selected = 0; struct camu_codec_stream *stream; al_array_foreach_ptr(client->streams, i, stream) { - switch (stream->type) { - case CAMU_STREAM_AUDIO: - if (have_audio || !(client->prefs.enabled_mask & CAMU_MASK_AUDIO)) { - continue; - } - have_audio = true; - break; - case CAMU_STREAM_VIDEO: - if (have_video || !(client->prefs.enabled_mask & CAMU_MASK_VIDEO)) { - continue; - } - have_video = true; - break; - case CAMU_STREAM_SUBTITLE: - if (have_subs || !(client->prefs.enabled_mask & CAMU_MASK_SUBTITLE)) { - continue; - } - have_subs = true; - break; + u8 type_mask = type_to_mask[stream->type]; + if ((selected & type_mask) || !(client->prefs.enabled_mask & type_mask)) { + continue; } - + selected |= type_mask; client->mask |= 1 << stream->index; - struct lia_vcr_track *track = al_alloc_object(struct lia_vcr_track); track->stream = stream; track->client = lia_handler_by_name(&liana)->create_client_handler(); track->client->callback = client->callback; track->client->userdata = client->userdata; - if (!track->client->init(track->client, client->renderer, track->stream)) { track->client->free(&track->client); al_free(track); continue; } - client->callback(client->userdata, LIANA_CLIENT_CONFIGURE, stream, track); - lia_vcr_add_track(&client->vcr, track); } - client->callback(client->userdata, LIANA_CLIENT_CONFIGURE_COMPLETE, NULL, NULL); } @@ -185,12 +179,12 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream) .pause = LIANA_PAUSE_NONE }; client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time); - if (client->mask == 0) { + bool unconfigured = client->mask == 0; + if (unconfigured) { al_assert(client->connection_id == 0); al_log_warn("liana", "Handling reconnect on unconfigured client."); - } else { - client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, NULL); } + client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &unconfigured); } else { al_assert(client->connection_id == 0); } @@ -218,9 +212,7 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream * } else { lia_vcr_close_all(&client->vcr); } - // REMOVE_BUFFERS can possibly run the event loop while waiting. - // This should be accounted for in the client code here to not cause - // any unexpected behavior. + // We need to account for REMOVE_BUFFERS possibly running the event loop to wait. client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &client->reconnect); if (client->reconnect == RECONNECT_ON_CONNECTION_CLOSED) { // If reconnect() errors, this will close the client on recursion. @@ -236,7 +228,7 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream * } void lia_client_connect(struct lia_client *client, struct nn_event_loop *loop, - u8 type, str *addr, u16 port, u32 node_id, u64 pos, struct camu_renderer *renderer) + u8 type, str *addr, u16 port, u32 node_id, u64 pos) { client->loop = loop; client->node_id = node_id; @@ -249,7 +241,6 @@ void lia_client_connect(struct lia_client *client, struct nn_event_loop *loop, client->port = port; client->connection_id = 0; nn_packet_stream_init(&client->data, connection_callback, connection_closed_callback, client); - client->renderer = renderer; #ifdef CAMU_DIRECT_MODE (void)type; nn_multiplex_direct_connect(&client->data, CAMU_MULTIPLEX_LIANA); diff --git a/src/liana/client.h b/src/liana/client.h index 9e49d0e..4e2ff9e 100644 --- a/src/liana/client.h +++ b/src/liana/client.h @@ -45,7 +45,7 @@ struct lia_client { }; void lia_client_connect(struct lia_client *client, struct nn_event_loop *loop, - u8 type, str *addr, u16 port, u32 node_id, u64 pos, struct camu_renderer *renderer); + u8 type, str *addr, u16 port, u32 node_id, u64 pos); 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/handlers/codec_client.c b/src/liana/handlers/codec_client.c index 6f7e9f4..a412b5c 100644 --- a/src/liana/handlers/codec_client.c +++ b/src/liana/handlers/codec_client.c @@ -46,8 +46,8 @@ static bool push_av_packet(struct lia_codec_client *codec, AVPacket *pkt) static void passthrough_subtitle(struct lia_codec_client *codec, AVPacket *pkt) { - struct camu_codec_packet packet = { .av.pkt = pkt }; struct camu_codec_stream *stream = codec->handler.stream; + struct camu_codec_packet packet = { .av.pkt = pkt }; codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_SUBTITLE, stream, &packet); } #endif diff --git a/src/liana/list.c b/src/liana/list.c index 650b6c5..47b8545 100644 --- a/src/liana/list.c +++ b/src/liana/list.c @@ -1,18 +1,17 @@ #include #include #include +#define AL_LOG_SECTION "list" +//#define AL_LOG_ENABLE_TRACE #include #include "list.h" #include "list_cmp.h" -//#define LIANA_LIST_TRACE - enum { ADD_SINK = 0, REMOVE_SINK, ADD, - UNSET, SKIPTO, SKIP, TOGGLE_PAUSE, @@ -21,6 +20,7 @@ enum { REVERSE, SORT, SHUFFLE, + UNSET, CLEAR }; @@ -73,17 +73,17 @@ static bool entry_load_and_get_duration(struct lia_list *list, struct lia_list_e if (status == LIANA_ENTRY_ERRORED) { if (sequence >= 0) { al_array_remove_at(list->entries, (u32)sequence); - if (list->cmd->sequence == sequence) { - // @TODO: This command has to be effectively discarded. - list->cmd->sequence = -1; - } else if (list->cmd->sequence > sequence) { - list->cmd->sequence--; - } - if (list->current >= sequence) { + if (list->current > sequence) { list->current--; - // This maps to the behavior of only skipping ahead on errors. - if ((u32)list->current == list->entries.count) { - list->idle = true; + } + struct lia_list_cmd *cmd = list->cmd; + al_assert(cmd->sequence != sequence); + if (cmd->sequence > sequence) { + cmd->sequence--; + } + al_array_foreach(list->command_queue, i, cmd) { + if (cmd->sequence > sequence) { + cmd->sequence--; } } } @@ -150,8 +150,7 @@ static void set_current(struct lia_list *list, struct lia_list_entry *entry, s32 entry_ref(list, entry); list->current = sequence; list->idle = false; - // @TODO: This is really wrong. Switching back and forth between 2 entries - // will unload everything else. + // @TODO: Skipping back and forth between 2 entries will unload everything else. unref_all_entries(list); struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { @@ -164,14 +163,8 @@ static void pump_queue(struct lia_list *list); static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) { - struct lia_list_entry *current; if (list->current >= 0 && !list->idle) { - current = al_array_at(list->entries, list->current); - bool error; - if (!entry_load_and_get_duration(list, current, list->current, &error)) { - if (error) pump_queue(list); - return false; - } + struct lia_list_entry *current = al_array_at(list->entries, list->current); u64 now = nn_get_timestamp(); u8 pause; u64 at = LIANA_TIMESTAMP_INVALID; @@ -205,19 +198,33 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) static void handle_remove_sink(struct lia_list *list, void *userdata) { + struct lia_list_cmd *cmd = list->cmd; + if (cmd && cmd->op == ADD_SINK && cmd->sink->userdata == userdata) { + al_free(cmd->sink); + al_free(cmd); + list->cmd = NULL; + return; + } + al_array_foreach(list->command_queue, i, cmd) { + if (cmd->op == ADD_SINK && cmd->sink->userdata == userdata) { + al_free(cmd->sink); + al_free(cmd); + al_array_remove_at(list->command_queue, i); + return; + } + } struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { if (sink->userdata == userdata) { al_array_remove_at(list->sinks, i); al_free(sink); - break; + return; } } } static bool handle_add(struct lia_list *list, struct lia_list_entry *entry) { - // The list being idle doesn't mean list->current or sink->set aren't set. if (list->idle) { bool error; if (!entry_load_and_get_duration(list, entry, -1, &error)) { @@ -241,28 +248,6 @@ static bool handle_add(struct lia_list *list, struct lia_list_entry *entry) return true; } -static void unset_all(struct lia_list *list) -{ - list->current = -1; - list->queued = -1; - struct lia_list_sink *sink; - al_array_foreach(list->sinks, i, sink) { - sink->set = -1; - sink->queued = -1; - } -} - -static void handle_unset(struct lia_list *list) -{ - unset_all(list); - struct lia_list_sink *sink; - al_array_foreach(list->sinks, i, sink) { - sink_unset(sink); - } - list->current = list->entries.count - 1; - list->idle = true; -} - static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32 sequence) { s32 size = (s32)list->entries.count; @@ -274,7 +259,7 @@ static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32 static s32 get_sequence_from_entry_id(struct lia_list *list, u32 id) { - // Not returning the entry pointer here seems wasteful but it's a meaningful simplification. + // Not returning the entry pointer here is a meaningful simplification. struct lia_list_entry *entry; al_array_foreach(list->entries, i, entry) { if (entry->id == id) { @@ -369,6 +354,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) // // @TODO: Why not hold an ended entry? The idea of ended entries being // unpaused but never held seems like a over-complication. + // sink-side ended should only reference buffer state? i.e. pause the entries clock even if ended. if (!current_ended && current->paused_at == LIANA_TIMESTAMP_INVALID) { current->paused_at = at; if (current->paused_at < current->start) { @@ -379,10 +365,8 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) current->held = true; } -#ifdef LIANA_LIST_TRACE - al_log_info("list", "skipto [#%u-#%u]: pause: %hhu, held: %s, current_ended: %s, target_ended: %s.", + trace("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)); -#endif struct lia_timing time = { .at = at, @@ -424,6 +408,7 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) u64 at; bool ended = assume_ended(entry, now); if (ended) return; + switch (pause) { case LIANA_PAUSE_PAUSE: al_assert(entry->start != LIANA_TIMESTAMP_INVALID); @@ -443,11 +428,7 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) break; } -#ifdef LIANA_LIST_TRACE - al_log_info("list", "toggle_pause [#%u]: pts: %f, pause: %hhu.", entry->id, pts, pause); -#else - (void)pts; -#endif + trace("toggle_pause (#%u): pts: %f, pause: %hhu.", entry->id, pts, pause); struct lia_timing time = { .at = at, @@ -462,32 +443,30 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) } } -static void handle_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent) +static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos) { - struct lia_list_entry *entry; if (sequence == LIANA_SEQUENCE_ANY) { sequence = list->current; } else { sequence = get_sequence_from_entry_id(list, id); } - if (sequence < 0) return; - entry = get_entry_from_sequence(list, sequence); + struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); al_assert(entry); - if (entry->duration == LIANA_TIMESTAMP_INVALID) { - al_log_warn("list", "Skipping seek on entry with no duration."); + if (entry->duration == LIANA_TIMESTAMP_INVALID || entry->duration == 0) { + warn("Skipping seek on entry with no duration."); return; } - entry->ended = false; - entry->reset_id = get_incremental_id(list); - + pos = CLAMP(pos, (u64)0, entry->duration); u64 now = nn_get_timestamp(); - u64 pos = (u64)(entry->duration * percent); u64 at = now + LIANA_BASE_DELAY; u8 pause = entry->paused_at == LIANA_TIMESTAMP_INVALID ? LIANA_PAUSE_RESUME : LIANA_PAUSE_NONE; + + entry->ended = false; + entry->reset_id = get_incremental_id(list); entry->offset = pos; if (pause == LIANA_PAUSE_RESUME) { entry->start = at; @@ -495,9 +474,7 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent list->idle = false; -#ifdef LIANA_LIST_TRACE - al_log_info("list", "seek [#%u]: pos: %f.", entry->id, pos / 1000000.0); -#endif + trace("seek (#%u): pos: %f, pause: %hhu.", entry->id, pos / 1000000.0, pause); struct lia_timing time = { .at = at, @@ -524,18 +501,16 @@ static bool handle_end(struct lia_list *list, u32 id, u32 reset_id) struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); if (reset_id != entry->reset_id) { - al_log_warn("list", "Got end() with out of order or incorrect reset id, ignoring."); + warn("Got end() with out of order or incorrect reset id, ignoring."); return true; } if (entry->ended) { - al_log_warn("list", "Got end() from an already ended resource, ignoring."); + warn("Got end() from an already ended resource, ignoring."); return true; } -#ifdef LIANA_LIST_TRACE - al_log_info("list", "end [#%u].", entry->id); -#endif + trace("end (#%u).", entry->id); entry->ended = true; entry->offset = entry->duration; @@ -582,7 +557,7 @@ static bool adjust_current(struct lia_list *list, struct lia_list_entry *previou { al_assert(list->current >= 0); struct lia_list_cmd *cmd = list->cmd; - struct lia_list_entry *entry; + struct lia_list_entry *entry = NULL; al_array_foreach(list->entries, i, entry) { if (entry->opaque == previous->opaque) { if (i == (u32)list->current) { @@ -595,7 +570,7 @@ static bool adjust_current(struct lia_list *list, struct lia_list_entry *previou break; } } - al_assert(cmd->op == SKIPTO && !entry->held); + al_assert(cmd->op == SKIPTO && entry && !entry->held); pump_queue(list); return false; } @@ -651,6 +626,20 @@ static bool handle_shuffle(struct lia_list *list) return adjust_current(list, previous); } +static void handle_unset(struct lia_list *list) +{ + // @TODO: Unset behavior (flag on list): + // SKIP: Based on previous current. + // ADD: Skip to added entry. + // SEEK: Set and seek previous current. + // Explicitly ignore all other events. + struct lia_list_sink *sink; + al_array_foreach(list->sinks, i, sink) { + sink->set = -1; + sink_unset(sink); + } +} + static void run_queue(struct lia_list *list) { // @TODO: What does current = -1/currentless really mean. @@ -673,9 +662,6 @@ static void run_queue(struct lia_list *list) return; } break; - case UNSET: - handle_unset(list); - break; case SKIPTO: if (!handle_skipto(list, cmd->sequence, cmd->arg0.i)) { // Target entry not loaded. @@ -692,7 +678,7 @@ static void run_queue(struct lia_list *list) handle_toggle_pause(list, cmd->sequence, cmd->argf); break; case SEEK: - handle_seek(list, cmd->sequence, cmd->arg0.u, cmd->argf); + handle_seek(list, cmd->sequence, cmd->arg0.u, cmd->arg1.u); break; case END: if (!handle_end(list, cmd->arg0.u, cmd->arg1.u)) { @@ -715,6 +701,9 @@ static void run_queue(struct lia_list *list) return; } break; + case UNSET: + handle_unset(list); + break; case CLEAR: break; } @@ -775,13 +764,6 @@ void lia_list_add(struct lia_list *list, void *opaque, u64 duration, wstr *name) pump_queue(list); } -void lia_list_unset(struct lia_list *list) -{ - struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd); - cmd->op = UNSET; - al_array_push(list->command_queue, cmd); - pump_queue(list); -} void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index) { @@ -813,13 +795,13 @@ void lia_list_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) pump_queue(list); } -void lia_list_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent) +void lia_list_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos) { struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd); cmd->op = SEEK; cmd->sequence = sequence; cmd->arg0.u = id; - cmd->argf = percent; + cmd->arg1.u = pos; al_array_push(list->command_queue, cmd); pump_queue(list); } @@ -858,6 +840,14 @@ void lia_list_shuffle(struct lia_list *list) pump_queue(list); } +void lia_list_unset(struct lia_list *list) +{ + struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd); + cmd->op = UNSET; + al_array_push(list->command_queue, cmd); + pump_queue(list); +} + /* void lia_list_clear(struct lia_list *list) { diff --git a/src/liana/list.h b/src/liana/list.h index 0d80dde..754a4bd 100644 --- a/src/liana/list.h +++ b/src/liana/list.h @@ -130,7 +130,7 @@ void lia_list_unset(struct lia_list *list); void lia_list_skipto(struct lia_list *list, s32 sequence, s32 i); void lia_list_skip(struct lia_list *list, s32 sequence, s32 n); void lia_list_toggle_pause(struct lia_list *list, s32 sequence, f64 pts); -void lia_list_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent); +void lia_list_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos); void lia_list_end(struct lia_list *list, u32 id, u32 reset_id); void lia_list_reverse(struct lia_list *list); diff --git a/src/liana/server.c b/src/liana/server.c index 13ca0aa..9d3661c 100644 --- a/src/liana/server.c +++ b/src/liana/server.c @@ -450,7 +450,11 @@ static void duration_signal_callback(void *userdata) if (should_free_node(node)) { free_node(node); } else { - node->callback(node->userdata, LIANA_NODE_DURATION, node->duration); + if (node->errored) { + node->callback(node->userdata, LIANA_NODE_ERRORED, LIANA_TIMESTAMP_INVALID); + } else { + node->callback(node->userdata, LIANA_NODE_DURATION, node->duration); + } } } diff --git a/src/liana/server.h b/src/liana/server.h index 364008e..679d607 100644 --- a/src/liana/server.h +++ b/src/liana/server.h @@ -40,7 +40,8 @@ struct lia_node_connection { }; enum { - LIANA_NODE_DURATION = 0 + LIANA_NODE_DURATION = 0, + LIANA_NODE_ERRORED }; struct lia_node { diff --git a/src/liana/vcr.c b/src/liana/vcr.c index e06ff8e..a70a19a 100644 --- a/src/liana/vcr.c +++ b/src/liana/vcr.c @@ -250,9 +250,9 @@ static void update_metrics(struct lia_vcr *vcr, u32 size) return; } f32 kbps = (frame / 125.f) / (diff / 1000000.f); - f32 size = vcr->mark.buffered / (f32)MB(1); + f32 capacity = vcr->mark.buffered / (f32)MB(1); f32 buffered = al_atomic_load(u64)(&vcr->count, AL_ATOMIC_RELAXED) / (f32)MB(1); - al_log_info("vcr", "Receiving packets at %.2fkbps (%.2f/%.2fMB).", kbps, buffered, size); + al_log_info("vcr", "Receiving packets at %.2fkbps (%.2f/%.2fMB).", kbps, buffered, capacity); } } -- cgit v1.2.3-101-g0448