path: root/src/buffer/video.c

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

#ifdef CAMU_HAVE_FFMPEG
#include "../codec/ffmpeg/common.h"
#include "../codec/ffmpeg/scaler.h"
#endif

#include "video.h"
#include "common.h"
#include "common_internal.h"

// This used to scale based on avg_frame_duration, but that's actually
// quite incompatible with very low frame rates.
// 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 7 // Must be >1.
#define BUFFER_MARK_HIGH 8
// Keep MARK_RESET high because currently the decoder can produce an
// unpredictable amount of frames at any point.
#define BUFFER_MARK_RESET (BUFFER_MARK_HIGH * 3)
#endif

bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *clock)
{
    buf->clock = clock;
    buf->latency = 0.0;
    atomic_store(f64)(&buf->pts, -1.0, AL_ATOMIC_RELAXED);
    buf->seek_pts = -1.0;
    buf->is_static = true;
    buf->queue = NULL;
    buf->buffered = false;
    buf->weighted_first_read = false;
    atomic_store(u32)(&buf->flow, FLOWING, AL_ATOMIC_RELAXED);
#ifdef CAMU_SCREEN_THREADED
    atomic_store(bool)(&buf->ref, false, AL_ATOMIC_RELAXED);
#endif
    buf->view.mode = CAMU_VIEW_NONE;
    return true;
}

bool camu_video_buffer_configure(struct camu_video_buffer *buf, struct camu_codec_stream *stream,
    struct camu_renderer *renderer)
{
    buf->stream = stream;
    buf->queue = renderer->create_queue(renderer);
    buf->queue->buf = buf;
    struct camu_video_format *fmt = &buf->stream->video.fmt;
    const char *format_name = camu_pixel_format_name(fmt->format);
    switch (stream->mode) {
    case CAMU_NORMAL: {
        buf->is_static = true;
        buf->avg_frame_duration = 0.0;
        log_info("Stream: %s (%ux%u) IMAGE.", format_name, fmt->width, fmt->height);
        break;
    }
#ifdef CAMU_HAVE_FFMPEG
    case CAMU_FFMPEG_COMPAT: {
        AVRational frame_rate = stream->av.stream->avg_frame_rate;
        buf->is_static = stream->duration == 0 || frame_rate.den == 0;
        if (buf->is_static) {
            buf->avg_frame_duration = 0.0;
            log_info("Stream: %s (%ux%u) IMAGE.", format_name, fmt->width, fmt->height);
        } else {
            al_assert(frame_rate.den > 0);
            buf->avg_frame_duration = av_q2d(av_inv_q(frame_rate));
            buf->weighted_first_read = true;
            log_info("Stream: %s (%ux%u) VIDEO %.3ffps.", format_name, fmt->width, fmt->height, av_q2d(frame_rate));
        }
        break;
    }
#endif
    }
    struct camu_video_format *in = &buf->fmt.in;
    struct camu_video_format *req = &buf->fmt.req;
    camu_video_format_copy(in, fmt);
    camu_video_format_copy(req, in);
    if (fmt->width > 0 && fmt->height > 0 &&
#ifdef CAMU_VIDEO_BUFFER_REQUIRE_RGB
        (fmt->format != CAMU_PIXEL_FORMAT_RGBA && fmt->format != CAMU_PIXEL_FORMAT_RGB24 && fmt->format != CAMU_PIXEL_FORMAT_RGB32)
#else
        (fmt->format == CAMU_PIXEL_FORMAT_PAL8)
#endif
    ) {
        buf->fmt.scaler_needed = true;
        req->width = fmt->width;
        req->height = fmt->height;
#ifdef CAMU_HAVE_FFMPEG
        const AVPixFmtDescriptor *desc = av_pix_fmt_desc_get((enum AVPixelFormat)fmt->format);
        bool has_alpha = (desc->flags & AV_PIX_FMT_FLAG_ALPHA) != 0;
        req->format = has_alpha ? CAMU_PIXEL_FORMAT_RGBA : CAMU_PIXEL_FORMAT_RGB24;
        buf->scaler = camu_ff_scaler_create();
        if (!buf->scaler->init(buf->scaler, &buf->fmt)) {
            // Scaler will be freed in video_buffer_free().
            return false;
        }
#else
        return false;
#endif
        const char *req_format_name = camu_pixel_format_name(req->format);
        log_info("Scaling to: %s (%ux%u).", req_format_name, req->width, req->height);
    } else {
        buf->fmt.scaler_needed = false;
        camu_video_format_copy(req, in);
    }
    return true;
}

bool camu_video_buffer_configure_subtitles(struct camu_video_buffer *buf, struct camu_codec_stream *stream)
{
    if (!buf->queue) return false;
    struct camu_video_format *fmt = &buf->stream->video.fmt;
    return buf->queue->configure_subtitles(buf->queue, fmt->width, fmt->height, stream);
}

void camu_video_buffer_set_latency(struct camu_video_buffer *buf, f64 latency)
{
    buf->latency = latency;
}

#ifdef CAMU_HAVE_FFMPEG
static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame, u32 *count)
{
    AVStream *stream = buf->stream->av.stream;
    f64 pts = frame->best_effort_timestamp * av_q2d(stream->time_base);
    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->is_static && frame_is_late(buf->clock, base_pts, pts, duration)) {
        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);
    if (buf->fmt.scaler_needed) {
        if (!buf->scaler->scale(buf->scaler, (const u8 **)frame->data, frame->linesize)) {
            return false;
        }
        av_frame_free(&frame);
        frame = av_frame_clone(buf->scaler->get_frame(buf->scaler)->av.frame);
    }
    buf->queue->push_av_frame(buf->queue, frame, pts, count);
    return true;
}
#endif

