sync_queue.c

/*
 * This file is part of FFmpeg.
 *
 * FFmpeg is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * FFmpeg is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with FFmpeg; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 */

#include <stdint.h>
#include <string.h>

#include "libavutil/avassert.h"
#include "libavutil/channel_layout.h"
#include "libavutil/cpu.h"
#include "libavutil/error.h"
#include "libavutil/fifo.h"
#include "libavutil/mathematics.h"
#include "libavutil/mem.h"
#include "libavutil/samplefmt.h"
#include "libavutil/timestamp.h"

#include "objpool.h"
#include "sync_queue.h"

/*
 * How this works:
 * --------------
 * time:   0    1    2    3    4    5    6    7    8    9    10   11   12   13
 *         -------------------------------------------------------------------
 *         |    |    |    |    |    |    |    |    |    |    |    |    |    |
 *         |    ┌───┐┌────────┐┌───┐┌─────────────┐
 * stream 0|    │d=1││  d=2   ││d=1││    d=3      │
 *         |    └───┘└────────┘└───┘└─────────────┘
 *         ┌───┐               ┌───────────────────────┐
 * stream 1│d=1│               │         d=5           │
 *         └───┘               └───────────────────────┘
 *         |    ┌───┐┌───┐┌───┐┌───┐
 * stream 2|    │d=1││d=1││d=1││d=1│ <- stream 2 is the head stream of the queue
 *         |    └───┘└───┘└───┘└───┘
 *                  ^              ^
 *          [stream 2 tail] [stream 2 head]
 *
 * We have N streams (N=3 in the diagram), each stream is a FIFO. The *tail* of
 * each FIFO is the frame with smallest end time, the *head* is the frame with
 * the largest end time. Frames submitted to the queue with sq_send() are placed
 * after the head, frames returned to the caller with sq_receive() are taken
 * from the tail.
 *
 * The head stream of the whole queue (SyncQueue.head_stream) is the limiting
 * stream with the *smallest* head timestamp, i.e. the stream whose source lags
 * furthest behind all other streams. It determines which frames can be output
 * from the queue.
 *
 * In the diagram, the head stream is 2, because it head time is t=5, while
 * streams 0 and 1 end at t=8 and t=9 respectively. All frames that _end_ at
 * or before t=5 can be output, i.e. the first 3 frames from stream 0, first
 * frame from stream 1, and all 4 frames from stream 2.
 */

typedef struct SyncQueueStream {
    AVFifo          *fifo;
    AVRational       tb;

    /* number of audio samples in fifo */
    uint64_t         samples_queued;
    /* stream head: largest timestamp seen */
    int64_t          head_ts;
    int              limiting;
    /* no more frames will be sent for this stream */
    int              finished;

    uint64_t         frames_sent;
    uint64_t         samples_sent;
    uint64_t         frames_max;
    int              frame_samples;
} SyncQueueStream;

struct SyncQueue {
    enum SyncQueueType type;

    void *logctx;

    /* no more frames will be sent for any stream */
    int finished;
    /* sync head: the stream with the _smallest_ head timestamp
     * this stream determines which frames can be output */
    int head_stream;
    /* the finished stream with the smallest finish timestamp or -1 */
    int head_finished_stream;

    // maximum buffering duration in microseconds
    int64_t buf_size_us;

    SyncQueueStream *streams;
    unsigned int  nb_streams;

    // pool of preallocated frames to avoid constant allocations
    ObjPool *pool;

    int have_limiting;

    uintptr_t align_mask;
};

static void frame_move(const SyncQueue *sq, SyncQueueFrame dst,
                       SyncQueueFrame src)
{
    if (sq->type == SYNC_QUEUE_PACKETS)
        av_packet_move_ref(dst.p, src.p);
    else
        av_frame_move_ref(dst.f, src.f);
}

/**
 * Compute the end timestamp of a frame. If nb_samples is provided, consider
 * the frame to have this number of audio samples, otherwise use frame duration.
 */
