Audio buffer, mixer and sink improvements

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

6 files changed, 107 insertions, 77 deletions

diff --git a/src/buffer/audio.c b/src/buffer/audio.c
index d438d25..707219f 100644
--- a/src/buffer/audio.c
+++ b/src/buffer/audio.c
@@ -20,7 +20,7 @@
 enum {
     PAUSE_PAUSED = 0,
 #ifdef CAMU_AUDIO_BUFFER_FADE
-    PAUSE_FADING,
+    PAUSE_FADING_OUT,
 #endif
     PAUSE_PLAYING
 };
@@ -33,7 +33,7 @@ static void reset_buffer_state(struct camu_audio_buffer *buf)
     al_atomic_store(s32)(&buf->volume.set, 0, AL_ATOMIC_RELAXED);
 #ifdef CAMU_AUDIO_BUFFER_FADE
     buf->fade.offset = 0;
-    buf->fade.volume = 1.f;
+    buf->fade.volume = -1.f;
 #endif
     buf->buffered = false;
     al_atomic_store(u8)(&buf->flow, FLOWING, AL_ATOMIC_RELAXED);
@@ -43,9 +43,10 @@ static void reset_buffer_state(struct camu_audio_buffer *buf)
 bool camu_audio_buffer_init(struct camu_audio_buffer *buf, struct camu_clock *clock)
 {
     buf->clock = clock;
-    reset_buffer_state(buf);
-    buf->ignore_desync = false;
     buf->latency = 0.0;
+    buf->ignore_desync = false;
+    al_atomic_store(bool)(&buf->no_video, false, AL_ATOMIC_RELAXED);
+    reset_buffer_state(buf);
 #ifdef CAMU_MIXER_THREADED
     al_atomic_store(u8)(&buf->ref, 0, AL_ATOMIC_RELAXED);
 #endif
@@ -82,7 +83,7 @@ bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_code
 
     buf->mark.min = camu_audio_format_sec_to_bytes(&buf->fmt.req, BUFFER_MARK_MIN);
     buf->mark.buffered = camu_audio_format_sec_to_bytes(&buf->fmt.req, BUFFER_MARK_BUFFERED);
-    camu_peak_buffer_init(&buf->peak, 1024 * 16);
+    camu_peak_buffer_init(&buf->peak, KB(16));
 
     buf->stream = stream;
 
@@ -100,6 +101,16 @@ void camu_audio_buffer_set_latency(struct camu_audio_buffer *buf, f64 latency)
     buf->latency = latency;
 }
 
+void camu_audio_buffer_set_ignore_desync(struct camu_audio_buffer *buf, bool ignore_desync)
+{
+    buf->ignore_desync = ignore_desync;
+}
+
+void camu_audio_buffer_set_no_video(struct camu_audio_buffer *buf, bool no_video)
+{
+    al_atomic_store(bool)(&buf->no_video, no_video, AL_ATOMIC_RELAXED);
+}
+
 #ifdef _DEBUG_
 #define BUFFERED_SECONDS_DEBUG(buf) \
     camu_audio_format_bytes_to_sec(&buf->fmt.req, al_ring_buffer_occupied(&buf->rb))
@@ -113,26 +124,37 @@ static inline bool frame_is_late(struct camu_clock *clock, f64 base, f64 pts, f6
     return pts + duration < base;
 }
 
