From e9475ce94ba69bd437d8cf0cf3f78062be928568 Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Sat, 7 Jun 2025 11:54:55 -0400 Subject: Clarify and fix various sink behaviors Signed-off-by: Andrew Opalach --- src/libsink/input_simulator.c | 3 +- src/libsink/sink.c | 100 ++++++++++++++++++++++++------------------ src/libsink/sink.h | 1 + 3 files changed, 61 insertions(+), 43 deletions(-) (limited to 'src/libsink') diff --git a/src/libsink/input_simulator.c b/src/libsink/input_simulator.c index 80fde02..871f0ce 100644 --- a/src/libsink/input_simulator.c +++ b/src/libsink/input_simulator.c @@ -21,8 +21,9 @@ enum { static nn_thread_result NNWT_THREADCALL input_simulation_thread(void *userdata) { struct camu_sink *sink = (struct camu_sink *)userdata; + nn_thread_set_name("input_simulator"); while (!quit) { - nn_thread_sleep(NNWT_TS_FROM_USEC(30000)); + nn_thread_sleep(NNWT_TS_FROM_USEC(60000)); switch (al_rand() % MARK) { case SKIP: camu_sink_skip(sink, (al_rand() % 5)); diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 29af2f5..94ee3c6 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -204,7 +204,7 @@ static void remove_entry_video_buffer(struct camu_sink_entry *entry) // It's possible for some of an entry's buffers to be ENDED while others are still ADDED. // This means entry->ended and BUFFER_ENDED have two distinct considerations. -// entry->ended: Completely ignored and needs special consideration in CLIENT_REMOVE_BUFFERS. +// entry->ended: Completely ignored and needs special handling in CLIENT_REMOVE_BUFFERS. // BUFFER_ENDED: No-op'd in remove_entry_buffers() and add_or_queue_entry() but otherwise unchanged. static void remove_entry_buffers(struct camu_sink_entry *entry) { @@ -279,19 +279,15 @@ 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) +static inline s32 get_sequence_for_command(struct camu_sink_entry *entry) { // SEQUENCE_ANY resolves order on the server. s32 sequence = LIANA_SEQUENCE_ANY; // If we're local this could only lead to feeling like your inputs were eaten. #ifndef CAMU_SINK_LOCAL - if (sink->target && sink->target != (struct camu_sink_entry *)0xb00b) { - sequence = sink->target->sequence; - } else if (sink->current) { - sequence = sink->current->sequence; - } + if (entry) sequence = entry->sequence; #else - (void)sink; + (void)entry; #endif return sequence; } @@ -393,7 +389,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); nn_packet_write_u8(packet, CAMU_LIST_SKIP); - nn_packet_write_s32(packet, get_sequence_for_command(sink)); + nn_packet_write_s32(packet, get_sequence_for_command((struct camu_sink_entry *)cmd->opaque)); nn_packet_write_s32(packet, (s32)cmd->value.i); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); break; @@ -406,7 +402,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); nn_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE); - nn_packet_write_s32(packet, get_sequence_for_command(sink)); + nn_packet_write_s32(packet, get_sequence_for_command((struct camu_sink_entry *)cmd->opaque)); nn_packet_write_f64(packet, cmd->value.f); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); #endif @@ -674,7 +670,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) if (detached) { al_assert(detached == current); sink->detached = NULL; - log_warn("Unset detached as a substitute for remove."); + log_debug("Unset detached as a substitute for remove."); // This should only matter if detached was unconfigured. if (AUDIO_EMPTY(detached)) AUDIO_STATE(detached) = BUFFER_INIT; if (VIDEO_EMPTY(detached)) VIDEO_STATE(detached) = BUFFER_INIT; @@ -690,7 +686,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) remove_previous_if_contains(sink, target); add_or_queue_entry(target); } else { - if (VIDEO_IS_SINGLE_FRAME(target)) { + if (!VIDEO_EMPTY(target) && VIDEO_IS_SINGLE_FRAME(target)) { add_video_if_set_and_buffered(target); } stop_video = true; @@ -698,7 +694,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) } if (ensure_removed) { - if (VIDEO_IS_SINGLE_FRAME(current)) { + if (!VIDEO_EMPTY(current) && VIDEO_IS_SINGLE_FRAME(current)) { remove_entry_video_buffer(current); } al_assert(AUDIO_NOT_ADDED(current)); @@ -849,24 +845,25 @@ static void video_buffer_callback(void *userdata, u8 op) break; case CAMU_BUFFER_EOF: log_debug("Video EOF."); - bool single_frame = VIDEO_IS_SINGLE_FRAME(entry); - bool swapped = false; nn_mutex_lock(&sink->lock); // This buffer's state could be ADDED, SET_OR_BUFFERED, or CONFIGURED. if (!AUDIO_EMPTY(entry)) { camu_audio_buffer_set_no_video(&entry->audio.buf, true); } - if (!single_frame) { - if (VIDEO_STATE(entry) == BUFFER_ADDED) { - remove_entry_video_buffer(entry); - } - VIDEO_STATE(entry) = BUFFER_ENDED; - if (AUDIO_ENDED_OR_EMPTY(entry)) { - swapped = end_entry_and_advance_queue(sink, entry); - } + if (VIDEO_IS_SINGLE_FRAME(entry)) { + nn_mutex_unlock(&sink->lock); + return; + } + if (VIDEO_STATE(entry) == BUFFER_ADDED) { + remove_entry_video_buffer(entry); + } + VIDEO_STATE(entry) = BUFFER_ENDED; + bool swapped = false; + if (AUDIO_ENDED_OR_EMPTY(entry)) { + swapped = end_entry_and_advance_queue(sink, entry); } nn_mutex_unlock(&sink->lock); - if (!swapped && !single_frame) { + if (!swapped) { queue_cmd(sink, (struct camu_sink_cmd){ .op = STOP, .value.i = CAMU_SINK_VIDEO @@ -907,8 +904,8 @@ static void clock_callback(void *userdata, u8 op) static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry) { -#ifdef CAMU_SINK_LOCAL bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); +#ifdef CAMU_SINK_LOCAL if (!AUDIO_EMPTY(entry) && !ignore_video) { f64 audio = camu_mixer_get_latency(sink->audio.mixer); s32 frames = audio / entry->video.buf.avg_frame_duration; @@ -923,13 +920,12 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s // To sync clients with differing audio latencies our only option is to factor the mixer // latency directly into the audio buffer. f64 audio = camu_mixer_get_latency(sink->audio.mixer); - if (!VIDEO_EMPTY(entry)) { + if (!ignore_video) { s32 frames = audio / entry->video.buf.avg_frame_duration; struct camu_renderer *renderer = sink->video.renderer; if (renderer) frames += renderer->get_latency(renderer); camu_video_buffer_set_latency(&entry->video.buf, frames); } - bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); camu_audio_buffer_set_latency(&entry->audio.buf, audio); camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); #endif @@ -1051,6 +1047,13 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str nn_mutex_lock(&sink->lock); log_trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(rec->reconnect), BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current)); + // Immediately switch to a potential target to avoid excessive delay/catchup + // that would be caused by this entry being re-added with an in-between clock state. + if (sink->target) { + switch_to(sink, sink->target); + sink->target = NULL; + } + // This entry might be in previous if it was added to previous then, // 1. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered. // 2. it was seeked. @@ -1077,8 +1080,8 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str remove_entry_audio_buffer(entry); } // Slight optimization. A duplicate frame will still be sent but discarded in the video buffer. - bool ignore_video = rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry); - if (!ignore_video && !VIDEO_ENDED_OR_EMPTY(entry)) { + bool skip_video = rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry); + if (!skip_video && !VIDEO_ENDED_OR_EMPTY(entry)) { remove_entry_video_buffer(entry); remove_entry_video_buffer(entry); } @@ -1090,7 +1093,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str // Unlock to wait. nn_mutex_unlock(&sink->lock); - while (entry_audio_buffer_held(entry) || (!ignore_video && entry_video_buffer_held(entry))) { + while (entry_audio_buffer_held(entry) || (!skip_video && entry_video_buffer_held(entry))) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); } @@ -1107,7 +1110,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str if (VIDEO_STATE(entry) == BUFFER_ENDED) { al_assert(!VIDEO_EMPTY(entry)); VIDEO_STATE(entry) = BUFFER_CONFIGURED; - } else if (entry->ended && !ignore_video && VIDEO_EMPTY(entry)) { + } else if (entry->ended && VIDEO_EMPTY(entry)) { VIDEO_STATE(entry) = BUFFER_INIT; } entry->ended = false; @@ -1118,22 +1121,25 @@ 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; - log_trace("resume_at("ENTRY_FMT"), paused_at: %f.", ENTRY_ARG(entry), entry->clock.paused_at); + log_trace("resume_at("ENTRY_FMT"), seek_pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->seek_pos / 1000000.0, entry->clock.paused_at); // These buffers won't be re-added until after a CLIENT_RECONNECTED event. + bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); if (!AUDIO_EMPTY(entry)) { camu_audio_buffer_reset(&entry->audio.buf); + // no_video is set in video BUFFER_EOF as a fail-safe. Reset it here. + camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); } - if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) { + if (!ignore_video) { camu_video_buffer_reset(&entry->video.buf, time->seek_pos); } -#ifdef CAMU_SINK_LOCAL - time->at = 0; -#endif nn_mutex_lock(&sink->lock); #ifdef LIANA_LIST_SCUFFED_LOOP if (time->seek_pos == 0) { camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock)); } else { +#endif +#ifdef CAMU_SINK_LOCAL + time->at = 0; #endif camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at); #ifdef LIANA_LIST_SCUFFED_LOOP @@ -1198,7 +1204,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); } } - // @TODO: Mark for removal here instead. bool removed = al_array_remove(sink->entries, entry); remove_from_queue_by_opaque(sink, entry); @@ -1525,7 +1530,6 @@ static struct nn_rpc_command commands[] = { { .op = CAMU_SINK_SEEK, .callback = seek_command_callback, .userdata = NULL } }; -// @TODO: Send sink id from list. static void identify_callback(void *userdata, struct nn_rpc_connection *conn, struct nn_packet *packet) { struct camu_sink *sink = (struct camu_sink *)userdata; @@ -1607,18 +1611,30 @@ void camu_sink_return_current(struct camu_sink *sink) nn_mutex_unlock(&sink->lock); } +static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink) +{ + if (sink->target && sink->target != (struct camu_sink_entry *)0xb00b) { + return sink->target; + } + return sink->current; +} + void camu_sink_skip(struct camu_sink *sink, s32 n) { + nn_mutex_lock(&sink->lock); + struct camu_sink_entry *current = get_entry_for_command(sink); + nn_mutex_unlock(&sink->lock); queue_cmd(sink, (struct camu_sink_cmd){ .op = SKIP, - .value.i = n + .value.i = n, + .opaque = current }); } void camu_sink_toggle_pause(struct camu_sink *sink) { nn_mutex_lock(&sink->lock); - struct camu_sink_entry *current = sink->current; + struct camu_sink_entry *current = get_entry_for_command(sink); nn_mutex_unlock(&sink->lock); if (!current) return; queue_cmd(sink, (struct camu_sink_cmd){ @@ -1631,7 +1647,7 @@ void camu_sink_toggle_pause(struct camu_sink *sink) void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode) { nn_mutex_lock(&sink->lock); - struct camu_sink_entry *current = sink->current; + struct camu_sink_entry *current = get_entry_for_command(sink); u64 duration = 0; f64 pts = 0.0; if (current) { @@ -1713,7 +1729,7 @@ void camu_sink_close(struct camu_sink *sink) void camu_sink_free(struct camu_sink *sink) { - al_assert(sink->entries.count == 0); + al_assert(!sink->entries.count); al_array_free(sink->entries); nn_rpc_free(&sink->client); camu_queue_free(sink->queue); diff --git a/src/libsink/sink.h b/src/libsink/sink.h index c2ea6a1..40bd0a4 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -13,6 +13,7 @@ #ifndef CAMU_SINK_NO_VIDEO #include "../buffer/video.h" #else +// Stub video_buffer usage within libsink only. #include "../buffer/video_null.h" #endif -- cgit v1.2.3-101-g0448