static int64_t frame_end(const SyncQueue *sq, SyncQueueFrame frame, int nb_samples)
{
    if (nb_samples) {
        int64_t d = av_rescale_q(nb_samples, (AVRational){ 1, frame.f->sample_rate},
                                 frame.f->time_base);
        return frame.f->pts + d;
    }

    return (sq->type == SYNC_QUEUE_PACKETS) ?
           frame.p->pts + frame.p->duration :
           frame.f->pts + frame.f->duration;
}

static int frame_samples(const SyncQueue *sq, SyncQueueFrame frame)
{
    return (sq->type == SYNC_QUEUE_PACKETS) ? 0 : frame.f->nb_samples;
}

static int frame_null(const SyncQueue *sq, SyncQueueFrame frame)
{
    return (sq->type == SYNC_QUEUE_PACKETS) ? (frame.p == NULL) : (frame.f == NULL);
}

static void tb_update(const SyncQueue *sq, SyncQueueStream *st,
                      const SyncQueueFrame frame)
{
    AVRational tb = (sq->type == SYNC_QUEUE_PACKETS) ?
                    frame.p->time_base : frame.f->time_base;

    av_assert0(tb.num > 0 && tb.den > 0);

    if (tb.num == st->tb.num && tb.den == st->tb.den)
        return;

    // timebase should not change after the first frame
    av_assert0(!av_fifo_can_read(st->fifo));

    if (st->head_ts != AV_NOPTS_VALUE)
        st->head_ts = av_rescale_q(st->head_ts, st->tb, tb);

    st->tb = tb;
}

static void finish_stream(SyncQueue *sq, unsigned int stream_idx)
{
    SyncQueueStream *st = &sq->streams[stream_idx];

    if (!st->finished)
        av_log(sq->logctx, AV_LOG_DEBUG,
               "sq: finish %u; head ts %s\n", stream_idx,
               av_ts2timestr(st->head_ts, &st->tb));

    st->finished = 1;

    if (st->limiting && st->head_ts != AV_NOPTS_VALUE) {
        /* check if this stream is the new finished head */
        if (sq->head_finished_stream < 0 ||
            av_compare_ts(st->head_ts, st->tb,
                          sq->streams[sq->head_finished_stream].head_ts,
                          sq->streams[sq->head_finished_stream].tb) < 0) {
            sq->head_finished_stream = stream_idx;
        }

        /* mark as finished all streams that should no longer receive new frames,
         * due to them being ahead of some finished stream */
        st = &sq->streams[sq->head_finished_stream];
        for (unsigned int i = 0; i < sq->nb_streams; i++) {
            SyncQueueStream *st1 = &sq->streams[i];
            if (st != st1 && st1->head_ts != AV_NOPTS_VALUE &&
                av_compare_ts(st->head_ts, st->tb, st1->head_ts, st1->tb) <= 0) {
                if (!st1->finished)
                    av_log(sq->logctx, AV_LOG_DEBUG,
                           "sq: finish secondary %u; head ts %s\n", i,
                           av_ts2timestr(st1->head_ts, &st1->tb));

                st1->finished = 1;
            }
        }
    }

    /* mark the whole queue as finished if all streams are finished */
    for (unsigned int i = 0; i < sq->nb_streams; i++) {
        if (!sq->streams[i].finished)
            return;
    }
    sq->finished = 1;

    av_log(sq->logctx, AV_LOG_DEBUG, "sq: finish queue\n");
}

static void queue_head_update(SyncQueue *sq)
{
    av_assert0(sq->have_limiting);

    if (sq->head_stream < 0) {
        unsigned first_limiting = UINT_MAX;

        /* wait for one timestamp in each stream before determining
         * the queue head */
        for (unsigned int i = 0; i < sq->nb_streams; i++) {
            SyncQueueStream *st = &sq->streams[i];
            if (!st->limiting)
                continue;
            if (st->head_ts == AV_NOPTS_VALUE)
                return;
            if (first_limiting == UINT_MAX)
                first_limiting = i;
        }

        // placeholder value, correct one will be found below
        av_assert0(first_limiting < UINT_MAX);
        sq->head_stream = first_limiting;
    }

    for (unsigned int i = 0; i < sq->nb_streams; i++) {
        SyncQueueStream *st_head  = &sq->streams[sq->head_stream];
        SyncQueueStream *st_other = &sq->streams[i];
        if (st_other->limiting && st_other->head_ts != AV_NOPTS_VALUE &&
            av_compare_ts(st_other->head_ts, st_other->tb,
                          st_head->head_ts,  st_head->tb) < 0)
            sq->head_stream = i;
    }
}

