/* * 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++) {        const 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); 

Lines 1–500 of 719. Load more