path: root/src/buffer/audio.c

#define AL_LOG_SECTION "audio_buffer"
#include <al/log.h>

#ifdef CAMU_HAVE_FFMPEG
#include "../codec/ffmpeg/resampler.h"
#endif

#include "audio.h"
#include "volume.h"
#include "common.h"
#include "common_internal.h"

#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

#ifdef CAMU_AUDIO_BUFFER_FADE
#define FADE_STEP(fmt, down) ((down ? -4.75f : 2.25f) / (fmt)->sample_rate)
#define FADE_MIN 0.25f
#define FADE_TAIL 1
#endif

enum {
    // 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,
    // Return silence fade.tail times then pause.
    PAUSE_FADE_COMPLETE,
#endif
    PAUSE_PLAYING
};

static void reset_buffer_state(struct camu_audio_buffer *buf)
{
    atomic_store(f64)(&buf->pts, -1.0, AL_ATOMIC_RELAXED);
    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);
    buf->buffered = false;
    atomic_store(u32)(&buf->flow, FLOWING, AL_ATOMIC_RELAXED);
    atomic_store(ptrdiff_t)(&buf->uncork_at, 0, AL_ATOMIC_RELAXED);
}

bool camu_audio_buffer_init(struct camu_audio_buffer *buf, struct camu_clock *clock)
{
    buf->clock = clock;
    buf->latency = 0.0;
    buf->ignore_desync = false;
    atomic_store(bool)(&buf->no_video, false, AL_ATOMIC_RELAXED);
    reset_buffer_state(buf);
#ifdef CAMU_AUDIO_BUFFER_FADE
    // Persist fade volume across resets.
    buf->fade.volume = -1.f;
#endif
#ifdef CAMU_MIXER_THREADED
    atomic_store(bool)(&buf->ref, false, AL_ATOMIC_RELAXED);
#endif
    return true;
}

bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_codec_stream *stream,
    struct camu_mixer *mixer)
{
    buf->stream = stream;

    struct camu_audio_format *fmt = &stream->audio.fmt;
    const char *format_name = camu_audio_format_name(fmt->format);
    log_info("Stream: %s (%dch) %dHz.", format_name, fmt->channel_count, fmt->sample_rate);

    struct camu_audio_format *in = &buf->fmt.in;
    struct camu_audio_format *req = &buf->fmt.req;
    camu_audio_format_copy(in, fmt);
    camu_mixer_pick_format(mixer, &buf->fmt);
    buf->fmt.resampler_needed = !camu_resampler_format_matches(&buf->fmt);
    if (buf->fmt.resampler_needed) {
#ifdef CAMU_HAVE_FFMPEG
        buf->resampler = camu_ff_resampler_create();
        if (!buf->resampler->init(buf->resampler, &buf->fmt)) {
            // Resampler will be freed in audio_buffer_free().
            return false;
        }
#else
        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);
    }

    buf->size = (ptrdiff_t)camu_audio_format_sec_to_bytes(req, BUFFER_SIZE);
    buf->data = (u8 *)al_malloc(buf->size);
    al_ring_buffer_init(&buf->rb, buf->data, buf->size);

    buf->mark.min = (ptrdiff_t)camu_audio_format_sec_to_bytes(req, BUFFER_MARK_MIN);
    buf->mark.buffered = (ptrdiff_t)camu_audio_format_sec_to_bytes(req, BUFFER_MARK_BUFFERED);
    camu_peak_buffer_init(&buf->peak, KB(16));

    return true;
}

void camu_audio_buffer_set_volume(struct camu_audio_buffer *buf, f32 volume)
{
    atomic_store(f32)(&buf->volume.queued, volume, AL_ATOMIC_RELAXED);
    atomic_add(u32)(&buf->volume.set, 1, AL_ATOMIC_RELAXED);
}

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)
{
    atomic_store(bool)(&buf->no_video, no_video, AL_ATOMIC_RELAXED);
}