/* update this stream's head timestamp */
static void stream_update_ts(SyncQueue *sq, unsigned int stream_idx, int64_t ts)
{
    SyncQueueStream *st = &sq->streams[stream_idx];

    if (ts == AV_NOPTS_VALUE ||
        (st->head_ts != AV_NOPTS_VALUE && st->head_ts >= ts))
        return;

    st->head_ts = ts;

    /* if this stream is now ahead of some finished stream, then
     * this stream is also finished */
    if (sq->head_finished_stream >= 0 &&
        av_compare_ts(sq->streams[sq->head_finished_stream].head_ts,
                      sq->streams[sq->head_finished_stream].tb,
                      ts, st->tb) <= 0)
        finish_stream(sq, stream_idx);

    /* update the overall head timestamp if it could have changed */
    if (st->limiting &&
        (sq->head_stream < 0 || sq->head_stream == stream_idx))
        queue_head_update(sq);
}

/* If the queue for the given stream (or all streams when stream_idx=-1)
 * is overflowing, trigger a fake heartbeat on lagging streams.
 *
 * @return 1 if heartbeat triggered, 0 otherwise
 */
static int overflow_heartbeat(SyncQueue *sq, int stream_idx)
{
    SyncQueueStream *st;
    SyncQueueFrame frame;
    int64_t tail_ts = AV_NOPTS_VALUE;

    /* if no stream specified, pick the one that is most ahead */
    if (stream_idx < 0) {
        int64_t ts = AV_NOPTS_VALUE;

        for (int i = 0; i < sq->nb_streams; i++) {
            st = &sq->streams[i];
            if (st->head_ts != AV_NOPTS_VALUE &&
                (ts == AV_NOPTS_VALUE ||
                 av_compare_ts(ts, sq->streams[stream_idx].tb,
                               st->head_ts, st->tb) < 0)) {
                ts = st->head_ts;
                stream_idx = i;
            }
        }
        /* no stream has a timestamp yet -> nothing to do */
        if (stream_idx < 0)
            return 0;
    }

    st = &sq->streams[stream_idx];

    /* get the chosen stream's tail timestamp */
    for (size_t i = 0; tail_ts == AV_NOPTS_VALUE &&
                       av_fifo_peek(st->fifo, &frame, 1, i) >= 0; i++)
        tail_ts = frame_end(sq, frame, 0);

    /* overflow triggers when the tail is over specified duration behind the head */
    if (tail_ts == AV_NOPTS_VALUE || tail_ts >= st->head_ts ||
        av_rescale_q(st->head_ts - tail_ts, st->tb, AV_TIME_BASE_Q) < sq->buf_size_us)
        return 0;

    /* signal a fake timestamp for all streams that prevent tail_ts from being output */
    tail_ts++;
    for (unsigned int i = 0; i < sq->nb_streams; i++) {
        SyncQueueStream *st1 = &sq->streams[i];
        int64_t ts;

        if (st == st1 || st1->finished ||
            (st1->head_ts != AV_NOPTS_VALUE &&
             av_compare_ts(tail_ts, st->tb, st1->head_ts, st1->tb) <= 0))
            continue;

        ts = av_rescale_q(tail_ts, st->tb, st1->tb);
        if (st1->head_ts != AV_NOPTS_VALUE)
            ts = FFMAX(st1->head_ts + 1, ts);

        av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u overflow heardbeat %s -> %s\n",
               i, av_ts2timestr(st1->head_ts, &st1->tb), av_ts2timestr(ts, &st1->tb));

        stream_update_ts(sq, i, ts);
    }

    return 1;
}

