diff options
30 files changed, 967 insertions, 572 deletions
@@ -41,9 +41,27 @@ - [x] al_atomic_* -> atomic_* - [x] Test pause -> cmv-clear -> add audio only - [x] libnaunet zlib wrap - - [ ] Remove entry->ended from sink. + - [x] Runtime and simpler SINK_LOCAL. + - [x] Is there any race condition around entry->paused? + - [x] Audio buffer local pause. + - [ ] fade-out on seek? fade-in on any full cut-out? + - [ ] win32 -municode -mwindows build fixes. + - win32 resize stuff + - https://stackoverflow.com/questions/50898990/reduce-flickering-when-using-setwindowpos-to-change-the-left-edge-of-a-window + - https://stackoverflow.com/questions/53000291/how-to-smooth-ugly-jitter-flicker-jumping-when-resizing-windows-especially-drag + - https://www.gamedev.net/forums/topic/708865-dxgi-flip-model-flickering-during-live-resize/ + - https://github.com/Blinue/Magpie/pull/1071 + - https://gist.github.com/apkipa/20cae438aef2a8633f99e10e0b90b11e + - [ ] Why does harfbuz use a different ar than what I define. + - [ ] Implement resize to buffer. + - [ ] Windows test: Have second client do input simulation while causing Select pause on windows cmd. + - [ ] PTS gets cooked at some point (u64)-1. + - [ ] Log overall bitrate from FFmpeg. + - [ ] Is it possibly to cleanly fold DIRECT_MODE more into nn_multiplex/nn_rpc/nn_packet_stream. - [ ] Pass over logging. + - [ ] TCP throughput testing. - [ ] list_remove(). + - [ ] Remove entry->ended from sink. - [ ] Do ntp like latency evaluation. - [ ] packet_cache_v2. - [ ] Implement queued. @@ -51,7 +69,8 @@ - [ ] Header cleanup. - [ ] https://github.com/JuliaStrings/utf8proc/blob/master/utf8proc.h - [ ] ustr for ui elements. - - [ ] remove wstr. + - [ ] Remove wstr. + - [ ] Consider usage in win32 sections. - [ ] Packet benchmark. - Create packet, write bunch of ints, send to another thread, thread adds everything. - [ ] Startup timing information. @@ -105,6 +124,21 @@ - [ ] Mark entries for removal rather than remove in-place. - [ ] More optimal command queue. +// @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). + === Cache - See cache/entry.h. - [ ] Don't use file_mapped for large files. @@ -125,6 +159,8 @@ - [ ] Internalize "packet cache" logic. === Codecs + - [ ] Cleanup hwaccel stuff. + - [ ] Separate ff_decoder from hwaccel specific stuff. - [ ] Negotiate codec with sinks. - [ ] Better wuffs support. - [ ] https://snisurset.net/code/abydos diff --git a/src/buffer/audio.c b/src/buffer/audio.c index f066f18..7c10dda 100644 --- a/src/buffer/audio.c +++ b/src/buffer/audio.c @@ -15,15 +15,21 @@ #define BUFFER_MARK_BUFFERED 0.35 #ifdef CAMU_AUDIO_BUFFER_FADE -#define FADE_STEP(fmt, down) ((down ? -2.5f : 2.5f) / (fmt)->sample_rate) -#define FADE_MIN 0.175f +#define FADE_STEP(fmt, down) ((down ? -4.75f : 2.25f) / (fmt)->sample_rate) +#define FADE_MIN 0.25f +#define FADE_TAIL 1 #endif enum { - PAUSE_PAUSED = 0, + // Resync (skip/delay) on read(), skip fade and signal BUFFER_PAUSED on pause. + PAUSE_SYNC = 0, + // Already signaled BUFFER_PAUSED, resync on resume. + PAUSE_PAUSED, #ifdef CAMU_AUDIO_BUFFER_FADE + // Fade out but don't consume any data from the buffer. PAUSE_FADING, - PAUSE_FADING_COMPLETE, + // Return silence fade.tail times then pause. + PAUSE_FADE_COMPLETE, #endif PAUSE_PLAYING }; @@ -31,7 +37,7 @@ enum { static void reset_buffer_state(struct camu_audio_buffer *buf) { atomic_store(f64)(&buf->pts, -1.0, AL_ATOMIC_RELAXED); - buf->pause = PAUSE_PAUSED; + buf->pause = PAUSE_SYNC; atomic_store(u32)(&buf->unpause, 0, AL_ATOMIC_RELAXED); buf->logged_delay = false; atomic_store(u32)(&buf->volume.set, 0, AL_ATOMIC_RELAXED); @@ -125,6 +131,7 @@ static bool push_internal(struct camu_audio_buffer *buf, f64 pts, u8 **data, s32 if (data && sample_count > 0) { f64 duration = camu_audio_format_samples_to_sec(&buf->fmt.in, sample_count); if (frame_is_late(buf->clock, base_pts, pts, duration)) { + log_trace("Discarding late frame with PTS %f.", pts); return true; } if (buf->fmt.resampler_needed) { @@ -281,20 +288,25 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif ptrdiff_t ret, signal = req; f64 base_pts = atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE); - bool set_clock = atomic_load(bool)(&buf->no_video, AL_ATOMIC_RELAXED); - f64 pts = camu_clock_get_pts(buf->clock, buf->latency, set_clock); + bool allow_set = atomic_load(bool)(&buf->no_video, AL_ATOMIC_RELAXED); + bool armed_for_pause = false; + f64 pts = camu_clock_get_pts(buf->clock, buf->latency, allow_set, &armed_for_pause); if (pts == CAMU_PTS_SIGNAL_PAUSE) { return 0; - } else if (pts == CAMU_PTS_PAUSED) { + } else if (pts == CAMU_PTS_PAUSED || armed_for_pause) { #ifdef CAMU_AUDIO_BUFFER_FADE + if (buf->pause == PAUSE_SYNC) { + buf->pause = PAUSE_FADE_COMPLETE; + buf->fade.tail = FADE_TAIL; + buf->logged_delay = false; + } if (buf->pause == PAUSE_PLAYING) { buf->pause = PAUSE_FADING; // fade offset should only ever be read when pause = FADING. buf->fade.offset = 0; - } else if (buf->pause == PAUSE_PAUSED || buf->pause == PAUSE_FADING_COMPLETE || - (buf->fade.volume == 0.f)) { + } else if (buf->pause == PAUSE_PAUSED || buf->pause == PAUSE_FADE_COMPLETE || buf->fade.volume == 0.f) { al_memset(data, 0, req); - if (buf->pause == PAUSE_FADING_COMPLETE) { + if (buf->pause == PAUSE_FADE_COMPLETE) { if (!--buf->fade.tail) { buf->callback(buf->userdata, CAMU_BUFFER_PAUSED); buf->pause = PAUSE_PAUSED; @@ -302,19 +314,24 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif } else if (buf->pause != PAUSE_PAUSED) { // Don't signal PAUSED until the next read to ensure at least 1 silent // frame is included in the fade out. - buf->pause = PAUSE_FADING_COMPLETE; + buf->pause = PAUSE_FADE_COMPLETE; // How many frames of silence to append after fading. - buf->fade.tail = 1; + buf->fade.tail = FADE_TAIL; } return req; } - } else if (buf->pause == PAUSE_FADING || buf->pause == PAUSE_FADING_COMPLETE) { + } else if (buf->pause == PAUSE_FADING || buf->pause == PAUSE_FADE_COMPLETE) { // If unpaused during a fade out, resume immediately to not break the stream. - buf->pause = PAUSE_PLAYING; - // Though, we will still jump back (possible pop) unless we are ignoring sync. + // If we're ignoring desync, we can move the buffer up the the current + // fade position so there's no break _or_ jump in the stream. if (buf->ignore_desync) { + buf->pause = PAUSE_PLAYING; ret = al_ring_buffer_discard(&buf->rb, buf->fade.offset); - base_pts += camu_audio_format_bytes_to_sec(fmt, ret); + f64 skip = camu_audio_format_bytes_to_sec(fmt, buf->fade.offset); + base_pts += skip; + log_info("Resuming at fade position, skipping %fs of audio (%zd bytes).", skip, ret); + } else { + buf->pause = PAUSE_SYNC; } #else // No fade and clock paused. al_memset(data, 0, req); @@ -336,8 +353,8 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif if (!buf->ignore_desync && atomic_load(u32)(&buf->unpause, AL_ATOMIC_ACQUIRE) > 0) { // Queuing multiple resyncs before resuming the stream will cause pops! - log_debug("Forcing resync."); - buf->pause = PAUSE_PAUSED; + log_info("Forcing resync."); + buf->pause = PAUSE_SYNC; atomic_sub(u32)(&buf->unpause, 1, AL_ATOMIC_RELEASE); } @@ -353,13 +370,13 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif // Unpause and possibly attempt syncing to the clock. // PAUSE_PAUSED signifies that the last read was silence, meaning we can skip around // in the buffer without worrying about pops. - if (UNLIKELY(buf->pause == PAUSE_PAUSED)) { + if (UNLIKELY(buf->pause == PAUSE_SYNC || buf->pause == PAUSE_PAUSED)) { if (!buf->ignore_desync) { pts -= base_pts; if (pts > 0.0) { // Skip. ret = MIN((ptrdiff_t)camu_audio_format_sec_to_bytes(fmt, pts), have); - log_info("Skipping %fs of audio (%zd bytes).", pts, ret); ret = al_ring_buffer_discard(&buf->rb, ret); + log_info("Skipping %fs of audio (%zd bytes).", pts, ret); have -= ret; base_pts += camu_audio_format_bytes_to_sec(fmt, ret); // Could go on to underrun. @@ -367,7 +384,7 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif pts = -pts; ret = MIN((ptrdiff_t)camu_audio_format_sec_to_bytes(fmt, pts), req); if (!buf->logged_delay) { - log_info("Delaying audio by %fs.", pts); + log_info("Delaying audio by %fs (%zd bytes).", pts, ret); buf->logged_delay = true; } al_memset(data, 0, ret); @@ -410,7 +427,7 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif // If we have data by the next read(), try to skip ahead to maintain sync. // This might exacerbate the underrun issue but an underrun is already // unexpected behavior, trying to stay in sync comes first. - buf->pause = PAUSE_PAUSED; + buf->pause = PAUSE_SYNC; } // If flow = SIGNALED, we are waiting to be removed or reset. } diff --git a/src/buffer/clock.c b/src/buffer/clock.c index 356ac56..86d6739 100644 --- a/src/buffer/clock.c +++ b/src/buffer/clock.c @@ -1,16 +1,20 @@ -#include <nnwt/thread.h> +#include <al/math.h> +#include <nnwt/time.h> #include "../liana/common.h" #include "clock.h" -#define RUNNING 0.0 -#define PAUSED -DBL_MAX +// clock->tick +#define SET (-HUGE_VAL) -// @TODO: -// Stop throwing away precision before we need to. -// Assume calc_tick_offset() could give a negative result at any point. -// Use a tick offset to set local buffer latency in a way that makes more sense. +// clock->pause +#define RUNNING HUGE_VAL +#define PAUSED (-HUGE_VAL) + +// clock->paused_at +#define NOT_STARTED HUGE_VAL +#define NOT_PAUSED (-HUGE_VAL) void camu_clock_init(struct camu_clock *clock, void (*callback)(void *, u8), void *userdata) { @@ -18,16 +22,16 @@ void camu_clock_init(struct camu_clock *clock, void (*callback)(void *, u8), voi clock->userdata = userdata; } -static f64 calc_tick_offset(f64 tick, u64 now, u64 target) +static inline f64 calc_tick_offset(f64 tick, u64 now, u64 target) { if (now > target) { - return tick - (now - target) / 1000000.0; + return tick - ((now - target) / 1000000.0); } else { - return tick + (target - now) / 1000000.0; + return tick + ((target - now) / 1000000.0); } } -static void offset_tick(struct camu_clock *clock, f64 diff) +static inline void offset_tick(struct camu_clock *clock, f64 diff) { diff += atomic_load(f64)(&clock->tick, AL_ATOMIC_ACQUIRE); atomic_store(f64)(&clock->tick, diff, AL_ATOMIC_RELEASE); @@ -36,20 +40,27 @@ static void offset_tick(struct camu_clock *clock, f64 diff) void camu_clock_set(struct camu_clock *clock, f64 base) { clock->base = base; - atomic_store(f64)(&clock->tick, -1.0, AL_ATOMIC_RELAXED); + atomic_store(f64)(&clock->tick, SET, AL_ATOMIC_RELAXED); atomic_store(f64)(&clock->pause, PAUSED, AL_ATOMIC_RELAXED); - clock->paused_at = 0.0; + atomic_store(bool)(&clock->external_pause, false, AL_ATOMIC_RELAXED); + clock->paused_at = NOT_STARTED; atomic_store(f64)(&clock->last_pts, base, AL_ATOMIC_RELAXED); } +// Setting an offset will unconditionally break sync. Only used in local mode. +void camu_clock_offset(struct camu_clock *clock, f64 offset) +{ + clock->offset = offset; +} + void camu_clock_seek(struct camu_clock *clock, f64 base, u64 target) { clock->base = base; atomic_store(f64)(&clock->last_pts, base, AL_ATOMIC_RELAXED); - if (clock->paused_at == -1.0) { + if (clock->paused_at == NOT_PAUSED) { // We are safe to directly edit the tick here. if (target == 0) { - atomic_store(f64)(&clock->tick, -1.0, AL_ATOMIC_RELAXED); + atomic_store(f64)(&clock->tick, SET, AL_ATOMIC_RELAXED); } else { f64 tick = nn_get_tick(); tick = calc_tick_offset(tick, nn_get_timestamp(), target); @@ -63,62 +74,73 @@ void camu_clock_seek(struct camu_clock *clock, f64 base, u64 target) // Likely confusing an entry's buffers and causing excessive catchup or delay. // We mitigate this sink-side by immediately switching to a potential target in // CLIENT_REMOVE_BUFFERS, which is always called before an entry is seeked. - clock->paused_at = 0.0; + clock->paused_at = NOT_STARTED; } } -// Seek to 0 but include the time it took to perform the seek. -void camu_clock_loop(struct camu_clock *clock, f64 last_pts) -{ - clock->base = 0.0; - offset_tick(clock, last_pts - clock->base); -} - // If late, pause immediately and on resume() there will be a long delay. -void camu_clock_pause(struct camu_clock *clock, u64 target) +bool camu_clock_pause(struct camu_clock *clock, u64 target) { - al_assert(clock->paused_at == -1.0); + al_assert(clock->paused_at == NOT_PAUSED); - f64 now = nn_get_tick(), tick = now; + f64 tick = atomic_load(f64)(&clock->tick, AL_ATOMIC_RELAXED); + + f64 current = nn_get_tick(), pause = current; if (target > 0) { - tick = calc_tick_offset(tick, nn_get_timestamp(), target); + pause = calc_tick_offset(pause, nn_get_timestamp(), target); } - atomic_store(f64)(&clock->pause, (tick > now) ? tick : -tick, AL_ATOMIC_RELAXED); + clock->paused_at = pause; - clock->paused_at = tick; + if ((tick != SET && tick >= current) || pause <= current) { // Immediate pause. + // This could be -0.0, meaning 0.0 has to be considered paused by get_pts(). + atomic_store(f64)(&clock->pause, -pause, AL_ATOMIC_RELAXED); + return true; + } else { // Arm for pause. + // This can never be 0.0. + al_assert(pause != 0.0); + atomic_store(f64)(&clock->pause, pause, AL_ATOMIC_RELAXED); + return false; + } +} + +void camu_clock_external_pause(struct camu_clock *clock) +{ + atomic_store(bool)(&clock->external_pause, true, AL_ATOMIC_RELAXED); } // If late, this will be a catchup. void camu_clock_resume(struct camu_clock *clock, u64 target) { - al_assert(clock->paused_at != -1.0); + al_assert(clock->paused_at != NOT_PAUSED); + + atomic_store(bool)(&clock->external_pause, false, AL_ATOMIC_RELAXED); f64 tick = nn_get_tick(); if (target > 0) { tick = calc_tick_offset(tick, nn_get_timestamp(), target); } - if (clock->paused_at == 0.0) { + if (clock->paused_at == NOT_STARTED) { if (target == 0) { // target = 0 can never be synced. - atomic_store(f64)(&clock->tick, -1.0, AL_ATOMIC_RELAXED); + atomic_store(f64)(&clock->tick, SET, AL_ATOMIC_RELAXED); } else { atomic_store(f64)(&clock->tick, tick, AL_ATOMIC_RELAXED); } } else { f64 pause = atomic_load(f64)(&clock->pause, AL_ATOMIC_ACQUIRE); - offset_tick(clock, tick - ((pause != PAUSED) ? -pause : clock->paused_at)); + al_assert(pause != PAUSED); + /* Using pause instead of clock->paused_at is incorrect for sync. + if (pause < 0.0) pause = -pause; + offset_tick(clock, tick - pause); + */ + offset_tick(clock, tick - clock->paused_at); } - atomic_store(f64)(&clock->pause, RUNNING, AL_ATOMIC_RELEASE); - - clock->paused_at = -1.0; -} + clock->paused_at = NOT_PAUSED; -bool camu_clock_is_paused(struct camu_clock *clock) -{ - return atomic_load(f64)(&clock->pause, AL_ATOMIC_RELAXED) < 0.0; + atomic_store(f64)(&clock->pause, RUNNING, AL_ATOMIC_RELEASE); } f64 camu_clock_get_base_pts(struct camu_clock *clock) @@ -126,26 +148,28 @@ f64 camu_clock_get_base_pts(struct camu_clock *clock) return clock->base; } -f64 camu_clock_get_pts(struct camu_clock *clock, f64 latency, bool allow_set) +f64 camu_clock_get_pts(struct camu_clock *clock, f64 latency, bool allow_set, bool *armed_for_pause) { + // Treat pause = 0.0 or -0.0 as paused. f64 pause = atomic_load(f64)(&clock->pause, AL_ATOMIC_RELAXED); - if (pause < 0.0) return CAMU_PTS_PAUSED; + if (pause <= 0.0) return CAMU_PTS_PAUSED; f64 current = nn_get_tick(); f64 tick = atomic_load(f64)(&clock->tick, AL_ATOMIC_RELAXED); - if (tick == -1.0) { + if (tick == SET) { if (allow_set) { - tick = atomic_compare_and_swap(f64)(&clock->tick, -1.0, current); - if (tick == -1.0) tick = current; + tick = atomic_compare_and_swap(f64)(&clock->tick, SET, current); + if (tick == SET) tick = current; } else { return CAMU_PTS_PAUSED; } } - f64 pts = clock->base + (current - tick); + f64 pts = (clock->base - clock->offset) + (current - tick); bool signal_pause = false; - if (pause > 0.0 && current > pause) { // pause > 0.0 = RUNNING or armed for pause. + *armed_for_pause = pause != RUNNING || atomic_load(bool)(&clock->external_pause, AL_ATOMIC_RELAXED); + if (pause > 0.0 && current > pause) { f64 paused_at = -current; pause = atomic_compare_and_swap(f64)(&clock->pause, pause, paused_at); if (pause != paused_at) { diff --git a/src/buffer/clock.h b/src/buffer/clock.h index 8324b6a..eb3dcaf 100644 --- a/src/buffer/clock.h +++ b/src/buffer/clock.h @@ -33,8 +33,10 @@ enum { struct camu_clock { f64 base; + f64 offset; atomic(f64) tick; atomic(f64) pause; + atomic(bool) external_pause; f64 paused_at; atomic(f64) last_pts; void (*callback)(void *, u8); @@ -43,14 +45,14 @@ struct camu_clock { void camu_clock_init(struct camu_clock *clock, void (*callback)(void *, u8), void *userdata); void camu_clock_set(struct camu_clock *clock, f64 base); +void camu_clock_offset(struct camu_clock *clock, f64 offset); void camu_clock_seek(struct camu_clock *clock, f64 base, u64 target); -void camu_clock_loop(struct camu_clock *clock, f64 last_pts); -void camu_clock_pause(struct camu_clock *clock, u64 target); +bool camu_clock_pause(struct camu_clock *clock, u64 target); +void camu_clock_external_pause(struct camu_clock *clock); void camu_clock_resume(struct camu_clock *clock, u64 target); -bool camu_clock_is_paused(struct camu_clock *clock); f64 camu_clock_get_base_pts(struct camu_clock *clock); -f64 camu_clock_get_pts(struct camu_clock *clock, f64 latency, bool allow_set); +f64 camu_clock_get_pts(struct camu_clock *clock, f64 latency, bool allow_set, bool *armed_for_pause); f64 camu_clock_get_last_pts(struct camu_clock *clock); diff --git a/src/buffer/video.c b/src/buffer/video.c index 93356ac..5a469a4 100644 --- a/src/buffer/video.c +++ b/src/buffer/video.c @@ -15,10 +15,17 @@ // MARK_LOW is considered directly after reading a frame, so in other words, // it will trigger at the point where there is about (LOW+1) frames left. Even at // something like 240fps that's still ~(4.16*(LOW+1))ms to uncork and produce a new frame. +#ifdef CAMU_HUGE_VIDEO_BUFFER +#define BUFFER_MARK_LOW 24 +#define BUFFER_MARK_BUFFERED 36 // Must be >1. +#define BUFFER_MARK_HIGH 48 +#define BUFFER_MARK_RESET (BUFFER_MARK_HIGH * 2) +#else #define BUFFER_MARK_LOW 4 #define BUFFER_MARK_BUFFERED 6 // Must be >1. #define BUFFER_MARK_HIGH 8 #define BUFFER_MARK_RESET (BUFFER_MARK_HIGH * 2) +#endif bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *clock) { @@ -125,7 +132,7 @@ static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame f64 base_pts = atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE); f64 duration = camu_ff_frame_duration(frame) * av_q2d(stream->time_base); if (!buf->single_frame && frame_is_late(buf->clock, base_pts, pts, duration)) { - log_debug("Discarding late frame with PTS %f.", pts); + log_trace("Discarding late frame with PTS %f.", pts); return false; } if (base_pts == -1.0) atomic_store(f64)(&buf->pts, pts, AL_ATOMIC_RELEASE); diff --git a/src/codec/codec.h b/src/codec/codec.h index 730e8cb..f1c979b 100644 --- a/src/codec/codec.h +++ b/src/codec/codec.h @@ -12,6 +12,8 @@ #include "codecs.h" +//#define CAMU_HUGE_VIDEO_BUFFER + enum { CAMU_LANG_UNKNOWN = 0, CAMU_LANG_ENGLISH, @@ -208,7 +210,8 @@ struct camu_demuxer { bool (*seek)(struct camu_demuxer *, u64, bool); void (*free)(struct camu_demuxer **); array(struct camu_codec_stream) streams; - u32 subscribed; + // @TODO: .mkv can probably have more than 64 streams. + u64 subscribed; }; struct camu_renderer; diff --git a/src/codec/ffmpeg/common.c b/src/codec/ffmpeg/common.c index e697a00..59f59b7 100644 --- a/src/codec/ffmpeg/common.c +++ b/src/codec/ffmpeg/common.c @@ -65,3 +65,31 @@ void camu_ff_common_init(void) #endif camu_ff_set_log_callback(av_log_callback); } + +AVCodecContext *camu_ff_alloc_codec_context(const AVCodec *codec, AVCodecParameters *codecpar) +{ + AVCodecContext *codec_context = avcodec_alloc_context3(codec); + if (!codec_context) { + log_error("Failed to alloc codec context."); + return NULL; + } + + s32 ret = avcodec_parameters_to_context(codec_context, codecpar); + if (ret < 0) { + log_error("Failed to copy codec parameters (%s).", av_err2str(ret)); + return NULL; + } + + return codec_context; +} + +bool camu_ff_open_avcodec(AVCodecContext *codec_context, const AVCodec *codec, AVDictionary *opts) +{ + s32 ret = avcodec_open2(codec_context, codec, &opts); + if (ret < 0) { + log_error("Failed to open codec %s (%s).", codec->name, av_err2str(ret)); + return false; + } + av_dict_free(&opts); + return true; +} diff --git a/src/codec/ffmpeg/common.h b/src/codec/ffmpeg/common.h index 639ca38..26f807d 100644 --- a/src/codec/ffmpeg/common.h +++ b/src/codec/ffmpeg/common.h @@ -2,9 +2,13 @@ #include <al/types.h> #include <libavutil/frame.h> +#include <libavcodec/avcodec.h> s64 camu_ff_frame_duration(AVFrame *frame); // userdata, level, fmt, args void camu_ff_set_log_callback(void (*callback)(void *, int, const char *, va_list)); void camu_ff_common_init(void); + +AVCodecContext *camu_ff_alloc_codec_context(const AVCodec *codec, AVCodecParameters *codecpar); +bool camu_ff_open_avcodec(AVCodecContext *codec_context, const AVCodec *codec, AVDictionary *opts); diff --git a/src/codec/ffmpeg/decoder.c b/src/codec/ffmpeg/decoder.c index 7d50bca..3da0027 100644 --- a/src/codec/ffmpeg/decoder.c +++ b/src/codec/ffmpeg/decoder.c @@ -4,8 +4,11 @@ #include <libavutil/cpu.h> #include "decoder.h" +#include "common.h" #ifdef CAMU_FF_DECODER_HWACCEL +// If hwaccel throughput is worse then expected on linux, test with +// your GPU set to "highest clocks"/power_dpm_force_performance_level = high. #include "../../render/renderer.h" static const char *hwdevices[] = { #ifdef CAMU_HAVE_VULKAN_DECODE @@ -23,31 +26,14 @@ static const char *hwdevices[] = { }; #endif -static bool alloc_codec_context_internal(struct camu_ff_decoder *av, const AVCodec *codec, AVCodecParameters *codecpar) +static void log_codec_name(const AVCodec *codec, const char *hwdevice_name) { - if (!(av->codec_context = avcodec_alloc_context3(codec))) { - log_error("Failed to alloc codec context."); - return false; - } - - if (avcodec_parameters_to_context(av->codec_context, codecpar) < 0) { - log_error("Failed to copy codec parameters."); - return false; + const char *name = codec->long_name ? codec->long_name : codec->name; + if (hwdevice_name) { + log_info("Codec: %s (via %s hwaccel).", name, hwdevice_name); + } else { + log_info("Codec: %s.", name); } - - return true; -} - -static bool open_avcodec_internal(struct camu_ff_decoder *av, const AVCodec *codec) -{ - AVDictionary *opts = NULL; - s32 ret = avcodec_open2(av->codec_context, codec, &opts); - if (ret < 0) { - log_error("Failed to open codec %s (%s).", codec->name, av_err2str(ret)); - return false; - } - av_dict_free(&opts); - return true; } #ifdef CAMU_FF_DECODER_HWACCEL @@ -93,39 +79,47 @@ static s32 init_hwframe_context(struct camu_ff_decoder *av, AVCodecContext *cont return ret; } +static bool fallback_to_sw_codec(struct camu_ff_decoder *av, const AVCodec *codec) +{ + log_warn("Attempting software decoding fallback..."); + // Remake AVCodecContext as a software decoder. + AVCodecParameters *codecpar = av->codecpar; + av->errored_hw_context = av->codec_context; + av->codec_context = camu_ff_alloc_codec_context(codec, codecpar); + if (!av->codec_context) { + return false; + } + // Note about thread_type choice in ff_decoder_init(). + av->codec_context->thread_type = FF_THREAD_FRAME; + av->codec_context->thread_count = av->thread_count; + if (!camu_ff_open_avcodec(av->codec_context, codec, NULL)) { + return false; + } + log_codec_name(codec, NULL); + log_info("Using software fallback with %i threads for decoder.", av->thread_count); + return true; +} + static enum AVPixelFormat get_hw_format(AVCodecContext *context, const enum AVPixelFormat *pix_fmts) { struct camu_ff_decoder *av = (struct camu_ff_decoder *)context->opaque; - const AVCodec *codec = av->codec_context->codec; - const char *hwdevice_name = av_hwdevice_get_type_name(av->hw_device_type); - const enum AVPixelFormat *fmt = pix_fmts; for (; *fmt != AV_PIX_FMT_NONE; fmt++) { if (*fmt == av->hw_pix_fmt) { if (av->use_frames_context) { init_hwframe_context(av, av->codec_context, av->hw_context); } - log_info("Using %s hardware decoding.", hwdevice_name); - goto out; + return *fmt; } } + const char *hwdevice_name = av_hwdevice_get_type_name(av->hw_device_type); log_error("Failed to get %s HW surface format.", hwdevice_name); - // Remake AVCodecContext as a software decoder. - 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(). - av->codec_context->thread_type = FF_THREAD_FRAME; - av->codec_context->thread_count = av->thread_count; - if (open_avcodec_internal(av, codec)) { - log_info("Software fallback, using %i threads for decoder.", av->thread_count); - } + fallback_to_sw_codec(av, av->sw_codec); -out: - return *fmt; + return AV_PIX_FMT_NONE; } static s32 init_hwdevice_context(struct camu_ff_decoder *av, AVCodecContext *context) @@ -137,14 +131,17 @@ static s32 init_hwdevice_context(struct camu_ff_decoder *av, AVCodecContext *con if (av->hw_device_type == AV_HWDEVICE_TYPE_D3D11VA) { log_info("av_hwdevice_ctx_create() returns."); } + const char *hwdevice_name = av_hwdevice_get_type_name(av->hw_device_type); if (ret < 0) { - log_error("Failed to create specified HW device."); + log_error("Failed to create %s HW device.", hwdevice_name); return ret; } context->hw_device_ctx = av_buffer_ref(av->hw_context); +#ifdef CAMU_HUGE_VIDEO_BUFFER // Note that context->extra_hw_frames has the ability to cause corruption. - context->extra_hw_frames = 12; + context->extra_hw_frames = 48; +#endif return ret; } @@ -188,8 +185,12 @@ static bool get_hwdevice_config(struct camu_ff_decoder *av, const AVCodec *codec return true; } } - const char *hwdevice_name = av_hwdevice_get_type_name(hw_device_type); - log_warn("No hw_config for device type %s (type: %d).", hwdevice_name, hw_device_type); + if (hw_device_type == AV_HWDEVICE_TYPE_NONE) { + log_warn("No hw_config for codec %s.", codec->name); + } else { + const char *hwdevice_name = av_hwdevice_get_type_name(hw_device_type); + log_warn("No hw_config for device type %s.", hwdevice_name); + } return false; } @@ -243,10 +244,21 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend log_error("Failed to find decoder."); return false; } +#ifdef CAMU_FF_DECODER_HWACCEL + av->sw_codec = codec; + // @TODO: How do you correctly probe avcodec_get_hw_config() if find_decoder() + // always returns libdav1d first? + if (al_strscmp(codec->name, "libdav1d") == 0) { + if (!(codec = avcodec_find_decoder_by_name("av1"))) { + codec = av->sw_codec; + } + } +#endif bool is_video = codecpar->codec_type == AVMEDIA_TYPE_VIDEO && stream->duration > 0; #ifdef CAMU_FF_DECODER_HWACCEL + bool hw_codec_selected = false; if (is_video) { collect_supported_hwaccels(av); av->hw_device_type = AV_HWDEVICE_TYPE_NONE; @@ -254,21 +266,22 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend 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 { + if (get_hwdevice_config(av, codec, av->hw_device_type)) { break; + } else { + av->hw_device_type = AV_HWDEVICE_TYPE_NONE; } } if (av->hw_device_type == AV_HWDEVICE_TYPE_NONE) { const AVCodec *hw_codec; al_array_foreach(av->supported_hw_codecs, i, hw_codec) { if (hw_codec->id == codecpar->codec_id) { - // @TODO: hw_codec's without a hwdevice (v4l2m2m, cuvid, amf). + // @TODO: hw_codecs without a hwdevice (v4l2m2m, cuvid, amf). size_t length = al_strlen(hw_codec->wrapper_name); av->hw_device_type = hw_device_supported_by_name(av, hw_codec->wrapper_name, length); if (get_hwdevice_config(av, hw_codec, av->hw_device_type)) { codec = hw_codec; + hw_codec_selected = true; break; } av->hw_device_type = AV_HWDEVICE_TYPE_NONE; @@ -281,7 +294,8 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend } #endif - if (!alloc_codec_context_internal(av, codec, codecpar)) { + av->codec_context = camu_ff_alloc_codec_context(codec, codecpar); + if (!av->codec_context) { return false; } @@ -302,14 +316,18 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend #endif if (is_video) { - av->thread_count = MIN(8, MAX(1, av_cpu_count() / 2)); +#ifdef CAMU_HUGE_VIDEO_BUFFER + av->thread_count = av_cpu_count() * 2; +#else + av->thread_count = MIN(12, MAX(1, av_cpu_count())); +#endif #ifdef CAMU_FF_DECODER_HWACCEL - if (av->hw_device_type == AV_HWDEVICE_TYPE_NONE) { + if (!(hw_codec_selected || av->hw_device_type != AV_HWDEVICE_TYPE_NONE)) { #endif // FF_THREAD_FRAME or FF_THREAD_SLICE. // Both modes can get choked on certain videos, higher thread_count with // FRAME seems like the safest default. - // @TODO: Check ffmpeg source, SLICE + FRAME? + // @TODO: Check ffmpeg source, is SLICE + FRAME real? // - https://ffmpeg.org/ffmpeg-codecs.html#Codec-Options av->codec_context->thread_type = FF_THREAD_FRAME; av->codec_context->thread_count = av->thread_count; @@ -322,13 +340,26 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend //av->codec_context->flags |= AV_CODEC_FLAG_BITEXACT; //av->codec_context->flags2 |= AV_CODEC_FLAG2_FAST; - if (!open_avcodec_internal(av, codec)) { + if (!camu_ff_open_avcodec(av->codec_context, codec, NULL)) { +#ifdef CAMU_FF_DECODER_HWACCEL + if (hw_codec_selected && fallback_to_sw_codec(av, av->sw_codec)) { + codec = av->sw_codec; + hw_codec_selected = false; + } else { + return false; + } +#else return false; +#endif } - log_info("Codec: %s (%s) %lldkbps.", codec->name, - codec->long_name ? codec->long_name : codec->name, - (av->codec_context->bit_rate > 0) ? av->codec_context->bit_rate / 1000 : 0); + const char *hwdevice_name = NULL; +#ifdef CAMU_FF_DECODER_HWACCEL + if (av->hw_device_type != AV_HWDEVICE_TYPE_NONE) { + hwdevice_name = av_hwdevice_get_type_name(av->hw_device_type); + } +#endif + log_codec_name(codec, hwdevice_name); av->callback = callback; av->userdata = userdata; diff --git a/src/codec/ffmpeg/decoder.h b/src/codec/ffmpeg/decoder.h index 0c3d4d7..a422941 100644 --- a/src/codec/ffmpeg/decoder.h +++ b/src/codec/ffmpeg/decoder.h @@ -15,6 +15,7 @@ struct camu_ff_decoder { s32 thread_count; #ifdef CAMU_FF_DECODER_HWACCEL struct camu_renderer *renderer; + const AVCodec *sw_codec; array(const AVCodec *) supported_hw_codecs; array(enum AVHWDeviceType) supported_hw_devices; AVBufferRef *hw_context; diff --git a/src/codec/ffmpeg/demuxer.c b/src/codec/ffmpeg/demuxer.c index 32068f3..c49ced1 100644 --- a/src/codec/ffmpeg/demuxer.c +++ b/src/codec/ffmpeg/demuxer.c @@ -158,6 +158,7 @@ static bool ff_demuxer_init(struct camu_demuxer *demux, struct cch_handle *handl static bool subscribed_to_index(struct camu_demuxer *demux, s32 index) { + al_assert(index < 64); return (demux->subscribed >> index) & 1; } diff --git a/src/fruits/cmsrv/cmsrv.c b/src/fruits/cmsrv/cmsrv.c index babd5ef..53fbc98 100644 --- a/src/fruits/cmsrv/cmsrv.c +++ b/src/fruits/cmsrv/cmsrv.c @@ -91,6 +91,8 @@ static u8 server_line_callback(void *userdata, str *line) lia_list_reverse(list); } else if (al_str_eq(line, &al_str_c(";CLEAR"))) { lia_list_clear(list); + } else if (al_str_eq(line, &al_str_c(";DISCONNECT"))) { + lia_server_force_disconnect_nodes(&s->server.data.server); } else { struct nn_packet *packet = nn_packet_create(); nn_packet_write_str(packet, line); diff --git a/src/liana/client.c b/src/liana/client.c index 4d5689f..2b62612 100644 --- a/src/liana/client.c +++ b/src/liana/client.c @@ -13,8 +13,10 @@ enum { RECONNECT_NONE = 0, - RECONNECT_ON_CONNECTION_CLOSED, + RECONNECT_RECOVER, + RECONNECT_SEEK, RECONNECT_SIGNAL_CLIENT, + RECONNECT_DISREGUARD, RECONNECT_DISCONNECTED }; @@ -48,7 +50,7 @@ static void collect_streams(struct lia_client *client, struct nn_packet *packet) u8 type = nn_packet_read_u8(packet); u64 duration = nn_packet_read_u64(packet); s32 index = nn_packet_read_s32(packet); - al_assert(index < 32); + al_assert(index < 64); switch (mode) { case CAMU_NORMAL: { if (type == CAMU_STREAM_AUDIO) { @@ -147,10 +149,6 @@ static void parse_info_packet(struct lia_client *client, struct nn_packet *packe struct camu_codec_stream *stream; al_array_foreach_ptr(client->streams, i, stream) { u8 type = stream->type; - if (type == CAMU_STREAM_SUBTITLE && client->mask == 0) { - log_warn("Ignoring subtitle-only resource."); - goto out; - } if ((selected & (1 << type)) || !(prefs->enabled & (1 << type))) { continue; } @@ -198,6 +196,10 @@ static void parse_info_packet(struct lia_client *client, struct nn_packet *packe lia_vcr_add_track(&client->vcr, track); } } + if (client->mask == (1 << CAMU_STREAM_SUBTITLE)) { + log_warn("Ignoring subtitle-only resource."); + client->mask = 0; + } out: client->callback(client->userdata, LIANA_CLIENT_CONFIGURE_COMPLETE, NULL, NULL); } @@ -211,13 +213,15 @@ static void info_packet_callback(void *userdata, struct nn_packet_stream *stream if (client->mask == 0 || lia_vcr_is_empty(&client->vcr)) { log_warn("Discarding resource with no applicable streams."); al_assert(client->reconnect == RECONNECT_NONE); + client->reconnect = RECONNECT_DISREGUARD; nn_packet_stream_disconnect(&client->data); return; } stream->packet_callback = data_packet_callback; struct nn_packet *rpacket = nn_packet_create(); - nn_packet_write_u32(rpacket, client->mask); + nn_packet_write_u64(rpacket, client->mask); nn_packet_stream_send_packet(stream, rpacket); + client->reconnect = RECONNECT_RECOVER; lia_vcr_start(&client->vcr); } @@ -239,6 +243,7 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream) .pos = client->pos, .pause = LIANA_PAUSE_NONE }; + client->at = LIANA_TIMESTAMP_INVALID; client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time); // The value of client->mask will not have changed since connection_closed_callback(). if (client->rec.unconfigured) { @@ -247,18 +252,19 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream) } client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &client->rec); } else { - al_assert(client->connection_id == 0); + al_assert(client->connection_id == 0 && client->reconnect == RECONNECT_NONE); } stream->packet_sent_callback = packet_sent_callback; struct nn_packet *packet = nn_packet_create(); nn_packet_write_u32(packet, client->node_id); nn_packet_write_u32(packet, client->connection_id); - nn_packet_write_u32(packet, client->mask); + nn_packet_write_u64(packet, client->mask); nn_packet_write_u64(packet, client->pos); if (client->mask == 0) { stream->packet_callback = info_packet_callback; } else { stream->packet_callback = data_packet_callback; + client->reconnect = RECONNECT_RECOVER; lia_vcr_start(&client->vcr); } nn_packet_stream_send_packet(stream, packet); @@ -268,15 +274,26 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream) static void connection_closed_callback(void *userdata, struct nn_packet_stream *stream) { struct lia_client *client = (struct lia_client *)userdata; - if (client->reconnect == RECONNECT_ON_CONNECTION_CLOSED) { + + bool reconnect = client->reconnect == RECONNECT_RECOVER || client->reconnect == RECONNECT_SEEK; + if (client->reconnect == RECONNECT_RECOVER) { + al_assert(client->at == LIANA_TIMESTAMP_INVALID); + struct lia_timing time; + client->callback(client->userdata, LIANA_CLIENT_RECOVER_TO, NULL, &time); + client->pos = time.pos; + client->at = time.at; + } + + if (reconnect) { lia_vcr_flush(&client->vcr); } else { lia_vcr_close_all(&client->vcr); } + // 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, + .reconnect = reconnect, .unconfigured = client->mask == 0, .mask = client->mask }; @@ -295,10 +312,13 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream * } al_array_free(client->rec.detached); } - if (client->reconnect == RECONNECT_ON_CONNECTION_CLOSED) { - // If stream_reconnect() errors, the client will be closed on recursion. + + if (reconnect) { + // If stream_reconnect() errors or is aborted, the client will be closed on recursion. client->reconnect = RECONNECT_SIGNAL_CLIENT; - if (!client->mask) { + // A client being seeked before an info packet is another reason mask may + // be unset here. In that case we don't want to forcefully close. + if (!client->rec.unconfigured && !client->mask) { connection_closed_callback(userdata, stream); } else { #ifdef CAMU_DIRECT_MODE @@ -318,6 +338,7 @@ void lia_client_connect(struct lia_client *client, struct nn_event_loop *loop, client->loop = loop; client->node_id = node_id; client->pos = pos; + client->at = LIANA_TIMESTAMP_INVALID; client->mask = 0; al_array_init(client->streams); client->reconnect = RECONNECT_NONE; @@ -336,13 +357,16 @@ void lia_client_connect(struct lia_client *client, struct nn_event_loop *loop, void lia_client_seek(struct lia_client *client, u64 pos, u64 at) { - if (client->reconnect == RECONNECT_DISCONNECTED) return; - // If reconnect = ON_CONNECTION_CLOSED or SIGNAL_CLIENT, we are safe to edit pos + u8 reconnect = client->reconnect; + if (reconnect == RECONNECT_DISCONNECTED || reconnect == RECONNECT_DISREGUARD) { + return; + } + // If reconnect = SEEK, RECOVER or SIGNAL_CLIENT, we are safe to edit pos // and at in-place because they aren't evaluated until connection_callback(). client->pos = pos; client->at = at; - if (client->reconnect == RECONNECT_NONE) { - client->reconnect = RECONNECT_ON_CONNECTION_CLOSED; + if (reconnect == RECONNECT_NONE || reconnect == RECONNECT_RECOVER) { + client->reconnect = RECONNECT_SEEK; nn_packet_stream_disconnect(&client->data); } } @@ -352,7 +376,9 @@ void lia_client_disconnect(struct lia_client *client) u8 reconnect = client->reconnect; al_assert(reconnect != RECONNECT_DISCONNECTED); client->reconnect = RECONNECT_DISCONNECTED; - if (reconnect != RECONNECT_ON_CONNECTION_CLOSED) { + // If reconnect == SIGNAL_CLIENT, stream_disconnect() needs to ensure + // connection_callback() is never called. + if (reconnect != RECONNECT_SEEK && reconnect != RECONNECT_DISREGUARD) { nn_packet_stream_disconnect(&client->data); } } diff --git a/src/liana/client.h b/src/liana/client.h index 681b134..452150d 100644 --- a/src/liana/client.h +++ b/src/liana/client.h @@ -12,6 +12,7 @@ enum { LIANA_CLIENT_DATA, LIANA_CLIENT_SUBTITLE, LIANA_CLIENT_REMOVE_BUFFERS, + LIANA_CLIENT_RECOVER_TO, LIANA_CLIENT_RESUME_AT, LIANA_CLIENT_RECONNECTED, LIANA_CLIENT_EOF, @@ -22,7 +23,7 @@ enum { struct lia_reconnect_info { bool reconnect; bool unconfigured; - u32 mask; + u64 mask; array(struct camu_codec_stream *) detached; }; @@ -39,7 +40,7 @@ struct lia_client { u32 node_id; struct lia_prefs prefs; array(struct camu_codec_stream) streams; - u32 mask; + u64 mask; u64 pos; u64 at; u8 reconnect; diff --git a/src/liana/handlers/cdio_server.c b/src/liana/handlers/cdio_server.c index 54c63ab..c67330d 100644 --- a/src/liana/handlers/cdio_server.c +++ b/src/liana/handlers/cdio_server.c @@ -42,7 +42,7 @@ static void cdio_server_write_info(struct lia_server_handler *handler, struct nn nn_packet_write_s32(packet, cdio->fmt.channel_count); } -static void cdio_server_subscribe(struct lia_server_handler *handler, u32 mask) +static void cdio_server_subscribe(struct lia_server_handler *handler, u64 mask) { (void)handler; (void)mask; diff --git a/src/liana/handlers/codec_server.c b/src/liana/handlers/codec_server.c index fbb1e72..a7fc587 100644 --- a/src/liana/handlers/codec_server.c +++ b/src/liana/handlers/codec_server.c @@ -79,7 +79,7 @@ static void codec_server_write_info(struct lia_server_handler *handler, struct n } } -static void codec_server_subscribe(struct lia_server_handler *handler, u32 mask) +static void codec_server_subscribe(struct lia_server_handler *handler, u64 mask) { struct lia_codec_server *codec = (struct lia_codec_server *)handler; codec->demux->subscribed = mask; diff --git a/src/liana/handlers/dvd_server.c b/src/liana/handlers/dvd_server.c index 5a9c793..53ef323 100644 --- a/src/liana/handlers/dvd_server.c +++ b/src/liana/handlers/dvd_server.c @@ -53,7 +53,7 @@ static void dvd_server_write_info(struct lia_server_handler *handler, struct nn_ (void)packet; } -static void dvd_server_subscribe(struct lia_server_handler *handler, u32 mask) +static void dvd_server_subscribe(struct lia_server_handler *handler, u64 mask) { (void)handler; (void)mask; diff --git a/src/liana/handlers/handler.h b/src/liana/handlers/handler.h index 48fcef3..ec2f3fe 100644 --- a/src/liana/handlers/handler.h +++ b/src/liana/handlers/handler.h @@ -14,7 +14,7 @@ enum { struct lia_server_handler { bool (*init)(struct lia_server_handler *, struct cch_handle *); void (*write_info)(struct lia_server_handler *, struct nn_packet *); - void (*subscribe)(struct lia_server_handler *, u32); + void (*subscribe)(struct lia_server_handler *, u64); u64 (*get_duration)(struct lia_server_handler *); bool (*seek)(struct lia_server_handler *, u64); void (*step)(struct lia_server_handler *); diff --git a/src/liana/server.c b/src/liana/server.c index 04ef54b..5065425 100644 --- a/src/liana/server.c +++ b/src/liana/server.c @@ -1,3 +1,6 @@ +#define AL_LOG_SECTION "liana" +#include <al/log.h> + #include "server.h" #include "handlers.h" #include "list.h" @@ -166,7 +169,7 @@ static void data_connection_closed_callback(void *userdata, struct nn_packet_str } } -static void start_connection_handler(struct lia_node_connection *conn, u32 mask) +static void start_connection_handler(struct lia_node_connection *conn, u64 mask) { struct nn_packet_stream *stream = conn->stream; conn->handler->subscribe(conn->handler, mask); @@ -180,7 +183,7 @@ static void start_connection_handler(struct lia_node_connection *conn, u32 mask) static void subscribe_packet_callback(void *userdata, struct nn_packet_stream *stream, struct nn_packet *packet) { struct lia_node_connection *conn = (struct lia_node_connection *)userdata; - u32 mask = nn_packet_read_u32(packet); + u64 mask = nn_packet_read_u64(packet); nn_packet_stream_return_packet(stream, packet); start_connection_handler(conn, mask); } @@ -203,7 +206,7 @@ static void handle_connection(struct lia_node_connection *conn, struct nn_packet { struct nn_packet_stream *stream = conn->stream; - u32 mask = nn_packet_read_u32(packet); + u64 mask = nn_packet_read_u64(packet); u64 seek_pos = nn_packet_read_u64(packet); al_assert(!conn->ref); @@ -487,7 +490,7 @@ void lia_node_close(struct lia_node *node) al_array_foreach_rev(node->connections, i, conn) { if (conn->stream) { conn->disconnected = true; - nn_packet_stream_disconnect(conn->stream); + //nn_packet_stream_disconnect(conn->stream); } else { free_connection(conn); } @@ -503,6 +506,19 @@ void lia_server_close(struct lia_server *server) } } +void lia_server_force_disconnect_nodes(struct lia_server *server) +{ + struct lia_node *node; + al_array_foreach(server->nodes, i, node) { + struct lia_node_connection *conn; + al_array_foreach_rev(node->connections, j, conn) { + if (conn->stream) { + nn_packet_stream_disconnect(conn->stream); + } + } + } +} + void lia_server_free(struct lia_server *server) { // Assuming we joined on the event loop, server->nodes should be empty. diff --git a/src/liana/server.h b/src/liana/server.h index 8035ee1..bafe95b 100644 --- a/src/liana/server.h +++ b/src/liana/server.h @@ -61,4 +61,5 @@ struct lia_node *lia_server_create_node(struct lia_server *server, struct cch_en void lia_node_get_duration(struct lia_node *node); void lia_node_close(struct lia_node *node); void lia_server_close(struct lia_server *server); +void lia_server_force_disconnect_nodes(struct lia_server *server); void lia_server_free(struct lia_server *server); diff --git a/src/liana/vcr.c b/src/liana/vcr.c index 4fa548c..f9dad7b 100644 --- a/src/liana/vcr.c +++ b/src/liana/vcr.c @@ -1,15 +1,20 @@ #define AL_LOG_SECTION "vcr" //#define AL_LOG_ENABLE_TRACE #include <al/log.h> +#include <nnwt/time.h> #include "handlers/handler.h" #include "vcr.h" #include "common.h" -#define VCR_BUFFER_BUFFERED MB(6LL) -#define VCR_BUFFER_GROW_FACTOR 8LL -#define VCR_BUFFER_LOW_OFFSET KB(500LL) +#define VCR_BUFFER_BUFFERED MB((u64)6) +#ifdef CAMU_HUGE_VIDEO_BUFFER +#define VCR_BUFFER_GROW_FACTOR ((u64)24) +#else +#define VCR_BUFFER_GROW_FACTOR ((u64)8) +#endif +#define VCR_BUFFER_LOW_OFFSET KB((u64)500) AL_STATIC_ASSERT(buf_gt_low_offset, VCR_BUFFER_BUFFERED * VCR_BUFFER_GROW_FACTOR, >, VCR_BUFFER_LOW_OFFSET); enum { @@ -97,6 +102,8 @@ void lia_vcr_init(struct lia_vcr *vcr, struct nn_event_loop *loop, struct nn_pac vcr->corked = false; nn_signal_init(&vcr->signal, loop, signal_callback, vcr); reset_metrics(vcr); +#else + (void)loop; #endif } @@ -128,7 +135,6 @@ static nn_thread_result NNWT_THREADCALL vcr_track_thread(void *userdata) al_snprintf((char *)thread_name, sizeof(thread_name), "vcr:%hu_%d", vcr->node_id, track->stream->index); nn_thread_set_name(thread_name); - s32 state; bool corked; u32 packets, index = 0; struct nn_packet *packet = NULL; @@ -161,7 +167,7 @@ static nn_thread_result NNWT_THREADCALL vcr_track_thread(void *userdata) #endif } - nn_mutex_lock(&track->mutex); + nn_mutex_lock(&track->lock); // NULL packet means flush. bool success = track->client->handle_packet(track->client, packet); @@ -171,19 +177,21 @@ static nn_thread_result NNWT_THREADCALL vcr_track_thread(void *userdata) return_entire_cache(track); track->cache.disabled = true; nn_packet_cache_unlock(&track->cache); - nn_mutex_unlock(&track->mutex); + nn_mutex_unlock(&track->lock); return 0; } - // Take state again because handle_packet() could have caused the track to be corked. - state = atomic_load(s32)(&track->state, AL_ATOMIC_RELAXED); + if (!packet) { + // Wait on EOF. + track->state = VCR_TRACK_STOPPED; + } - // We wait if corked (TRACK_STOPPED) or EOF. - corked = (packet && state == VCR_TRACK_STOPPED) || !packet; + // Track could have been corked from within handle_packet(). + corked = track->state == VCR_TRACK_STOPPED; // Don't wait, continue processing packets from the current set. if (!corked) { - nn_mutex_unlock(&track->mutex); + nn_mutex_unlock(&track->lock); continue; } @@ -196,12 +204,12 @@ static nn_thread_result NNWT_THREADCALL vcr_track_thread(void *userdata) nn_packet_cache_unlock(&track->cache); // Wait for uncork. - nn_cond_wait(&track->cond, &track->mutex); + nn_cond_wait(&track->cond, &track->lock); // Check for possibly updated state. - state = atomic_load(s32)(&track->state, AL_ATOMIC_RELAXED); + u32 state = track->state; - nn_mutex_unlock(&track->mutex); + nn_mutex_unlock(&track->lock); if (state == VCR_TRACK_CLOSED) { // We already unlocked the cache. @@ -253,12 +261,12 @@ void lia_vcr_add_track(struct lia_vcr *vcr, struct lia_vcr_track *track) { track->vcr = vcr; nn_cond_init(&track->cond); - nn_mutex_init(&track->mutex); + nn_mutex_init(&track->lock); 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); - atomic_store(s32)(&track->state, VCR_TRACK_RUNNING, AL_ATOMIC_RELAXED); + track->state = VCR_TRACK_RUNNING; } bool lia_vcr_remove_track_by_stream(struct lia_vcr *vcr, struct camu_codec_stream *stream) @@ -304,7 +312,7 @@ static void cork_if_buffered(struct lia_vcr *vcr, u64 buffer) if (vcr->expand == VCR_EXPAND_UNTOUCHED) { vcr->mark.buffered = buffer * VCR_BUFFER_GROW_FACTOR; vcr->expand = VCR_EXPAND_GROWN; - log_debug("Expanded buffer to size %.2fMB.", vcr->mark.buffered / (f32)MB(1)); + log_info("Expanded buffer to size %.2fMB.", vcr->mark.buffered / (f32)MB(1)); return; } if (!vcr->corked) { @@ -398,38 +406,46 @@ void lia_vcr_set_buffered(struct lia_vcr_track *track) atomic_store(bool)(&track->buffered, true, AL_ATOMIC_RELAXED); } +// cork() is only ever called from within handle_packet() in vcr_track_thread(). +// Meaning track->lock will be held. void lia_vcr_cork(struct lia_vcr_track *track) { - atomic_store(s32)(&track->state, VCR_TRACK_STOPPED, AL_ATOMIC_RELAXED); + track->state = VCR_TRACK_STOPPED; } void lia_vcr_uncork(struct lia_vcr_track *track) { - if (atomic_load(s32)(&track->state, AL_ATOMIC_ACQUIRE) != VCR_TRACK_STOPPED) { - // We will get here during normal operation. Early returning is historically - // tricky in vcr_uncork(). If I'm understanding correctly, asserting that - // cond_is_waiting() just below means we are safe. + // We need to avoid a race with cork() _and_ flush(). + // Possible race with flush() if locking after checking if state != STOPPED: + // - cork() -> flush() -> uncork(). + // - In uncork() we evaluate track->state to be STOPPED then wait on the lock + // being held by flush(). flush() sets the state to CLOSED and signals the + // cond. Now nn_cond_is_waiting() is false at the point uncork() acquires the lock. + nn_mutex_lock(&track->lock); + if (track->state != VCR_TRACK_STOPPED) { + nn_mutex_unlock(&track->lock); return; } - // Lock before setting track->state to avoid a race with cork(). - nn_mutex_lock(&track->mutex); - atomic_store(s32)(&track->state, VCR_TRACK_RUNNING, AL_ATOMIC_RELEASE); + track->state = VCR_TRACK_RUNNING; al_assert(nn_cond_is_waiting(&track->cond)); nn_cond_signal(&track->cond); - nn_mutex_unlock(&track->mutex); + nn_mutex_unlock(&track->lock); } static void vcr_track_close_internal(struct lia_vcr_track *track) { struct lia_vcr *vcr = track->vcr; - // Calling packet_cache_disable() while holding the track mutex can very possibly deadlock. + // Calling packet_cache_disable() while holding track->lock can very possibly deadlock. nn_packet_cache_disable(&track->cache); - nn_mutex_lock(&track->mutex); - atomic_store(s32)(&track->state, VCR_TRACK_CLOSED, AL_ATOMIC_RELAXED); - if (nn_cond_is_waiting(&track->cond)) { + nn_mutex_lock(&track->lock); + if (track->state == VCR_TRACK_STOPPED) { + al_assert(nn_cond_is_waiting(&track->cond)); nn_cond_signal(&track->cond); + } else { + al_assert(!nn_cond_is_waiting(&track->cond)); } - nn_mutex_unlock(&track->mutex); + track->state = VCR_TRACK_CLOSED; + nn_mutex_unlock(&track->lock); if (vcr->started) { al_assert(track->running); nn_thread_join(&track->thread); @@ -463,7 +479,7 @@ void lia_vcr_flush(struct lia_vcr *vcr) atomic_store(bool)(&track->buffered, false, AL_ATOMIC_RELAXED); track->client->flush(track->client); nn_packet_cache_enable(&track->cache); - atomic_store(s32)(&track->state, VCR_TRACK_RUNNING, AL_ATOMIC_RELAXED); + track->state = VCR_TRACK_RUNNING; } else { track->client->flush(track->client); } diff --git a/src/liana/vcr.h b/src/liana/vcr.h index 1ab58da..5b8b9df 100644 --- a/src/liana/vcr.h +++ b/src/liana/vcr.h @@ -14,11 +14,11 @@ struct lia_vcr_track { struct camu_codec_stream *stream; struct lia_client_handler *client; struct nn_packet_cache cache; - atomic(s32) state; + u32 state; atomic(bool) buffered; bool running; struct nn_cond cond; - struct nn_mutex mutex; + struct nn_mutex lock; struct nn_thread thread; struct lia_vcr *vcr; }; diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 0db5da2..80f40cd 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -1,6 +1,7 @@ #define AL_LOG_SECTION "sink" //#define AL_LOG_ENABLE_TRACE #include <al/log.h> +#include <al/random.h> #include <nnwt/multiplex.h> #include "../server/common.h" @@ -10,13 +11,11 @@ #include "sink.h" #include "common.h" -//#define CAMU_SINK_LOCAL //#define CAMU_SINK_ONESHOT // Requested state of the sinks outputs. enum { - SINK_EMPTY = 0, - SINK_PAUSED, + SINK_PAUSED = 0, SINK_PLAYING }; @@ -75,7 +74,7 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); // printf format for entries. #ifdef AL_DEBUG -#define ENTRY_FMT "#%u(%p)" +#define ENTRY_FMT "#%u(%p)" // @TODO: Put static and ended here at least. #define ENTRY_ARG(entry) (ENTRY_IS_VALID(entry) ? REMOTE_ENTRY_ID((entry)->id) : 0), ((entry) ? (entry) : 0x0) #else #define ENTRY_FMT "#%u" @@ -93,12 +92,15 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); #define AUDIO_ENDED(entry) (AUDIO_STATE(entry) >= BUFFER_ENDED) #define VIDEO_ENDED(entry) (VIDEO_STATE(entry) >= BUFFER_ENDED) -/* -#define ENTRY_ENDED(entry) \ + +#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)) + +#define ENTRY_EVAL_ENDED(entry) \ ((AUDIO_ENDED(entry) && (VIDEO_ENDED(entry) || VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry))) || \ (AUDIO_EMPTY(entry) && VIDEO_ENDED(entry))) -*/ -#define ENTRY_ENDED(entry) entry->ended +#define ENTRY_ENDED(entry) (al_assert(entry->ended == ENTRY_EVAL_ENDED(entry)), entry->ended) // IGNORED = ENDED or EMPTY. #define AUDIO_ADDED_OR_IGNORED(entry) (AUDIO_STATE(entry) >= BUFFER_ADDED || AUDIO_EMPTY(entry)) @@ -107,14 +109,10 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); #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 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) +#define BLOCKING_SLEEP(sink, delay) ((void)sink, nn_thread_sleep(delay)) #else -#define BLOCKING_SLEEP(delay) nn_event_loop_sleep(sink->loop, delay) +#define BLOCKING_SLEEP(sink, delay) nn_event_loop_sleep(sink->loop, delay) #endif #define CMD(cmd, ...) ((struct camu_sink_cmd){ .op = cmd, __VA_ARGS__ }) @@ -178,6 +176,7 @@ static void refresh_video_output(struct camu_sink *sink) static inline void add_entry_audio_buffer(struct camu_sink_entry *entry) { + log_trace("add_entry_audio_buffer("ENTRY_FMT").", ENTRY_ARG(entry)); struct camu_sink *sink = entry->sink; #ifdef CAMU_MIXER_THREADED_START_STOP sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); @@ -189,6 +188,7 @@ static inline void add_entry_audio_buffer(struct camu_sink_entry *entry) static inline void add_entry_video_buffer(struct camu_sink_entry *entry) { #ifndef CAMU_SINK_NO_VIDEO + log_trace("add_entry_video_buffer("ENTRY_FMT").", ENTRY_ARG(entry)); struct camu_sink *sink = entry->sink; sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); #else @@ -200,6 +200,8 @@ static void remove_entry_audio_buffer(struct camu_sink_entry *entry) { al_assert(!AUDIO_ENDED(entry)); al_assert(AUDIO_STATE(entry) != BUFFER_INIT); + log_trace("remove_entry_audio_buffer("ENTRY_FMT"), do_remove: %s.", ENTRY_ARG(entry), + BOOLSTR(AUDIO_STATE(entry) == BUFFER_ADDED)); switch (AUDIO_STATE(entry)) { case BUFFER_ADDED: AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED; @@ -224,6 +226,8 @@ 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_ENDED(entry)); al_assert(VIDEO_STATE(entry) != BUFFER_INIT); + log_trace("remove_entry_video_buffer("ENTRY_FMT"), do_remove: %s.", ENTRY_ARG(entry), + BOOLSTR(VIDEO_STATE(entry) == BUFFER_ADDED)); switch (VIDEO_STATE(entry)) { case BUFFER_ADDED: VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED; @@ -251,32 +255,12 @@ static void remove_entry_video_buffer(struct camu_sink_entry *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)); + log_trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", + ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(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); -static void add_video_if_set_and_buffered(struct camu_sink_entry *entry); - -static void add_or_queue_entry(struct camu_sink_entry *entry) -{ - log_trace("add_or_queue_entry("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); - al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); - al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); - // Either buffer could be DETACHED. - if (AUDIO_STATE(entry) == BUFFER_INIT) { - AUDIO_STATE(entry) = BUFFER_QUEUED; - } else if (!AUDIO_ENDED_OR_EMPTY(entry)) { - add_audio_if_set_and_buffered(entry); - } - if (VIDEO_STATE(entry) == BUFFER_INIT) { - VIDEO_STATE(entry) = BUFFER_QUEUED; - } else if (!VIDEO_ENDED_OR_EMPTY(entry)) { - add_video_if_set_and_buffered(entry); - } -} - // Disconnecting a packet stream twice before a reconnect is an error. static void maybe_disconnect_entry(struct camu_sink_entry *entry) { @@ -287,57 +271,20 @@ static void maybe_disconnect_entry(struct camu_sink_entry *entry) } } -#ifdef CAMU_SINK_LOCAL -static void local_entry_pause(struct camu_sink *sink, struct camu_sink_entry *entry) -{ - 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_ENDED_OR_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) { -#ifndef CAMU_SINK_NO_VIDEO - sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL); -#endif - sink->video.state = SINK_PAUSED; - } - } else { - entry->paused = false; - camu_clock_resume(&entry->clock, 0); - log_info("Clock resumed."); - if (!VIDEO_ENDED_OR_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_ENDED_OR_EMPTY(entry) && sink->audio.state == SINK_PAUSED) { - sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); - sink->audio.state = SINK_PLAYING; - } - } -} -#endif - // 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; } -static inline s32 get_sequence_for_command(struct camu_sink_entry *entry) +static inline s32 get_sequence_for_command(struct camu_sink *sink, struct camu_sink_entry *entry) { + // If local, using entry->sequence could only lead to feeling like your inputs were eaten. + if (!sink->local && entry) { + return entry->sequence; + } // SEQUENCE_ANY resolves order on the server. - s32 sequence = LIANA_SEQUENCE_ANY; - // 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; -#endif - return sequence; + return LIANA_SEQUENCE_ANY; } static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) @@ -433,33 +380,29 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) return; } case SKIP: { - if (!sink->conn) return; + if (!sink->connected) 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(entry)); + nn_packet_write_s32(packet, get_sequence_for_command(sink, entry)); nn_packet_write_s32(packet, (s32)cmd->v.i); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); break; } case TOGGLE_PAUSE: { + if (!sink->connected) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; -#ifdef CAMU_SINK_LOCAL - if (!entry->held) local_entry_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(entry)); + nn_packet_write_s32(packet, get_sequence_for_command(sink, entry)); nn_packet_write_f64(packet, cmd->v.f); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); -#endif break; } case SEEK: { - if (!sink->conn) return; + if (!sink->connected) 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); @@ -471,13 +414,12 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case RESEEK: { - if (!sink->conn) return; // Crude way to trigger "re-add sink to list". - nn_rpc_conn_disconnect(sink->conn); + if (sink->conn) nn_rpc_conn_disconnect(sink->conn); break; } case SHUFFLE: { - if (!sink->conn) return; + if (!sink->connected) 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_SHUFFLE); @@ -485,7 +427,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case END: { - if (!sink->conn) return; + if (!sink->connected) 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); @@ -513,14 +455,13 @@ 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 explicitly STOP the audio in places we - // expect the mixer to be empty (rely on MIXER_EMPTY). if (op == CAMU_MIXER_EMPTY) { log_info("Mixer empty."); - // 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. + // Regardless of if we are checking an entry's state, 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. + // This check is too loose. Still about as good as any naive implementation + // of SINK_EMPTY, though. if (!(sink->current && AUDIO_STATE(sink->current) == BUFFER_ADDED)) { queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO)); } @@ -572,13 +513,13 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink) } } -static void maybe_remove_previous(struct camu_sink *sink) +static void maybe_run_previous(struct camu_sink *sink) { - log_trace("maybe_remove_previous(), previous_count: %u.", sink->previous.count); + log_trace("maybe_run_previous(), previous_count: %u.", sink->previous.count); 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 + // Cleanup entries from old connections. Especially important to // keep reseek() from being overly wasteful. if (CONNECTION_NUMBER(previous->id) != sink->connection_number) { queue_cmd(sink, CMD(EJECT_ENTRY, .opaque = previous)); @@ -590,29 +531,43 @@ static void maybe_remove_previous(struct camu_sink *sink) // 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. 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) +static void run_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key) { bool removed = false; struct camu_sink_entry *previous; al_array_foreach(sink->previous, i, previous) { if (previous == key) { - maybe_remove_previous(sink); + maybe_run_previous(sink); removed = true; break; } } - log_trace("remove_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed)); + log_trace("run_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed)); +} + +static bool maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry) +{ + bool removed = false; + struct camu_sink_entry *previous; + al_array_foreach(sink->previous, i, previous) { + if (previous == entry) { + al_array_remove_at(sink->previous, i); + removed = true; + break; + } + } + log_trace("maybe_remove_from_previous("ENTRY_FMT"), removed: %s.", ENTRY_ARG(entry), BOOLSTR(removed)); + return removed; } // Every call to maybe_add_to_previous() must map to a remove_entry_buffers(). static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous, struct camu_sink_entry *target) { - log_trace("maybe_add_to_previous("ENTRY_FMT", "ENTRY_FMT").", ENTRY_ARG(previous), ENTRY_ARG(target)); + log_trace("maybe_add_to_previous("ENTRY_FMT"), target: "ENTRY_FMT".", ENTRY_ARG(previous), ENTRY_ARG(target)); al_assert(previous != target); - // If none of an entry's buffers are added, we don't care about adding it to previous. - bool dangling_target = target == (struct camu_sink_entry *)0xb00b; - if (dangling_target || ENTRY_ENDED(target) || (AUDIO_STATE(previous) != BUFFER_ADDED && VIDEO_STATE(previous) != BUFFER_ADDED)) { + bool ignore_target = target == (struct camu_sink_entry *)0xb00b || ENTRY_ENDED(target); + if (ignore_target || (AUDIO_STATE(previous) != BUFFER_ADDED && VIDEO_STATE(previous) != BUFFER_ADDED)) { remove_entry_buffers(previous); return; } @@ -621,15 +576,14 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry static void after_add_entry(struct camu_sink_entry *entry, bool skip_audio, bool skip_video) { - maybe_remove_previous(entry->sink); + struct camu_sink *sink = entry->sink; + maybe_run_previous(sink); queue_cmds(entry->sink, 2, CMD((skip_video || entry->paused) ? STOP : START, .v.u = CAMU_SINK_VIDEO), CMD((skip_audio || entry->paused) ? STOP : START, .v.u = CAMU_SINK_AUDIO) ); - if (VIDEO_ENDED_OR_EMPTY(entry)) { - // Clear the screen if skipping from a video to an audio-only entry. - refresh_video_output(entry->sink); - } + // Clear the screen if skipping from a video to an audio-only entry. + if (VIDEO_ENDED_OR_EMPTY(entry)) refresh_video_output(sink); } // Call this after setting state to ADDED because this entry might be in previous. @@ -638,13 +592,16 @@ 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); + if (!skip_audio && !skip_video) { + al_assert(VIDEO_STATE(entry) == AUDIO_STATE(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) +static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) { al_assert(!ENTRY_ENDED(entry)); al_assert(AUDIO_STATE(entry) != BUFFER_INIT); @@ -664,7 +621,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) } } -void add_video_if_set_and_buffered(struct camu_sink_entry *entry) +static void add_video_if_set_and_buffered(struct camu_sink_entry *entry) { // Single frame entries will be added/removed with ended set. if (ENTRY_ENDED(entry)) al_assert(VIDEO_IS_SINGLE_FRAME(entry)); @@ -691,6 +648,35 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) } } +static void add_or_queue_entry(struct camu_sink_entry *entry) +{ + log_trace("add_or_queue_entry("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", + ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); + al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); + al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); + // Either buffer could be DETACHED. + if (AUDIO_STATE(entry) == BUFFER_INIT) { + AUDIO_STATE(entry) = BUFFER_QUEUED; + } else if (!AUDIO_ENDED_OR_EMPTY(entry)) { + add_audio_if_set_and_buffered(entry); + } + if (VIDEO_STATE(entry) == BUFFER_INIT) { + VIDEO_STATE(entry) = BUFFER_QUEUED; + } else if (!VIDEO_ENDED_OR_EMPTY(entry)) { + add_video_if_set_and_buffered(entry); + } +} + +static void ensure_single_frame_removed(struct camu_sink_entry *entry) +{ + if (!VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) { + remove_entry_video_buffer(entry); + struct camu_sink *sink = entry->sink; + while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000)); } + } +} + +// This is the only function that sets sink->current. static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) { struct camu_sink_entry *current = sink->current; @@ -703,8 +689,13 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) if (suspended) { al_assert(suspended == current); // It should be impossible for a buffer to be QUEUED while it's entry is suspended. + // Even for a single frame video buffer because we wouldn't even know if it only + // has a single frame yet. al_assert(AUDIO_STATE(suspended) != BUFFER_QUEUED); al_assert(VIDEO_STATE(suspended) != BUFFER_QUEUED); + ensure_single_frame_removed(suspended); + al_assert(AUDIO_STATE(suspended) != BUFFER_ADDED); + al_assert(VIDEO_STATE(suspended) != BUFFER_ADDED); sink->suspended = NULL; log_warn("Unset suspended entry as a substitute for remove."); } else { @@ -713,46 +704,34 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) } bool dangling_target = target == (struct camu_sink_entry *)0xb00b; + if (dangling_target) log_trace("Ignoring dangling target."); bool stop_video = dangling_target; - if (dangling_target) { - log_trace("Ignored dangling target."); - } else { - remove_previous_if_contains(sink, target); - add_or_queue_entry(target); - stop_video = VIDEO_ENDED_OR_EMPTY(target) || VIDEO_IS_SINGLE_FRAME(target); - } - - if (dangling_target) { - sink->current = NULL; - } else { + if (!dangling_target) { target->audio.ignore_paused = false; - if (!target->paused) { + if (!sink->local && !target->paused) { camu_audio_buffer_resync(&target->audio.buf); } - sink->current = target; + if (!maybe_remove_from_previous(sink, target)) { + add_or_queue_entry(target); + } + al_assert(AUDIO_STATE(target) != BUFFER_INIT); + al_assert(VIDEO_STATE(target) != BUFFER_INIT); + // If target was just created, it's AUDIO/VIDEO_STATE() will be QUEUED. + // Meaning, at this point, it's video buffer would be considered empty + // as well as being too early to tell if it's static or not. + stop_video = VIDEO_ENDED_OR_EMPTY(target) || VIDEO_IS_SINGLE_FRAME(target); } if (stop_video) { queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO)); - refresh_video_output(sink); + if (dangling_target || VIDEO_ENDED_OR_EMPTY(target)) { + refresh_video_output(sink); + } } + + sink->current = dangling_target ? NULL : 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) { struct camu_sink_entry *current = sink->current; @@ -760,34 +739,29 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta ENTRY_ARG(target), at / 1000000.0, ENTRY_ARG(current)); al_assert(target != current); al_assert(!sink->target); - // This is extra verbose because the order is important. - // 1. sink->target has to be set before calling clock_pause(). - if (current && !ENTRY_ENDED(current)) { - sink->target = target; - } - // 2. current must still be paused, even if it's ended. + sink->target = target; + bool immediate = !current || ENTRY_ENDED(current); if (current) { current->audio.ignore_paused = true; - camu_clock_pause(¤t->clock, at); + immediate |= camu_clock_pause(¤t->clock, at); } - // 3. In the immediate swap case, switch_to() has to come last. - if (!current || ENTRY_ENDED(current)) { - switch_to(sink, target); + if (immediate) { + switch_to(sink, sink->target); + sink->target = NULL; } } -#endif static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry) { log_debug("Entry ("ENTRY_FMT") ended.", ENTRY_ARG(entry)); - remove_previous_if_contains(sink, entry); + run_previous_if_contains(sink, entry); #ifdef CAMU_SINK_ONESHOT sink->callback(sink->userdata, CAMU_SINK_MOCK_CLOSE, 0, NULL); return false; #endif // This entry's buffers cannot be added again until after a reset. entry->ended = true; - al_assert(ENTRY_ENDED(entry)); + al_assert(ENTRY_EVAL_ENDED(entry)); queue_cmd(sink, CMD(END, .v.u = entry->reset_token, .opaque = entry)); #ifdef LIANA_LIST_SCUFFED_LOOP log_info("Looping."); @@ -819,8 +793,13 @@ static void audio_buffer_callback(void *userdata, u8 op) case CAMU_BUFFER_UNCORK: lia_vcr_uncork(entry->audio.track); break; - case CAMU_BUFFER_PAUSED: + case CAMU_BUFFER_PAUSED: // Comes from audio read() thread. nn_mutex_lock(&sink->lock); + // entry->paused could plausibly be false here if the lock was held + // by pause_command_callback() to resume. This can be simulated by + // calling list_toggle_pause() twice in server/list_action_callback() + // for each sink request. Spaced by an nn_event_loop_sleep(~15500us) + // (no_video, MINIAUDIO_LOW_LATENCY mode). if (!entry->audio.ignore_paused && entry->paused) { log_info("Audio buffer paused."); queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO)); @@ -929,30 +908,31 @@ static void clock_callback(void *userdata, u8 op) static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry) { 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 - if (!AUDIO_EMPTY(entry) && !ignore_video) { - 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; - camu_audio_buffer_set_latency(&entry->audio.buf, -audio); - camu_video_buffer_set_latency(&entry->video.buf, -video); - } - // If we're local we don't have to worry about syncing audio-only entries. - camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video); - camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); -#else - // When trying to sync clients with different audio/video latencies (common case), - // our only option is to shift each buffer forward directly by their latency. f64 audio = camu_mixer_get_latency(sink->audio.mixer); + u32 frames = 0; + f64 video = 0.0; if (!ignore_video) { struct camu_renderer *renderer = sink->video.renderer; - f64 video = renderer->get_latency(renderer) * avg_frame_duration; - camu_video_buffer_set_latency(&entry->video.buf, video); + if (renderer) { + f64 avg_frame_duration = entry->video.buf.avg_frame_duration; + frames = renderer->get_latency(renderer); + video = frames * avg_frame_duration; + } } camu_audio_buffer_set_latency(&entry->audio.buf, audio); camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); -#endif + camu_video_buffer_set_latency(&entry->video.buf, video); + log_info("video_latency: %f (%u frames), audio_latency: %f.", video, frames, audio); + if (sink->local) { + camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video); + // When the video buffer starts the clock, we have to consider the audio + // buffer is treating the last period of silence sent during a paused clock + // as part of the stream. This poses an issue for sync because it requires more + // than a period-length offset to not skip data at the start. Combined with the + // fact that the timing of a request for the next period doesn't have be uniform. + // Could make a diagram of this. // @TODO + camu_clock_offset(&entry->clock, MAX(audio, video)); + } } static void run_queue_by_opaque(struct camu_sink *sink, void *opaque) @@ -1074,29 +1054,33 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str case LIANA_CLIENT_REMOVE_BUFFERS: { struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque; 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)); + log_trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s, single_frame: %s.", + ENTRY_ARG(entry), BOOLSTR(rec->reconnect), + BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current), + BOOLSTR(!VIDEO_EMPTY(entry) ? VIDEO_IS_SINGLE_FRAME(entry) : false)); if (entry == sink->current) { - al_assert(AUDIO_STATE(entry) != BUFFER_INIT); - al_assert(VIDEO_STATE(entry) != BUFFER_INIT); + if (rec->reconnect) { + // Else, possible failed or aborted reconnect (Empty buffer could be INIT). + 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; + } else if (rec->reconnect) { + // If this entry is still current on CLIENT_RECONNECTED, re-add it's buffers. + sink->suspended = entry; } } - // If this entry is still current on CLIENT_RECONNECTED, re-add it's buffers. - if (entry == sink->current && rec->reconnect) { - sink->suspended = 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. + run_previous_if_contains(sink, entry); + // AUDIO/VIDEO_STATE() could be INIT at this point, even if entry = current. // 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() @@ -1104,7 +1088,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str 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. + // Don't consider unconfigured entries past this point. if (rec->unconfigured) { al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry)); nn_mutex_unlock(&sink->lock); @@ -1136,7 +1120,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str nn_mutex_unlock(&sink->lock); while ((!skip_audio && entry_audio_buffer_held(entry)) || (!skip_video && entry_video_buffer_held(entry))) { - BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); + BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000)); } // At this point we can be sure that the entry's buffers are no longer in use. @@ -1150,13 +1134,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } 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. + } else if (VIDEO_STATE(entry) == BUFFER_ADDED && VIDEO_IS_SINGLE_FRAME(entry)) { + // Try to not request a duplicate frame. We can only be sure the frame wasn't + // dropped by the VCR if it's ADDED. rec->mask &= ~(1 << VIDEO_STREAM(entry)->index); } } - // Finalize the removal of the buffers by handling ended (ENDED or ERRORED) buffers. + // 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) { @@ -1178,13 +1163,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str break; } - case LIANA_CLIENT_RESUME_AT: { // This is called after the client reconnects, before CLIENT_RECONNECTED. + case LIANA_CLIENT_RECOVER_TO: { 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->pos = camu_clock_get_last_pts(&entry->clock) * (u64)1000000; time->at = 0; -#endif - log_trace("resume_at("ENTRY_FMT"), pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->pos / 1000000.0, entry->clock.paused_at); + break; + } + case LIANA_CLIENT_RESUME_AT: { // This is called after the client reconnects, before CLIENT_RECONNECTED. + struct lia_timing *time = (struct lia_timing *)opaque; + if (sink->local) time->at = 0; + f64 pos = time->pos / 1000000.0; + log_trace("resume_at("ENTRY_FMT"), pos: %f, paused_at: %f.", ENTRY_ARG(entry), pos, entry->clock.paused_at); nn_mutex_lock(&sink->lock); bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); if (!AUDIO_EMPTY(entry)) { @@ -1193,24 +1182,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); } if (!ignore_video) { - camu_video_buffer_reset(&entry->video.buf, time->pos / 1000000.0); - } -#ifdef LIANA_LIST_SCUFFED_LOOP - if (time->pos == 0) { - camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock)); - } else { -#endif - camu_clock_seek(&entry->clock, time->pos / 1000000.0, time->at); -#ifdef LIANA_LIST_SCUFFED_LOOP + camu_video_buffer_reset(&entry->video.buf, pos); } -#endif + camu_clock_seek(&entry->clock, pos, time->at); nn_mutex_unlock(&sink->lock); break; } case LIANA_CLIENT_RECONNECTED: { struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque; nn_mutex_lock(&sink->lock); - log_trace("reconnected("ENTRY_FMT"), suspended: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), ENTRY_ARG(sink->suspended), AUDIO_STATE(entry), VIDEO_STATE(entry)); + log_trace("reconnected("ENTRY_FMT"), suspended: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.", + ENTRY_ARG(entry), ENTRY_ARG(sink->suspended), AUDIO_STATE(entry), VIDEO_STATE(entry)); if (entry == sink->suspended) { al_assert(entry == sink->current); // Let this be the only other explicit BUFFER_ENDED check, or this will @@ -1262,11 +1244,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str nn_mutex_lock(&sink->lock); // We can be assured that CLIENT_REMOVE_BUFFERS has been called on this 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)); } - } + // If a client is closed after a failed reconnect, a single frame + // video buffer could still be added + if (rec->reconnect) ensure_single_frame_removed(entry); al_assert(AUDIO_STATE(entry) != BUFFER_ADDED); al_assert(VIDEO_STATE(entry) != BUFFER_ADDED); @@ -1291,7 +1271,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } } // If current was never fully added we need to call this here. - maybe_remove_previous(sink); + maybe_run_previous(sink); } nn_mutex_unlock(&sink->lock); @@ -1316,7 +1296,6 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id) entry->ended = false; camu_clock_init(&entry->clock, clock_callback, entry); - entry->held = false; entry->client.callback = client_callback; entry->client.userdata = entry; @@ -1412,25 +1391,8 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, goto out; } -#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)); - if (current) { - current->audio.ignore_paused = true; - if (!current->paused) { - // paused has to map directly to the clock state here. - al_assert(!camu_clock_is_paused(¤t->clock)); - camu_clock_pause(¤t->clock, 0); - } - current->held = true; - } - entry->held = false; - if (!entry->paused) { - camu_clock_resume(&entry->clock, 0); - } - switch_to(sink, entry); -#else + if (sink->local) at = 0; + // For target to be set that must mean current is set, armed to pause, and not ended. struct camu_sink_entry *prev_target = sink->target; log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry), @@ -1501,7 +1463,6 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, camu_clock_resume(&entry->clock, at); break; } -#endif out: nn_mutex_unlock(&sink->lock); @@ -1532,10 +1493,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con al_assert(entry->sequence == sequence); 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; - if (!entry->held) local_entry_pause(sink, entry); -#else + if (sink->local) at = 0; switch (pause) { case LIANA_PAUSE_PAUSE: { entry->paused = true; @@ -1553,13 +1511,14 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_VIDEO)); } if (!AUDIO_ENDED_OR_EMPTY(entry)) { - camu_audio_buffer_resync(&entry->audio.buf); + if (!sink->local) { + camu_audio_buffer_resync(&entry->audio.buf); + } queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_AUDIO)); } break; } } -#endif nn_mutex_unlock(&sink->lock); out: @@ -1629,8 +1588,12 @@ static void identify_on_connection(struct camu_sink *sink) static void connection_callback(void *userdata, struct nn_rpc_connection *conn) { struct camu_sink *sink = (struct camu_sink *)userdata; + nn_timer_stop(&sink->reconnect_timer); + if (sink->conn) al_assert(sink->conn == conn); sink->conn = conn; - sink->connection_number++; + sink->connected = true; + sink->connection_number = al_u16_inc_wrap(sink->connection_number); + if (sink->connection_number == 0) sink->connection_number = 1; identify_on_connection(sink); } @@ -1638,25 +1601,31 @@ static void reconnect_timer_callback(void *userdata, struct nn_timer *timer) { struct camu_sink *sink = (struct camu_sink *)userdata; (void)timer; - nn_timer_stop(&sink->reconnect_timer); - nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); + if (sink->conn) { + // If the client was connecting, reconnect() will force a disconnect before reconnecting. + log_warn("Forcing reconnect due to timeout."); + } + sink->conn = nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); } static void connection_closed_callback(void *userdata, struct nn_rpc_connection *conn) { struct camu_sink *sink = (struct camu_sink *)userdata; - bool reconnect = !sink->reconnect_timer.disabled; + bool reconnect = sink->conn != NULL; bool disconnected = sink->conn && sink->connection_number > 0; if (sink->conn) { al_assert(sink->conn == conn); sink->conn = NULL; + sink->connected = false; + } else { + al_assert(!reconnect || !sink->connected); } if (reconnect) { if (disconnected) { - log_info("Connection to server closed, attempting reconnect..."); + log_warn("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..."); + log_warn("Failed to connect to server, trying again..."); nn_timer_again(&sink->reconnect_timer); } } @@ -1671,7 +1640,8 @@ 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(1500000)); + // This is also effectively a timeout for attempted reconnects. + nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1000000)); nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink); nn_signal_start(&sink->queue_signal); camu_queue_init(sink->queue); @@ -1681,7 +1651,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, al_array_init(sink->previous); al_array_init(sink->entries); // Start high to exercise the wrapping path. - sink->lru = SINK_LRU_MAX - 2; + sink->lru = SINK_LRU_MAX - al_random_int(0, ENTRY_MAX_AGE); mixer->callback = mixer_callback; mixer->userdata = sink; sink->audio.state = SINK_PAUSED; @@ -1704,6 +1674,7 @@ bool camu_sink_connect(struct camu_sink *sink, str *name, u8 type, str *addr, u1 nn_rpc_add_command(&sink->client, &commands[i]); } #ifdef CAMU_DIRECT_MODE + // Note that direct_connect() runs connection_callback() directly. nn_multiplex_direct_connect(sink->conn->stream, CAMU_MULTIPLEX_RPC); #else nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port); @@ -1736,7 +1707,9 @@ void camu_sink_toggle_pause(struct camu_sink *sink) struct camu_sink_entry *current = get_entry_for_command(sink); nn_mutex_unlock(&sink->lock); if (!current) return; - f64 pts = camu_clock_get_pts(¤t->clock, 0.0, false); + bool armed_for_pause = false; + camu_clock_external_pause(¤t->clock); + f64 pts = camu_clock_get_pts(¤t->clock, 0.0, false, &armed_for_pause); queue_cmd(sink, CMD(TOGGLE_PAUSE, .v.f = pts, .opaque = current)); } @@ -1791,7 +1764,7 @@ void camu_sink_shuffle(struct camu_sink *sink) void camu_sink_stop(struct camu_sink *sink) { queue_cmds(sink, 3, - CMD(STOP, .v.u = CAMU_SINK_AUDIO), + CMD(STOP, .v.u = CAMU_SINK_AUDIO), CMD(CLEAR_BUFFERS, .v.u = CAMU_SINK_AUDIO), CMD(CLOSE) ); @@ -1800,8 +1773,11 @@ void camu_sink_stop(struct camu_sink *sink) void camu_sink_close(struct camu_sink *sink) { nn_timer_stop(&sink->reconnect_timer); - nn_timer_disable(&sink->reconnect_timer); - if (sink->conn) nn_rpc_conn_disconnect(sink->conn); + if (sink->conn) { + struct nn_rpc_connection *conn = sink->conn; + sink->conn = NULL; // Signal to connection_closed_callback() we're done. + nn_rpc_conn_disconnect(conn); + } struct camu_sink_entry *entry; al_array_foreach_rev(sink->entries, i, entry) { al_array_remove_at(sink->entries, i); diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 1e8bbf1..acb6a51 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -52,7 +52,6 @@ struct camu_sink_entry { bool ended; u32 reset_token; struct camu_clock clock; - bool held; // Only used in SINK_LOCAL. bool paused; struct { u8 state; @@ -83,7 +82,9 @@ struct camu_sink { u16 port; struct nn_rpc client; struct nn_rpc_connection *conn; + bool connected; u16 connection_number; + bool local; struct nn_mutex lock; struct nn_timer reconnect_timer; struct nn_signal queue_signal; diff --git a/src/render/queue_libplacebo.c b/src/render/queue_libplacebo.c index e3540c3..408af51 100644 --- a/src/render/queue_libplacebo.c +++ b/src/render/queue_libplacebo.c @@ -13,27 +13,63 @@ static bool queue_lp_configure_subtitles(struct camu_frame_queue *queue, u32 wid { struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue; #ifdef CAMU_HAVE_SUBTITLES - if (!lq->ass) return false; - lq->ass_renderer = ass_renderer_init(lq->ass); - if (!lq->ass_renderer) { + ASS_Library *ass = lq->subs.ass; + if (!(lq->subs.renderer = ass_renderer_init(ass))) { log_error("Failed to initialize ass renderer."); return false; } - ass_set_shaper(lq->ass_renderer, ASS_SHAPING_COMPLEX); - lq->ass_track = ass_new_track(lq->ass); - if (!lq->ass_track) { + if (!(lq->subs.track = ass_new_track(ass))) { log_error("Failed to create ass track."); return false; } - ass_set_frame_size(lq->ass_renderer, width, height); - ass_set_storage_size(lq->ass_renderer, width, height); - ass_set_fonts(lq->ass_renderer, NULL, NULL, ASS_FONTPROVIDER_AUTODETECT, NULL, 0); - ass_set_hinting(lq->ass_renderer, ASS_HINTING_NONE); + ASS_Renderer *renderer = lq->subs.renderer; + ASS_Track *track = lq->subs.track; + ass_set_shaper(renderer, ASS_SHAPING_COMPLEX); + ass_set_frame_size(renderer, width, height); + ass_set_storage_size(renderer, width, height); + ass_set_fonts(renderer, NULL, NULL, ASS_FONTPROVIDER_AUTODETECT, NULL, 0); + ass_set_hinting(renderer, ASS_HINTING_NONE); AVCodecParameters *codecpar = stream->av.stream->codecpar; - ass_process_codec_private(lq->ass_track, (const char *)codecpar->extradata, - codecpar->extradata_size); - nn_mutex_init(&lq->subtitle_lock); - lq->has_subtitles = true; + lq->subs.conversion_needed = codecpar->codec_id != AV_CODEC_ID_ASS; + if (lq->subs.conversion_needed) { + const AVCodecDescriptor *desc = avcodec_descriptor_get(codecpar->codec_id); + if (!(desc->props & AV_CODEC_PROP_TEXT_SUB)) { + log_error("Cannot convert non-text (bitmap) subtitles."); + return false; + } + const AVCodec *codec = avcodec_find_decoder(codecpar->codec_id); + if (!codec) { + log_error("Failed to find subtitle decoder."); + return false; + } + if (!(lq->subs.converter = camu_ff_alloc_codec_context(codec, codecpar))) { + return false; + } + AVCodecContext *converter = lq->subs.converter; + converter->sub_charenc_mode = FF_SUB_CHARENC_MODE_IGNORE; + if (!camu_ff_open_avcodec(converter, codec, NULL)) { + return false; + } + const AVCodecDescriptor *ass_desc = avcodec_descriptor_get(AV_CODEC_ID_ASS); + log_info("Subtitle conversion: %s -> %s.", desc->long_name, ass_desc->long_name); + // https://github.com/mpv-player/mpv/blob/266cb79f38fd1a5fd448b453dee5971795a145ca/sub/sd_ass.c#L634 +#define MP_ASS_FONT_PLAYRESX 384.0 + track->PlayResX = track->PlayResY * (f64)width / MAX(height, (u32)1); + f64 fix_margins = track->PlayResX / MP_ASS_FONT_PLAYRESX; + f64 font_scale = 1.0; + for (s32 n = 0; n < track->n_styles; n++) { + track->styles[n].MarginL = track->styles[n].MarginL * fix_margins; + track->styles[n].MarginR = track->styles[n].MarginR * fix_margins; + track->styles[n].MarginV = track->styles[n].MarginV * font_scale; + } + ass_process_codec_private(track, (const char *)converter->subtitle_header, + converter->subtitle_header_size); + } else { + ass_process_codec_private(track, (const char *)codecpar->extradata, + codecpar->extradata_size); + } + nn_mutex_init(&lq->subs.lock); + lq->subs.are_present = true; return true; #else (void)lq; @@ -65,11 +101,6 @@ static bool map_frame(pl_gpu gpu, pl_tex *tex, const struct pl_source_frame *src masks[2] = 0x00ff0000; pixel_stride = 3; break; - case CAMU_PIXEL_FORMAT_GREYA: - masks[0] = 0x000000ff; - masks[1] = 0x0000ff00; - pixel_stride = 2; - break; case CAMU_PIXEL_FORMAT_GREY: masks[0] = 0x000000ff; pixel_stride = 1; @@ -79,25 +110,26 @@ static bool map_frame(pl_gpu gpu, pl_tex *tex, const struct pl_source_frame *src al_assert_and_return(false); } - struct pl_plane_data data = { - .type = PL_FMT_UNORM, - .width = frame->video.width, - .height = frame->video.height, - .pixel_stride = pixel_stride, - .row_stride = frame->video.width * pixel_stride, - .pixels = frame->data - }; - pl_plane_data_from_mask(&data, masks); + if (frame->format == CAMU_PIXEL_FORMAT_GREY) { + out_frame->repr.sys = pl_system_from_av(AVCOL_SPC_BT470BG); + out_frame->repr.levels = pl_levels_from_av(AVCOL_RANGE_JPEG); + } else { + out_frame->repr = pl_color_repr_unknown; + } + out_frame->color = pl_color_space_unknown; + out_frame->crop = (pl_rect2df){ 0.f, 0.f, frame->video.width, frame->video.height }; - struct pl_plane plane = { 0 }; - bool ok = pl_upload_plane(gpu, &plane, tex, &data); + struct pl_plane_data data[4] = { 0 }; + data[0].type = PL_FMT_UNORM; + data[0].width = frame->video.width; + data[0].height = frame->video.height; + data[0].pixel_stride = pixel_stride; + data[0].row_stride = frame->video.width * pixel_stride; + data[0].pixels = frame->data; + pl_plane_data_from_mask(&data[0], masks); - out_frame->num_planes = 1; - out_frame->planes[0] = plane; - out_frame->repr = pl_color_repr_unknown; - out_frame->color = pl_color_space_monitor; - out_frame->crop = (pl_rect2df){ 0.f, 0.f, frame->video.width, frame->video.height }; - //out_frame->profile = + struct pl_plane plane = { 0 }; + bool ok = pl_upload_plane(gpu, &plane, tex, &data[0]); camu_codec_frame_discard(frame); @@ -106,6 +138,9 @@ static bool map_frame(pl_gpu gpu, pl_tex *tex, const struct pl_source_frame *src return false; } + out_frame->num_planes = 1; + out_frame->planes[0] = plane; + return true; } @@ -153,7 +188,7 @@ static void free_overlay(pl_gpu gpu, struct camu_overlay *overlay) static struct camu_overlay *first_nonref_overlay(struct camu_frame_queue_lp *lq, u32 *index) { struct camu_overlay *overlay; - al_array_foreach(lq->subtitles, i, overlay) { + al_array_foreach(lq->subs.overlays, i, overlay) { if (!overlay->ref) { *index = i; return overlay; @@ -224,10 +259,10 @@ static struct camu_overlay *create_subtitle_overlay(struct camu_overlay *prev, p ass_frame->dst_y + ass_frame->h }; u32 c = ass_frame->color; - current_part->color[0] = (c >> 24) / 255.0; - current_part->color[1] = ((c >> 16) & 0xff) / 255.0; - current_part->color[2] = ((c >> 8) & 0xff) / 255.0; - current_part->color[3] = 1.0 - (c & 0xff) / 255.0; + current_part->color[0] = (c >> 24) / 255.f; + current_part->color[1] = ((c >> 16) & 0xff) / 255.f; + current_part->color[2] = ((c >> 8) & 0xff) / 255.f; + current_part->color[3] = 1.f - (c & 0xff) / 255.f; current->parts = current_part; current->num_parts = 1; @@ -265,7 +300,7 @@ static bool map_av_frame(pl_gpu gpu, pl_tex *tex, const struct pl_source_frame * // pl_map_avframe_derived() will fail under Xwayland because of a lack of EGL_EXT_image_dma_buf_import. if (!ok && frame->hw_frames_ctx) { if (!lq->copy_frame_fallback) { - log_warn("Falling back to software copy."); + log_warn("Failed to map hwframe, falling back to software copy."); lq->copy_frame_fallback = true; lq->swframe = av_frame_alloc(); enum AVPixelFormat *fmts; @@ -290,27 +325,27 @@ static bool map_av_frame(pl_gpu gpu, pl_tex *tex, const struct pl_source_frame * ((struct pl_source_frame *)src)->frame_data = NULL; #ifdef CAMU_HAVE_SUBTITLES - if (ok && lq->has_subtitles) { + if (ok && lq->subs.are_present) { s64 now = av_rescale_q(frame->best_effort_timestamp, stream->time_base, (AVRational){ 1, 1000 }); s32 change = -1; // 1 = different position, 2 = different content. - nn_mutex_lock(&lq->subtitle_lock); - ASS_Image *ass_frame = ass_render_frame(lq->ass_renderer, lq->ass_track, now, &change); + nn_mutex_lock(&lq->subs.lock); + ASS_Image *ass_frame = ass_render_frame(lq->subs.renderer, lq->subs.track, now, &change); al_assert(change >= 0); - // ass_frame = NULL and lq->ass_track->n_events = 0 can also mean parsing completely failed. + // ass_frame = NULL and lq->subs.track->n_events = 0 can also mean parsing completely failed. if (ass_frame) { - struct camu_overlay *overlay = lq->last_subtitle; + struct camu_overlay *overlay = lq->subs.last; if (change != 0) { u32 reused = AL_ARRAY_NO_INDEX; overlay = create_subtitle_overlay(first_nonref_overlay(lq, &reused), gpu, ass_frame); // If all texture uploads failed, overlay->num will be 0. - lq->last_subtitle = overlay; + lq->subs.last = overlay; if (overlay) { if (reused != AL_ARRAY_NO_INDEX) { // Pointer may have changed due to reallocation. - al_array_at(lq->subtitles, reused) = overlay; + al_array_at(lq->subs.overlays, reused) = overlay; } else { // New overlay object. - overlay->lock = &lq->subtitle_lock; - al_array_push(lq->subtitles, overlay); + overlay->lock = &lq->subs.lock; + al_array_push(lq->subs.overlays, overlay); } } } @@ -322,7 +357,7 @@ static bool map_av_frame(pl_gpu gpu, pl_tex *tex, const struct pl_source_frame * out_frame->overlays = NULL; } } - nn_mutex_unlock(&lq->subtitle_lock); + nn_mutex_unlock(&lq->subs.lock); } #endif @@ -373,15 +408,40 @@ static void queue_lp_push_av_frame(struct camu_frame_queue *queue, AVFrame *fram } #endif +static bool convert_subtitle(struct camu_frame_queue_lp *lq, AVPacket *pkt) +{ + AVSubtitle sub = { 0 }; + s32 got_sub = 0; + s32 ret = avcodec_decode_subtitle2(lq->subs.converter, &sub, &got_sub, pkt); + if (ret < 0) { + log_error("Failed to convert subtitle (%s).", av_err2str(ret)); + } else if (got_sub) { + for (u32 i = 0; i < sub.num_rects; i++) { + AVSubtitleRect *rect = sub.rects[i]; + size_t size = al_strlen(rect->ass); + if (rect->ass[size - 1] == 0x02) size--; // Not sure what this is... + ass_process_chunk(lq->subs.track, rect->ass, size, pkt->pts, pkt->duration); + } + avsubtitle_free(&sub); + return true; + } + return false; +} + static void queue_lp_push_subtitle(struct camu_frame_queue *queue, struct camu_codec_packet *packet) { struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue; #ifdef CAMU_HAVE_SUBTITLES + if (!lq->subs.are_present) return; AVPacket *pkt = packet->av.pkt; - nn_mutex_lock(&lq->subtitle_lock); - al_assert(lq->ass_track && lq->ass_renderer); - ass_process_chunk(lq->ass_track, (const char *)pkt->data, pkt->size, pkt->pts, pkt->duration); - nn_mutex_unlock(&lq->subtitle_lock); + nn_mutex_lock(&lq->subs.lock); + al_assert(lq->subs.track && lq->subs.renderer); + if (lq->subs.conversion_needed) { + convert_subtitle(lq, pkt); + } else { + ass_process_chunk(lq->subs.track, (const char *)pkt->data, pkt->size, pkt->pts, pkt->duration); + } + nn_mutex_unlock(&lq->subs.lock); #else (void)lq; (void)packet; @@ -392,6 +452,16 @@ static void queue_lp_flush(struct camu_frame_queue *queue) { struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue; pl_queue_push(lq->queue, NULL); +#ifdef CAMU_HAVE_SUBTITLES + if (lq->subs.are_present) { + if (lq->subs.conversion_needed) { + AVPacket *empty_pkt = av_packet_alloc(); + while (convert_subtitle(lq, empty_pkt)) { } + av_packet_free(&empty_pkt); + } + ass_flush_events(lq->subs.track); + } +#endif } static s32 queue_lp_count(struct camu_frame_queue *queue) @@ -422,7 +492,7 @@ static u8 queue_lp_read(struct camu_frame_queue *queue, f64 pts, void *out) static void queue_lp_reset(struct camu_frame_queue *queue) { struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue; - pl_queue_push(lq->queue, NULL); + queue_lp_flush(&lq->q); pl_queue_reset(lq->queue); } @@ -436,15 +506,22 @@ static void queue_lp_free(struct camu_frame_queue **queue) } #endif #ifdef CAMU_HAVE_SUBTITLES - if (lq->has_subtitles) { - struct camu_overlay *overlay; - al_array_foreach_rev(lq->subtitles, i, overlay) { - free_overlay(lq->gpu, overlay); - } - al_array_free(lq->subtitles); - ass_free_track(lq->ass_track); - ass_renderer_done(lq->ass_renderer); - nn_mutex_destroy(&lq->subtitle_lock); + struct camu_overlay *overlay; + al_array_foreach_rev(lq->subs.overlays, i, overlay) { + free_overlay(lq->gpu, overlay); + } + al_array_free(lq->subs.overlays); + if (lq->subs.are_present) { + nn_mutex_destroy(&lq->subs.lock); + } + if (lq->subs.converter) { + avcodec_free_context(&lq->subs.converter); + } + if (lq->subs.track) { + ass_free_track(lq->subs.track); + } + if (lq->subs.renderer) { + ass_renderer_done(lq->subs.renderer); } #endif al_free(lq); @@ -455,9 +532,9 @@ struct camu_frame_queue *camu_frame_queue_lp_create(void) { struct camu_frame_queue_lp *lq = al_alloc_object(struct camu_frame_queue_lp); #ifdef CAMU_HAVE_SUBTITLES - lq->last_subtitle = NULL; - al_array_init(lq->subtitles); - lq->has_subtitles = false; + lq->subs.are_present = false; + lq->subs.last = NULL; + al_array_init(lq->subs.overlays); #endif lq->q.configure_subtitles = queue_lp_configure_subtitles; lq->q.push = queue_lp_push; diff --git a/src/render/queue_libplacebo.h b/src/render/queue_libplacebo.h index 0d49000..f72cb88 100644 --- a/src/render/queue_libplacebo.h +++ b/src/render/queue_libplacebo.h @@ -4,7 +4,6 @@ #ifdef CAMU_HAVE_FFMPEG #include <al/lib.h> AL_IGNORE_WARNING("-Wswitch") -AL_IGNORE_WARNING("-Wunused-parameter") // PL_LIBAV_IMPLEMENTATION defined in queue_libplacebo.c. #include <libplacebo/utils/libav.h> AL_IGNORE_WARNING_END @@ -25,13 +24,17 @@ struct camu_frame_queue_lp { AVFrame *swframe; #endif #ifdef CAMU_HAVE_SUBTITLES - ASS_Library *ass; - ASS_Renderer *ass_renderer; - ASS_Track *ass_track; - struct nn_mutex subtitle_lock; - struct camu_overlay *last_subtitle; - array(struct camu_overlay *) subtitles; - bool has_subtitles; + struct { + bool are_present; + bool conversion_needed; + AVCodecContext *converter; + ASS_Library *ass; + ASS_Renderer *renderer; + ASS_Track *track; + struct camu_overlay *last; + array(struct camu_overlay *) overlays; + struct nn_mutex lock; + } subs; #endif }; diff --git a/src/render/renderer_libplacebo.c b/src/render/renderer_libplacebo.c index 0995f6f..48b7ea6 100644 --- a/src/render/renderer_libplacebo.c +++ b/src/render/renderer_libplacebo.c @@ -20,7 +20,14 @@ #define RENDERER_DEBUG 0 #endif +#define CRT_BACKGROUND +#define CRT_BACKGROUND_COLOR (1.f / 7.5f) + +#ifdef CRT_BACKGROUND +static f32 clear_color[4] = { CRT_BACKGROUND_COLOR, CRT_BACKGROUND_COLOR, CRT_BACKGROUND_COLOR, 1.f }; +#else static f32 clear_color[4] = { 0.f, 0.f, 0.f, 1.f }; +#endif static struct pl_render_params default_params; @@ -146,9 +153,11 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid { struct camu_renderer_lp *lr = (struct camu_renderer_lp *)renderer; +#ifndef CRT_BACKGROUND clear_color[0] = ((global_color_palette.colors[0] >> 16) & 0xff) / 255.f; clear_color[1] = ((global_color_palette.colors[0] >> 8) & 0xff) / 255.f; clear_color[2] = ((global_color_palette.colors[0]) & 0xff) / 255.f; +#endif lr->logger = pl_log_create(PL_API_VER, pl_log_params( .log_cb = log_callback, @@ -173,13 +182,13 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid )); if (!lr->vk_inst) { log_error("Failed to create vulkan instance."); - goto err; + return false; } VkResult ret = vk_create_surface(priv, lr->vk_inst->instance, &lr->surface); if (ret != VK_SUCCESS) { log_error("Failed to create vulkan surface."); - goto err; + return false; } lr->vk = pl_vulkan_create(lr->logger, pl_vulkan_params( @@ -190,7 +199,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid )); if (!lr->vk) { log_error("Failed to create vulkan device."); - goto err; + return false; } lr->swapchain = pl_vulkan_create_swapchain(lr->vk, pl_vulkan_swapchain_params( @@ -200,7 +209,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid )); if (!lr->swapchain) { log_error("Failed to create vulkan swapchain."); - goto err; + return false; } lr->gpu = lr->vk->gpu; @@ -213,18 +222,21 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid )); if (!lr->d3d11) { log_error("Failed to create D3D11 device."); - goto err; + return false; } // https://learn.microsoft.com/en-us/windows/win32/api/dxgi/ne-dxgi-dxgi_swap_chain_flag + // https://learn.microsoft.com/en-us/windows/win32/api/dxgi/ne-dxgi-dxgi_swap_effect + char *FLIP = getenv("SINK_DXGI_FLIP"); + if (!FLIP) FLIP = "0"; lr->swapchain = pl_d3d11_create_swapchain(lr->d3d11, pl_d3d11_swapchain_params( .window = win32_window, - //.blit = true, + .blit = al_strscmp(FLIP, "0") == 0, //.flags = DXGI_SWAP_CHAIN_FLAG_ALLOW_TEARING )); if (!lr->swapchain) { log_error("Failed to create D3D11 swapchain."); - goto err; + return false; } lr->gpu = lr->d3d11->gpu; @@ -247,7 +259,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid )); if (!lr->gl) { log_error("Failed to create GL device."); - goto err; + return false; } // max_swapchain_depth is returned as-is by pl_swapchain_latency(). @@ -259,7 +271,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid )); if (!lr->swapchain) { log_error("Failed to create GL swapchain."); - goto err; + return false; } lr->gpu = lr->gl->gpu; @@ -307,15 +319,16 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, u32 *wid #ifdef CAMU_HAVE_SUBTITLES lr->ass = ass_library_init(); + if (!lr->ass) { + log_error("Failed to initialize libass."); + return false; + } ass_set_message_cb(lr->ass, ass_log_callback, lr); #endif lr->last_swap_tick = nn_get_tick(); return true; -err: - lr->r.free((struct camu_renderer **)&lr); - return false; } static struct camu_frame_queue *renderer_lp_create_queue(struct camu_renderer *renderer) @@ -333,7 +346,7 @@ static struct camu_frame_queue *renderer_lp_create_queue(struct camu_renderer *r lq->copy_frame_fallback = false; #endif #ifdef CAMU_HAVE_SUBTITLES - lq->ass = lr->ass; + lq->subs.ass = lr->ass; #endif return queue; } @@ -385,8 +398,13 @@ static bool renderer_lp_render(struct camu_renderer *renderer, struct camu_scree struct pl_swapchain_frame frame; if (pl_swapchain_start_frame(lr->swapchain, &frame)) { pl_frame_from_swapchain(&lr->target, &frame); - pl_frame_clear_rgba(lr->gpu, &lr->target, clear_color); lr->have_frame = true; +#ifdef CRT_BACKGROUND + lr->frame_cleared = false; +#else + pl_frame_clear_rgba(lr->gpu, &lr->target, clear_color); + lr->frame_cleared = true; +#endif } else { log_error("Failed to start frame, fatal."); return false; @@ -473,6 +491,15 @@ static bool renderer_lp_render(struct camu_renderer *renderer, struct camu_scree target->color.hdr.min_luma = (PL_COLOR_SDR_WHITE - 50.f) / PL_COLOR_SDR_CONTRAST; } */ +#ifdef CRT_BACKGROUND + if (!lr->frame_cleared) { + clear_color[0] = 0.03f; + clear_color[1] = 0.03f; + clear_color[2] = 0.03f; + pl_frame_clear_rgba(lr->gpu, &lr->target, clear_color); + lr->frame_cleared = true; + } +#endif pl_render_image_mix(lr->renderer, &mix, target, &lr->params); } } else { @@ -503,6 +530,16 @@ static bool renderer_lp_render(struct camu_renderer *renderer, struct camu_scree return true; } +#ifdef CRT_BACKGROUND + if (!lr->frame_cleared) { + clear_color[0] = CRT_BACKGROUND_COLOR; + clear_color[1] = CRT_BACKGROUND_COLOR; + clear_color[2] = CRT_BACKGROUND_COLOR; + pl_frame_clear_rgba(lr->gpu, &lr->target, clear_color); + lr->frame_cleared = true; + } +#endif + if (pl_swapchain_submit_frame(lr->swapchain)) { lr->have_frame = false; } else { diff --git a/src/render/renderer_libplacebo.h b/src/render/renderer_libplacebo.h index 921c7ac..532963f 100644 --- a/src/render/renderer_libplacebo.h +++ b/src/render/renderer_libplacebo.h @@ -33,6 +33,7 @@ struct camu_renderer_lp { pl_swapchain swapchain; struct pl_frame target; bool have_frame; + bool frame_cleared; pl_renderer renderer; f64 last_swap_tick; struct pl_render_params params; diff --git a/src/screen/screen.c b/src/screen/screen.c index 0a0473d..cdc42bf 100644 --- a/src/screen/screen.c +++ b/src/screen/screen.c @@ -3,6 +3,7 @@ #include <al/log.h> #include <al/math.h> #include <nnwt/thread.h> +#include <nnwt/time.h> #include "view.h" #include "screen.h" @@ -15,6 +16,7 @@ #define SCREEN_IS_DRAGGING(scr) ((scr)->flags & CAMU_SCREEN_DRAGGING) #define SCREEN_IS_ZOOMING(scr) ((scr)->flags & CAMU_SCREEN_ZOOMING) +#define SCREEN_IS_FROZEN(scr) ((scr)->flags & CAMU_SCREEN_FROZEN) #define SCREEN_ZOOM_MODE(scr, mode) ((scr)->flags & mode) #define SCREEN_DEFAULT_CURSOR(scr) ((scr->fullscreen || scr->vr_emulation) ? STELA_CURSOR_HIDDEN : STELA_CURSOR_NORMAL) @@ -91,7 +93,7 @@ static bool pointer_pos_callback(void *userdata, f64 x, f64 y) if (SCREEN_IS_DRAGGING(scr)) { #ifdef CAMU_SCREEN_DRAG_SEEK u64 now = nn_get_timestamp(); - if (view && now - scr->last_seek_ts > 110000) { + if (view && now - scr->last_seek_ts > 100000) { seek_to_percent_at_pointer(scr, x); scr->last_seek_ts = now; } @@ -104,7 +106,7 @@ static bool pointer_pos_callback(void *userdata, f64 x, f64 y) view->mode = CAMU_VIEW_DETACHED; queue_refresh = true; } - if (fabs(dx) + fabs(dy) > 3.0) { + if (fabs(dx) + fabs(dy) > 6.0) { scr->last_click_ts = SCREEN_INVALID_TS; } } @@ -120,57 +122,109 @@ static bool scroll_callback(void *userdata, f64 y); static bool touch_callback(void *userdata, s32 index, u8 phase, f64 x, f64 y) { - struct camu_screen *scr = (struct camu_screen *)userdata; // Rough testing stuff. +#define distance(x1, x2, y1, y2) sqrt(((x2 - x1) * (x2 - x1)) + ((y2 - y1) * (y2 - y1))) + struct camu_screen *scr = (struct camu_screen *)userdata; switch (phase) { case STELA_TOUCH_BEGAN: { if (index == 0) { + scr->last_touch_x[0] = x; + scr->last_touch_y[0] = y; + } else if (index == 1) { + scr->last_touch_x[1] = x; + scr->last_touch_y[1] = y; + } + if (!SCREEN_IS_DRAGGING(scr)) { scr->flags |= CAMU_SCREEN_DRAGGING; scr->last_click_ts = nn_get_timestamp(); - scr->last_pointer_x = x; - scr->last_pointer_y = y; - } else if (index == 1) { - scr->flags &= ~CAMU_SCREEN_DRAGGING; + } else if (!SCREEN_IS_ZOOMING(scr)) { scr->flags |= CAMU_SCREEN_ZOOMING; - } else if (index == 2) { - scr->flags &= ~CAMU_SCREEN_ZOOMING; + f64 lx = scr->last_touch_x[0], lx2 = scr->last_touch_x[1]; + f64 ly = scr->last_touch_y[0], ly2 = scr->last_touch_y[1]; + scr->last_distance = distance(lx, lx2, ly, ly2); + scr->midpoint_x = (lx + lx2) / 2.0; + scr->midpoint_y = (ly + ly2) / 2.0; + } else if (!SCREEN_IS_FROZEN(scr)) { + scr->flags |= CAMU_SCREEN_FROZEN; struct camu_view *view = get_view_from_mouse_pos(scr); - view->mode = CAMU_DEFAULT_VIEW; - camu_view_calculate(view, scr->width, scr->height); + if (view) { + view->mode = CAMU_DEFAULT_VIEW; + camu_view_calculate(view, scr->width, scr->height); + } } break; } case STELA_TOUCH_MOVED: { - if (index == 0) { - if (SCREEN_IS_ZOOMING(scr)) { - f64 dy = (y - scr->last_pointer_y) / 50.0; - scr->last_pointer_x = x; - scr->last_pointer_y = y; - return scroll_callback(scr, dy); - } else if (SCREEN_IS_DRAGGING(scr)) { - return pointer_pos_callback(scr, x, y); + // @TODO: Track current index 1 and 2. On release search for next non-0 index in mask. + if (SCREEN_IS_FROZEN(scr)) { + return false; + } else if (SCREEN_IS_ZOOMING(scr)) { + if (index == 0) { + scr->last_touch_x[0] = x; + scr->last_touch_y[0] = y; + } else if (index == 1) { + scr->last_touch_x[1] = x; + scr->last_touch_y[1] = y; } + f64 lx = scr->last_touch_x[0], lx2 = scr->last_touch_x[1]; + f64 ly = scr->last_touch_y[0], ly2 = scr->last_touch_y[1]; + f64 dist = distance(lx, lx2, ly, ly2); + f64 mx = x = (lx + lx2) / 2.0, my = (ly + ly2) / 2.0; + bool refresh = false; + struct camu_view *view = get_view_from_mouse_pos(scr); + if (view) { + refresh |= camu_view_pan_simple(view, scr->width, scr->height, mx - scr->midpoint_x, my - scr->midpoint_y); + } + scr->midpoint_x = mx; + scr->midpoint_y = my; + scr->last_pointer_x = scr->midpoint_x; + scr->last_pointer_y = scr->midpoint_y; + refresh |= scroll_callback(scr, (dist - scr->last_distance) / 120.0); + scr->last_distance = dist; + return refresh; + } else if (SCREEN_IS_DRAGGING(scr)) { + if (index == 0) { + scr->last_pointer_x = scr->last_touch_x[0]; + scr->last_pointer_y = scr->last_touch_y[0]; + } else if (index == 1) { + scr->last_pointer_x = scr->last_touch_x[1]; + scr->last_pointer_y = scr->last_touch_y[1]; + } + bool refresh = pointer_pos_callback(scr, x, y); + if (index == 0) { + scr->last_touch_x[0] = x; + scr->last_touch_y[0] = y; + } else if (index == 1) { + scr->last_touch_x[1] = x; + scr->last_touch_y[1] = y; + } + return refresh; } break; } case STELA_TOUCH_ENDED: { - if (index == 0) { - scr->flags &= ~(CAMU_SCREEN_DRAGGING | CAMU_SCREEN_ZOOMING); + if (SCREEN_IS_FROZEN(scr)) { + scr->flags &= ~CAMU_SCREEN_FROZEN; + seek_to_percent_at_pointer(scr, x); + } else if (SCREEN_IS_ZOOMING(scr)) { + scr->flags &= ~CAMU_SCREEN_ZOOMING; + } else if (SCREEN_IS_DRAGGING(scr)) { + scr->flags &= ~CAMU_SCREEN_DRAGGING; if (SCREEN_LAST_CLICK_WITHIN(200000)) { - s32 n = (x >= scr->width / 2.0) ? 1 : -1; - scr->callback(scr->userdata, CAMU_SCREEN_SKIP, &n); - } - } else if (index == 1) { - // We want to ignore this if there was a 3rd touch. - if (SCREEN_IS_ZOOMING(scr)) { - scr->flags &= ~CAMU_SCREEN_ZOOMING; - scr->flags |= CAMU_SCREEN_DRAGGING; + f64 third = scr->width / 3.0; + if (x >= third && x <= third * 2.0) { + scr->callback(scr->userdata, CAMU_SCREEN_TOGGLE_PAUSE, NULL); + } else { + s32 n = (x >= third) ? 1 : -1; + scr->callback(scr->userdata, CAMU_SCREEN_SKIP, &n); + } } } break; } } return false; +#undef distance } static bool mouse_button_callback(void *userdata, u8 state, u8 button) @@ -373,6 +427,13 @@ static bool key_callback(void *userdata, u8 state, u16 button) } break; } + case STELA_KEY_T: { + struct camu_view *view = get_view_from_mouse_pos(scr); + if (view) { + scr->window->resize(scr->window, ceil(view->width * view->zoom), ceil(view->height * view->zoom)); + } + break; + } case STELA_KEY_EQUAL: { struct camu_view *view = get_view_from_mouse_pos(scr); if (view) { @@ -455,7 +516,7 @@ static bool key_callback(void *userdata, u8 state, u16 button) return false; } -static void key_immediate_callback(void *userdata, u8 state, u16 button) +static bool key_immediate_callback(void *userdata, u8 state, u16 button) { struct camu_screen *scr = (struct camu_screen *)userdata; switch (state) { @@ -468,17 +529,32 @@ static void key_immediate_callback(void *userdata, u8 state, u16 button) scr->window->set_cursor(scr->window, scr->fullscreen ? STELA_CURSOR_HIDDEN : STELA_CURSOR_NORMAL); } //#ifdef STELA_API_DX11 -#if 0 // Stela's window_win32 fullscreen mode (borderless) is probably more desirable here. +#if 0 // Stela's window_win32 fullscreen mode (borderless) is likely more desirable here. scr->renderer->set(scr->renderer, CAMU_RENDERER_FULLSCREEN, scr->fullscreen); #else scr->window->toggle_fullscreen(scr->window); #endif + return true; + case STELA_KEY_LEFT: + if (scr->touch_mode) { + s32 n = -1; + scr->callback(scr->userdata, CAMU_SCREEN_SKIP, &n); + return true; + } + break; + case STELA_KEY_RIGHT: + if (scr->touch_mode) { + s32 n = 1; + scr->callback(scr->userdata, CAMU_SCREEN_SKIP, &n); + return true; + } break; } break; default: break; } + return false; } bool camu_screen_init(struct camu_screen *scr) @@ -509,6 +585,7 @@ bool camu_screen_init(struct camu_screen *scr) #ifdef CAMU_SCREEN_DRAG_SEEK scr->last_seek_ts = 0; #endif + scr->touch_mode = false; scr->vr_emulation = CAMU_VR_DISABLED; scr->vr_left_eye = false; scr->vr_calibrate_x = 0.5; diff --git a/src/screen/screen.h b/src/screen/screen.h index b92cfe5..0d40a02 100644 --- a/src/screen/screen.h +++ b/src/screen/screen.h @@ -27,7 +27,8 @@ enum { CAMU_SCREEN_MOUSE_MOD = 1 << 3, CAMU_SCREEN_ZOOMING = 1 << 4, CAMU_SCREEN_ZOOM_PAN_SIMPLE = 1 << 5, - CAMU_SCREEN_ZOOM_FOV = 1 << 6 + CAMU_SCREEN_ZOOM_FOV = 1 << 6, + CAMU_SCREEN_FROZEN = 1 << 7 }; enum { @@ -76,9 +77,15 @@ struct camu_screen { u64 last_click_ts; f64 last_pointer_y; f64 last_pointer_x; + f64 last_touch_x[2]; + f64 last_touch_y[2]; + f64 last_distance; + f64 midpoint_x; + f64 midpoint_y; #ifdef CAMU_SCREEN_DRAG_SEEK u64 last_seek_ts; #endif + bool touch_mode; u8 vr_emulation; bool vr_left_eye; f64 vr_calibrate_x; |