From 46d5660a2582323e1c9c33a56694c322bc21b7bc Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Wed, 5 Mar 2025 14:37:12 -0500 Subject: More robust disconnect for errored entries - Improve >1 width character handling in tui input. Signed-off-by: Andrew Opalach --- src/libsink/sink.c | 135 ++++++++++++++++++++++++++++++++--------------------- src/libsink/sink.h | 3 +- 2 files changed, 83 insertions(+), 55 deletions(-) (limited to 'src/libsink') diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 3528e6b..3f337d5 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -35,12 +35,15 @@ enum { // Command queue commands. enum { + // Sink operations. START = 0, STOP, ADD_BUFFER, REMOVE_BUFFER, - CLEAR, + CLEAR_BUFFERS, + ERROR_OUT, CLOSE, + // List actions. TOGGLE_PAUSE, SEEK, SKIP, @@ -224,6 +227,16 @@ static void add_or_queue_entry(struct camu_sink_entry *entry) #endif } +// Disconnecting a packet stream twice before a reconnect is an error. +static void maybe_disconnect_entry(struct camu_sink_entry *entry) +{ + if (!entry->disconnected) { + // disconnect() could free entry. + entry->disconnected = true; + lia_client_disconnect(&entry->client); + } +} + #ifdef CAMU_SINK_LOCAL static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *entry) { @@ -342,7 +355,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) } break; } - case CLEAR: { + case CLEAR_BUFFERS: { switch (cmd->value.i) { case CAMU_SINK_AUDIO: sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_AUDIO, NULL); @@ -355,6 +368,11 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) } break; } + case ERROR_OUT: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + maybe_disconnect_entry(entry); + break; + } case CLOSE: { nn_signal_stop(&sink->queue_signal); sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL); @@ -472,7 +490,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, ) { sink->loop = loop; - nn_mutex_init(&sink->mutex); + nn_mutex_init(&sink->lock); nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink); nn_signal_start(&sink->queue_signal); camu_queue_init(sink->queue); @@ -580,7 +598,7 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink) // 0 1 2 3 4 if (entry->lru > sink->lru && ((UINT16_MAX - entry->lru) + 1) + sink->lru >= ENTRY_MAX_AGE) { al_array_remove_at(sink->entries, i); - lia_client_disconnect(&entry->client); + maybe_disconnect_entry(entry); } if (sink->entries.count <= ENTRY_MAX_AGE) return; } @@ -592,7 +610,7 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink) al_assert(sink->lru >= entry->lru); if (sink->lru - entry->lru >= ENTRY_MAX_AGE) { al_array_remove_at(sink->entries, i); - lia_client_disconnect(&entry->client); + maybe_disconnect_entry(entry); } if (sink->entries.count <= ENTRY_MAX_AGE) return; } @@ -809,9 +827,9 @@ static void audio_buffer_callback(void *userdata, u8 op) switch (op) { case CAMU_BUFFER_BUFFERED: lia_vcr_set_buffered(entry->audio.track); - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); add_audio_if_set_and_buffered(entry); - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; case CAMU_BUFFER_CORK: lia_vcr_cork(entry->audio.track); @@ -820,7 +838,7 @@ static void audio_buffer_callback(void *userdata, u8 op) lia_vcr_uncork(entry->audio.track); break; case CAMU_BUFFER_PAUSED: - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); if (!entry->audio.ignore_paused && entry->buffers_paused) { al_log_info("sink", "Audio buffer paused."); queue_cmd(sink, (struct camu_sink_cmd){ @@ -828,10 +846,10 @@ static void audio_buffer_callback(void *userdata, u8 op) .value.i = CAMU_SINK_AUDIO }); } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; case CAMU_BUFFER_EOF: { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); al_log_info("sink", "Audio EOF."); // @TODO: This is not well synced. EOF can happen at any time // while other stuff is happening in the sink. For example @@ -853,11 +871,15 @@ static void audio_buffer_callback(void *userdata, u8 op) if (end_entry) { end_entry_and_advance_queue(sink, entry); } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; } case CAMU_BUFFER_ERRORED: al_log_error("sink", "Audio buffer errored."); + queue_cmd(sink, (struct camu_sink_cmd){ + .op = ERROR_OUT, + .opaque = entry + }); break; } } @@ -870,9 +892,9 @@ static void video_buffer_callback(void *userdata, u8 op) switch (op) { case CAMU_BUFFER_BUFFERED: { lia_vcr_set_buffered(entry->video.track); - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); add_video_if_set_and_buffered(entry); - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; } case CAMU_BUFFER_CORK: @@ -884,7 +906,7 @@ static void video_buffer_callback(void *userdata, u8 op) case CAMU_BUFFER_EOF: { bool swapped = false; bool single_frame = VIDEO_IS_SINGLE_FRAME(entry); - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); al_log_info("sink", "Video EOF."); if (!AUDIO_EMPTY(entry)) { camu_audio_buffer_set_no_video(&entry->audio.buf, true); @@ -899,7 +921,7 @@ static void video_buffer_callback(void *userdata, u8 op) swapped = end_entry_and_advance_queue(sink, entry); } } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); if (!swapped && !single_frame) { queue_cmd(sink, (struct camu_sink_cmd){ .op = STOP, @@ -910,6 +932,10 @@ static void video_buffer_callback(void *userdata, u8 op) } case CAMU_BUFFER_ERRORED: al_log_error("sink", "Video buffer errored."); + queue_cmd(sink, (struct camu_sink_cmd){ + .op = ERROR_OUT, + .opaque = entry + }); break; } } @@ -923,7 +949,7 @@ static void clock_callback(void *userdata, u8 op) #ifdef CAMU_SINK_TRACE al_log_info("sink", "clock_callback(), target: 0x%llx", sink->target); #endif - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); if (entry == sink->current) { if (sink->target) { switch_to(sink, sink->target); @@ -937,7 +963,7 @@ static void clock_callback(void *userdata, u8 op) #endif } } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); } } @@ -1004,10 +1030,10 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str entry->audio.track = (struct lia_vcr_track *)opaque; 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); + maybe_disconnect_entry(entry); return; } - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); if (entry->audio.state == BUFFER_QUEUED) { entry->audio.state = BUFFER_SET_OR_BUFFERED; } else if (entry->audio.state == BUFFER_INIT) { @@ -1015,17 +1041,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } else { al_assert(false); } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; #ifndef CAMU_SINK_NO_VIDEO case CAMU_STREAM_VIDEO: entry->video.track = (struct lia_vcr_track *)opaque; 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); + maybe_disconnect_entry(entry); return; } - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); if (entry->video.state == BUFFER_QUEUED) { entry->video.state = BUFFER_SET_OR_BUFFERED; } else if (entry->video.state == BUFFER_INIT) { @@ -1033,7 +1059,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } else { al_assert(false); } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; case CAMU_STREAM_SUBTITLE: if (!camu_video_buffer_configure_subtitles(&entry->video.buf, stream)) { @@ -1051,9 +1077,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } case LIANA_CLIENT_CONFIGURE_COMPLETE: { // All present buffers were configured. - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); evaluate_and_set_buffer_params(sink, entry); - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; } case LIANA_CLIENT_DATA: { @@ -1093,7 +1119,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str case LIANA_CLIENT_REMOVE_BUFFERS: { bool reconnect = *(bool *)opaque; - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); #ifdef CAMU_SINK_TRACE al_log_info("sink", "remove_buffers(%s), entry == current: %s", BOOLSTR(reconnect), BOOLSTR(entry == sink->current)); @@ -1133,7 +1159,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str // MIXER_THREADED_START_STOP is not set. run_queue_by_opaque(sink, entry); - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); while ( // Block until buffers are removed. #ifndef CAMU_SINK_NO_VIDEO @@ -1145,7 +1171,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str // Reset possible ENDED state here in case the entry ended at // some point after unlocking to block above. - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); if (reconnect && entry->ended) { entry->ended = false; @@ -1172,7 +1198,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } #endif - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); if (reconnect) { if (!AUDIO_EMPTY(entry)) { @@ -1189,10 +1215,10 @@ 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; - nn_mutex_lock(&sink->mutex); #ifdef CAMU_SINK_TRACE al_log_info("sink", "resume_at(%llu, %llu)", time->seek_pos, time->at); #endif + nn_mutex_lock(&sink->lock); #if defined LIANA_LIST_SCUFFED_LOOP && !defined CAMU_SINK_NO_VIDEO struct camu_video_buffer *buf = &entry->video.buf; if (time->seek_pos == 0 && buf->last_pts >= 0.0) { @@ -1203,11 +1229,11 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str #else camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at); #endif - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; } case LIANA_CLIENT_RECONNECTED: { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); if (entry == sink->reconnecting) { al_assert(entry == sink->current); if (entry->audio.state > BUFFER_QUEUED) { @@ -1226,7 +1252,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str #endif sink->reconnecting = NULL; } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); break; } case LIANA_CLIENT_EOF: { @@ -1251,7 +1277,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } case LIANA_CLIENT_CLOSED: { // LIANA_CLIENT_REMOVE_BUFFERS has been called on this entry before we're here. - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); bool removed; al_array_remove_checked(sink->entries, entry, removed); @@ -1288,7 +1314,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str sink->queued = NULL; } - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); lia_client_free(&entry->client); camu_audio_buffer_free(&entry->audio.buf); @@ -1308,6 +1334,7 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id) struct camu_sink_entry *entry = al_alloc_object(struct camu_sink_entry); entry->sink = sink; entry->id = id; + entry->disconnected = false; entry->ended = false; camu_clock_init(&entry->clock, clock_callback, entry); @@ -1353,7 +1380,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, (void)conn; (void)rpacket; - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); u8 op = nn_packet_read_u8(packet); if (op == LIANA_SINK_UNSET) { @@ -1503,7 +1530,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, #endif out: - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); if (op != LIANA_SINK_BUFFER) { maybe_cleanup_old_entries(sink); } @@ -1529,7 +1556,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con if (!entry) goto out; al_assert(entry->sequence == sequence); - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); #ifdef CAMU_SINK_LOCAL (void)at; (void)pause; @@ -1563,7 +1590,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con break; } #endif - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); out: nn_packet_stream_return_packet(conn->stream, packet); @@ -1670,13 +1697,13 @@ bool camu_sink_connect(struct camu_sink *sink, u8 type, str *addr, u16 port, str struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink) { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); return sink->current; } void camu_sink_return_current(struct camu_sink *sink) { - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); } void camu_sink_skip(struct camu_sink *sink, s32 n) @@ -1696,9 +1723,9 @@ void camu_sink_shuffle(struct camu_sink *sink) void camu_sink_toggle_pause(struct camu_sink *sink) { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); if (current) { queue_cmd(sink, (struct camu_sink_cmd){ .op = TOGGLE_PAUSE, @@ -1710,9 +1737,9 @@ void camu_sink_toggle_pause(struct camu_sink *sink) void camu_sink_seek(struct camu_sink *sink, f64 precent) { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); queue_cmd(sink, (struct camu_sink_cmd){ .op = SEEK, .value.f = precent, @@ -1734,9 +1761,9 @@ static f64 tmp_get_entry_pts(struct camu_sink_entry *entry) void camu_sink_relative_seek(struct camu_sink *sink, f64 offset) { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); if (!current) return; f64 pts = tmp_get_entry_pts(current) + offset; f64 duration = current->client.duration / 1000000.0; @@ -1749,9 +1776,9 @@ void camu_sink_relative_seek(struct camu_sink *sink, f64 offset) void camu_sink_reseek(struct camu_sink *sink) { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); if (!current) return; queue_cmd(sink, (struct camu_sink_cmd){ .op = RESEEK, @@ -1780,9 +1807,9 @@ static char status[128]; void camu_sink_status(struct camu_sink *sink) { - nn_mutex_lock(&sink->mutex); + nn_mutex_lock(&sink->lock); struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->mutex); + nn_mutex_unlock(&sink->lock); if (!current) { al_log_info("sink", "Nothing playing."); return; @@ -1819,7 +1846,7 @@ void camu_sink_stop(struct camu_sink *sink) .value.i = CAMU_SINK_AUDIO }); queue_cmd(sink, (struct camu_sink_cmd){ - .op = CLEAR, + .op = CLEAR_BUFFERS, .value.i = CAMU_SINK_AUDIO }); queue_cmd(sink, (struct camu_sink_cmd){ @@ -1835,7 +1862,7 @@ void camu_sink_close(struct camu_sink *sink) struct camu_sink_entry *entry; al_array_foreach_rev(sink->entries, i, entry) { al_array_remove_at(sink->entries, i); - lia_client_disconnect(&entry->client); + maybe_disconnect_entry(entry); } } @@ -1848,7 +1875,7 @@ void camu_sink_free(struct camu_sink *sink) al_array_free(sink->entries); nn_rpc_free(&sink->client); camu_queue_free(sink->queue); - nn_mutex_destroy(&sink->mutex); + nn_mutex_destroy(&sink->lock); al_str_free(&sink->addr); al_str_free(&sink->name); } diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 4e97f3a..ae8e2e4 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -43,6 +43,7 @@ struct camu_sink_entry { struct lia_client client; s32 sequence; u16 lru; + bool disconnected; bool ended; u32 reset_id; struct camu_clock clock; @@ -78,7 +79,7 @@ struct camu_sink { u16 port; struct nn_rpc client; struct nn_rpc_connection *conn; - struct nn_mutex mutex; + struct nn_mutex lock; struct nn_timer periodic_timer; struct nn_signal queue_signal; queue(struct camu_sink_cmd) queue; -- cgit v1.2.3-101-g0448