int sq_send(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame)
{
    SyncQueueStream *st;
    SyncQueueFrame dst;
    int64_t ts;
    int ret, nb_samples;

    av_assert0(stream_idx < sq->nb_streams);
    st = &sq->streams[stream_idx];

    if (frame_null(sq, frame)) {
        av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u EOF\n", stream_idx);
        finish_stream(sq, stream_idx);
        return 0;
    }
    if (st->finished)
        return AVERROR_EOF;

    tb_update(sq, st, frame);

    ret = objpool_get(sq->pool, (void**)&dst);
    if (ret < 0)
        return ret;

    frame_move(sq, dst, frame);

    nb_samples = frame_samples(sq, dst);
    // make sure frame duration is consistent with sample count
    if (nb_samples) {
        av_assert0(dst.f->sample_rate > 0);
        dst.f->duration = av_rescale_q(nb_samples, (AVRational){ 1, dst.f->sample_rate },
                                       dst.f->time_base);
    }

    ts = frame_end(sq, dst, 0);

    av_log(sq->logctx, AV_LOG_DEBUG, "sq: send %u ts %s\n", stream_idx,
           av_ts2timestr(ts, &st->tb));

    ret = av_fifo_write(st->fifo, &dst, 1);
    if (ret < 0) {
        frame_move(sq, frame, dst);
        objpool_release(sq->pool, (void**)&dst);
        return ret;
    }

    stream_update_ts(sq, stream_idx, ts);

    st->samples_queued += nb_samples;
    st->samples_sent   += nb_samples;

    if (st->frame_samples)
        st->frames_sent = st->samples_sent / st->frame_samples;
    else
        st->frames_sent++;

    if (st->frames_sent >= st->frames_max) {
        av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u frames_max %"PRIu64" reached\n",
               stream_idx, st->frames_max);

        finish_stream(sq, stream_idx);
    }

    return 0;
}

static void offset_audio(AVFrame *f, int nb_samples)
{
    const int planar = av_sample_fmt_is_planar(f->format);
    const int planes = planar ? f->ch_layout.nb_channels : 1;
    const int    bps = av_get_bytes_per_sample(f->format);
    const int offset = nb_samples * bps * (planar ? 1 : f->ch_layout.nb_channels);

    av_assert0(bps > 0);
    av_assert0(nb_samples < f->nb_samples);

    for (int i = 0; i < planes; i++) {
        f->extended_data[i] += offset;
        if (i < FF_ARRAY_ELEMS(f->data))
            f->data[i] = f->extended_data[i];
    }
    f->linesize[0] -= offset;
    f->nb_samples  -= nb_samples;
    f->duration     = av_rescale_q(f->nb_samples, (AVRational){ 1, f->sample_rate },
                                   f->time_base);
    f->pts         += av_rescale_q(nb_samples,    (AVRational){ 1, f->sample_rate },
                                   f->time_base);
}

static int frame_is_aligned(const SyncQueue *sq, const AVFrame *frame)
{
    // only checks linesize[0], so only works for audio
    av_assert0(frame->nb_samples > 0);
    av_assert0(sq->align_mask);

    // only check data[0], because we always offset all data pointers
    // by the same offset, so if one is aligned, all are
    if (!((uintptr_t)frame->data[0] & sq->align_mask) &&
        !(frame->linesize[0]        & sq->align_mask) &&
        frame->linesize[0] > sq->align_mask)
        return 1;

    return 0;
}

