More robust disconnect for errored entries

 - Improve >1 width character handling in tui input.

Signed-off-by: Andrew Opalach <andrew@akon.city>

2 files changed, 83 insertions, 55 deletions

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;