Server resource unload, many tweaks and fixes

 - Initial liana client preferences.
 - Hook up libplacebo dx11 backend.
 - Make usage of FFmpeg hardware decoding api make some sense.

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

2 files changed, 138 insertions, 91 deletions

diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 716990e..0eca9f5 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -12,12 +12,6 @@
 //#define CAMU_SINK_ONESHOT
 //#define CAMU_SINK_TRACE
 
-#ifndef CAMU_SINK_NO_VIDEO
-#ifdef CAMU_RENDERER_LIBPLACEBO
-#include "../render/renderer_libplacebo.h"
-#endif
-#endif
-
 // Requested state of the sinks outputs.
 enum {
     SINK_EMPTY = 0,
@@ -57,7 +51,7 @@ enum {
 };
 
 // Number of entries to keep buffered at one time.
-#define ENTRY_MAX_AGE 7
+#define ENTRY_MAX_AGE 3
 
 // If a buffer is still INIT or QUEUED after the entry is configured, it's "empty".
 #define BUFFER_EMPTY(buf) ((buf)->state == BUFFER_INIT || (buf)->state == BUFFER_QUEUED)
@@ -149,8 +143,7 @@ static inline void add_entry_video_buffer(struct camu_sink_entry *entry)
 
 static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
-    al_assert(!entry->ended && entry->audio.state != BUFFER_ENDED);
-    al_assert(entry->audio.state != BUFFER_INIT);
+    al_assert(entry->audio.state != BUFFER_ENDED && entry->audio.state != BUFFER_INIT);
 
     if (entry->audio.state == BUFFER_ADDED) {
         entry->audio.state = BUFFER_SET_OR_BUFFERED;
@@ -174,8 +167,7 @@ static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_e
 static void remove_entry_video_buffer(struct camu_sink *sink, struct camu_sink_entry *entry)
 {
     // Don't assert !entry->ended here because of single frame handling.
-    al_assert(entry->video.state != BUFFER_ENDED);
-    al_assert(entry->video.state != BUFFER_INIT);
+    al_assert(entry->video.state != BUFFER_ENDED && entry->video.state != BUFFER_INIT);
 
     if (entry->video.state == BUFFER_ADDED) {
         entry->video.state = BUFFER_SET_OR_BUFFERED;
@@ -216,7 +208,6 @@ static void add_video_if_set_and_buffered(struct camu_sink_entry *entry);
 
 static void add_or_queue_entry(struct camu_sink_entry *entry)
 {
-    al_assert(!entry->ended);
     if (entry->audio.state == BUFFER_INIT) {
         entry->audio.state = BUFFER_QUEUED;
     } else {
@@ -264,12 +255,21 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent
 
 static inline s32 get_sequence_for_command(struct camu_sink *sink)
 {
+    // SEQUENCE_ANY resolves order on the server.
     s32 sequence = LIANA_SEQUENCE_ANY;
+#if 0
+    // Setting an explicit sequence makes skip and pause act on "what you see".
+    // This is likely not the expected behavior in the common case. The user might
+    // feel like their input was eaten if skipping after a different skip happens
+    // on the server but is yet to be reflected on their end.
     if (sink->target) {
         sequence = sink->target->sequence;
     } else if (sink->current) {
         sequence = sink->current->sequence;
     }
+#else
+    (void)sink;
+#endif
     return sequence;
 }
 
@@ -504,6 +504,17 @@ static void maybe_remove_previous(struct camu_sink *sink)
     sink->previous.count = 0;
 }
 
+static void remove_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key)
+{
+    struct camu_sink_entry *previous;
+    al_array_foreach(sink->previous, i, previous) {
+        if (previous == key) {
+            maybe_remove_previous(sink);
+            break;
+        }
+    }
+}
+
 // Due to the looseness of the previous queue, we may have to explicitly remove an entry
 // if it becomes incorrect to attempt removing it's buffers.
 // An obvious example of this is at the point an entry gets freed.
@@ -526,15 +537,6 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
 
     al_assert(previous != target && !previous->ended);
 
-    struct camu_sink_entry *entry;
-    al_array_foreach_rev(sink->previous, i, entry) {
-        // If the entry we are about to set is in previous, run the queue.
-        if (entry == target) {
-            maybe_remove_previous(sink);
-            break;
-        }
-    }
-
     // If neither of the entries audio or video buffer is ADDED, we don't care about adding it
     // to previous (waiting for the next added entry to remove it).
 #ifndef CAMU_SINK_NO_VIDEO
@@ -614,13 +616,6 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
     } else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) {
         entry->audio.state = BUFFER_ADDED;
 
-#ifndef CAMU_SINK_NO_VIDEO
-        if (VIDEO_EMPTY(entry)) {
-            struct camu_sink *sink = entry->sink;
-            sink->callback(sink->userdata, CAMU_SINK_REFRESH_VIDEO, CAMU_SINK_VIDEO, NULL);
-        }
-#endif
-
         if (VIDEO_ADDED_OR_EMPTY(entry)) {
 #ifndef CAMU_SINK_NO_VIDEO
             bool single_frame = VIDEO_IS_SINGLE_FRAME(entry);
@@ -632,7 +627,9 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
 #endif
             add_entry_audio_buffer(entry);
 
-            // This entry could be in previous. This is well defined but confusing.
+            // This must be called after setting the buffer's state to ADDED because
+            // this entry might be in previous. It's confusing but well defined,
+            // although could probably be avoided by a greater simplification.
             maybe_remove_previous(entry->sink);
 
 #ifndef CAMU_SINK_NO_VIDEO
@@ -727,6 +724,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
     }
 
     if (!target->ended) {
+        remove_previous_if_contains(sink, target);
         add_or_queue_entry(target);
     } else {
 #ifndef CAMU_SINK_NO_VIDEO
@@ -734,17 +732,12 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
             add_video_if_set_and_buffered(target);
         } else {
             // @TODO: current-less
-            sink->callback(sink->userdata, CAMU_SINK_REFRESH_VIDEO, CAMU_SINK_VIDEO, NULL);
             queue_cmd(sink, (struct camu_sink_cmd){
                 .op = STOP,
                 .value.i = CAMU_SINK_VIDEO
             });
         }
 #endif
-        queue_cmd(sink, (struct camu_sink_cmd){
-            .op = STOP,
-            .value.i = CAMU_SINK_AUDIO
-        });
     }
 
     sink->current = target;
@@ -777,12 +770,12 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink
     return false;
 #endif
     entry->ended = true;
+    maybe_remove_from_previous(sink, entry);
     queue_cmd(sink, (struct camu_sink_cmd){
         .op = END,
         .value.u = entry->reset_id,
         .opaque = entry
     });
-    maybe_remove_from_previous(sink, entry);
 #ifdef LIANA_LIST_SCUFFED_LOOP
     al_log_info("sink", "Looping.");
     return true;
@@ -818,8 +811,8 @@ static void audio_buffer_callback(void *userdata, u8 op)
         break;
     case CAMU_BUFFER_PAUSED:
         nn_mutex_lock(&sink->mutex);
-        al_log_info("sink", "Audio buffer paused.");
-        if (!entry->audio.ignore_paused) {
+        if (!entry->audio.ignore_paused && entry->buffers_paused) {
+            al_log_info("sink", "Audio buffer paused.");
             queue_cmd(sink, (struct camu_sink_cmd){
                 .op = STOP,
                 .value.i = CAMU_SINK_AUDIO
@@ -920,9 +913,18 @@ static void clock_callback(void *userdata, u8 op)
         al_log_info("sink", "clock_callback(), target: 0x%llx", sink->target);
 #endif
         nn_mutex_lock(&sink->mutex);
-        if (entry == sink->current && sink->target) {
-            switch_to(sink, sink->target);
-            sink->target = NULL;
+        if (entry == sink->current) {
+            if (sink->target) {
+                switch_to(sink, sink->target);
+                sink->target = NULL;
+            } else if (entry->buffers_paused) {
+#ifndef CAMU_SINK_NO_VIDEO
+                queue_cmd(entry->sink, (struct camu_sink_cmd){
+                    .op = STOP,
+                    .value.i = CAMU_SINK_VIDEO
+                });
+#endif
+            }
         }
         nn_mutex_unlock(&sink->mutex);
     }
@@ -1023,21 +1025,13 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             nn_mutex_unlock(&sink->mutex);
             break;
         case CAMU_STREAM_SUBTITLE:
-#ifdef CAMU_HAVE_FFMPEG
-            if (!camu_video_buffer_configure_subtitles(&entry->video.buf, stream->av.stream->codecpar)) {
+            if (!camu_video_buffer_configure_subtitles(&entry->video.buf, stream)) {
                 al_log_warn("sink", "Video buffer failed to configure subtitles.");
             }
-#endif
             break;
         case CAMU_STREAM_ATTACHMENT: {
-#ifdef CAMU_HAVE_FFMPEG
-#ifdef CAMU_RENDERER_LIBPLACEBO
-            struct camu_renderer_lp *lp = (struct camu_renderer_lp *)sink->video.renderer;
-            AVDictionaryEntry *entry = av_dict_get(stream->av.stream->metadata, "filename", NULL, 0);
-            AVCodecParameters *codecpar = stream->av.stream->codecpar;
-            ass_add_font(lp->ass, entry->value, (const char *)codecpar->extradata, codecpar->extradata_size);
-#endif
-#endif
+            struct camu_renderer *renderer = sink->video.renderer;
+            renderer->add_font(renderer, stream);
             break;
         }
 #endif
@@ -1074,20 +1068,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         camu_codec_frame_discard(frame);
         break;
     }
+#ifndef CAMU_SINK_NO_VIDEO
     case LIANA_CLIENT_SUBTITLE: {
         switch (stream->type) {
         case CAMU_STREAM_SUBTITLE: {
-#ifndef CAMU_SINK_NO_VIDEO
-#ifdef CAMU_HAVE_FFMPEG
-            AVPacket *pkt = (AVPacket *)opaque;
-            camu_video_buffer_push_subtitle(&entry->video.buf, pkt);
-#endif
-#endif
+            camu_video_buffer_push_subtitle(&entry->video.buf, (struct camu_codec_packet *)opaque);
             break;
         }
         }
         break;
     }
+#endif
     case LIANA_CLIENT_REMOVE_BUFFERS: {
         bool reconnect = *(bool *)opaque;
 
@@ -1106,14 +1097,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         // Entry might be in previous here if it 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);
-        struct camu_sink_entry *previous;
-        al_array_foreach(sink->previous, i, previous) {
-            if (previous == entry) {
-                maybe_remove_previous(sink);
-                break;
-            }
-        }
+        remove_previous_if_contains(sink, entry);
 
         bool skip_audio = sink->audio.state == SINK_PAUSED;
         if (entry->audio.state == BUFFER_ADDED) {
@@ -1148,8 +1132,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         // some point after unlocking to block above.
         nn_mutex_lock(&sink->mutex);
 
-        if (reconnect) {
+        if (reconnect && entry->ended) {
             entry->ended = false;
+            if (entry->audio.state == BUFFER_QUEUED) {
+                entry->audio.state = BUFFER_INIT;
+            }
+#ifndef CAMU_SINK_NO_VIDEO
+            if (entry->video.state == BUFFER_QUEUED) {
+                entry->video.state = BUFFER_INIT;
+            }
+#endif
         }
 
         // SET_OR_BUFFERED, ADDED, or ENDED.
@@ -1202,10 +1194,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             al_assert(entry == sink->current);
             if (entry->audio.state > BUFFER_QUEUED) {
                 add_audio_if_set_and_buffered(entry);
+            } else {
+                entry->audio.state = BUFFER_QUEUED;
             }
 #ifndef CAMU_SINK_NO_VIDEO
-            if (entry->video.state > BUFFER_QUEUED && !VIDEO_IS_SINGLE_FRAME(entry)) {
-                add_video_if_set_and_buffered(entry);
+            if (entry->video.state > BUFFER_QUEUED) {
+                if (!VIDEO_IS_SINGLE_FRAME(entry)) {
+                    add_video_if_set_and_buffered(entry);
+                }
+            } else {
+                entry->video.state = BUFFER_QUEUED;
             }
 #endif
             sink->reconnecting = NULL;
@@ -1237,6 +1235,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
         // LIANA_CLIENT_REMOVE_BUFFERS has been called on this entry before we're here.
         nn_mutex_lock(&sink->mutex);
 
+        bool removed;
+        al_array_remove_checked(sink->entries, entry, removed);
+
         if (entry == sink->target) {
             // @TODO: current-less
             // current could be paused and targeting this entry.
@@ -1254,15 +1255,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             } else {
                 // @TODO: current-less
                 sink->current = NULL;
-                sink->callback(sink->userdata, CAMU_SINK_REFRESH_VIDEO, CAMU_SINK_VIDEO, NULL);
-                queue_cmd(entry->sink, (struct camu_sink_cmd){
-                    .op = STOP,
-                    .value.i = CAMU_SINK_VIDEO
-                });
-                queue_cmd(entry->sink, (struct camu_sink_cmd){
-                    .op = STOP,
-                    .value.i = CAMU_SINK_AUDIO
-                });
+#ifndef CAMU_SINK_NO_VIDEO
+                if (removed) { // Don't stop video on exit.
+                    queue_cmd(entry->sink, (struct camu_sink_cmd){
+                        .op = STOP,
+                        .value.i = CAMU_SINK_VIDEO
+                    });
+                }
+#endif
             }
             // If current was never fully added we need to call this here.
             maybe_remove_previous(sink);
@@ -1270,9 +1270,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
             sink->queued = NULL;
         }
 
-        bool removed;
-        al_array_remove_checked(sink->entries, entry, removed);
-
         nn_mutex_unlock(&sink->mutex);
 
         lia_client_free(&entry->client);
@@ -1280,7 +1277,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 #ifndef CAMU_SINK_NO_VIDEO
         camu_video_buffer_free(&entry->video.buf);
 #endif
-        al_log_info("sink", "Entry (0x%llx) closed by %s.", entry, removed ? "disconnect" : "cleanup");
+        al_log_info("sink", "Entry (0x%llx) closed by %s.", entry, removed ? "cleanup" : "exit");
         al_free(entry);
 
         break;
@@ -1313,6 +1310,7 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id)
 
     entry->client.callback = client_callback;
     entry->client.userdata = entry;
+    entry->client.prefs = sink->prefs;
 
     al_array_push(sink->entries, entry);
 
@@ -1523,12 +1521,8 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
     case LIANA_PAUSE_PAUSE:
         entry->buffers_paused = true;
         camu_clock_pause(&entry->clock, at);
-        if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
-            queue_cmd(entry->sink, (struct camu_sink_cmd){
-                .op = STOP,
-                .value.i = CAMU_SINK_VIDEO
-            });
-        }
+        // Audio will be stopped in a BUFFER_PAUSED callback and video
+        // will be stopped in clock_callback().
         break;
     case LIANA_PAUSE_RESUME:
         entry->buffers_paused = false;
@@ -1540,12 +1534,14 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
                 .value.i = CAMU_SINK_AUDIO
             });
         }
+#ifndef CAMU_SINK_NO_VIDEO
         if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
             queue_cmd(entry->sink, (struct camu_sink_cmd){
                 .op = START,
                 .value.i = CAMU_SINK_VIDEO
             });
         }
+#endif
         break;
     }
 #endif
@@ -1632,6 +1628,8 @@ bool camu_sink_connect(struct camu_sink *sink, u8 type, str *addr, u16 port, str
 {
     al_str_clone(&sink->name, name);
     sink->type = type;
+    al_str_clone(&sink->addr, addr);
+    sink->port = port;
     nn_timer_init(&sink->periodic_timer, sink->loop, periodic_timer_callback, sink);
     nn_timer_set_repeat(&sink->periodic_timer, NNWT_TS_FROM_USEC(1000000));
     nn_rpc_init(&sink->client, sink->loop, connection_callback, connection_closed_callback, sink);
@@ -1641,8 +1639,8 @@ bool camu_sink_connect(struct camu_sink *sink, u8 type, str *addr, u16 port, str
         nn_rpc_add_command(&sink->client, &commands[i]);
     }
     nn_rpc_prepare_client(&sink->client);
-    al_str_clone(&sink->addr, addr);
-    sink->port = port;
+    // Directly set conn here to handle never connected case.
+    sink->conn = sink->client.conn;
 #ifdef CAMU_DIRECT_MODE
     struct nn_rpc_connection *conn = sink->client.conn;
     nn_multiplex_direct_connect(conn->stream, CAMU_MULTIPLEX_RPC);
@@ -1722,11 +1720,57 @@ void camu_sink_unset(struct camu_sink *sink)
     });
 }
 
-void camu_sink_offset_volume(struct camu_sink *sink, f64 amount)
+void camu_sink_set_volume(struct camu_sink *sink, f32 volume)
+{
+    camu_mixer_set_volume(sink->audio.mixer, volume);
+}
+
+void camu_sink_offset_volume(struct camu_sink *sink, f32 amount)
 {
     camu_mixer_offset_volume(sink->audio.mixer, amount);
 }
 
+static char status[128];
+
+void camu_sink_status(struct camu_sink *sink)
+{
+    nn_mutex_lock(&sink->mutex);
+    struct camu_sink_entry *current = sink->current;
+    nn_mutex_unlock(&sink->mutex);
+    if (!current) {
+        al_log_info("sink", "Nothing playing.");
+        return;
+    }
+    f64 pts = camu_clock_get_pts(&current->clock, 0.0, false);
+    if (pts == -1.0) {
+#ifndef CAMU_SINK_NO_VIDEO
+        pts = al_atomic_load(f64)(&current->video.buf.pts, AL_ATOMIC_RELAXED);
+#endif
+    }
+    f64 duration = current->client.duration / 1000000.0;
+    s32 text = 0;
+    u32 minute = (u32)(pts / 60);
+    u32 hour = minute / 60;
+    minute -= hour * 60;
+    if (camu_clock_is_paused(&current->clock)) {
+        text += al_snprintf(status + text, sizeof(status) - text, "⏸ ");
+    } else {
+        text += al_snprintf(status + text, sizeof(status) - text, "⏵ ");
+    }
+    text += al_snprintf(status + text, sizeof(status) - text, "[");
+    bool show_hour = duration >= 60.0 * 60.0;
+    if (show_hour) text += al_snprintf(status + text, sizeof(status) - text, "%.2d:", hour);
+    text += al_snprintf(status + text, sizeof(status) - text, "%.2d:%.2d/", minute, (u32)pts % 60);
+    minute = (u32)(duration / 60);
+    hour = minute / 60;
+    minute -= hour * 60;
+    if (show_hour) text += al_snprintf(status + text, sizeof(status) - text, "%.2d:", hour);
+    text += al_snprintf(status + text, sizeof(status) - text, "%.2d:%.2d", minute, (u32)duration % 60);
+    text += al_snprintf(status + text, sizeof(status) - text, "]");
+    status[text] = '\0';
+    al_log_info("sink", "%s", status);
+}
+
 void camu_sink_stop(struct camu_sink *sink)
 {
     queue_cmd(sink, (struct camu_sink_cmd){
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 9bb408d..86c6a04 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -100,6 +100,7 @@ struct camu_sink {
         struct camu_renderer *renderer;
     } video;
 #endif
+    struct lia_prefs prefs;
     struct lia_server *local_server;
     u8 (*callback)(void *, u8, u8, void *);
     void *userdata;
@@ -120,7 +121,9 @@ void camu_sink_toggle_pause(struct camu_sink *sink);
 void camu_sink_seek(struct camu_sink *sink, f64 pos);
 void camu_sink_reseek(struct camu_sink *sink);
 void camu_sink_unset(struct camu_sink *sink);
-void camu_sink_offset_volume(struct camu_sink *sink, f64 amount);
+void camu_sink_set_volume(struct camu_sink *sink, f32 volume);
+void camu_sink_offset_volume(struct camu_sink *sink, f32 amount);
+void camu_sink_status(struct camu_sink *sink);
 void camu_sink_stop(struct camu_sink *sink);
 void camu_sink_close(struct camu_sink *sink);
 void camu_sink_free(struct camu_sink *sink);