// 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)
{
    f64 base_pts = atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);

    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) {
            sample_count = buf->resampler->convert(buf->resampler, (const u8 **)data, sample_count);
            data = buf->resampler->get_data(buf->resampler);
        }
    } else {
        if (buf->fmt.resampler_needed) {
            sample_count = buf->resampler->flush(buf->resampler);
            data = buf->resampler->get_data(buf->resampler);
        }
    }

    if (base_pts == -1.0) { // This push() has to set buf->pts even if sample_count = 0.
        al_assert(!buf->buffered);
        atomic_store(f64)(&buf->pts, pts, AL_ATOMIC_RELEASE);
    }

    if (sample_count == 0) {
        // The resampler is holding all the data.
        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) {
        buf->buffered = true;
        log_debug("Buffered (mark: %.1fKB).", buf->mark.buffered / 1024.0);
        buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED);
    }

    ptrdiff_t have = (ptrdiff_t)camu_audio_format_samples_to_bytes(&buf->fmt.req, sample_count);
    ptrdiff_t peak = camu_peak_buffer_get_size(&buf->peak);
    if (space < have + peak) {
        al_assert(have < buf->mark.min);
        camu_peak_buffer_push(&buf->peak, data[0], have);
        peak += have;
        if (peak >= buf->mark.min) {
            // If this happens the writer of this buffer is taking way too long to stop.
            log_warn("Overrun likely, discarding peak buffer (audio will be desynced).");
            camu_peak_buffer_flush(&buf->peak);
            peak = 0;
        }
        al_assert(peak <= buf->mark.min);
        // We adjust the min mark by the peak buffer size just for consistency.
        atomic_store(ptrdiff_t)(&buf->uncork_at, buf->mark.min - peak, AL_ATOMIC_RELAXED);
        buf->callback(buf->userdata, CAMU_BUFFER_CORK);
        return false;
    }

    if (peak > 0) {
        al_ring_buffer_write(&buf->rb, camu_peak_buffer_flush(&buf->peak), peak);
    }

    al_ring_buffer_write(&buf->rb, data[0], have);

    return true;
}

#ifdef CAMU_HAVE_FFMPEG
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);
    av_frame_free(&frame);
}
#endif

// A reset() must finish before any data is pushed.
void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_frame *frame)
{
    u8 flow = atomic_load(u32)(&buf->flow, AL_ATOMIC_RELAXED);
    // flow could be ERRORED here.
    if (flow != FLOWING) {
        // Assert that push() is never called after flush().
        al_assert(flow != FLUSHED);
        camu_codec_frame_discard(frame);
        return;
    }

    switch (frame->mode) {
    case CAMU_NORMAL: {
        s32 sample_count = frame->audio.sample_count;
        u8 *planes[CAMU_PLANAR_DATA_POINTERS] = { 0 };
        planes[0] = frame->data;
        push_internal(buf, frame->pts, planes, sample_count);
        break;
    }
#ifdef CAMU_HAVE_FFMPEG
    case CAMU_FFMPEG_COMPAT: {
        push_av_frame_internal(buf, frame->av.frame);
        break;
    }
#endif
    }

    al_free(frame);

#ifdef CAMU_BUFFER_SPORADIC_ERRORS
    ROLL_FOR_BUFFER_ERROR(buf);
#endif
}

// flush() always comes from the same thread as push().
void camu_audio_buffer_flush(struct camu_audio_buffer *buf, bool error)
{
    log_debug("Flush requested.");
    u8 flow = error ? FLUSHED_ERROR : FLUSHED;
    atomic_store(u32)(&buf->flow, flow, AL_ATOMIC_RELAXED);
    if (!push_internal(buf, 0.0, NULL, 0)) {
        log_debug("Buffer filled by flush.");
    }
    if (!buf->buffered) {
        buf->buffered = true;
        log_debug("Buffered (flush).");
        buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED);
    }
}

// Not thread-safe, must be called while the buffer is not being read from or written to.
void camu_audio_buffer_reset(struct camu_audio_buffer *buf)
{
    reset_buffer_state(buf);
    al_ring_buffer_reset(&buf->rb);
    camu_peak_buffer_flush(&buf->peak);
    if (buf->fmt.resampler_needed) {
        buf->resampler->flush(buf->resampler);
    }
}

void camu_audio_buffer_resync(struct camu_audio_buffer *buf)
{
    atomic_add(u32)(&buf->unpause, 1, AL_ATOMIC_RELAXED);
}

ptrdiff_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, ptrdiff_t req)
{
    // Assert this buffer isn't being read before we signaled BUFFER_BUFFERED.
    al_assert(buf->buffered);

    u8 flow = atomic_load(u32)(&buf->flow, AL_ATOMIC_ACQUIRE);
    if (UNLIKELY(flow == ERRORED || flow == FLUSHED_ERROR)) {
        if (flow == FLUSHED_ERROR) {
            buf->callback(buf->userdata, CAMU_BUFFER_ERRORED);
            atomic_store(u32)(&buf->flow, ERRORED, AL_ATOMIC_RELEASE);
        }
        al_memset(data, 0, req);
        return req;
    }

    struct camu_audio_format *fmt = &buf->fmt.req;
    ptrdiff_t ret, signal = req;

    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);
    if (pts == CAMU_PTS_SIGNAL_PAUSE) {
        return 0;
    } 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_FADE_COMPLETE || buf->fade.volume == 0.f) {
            al_memset(data, 0, req);
            if (buf->pause == PAUSE_FADE_COMPLETE) {
                if (!--buf->fade.tail) {
                    buf->callback(buf->userdata, CAMU_BUFFER_PAUSED);
                    buf->pause = PAUSE_PAUSED;
                }
            } 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_FADE_COMPLETE;
                // How many frames of silence to append after fading.
                buf->fade.tail = FADE_TAIL;
            }
            return req;
        }
    } else if (buf->pause == PAUSE_FADING || buf->pause == PAUSE_FADE_COMPLETE) {
        // If unpaused during a fade out, resume immediately to not break the stream.
        // 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);
            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);
        if (buf->pause != PAUSE_PAUSED) {
            buf->pause = PAUSE_PAUSED;
            buf->callback(buf->userdata, CAMU_BUFFER_PAUSED);
        }
        return req;