static int receive_samples(SyncQueue *sq, SyncQueueStream *st,
                           AVFrame *dst, int nb_samples)
{
    SyncQueueFrame src;
    int ret;

    av_assert0(st->samples_queued >= nb_samples);

    ret = av_fifo_peek(st->fifo, &src, 1, 0);
    av_assert0(ret >= 0);

    // peeked frame has enough samples and its data is aligned
    // -> we can just make a reference and limit its sample count
    if (src.f->nb_samples > nb_samples && frame_is_aligned(sq, src.f)) {
        ret = av_frame_ref(dst, src.f);
        if (ret < 0)
            return ret;

        dst->nb_samples = nb_samples;
        offset_audio(src.f, nb_samples);
        st->samples_queued -= nb_samples;

        goto finish;
    }

    // otherwise allocate a new frame and copy the data
    ret = av_channel_layout_copy(&dst->ch_layout, &src.f->ch_layout);
    if (ret < 0)
        return ret;

    dst->format     = src.f->format;
    dst->nb_samples = nb_samples;

    ret = av_frame_get_buffer(dst, 0);
    if (ret < 0)
        goto fail;

    ret = av_frame_copy_props(dst, src.f);
    if (ret < 0)
        goto fail;

    dst->nb_samples = 0;
    while (dst->nb_samples < nb_samples) {
        int to_copy;

        ret = av_fifo_peek(st->fifo, &src, 1, 0);
        av_assert0(ret >= 0);

        to_copy = FFMIN(nb_samples - dst->nb_samples, src.f->nb_samples);

        av_samples_copy(dst->extended_data, src.f->extended_data, dst->nb_samples,
                        0, to_copy, dst->ch_layout.nb_channels, dst->format);

        if (to_copy < src.f->nb_samples)
            offset_audio(src.f, to_copy);
        else {
            av_frame_unref(src.f);
            objpool_release(sq->pool, (void**)&src);
            av_fifo_drain2(st->fifo, 1);
        }
        st->samples_queued -= to_copy;

        dst->nb_samples += to_copy;
    }

finish:
    dst->duration   = av_rescale_q(nb_samples, (AVRational){ 1, dst->sample_rate },
                                   dst->time_base);

    return 0;

fail:
    av_frame_unref(dst);
    return ret;
}

static int receive_for_stream(SyncQueue *sq, unsigned int stream_idx,
                              SyncQueueFrame frame)
{
    SyncQueueStream *st_head = sq->head_stream >= 0 ?
                               &sq->streams[sq->head_stream] : NULL;
    SyncQueueStream *st;

    av_assert0(stream_idx < sq->nb_streams);
    st = &sq->streams[stream_idx];

    if (av_fifo_can_read(st->fifo) &&
        (st->frame_samples <= st->samples_queued || st->finished)) {
        int nb_samples = st->frame_samples;
        SyncQueueFrame peek;
        int64_t ts;
        int cmp = 1;

        if (st->finished)
            nb_samples = FFMIN(nb_samples, st->samples_queued);

        av_fifo_peek(st->fifo, &peek, 1, 0);
        ts = frame_end(sq, peek, nb_samples);

        /* check if this stream's tail timestamp does not overtake
         * the overall queue head */
        if (ts != AV_NOPTS_VALUE && st_head)
            cmp = av_compare_ts(ts, st->tb, st_head->head_ts, st_head->tb);

        /* We can release frames that do not end after the queue head.
         * Frames with no timestamps are just passed through with no conditions.
         * Frames are also passed through when there are no limiting streams.
         */
        if (cmp <= 0 || ts == AV_NOPTS_VALUE || !sq->have_limiting) {
            if (nb_samples &&
                (nb_samples != peek.f->nb_samples || !frame_is_aligned(sq, peek.f))) {
                int ret = receive_samples(sq, st, frame.f, nb_samples);
                if (ret < 0)
                    return ret;
            } else {
                frame_move(sq, frame, peek);
                objpool_release(sq->pool, (void**)&peek);
                av_fifo_drain2(st->fifo, 1);
                av_assert0(st->samples_queued >= frame_samples(sq, frame));
                st->samples_queued -= frame_samples(sq, frame);
            }

            av_log(sq->logctx, AV_LOG_DEBUG,
                   "sq: receive %u ts %s queue head %d ts %s\n", stream_idx,
                   av_ts2timestr(frame_end(sq, frame, 0), &st->tb),
                   sq->head_stream,
                   st_head ? av_ts2timestr(st_head->head_ts, &st_head->tb) : "N/A");

            return 0;
        }
    }

    return (sq->finished || (st->finished && !av_fifo_can_read(st->fifo))) ?
            AVERROR_EOF : AVERROR(EAGAIN);
}