-// return value of false indicates we pushed to the peak buffer.
-static bool push_internal(struct camu_audio_buffer *buf, f64 pts, u8 **data, s32 sample_count, bool flush)
+// A return value of false signals that we pushed to the peak buffer.
+static bool push_internal(struct camu_audio_buffer *buf, f64 pts, u8 **data, s32 sample_count)
 {
-    if (!flush) {
+    f64 base_pts = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
+
+    if (sample_count > 0) {
+        al_assert(data);
         f64 duration = camu_audio_format_samples_to_sec(&buf->fmt.in, sample_count);
-        f64 base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
-        if (frame_is_late(buf->clock, base, pts, duration)) return true;
+        if (frame_is_late(buf->clock, base_pts, pts, duration)) {
+            return true;
+        }
         if (buf->fmt.resampler_needed) {
             sample_count = buf->resampler->convert(buf->resampler, (const u8 **)data, sample_count);
             data = buf->resampler->get_data(buf->resampler);
         }
-        if (sample_count == 0) return true;
-        if (base == -1.0) al_atomic_store(f64)(&buf->pts, pts, AL_ATOMIC_RELEASE);
     } else {
-        al_assert(!data && sample_count == 0);
+        al_assert(!data);
         if (buf->fmt.resampler_needed) {
             sample_count = buf->resampler->flush(buf->resampler);
             data = buf->resampler->get_data(buf->resampler);
         }
-        if (sample_count == 0) return true;
+    }
+
+    if (base_pts == -1.0) {
+        // We want this push() to set the pts even if sample_count = 0.
+        al_atomic_store(f64)(&buf->pts, pts, AL_ATOMIC_RELEASE);
+    }
+
+    if (sample_count == 0) {
+        // The resampler is holding all the data.
+        return true;
     }
 
     size_t space = al_ring_buffer_space(&buf->rb);
@@ -177,7 +199,7 @@ static void push_av_frame_internal(struct camu_audio_buffer *buf, AVFrame *frame
     s32 sample_count = frame->nb_samples;
     AVStream *stream = buf->stream->av.stream;
     f64 pts = frame->best_effort_timestamp * av_q2d(stream->time_base);
-    push_internal(buf, pts, frame->data, sample_count, false);
+    push_internal(buf, pts, frame->data, sample_count);
     av_frame_free(&frame);
 }
 #endif
@@ -186,13 +208,13 @@ static void push_av_frame_internal(struct camu_audio_buffer *buf, AVFrame *frame
 void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_frame *frame)
 {
     al_assert(al_atomic_load(u8)(&buf->flow, AL_ATOMIC_RELAXED) == FLOWING);
+
     switch (frame->mode) {
     case CAMU_NORMAL: {
         s32 sample_count = frame->audio.sample_count;
-        f64 pts = frame->pts;
         u8 *planes[CAMU_PLANAR_DATA_POINTERS] = { 0 };
         planes[0] = frame->data;
-        push_internal(buf, pts, planes, sample_count, false);
+        push_internal(buf, frame->pts, planes, sample_count);
         break;
     }
 #ifdef CAMU_HAVE_FFMPEG
@@ -202,6 +224,7 @@ void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_fra
     }
 #endif
     }
+
     al_free(frame);
 }
 
@@ -209,14 +232,17 @@ void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_fra
 void camu_audio_buffer_flush(struct camu_audio_buffer *buf)
 {
     al_log_debug("audio_buffer", "Flush requested.");
-    if (!push_internal(buf, 0.0, NULL, 0, true)) {
+
+    if (!push_internal(buf, 0.0, NULL, 0)) {
         al_log_debug("audio_buffer", "Buffer filled by flush.");
     }
+
     if (!buf->buffered) {
         al_log_debug("audio_buffer", "Buffered (mark: %.2fs).", BUFFERED_SECONDS_DEBUG(buf));
         buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED);
         buf->buffered = true;
     }
+
     al_atomic_store(u8)(&buf->flow, FLUSHED, AL_ATOMIC_RELAXED);
 }
 
@@ -236,14 +262,6 @@ void camu_audio_buffer_unpause(struct camu_audio_buffer *buf)
     al_atomic_add(s32)(&buf->unpause, 1, AL_ATOMIC_RELAXED);
 }
 
-static void increment_pts(struct camu_audio_buffer *buf, f64 amount, f64 *base)
-{
-    *base += amount;
-    al_atomic_store(f64)(&buf->pts, *base, AL_ATOMIC_RELEASE);
-}
-
-#define NOT_PAUSED(pause) (pause != PAUSE_PAUSED)
-
 // PAUSE_PAUSED signifies that the last read was silence. Meaning we can skip
 // around in the buffer without worrying about pops.
 size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t req)
@@ -258,25 +276,31 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
         al_atomic_sub(s32)(&buf->volume.set, 1, AL_ATOMIC_RELEASE);
     }
 
