1 files changed, 425 insertions, 0 deletions

diff --git a/fftools/sync_queue.c b/fftools/sync_queue.c
new file mode 100644
index 0000000000..ab654ca790
--- /dev/null
+++ b/fftools/sync_queue.c
@@ -0,0 +1,425 @@
+/*
+ * 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/error.h"
+#include "libavutil/fifo.h"
+#include "libavutil/mathematics.h"
+#include "libavutil/mem.h"
+
+#include "objpool.h"
+#include "sync_queue.h"
+
+typedef struct SyncQueueStream {
+    AVFifo          *fifo;
+    AVRational       tb;
+
+    /* stream head: largest timestamp seen */
+    int64_t          head_ts;
+    /* no more frames will be sent for this stream */
+    int              finished;
+} SyncQueueStream;
+
+struct SyncQueue {
+    enum SyncQueueType type;
+
+    /* 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;
+};
+
+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);
+}
+
+static int64_t frame_ts(const SyncQueue *sq, SyncQueueFrame frame)
+{
+    return (sq->type == SYNC_QUEUE_PACKETS) ?
+           frame.p->pts + frame.p->duration :
+           frame.f->pts + frame.f->pkt_duration;
+}
+
+static int frame_null(const SyncQueue *sq, SyncQueueFrame frame)
+{
+    return (sq->type == SYNC_QUEUE_PACKETS) ? (frame.p == NULL) : (frame.f == NULL);
+}
+
+static void finish_stream(SyncQueue *sq, unsigned int stream_idx)
+{
+    SyncQueueStream *st = &sq->streams[stream_idx];
+
+    st->finished = 1;
+
+    if (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)
+                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;
+}
+
+static void queue_head_update(SyncQueue *sq)
+{
+    if (sq->head_stream < 0) {
+        /* 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->head_ts == AV_NOPTS_VALUE)
+                return;
+        }
+
+        // placeholder value, correct one will be found below
+        sq->head_stream = 0;
+    }
+
+    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->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 (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_ts(sq, frame);
+
+    /* 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);
+
+        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;
+
+    av_assert0(stream_idx < sq->nb_streams);
+    st = &sq->streams[stream_idx];
+
+    av_assert0(st->tb.num > 0 && st->tb.den > 0);
+
+    if (frame_null(sq, frame)) {
+        finish_stream(sq, stream_idx);
+        return 0;
+    }
+    if (st->finished)
+        return AVERROR_EOF;
+
+    ret = objpool_get(sq->pool, (void**)&dst);
+    if (ret < 0)
+        return ret;
+
+    frame_move(sq, dst, frame);
+
+    ts = frame_ts(sq, dst);
+
+    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);
+
+    return 0;
+}
+
+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)) {
+        SyncQueueFrame peek;
+        int64_t ts;
+        int cmp = 1;
+
+        av_fifo_peek(st->fifo, &peek, 1, 0);
+        ts = frame_ts(sq, peek);
+
+        /* 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.
+         */
+        if (cmp <= 0 || ts == AV_NOPTS_VALUE) {
+            frame_move(sq, frame, peek);
+            objpool_release(sq->pool, (void**)&peek);
+            av_fifo_drain2(st->fifo, 1);
+            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)
+{
+    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;
+
+    return sq->nb_streams++;
+}
+
+void sq_set_tb(SyncQueue *sq, unsigned int stream_idx, AVRational tb)
+{
+    SyncQueueStream *st;
+
+    av_assert0(stream_idx < sq->nb_streams);
+    st = &sq->streams[stream_idx];
+
+    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;
+}
+
+SyncQueue *sq_alloc(enum SyncQueueType type, int64_t buf_size_us)
+{
+    SyncQueue *sq = av_mallocz(sizeof(*sq));
+
+    if (!sq)
+        return NULL;
+
+    sq->type                 = type;
+    sq->buf_size_us          = buf_size_us;
+
+    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);
+}