static int receive_internal(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
{
    int nb_eof = 0;
    int ret;

    /* read a frame for a specific stream */
    if (stream_idx >= 0) {
        ret = receive_for_stream(sq, stream_idx, frame);
        return (ret < 0) ? ret : stream_idx;
    }

    /* read a frame for any stream with available output */
    for (unsigned int i = 0; i < sq->nb_streams; i++) {
        ret = receive_for_stream(sq, i, frame);
        if (ret == AVERROR_EOF || ret == AVERROR(EAGAIN)) {
            nb_eof += (ret == AVERROR_EOF);
            continue;
        }
        return (ret < 0) ? ret : i;
    }

    return (nb_eof == sq->nb_streams) ? AVERROR_EOF : AVERROR(EAGAIN);
}

int sq_receive(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
{
    int ret = receive_internal(sq, stream_idx, frame);

    /* try again if the queue overflowed and triggered a fake heartbeat
     * for lagging streams */
    if (ret == AVERROR(EAGAIN) && overflow_heartbeat(sq, stream_idx))
        ret = receive_internal(sq, stream_idx, frame);

    return ret;
}

int sq_add_stream(SyncQueue *sq, int limiting)
{
    SyncQueueStream *tmp, *st;

    tmp = av_realloc_array(sq->streams, sq->nb_streams + 1, sizeof(*sq->streams));
    if (!tmp)
        return AVERROR(ENOMEM);
    sq->streams = tmp;

    st = &sq->streams[sq->nb_streams];
    memset(st, 0, sizeof(*st));

    st->fifo = av_fifo_alloc2(1, sizeof(SyncQueueFrame), AV_FIFO_FLAG_AUTO_GROW);
    if (!st->fifo)
        return AVERROR(ENOMEM);

    /* we set a valid default, so that a pathological stream that never
     * receives even a real timebase (and no frames) won't stall all other
     * streams forever; cf. overflow_heartbeat() */
    st->tb      = (AVRational){ 1, 1 };
    st->head_ts = AV_NOPTS_VALUE;
    st->frames_max = UINT64_MAX;
    st->limiting   = limiting;

    sq->have_limiting |= limiting;

    return sq->nb_streams++;
}

void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames)
{
    SyncQueueStream *st;

    av_assert0(stream_idx < sq->nb_streams);
    st = &sq->streams[stream_idx];

    st->frames_max = frames;
    if (st->frames_sent >= st->frames_max)
        finish_stream(sq, stream_idx);
}

void sq_frame_samples(SyncQueue *sq, unsigned int stream_idx,
                      int frame_samples)
{
    SyncQueueStream *st;

    av_assert0(sq->type == SYNC_QUEUE_FRAMES);
    av_assert0(stream_idx < sq->nb_streams);
    st = &sq->streams[stream_idx];

    st->frame_samples = frame_samples;

    sq->align_mask = av_cpu_max_align() - 1;
}

SyncQueue *sq_alloc(enum SyncQueueType type, int64_t buf_size_us, void *logctx)
{
    SyncQueue *sq = av_mallocz(sizeof(*sq));

    if (!sq)
        return NULL;

    sq->type                 = type;
    sq->buf_size_us          = buf_size_us;
    sq->logctx               = logctx;

    sq->head_stream          = -1;
    sq->head_finished_stream = -1;

    sq->pool = (type == SYNC_QUEUE_PACKETS) ? objpool_alloc_packets() :
                                              objpool_alloc_frames();
    if (!sq->pool) {
        av_freep(&sq);
        return NULL;
    }

    return sq;
}

void sq_free(SyncQueue **psq)
{
    SyncQueue *sq = *psq;

    if (!sq)
        return;

    for (unsigned int i = 0; i < sq->nb_streams; i++) {
        SyncQueueFrame frame;
        while (av_fifo_read(sq->streams[i].fifo, &frame, 1) >= 0)
            objpool_release(sq->pool, (void**)&frame);

        av_fifo_freep2(&sq->streams[i].fifo);
    }

    av_freep(&sq->streams);

    objpool_free(&sq->pool);

    av_freep(psq);
}