-    if (camu_clock_is_paused(buf->clock)) {
+    bool set_clock = al_atomic_load(bool)(&buf->no_video, AL_ATOMIC_RELAXED);
+    f64 pts = camu_clock_get_pts(buf->clock, buf->latency, set_clock);
+    if (pts == -1.0) {
 #ifdef CAMU_AUDIO_BUFFER_FADE
         if (buf->pause == PAUSE_PLAYING) {
-            buf->pause = PAUSE_FADING;
-            buf->fade.offset = 0;
-        } else if (buf->fade.volume == 0.f || buf->pause == PAUSE_PAUSED) {
+            buf->pause = PAUSE_FADING_OUT;
             buf->fade.offset = 0;
+        } else if (buf->pause == PAUSE_PAUSED || buf->fade.volume == 0.f) {
             al_memset(data, 0, req);
-            if (NOT_PAUSED(buf->pause)) {
+            if (buf->pause != PAUSE_PAUSED) {
+                // Fade out finished.
                 buf->pause = PAUSE_PAUSED;
                 buf->callback(buf->userdata, CAMU_BUFFER_PAUSED);
             }
+            buf->fade.offset = 0;
             return req;
         }
-    } else if (buf->pause == PAUSE_FADING || buf->pause == PAUSE_PAUSED) {
-        if (buf->pause == PAUSE_FADING) buf->pause = PAUSE_PLAYING;
-#else
+    } else if (buf->pause == PAUSE_FADING_OUT || buf->pause == PAUSE_PAUSED) {
+        if (buf->pause == PAUSE_FADING_OUT) {
+            // If unpaused during a fade out, resume immediately to not break the stream.
+            buf->pause = PAUSE_PLAYING;
+        }
+#else // No fade and clock paused.
         al_memset(data, 0, req);
-        if (NOT_PAUSED(buf->pause)) {
+        if (buf->pause != PAUSE_PAUSED) {
             buf->pause = PAUSE_PAUSED;
             buf->callback(buf->userdata, CAMU_BUFFER_PAUSED);
         }
@@ -285,36 +309,36 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
     }
 
     if (al_atomic_load(s32)(&buf->unpause, AL_ATOMIC_ACQUIRE) > 0) {
-        // If pause != PLAYING here, this will cause unexpected behavior.
+        // If pause != PLAYING here this will likely cause unexpected behavior.
         buf->pause = PAUSE_PAUSED;
         al_atomic_sub(s32)(&buf->unpause, 1, AL_ATOMIC_RELEASE);
     }
 
-    f64 base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
-    f64 pts = camu_clock_get_pts(buf->clock, buf->latency, false);
+    struct camu_audio_format *fmt = &buf->fmt.req;
     size_t ret, signal = req;
     size_t have = al_ring_buffer_occupied(&buf->rb);
 #ifdef CAMU_AUDIO_BUFFER_FADE
     // Cut off fade if it's reaching too far.
     if (have < buf->fade.offset) have = 0;
     else have -= buf->fade.offset;
-    if (buf->pause != PAUSE_FADING && buf->pause != PAUSE_PLAYING) {
 #endif
-        pts -= base;
-        // Attempt syncing to the clock.
-        // For this to work the mixer must report a reasonably accurate value for latency.
-        if (UNLIKELY(!buf->ignore_desync && buf->pause == PAUSE_PAUSED)) {
+
+    f64 base_pts = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
+    // Unpause and possibly attempt syncing to the clock.
+    if (UNLIKELY(buf->pause == PAUSE_PAUSED)) {
+        if (!buf->ignore_desync) {
+            pts -= base_pts;
             if (pts > 0.0) { // Skip.
-                ret = camu_audio_format_sec_to_bytes(&buf->fmt.req, pts);
+                ret = camu_audio_format_sec_to_bytes(fmt, pts);
                 ret = MIN(ret, have);
                 al_log_info("audio_buffer", "Skipping %fs of audio (%zu bytes).", pts, ret);
                 ret = al_ring_buffer_discard(&buf->rb, ret);
                 have -= ret;
-                increment_pts(buf, camu_audio_format_bytes_to_sec(&buf->fmt.req, ret), &base);
+                base_pts += camu_audio_format_bytes_to_sec(fmt, ret);
                 // Could go on to underrun.
             } else if (pts < 0.0) { // Delay.
                 pts = -pts;
-                ret = camu_audio_format_sec_to_bytes(&buf->fmt.req, pts);
+                ret = camu_audio_format_sec_to_bytes(fmt, pts);
                 ret = MIN(ret, req);
                 al_log_info("audio_buffer", "Delaying audio by %fs.", pts);
                 al_memset(data, 0, ret);
@@ -324,13 +348,11 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
                 if (req == 0) return signal;
             }
         }
-        if (buf->pause != PAUSE_PLAYING) buf->pause = PAUSE_PLAYING;
-#ifdef CAMU_AUDIO_BUFFER_FADE
+        buf->pause = PAUSE_PLAYING;
     }
-#endif
 
     u8 flow = al_atomic_load(u8)(&buf->flow, AL_ATOMIC_ACQUIRE);
-    if (have < req) { // We don't have enough data to fulfill our request.
+    if (UNLIKELY(have < req)) { // We don't have enough data to fulfill our request.
         if (flow == FLUSHED) { // Stream is flushed.
             // Check peak buffer for any remaining data.
             ret = camu_peak_buffer_get_size(&buf->peak);
@@ -340,17 +362,16 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
                 al_ring_buffer_write(&buf->rb, camu_peak_buffer_flush(&buf->peak), ret);
                 have += ret;
             }
-            // Only signal EOF if what, if any, data from the peak buffer
-            // still isn't enough for this request.
+            // Only signal EOF if the data from the peak buffer, if any, wasn't
+            // enough for this request.
             if (have < req) {
                 signal = have;
                 al_log_debug("audio_buffer", "Flushed (signal: %zu).", signal);
                 buf->callback(buf->userdata, CAMU_BUFFER_EOF);
-                flow = SIGNALED;
                 al_atomic_store(u8)(&buf->flow, SIGNALED, AL_ATOMIC_RELEASE);
             }
         } else {
-            // Silence remainder of the request.
+            // Silence the remainder of the request.
             al_memset(data + have, 0, req - have);
             if (flow == FLOWING) {
                 // We hit an underrun because data wasn't coming in fast enough.
@@ -368,11 +389,11 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
         req = MIN(req, have);
     }
 
-    if (req > 0) {
+    if (LIKELY(req > 0)) {
 #ifdef CAMU_AUDIO_BUFFER_FADE
-        if (buf->pause == PAUSE_FADING) {
-            // Peek into the ring buffer. This won't consume the data, so when
-            // we resume it will fade in from the same position the fade out started.
+        if (buf->pause == PAUSE_FADING_OUT) {
+            // Peeking into the ring buffer won't consume data. So, when resumed we
+            // will fade in from the same position that the fade out started.
             ret = al_ring_buffer_peek(&buf->rb, data, buf->fade.offset, req);
             buf->fade.offset += ret;
         } else {
@@ -380,32 +401,37 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
             // Read as much as we determined we can.
             ret = al_ring_buffer_read(&buf->rb, data, req);
             al_assert(ret == req);
-            increment_pts(buf, camu_audio_format_bytes_to_sec(&buf->fmt.req, ret), &base);
+            base_pts += camu_audio_format_bytes_to_sec(fmt, ret);
 #ifdef CAMU_AUDIO_BUFFER_FADE
         }
+
         f32 step = 0.f;
-        if (buf->pause == PAUSE_FADING || buf->fade.volume > buf->volume.user) {
-            step = -FADE_STEP(&buf->fmt.req);
+        if (buf->pause == PAUSE_FADING_OUT || buf->fade.volume > buf->volume.user) {
+            step = -FADE_STEP(fmt);
         } else if (buf->fade.volume < buf->volume.user) {
-            step = FADE_STEP(&buf->fmt.req);
+            step = FADE_STEP(fmt);
         }
+
         if (step != 0.f || buf->fade.volume != 1.f) {
             if (buf->volume.user > 1.f) {
-                // Only adjust step if it's an increase.
+                // Only adjust the step if it's an increase to avoid
+                // drawn-out fades if the volume is low.
                 step *= buf->volume.user;
             }
-            buf->fade.volume = apply_volume(data, req, &buf->fmt.req, buf->fade.volume, buf->volume.user, step);
+            buf->fade.volume = apply_volume(data, req, fmt, buf->fade.volume, buf->volume.user, step);
         }
 #else
         if (buf->volume.user != 1.f) {
-            apply_volume(data, req, &buf->fmt.req, buf->volume.user, 0.f, 0.f);
+            apply_volume(data, req, fmt, buf->volume.user, 0.f, 0.f);
         }
 #endif
     }
 
+    al_atomic_store(f64)(&buf->pts, base_pts, AL_ATOMIC_RELEASE);
+
     // Check if we should request to uncork.
 #ifdef CAMU_AUDIO_BUFFER_FADE
-    if (flow == FLOWING && buf->pause != PAUSE_FADING) {
+    if (flow == FLOWING && buf->pause != PAUSE_FADING_OUT) {
 #else
     if (flow == FLOWING) {
 #endif
diff --git a/src/buffer/audio.h b/src/buffer/audio.h
index 6fc3cca..ca2d405 100644
--- a/src/buffer/audio.h
+++ b/src/buffer/audio.h
@@ -14,12 +14,14 @@
 struct camu_audio_buffer {
     struct camu_codec_stream *stream;
 
+    struct camu_clock *clock;
     atomic(f64) pts;
+    f64 latency;
+    bool ignore_desync;
+    atomic(bool) no_video;
+
     u8 pause;
     atomic(s32) unpause;
-    struct camu_clock *clock;
-    bool ignore_desync;
-    f64 latency;
 
     struct camu_resampler_format fmt;
     struct camu_resampler *resampler;
@@ -61,6 +63,8 @@ bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_code
     struct camu_mixer *mixer);
 void camu_audio_buffer_set_volume(struct camu_audio_buffer *buf, f32 volume);
 void camu_audio_buffer_set_latency(struct camu_audio_buffer *buf, f64 latency);
+void camu_audio_buffer_set_ignore_desync(struct camu_audio_buffer *buf, bool ignore_desync);
+void camu_audio_buffer_set_no_video(struct camu_audio_buffer *buf, bool no_video);
 void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_frame *frame);
 void camu_audio_buffer_flush(struct camu_audio_buffer *buf);
 void camu_audio_buffer_reset(struct camu_audio_buffer *buf);
diff --git a/src/buffer/clock.c b/src/buffer/clock.c
index b711a27..3e3efc0 100644
--- a/src/buffer/clock.c
+++ b/src/buffer/clock.c
@@ -108,7 +108,7 @@ f64 camu_clock_get_base_pts(struct camu_clock *clock)
     return clock->base;
 }
 
-f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset, bool skip_set)
+f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset, bool allow_set)
 {
     f64 pause = al_atomic_load(f64)(&clock->pause, AL_ATOMIC_ACQUIRE);
     if (pause == PAUSED) return -1.0;
@@ -117,11 +117,11 @@ f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset, bool skip_set)
 
     f64 tick = al_atomic_load(f64)(&clock->tick, AL_ATOMIC_RELAXED);
     if (tick == -1.0) {
-        if (skip_set) {
-            return -1.0;
-        } else {
+        if (allow_set) {
             tick = al_atomic_compare_and_swap(f64)(&clock->tick, -1.0, current);
             if (tick == -1.0) tick = current;
+        } else {
+            return -1.0;
         }
     }
 
diff --git a/src/buffer/clock.h b/src/buffer/clock.h
index edad287..a2762a6 100644
--- a/src/buffer/clock.h
+++ b/src/buffer/clock.h
@@ -48,4 +48,4 @@ bool camu_clock_is_armed(struct camu_clock *clock);
 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 offset, bool skip_set);
+f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset, bool allow_set);
diff --git a/src/buffer/video.c b/src/buffer/video.c
index e2b6f8e..47c1ea9 100644
--- a/src/buffer/video.c
+++ b/src/buffer/video.c
@@ -28,7 +28,7 @@ bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *cl
     buf->avg_frame_duration = 0.0;
     buf->queue = NULL;
     buf->buffered = false;
-    buf->weighted_read = false;
+    buf->weighted_read = true;
     al_atomic_store(u8)(&buf->flow, FLOWING, AL_ATOMIC_RELAXED);
 #ifdef CAMU_SCREEN_THREADED
     al_atomic_store(u8)(&buf->ref, 0, AL_ATOMIC_RELAXED);
@@ -242,7 +242,7 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weig
 {
     f64 base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
     if (!buf->single_frame) {
-        f64 pts = camu_clock_get_pts(buf->clock, buf->latency, buf->weighted_read);
+        f64 pts = camu_clock_get_pts(buf->clock, buf->latency, !buf->weighted_read);
         if (pts > base) {
             base = pts;
             al_atomic_store(f64)(&buf->pts, base, AL_ATOMIC_RELEASE);
diff --git a/src/buffer/video.h b/src/buffer/video.h
index 5313f75..632a28f 100644
--- a/src/buffer/video.h
+++ b/src/buffer/video.h
@@ -15,8 +15,8 @@ struct camu_video_buffer {
     struct camu_codec_stream *stream;
 
     struct camu_clock *clock;
-    f64 latency;
     atomic(f64) pts;
+    f64 latency;
     f64 last_pts;
 
     bool single_frame;