From 8f208c26b6fa1a9f3372679c047cab559c06e26b Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Mon, 14 Sep 2026 08:57:42 -0400 Subject: Server-side fixes from DIRECT_MODE testing Signed-off-by: Andrew Opalach --- src/buffer/audio.c | 32 +++++++++++++++++++------------- src/buffer/clock.c | 37 ++++++++++++++++++++++--------------- src/buffer/clock.h | 23 ++++++++++++++++------- src/buffer/common.h | 11 ----------- src/buffer/common_internal.h | 11 +++++++++++ src/buffer/video.c | 16 +++++++++------- 6 files changed, 77 insertions(+), 53 deletions(-) (limited to 'src/buffer') diff --git a/src/buffer/audio.c b/src/buffer/audio.c index a815c66..1b866a9 100644 --- a/src/buffer/audio.c +++ b/src/buffer/audio.c @@ -12,7 +12,7 @@ #define BUFFER_SIZE 7.0 #define BUFFER_MARK_MIN 3.25 // Must be a most half of the buffer size. -#define BUFFER_MARK_BUFFERED 0.35 +#define BUFFER_MARK_BUFFERED 0.5 #ifdef CAMU_AUDIO_BUFFER_FADE #define FADE_STEP(fmt, down) ((down ? -4.75f : 2.25f) / (fmt)->sample_rate) @@ -88,7 +88,8 @@ bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_code return false; #endif const char *req_format_name = camu_audio_format_name(req->format); - log_info("Resampling to: %s (%dch) %dHz.", req_format_name, req->channel_count, req->sample_rate); + log_info("Resampling to: %s (%dch) %dHz.", req_format_name, + req->channel_count, req->sample_rate); } buf->size = (ptrdiff_t)camu_audio_format_sec_to_bytes(req, BUFFER_SIZE); @@ -155,11 +156,11 @@ static bool push_internal(struct camu_audio_buffer *buf, f64 pts, u8 **data, s32 return true; } - // The maximum space is buf->size - 1. - ptrdiff_t space = al_ring_buffer_space(&buf->rb); - if (!buf->buffered && (buf->size - 1) - space >= buf->mark.buffered) { + // The maximum space in a ring buffer is size - 1. + ptrdiff_t occupied, space = al_ring_buffer_space(&buf->rb); + if (!buf->buffered && (occupied = (buf->size - 1) - space) >= buf->mark.buffered) { buf->buffered = true; - log_debug("Buffered (mark: %.1fKB).", buf->mark.buffered / 1024.0); + log_debug("Buffered (%.1fKB).", occupied / 1024.0); buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); } @@ -241,14 +242,19 @@ void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_fra void camu_audio_buffer_flush(struct camu_audio_buffer *buf, bool error) { log_debug("Flush requested."); - u8 flow = error ? FLUSHED_ERROR : FLUSHED; - atomic_store(u32)(&buf->flow, flow, AL_ATOMIC_RELAXED); + u8 flow = atomic_load(u32)(&buf->flow, AL_ATOMIC_ACQUIRE); + error |= flow == FLUSHED_ERROR; + flow = error ? FLUSHED_ERROR : FLUSHED; + atomic_store(u32)(&buf->flow, flow, AL_ATOMIC_RELEASE); if (!push_internal(buf, 0.0, NULL, 0)) { log_debug("Buffer filled by flush."); } if (!buf->buffered) { buf->buffered = true; - log_debug("Buffered (flush)."); +#ifdef AL_LOG_ENABLE_DEBUG + ptrdiff_t occupied = al_ring_buffer_occupied(&buf->rb); + log_debug("Buffered (%.1fKB).", occupied / 1024.0); +#endif buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); } } @@ -289,11 +295,11 @@ ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdif f64 base_pts = atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE); 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); + u8 status = 0; + f64 pts = camu_clock_get_pts(buf->clock, buf->latency, allow_set, &status); if (pts == CAMU_PTS_SIGNAL_PAUSE) { return 0; - } else if (pts == CAMU_PTS_PAUSED || armed_for_pause) { + } else if (CAMU_PTS_CONSIDER_PAUSED(pts) || (status & CAMU_CLOCK_EXTERNAL_PAUSE)) { #ifdef CAMU_AUDIO_BUFFER_FADE if (buf->pause == PAUSE_SYNC) { buf->pause = PAUSE_FADE_COMPLETE; @@ -353,7 +359,7 @@ 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_info("Forcing resync."); + log_debug("Forcing resync."); buf->pause = PAUSE_SYNC; atomic_sub(u32)(&buf->unpause, 1, AL_ATOMIC_RELEASE); } diff --git a/src/buffer/clock.c b/src/buffer/clock.c index 19f721c..1adb945 100644 --- a/src/buffer/clock.c +++ b/src/buffer/clock.c @@ -43,6 +43,7 @@ void camu_clock_set(struct camu_clock *clock, f64 base) clock->offset = 0.0; atomic_store(f64)(&clock->tick, SET, AL_ATOMIC_RELAXED); atomic_store(f64)(&clock->pause, PAUSED, AL_ATOMIC_RELAXED); + atomic_store(bool)(&clock->pause_for_swap, false, AL_ATOMIC_RELAXED); 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); @@ -106,7 +107,12 @@ bool camu_clock_pause(struct camu_clock *clock, u64 target) } } -void camu_clock_external_pause(struct camu_clock *clock) +void camu_clock_set_pause_for_swap(struct camu_clock *clock) +{ + atomic_store(bool)(&clock->pause_for_swap, true, AL_ATOMIC_RELAXED); +} + +void camu_clock_set_external_pause(struct camu_clock *clock) { atomic_store(bool)(&clock->external_pause, true, AL_ATOMIC_RELAXED); } @@ -116,8 +122,6 @@ void camu_clock_resume(struct camu_clock *clock, u64 target) { 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); @@ -142,15 +146,10 @@ void camu_clock_resume(struct camu_clock *clock, u64 target) clock->paused_at = NOT_PAUSED; - atomic_store(f64)(&clock->pause, RUNNING, AL_ATOMIC_RELEASE); -} + atomic_store(bool)(&clock->pause_for_swap, false, AL_ATOMIC_RELAXED); + atomic_store(bool)(&clock->external_pause, false, AL_ATOMIC_RELAXED); -// @TODO: This only works for local sink. -bool camu_clock_is_user_paused(struct camu_clock *clock) -{ - f64 tick = atomic_load(f64)(&clock->tick, AL_ATOMIC_RELAXED); - f64 pause = atomic_load(f64)(&clock->pause, AL_ATOMIC_RELAXED); - return tick != SET && pause <= 0.0; + atomic_store(f64)(&clock->pause, RUNNING, AL_ATOMIC_RELEASE); } f64 camu_clock_get_base_pts(struct camu_clock *clock) @@ -158,7 +157,7 @@ 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, bool *armed_for_pause) +f64 camu_clock_get_pts(struct camu_clock *clock, f64 latency, bool allow_set, u8 *status) { // Treat pause = 0.0 or -0.0 as paused. f64 pause = atomic_load(f64)(&clock->pause, AL_ATOMIC_RELAXED); @@ -171,15 +170,23 @@ f64 camu_clock_get_pts(struct camu_clock *clock, f64 latency, bool allow_set, bo tick = atomic_compare_and_swap(f64)(&clock->tick, SET, current); if (tick == SET) tick = current; } else { - return CAMU_PTS_PAUSED; + return CAMU_PTS_UNSET; } } + bool allow_pause = !status || *status != CAMU_CLOCK_NO_SIGNAL_PAUSE; + if (status) { + *status = 0; + bool pause_for_swap = atomic_load(bool)(&clock->pause_for_swap, AL_ATOMIC_RELAXED); + bool external_pause = atomic_load(bool)(&clock->external_pause, AL_ATOMIC_RELAXED); + if (pause_for_swap) *status |= CAMU_CLOCK_PAUSE_FOR_SWAP; + if (external_pause) *status |= CAMU_CLOCK_EXTERNAL_PAUSE; + } + f64 pts = clock->base + (current - tick); bool signal_pause = false; - *armed_for_pause = pause != RUNNING || atomic_load(bool)(&clock->external_pause, AL_ATOMIC_RELAXED); - if (pause > 0.0 && current > pause) { + if (allow_pause && 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 e1a2d9e..3169c61 100644 --- a/src/buffer/clock.h +++ b/src/buffer/clock.h @@ -23,21 +23,30 @@ // Next/Prev: // - User input -> swap -#define CAMU_PTS_PAUSED ((f64)0xffffffffffffffff) -#define CAMU_PTS_SIGNAL_PAUSE ((f64)0x7fffffffffffffff) -#define CAMU_PTS_CONSIDER_PAUSED(pts) (pts == CAMU_PTS_PAUSED || pts == CAMU_PTS_SIGNAL_PAUSE) +// Values that PTS can never normally be. +#define CAMU_PTS_PAUSED 0x1.fffffffffffffp+1023 +#define CAMU_PTS_SIGNAL_PAUSE 0x1.fffffffffffffp+1022 +#define CAMU_PTS_UNSET 0x1.fffffffffffffp+1021 +#define CAMU_PTS_CONSIDER_PAUSED(pts) (pts == CAMU_PTS_PAUSED || pts == CAMU_PTS_SIGNAL_PAUSE || pts == CAMU_PTS_UNSET) enum { CAMU_CLOCK_PAUSED = 0 }; +enum { + CAMU_CLOCK_PAUSE_FOR_SWAP = 1, + CAMU_CLOCK_EXTERNAL_PAUSE = 1 << 1, + CAMU_CLOCK_NO_SIGNAL_PAUSE = 1 << 7 +}; + struct camu_clock { f64 base; f64 offset; atomic(f64) tick; atomic(f64) pause; - atomic(bool) external_pause; f64 paused_at; + atomic(bool) pause_for_swap; + atomic(bool) external_pause; atomic(f64) last_pts; void (*callback)(void *, u8); void *userdata; @@ -50,10 +59,10 @@ void camu_clock_offset(struct camu_clock *clock, f64 offset); void camu_clock_seek(struct camu_clock *clock, f64 base, u64 target); bool camu_clock_pause(struct camu_clock *clock, u64 target); -void camu_clock_external_pause(struct camu_clock *clock); +void camu_clock_set_pause_for_swap(struct camu_clock *clock); +void camu_clock_set_external_pause(struct camu_clock *clock); void camu_clock_resume(struct camu_clock *clock, u64 target); -bool camu_clock_is_user_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, bool *armed_for_pause); +f64 camu_clock_get_pts(struct camu_clock *clock, f64 latency, bool allow_set, u8 *status); f64 camu_clock_get_last_pts(struct camu_clock *clock); diff --git a/src/buffer/common.h b/src/buffer/common.h index aaf53ea..24964ba 100644 --- a/src/buffer/common.h +++ b/src/buffer/common.h @@ -1,16 +1,5 @@ #pragma once -//#define CAMU_BUFFER_SPORADIC_ERRORS -#ifdef CAMU_BUFFER_SPORADIC_ERRORS -#include -#define ROLL_FOR_BUFFER_ERROR(buf) do { \ - if (al_random_int(0, 254) == 72) { \ - log_warn("Random buffer error proc."); \ - atomic_store(u32)(&(buf)->flow, FLUSHED_ERROR, AL_ATOMIC_RELAXED); \ - } \ -} while (0) -#endif - enum { CAMU_BUFFER_BUFFERED = 0, CAMU_BUFFER_CORK, diff --git a/src/buffer/common_internal.h b/src/buffer/common_internal.h index a9e6ece..6685a3e 100644 --- a/src/buffer/common_internal.h +++ b/src/buffer/common_internal.h @@ -17,6 +17,17 @@ enum { ERRORED }; +//#define CAMU_BUFFER_SPORADIC_ERRORS +#ifdef CAMU_BUFFER_SPORADIC_ERRORS +#include +#define ROLL_FOR_BUFFER_ERROR(buf) do { \ + if (al_random_int(0, 254) == 72) { \ + log_warn("Random buffer error proc."); \ + atomic_store(u32)(&(buf)->flow, FLUSHED_ERROR, AL_ATOMIC_RELAXED); \ + } \ +} while (0) +#endif + static inline bool frame_is_late(struct camu_clock *clock, f64 base, f64 pts, f64 duration) { return pts + duration < ((base == -1.0) ? camu_clock_get_base_pts(clock) : base); diff --git a/src/buffer/video.c b/src/buffer/video.c index 19d7aa7..9125dab 100644 --- a/src/buffer/video.c +++ b/src/buffer/video.c @@ -178,7 +178,8 @@ void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_fra if (flow != FLOWING) { // A static buffer will be FLUSHED after any push(). if (buf->is_static) { - log_error("Unexpected duplicate frame received."); + log_error("Expected a single frame, but received another."); + atomic_store(u32)(&buf->flow, FLUSHED_ERROR, AL_ATOMIC_RELEASE); } camu_codec_frame_discard(frame); return; @@ -214,7 +215,7 @@ void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_fra } buf->buffered = true; buf->buffered_with_one_frame = count == 1; - log_debug("Buffered (mark: %.2fs).", count * buf->avg_frame_duration); + log_debug("Buffered (%.2fs).", count * buf->avg_frame_duration); buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); } else if (count >= BUFFER_MARK_HIGH) { buf->callback(buf->userdata, CAMU_BUFFER_CORK); @@ -234,14 +235,16 @@ void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struct camu_ void camu_video_buffer_flush(struct camu_video_buffer *buf, bool error) { log_debug("Flush requested."); - u8 flow = error ? FLUSHED_ERROR : FLUSHED; - atomic_store(u32)(&buf->flow, flow, AL_ATOMIC_RELAXED); + u8 flow = atomic_load(u32)(&buf->flow, AL_ATOMIC_ACQUIRE); + error |= flow == FLUSHED_ERROR; + flow = error ? FLUSHED_ERROR : FLUSHED; + atomic_store(u32)(&buf->flow, flow, AL_ATOMIC_RELEASE); u32 count; buf->queue->flush(buf->queue, &count); if (!buf->buffered) { buf->buffered = true; buf->buffered_with_one_frame = count == 1; - log_debug("Buffered (mark: %.2fs).", count * buf->avg_frame_duration); + log_debug("Buffered (%.2fs).", count * buf->avg_frame_duration); buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); } } @@ -266,8 +269,7 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weig f64 base_pts = atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE); if (!buf->is_static) { bool allow_set = !buf->weighted_first_read; - bool armed_for_pause = false; - f64 pts = camu_clock_get_pts(buf->clock, buf->latency, allow_set, &armed_for_pause); + f64 pts = camu_clock_get_pts(buf->clock, buf->latency, allow_set, NULL); if (!CAMU_PTS_CONSIDER_PAUSED(pts) && pts > base_pts) { base_pts = pts; } -- cgit v1.2.3-101-g0448