#endif
    }

    if (atomic_load(u32)(&buf->volume.set, AL_ATOMIC_ACQUIRE) > 0) {
        buf->volume.user = atomic_load(f32)(&buf->volume.queued, AL_ATOMIC_RELAXED);
#ifdef CAMU_AUDIO_BUFFER_FADE
        if (buf->fade.volume == -1.f) buf->fade.volume = buf->volume.user;
#endif
        atomic_sub(u32)(&buf->volume.set, 1, AL_ATOMIC_RELEASE);
    }

    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.");
        buf->pause = PAUSE_SYNC;
        atomic_sub(u32)(&buf->unpause, 1, AL_ATOMIC_RELEASE);
    }

    ptrdiff_t have = al_ring_buffer_occupied(&buf->rb);
#ifdef CAMU_AUDIO_BUFFER_FADE
    if (buf->pause == PAUSE_FADING) {
        // Cut off fade if it's reaching too far.
        if (have < buf->fade.offset) have = 0;
        else have -= buf->fade.offset;
    }
#endif

    // 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_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);
                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.
            } else if (pts < 0.0) { // Delay.
                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 (%zd bytes).", pts, ret);
                    buf->logged_delay = true;
                }
                al_memset(data, 0, ret);
                data += ret;
                req -= ret;
                // We can continue to delay.
                if (req == 0) goto out;
            }
            buf->logged_delay = false;
        }
        buf->pause = PAUSE_PLAYING;
    }

    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);
            if (ret > 0) {
                // This doesn't break single reader, single writer because
                // pushing data after a flush is not allowed.
                al_ring_buffer_write(&buf->rb, camu_peak_buffer_flush(&buf->peak), ret);
                have += ret;
            }
            // Only signal EOF if the data from the peak buffer, if any, wasn't
            // enough for this request.
            if (have < req) {
                signal = have;
                log_debug("Flushed (signal: %zd).", signal);
                buf->callback(buf->userdata, CAMU_BUFFER_EOF);
                atomic_store(u32)(&buf->flow, SIGNALED, AL_ATOMIC_RELEASE);
            }
        } else {
            // 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.
                // An underrun can also happen in the audio output if read() (this function)
                // takes too long. That isn't checked here.
                log_warn("Underrun (req: %zd, have: %zd).", req, have);
                // 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_SYNC;
            }
            // If flow = SIGNALED, we are waiting to be removed or reset.
        }
        // The amount of available data could've increased in the flow = FLUSHED case.
        req = MIN(req, have);
    }

    if (LIKELY(req > 0)) {
#ifdef CAMU_AUDIO_BUFFER_FADE
        if (buf->pause == PAUSE_FADING) {
            // 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 {
#endif
            // Read as much as we determined we can.
            ret = al_ring_buffer_read(&buf->rb, data, req);
            al_assert(ret == req);
            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) {
            bool down = (buf->pause == PAUSE_FADING || buf->volume.user < buf->fade.volume);
            step = FADE_STEP(fmt, down) * MAX(buf->fade.volume, FADE_MIN) *
                ((buf->volume.user > 1.f) ? buf->volume.user : 1.f); // Move faster at >100%.
        }
        if (step != 0.f || buf->fade.volume != 1.f) {
            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, fmt, buf->volume.user, 0.f, 0.f);
#endif
    }

    // Check if we should request to uncork.
#ifdef CAMU_AUDIO_BUFFER_FADE
    if (flow == FLOWING && buf->pause != PAUSE_FADING) {
#else
    if (flow == FLOWING) {
#endif
        ret = atomic_load(ptrdiff_t)(&buf->uncork_at, AL_ATOMIC_RELAXED);
        if (ret && have - req <= ret) {
            buf->callback(buf->userdata, CAMU_BUFFER_UNCORK);
        }
    }

out:
    // Updating buf->pts from the read() thread has to be done in a single step.
    // atomic_add() at the points where base_pts is incremented would be incorrect.
    atomic_store(f64)(&buf->pts, base_pts, AL_ATOMIC_RELEASE);

    // To signal EOF, return less then req.
    return signal;
}

void camu_audio_buffer_free(struct camu_audio_buffer *buf)
{
    if (buf->fmt.resampler_needed) {
        buf->resampler->free(&buf->resampler);
    }
    if (buf->data) al_free(buf->data);
    camu_peak_buffer_free(&buf->peak);
}