fftools/sync_queue.c

   1 /*
   2  * This file is part of FFmpeg.
   3  *
   4  * FFmpeg is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Lesser General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2.1 of the License, or (at your option) any later version.
   8  *
   9  * FFmpeg is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Lesser General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Lesser General Public
  15  * License along with FFmpeg; if not, write to the Free Software
  16  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  17  */
  18
  19 #include <stdint.h>
  20 #include <string.h>
  21
  22 #include "libavutil/avassert.h"
  23 #include "libavutil/container_fifo.h"
  24 #include "libavutil/channel_layout.h"
  25 #include "libavutil/cpu.h"
  26 #include "libavutil/error.h"
  27 #include "libavutil/mathematics.h"
  28 #include "libavutil/mem.h"
  29 #include "libavutil/samplefmt.h"
  30 #include "libavutil/timestamp.h"
  31
  32 #include "sync_queue.h"
  33
  34 /*
  35  * How this works:
  36  * --------------
  37  * time:   0    1    2    3    4    5    6    7    8    9    10   11   12   13
  38  *         -------------------------------------------------------------------
  39  *         |    |    |    |    |    |    |    |    |    |    |    |    |    |
  40  *         |    ┌───┐┌────────┐┌───┐┌─────────────┐
  41  * stream 0|    │d=1││  d=2   ││d=1││    d=3      │
  42  *         |    └───┘└────────┘└───┘└─────────────┘
  43  *         ┌───┐               ┌───────────────────────┐
  44  * stream 1│d=1│               │         d=5           │
  45  *         └───┘               └───────────────────────┘
  46  *         |    ┌───┐┌───┐┌───┐┌───┐
  47  * stream 2|    │d=1││d=1││d=1││d=1│ <- stream 2 is the head stream of the queue
  48  *         |    └───┘└───┘└───┘└───┘
  49  *                  ^              ^
  50  *          [stream 2 tail] [stream 2 head]
  51  *
  52  * We have N streams (N=3 in the diagram), each stream is a FIFO. The *tail* of
  53  * each FIFO is the frame with smallest end time, the *head* is the frame with
  54  * the largest end time. Frames submitted to the queue with sq_send() are placed
  55  * after the head, frames returned to the caller with sq_receive() are taken
  56  * from the tail.
  57  *
  58  * The head stream of the whole queue (SyncQueue.head_stream) is the limiting
  59  * stream with the *smallest* head timestamp, i.e. the stream whose source lags
  60  * furthest behind all other streams. It determines which frames can be output
  61  * from the queue.
  62  *
  63  * In the diagram, the head stream is 2, because it head time is t=5, while
  64  * streams 0 and 1 end at t=8 and t=9 respectively. All frames that _end_ at
  65  * or before t=5 can be output, i.e. the first 3 frames from stream 0, first
  66  * frame from stream 1, and all 4 frames from stream 2.
  67  */
  68
  69 #define SQPTR(sq, frame) ((sq->type == SYNC_QUEUE_FRAMES) ? \
  70                           (void*)frame.f : (void*)frame.p)
  71
  72 typedef struct SyncQueueStream {
  73     AVContainerFifo *fifo;
  74     AVRational       tb;
  75
  76     /* number of audio samples in fifo */
  77     uint64_t         samples_queued;
  78     /* stream head: largest timestamp seen */
  79     int64_t          head_ts;
  80     int              limiting;
  81     /* no more frames will be sent for this stream */
  82     int              finished;
  83
  84     uint64_t         frames_sent;
  85     uint64_t         samples_sent;
  86     uint64_t         frames_max;
  87     int              frame_samples;
  88 } SyncQueueStream;
  89
  90 struct SyncQueue {
  91     enum SyncQueueType type;
  92
  93     void *logctx;
  94
  95     /* no more frames will be sent for any stream */
  96     int finished;
  97     /* sync head: the stream with the _smallest_ head timestamp
  98      * this stream determines which frames can be output */
  99     int head_stream;
 100     /* the finished stream with the smallest finish timestamp or -1 */
 101     int head_finished_stream;
 102
 103     // maximum buffering duration in microseconds
 104     int64_t buf_size_us;
 105
 106     SyncQueueStream *streams;
 107     unsigned int  nb_streams;
 108
 109     int have_limiting;
 110
 111     uintptr_t align_mask;
 112 };
 113
 114 /**
 115  * Compute the end timestamp of a frame. If nb_samples is provided, consider
 116  * the frame to have this number of audio samples, otherwise use frame duration.
 117  */
 118 static int64_t frame_end(const SyncQueue *sq, SyncQueueFrame frame, int nb_samples)
 119 {
 120     if (nb_samples) {
 121         int64_t d = av_rescale_q(nb_samples, (AVRational){ 1, frame.f->sample_rate},
 122                                  frame.f->time_base);
 123         return frame.f->pts + d;
 124     }
 125
 126     return (sq->type == SYNC_QUEUE_PACKETS) ?
 127            frame.p->pts + frame.p->duration :
 128            frame.f->pts + frame.f->duration;
 129 }
 130
 131 static int frame_samples(const SyncQueue *sq, SyncQueueFrame frame)
 132 {
 133     return (sq->type == SYNC_QUEUE_PACKETS) ? 0 : frame.f->nb_samples;
 134 }
 135
 136 static int frame_null(const SyncQueue *sq, SyncQueueFrame frame)
 137 {
 138     return (sq->type == SYNC_QUEUE_PACKETS) ? (frame.p == NULL) : (frame.f == NULL);
 139 }
 140
 141 static void tb_update(const SyncQueue *sq, SyncQueueStream *st,
 142                       const SyncQueueFrame frame)
 143 {
 144     AVRational tb = (sq->type == SYNC_QUEUE_PACKETS) ?
 145                     frame.p->time_base : frame.f->time_base;
 146
 147     av_assert0(tb.num > 0 && tb.den > 0);
 148
 149     if (tb.num == st->tb.num && tb.den == st->tb.den)
 150         return;
 151
 152     // timebase should not change after the first frame
 153     av_assert0(!av_container_fifo_can_read(st->fifo));
 154
 155     if (st->head_ts != AV_NOPTS_VALUE)
 156         st->head_ts = av_rescale_q(st->head_ts, st->tb, tb);
 157
 158     st->tb = tb;
 159 }
 160
 161 static void finish_stream(SyncQueue *sq, unsigned int stream_idx)
 162 {
 163     SyncQueueStream *st = &sq->streams[stream_idx];
 164
 165     if (!st->finished)
 166         av_log(sq->logctx, AV_LOG_DEBUG,
 167                "sq: finish %u; head ts %s\n", stream_idx,
 168                av_ts2timestr(st->head_ts, &st->tb));
 169
 170     st->finished = 1;
 171
 172     if (st->limiting && st->head_ts != AV_NOPTS_VALUE) {
 173         /* check if this stream is the new finished head */
 174         if (sq->head_finished_stream < 0 ||
 175             av_compare_ts(st->head_ts, st->tb,
 176                           sq->streams[sq->head_finished_stream].head_ts,
 177                           sq->streams[sq->head_finished_stream].tb) < 0) {
 178             sq->head_finished_stream = stream_idx;
 179         }
 180
 181         /* mark as finished all streams that should no longer receive new frames,
 182          * due to them being ahead of some finished stream */
 183         st = &sq->streams[sq->head_finished_stream];
 184         for (unsigned int i = 0; i < sq->nb_streams; i++) {
 185             SyncQueueStream *st1 = &sq->streams[i];
 186             if (st != st1 && st1->head_ts != AV_NOPTS_VALUE &&
 187                 av_compare_ts(st->head_ts, st->tb, st1->head_ts, st1->tb) <= 0) {
 188                 if (!st1->finished)
 189                     av_log(sq->logctx, AV_LOG_DEBUG,
 190                            "sq: finish secondary %u; head ts %s\n", i,
 191                            av_ts2timestr(st1->head_ts, &st1->tb));
 192
 193                 st1->finished = 1;
 194             }
 195         }
 196     }
 197
 198     /* mark the whole queue as finished if all streams are finished */
 199     for (unsigned int i = 0; i < sq->nb_streams; i++) {
 200         if (!sq->streams[i].finished)
 201             return;
 202     }
 203     sq->finished = 1;
 204
 205     av_log(sq->logctx, AV_LOG_DEBUG, "sq: finish queue\n");
 206 }
 207
 208 static void queue_head_update(SyncQueue *sq)
 209 {
 210     av_assert0(sq->have_limiting);
 211
 212     if (sq->head_stream < 0) {
 213         unsigned first_limiting = UINT_MAX;
 214
 215         /* wait for one timestamp in each stream before determining
 216          * the queue head */
 217         for (unsigned int i = 0; i < sq->nb_streams; i++) {
 218             SyncQueueStream *st = &sq->streams[i];
 219             if (!st->limiting)
 220                 continue;
 221             if (st->head_ts == AV_NOPTS_VALUE)
 222                 return;
 223             if (first_limiting == UINT_MAX)
 224                 first_limiting = i;
 225         }
 226
 227         // placeholder value, correct one will be found below
 228         av_assert0(first_limiting < UINT_MAX);
 229         sq->head_stream = first_limiting;
 230     }
 231
 232     for (unsigned int i = 0; i < sq->nb_streams; i++) {
 233         SyncQueueStream *st_head  = &sq->streams[sq->head_stream];
 234         SyncQueueStream *st_other = &sq->streams[i];
 235         if (st_other->limiting && st_other->head_ts != AV_NOPTS_VALUE &&
 236             av_compare_ts(st_other->head_ts, st_other->tb,
 237                           st_head->head_ts,  st_head->tb) < 0)
 238             sq->head_stream = i;
 239     }
 240 }
 241
 242 /* update this stream's head timestamp */
 243 static void stream_update_ts(SyncQueue *sq, unsigned int stream_idx, int64_t ts)
 244 {
 245     SyncQueueStream *st = &sq->streams[stream_idx];
 246
 247     if (ts == AV_NOPTS_VALUE ||
 248         (st->head_ts != AV_NOPTS_VALUE && st->head_ts >= ts))
 249         return;
 250
 251     st->head_ts = ts;
 252
 253     /* if this stream is now ahead of some finished stream, then
 254      * this stream is also finished */
 255     if (sq->head_finished_stream >= 0 &&
 256         av_compare_ts(sq->streams[sq->head_finished_stream].head_ts,
 257                       sq->streams[sq->head_finished_stream].tb,
 258                       ts, st->tb) <= 0)
 259         finish_stream(sq, stream_idx);
 260
 261     /* update the overall head timestamp if it could have changed */
 262     if (st->limiting &&
 263         (sq->head_stream < 0 || sq->head_stream == stream_idx))
 264         queue_head_update(sq);
 265 }
 266
 267 /* If the queue for the given stream (or all streams when stream_idx=-1)
 268  * is overflowing, trigger a fake heartbeat on lagging streams.
 269  *
 270  * @return 1 if heartbeat triggered, 0 otherwise
 271  */
 272 static int overflow_heartbeat(SyncQueue *sq, int stream_idx)
 273 {
 274     SyncQueueStream *st;
 275     SyncQueueFrame frame;
 276     int64_t tail_ts = AV_NOPTS_VALUE;
 277
 278     /* if no stream specified, pick the one that is most ahead */
 279     if (stream_idx < 0) {
 280         int64_t ts = AV_NOPTS_VALUE;
 281
 282         for (int i = 0; i < sq->nb_streams; i++) {
 283             st = &sq->streams[i];
 284             if (st->head_ts != AV_NOPTS_VALUE &&
 285                 (ts == AV_NOPTS_VALUE ||
 286                  av_compare_ts(ts, sq->streams[stream_idx].tb,
 287                                st->head_ts, st->tb) < 0)) {
 288                 ts = st->head_ts;
 289                 stream_idx = i;
 290             }
 291         }
 292         /* no stream has a timestamp yet -> nothing to do */
 293         if (stream_idx < 0)
 294             return 0;
 295     }
 296
 297     st = &sq->streams[stream_idx];
 298
 299     /* get the chosen stream's tail timestamp */
 300     for (size_t i = 0; tail_ts == AV_NOPTS_VALUE &&
 301                        av_container_fifo_peek(st->fifo, (void**)&frame, i) >= 0; i++)
 302         tail_ts = frame_end(sq, frame, 0);
 303
 304     /* overflow triggers when the tail is over specified duration behind the head */
 305     if (tail_ts == AV_NOPTS_VALUE || tail_ts >= st->head_ts ||
 306         av_rescale_q(st->head_ts - tail_ts, st->tb, AV_TIME_BASE_Q) < sq->buf_size_us)
 307         return 0;
 308
 309     /* signal a fake timestamp for all streams that prevent tail_ts from being output */
 310     tail_ts++;
 311     for (unsigned int i = 0; i < sq->nb_streams; i++) {
 312         const SyncQueueStream *st1 = &sq->streams[i];
 313         int64_t ts;
 314
 315         if (st == st1 || st1->finished ||
 316             (st1->head_ts != AV_NOPTS_VALUE &&
 317              av_compare_ts(tail_ts, st->tb, st1->head_ts, st1->tb) <= 0))
 318             continue;
 319
 320         ts = av_rescale_q(tail_ts, st->tb, st1->tb);
 321         if (st1->head_ts != AV_NOPTS_VALUE)
 322             ts = FFMAX(st1->head_ts + 1, ts);
 323
 324         av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u overflow heardbeat %s -> %s\n",
 325                i, av_ts2timestr(st1->head_ts, &st1->tb), av_ts2timestr(ts, &st1->tb));
 326
 327         stream_update_ts(sq, i, ts);
 328     }
 329
 330     return 1;
 331 }
 332
 333 int sq_send(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame)
 334 {
 335     SyncQueueStream *st;
 336     int64_t ts;
 337     int ret, nb_samples;
 338
 339     av_assert0(stream_idx < sq->nb_streams);
 340     st = &sq->streams[stream_idx];
 341
 342     if (frame_null(sq, frame)) {
 343         av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u EOF\n", stream_idx);
 344         finish_stream(sq, stream_idx);
 345         return 0;
 346     }
 347     if (st->finished)
 348         return AVERROR_EOF;
 349
 350     tb_update(sq, st, frame);
 351
 352     nb_samples = frame_samples(sq, frame);
 353     // make sure frame duration is consistent with sample count
 354     if (nb_samples) {
 355         av_assert0(frame.f->sample_rate > 0);
 356         frame.f->duration = av_rescale_q(nb_samples, (AVRational){ 1, frame.f->sample_rate },
 357                                          frame.f->time_base);
 358     }
 359
 360     ts = frame_end(sq, frame, 0);
 361
 362     av_log(sq->logctx, AV_LOG_DEBUG, "sq: send %u ts %s\n", stream_idx,
 363            av_ts2timestr(ts, &st->tb));
 364
 365     ret = av_container_fifo_write(st->fifo, SQPTR(sq, frame), 0);
 366     if (ret < 0)
 367         return ret;
 368
 369     stream_update_ts(sq, stream_idx, ts);
 370
 371     st->samples_queued += nb_samples;
 372     st->samples_sent   += nb_samples;
 373
 374     if (st->frame_samples)
 375         st->frames_sent = st->samples_sent / st->frame_samples;
 376     else
 377         st->frames_sent++;
 378
 379     if (st->frames_sent >= st->frames_max) {
 380         av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u frames_max %"PRIu64" reached\n",
 381                stream_idx, st->frames_max);
 382
 383         finish_stream(sq, stream_idx);
 384     }
 385
 386     return 0;
 387 }
 388
 389 static void offset_audio(AVFrame *f, int nb_samples)
 390 {
 391     const int planar = av_sample_fmt_is_planar(f->format);
 392     const int planes = planar ? f->ch_layout.nb_channels : 1;
 393     const int    bps = av_get_bytes_per_sample(f->format);
 394     const int offset = nb_samples * bps * (planar ? 1 : f->ch_layout.nb_channels);
 395
 396     av_assert0(bps > 0);
 397     av_assert0(nb_samples < f->nb_samples);
 398
 399     for (int i = 0; i < planes; i++) {
 400         f->extended_data[i] += offset;
 401         if (i < FF_ARRAY_ELEMS(f->data))
 402             f->data[i] = f->extended_data[i];
 403     }
 404     f->linesize[0] -= offset;
 405     f->nb_samples  -= nb_samples;
 406     f->duration     = av_rescale_q(f->nb_samples, (AVRational){ 1, f->sample_rate },
 407                                    f->time_base);
 408     f->pts         += av_rescale_q(nb_samples,    (AVRational){ 1, f->sample_rate },
 409                                    f->time_base);
 410 }
 411
 412 static int frame_is_aligned(const SyncQueue *sq, const AVFrame *frame)
 413 {
 414     // only checks linesize[0], so only works for audio
 415     av_assert0(frame->nb_samples > 0);
 416     av_assert0(sq->align_mask);
 417
 418     // only check data[0], because we always offset all data pointers
 419     // by the same offset, so if one is aligned, all are
 420     if (!((uintptr_t)frame->data[0] & sq->align_mask) &&
 421         !(frame->linesize[0]        & sq->align_mask) &&
 422         frame->linesize[0] > sq->align_mask)
 423         return 1;
 424
 425     return 0;
 426 }
 427
 428 static int receive_samples(SyncQueue *sq, SyncQueueStream *st,
 429                            AVFrame *dst, int nb_samples)
 430 {
 431     SyncQueueFrame src;
 432     int ret;
 433
 434     av_assert0(st->samples_queued >= nb_samples);
 435
 436     ret = av_container_fifo_peek(st->fifo, (void**)&src, 0);
 437     av_assert0(ret >= 0);
 438
 439     // peeked frame has enough samples and its data is aligned
 440     // -> we can just make a reference and limit its sample count
 441     if (src.f->nb_samples > nb_samples && frame_is_aligned(sq, src.f)) {
 442         ret = av_frame_ref(dst, src.f);
 443         if (ret < 0)
 444             return ret;
 445
 446         dst->nb_samples = nb_samples;
 447         offset_audio(src.f, nb_samples);
 448         st->samples_queued -= nb_samples;
 449
 450         goto finish;
 451     }
 452
 453     // otherwise allocate a new frame and copy the data
 454     ret = av_channel_layout_copy(&dst->ch_layout, &src.f->ch_layout);
 455     if (ret < 0)
 456         return ret;
 457
 458     dst->format     = src.f->format;
 459     dst->nb_samples = nb_samples;
 460
 461     ret = av_frame_get_buffer(dst, 0);
 462     if (ret < 0)
 463         goto fail;
 464
 465     ret = av_frame_copy_props(dst, src.f);
 466     if (ret < 0)
 467         goto fail;
 468
 469     dst->nb_samples = 0;
 470     while (dst->nb_samples < nb_samples) {
 471         int to_copy;
 472
 473         ret = av_container_fifo_peek(st->fifo, (void**)&src, 0);
 474         av_assert0(ret >= 0);
 475
 476         to_copy = FFMIN(nb_samples - dst->nb_samples, src.f->nb_samples);
 477
 478         av_samples_copy(dst->extended_data, src.f->extended_data, dst->nb_samples,
 479                         0, to_copy, dst->ch_layout.nb_channels, dst->format);
 480
 481         if (to_copy < src.f->nb_samples)
 482             offset_audio(src.f, to_copy);
 483         else
 484             av_container_fifo_drain(st->fifo, 1);
 485
 486         st->samples_queued -= to_copy;
 487
 488         dst->nb_samples += to_copy;
 489     }
 490
 491 finish:
 492     dst->duration   = av_rescale_q(nb_samples, (AVRational){ 1, dst->sample_rate },
 493                                    dst->time_base);
 494
 495     return 0;
 496
 497 fail:
 498     av_frame_unref(dst);
 499     return ret;
 500 }
 501
 502 static int receive_for_stream(SyncQueue *sq, unsigned int stream_idx,
 503                               SyncQueueFrame frame)
 504 {
 505     const SyncQueueStream *st_head = sq->head_stream >= 0 ?
 506                                      &sq->streams[sq->head_stream] : NULL;
 507     SyncQueueStream *st;
 508
 509     av_assert0(stream_idx < sq->nb_streams);
 510     st = &sq->streams[stream_idx];
 511
 512     if (av_container_fifo_can_read(st->fifo) &&
 513         (st->frame_samples <= st->samples_queued || st->finished)) {
 514         int nb_samples = st->frame_samples;
 515         SyncQueueFrame peek;
 516         int64_t ts;
 517         int cmp = 1;
 518
 519         if (st->finished)
 520             nb_samples = FFMIN(nb_samples, st->samples_queued);
 521
 522         av_container_fifo_peek(st->fifo, (void**)&peek, 0);
 523         ts = frame_end(sq, peek, nb_samples);
 524
 525         /* check if this stream's tail timestamp does not overtake
 526          * the overall queue head */
 527         if (ts != AV_NOPTS_VALUE && st_head)
 528             cmp = av_compare_ts(ts, st->tb, st_head->head_ts, st_head->tb);
 529
 530         /* We can release frames that do not end after the queue head.
 531          * Frames with no timestamps are just passed through with no conditions.
 532          * Frames are also passed through when there are no limiting streams.
 533          */
 534         if (cmp <= 0 || ts == AV_NOPTS_VALUE || !sq->have_limiting) {
 535             if (nb_samples &&
 536                 (nb_samples != peek.f->nb_samples || !frame_is_aligned(sq, peek.f))) {
 537                 int ret = receive_samples(sq, st, frame.f, nb_samples);
 538                 if (ret < 0)
 539                     return ret;
 540             } else {
 541                 int ret = av_container_fifo_read(st->fifo, SQPTR(sq, frame), 0);
 542                 av_assert0(ret >= 0);
 543
 544                 av_assert0(st->samples_queued >= frame_samples(sq, frame));
 545                 st->samples_queued -= frame_samples(sq, frame);
 546             }
 547
 548             av_log(sq->logctx, AV_LOG_DEBUG,
 549                    "sq: receive %u ts %s queue head %d ts %s\n", stream_idx,
 550                    av_ts2timestr(frame_end(sq, frame, 0), &st->tb),
 551                    sq->head_stream,
 552                    st_head ? av_ts2timestr(st_head->head_ts, &st_head->tb) : "N/A");
 553
 554             return 0;
 555         }
 556     }
 557
 558     return (sq->finished || (st->finished && !av_container_fifo_can_read(st->fifo))) ?
 559             AVERROR_EOF : AVERROR(EAGAIN);
 560 }
 561
 562 static int receive_internal(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
 563 {
 564     int nb_eof = 0;
 565     int ret;
 566
 567     /* read a frame for a specific stream */
 568     if (stream_idx >= 0) {
 569         ret = receive_for_stream(sq, stream_idx, frame);
 570         return (ret < 0) ? ret : stream_idx;
 571     }
 572
 573     /* read a frame for any stream with available output */
 574     for (unsigned int i = 0; i < sq->nb_streams; i++) {
 575         ret = receive_for_stream(sq, i, frame);
 576         if (ret == AVERROR_EOF || ret == AVERROR(EAGAIN)) {
 577             nb_eof += (ret == AVERROR_EOF);
 578             continue;
 579         }
 580         return (ret < 0) ? ret : i;
 581     }
 582
 583     return (nb_eof == sq->nb_streams) ? AVERROR_EOF : AVERROR(EAGAIN);
 584 }
 585
 586 int sq_receive(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
 587 {
 588     int ret = receive_internal(sq, stream_idx, frame);
 589
 590     /* try again if the queue overflowed and triggered a fake heartbeat
 591      * for lagging streams */
 592     if (ret == AVERROR(EAGAIN) && overflow_heartbeat(sq, stream_idx))
 593         ret = receive_internal(sq, stream_idx, frame);
 594
 595     return ret;
 596 }
 597
 598 int sq_add_stream(SyncQueue *sq, int limiting)
 599 {
 600     SyncQueueStream *tmp, *st;
 601
 602     tmp = av_realloc_array(sq->streams, sq->nb_streams + 1, sizeof(*sq->streams));
 603     if (!tmp)
 604         return AVERROR(ENOMEM);
 605     sq->streams = tmp;
 606
 607     st = &sq->streams[sq->nb_streams];
 608     memset(st, 0, sizeof(*st));
 609
 610     st->fifo = (sq->type == SYNC_QUEUE_FRAMES) ?
 611                av_container_fifo_alloc_avframe(0) : av_container_fifo_alloc_avpacket(0);
 612     if (!st->fifo)
 613         return AVERROR(ENOMEM);
 614
 615     /* we set a valid default, so that a pathological stream that never
 616      * receives even a real timebase (and no frames) won't stall all other
 617      * streams forever; cf. overflow_heartbeat() */
 618     st->tb      = (AVRational){ 1, 1 };
 619     st->head_ts = AV_NOPTS_VALUE;
 620     st->frames_max = UINT64_MAX;
 621     st->limiting   = limiting;
 622
 623     sq->have_limiting |= limiting;
 624
 625     return sq->nb_streams++;
 626 }
 627
 628 void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames)
 629 {
 630     SyncQueueStream *st;
 631
 632     av_assert0(stream_idx < sq->nb_streams);
 633     st = &sq->streams[stream_idx];
 634
 635     st->frames_max = frames;
 636     if (st->frames_sent >= st->frames_max)
 637         finish_stream(sq, stream_idx);
 638 }
 639
 640 void sq_frame_samples(SyncQueue *sq, unsigned int stream_idx,
 641                       int frame_samples)
 642 {
 643     SyncQueueStream *st;
 644
 645     av_assert0(sq->type == SYNC_QUEUE_FRAMES);
 646     av_assert0(stream_idx < sq->nb_streams);
 647     st = &sq->streams[stream_idx];
 648
 649     st->frame_samples = frame_samples;
 650
 651     sq->align_mask = av_cpu_max_align() - 1;
 652 }
 653
 654 SyncQueue *sq_alloc(enum SyncQueueType type, int64_t buf_size_us, void *logctx)
 655 {
 656     SyncQueue *sq = av_mallocz(sizeof(*sq));
 657
 658     if (!sq)
 659         return NULL;
 660
 661     sq->type                 = type;
 662     sq->buf_size_us          = buf_size_us;
 663     sq->logctx               = logctx;
 664
 665     sq->head_stream          = -1;
 666     sq->head_finished_stream = -1;
 667
 668     return sq;
 669 }
 670
 671 void sq_free(SyncQueue **psq)
 672 {
 673     SyncQueue *sq = *psq;
 674
 675     if (!sq)
 676         return;
 677
 678     for (unsigned int i = 0; i < sq->nb_streams; i++)
 679         av_container_fifo_free(&sq->streams[i].fifo);
 680
 681     av_freep(&sq->streams);
 682
 683     av_freep(psq);
 684 }