Server-side fixes from DIRECT_MODE testing

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

6 files changed, 77 insertions, 53 deletions

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 <al/random.h>
-#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 <al/random.h>
+#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;
         }