void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_frame *frame)
{
    u8 flow = atomic_load(u32)(&buf->flow, AL_ATOMIC_ACQUIRE);
    if (flow != FLOWING) {
        // A static buffer will be FLUSHED after any push().
        if (buf->is_static) {
            log_warn("Unexpected duplicate frame received.");
        }
        camu_codec_frame_discard(frame);
        return;
    }

    u32 count;
    switch (frame->mode) {
    case CAMU_NORMAL: {
        f64 base_pts = atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
        if (base_pts == -1.0) {
            atomic_store(f64)(&buf->pts, 0.0, AL_ATOMIC_RELEASE);
        }
        buf->queue->push(buf->queue, frame, 0.0, &count);
        break;
    }
#ifdef CAMU_HAVE_FFMPEG
    case CAMU_FFMPEG_COMPAT: {
        if (!push_av_frame_internal(buf, frame->av.frame, &count)) {
            camu_codec_frame_discard(frame);
            return;
        }
        al_free(frame);
        break;
    }
#endif
    }

    if (!buf->buffered && (buf->is_static || count >= BUFFER_MARK_BUFFERED)) {
        // Preserve order of: set flow -> flush -> callback, for static buffers.
        if (buf->is_static) {
            atomic_store(u32)(&buf->flow, FLUSHED, AL_ATOMIC_RELEASE);
            buf->queue->flush(buf->queue, &count);
        }
        buf->buffered = true;
        buf->buffered_with_one_frame = count == 1;
        log_debug("Buffered (mark: %.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);
    }

#ifdef CAMU_BUFFER_SPORADIC_ERRORS
    ROLL_FOR_BUFFER_ERROR(buf);
#endif
}

void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struct camu_codec_packet *packet)
{
    if (buf->queue) buf->queue->push_subtitle(buf->queue, packet);
}

// flush() always comes from the same thread as push().
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);
    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);
        buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED);
    }
}

// Not thread-safe, must be called while the buffer is not being read from or written to.
// If a queue->reset() happened from the push() thread while the buffer was active,
// the latest read() frame could be freed before it was used.
void camu_video_buffer_reset(struct camu_video_buffer *buf, f64 pts)
{
    buf->seek_pts = pts;
    atomic_store(f64)(&buf->pts, -1.0, AL_ATOMIC_RELAXED);
    if (buf->queue) buf->queue->reset(buf->queue);
    buf->buffered = false;
    atomic_store(u32)(&buf->flow, FLOWING, AL_ATOMIC_RELAXED);
}

bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weighted)
{
    // Assert this buffer isn't being read before we signaled BUFFER_BUFFERED.
    al_assert(buf->buffered);

    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);
        if (!CAMU_PTS_CONSIDER_PAUSED(pts) && pts > base_pts) {
            base_pts = pts;
        }
    }

    u8 flow = atomic_load(u32)(&buf->flow, AL_ATOMIC_ACQUIRE);
    if (flow == ERRORED) return false;

    u32 count;
    u8 ret = buf->queue->read(buf->queue, base_pts, out, &count);
    if (ret == CAMU_QUEUE_ERR || flow == FLUSHED_ERROR) {
        buf->callback(buf->userdata, CAMU_BUFFER_ERRORED);
        atomic_store(u32)(&buf->flow, ERRORED, AL_ATOMIC_RELEASE);
        return false;
    }

    // We must update buf->pts on every read(), even on QUEUE_MORE.
    // Consdier a case where the PTS we call queue->read() with does nothing but
    // drop frames and returns QUEUE_MORE. If we didn't update buf->pts during
    // that call and on the next call to read() the clock is paused, base_pts
    // will erroneously be less than PTS we just used on the previous read().
    atomic_store(f64)(&buf->pts, base_pts, AL_ATOMIC_RELEASE);

    // If a buffer only ever has 1 frame, libplacebo will never return EOF.
    // In our code that 1 frame buffer might be a video, in which case that behavior is erroneous.
    bool eof = ret == CAMU_QUEUE_EOF || (buf->buffered_with_one_frame && ret == CAMU_QUEUE_OK);
    if (flow == FLUSHED && eof) {
        log_debug("Flushed.");
        buf->callback(buf->userdata, CAMU_BUFFER_EOF);
        atomic_store(u32)(&buf->flow, SIGNALED, AL_ATOMIC_RELEASE);
    } else if (flow == FLOWING) {
        // Never attempt to uncork if flow = FLUSHED.
        if (count <= BUFFER_MARK_LOW) {
            buf->callback(buf->userdata, CAMU_BUFFER_UNCORK);
        } else if (count >= BUFFER_MARK_RESET) {
            log_warn("Buffer overflow, resetting.");
            ret = CAMU_QUEUE_ERR; // Don't use the frame written to out.
            buf->queue->reset(buf->queue);
        }
    }

    if (UNLIKELY(ret == CAMU_QUEUE_OK && buf->weighted_first_read)) {
        *weighted = true;
        buf->weighted_first_read = false;
    } else if (ret == CAMU_QUEUE_MORE) {
        log_warn("Underrun.");
    }

    return ret == CAMU_QUEUE_OK || ret == CAMU_QUEUE_MORE;
}

void camu_video_buffer_free(struct camu_video_buffer *buf)
{
    if (buf->fmt.scaler_needed) buf->scaler->free(&buf->scaler);
    if (buf->queue) buf->queue->free(&buf->queue);
}