diff options
| -rw-r--r-- | src/buffer/audio.c | 38 | ||||
| -rw-r--r-- | src/buffer/audio.h | 2 | ||||
| -rw-r--r-- | src/buffer/common.h | 10 | ||||
| -rw-r--r-- | src/buffer/common_internal.h | 2 | ||||
| -rw-r--r-- | src/buffer/video.c | 66 | ||||
| -rw-r--r-- | src/buffer/video.h | 8 | ||||
| -rw-r--r-- | src/buffer/video_null.h | 6 | ||||
| -rw-r--r-- | src/codec/ffmpeg/decoder.c | 26 | ||||
| -rw-r--r-- | src/codec/ffmpeg/decoder.h | 1 | ||||
| -rw-r--r-- | src/liana/client.c | 74 | ||||
| -rw-r--r-- | src/liana/client.h | 5 | ||||
| -rw-r--r-- | src/liana/handlers/codec_client.c | 39 | ||||
| -rw-r--r-- | src/liana/list.c | 6 | ||||
| -rw-r--r-- | src/liana/vcr.c | 196 | ||||
| -rw-r--r-- | src/liana/vcr.h | 16 | ||||
| -rw-r--r-- | src/libsink/sink.c | 508 | ||||
| -rw-r--r-- | src/libsink/sink.h | 1 | ||||
| -rw-r--r-- | src/render/renderer_libplacebo.c | 4 |
18 files changed, 624 insertions, 384 deletions
diff --git a/src/buffer/audio.c b/src/buffer/audio.c index 699b05e..377877e 100644 --- a/src/buffer/audio.c +++ b/src/buffer/audio.c @@ -10,7 +10,7 @@ #include "common.h" #include "common_internal.h" -#define BUFFER_SIZE 8.0 +#define BUFFER_SIZE 6.0 #define BUFFER_MARK_MIN 3.25 // Must be a most half of the buffer size. #define BUFFER_MARK_BUFFERED 0.35 @@ -151,9 +151,9 @@ static bool push_internal(struct camu_audio_buffer *buf, f64 pts, u8 **data, s32 // The maximum space is buf->size - 1. ptrdiff_t space = al_ring_buffer_space(&buf->rb); if (!buf->buffered && (buf->size - 1) - space >= buf->mark.buffered) { + buf->buffered = true; log_debug("Buffered (mark: %.1fKB).", buf->mark.buffered / 1024.0); buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); - buf->buffered = true; } ptrdiff_t have = (ptrdiff_t)camu_audio_format_samples_to_bytes(&buf->fmt.req, sample_count); @@ -198,7 +198,15 @@ static void push_av_frame_internal(struct camu_audio_buffer *buf, AVFrame *frame // A reset() must finish before any data is pushed. void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_frame *frame) { - al_assert(al_atomic_load(u8)(&buf->flow, AL_ATOMIC_RELAXED) == FLOWING); + u8 flow = al_atomic_load(u8)(&buf->flow, AL_ATOMIC_RELAXED); + // flow could be ERRORED here. + if (flow != FLOWING) { + // Assert that push() is never called after flush(). + al_assert(flow != FLUSHED); + camu_codec_frame_discard(frame); + return; + } + switch (frame->mode) { case CAMU_NORMAL: { s32 sample_count = frame->audio.sample_count; @@ -214,22 +222,28 @@ void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_fra } #endif } + al_free(frame); + +#ifdef CAMU_BUFFER_SPORADIC_ERRORS + ROLL_FOR_BUFFER_ERROR(buf); +#endif } // flush() always comes from the same thread as push(). -void camu_audio_buffer_flush(struct camu_audio_buffer *buf) +void camu_audio_buffer_flush(struct camu_audio_buffer *buf, bool error) { log_debug("Flush requested."); + u8 flow = error ? FLUSHED_ERROR : FLUSHED; + al_atomic_store(u8)(&buf->flow, flow, AL_ATOMIC_RELAXED); if (!push_internal(buf, 0.0, NULL, 0)) { log_debug("Buffer filled by flush."); } if (!buf->buffered) { + buf->buffered = true; log_debug("Buffered (flush)."); buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); - buf->buffered = true; } - al_atomic_store(u8)(&buf->flow, FLUSHED, AL_ATOMIC_RELAXED); } // Not thread-safe, must be called while the buffer is not being read from or written to. @@ -250,8 +264,19 @@ void camu_audio_buffer_resync(struct camu_audio_buffer *buf) ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdiff_t req) { + // Assert this buffer isn't being read before we signaled BUFFER_BUFFERED. al_assert(buf->buffered); + u8 flow = al_atomic_load(u8)(&buf->flow, AL_ATOMIC_ACQUIRE); + if (UNLIKELY(flow == ERRORED || flow == FLUSHED_ERROR)) { + if (flow == FLUSHED_ERROR) { + buf->callback(buf->userdata, CAMU_BUFFER_ERRORED); + al_atomic_store(u8)(&buf->flow, ERRORED, AL_ATOMIC_RELEASE); + } + al_memset(data, 0, req); + return req; + } + struct camu_audio_format *fmt = &buf->fmt.req; ptrdiff_t ret, signal = req; @@ -354,7 +379,6 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif buf->pause = PAUSE_PLAYING; } - u8 flow = al_atomic_load(u8)(&buf->flow, AL_ATOMIC_ACQUIRE); if (UNLIKELY(have < req)) { // We don't have enough data to fulfill our request. if (flow == FLUSHED) { // Stream is flushed. // Check peak buffer for any remaining data. diff --git a/src/buffer/audio.h b/src/buffer/audio.h index 78a12dc..a740c6c 100644 --- a/src/buffer/audio.h +++ b/src/buffer/audio.h @@ -68,7 +68,7 @@ void camu_audio_buffer_set_latency(struct camu_audio_buffer *buf, f64 latency); void camu_audio_buffer_set_ignore_desync(struct camu_audio_buffer *buf, bool ignore_desync); void camu_audio_buffer_set_no_video(struct camu_audio_buffer *buf, bool no_video); void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_frame *frame); -void camu_audio_buffer_flush(struct camu_audio_buffer *buf); +void camu_audio_buffer_flush(struct camu_audio_buffer *buf, bool error); void camu_audio_buffer_reset(struct camu_audio_buffer *buf); void camu_audio_buffer_resync(struct camu_audio_buffer *buf); ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdiff_t req); diff --git a/src/buffer/common.h b/src/buffer/common.h index 24964ba..bd9ca45 100644 --- a/src/buffer/common.h +++ b/src/buffer/common.h @@ -1,5 +1,15 @@ #pragma once +//#define CAMU_BUFFER_SPORADIC_ERRORS +#ifdef CAMU_BUFFER_SPORADIC_ERRORS +#include <al/random.h> +#define ROLL_FOR_BUFFER_ERROR(buf) do { \ + if (al_random_int(0, 254) == 72) { \ + al_atomic_store(u8)(&(buf)->flow, FLUSHED_ERROR, AL_ATOMIC_RELAXED); \ + } \ +} while (0) +#endif + enum { CAMU_BUFFER_BUFFERED = 0, CAMU_BUFFER_CORK, diff --git a/src/buffer/common_internal.h b/src/buffer/common_internal.h index 700e278..a9e6ece 100644 --- a/src/buffer/common_internal.h +++ b/src/buffer/common_internal.h @@ -9,6 +9,8 @@ enum { FLOWING, // Got request to flush. FLUSHED, + // Flushed because of an error. + FLUSHED_ERROR, // Realized flush. SIGNALED, // Errored. diff --git a/src/buffer/video.c b/src/buffer/video.c index 0e9ce28..d16a106 100644 --- a/src/buffer/video.c +++ b/src/buffer/video.c @@ -26,7 +26,7 @@ bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *cl buf->clock = clock; buf->latency = 0.0; al_atomic_store(f64)(&buf->pts, -1.0, AL_ATOMIC_RELAXED); - buf->reset_pts = -1.0; + buf->seek_pts = -1.0; buf->single_frame = true; buf->queue = NULL; buf->buffered = false; @@ -117,27 +117,6 @@ void camu_video_buffer_set_latency(struct camu_video_buffer *buf, f64 latency) buf->latency = latency; } -static void after_push_internal(struct camu_video_buffer *buf) -{ - // Note that queue->reset() must only be called from the read() thread. - // If a reset where to happen from the this thread, the latest read() - // frame could be freed before it was used. - s32 count = buf->queue->count(buf->queue); - if (!buf->buffered && (buf->single_frame || count >= BUFFER_MARK_BUFFERED)) { - // Preserve order of: set flow -> flush -> callback, for single frames. - if (buf->single_frame) { - al_atomic_store(u8)(&buf->flow, FLUSHED, AL_ATOMIC_RELAXED); - buf->queue->flush(buf->queue); - } - buf->buffered = true; - buf->buffered_with_one_frame = count == 1; - log_debug("Buffered (mark: %.2fs).", count * buf->avg_frame_duration); - buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); - } else if (count >= BUFFER_MARK_HIGH) { - buf->callback(buf->userdata, CAMU_BUFFER_CORK); - } -} - #ifdef CAMU_HAVE_FFMPEG static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame) { @@ -163,11 +142,16 @@ static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_frame *frame) { - if (buf->single_frame && buf->buffered) { - log_warn("Unexpected duplicate frame received."); + u8 flow = al_atomic_load(u8)(&buf->flow, AL_ATOMIC_ACQUIRE); + if (flow != FLOWING) { + // A single frame will be FLUSHED after any push(). + if (buf->single_frame) { + log_warn("Unexpected duplicate frame received."); + } camu_codec_frame_discard(frame); return; } + switch (frame->mode) { case CAMU_NORMAL: { f64 base_pts = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE); @@ -188,7 +172,25 @@ void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_fra } #endif } - after_push_internal(buf); + + s32 count = buf->queue->count(buf->queue); + if (!buf->buffered && (buf->single_frame || count >= BUFFER_MARK_BUFFERED)) { + // Preserve order of: set flow -> flush -> callback, for single frames. + if (buf->single_frame) { + al_atomic_store(u8)(&buf->flow, FLUSHED, AL_ATOMIC_RELEASE); + buf->queue->flush(buf->queue); + } + buf->buffered = true; + buf->buffered_with_one_frame = count == 1; + log_debug("Buffered (mark: %.2fs).", count * buf->avg_frame_duration); + buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); + } else if (count >= BUFFER_MARK_HIGH) { + buf->callback(buf->userdata, CAMU_BUFFER_CORK); + } + +#ifdef CAMU_BUFFER_SPORADIC_ERRORS + ROLL_FOR_BUFFER_ERROR(buf); +#endif } void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struct camu_codec_packet *packet) @@ -197,10 +199,11 @@ void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struct camu_ } // flush() always comes from the same thread as push(). -void camu_video_buffer_flush(struct camu_video_buffer *buf) +void camu_video_buffer_flush(struct camu_video_buffer *buf, bool error) { log_debug("Flush requested."); - al_atomic_store(u8)(&buf->flow, FLUSHED, AL_ATOMIC_RELAXED); + u8 flow = error ? FLUSHED_ERROR : FLUSHED; + al_atomic_store(u8)(&buf->flow, flow, AL_ATOMIC_RELAXED); buf->queue->flush(buf->queue); if (!buf->buffered) { s32 count = buf->queue->count(buf->queue); @@ -212,9 +215,11 @@ void camu_video_buffer_flush(struct camu_video_buffer *buf) } // Not thread-safe, must be called while the buffer is not being read from or written to. +// If a queue->reset() happened from the push() thread while the buffer was active, +// the latest read() frame could be freed before it was used. void camu_video_buffer_reset(struct camu_video_buffer *buf, f64 pts) { - buf->reset_pts = pts; + buf->seek_pts = pts; al_atomic_store(f64)(&buf->pts, -1.0, AL_ATOMIC_RELAXED); if (buf->queue) buf->queue->reset(buf->queue); buf->buffered = false; @@ -223,6 +228,7 @@ void camu_video_buffer_reset(struct camu_video_buffer *buf, f64 pts) bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weighted) { + // Assert this buffer isn't being read before we signaled BUFFER_BUFFERED. al_assert(buf->buffered); f64 base_pts = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE); @@ -232,9 +238,9 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weig } u8 flow = al_atomic_load(u8)(&buf->flow, AL_ATOMIC_ACQUIRE); - if (flow == ERRORED) return false; + if (flow == ERRORED) return false; u8 ret = buf->queue->read(buf->queue, base_pts, out); - if (ret == CAMU_QUEUE_ERR) { + if (ret == CAMU_QUEUE_ERR || flow == FLUSHED_ERROR) { buf->callback(buf->userdata, CAMU_BUFFER_ERRORED); al_atomic_store(u8)(&buf->flow, ERRORED, AL_ATOMIC_RELEASE); return false; diff --git a/src/buffer/video.h b/src/buffer/video.h index 380943e..fdd7d0b 100644 --- a/src/buffer/video.h +++ b/src/buffer/video.h @@ -17,7 +17,7 @@ struct camu_video_buffer { struct camu_clock *clock; f64 latency; atomic(f64) pts; - f64 reset_pts; + f64 seek_pts; bool single_frame; f64 avg_frame_duration; @@ -27,7 +27,7 @@ struct camu_video_buffer { struct camu_frame_queue *queue; bool buffered; - // We need to manually trigger EOF in read(). + // Need to manually trigger EOF in read(). bool buffered_with_one_frame; // Flush the renderer on this read and don't allow it to set the clock. bool weighted_read; @@ -38,7 +38,7 @@ struct camu_video_buffer { atomic(u8) ref; #endif - // Previous view, set from screen. + // Previous view, set from screen::add_buffer_internal(). struct camu_view view; void (*callback)(void *, u8); @@ -52,7 +52,7 @@ bool camu_video_buffer_configure_subtitles(struct camu_video_buffer *buf, struct void camu_video_buffer_set_latency(struct camu_video_buffer *buf, f64 latency); void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_frame *frame); void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struct camu_codec_packet *packet); -void camu_video_buffer_flush(struct camu_video_buffer *buf); +void camu_video_buffer_flush(struct camu_video_buffer *buf, bool error); void camu_video_buffer_reset(struct camu_video_buffer *buf, f64 pts); bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weighted); void camu_video_buffer_free(struct camu_video_buffer *buf); diff --git a/src/buffer/video_null.h b/src/buffer/video_null.h index 06b284c..f82a4d3 100644 --- a/src/buffer/video_null.h +++ b/src/buffer/video_null.h @@ -8,6 +8,7 @@ struct camu_video_buffer { struct camu_codec_stream *stream; + f64 seek_pts; bool single_frame; f64 avg_frame_duration; #ifdef CAMU_SCREEN_THREADED @@ -47,7 +48,7 @@ static bool camu_video_buffer_configure_subtitles(struct camu_video_buffer *buf, { (void)buf; (void)stream; - return true; + return false; } static void camu_video_buffer_set_latency(struct camu_video_buffer *buf, s32 frames) @@ -69,9 +70,10 @@ static void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struc (void)packet; } -static void camu_video_buffer_flush(struct camu_video_buffer *buf) +static void camu_video_buffer_flush(struct camu_video_buffer *buf, bool error) { (void)buf; + (void)error; } static void camu_video_buffer_reset(struct camu_video_buffer *buf, f64 pts) diff --git a/src/codec/ffmpeg/decoder.c b/src/codec/ffmpeg/decoder.c index 6eff28b..645a88a 100644 --- a/src/codec/ffmpeg/decoder.c +++ b/src/codec/ffmpeg/decoder.c @@ -7,7 +7,7 @@ #ifdef CAMU_FF_DECODER_HWACCEL #include "../../render/renderer.h" -static const char *hwdevces[] = { +static const char *hwdevices[] = { #ifdef CAMU_HAVE_VULKAN_DECODE "vulkan", #endif @@ -75,7 +75,7 @@ static s32 init_hwframe_context(struct camu_ff_decoder *av, AVCodecContext *cont frames_ctx->sw_format = av->codec_context->sw_pix_fmt; frames_ctx->width = av->codec_context->width; frames_ctx->height = av->codec_context->height; - frames_ctx->initial_pool_size = 20; + frames_ctx->initial_pool_size = 12; s32 ret = av_hwframe_ctx_init(hw_frames_ref); if (ret < 0) { @@ -114,8 +114,10 @@ static enum AVPixelFormat get_hw_format(AVCodecContext *context, const enum AVPi log_error("Failed to get %s HW surface format.", hwdevice_name); // Remake AVCodecContext as a software decoder. - AVCodecParameters *codecpar = avcodec_parameters_alloc(); - avcodec_parameters_from_context(codecpar, av->codec_context); + // This might log an error like "Your platform doesn't support hardware accelerated AV1 decoding.". + // Even though the problem is really a lack of any decoder hardware OR software. + // This happens with wine + d3d11va. Maybe a build configuration issue or FFmpeg bug? + AVCodecParameters *codecpar = av->codecpar; av->errored_hw_context = av->codec_context; alloc_codec_context_internal(av, codec, codecpar); // Note about thread_type choice in ff_decoder_init(). @@ -131,7 +133,13 @@ out: static s32 init_hwdevice_context(struct camu_ff_decoder *av, AVCodecContext *context) { + if (av->hw_device_type == AV_HWDEVICE_TYPE_D3D11VA) { + log_info("av_hwdevice_ctx_create()."); + } s32 ret = av_hwdevice_ctx_create(&av->hw_context, av->hw_device_type, NULL, NULL, 0); + if (av->hw_device_type == AV_HWDEVICE_TYPE_D3D11VA) { + log_info("av_hwdevice_ctx_create() returns."); + } if (ret < 0) { log_error("Failed to create specified HW device."); return ret; @@ -139,7 +147,7 @@ static s32 init_hwdevice_context(struct camu_ff_decoder *av, AVCodecContext *con context->hw_device_ctx = av_buffer_ref(av->hw_context); // Note that context->extra_hw_frames has the ability to cause corruption. - //context->extra_hw_frames = 18; + context->extra_hw_frames = 12; return ret; } @@ -231,6 +239,8 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend #endif AVCodecParameters *codecpar = stream->av.stream->codecpar; + av->codecpar = codecpar; + const AVCodec *codec = avcodec_find_decoder(codecpar->codec_id); if (!codec) { log_error("Failed to find decoder."); @@ -244,9 +254,9 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend collect_supported_hwaccels(av); av->hw_device_type = AV_HWDEVICE_TYPE_NONE; av->hw_pix_fmt = AV_PIX_FMT_NONE; - for (u32 i = 0; i < ARRAY_SIZE(hwdevces); i++) { - size_t length = sizeof(hwdevces[i]) - 1; - av->hw_device_type = hw_device_supported_by_name(av, hwdevces[i], length); + for (u32 i = 0; i < ARRAY_SIZE(hwdevices); i++) { + size_t length = sizeof(hwdevices[i]) - 1; + av->hw_device_type = hw_device_supported_by_name(av, hwdevices[i], length); if (!get_hwdevice_config(av, codec, av->hw_device_type)) { av->hw_device_type = AV_HWDEVICE_TYPE_NONE; } else { diff --git a/src/codec/ffmpeg/decoder.h b/src/codec/ffmpeg/decoder.h index ff513b9..0c3d4d7 100644 --- a/src/codec/ffmpeg/decoder.h +++ b/src/codec/ffmpeg/decoder.h @@ -10,6 +10,7 @@ struct camu_ff_decoder { struct camu_decoder dec; + AVCodecParameters *codecpar; AVCodecContext *codec_context; s32 thread_count; #ifdef CAMU_FF_DECODER_HWACCEL diff --git a/src/liana/client.c b/src/liana/client.c index 581bdba..9f7940f 100644 --- a/src/liana/client.c +++ b/src/liana/client.c @@ -114,16 +114,12 @@ static void collect_streams(struct lia_client *client, struct nn_packet *packet) al_array_sort(client->streams, struct camu_codec_stream, stream_compare); } -static u8 type_to_mask[] = { - [CAMU_STREAM_AUDIO] = CAMU_MASK_AUDIO, - [CAMU_STREAM_VIDEO] = CAMU_MASK_VIDEO, - [CAMU_STREAM_SUBTITLE] = CAMU_MASK_SUBTITLE -}; - -static const char *type_to_str[] = { +static const char *stream_type_to_str[] = { [CAMU_STREAM_AUDIO] = "audio", [CAMU_STREAM_VIDEO] = "video", - [CAMU_STREAM_SUBTITLE] = "subtitle" + [CAMU_STREAM_SUBTITLE] = "subtitle", + [CAMU_STREAM_ATTACHMENT] = "attachment", + [CAMU_STREAM_UNKNOWN] = "unknown" }; static void parse_info_packet(struct lia_client *client, struct nn_packet *packet) @@ -138,7 +134,7 @@ static void parse_info_packet(struct lia_client *client, struct nn_packet *packe for (; accept_defaults < 2; accept_defaults++) { struct camu_codec_stream *stream; al_array_foreach_ptr(client->streams, i, stream) { - u8 type_mask = type_to_mask[stream->type]; + u8 type_mask = 1 << stream->type; if ((selected & type_mask) || !(prefs->enabled_mask & type_mask)) continue; const char *title = NULL; #ifdef CAMU_HAVE_FFMPEG @@ -166,9 +162,9 @@ static void parse_info_packet(struct lia_client *client, struct nn_packet *packe } #endif if (title) { - log_info("Selected %s stream (index: %u, title: %s).", type_to_str[stream->type], stream->index, title); + log_info("Selected %s stream (index: %u, title: %s).", stream_type_to_str[stream->type], stream->index, title); } else { - log_info("Selected %s stream (index: %u).", type_to_str[stream->type], stream->index); + log_info("Selected %s stream (index: %u).", stream_type_to_str[stream->type], stream->index); } selected |= type_mask; client->mask |= 1 << stream->index; @@ -219,7 +215,7 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream) struct lia_client *client = (struct lia_client *)userdata; if (client->reconnect == RECONNECT_SIGNAL_CLIENT) { client->reconnect = RECONNECT_NONE; - // Even if seek() was called before the initial connection_callback(), + // Even if client_seek() was called before the initial connection_callback(), // we still want to call RESUME_AT here. struct lia_timing time = { .at = client->at, @@ -227,15 +223,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); - struct lia_reconnect_info rec = { - .reconnect = true, - .unconfigured = client->mask == 0 - }; - if (rec.unconfigured) { + // The value of client->mask will not have changed since connection_closed_callback(). + if (client->rec.unconfigured) { al_assert(client->connection_id == 0); log_warn("Handling reconnect on unconfigured client."); } - client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &rec); + client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &client->rec); } else { al_assert(client->connection_id == 0); } @@ -263,22 +256,42 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream * } else { lia_vcr_close_all(&client->vcr); } - struct lia_reconnect_info rec = { - .reconnect = client->reconnect == RECONNECT_ON_CONNECTION_CLOSED, - .unconfigured = client->mask == 0 - }; - // We need to account for REMOVE_BUFFERS possibly running the event loop to wait. - client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &rec); + // If reconnect = SIGNAL_CLIENT, we either never connected or recursed at the reconnect step. + if (client->reconnect != RECONNECT_SIGNAL_CLIENT) { + client->rec = (struct lia_reconnect_info){ + .reconnect = client->reconnect == RECONNECT_ON_CONNECTION_CLOSED, + .unconfigured = client->mask == 0, + .mask = client->mask + }; + al_array_init(client->rec.detached); + // CLIENT_REMOVE_BUFFERS should be allowed to run the event loop to wait and should + // attempt to maintain the same state if called consecutively. + client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &client->rec); + client->mask = client->rec.mask; + struct camu_codec_stream *detached; + al_array_foreach(client->rec.detached, i, detached) { + client->mask &= ~(1 << detached->index); + // We have to remove the track or it will erroneously receive a NULL packet on PACKET_EOF. + // It would also be wasteful to spin up a track_thread() for a removed track anyway. + bool removed = lia_vcr_remove_track_by_stream(&client->vcr, detached); + al_assert(removed); + } + al_array_free(client->rec.detached); + } if (client->reconnect == RECONNECT_ON_CONNECTION_CLOSED) { - // If reconnect() errors, this will close the client on recursion. + // If stream_reconnect() errors, the client will be closed on recursion. client->reconnect = RECONNECT_SIGNAL_CLIENT; + if (!client->mask) { + connection_closed_callback(userdata, stream); + } else { #ifdef CAMU_DIRECT_MODE - nn_multiplex_direct_reconnect(stream); + nn_multiplex_direct_reconnect(stream); #else - nn_packet_stream_reconnect(stream, &client->addr, client->port); + nn_packet_stream_reconnect(stream, &client->addr, client->port); #endif + } } else { - client->callback(client->userdata, LIANA_CLIENT_CLOSED, NULL, NULL); + client->callback(client->userdata, LIANA_CLIENT_CLOSED, NULL, &client->rec); } } @@ -317,11 +330,6 @@ void lia_client_seek(struct lia_client *client, u64 pos, u64 at) } } -void lia_client_reseek(struct lia_client *client) -{ - (void)client; -} - void lia_client_disconnect(struct lia_client *client) { u8 reconnect = client->reconnect; diff --git a/src/liana/client.h b/src/liana/client.h index eb84051..40093e0 100644 --- a/src/liana/client.h +++ b/src/liana/client.h @@ -15,12 +15,15 @@ enum { LIANA_CLIENT_RESUME_AT, LIANA_CLIENT_RECONNECTED, LIANA_CLIENT_EOF, + LIANA_CLIENT_ERRORED, LIANA_CLIENT_CLOSED }; struct lia_reconnect_info { bool reconnect; bool unconfigured; + u32 mask; + array(struct camu_codec_stream *) detached; }; struct lia_prefs { @@ -38,6 +41,7 @@ struct lia_client { u64 pos; u64 at; u8 reconnect; + struct lia_reconnect_info rec; str addr; u16 port; u32 connection_id; @@ -52,6 +56,5 @@ 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); 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); void lia_client_free(struct lia_client *client); diff --git a/src/liana/handlers/codec_client.c b/src/liana/handlers/codec_client.c index c246a24..abb5b9a 100644 --- a/src/liana/handlers/codec_client.c +++ b/src/liana/handlers/codec_client.c @@ -32,7 +32,12 @@ static bool codec_client_init(struct lia_client_handler *handler, struct camu_re #ifdef CAMU_HAVE_FFMPEG static bool push_av_packet(struct lia_codec_client *codec, AVPacket *pkt) { - return codec->dec->push_av_packet(codec->dec, pkt) == CAMU_OK; + s32 ret = codec->dec->push_av_packet(codec->dec, pkt); + if (ret == AVERROR(EAGAIN)) { + codec->dec->process(codec->dec); + ret = codec->dec->push_av_packet(codec->dec, pkt); + } + return ret == CAMU_OK; } static void passthrough_subtitle(struct lia_codec_client *codec, AVPacket *pkt) @@ -82,17 +87,17 @@ static bool codec_client_handle_packet(struct lia_client_handler *handler, struc u32 rindex = packet->rindex; bool success; u8 mode = nn_packet_read_u8(packet); + // @TODO: default: shouldn't be a case. We should check for an invalid packet. switch (mode) { case CAMU_NORMAL: { if (codec->dec) { - // @TODO: Store pointer over r/windex (64 bits) if needed. - //if (packet->opaque) { - // success = push_packet(codec, (struct nn_buffer *)packet->opaque); - //} else { + if (packet->opaque) { + success = push_packet(codec, (struct nn_buffer *)packet->opaque); + } else { struct nn_buffer buffer; nn_packet_read_buffer(packet, &buffer); success = push_packet(codec, &buffer); - //} + } } else { success = true; } @@ -101,13 +106,13 @@ static bool codec_client_handle_packet(struct lia_client_handler *handler, struc #ifdef CAMU_HAVE_FFMPEG case CAMU_FFMPEG_COMPAT: { AVPacket *pkt; - //if (packet->opaque) { - // pkt = (AVPacket *)packet->opaque; - //} else { + if (packet->opaque) { + pkt = (AVPacket *)packet->opaque; + } else { pkt = av_packet_alloc(); nn_packet_read_av_packet(packet, pkt); - // packet->opaque = pkt; - //} + packet->opaque = pkt; + } if (codec->dec) { success = push_av_packet(codec, pkt); if (!success) { @@ -137,20 +142,10 @@ static bool codec_client_handle_packet(struct lia_client_handler *handler, struc packet->rindex = rindex; if (!success) { - // Forcing in an EOF on an error is not necessary but behaves better in the sink. - codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_EOF, codec->handler.stream, NULL); + codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_ERRORED, codec->handler.stream, NULL); return false; } - // Process, if needed. - if (codec->dec) { - s32 ret = codec->dec->process(codec->dec); - if (!(ret == CAMU_ERR_AGAIN || ret == CAMU_ERR_EOF)) { - codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_EOF, codec->handler.stream, NULL); - return false; - } - } - return true; } diff --git a/src/liana/list.c b/src/liana/list.c index 9ac7e87..2430a9c 100644 --- a/src/liana/list.c +++ b/src/liana/list.c @@ -184,7 +184,11 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) } else { at = current->start; } - pause = LIANA_PAUSE_RESUME; + // This sink could have an entry set from a connection we no longer + // know about. In that case skipping to this entry with a pause_and_swap_to() + // would be better. If this sink is empty that's still okay because + // PAUSE_BOTH is required to handle that case sink-side. + pause = LIANA_PAUSE_BOTH; } struct lia_timing time = { .at = at, diff --git a/src/liana/vcr.c b/src/liana/vcr.c index e95922b..32c33cf 100644 --- a/src/liana/vcr.c +++ b/src/liana/vcr.c @@ -1,9 +1,11 @@ #define AL_LOG_SECTION "vcr" +//#define AL_LOG_ENABLE_TRACE #include <al/log.h> #include "handlers/handler.h" #include "vcr.h" +#include "list.h" #define VCR_BUFFER_BUFFERED MB(4) #define VCR_BUFFER_GROW_FACTOR 8 @@ -29,16 +31,58 @@ enum { static void signal_callback(void *userdata) { struct lia_vcr *vcr = (struct lia_vcr *)userdata; - nn_packet_stream_cork(vcr->data, false); + if (vcr->corked) { + nn_packet_stream_cork(vcr->data, false); + vcr->corked = false; + log_trace("Uncorked."); + u64 now = nn_get_timestamp(); + // This keeps the difference between `mark` and `now` equal to the amount + // of time we were actually receiving packets for. + al_assert(vcr->metrics.last_report_mark != LIANA_TIMESTAMP_INVALID); + vcr->metrics.last_report_mark += (now - vcr->metrics.last_cork_ts); + al_assert(vcr->metrics.last_report_mark < now); + vcr->metrics.last_cork_ts = LIANA_TIMESTAMP_INVALID; + } } -#endif static void reset_metrics(struct lia_vcr *vcr) { - vcr->metric.current_frame = 0; - vcr->metric.last_report_ts = 0; + vcr->metrics.current_frame = 0; + vcr->metrics.last_cork_ts = LIANA_TIMESTAMP_INVALID; + vcr->metrics.last_report_ts = LIANA_TIMESTAMP_INVALID; + vcr->metrics.last_report_mark = LIANA_TIMESTAMP_INVALID; + vcr->metrics.average_kbps = 0.f; } +static void update_metrics(struct lia_vcr *vcr, u64 size) +{ + vcr->metrics.current_frame += size; + log_trace("current_frame: %lu, corked: %s.", vcr->metrics.current_frame, BOOLSTR(vcr->corked)); + u64 now = nn_get_timestamp(); + if (vcr->metrics.last_report_mark == LIANA_TIMESTAMP_INVALID) { + vcr->metrics.last_report_ts = now; + vcr->metrics.last_report_mark = now; + return; + } + u64 diff = now - vcr->metrics.last_report_ts; + u64 mark = now - vcr->metrics.last_report_mark; + u64 frame = vcr->metrics.current_frame; + if (mark > 500000 || (diff > 2000000 && vcr->metrics.current_frame >= KB(500)) || (!size && frame > 0)) { + al_assert(frame > 0); + f32 kbps = (frame / 125.f) / (mark / 1000000.f); + f32 average_kbps = vcr->metrics.average_kbps; + average_kbps = average_kbps == 0.f ? kbps : (average_kbps + kbps) / 2.f; + f32 buffered = al_atomic_load(u64)(&vcr->count, AL_ATOMIC_RELAXED) / (f32)MB(1); + f32 capacity = vcr->mark.buffered / (f32)MB(1); + log_info("Receiving packets at %.2fkbps (%.2f/%.2fMB).", average_kbps, buffered, capacity); + vcr->metrics.average_kbps = average_kbps; + vcr->metrics.last_report_ts = now; + vcr->metrics.last_report_mark = now; + vcr->metrics.current_frame = 0; + } +} +#endif + void lia_vcr_init(struct lia_vcr *vcr, struct nn_event_loop *loop, struct nn_packet_stream *data, u16 node_id) { vcr->data = data; @@ -46,15 +90,14 @@ void lia_vcr_init(struct lia_vcr *vcr, struct nn_event_loop *loop, struct nn_pac al_array_init(vcr->tracks); al_atomic_store(u64)(&vcr->count, 0, AL_ATOMIC_RELAXED); vcr->mark.buffered = VCR_BUFFER_BUFFERED; - vcr->mark.low = 0; + al_atomic_store(u64)(&vcr->mark.low, 0, AL_ATOMIC_RELAXED); vcr->expand = VCR_EXPAND_UNTOUCHED; vcr->started = false; #ifndef CAMU_DIRECT_MODE + vcr->corked = false; nn_signal_init(&vcr->signal, loop, signal_callback, vcr); -#else - (void)loop; -#endif reset_metrics(vcr); +#endif } static void return_entire_cache(struct lia_vcr_track *track) @@ -65,6 +108,17 @@ static void return_entire_cache(struct lia_vcr_track *track) track->cache.cache.count = 0; } +// packet_cache_v2 +// - Each track has it's own cond. +// - Tracks get signaled in order of stream index. +// - Cache has to handle case where the next packet in order is +// from a track that has yet to call packet_cache_wait() (Immediate return). +// - Once track has read as much as it can. +// - If it read all avaliable packets, call packet_cache_wait() again and that entire +// section will be consumed. +// - If it was partial, call packet_cache_yield(<number_of_packets>) for that many packets +// to be consumed. + static nn_thread_result NNWT_THREADCALL vcr_track_thread(void *userdata) { nn_thread_set_priority(NNWT_THREAD_SCHED_FIFO, 32); @@ -98,7 +152,8 @@ static nn_thread_result NNWT_THREADCALL vcr_track_thread(void *userdata) u64 buffer = al_atomic_sub(u64)(&vcr->count, size, AL_ATOMIC_RELAXED); #ifndef CAMU_DIRECT_MODE bool buffered = al_atomic_load(bool)(&track->buffered, AL_ATOMIC_RELAXED); - if (buffered && buffer <= vcr->mark.low) { + u64 low = al_atomic_load(u64)(&vcr->mark.low, AL_ATOMIC_RELAXED); + if (buffered && (low && buffer <= low)) { nn_signal_send(&vcr->signal); } #else @@ -120,8 +175,7 @@ static nn_thread_result NNWT_THREADCALL vcr_track_thread(void *userdata) return 0; } - // Take state after handle_packet() because we may have been - // corked from within it. + // Take state again because handle_packet() could have caused the track to be corked. state = al_atomic_load(s32)(&track->state, AL_ATOMIC_RELAXED); // We wait if corked (TRACK_STOPPED) or EOF. @@ -201,11 +255,27 @@ void lia_vcr_add_track(struct lia_vcr *vcr, struct lia_vcr_track *track) nn_cond_init(&track->cond); nn_mutex_init(&track->mutex); al_atomic_store(bool)(&track->buffered, !VCR_TRACK_THREADED(track), AL_ATOMIC_RELAXED); + track->running = false; nn_packet_cache_init(&track->cache, 256); al_array_push(vcr->tracks, track); al_atomic_store(s32)(&track->state, VCR_TRACK_RUNNING, AL_ATOMIC_RELAXED); } +bool lia_vcr_remove_track_by_stream(struct lia_vcr *vcr, struct camu_codec_stream *stream) +{ + struct lia_vcr_track *track; + al_array_foreach(vcr->tracks, i, track) { + if (track->stream == stream) { + al_array_remove_at(vcr->tracks, i); + nn_packet_cache_free(&track->cache); + track->client->free(&track->client); + al_free(track); + return true; + } + } + return false; +} + bool lia_vcr_is_empty(struct lia_vcr *vcr) { return !vcr->tracks.count; @@ -215,7 +285,9 @@ static struct lia_vcr_track *get_track_from_index(struct lia_vcr *vcr, s32 index { struct lia_vcr_track *track; al_array_foreach(vcr->tracks, i, track) { - if (track->stream->index == index) return track; + if (track->stream->index == index) { + return track; + } } return NULL; } @@ -230,15 +302,23 @@ static void cork_if_buffered(struct lia_vcr *vcr, u64 buffer) } if (buffered) { if (vcr->expand == VCR_EXPAND_UNTOUCHED) { - vcr->mark.buffered = buffer * 8; + vcr->mark.buffered = buffer * VCR_BUFFER_GROW_FACTOR; vcr->expand = VCR_EXPAND_GROWN; log_info("Expanded buffer to size %.2fMB.", vcr->mark.buffered / (f32)MB(1)); return; - } else if (vcr->expand == VCR_EXPAND_GROWN) { - vcr->mark.low = vcr->mark.buffered - MB(2); + } + if (!vcr->corked) { + log_trace("Corked."); + nn_packet_stream_cork(vcr->data, true); + vcr->corked = true; + vcr->metrics.last_cork_ts = nn_get_timestamp(); + } + // Don't set low until after we corked so that vcr_track_thread() will never try uncorking + // until we know what the low mark is. + if (vcr->expand == VCR_EXPAND_GROWN) { + al_atomic_store(u64)(&vcr->mark.low, vcr->mark.buffered - VCR_BUFFER_LOW_OFFSET, AL_ATOMIC_RELAXED); vcr->expand = VCR_EXPAND_COMPLETE; } - nn_packet_stream_cork(vcr->data, true); al_array_foreach(vcr->tracks, i, track) { nn_packet_cache_flush(&track->cache); } @@ -246,29 +326,6 @@ static void cork_if_buffered(struct lia_vcr *vcr, u64 buffer) } #endif -static void update_metrics(struct lia_vcr *vcr, u32 size) -{ - vcr->metric.current_frame += size; - u64 now = nn_get_timestamp(); - if (!vcr->metric.last_report_ts) { - vcr->metric.last_report_ts = now; - return; - } - u64 diff; - if ((diff = now - vcr->metric.last_report_ts) > 1000000) { - vcr->metric.last_report_ts = now; - u64 frame = vcr->metric.current_frame; - vcr->metric.current_frame = 0; - if (diff > 3000000) { - return; - } - f32 kbps = (frame / 125.f) / (diff / 1000000.f); - f32 capacity = vcr->mark.buffered / (f32)MB(1); - f32 buffered = al_atomic_load(u64)(&vcr->count, AL_ATOMIC_RELAXED) / (f32)MB(1); - log_info("Receiving packets at %.2fkbps (%.2f/%.2fMB).", kbps, buffered, capacity); - } -} - void lia_vcr_push_packet(struct lia_vcr *vcr, struct nn_packet *packet) { struct lia_vcr_track *track; @@ -276,24 +333,24 @@ void lia_vcr_push_packet(struct lia_vcr *vcr, struct nn_packet *packet) switch (op) { case LIANA_PACKET_DATA: { u32 size = nn_packet_get_size(packet); +#ifndef CAMU_DIRECT_MODE update_metrics(vcr, size); +#endif s32 index = nn_packet_read_s32(packet); if (!(track = get_track_from_index(vcr, index))) { - log_debug("Received data from errored or unknown track (index: %d).", index); + log_error("Received data from an errored or unknown track (index: %d).", index); break; } if (VCR_TRACK_THREADED(track)) { - u64 buffer; - if ((buffer = al_atomic_add(u64)(&vcr->count, size, AL_ATOMIC_RELAXED)) >= vcr->mark.buffered) { + if (nn_packet_cache_send_packet(&track->cache, packet)) { + u64 buffer; + if ((buffer = al_atomic_add(u64)(&vcr->count, size, AL_ATOMIC_RELAXED)) >= vcr->mark.buffered) { #ifndef CAMU_DIRECT_MODE - cork_if_buffered(vcr, buffer); + cork_if_buffered(vcr, buffer); #endif + } + return; // Keep packet. } - if (!nn_packet_cache_send_packet(&track->cache, packet)) { - break; - } - // Keep packet. - return; } else { if (!track->client->handle_packet(track->client, packet)) { log_warn("Error handling non-buffered packet."); @@ -303,6 +360,10 @@ void lia_vcr_push_packet(struct lia_vcr *vcr, struct nn_packet *packet) } case LIANA_PACKET_EOF: case LIANA_PACKET_ERROR: { +#ifndef CAMU_DIRECT_MODE + update_metrics(vcr, 0); // Flush. +#endif + // @TODO: Should ERROR be passed down to LIANA_CLIENT_ERRORED? al_array_foreach(vcr->tracks, i, track) { nn_packet_cache_send_packet(&track->cache, NULL); } @@ -310,7 +371,7 @@ void lia_vcr_push_packet(struct lia_vcr *vcr, struct nn_packet *packet) nn_signal_stop(&vcr->signal); #endif if (op == LIANA_PACKET_EOF) { - log_debug("Received EOF."); + log_info("Received EOF."); } else if (op == LIANA_PACKET_ERROR) { log_warn("Forcing EOF because we got an error packet."); } @@ -367,13 +428,29 @@ static void vcr_track_close_internal(struct lia_vcr_track *track) } } -void lia_vcr_flush(struct lia_vcr *vcr) +static void vcr_close_all_internal(struct lia_vcr *vcr) { struct lia_vcr_track *track; al_array_foreach(vcr->tracks, i, track) { if (VCR_TRACK_THREADED(track)) { vcr_track_close_internal(track); return_entire_cache(track); + } + } + vcr->started = false; +#ifndef CAMU_DIRECT_MODE + vcr->corked = false; + nn_signal_stop(&vcr->signal); + reset_metrics(vcr); +#endif +} + +void lia_vcr_flush(struct lia_vcr *vcr) +{ + vcr_close_all_internal(vcr); + struct lia_vcr_track *track; + al_array_foreach(vcr->tracks, i, track) { + if (VCR_TRACK_THREADED(track)) { al_atomic_store(bool)(&track->buffered, false, AL_ATOMIC_RELAXED); track->client->flush(track->client); nn_packet_cache_enable(&track->cache); @@ -382,31 +459,16 @@ void lia_vcr_flush(struct lia_vcr *vcr) track->client->flush(track->client); } } - vcr->started = false; -#ifndef CAMU_DIRECT_MODE - nn_signal_stop(&vcr->signal); -#endif al_atomic_store(u64)(&vcr->count, 0, AL_ATOMIC_RELAXED); - reset_metrics(vcr); if (vcr->expand == VCR_EXPAND_COMPLETE) { - vcr->mark.low = 0; + al_atomic_store(u64)(&vcr->mark.low, 0, AL_ATOMIC_RELAXED); vcr->expand = VCR_EXPAND_GROWN; } } void lia_vcr_close_all(struct lia_vcr *vcr) { - struct lia_vcr_track *track; - al_array_foreach(vcr->tracks, i, track) { - if (VCR_TRACK_THREADED(track)) { - vcr_track_close_internal(track); - return_entire_cache(track); - } - } - vcr->started = false; -#ifndef CAMU_DIRECT_MODE - nn_signal_stop(&vcr->signal); -#endif + vcr_close_all_internal(vcr); } void lia_vcr_free(struct lia_vcr *vcr) diff --git a/src/liana/vcr.h b/src/liana/vcr.h index 167ce7f..1ab58da 100644 --- a/src/liana/vcr.h +++ b/src/liana/vcr.h @@ -13,12 +13,12 @@ struct lia_vcr_track { struct camu_codec_stream *stream; struct lia_client_handler *client; + struct nn_packet_cache cache; atomic(s32) state; atomic(bool) buffered; - struct nn_packet_cache cache; + bool running; struct nn_cond cond; struct nn_mutex mutex; - bool running; struct nn_thread thread; struct lia_vcr *vcr; }; @@ -28,21 +28,29 @@ struct lia_vcr { u16 node_id; array(struct lia_vcr_track *) tracks; atomic(u64) count; - struct { u64 buffered, low; } mark; + struct { + u64 buffered; + atomic(u64) low; + } mark; u8 expand; bool started; #ifndef CAMU_DIRECT_MODE + bool corked; struct nn_signal signal; #endif struct { u64 current_frame; + u64 last_cork_ts; u64 last_report_ts; - } metric; + u64 last_report_mark; + f32 average_kbps; + } metrics; }; void lia_vcr_init(struct lia_vcr *vcr, struct nn_event_loop *loop, struct nn_packet_stream *data, u16 node_id); void lia_vcr_start(struct lia_vcr *vcr); void lia_vcr_add_track(struct lia_vcr *vcr, struct lia_vcr_track *track); +bool lia_vcr_remove_track_by_stream(struct lia_vcr *vcr, struct camu_codec_stream *stream); bool lia_vcr_is_empty(struct lia_vcr *vcr); void lia_vcr_push_packet(struct lia_vcr *vcr, struct nn_packet *packet); void lia_vcr_set_buffered(struct lia_vcr_track *track); diff --git a/src/libsink/sink.c b/src/libsink/sink.c index f410636..aa9542e 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -25,6 +25,8 @@ enum { BUFFER_INIT = 0, // Set but not configured. BUFFER_QUEUED, + // Errored buffer was removed, treat it as empty from now on. + BUFFER_DETACHED, // Ready to receive data. BUFFER_CONFIGURED, // The next call to add can add the buffer. @@ -32,7 +34,9 @@ enum { // Treat the buffer like it's added, even if it might not be. BUFFER_ADDED, // Effectively SET_OR_BUFFERED but not addable until after a reset. - BUFFER_ENDED + BUFFER_ENDED, + // Same as BUFFER_ENDED but will be detached on a reset. + BUFFER_ERRORED }; // Command queue commands. @@ -54,18 +58,20 @@ enum { END }; -#define SINK_LRU_MAX UINT16_MAX - // Number of entries to keep buffered at one time. #define ENTRY_MAX_AGE 4 - -#define ENTRY_IS_VALID(entry) ((entry) && (entry) != (struct camu_sink_entry *)0xb00b) +#define SINK_LRU_MAX UINT16_MAX +AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); // Store connection number in the upper 16 bits so IDs don't conflict after a server restart. // If the sink disconnects but the server didn't restart, this will invalidate IDs that _do_ map -// to the same resource. So it's a trade off. -#define LOCAL_ENTRY_ID(sink, id) (((u64)(sink)->connection_number) << 47 | (u64)(id)) +// to the same resource. That could be wasteful. It's also currently (ab)used by reseek(). +#define LOCAL_ENTRY_ID(sink, id) (((u64)(sink)->connection_number) << 48 | (u64)(id)) #define REMOTE_ENTRY_ID(id) ((u32)(id & 0x7fffffff)) +#define CONNECTION_NUMBER(id) ((id >> 48) & 0xffff) + +// Only sink->target can ever be 0xb00b. +#define ENTRY_IS_VALID(entry) ((entry) && (entry) != (struct camu_sink_entry *)0xb00b) // printf format for entries. #ifdef AL_DEBUG @@ -80,21 +86,25 @@ enum { #define VIDEO_STATE(entry) ((entry)->video.state) // 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) -#define AUDIO_EMPTY(entry) BUFFER_EMPTY(&(entry)->audio) -#define VIDEO_EMPTY(entry) BUFFER_EMPTY(&(entry)->video) +// Also, if a buffer errors, it will be detached in CLIENT_REMOVE_BUFFERS. +// Empty is a state that cannot change while a buffer is being used (push()/read()). +#define AUDIO_EMPTY(entry) (AUDIO_STATE(entry) <= BUFFER_DETACHED) +#define VIDEO_EMPTY(entry) (VIDEO_STATE(entry) <= BUFFER_DETACHED) -#define AUDIO_NOT_ADDED(entry) ((entry)->audio.state != BUFFER_ADDED) -#define VIDEO_NOT_ADDED(entry) ((entry)->video.state != BUFFER_ADDED) +#define AUDIO_ENDED(entry) (AUDIO_STATE(entry) >= BUFFER_ENDED) +#define VIDEO_ENDED(entry) (VIDEO_STATE(entry) >= BUFFER_ENDED) +#define ENTRY_ENDED(entry) ((AUDIO_EMPTY(entry) || AUDIO_ENDED(entry)) && (VIDEO_EMPTY(entry) || VIDEO_ENDED(entry))) -#define AUDIO_ADDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->audio)) -#define VIDEO_ADDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->video)) +// IGNORED = ENDED or EMPTY. +#define AUDIO_ADDED_OR_IGNORED(entry) (AUDIO_STATE(entry) >= BUFFER_ADDED || AUDIO_EMPTY(entry)) +#define VIDEO_ADDED_OR_IGNORED(entry) (VIDEO_STATE(entry) >= BUFFER_ADDED || VIDEO_EMPTY(entry)) -#define AUDIO_ENDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->audio)) -#define VIDEO_ENDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->video)) +#define AUDIO_ENDED_OR_EMPTY(entry) (AUDIO_ENDED(entry) || AUDIO_EMPTY(entry)) +#define VIDEO_ENDED_OR_EMPTY(entry) (VIDEO_ENDED(entry) || VIDEO_EMPTY(entry)) -#define VIDEO_STREAM(entry) (entry)->video.buf.stream -#define VIDEO_IS_SINGLE_FRAME(entry) (entry)->video.buf.single_frame +#define AUDIO_STREAM(entry) ((al_assert(!AUDIO_EMPTY(entry)), (entry)->audio.buf.stream)) +#define VIDEO_STREAM(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.stream)) +#define VIDEO_IS_SINGLE_FRAME(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.single_frame)) #if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED #define BLOCKING_SLEEP(delay) nn_thread_sleep(delay) @@ -102,6 +112,17 @@ enum { #define BLOCKING_SLEEP(delay) nn_event_loop_sleep(sink->loop, delay) #endif +// Functions that might happen on separate threads. +// CAMU_MIXER_THREADED: +// audio_buffer_callback() +// mixer_callback() +// CAMU_SCREEN_THREADED: +// video_buffer_callback() +// CAMU_MIXER_THREADED or CAMU_SCREEN_THREADED: +// clock_callback() +// client_callback()::LIANA_CLIENT_DATA +// client_callback()::LIANA_CLIENT_EOF/ERRORED + static inline bool entry_audio_buffer_held(struct camu_sink_entry *entry) { #ifdef CAMU_MIXER_THREADED @@ -163,7 +184,7 @@ static inline void add_entry_video_buffer(struct camu_sink_entry *entry) static void remove_entry_audio_buffer(struct camu_sink_entry *entry) { - al_assert(AUDIO_STATE(entry) != BUFFER_ENDED); + al_assert(!AUDIO_ENDED(entry)); al_assert(AUDIO_STATE(entry) != BUFFER_INIT); switch (AUDIO_STATE(entry)) { case BUFFER_ADDED: @@ -191,7 +212,7 @@ static void remove_entry_audio_buffer(struct camu_sink_entry *entry) static void remove_entry_video_buffer(struct camu_sink_entry *entry) { // Don't assert !entry->ended here because of single frame handling. - al_assert(VIDEO_STATE(entry) != BUFFER_ENDED); + al_assert(!VIDEO_ENDED(entry)); al_assert(VIDEO_STATE(entry) != BUFFER_INIT); switch (VIDEO_STATE(entry)) { case BUFFER_ADDED: @@ -214,17 +235,16 @@ 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 handling in CLIENT_REMOVE_BUFFERS. -// BUFFER_ENDED: No-op'd in remove_entry_buffers() and add_or_queue_entry() but otherwise unchanged. +// BUFFER_ENDED: No-op'd in remove_entry_buffers(), add_or_queue_entry() and not added by do_add_entry(), otherwise unchanged. +// Another note: remove_entry_buffers() and add_or_queue_entry() are only ever called by switch_to(). +// switch_to() -> add_or_queue_entry(). +// switch_to() -> maybe_add_to_previous() -> (possibly delayed)remove_entry_buffers(). static void remove_entry_buffers(struct camu_sink_entry *entry) { log_trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); al_assert(!entry->ended); - if (AUDIO_STATE(entry) != BUFFER_ENDED) { - remove_entry_audio_buffer(entry); - } - if (VIDEO_STATE(entry) != BUFFER_ENDED) { - remove_entry_video_buffer(entry); - } + if (!AUDIO_ENDED(entry)) remove_entry_audio_buffer(entry); + if (!VIDEO_ENDED(entry)) remove_entry_video_buffer(entry); } static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry); @@ -237,12 +257,12 @@ static void add_or_queue_entry(struct camu_sink_entry *entry) al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); if (AUDIO_STATE(entry) == BUFFER_INIT) { AUDIO_STATE(entry) = BUFFER_QUEUED; - } else if (AUDIO_STATE(entry) != BUFFER_ENDED) { + } else if (!AUDIO_ENDED(entry)) { add_audio_if_set_and_buffered(entry); } if (VIDEO_STATE(entry) == BUFFER_INIT) { VIDEO_STATE(entry) = BUFFER_QUEUED; - } else if (VIDEO_STATE(entry) != BUFFER_ENDED) { + } else if (!VIDEO_ENDED(entry)) { add_video_if_set_and_buffered(entry); } } @@ -258,12 +278,12 @@ static void maybe_disconnect_entry(struct camu_sink_entry *entry) } #ifdef CAMU_SINK_LOCAL -static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *entry) +static void local_pause(struct camu_sink *sink, struct camu_sink_entry *entry) { - if (entry->held) return; if (!camu_clock_is_paused(&entry->clock)) { entry->paused = true; camu_clock_pause(&entry->clock, 0); + log_info("Clock paused."); // Audio stop will be handled by a BUFFER_PAUSED callback. if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) { #ifndef CAMU_SINK_NO_VIDEO @@ -274,16 +294,17 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent } else { entry->paused = false; camu_clock_resume(&entry->clock, 0); - if (!AUDIO_EMPTY(entry) && sink->audio.state == SINK_PAUSED) { - sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); - sink->audio.state = SINK_PLAYING; - } + log_info("Clock resumed."); if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PAUSED) { #ifndef CAMU_SINK_NO_VIDEO sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL); #endif sink->video.state = SINK_PLAYING; } + if (!AUDIO_EMPTY(entry) && sink->audio.state == SINK_PAUSED) { + sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); + sink->audio.state = SINK_PLAYING; + } } } #endif @@ -292,8 +313,10 @@ 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. + // If local, this could only lead to feeling like your inputs were eaten. #ifndef CAMU_SINK_LOCAL + // Entry is sent as an argument via get_entry_for_command(), so it can only + // be a valid entry or NULL and not a dangling target. if (entry) sequence = entry->sequence; #else (void)entry; @@ -395,23 +418,25 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) } case SKIP: { if (!sink->conn) return; + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; 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((struct camu_sink_entry *)cmd->opaque)); + nn_packet_write_s32(packet, get_sequence_for_command(entry)); nn_packet_write_s32(packet, (s32)cmd->value.i); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); break; } case TOGGLE_PAUSE: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; #ifdef CAMU_SINK_LOCAL - sink_local_pause(sink, (struct camu_sink_entry *)cmd->opaque); + if (!entry->held) local_pause(sink, entry); #else if (!sink->conn) return; 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((struct camu_sink_entry *)cmd->opaque)); + nn_packet_write_s32(packet, get_sequence_for_command(entry)); nn_packet_write_f64(packet, cmd->value.f); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); #endif @@ -419,10 +444,10 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) } case SEEK: { if (!sink->conn) return; + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; 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_SEEK); - struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; nn_packet_write_s32(packet, entry->sequence); nn_packet_write_u32(packet, REMOTE_ENTRY_ID(entry->id)); nn_packet_write_u64(packet, cmd->value.u); @@ -430,8 +455,9 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case RESEEK: { - struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; - lia_client_reseek(&entry->client); + if (!sink->conn) return; + // Crude way to trigger "re-add sink to list". + nn_rpc_conn_disconnect(sink->conn); break; } case SHUFFLE: { @@ -444,10 +470,10 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) } case END: { if (!sink->conn) return; + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; 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_END); - struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; nn_packet_write_u32(packet, REMOTE_ENTRY_ID(entry->id)); nn_packet_write_u32(packet, (u32)cmd->value.u); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); @@ -471,10 +497,11 @@ static void queue_signal_callback(void *userdata) static void mixer_callback(void *userdata, u8 op) { struct camu_sink *sink = (struct camu_sink *)userdata; + // @TODO: Isn't the idea of MIXER_EMPTY to not STOP audio when we know the mixer will be empty? if (op == CAMU_MIXER_EMPTY) { log_info("Mixer empty."); - // We need to sync with do_add_entry() because the order of - // the START/STOP's in the queue matters. + // Regardless of if we are checking an entry's state here, we have to sync with + // do_add_entry() because the order of START/STOPs in the queue matters. nn_mutex_lock(&sink->lock); // This feels a bit too loose. if (!(sink->current && AUDIO_STATE(sink->current) == BUFFER_ADDED)) { @@ -503,7 +530,6 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink) // indicative of the amount of entries we have loaded. // The most obvious reason being it's incremented when moving back // and forth between two entries. As well as for buffer and queue operations. - u16 max_age = ENTRY_MAX_AGE; // We have to handle sink->lru wrapping in a step before the default case. // 0 65532 65533 65534 65535 // 0 1 65533 65534 65535 @@ -512,21 +538,22 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink) // 0 1 2 3 4 struct camu_sink_entry *entry; al_array_foreach_rev(sink->entries, i, entry) { - if (entry->lru > sink->lru && ((SINK_LRU_MAX - entry->lru) + 1) + sink->lru >= max_age) { + if (sink->entries.count <= ENTRY_MAX_AGE) return; + u16 entry_age = (SINK_LRU_MAX - entry->lru) + sink->lru; + if (entry->lru > sink->lru && entry_age > ENTRY_MAX_AGE) { al_array_remove_at(sink->entries, i); maybe_disconnect_entry(entry); } - if (sink->entries.count <= max_age) return; } - // Make sure we don't have to consider wrapping here. - if (sink->lru >= max_age) { - al_array_foreach_rev(sink->entries, i, entry) { - al_assert(sink->lru >= entry->lru); - if (sink->lru - entry->lru >= max_age) { - al_array_remove_at(sink->entries, i); - maybe_disconnect_entry(entry); - } - if (sink->entries.count <= max_age) return; + // Make sure we don't have to consider wrapping in the second loop. + if (sink->lru < ENTRY_MAX_AGE) return; + al_array_foreach_rev(sink->entries, i, entry) { + al_assert(sink->lru >= entry->lru); + if (sink->entries.count <= ENTRY_MAX_AGE) return; + u16 entry_age = sink->lru - entry->lru; + if (entry_age > ENTRY_MAX_AGE) { + al_array_remove_at(sink->entries, i); + maybe_disconnect_entry(entry); } } } @@ -537,12 +564,21 @@ static void maybe_remove_previous(struct camu_sink *sink) struct camu_sink_entry *previous; al_array_foreach(sink->previous, i, previous) { remove_entry_buffers(previous); + // Cleanup entries from old connections. This is especially important to + // keep reseek() from being overly wasteful. + if (CONNECTION_NUMBER(previous->id) != sink->connection_number) { + queue_cmd(sink, (struct camu_sink_cmd){ + .op = EJECT_ENTRY, + .opaque = previous + }); + } } sink->previous.count = 0; } // 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. +// entry if it becomes incorrect to attempt removing it's buffers. The most obvious example +// being at the point it's freed. static void remove_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key) { bool removed = false; @@ -573,45 +609,49 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry al_array_push(sink->previous, previous); } -// Call this after setting the buffer's state to ADDED because this entry might be in previous. -static void do_add_entry(struct camu_sink_entry *entry) +static void after_add_entry(struct camu_sink_entry *entry, bool skip_audio, bool skip_video) { - // Single frames are unconditionally added in add_video_if_set_and_buffered(). - if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) { - add_entry_video_buffer(entry); - } - if (!AUDIO_EMPTY(entry)) { - add_entry_audio_buffer(entry); - } maybe_remove_previous(entry->sink); queue_cmd(entry->sink, (struct camu_sink_cmd){ - .op = (VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry) || entry->paused) ? STOP : START, + .op = (skip_video || entry->paused) ? STOP : START, .value.i = CAMU_SINK_VIDEO }); - if (VIDEO_EMPTY(entry)) { + if (VIDEO_ENDED_OR_EMPTY(entry)) { // Clear the screen if skipping from a video to an audio-only entry. refresh_video_output(entry->sink); } queue_cmd(entry->sink, (struct camu_sink_cmd){ - .op = (AUDIO_EMPTY(entry) || entry->paused) ? STOP : START, + .op = (skip_audio || entry->paused) ? STOP : START, .value.i = CAMU_SINK_AUDIO }); } +// Call this after setting state to ADDED because this entry might be in previous. +static void do_add_entry(struct camu_sink_entry *entry) +{ + bool skip_audio = AUDIO_ENDED_OR_EMPTY(entry); + // Single frames are unconditionally added in add_video_if_set_and_buffered(). + bool skip_video = VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); + // Whether to add audio or video first could be a consideration for responsiveness. + if (!skip_video) add_entry_video_buffer(entry); + if (!skip_audio) add_entry_audio_buffer(entry); + after_add_entry(entry, skip_audio, skip_video); +} + void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) { al_assert(!entry->ended); al_assert(AUDIO_STATE(entry) != BUFFER_INIT); al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); al_assert(AUDIO_STATE(entry) != BUFFER_ADDED); - al_assert(AUDIO_STATE(entry) != BUFFER_ENDED); + al_assert(!AUDIO_ENDED(entry)); switch (AUDIO_STATE(entry)) { case BUFFER_CONFIGURED: AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED; break; case BUFFER_SET_OR_BUFFERED: AUDIO_STATE(entry) = BUFFER_ADDED; - if (VIDEO_ADDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry)) { + if (VIDEO_ADDED_OR_IGNORED(entry) || VIDEO_IS_SINGLE_FRAME(entry)) { do_add_entry(entry); } break; @@ -625,7 +665,7 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) al_assert(VIDEO_STATE(entry) != BUFFER_INIT); al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); al_assert(VIDEO_STATE(entry) != BUFFER_ADDED); - al_assert(VIDEO_STATE(entry) != BUFFER_ENDED); + al_assert(!VIDEO_ENDED(entry)); switch (VIDEO_STATE(entry)) { case BUFFER_CONFIGURED: VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED; @@ -634,8 +674,11 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) VIDEO_STATE(entry) = BUFFER_ADDED; if (VIDEO_IS_SINGLE_FRAME(entry)) { add_entry_video_buffer(entry); - if (AUDIO_EMPTY(entry)) maybe_remove_previous(entry->sink); - } else if (AUDIO_ADDED_OR_EMPTY(entry)) { + bool skip_audio = AUDIO_ENDED_OR_EMPTY(entry); + if (skip_audio) { + after_add_entry(entry, skip_audio, true); + } + } else if (AUDIO_ADDED_OR_IGNORED(entry)) { do_add_entry(entry); } break; @@ -644,26 +687,31 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) { - log_trace("switch_to("ENTRY_FMT"), current: "ENTRY_FMT".", ENTRY_ARG(target), ENTRY_ARG(sink->current)); + struct camu_sink_entry *current = sink->current; + + log_trace("switch_to("ENTRY_FMT"(ended: %s)), current: "ENTRY_FMT"(ended: %s).", + ENTRY_ARG(target), BOOLSTR(ENTRY_IS_VALID(target) ? target->ended : false), + ENTRY_ARG(current), BOOLSTR(current ? current->ended : false)); bool ensure_removed = false; - struct camu_sink_entry *current = sink->current; if (current) { struct camu_sink_entry *detached = sink->detached; al_assert(current != target); ensure_removed = detached || current->ended; if (detached) { al_assert(detached == current); + // A buffer's state being QUEUED should be impossible while it's + // entry is reconnecting. + al_assert(AUDIO_STATE(detached) != BUFFER_QUEUED); + al_assert(VIDEO_STATE(detached) != BUFFER_QUEUED); sink->detached = NULL; - 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; + log_warn("Unset detached as a substitute for remove."); } else if (!current->ended) { maybe_add_to_previous(sink, current, target); } } + // @TODO: Cleanup stop_video? and log_trace lengths. bool stop_video = false; bool dangling_target = target == (struct camu_sink_entry *)0xb00b; if (!dangling_target) { @@ -671,19 +719,22 @@ 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_EMPTY(target) && VIDEO_IS_SINGLE_FRAME(target)) { + log_trace("Target ended in switch_to()."); + if (!VIDEO_ENDED_OR_EMPTY(target) && VIDEO_IS_SINGLE_FRAME(target)) { add_video_if_set_and_buffered(target); } stop_video = true; } + } else { + log_trace("Ignored dangling target."); } if (ensure_removed) { - if (!VIDEO_EMPTY(current) && VIDEO_IS_SINGLE_FRAME(current)) { + if (!VIDEO_ENDED_OR_EMPTY(current) && VIDEO_IS_SINGLE_FRAME(current)) { remove_entry_video_buffer(current); } - al_assert(AUDIO_NOT_ADDED(current)); - al_assert(VIDEO_NOT_ADDED(current)); + al_assert(AUDIO_STATE(current) != BUFFER_ADDED); + al_assert(VIDEO_STATE(current) != BUFFER_ADDED); } if (!dangling_target) { @@ -706,6 +757,20 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) } } +// @TODO: Could entry->ended be redundant? What about entry->held? +// - The catalyst for this is having to check entry->ended in CLIENT_REMOVE_BUFFERS breaking the ability +// to maintain the same state if called consecutively. +// - As long as end_entry_and_advance_queue() evaluates target, what is the point of +// ended except to say all of an entry's buffers are ended. +// (except to guess what the server thinks, which was it's original purpose). +// - Knowing if the outputs are paused is clearly necessary but what about it's relationship to held? +// - entry->paused basically means preempt any queued skip action. +// - On video or audio paused, check if SINK_EMPTY or PAUSED and run target (currently not handled at all). +// - Fix SINK_LOCAL relying on entry->paused hint from server (rely on clock state?). +// entry->paused reduced to nothing but a hint about if the entry is paused when skipping. +// Only consideration is for the sink outputs to never stay paused when playing and vise versa +// - Then entry->paused actually doesn't matter (sink state is above entry anyway and we should be able to +// rely on the clock per-entry). #ifndef CAMU_SINK_LOCAL static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at) { @@ -787,27 +852,26 @@ static void audio_buffer_callback(void *userdata, u8 op) nn_mutex_unlock(&sink->lock); break; case CAMU_BUFFER_EOF: - log_debug("Audio EOF."); + case CAMU_BUFFER_ERRORED: { + bool error = op == CAMU_BUFFER_ERRORED; + log_debug(error ? "Audio buffer errored." : "Audio EOF."); nn_mutex_lock(&sink->lock); - // Having threaded outputs means anything could have happened while waiting - // on the lock above. If we were locked in CLIENT_REMOVE_BUFFERS, state could very - // well be CONFIGURED here. + // EOF and ERRORED come from the outputs read() thread. So, having threaded + // outputs means anything could have happened while waiting on the lock above. + // For example, if we were locked in CLIENT_REMOVE_BUFFERS, state could have + // dropped all the way to CONFIGURED before we acquired the lock here. + // This should also maintain a consistent state in the more common case of switch_to() + // right before a buffer EOF. if (AUDIO_STATE(entry) == BUFFER_ADDED) { remove_entry_audio_buffer(entry); } - AUDIO_STATE(entry) = BUFFER_ENDED; + AUDIO_STATE(entry) = error ? BUFFER_ERRORED : BUFFER_ENDED; if (VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry)) { end_entry_and_advance_queue(sink, entry); } nn_mutex_unlock(&sink->lock); break; - case CAMU_BUFFER_ERRORED: - log_error("Audio buffer errored."); - queue_cmd(sink, (struct camu_sink_cmd){ - .op = EJECT_ENTRY, - .opaque = entry - }); - break; + } } } @@ -829,20 +893,23 @@ static void video_buffer_callback(void *userdata, u8 op) lia_vcr_uncork(entry->video.track); break; case CAMU_BUFFER_EOF: - log_debug("Video EOF."); + case CAMU_BUFFER_ERRORED: { + bool error = op == CAMU_BUFFER_ERRORED; + log_debug(error ? "Video buffer errored." : "Video EOF."); 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 (VIDEO_IS_SINGLE_FRAME(entry)) { + if (VIDEO_IS_SINGLE_FRAME(entry) && !error) { nn_mutex_unlock(&sink->lock); return; } + // Video state could be ADDED, SET_OR_BUFFERED, or CONFIGURED. + // See note about threaded outputs in audio_buffer_callback(EOF|ERRORED). if (VIDEO_STATE(entry) == BUFFER_ADDED) { remove_entry_video_buffer(entry); } - VIDEO_STATE(entry) = BUFFER_ENDED; + VIDEO_STATE(entry) = error ? BUFFER_ERRORED : BUFFER_ENDED; bool swapped = false; if (AUDIO_ENDED_OR_EMPTY(entry)) { swapped = end_entry_and_advance_queue(sink, entry); @@ -855,13 +922,7 @@ static void video_buffer_callback(void *userdata, u8 op) }); } break; - case CAMU_BUFFER_ERRORED: - log_error("Video buffer errored."); - queue_cmd(sink, (struct camu_sink_cmd){ - .op = EJECT_ENTRY, - .opaque = entry - }); - break; + } } } @@ -889,6 +950,7 @@ static void clock_callback(void *userdata, u8 op) static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry) { + // @TODO: Video can't be ENDED here right? bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); f64 avg_frame_duration = entry->video.buf.avg_frame_duration; #ifdef CAMU_SINK_LOCAL @@ -896,8 +958,8 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s f64 audio = camu_mixer_get_latency(sink->audio.mixer); struct camu_renderer *renderer = sink->video.renderer; f64 video = renderer->get_latency(renderer) * avg_frame_duration; - // Start either the audio or video early so we can start the clock - // as soon as possible while keeping A/V sync. + // Start either the audio or video early so we can start the clock as + // soon as possible while keeping A/V sync. if (audio > video) { camu_video_buffer_set_latency(&entry->video.buf, video - audio); } else if (video > audio) { @@ -1008,15 +1070,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } case LIANA_CLIENT_DATA: { struct camu_codec_frame *frame = (struct camu_codec_frame *)opaque; + // Even though !AUDIO/VIDEO_EMPTY() is a value that cannot change at this point, we + // still have to lock because AUDIO/VIDEO_STATE() is not atomic. switch (stream->type) { case CAMU_STREAM_AUDIO: - al_assert(!AUDIO_EMPTY(entry)); + nn_locked_assert(!AUDIO_EMPTY(entry), &sink->lock); camu_audio_buffer_push(&entry->audio.buf, frame); - return; + break; case CAMU_STREAM_VIDEO: - al_assert(!VIDEO_EMPTY(entry)); + nn_locked_assert(!VIDEO_EMPTY(entry), &sink->lock); camu_video_buffer_push(&entry->video.buf, frame); - return; + break; default: camu_codec_frame_discard(frame); break; @@ -1037,22 +1101,38 @@ 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; + if (entry == sink->current) { + if (!entry->ended) { + // It should only be possible for a buffer to be INIT if entry is not current or ended. + // Ended case is: (audio or video buffer empty) -> entry switched off of -> entry ended -> switch_to()'d. + // - The empty buffer doesn't get re-QUEUED because the entry is ended. + al_assert(AUDIO_STATE(entry) != BUFFER_INIT); + al_assert(VIDEO_STATE(entry) != BUFFER_INIT); + } + if (sink->target) { + // This is necessary to avoid re-adding an entry with an in-between clock state. + // See note in clock.c::camu_clock_seek(). + switch_to(sink, sink->target); + sink->target = NULL; + } + } + + // If this entry is still current on CLIENT_RECONNECTED, re-add it's buffers. + if (entry == sink->current && rec->reconnect) { + sink->detached = entry; } // 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. remove_previous_if_contains(sink, entry); + // AUDIO/VIDEO_STATE() could be INIT after this point, even if entry = current. - // If this entry is still current on CLIENT_RECONNECTED, re-add buffers. - if (rec->reconnect && entry == sink->current) { - sink->detached = entry; - } + // We should treat CLIENT_REMOVE_BUFFERS as a function that removes an entry's buffers and + // resets it's buffered state. For a non-empty entry that means remove_entry_audio/video_buffer() + // twice and for an empty buffer, knock it down to BUFFER_INIT. + if (AUDIO_STATE(entry) == BUFFER_QUEUED) AUDIO_STATE(entry) = BUFFER_INIT; + if (VIDEO_STATE(entry) == BUFFER_QUEUED) VIDEO_STATE(entry) = BUFFER_INIT; // Ignore unconfigured entries. if (rec->unconfigured) { @@ -1062,79 +1142,95 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str return; } - // An empty buffer is guaranteed to not be held. - if (!AUDIO_ENDED_OR_EMPTY(entry)) { + // An empty buffer is guaranteed to not be held. An ended buffer is already + // removed and will be further handled at the end of this case. + bool skip_audio = AUDIO_EMPTY(entry); + if (!skip_audio && !AUDIO_ENDED(entry)) { // Remove for re-add in CLIENT_RECONNECTED. remove_entry_audio_buffer(entry); // Remove again for another add in BUFFER_BUFFERED. remove_entry_audio_buffer(entry); } - bool skip_video = rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry); - if (!VIDEO_ENDED_OR_EMPTY(entry)) { // Single frames will never be ENDED. - if (!skip_video) { - remove_entry_video_buffer(entry); - remove_entry_video_buffer(entry); - } else { - // We are keeping the frame, so don't request a duplicate. - entry->client.mask &= ~(1 << VIDEO_STREAM(entry)->index); - } + bool skip_video = VIDEO_EMPTY(entry); + // Don't remove a single_frame if we are reconnecting, unless it's errored. + skip_video = skip_video || (rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry) && VIDEO_STATE(entry) != BUFFER_ERRORED); + if (!skip_video && !VIDEO_ENDED(entry)) { + remove_entry_video_buffer(entry); + remove_entry_video_buffer(entry); } - // Resolve any queued REMOVE_BUFFER requests before blocking. - // If MIXER_THREADED_START_STOP is not set, REMOVE_BUFFER happens from the sink's - // command queue. So, this is necessary for safely blocking the loop here. + // If MIXER_THREADED_START_STOP is not set, REMOVE_BUFFER is not thread-safe. + // Meaning it must be run on the event loop. So, resolve any queued REMOVE_BUFFER + // requests so we can safely block the loop. run_queue_by_opaque(sink, entry); // Unlock to wait. nn_mutex_unlock(&sink->lock); - while (entry_audio_buffer_held(entry) || (!skip_video && entry_video_buffer_held(entry))) { + while ((!skip_audio && entry_audio_buffer_held(entry)) || (!skip_video && entry_video_buffer_held(entry))) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); } + // At this point we can be sure that the entry's buffers are no longer in use. - // Reset possible ENDED state here in case the entry ended at some point after unlocking above. + // Re-lock to check an entries ended state. As it could have been set at some point after unlocking to wait. nn_mutex_lock(&sink->lock); + if (rec->reconnect) { + // Detach errored buffers. If we detach both streams, the entry will be closed. + if (AUDIO_STATE(entry) == BUFFER_ERRORED) { + al_array_push(rec->detached, AUDIO_STREAM(entry)); + } + if (VIDEO_STATE(entry) == BUFFER_ERRORED) { + al_array_push(rec->detached, VIDEO_STREAM(entry)); + } else if (!VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) { + // Don't request a duplicate single_frame. + rec->mask &= ~(1 << VIDEO_STREAM(entry)->index); + } + } + + // Finalize the removal of the buffers by handling ended (ENDED or ERRORED) buffers. + // ENDED: Set to CONFIGURED to emulate the two removes earlier in this case. + // ERRORED: Set to DETACHED and they will now be considered empty. if (AUDIO_STATE(entry) == BUFFER_ENDED) { al_assert(!AUDIO_EMPTY(entry)); AUDIO_STATE(entry) = BUFFER_CONFIGURED; - } else if (entry->ended && AUDIO_EMPTY(entry)) { - // If entry is ended, it may have forewent a remove_entry_buffers(). - AUDIO_STATE(entry) = BUFFER_INIT; + } else if (AUDIO_STATE(entry) == BUFFER_ERRORED) { + AUDIO_STATE(entry) = BUFFER_DETACHED; } if (VIDEO_STATE(entry) == BUFFER_ENDED) { al_assert(!VIDEO_EMPTY(entry)); VIDEO_STATE(entry) = BUFFER_CONFIGURED; - } else if (entry->ended && VIDEO_EMPTY(entry)) { - VIDEO_STATE(entry) = BUFFER_INIT; + } else if (VIDEO_STATE(entry) == BUFFER_ERRORED) { + VIDEO_STATE(entry) = BUFFER_DETACHED; } + // Unset ended, consistent with the logic in list. entry->ended = false; nn_mutex_unlock(&sink->lock); break; } - case LIANA_CLIENT_RESUME_AT: { + case LIANA_CLIENT_RESUME_AT: { // This is called after the client reconnects, before CLIENT_RECONNECTED. struct lia_timing *time = (struct lia_timing *)opaque; +#ifdef CAMU_SINK_LOCAL + // List has no concept of a local sink, ignore it's request. + time->at = 0; +#endif 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. + nn_mutex_lock(&sink->lock); 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. + // no_video is set to true in video BUFFER_EOF as a fail-safe. Reset it here. camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); } if (!ignore_video) { camu_video_buffer_reset(&entry->video.buf, time->seek_pos); } - 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 } @@ -1148,40 +1244,44 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str log_trace("reconnected("ENTRY_FMT"), detached: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), ENTRY_ARG(sink->detached), AUDIO_STATE(entry), VIDEO_STATE(entry)); if (entry == sink->detached) { al_assert(entry == sink->current); + // Let this be the only other explicit BUFFER_ENDED check, or this will + // become too complicated. al_assert(AUDIO_STATE(entry) != BUFFER_ENDED); al_assert(VIDEO_STATE(entry) != BUFFER_ENDED); + // The value of unconfigured remains consistent from CLIENT_REMOVE_BUFFERS. if (rec->unconfigured) { - // The value of unconfigured is consistent from CLIENT_REMOVE_BUFFERS. al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry)); - } else { - if (AUDIO_EMPTY(entry)) { - AUDIO_STATE(entry) = BUFFER_QUEUED; - } else { - add_audio_if_set_and_buffered(entry); - } - if (VIDEO_EMPTY(entry)) { - VIDEO_STATE(entry) = BUFFER_QUEUED; - } else if (!VIDEO_IS_SINGLE_FRAME(entry)) { - add_video_if_set_and_buffered(entry); - } + } + // An unconfigured entry would have still been queued if it was current. + if (AUDIO_STATE(entry) == BUFFER_INIT) { + AUDIO_STATE(entry) = BUFFER_QUEUED; + } else if (!AUDIO_EMPTY(entry)) { + add_audio_if_set_and_buffered(entry); + } + if (VIDEO_STATE(entry) == BUFFER_INIT) { + VIDEO_STATE(entry) = BUFFER_QUEUED; + } else if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) { + add_video_if_set_and_buffered(entry); } sink->detached = NULL; } nn_mutex_unlock(&sink->lock); break; } - case LIANA_CLIENT_EOF: { + case LIANA_CLIENT_EOF: + case LIANA_CLIENT_ERRORED: { + bool error = op == LIANA_CLIENT_ERRORED; switch (stream->type) { case CAMU_STREAM_AUDIO: { - if (!AUDIO_EMPTY(entry)) { - camu_audio_buffer_flush(&entry->audio.buf); - } + nn_locked_assert(!AUDIO_EMPTY(entry), &sink->lock); + camu_audio_buffer_flush(&entry->audio.buf, error); break; } case CAMU_STREAM_VIDEO: { + nn_locked_assert(!VIDEO_EMPTY(entry), &sink->lock); // Single frames are immediately flushed inside the buffer. - if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) { - camu_video_buffer_flush(&entry->video.buf); + if (!VIDEO_IS_SINGLE_FRAME(entry) || error) { + camu_video_buffer_flush(&entry->video.buf, error); } break; } @@ -1189,14 +1289,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str break; } case LIANA_CLIENT_CLOSED: { - // CLIENT_REMOVE_BUFFERS has been called on this entry before we're here. + struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque; nn_mutex_lock(&sink->lock); + // We can be assured that CLIENT_REMOVE_BUFFERS has been called on this entry. - if (VIDEO_STATE(entry) == BUFFER_ADDED) { - al_assert(VIDEO_IS_SINGLE_FRAME(entry)); + // Only possible case that a buffer could still be added. + if (rec->reconnect && !VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) { remove_entry_video_buffer(entry); while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); } } + al_assert(AUDIO_STATE(entry) != BUFFER_ADDED); + al_assert(VIDEO_STATE(entry) != BUFFER_ADDED); bool removed = al_array_remove(sink->entries, entry); remove_from_queue_by_opaque(sink, entry); @@ -1232,7 +1335,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str lia_client_free(&entry->client); camu_audio_buffer_free(&entry->audio.buf); camu_video_buffer_free(&entry->video.buf); - log_info("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "force" : "cleanup"); + log_warn("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "force" : "cleanup"); al_free(entry); break; @@ -1266,8 +1369,12 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id) camu_video_buffer_init(&entry->video.buf, &entry->clock); entry->video.buf.callback = video_buffer_callback; entry->video.buf.userdata = entry; + // Two entries in order could share the same pointer if the first + // entry was just freed. This has to be accounted for in the renderer + // cache or it won't update on the first frame of the new entry. + // This is most likely to happen when the sink reconnects to a server. union { f64 f; u64 u; } fv = { .u = id }; - entry->video.buf.reset_pts = fv.f; + entry->video.buf.seek_pts = fv.f; al_array_push(sink->entries, entry); @@ -1293,10 +1400,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, if (op == LIANA_SINK_UNSET) { nn_mutex_lock(&sink->lock); if (sink->current) { - queue_cmd(sink, (struct camu_sink_cmd){ - .op = EJECT_ENTRY, - .opaque = sink->current - }); + maybe_disconnect_entry(sink->current); } goto out; } @@ -1321,8 +1425,8 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, bool create = !entry; if (create) entry = create_entry(sink, id); entry->sequence = sequence; - sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX); entry->lru = sink->lru; + sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX); entry->reset_id = reset_id; if (create) { entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE; @@ -1330,8 +1434,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, seek_pos); } - // Don't lock before calling client_connect() or we could deadlock - // in CLIENT_CLOSED on a failed socket_connect(). + // Don't lock before client_connect() or we could deadlock in CLIENT_CLOSED on a failed socket_connect(). nn_mutex_lock(&sink->lock); if (op == LIANA_SINK_BUFFER) { @@ -1345,7 +1448,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, #ifdef CAMU_SINK_LOCAL (void)at; al_assert(entry != current); - log_trace("set("ENTRY_FMT"), %s[local], created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create)); + log_trace("set("ENTRY_FMT"), %s(local), created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create)); if (current) { current->audio.ignore_paused = true; if (!current->paused) { @@ -1458,29 +1561,31 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con log_trace("pause("ENTRY_FMT"), %s, audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), lia_pause_op_name(pause), AUDIO_STATE(entry), VIDEO_STATE(entry)); #ifdef CAMU_SINK_LOCAL (void)at; - sink_local_pause(sink, entry); + if (!entry->held) local_pause(sink, entry); #else switch (pause) { case LIANA_PAUSE_PAUSE: entry->paused = true; camu_clock_pause(&entry->clock, at); + log_info("Clock paused."); // Audio will be stopped in a BUFFER_PAUSED callback. // Video will be stopped in a CLOCK_PAUSED callback. break; case LIANA_PAUSE_RESUME: entry->paused = false; camu_clock_resume(&entry->clock, at); - if (!AUDIO_EMPTY(entry)) { - camu_audio_buffer_resync(&entry->audio.buf); + log_info("Clock resumed."); + if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) { queue_cmd(entry->sink, (struct camu_sink_cmd){ .op = START, - .value.i = CAMU_SINK_AUDIO + .value.i = CAMU_SINK_VIDEO }); } - if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) { + if (!AUDIO_EMPTY(entry)) { + camu_audio_buffer_resync(&entry->audio.buf); queue_cmd(entry->sink, (struct camu_sink_cmd){ .op = START, - .value.i = CAMU_SINK_VIDEO + .value.i = CAMU_SINK_AUDIO }); } break; @@ -1570,18 +1675,20 @@ static void connection_closed_callback(void *userdata, struct nn_rpc_connection { struct camu_sink *sink = (struct camu_sink *)userdata; bool reconnect = !sink->reconnect_timer.disabled; + bool disconnected = sink->conn && sink->connection_number > 0; + if (sink->conn) { + al_assert(sink->conn == conn); + sink->conn = NULL; + } if (reconnect) { - if (sink->conn && sink->connection_number > 0) { + if (disconnected) { log_info("Connection to server closed, attempting reconnect..."); + nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); } else { log_info("Failed to connect to server, trying again..."); + nn_timer_again(&sink->reconnect_timer); } } - if (sink->conn) { - al_assert(sink->conn == conn); - sink->conn = NULL; - } - if (reconnect) nn_timer_again(&sink->reconnect_timer); } bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, @@ -1593,7 +1700,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, sink->connection_number = 0; nn_mutex_init(&sink->lock); nn_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink); - nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(2000000)); + nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1500000)); nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink); nn_signal_start(&sink->queue_signal); camu_queue_init(sink->queue); @@ -1603,7 +1710,8 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, sink->detached = NULL; al_array_init(sink->previous); al_array_init(sink->entries); - sink->lru = 0; + // Start high to exercise the wrapping path. + sink->lru = SINK_LRU_MAX - 2; mixer->callback = mixer_callback; mixer->userdata = sink; sink->audio.state = SINK_PAUSED; @@ -1644,6 +1752,7 @@ void camu_sink_return_current(struct camu_sink *sink) nn_mutex_unlock(&sink->lock); } +// sink->current could be NULL. static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink) { return ENTRY_IS_VALID(sink->target) ? sink->target : sink->current; @@ -1714,13 +1823,8 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode) void camu_sink_reseek(struct camu_sink *sink) { - nn_mutex_lock(&sink->lock); - struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->lock); - if (!current) return; queue_cmd(sink, (struct camu_sink_cmd){ - .op = RESEEK, - .opaque = current + .op = RESEEK }); } diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 3face58..f2ae20f 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -91,6 +91,7 @@ struct camu_sink { struct camu_sink_entry *current; struct camu_sink_entry *queued; struct camu_sink_entry *target; + // @TODO: Rename detached to reconnecing. struct camu_sink_entry *detached; array(struct camu_sink_entry *) previous; array(struct camu_sink_entry *) entries; diff --git a/src/render/renderer_libplacebo.c b/src/render/renderer_libplacebo.c index 47b9ab4..3b69c64 100644 --- a/src/render/renderer_libplacebo.c +++ b/src/render/renderer_libplacebo.c @@ -422,9 +422,9 @@ static void renderer_lp_render(struct camu_renderer *renderer, struct camu_scree do_gpu_finish |= weighted; // Terrible hack. Lets us distinguish single frames with the same dimensions. // Tied to a libplacebo patch to consider info_priv in the hash. - // Also, add in reset_pts to account for buffer resets. intptr_t hash = (intptr_t)video->buf; - hash += float_64_hash(video->buf->reset_pts); + // Also add in seek_pts to account for frame queue resets (seek() while paused). + hash += float_64_hash(video->buf->seek_pts); if (vr_emulation) { hash += float_64_hash(mouse_x); hash += float_64_hash(mouse_y); |