FFmpegKit iOS / macOS / tvOS API 6.0
Loading...
Searching...
No Matches
fftools_sync_queue.c
Go to the documentation of this file.
1/*
2 * This file is part of FFmpeg.
3 * Copyright (c) 2023 ARTHENICA LTD
4 *
5 * FFmpeg is free software; you can redistribute it and/or
6 * modify it under the terms of the GNU Lesser General Public
7 * License as published by the Free Software Foundation; either
8 * version 2.1 of the License, or (at your option) any later version.
9 *
10 * FFmpeg is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * Lesser General Public License for more details.
14 *
15 * You should have received a copy of the GNU Lesser General Public
16 * License along with FFmpeg; if not, write to the Free Software
17 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18 */
19
20/*
21 * This file is the modified version of sync_queue.c file living in ffmpeg source code under the fftools folder. We
22 * manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied
23 * by us to develop ffmpeg-kit library.
24 *
25 * ffmpeg-kit changes by ARTHENICA LTD
26 *
27 * 07.2023
28 * --------------------------------------------------------
29 * - FFmpeg 6.0 changes migrated
30 * - fftools header names updated
31 */
32
33#include <stdint.h>
34#include <string.h>
35
36#include "libavutil/avassert.h"
37#include "libavutil/error.h"
38#include "libavutil/fifo.h"
39#include "libavutil/mathematics.h"
40#include "libavutil/mem.h"
41
42#include "fftools_objpool.h"
43#include "fftools_sync_queue.h"
44
45typedef struct SyncQueueStream {
46 AVFifo *fifo;
47 AVRational tb;
48
49 /* stream head: largest timestamp seen */
50 int64_t head_ts;
52 /* no more frames will be sent for this stream */
54
55 uint64_t frames_sent;
56 uint64_t frames_max;
58
59struct SyncQueue {
61
62 /* no more frames will be sent for any stream */
64 /* sync head: the stream with the _smallest_ head timestamp
65 * this stream determines which frames can be output */
67 /* the finished stream with the smallest finish timestamp or -1 */
69
70 // maximum buffering duration in microseconds
71 int64_t buf_size_us;
72
74 unsigned int nb_streams;
75
76 // pool of preallocated frames to avoid constant allocations
78};
79
80static void frame_move(const SyncQueue *sq, SyncQueueFrame dst,
82{
83 if (sq->type == SYNC_QUEUE_PACKETS)
84 av_packet_move_ref(dst.p, src.p);
85 else
86 av_frame_move_ref(dst.f, src.f);
87}
88
89static int64_t frame_ts(const SyncQueue *sq, SyncQueueFrame frame)
90{
91 return (sq->type == SYNC_QUEUE_PACKETS) ?
92 frame.p->pts + frame.p->duration :
93 frame.f->pts + frame.f->duration;
94}
95
96static int frame_null(const SyncQueue *sq, SyncQueueFrame frame)
97{
98 return (sq->type == SYNC_QUEUE_PACKETS) ? (frame.p == NULL) : (frame.f == NULL);
99}
100
101static void finish_stream(SyncQueue *sq, unsigned int stream_idx)
102{
103 SyncQueueStream *st = &sq->streams[stream_idx];
104
105 st->finished = 1;
106
107 if (st->limiting && st->head_ts != AV_NOPTS_VALUE) {
108 /* check if this stream is the new finished head */
109 if (sq->head_finished_stream < 0 ||
110 av_compare_ts(st->head_ts, st->tb,
112 sq->streams[sq->head_finished_stream].tb) < 0) {
113 sq->head_finished_stream = stream_idx;
114 }
115
116 /* mark as finished all streams that should no longer receive new frames,
117 * due to them being ahead of some finished stream */
118 st = &sq->streams[sq->head_finished_stream];
119 for (unsigned int i = 0; i < sq->nb_streams; i++) {
120 SyncQueueStream *st1 = &sq->streams[i];
121 if (st != st1 && st1->head_ts != AV_NOPTS_VALUE &&
122 av_compare_ts(st->head_ts, st->tb, st1->head_ts, st1->tb) <= 0)
123 st1->finished = 1;
124 }
125 }
126
127 /* mark the whole queue as finished if all streams are finished */
128 for (unsigned int i = 0; i < sq->nb_streams; i++) {
129 if (!sq->streams[i].finished)
130 return;
131 }
132 sq->finished = 1;
133}
134
136{
137 if (sq->head_stream < 0) {
138 /* wait for one timestamp in each stream before determining
139 * the queue head */
140 for (unsigned int i = 0; i < sq->nb_streams; i++) {
141 SyncQueueStream *st = &sq->streams[i];
142 if (st->limiting && st->head_ts == AV_NOPTS_VALUE)
143 return;
144 }
145
146 // placeholder value, correct one will be found below
147 sq->head_stream = 0;
148 }
149
150 for (unsigned int i = 0; i < sq->nb_streams; i++) {
151 SyncQueueStream *st_head = &sq->streams[sq->head_stream];
152 SyncQueueStream *st_other = &sq->streams[i];
153 if (st_other->limiting && st_other->head_ts != AV_NOPTS_VALUE &&
154 av_compare_ts(st_other->head_ts, st_other->tb,
155 st_head->head_ts, st_head->tb) < 0)
156 sq->head_stream = i;
157 }
158}
159
160/* update this stream's head timestamp */
161static void stream_update_ts(SyncQueue *sq, unsigned int stream_idx, int64_t ts)
162{
163 SyncQueueStream *st = &sq->streams[stream_idx];
164
165 if (ts == AV_NOPTS_VALUE ||
166 (st->head_ts != AV_NOPTS_VALUE && st->head_ts >= ts))
167 return;
168
169 st->head_ts = ts;
170
171 /* if this stream is now ahead of some finished stream, then
172 * this stream is also finished */
173 if (sq->head_finished_stream >= 0 &&
174 av_compare_ts(sq->streams[sq->head_finished_stream].head_ts,
176 ts, st->tb) <= 0)
177 finish_stream(sq, stream_idx);
178
179 /* update the overall head timestamp if it could have changed */
180 if (st->limiting &&
181 (sq->head_stream < 0 || sq->head_stream == stream_idx))
183}
184
185/* If the queue for the given stream (or all streams when stream_idx=-1)
186 * is overflowing, trigger a fake heartbeat on lagging streams.
187 *
188 * @return 1 if heartbeat triggered, 0 otherwise
189 */
190static int overflow_heartbeat(SyncQueue *sq, int stream_idx)
191{
192 SyncQueueStream *st;
193 SyncQueueFrame frame;
194 int64_t tail_ts = AV_NOPTS_VALUE;
195
196 /* if no stream specified, pick the one that is most ahead */
197 if (stream_idx < 0) {
198 int64_t ts = AV_NOPTS_VALUE;
199
200 for (int i = 0; i < sq->nb_streams; i++) {
201 st = &sq->streams[i];
202 if (st->head_ts != AV_NOPTS_VALUE &&
203 (ts == AV_NOPTS_VALUE ||
204 av_compare_ts(ts, sq->streams[stream_idx].tb,
205 st->head_ts, st->tb) < 0)) {
206 ts = st->head_ts;
207 stream_idx = i;
208 }
209 }
210 /* no stream has a timestamp yet -> nothing to do */
211 if (stream_idx < 0)
212 return 0;
213 }
214
215 st = &sq->streams[stream_idx];
216
217 /* get the chosen stream's tail timestamp */
218 for (size_t i = 0; tail_ts == AV_NOPTS_VALUE &&
219 av_fifo_peek(st->fifo, &frame, 1, i) >= 0; i++)
220 tail_ts = frame_ts(sq, frame);
221
222 /* overflow triggers when the tail is over specified duration behind the head */
223 if (tail_ts == AV_NOPTS_VALUE || tail_ts >= st->head_ts ||
224 av_rescale_q(st->head_ts - tail_ts, st->tb, AV_TIME_BASE_Q) < sq->buf_size_us)
225 return 0;
226
227 /* signal a fake timestamp for all streams that prevent tail_ts from being output */
228 tail_ts++;
229 for (unsigned int i = 0; i < sq->nb_streams; i++) {
230 SyncQueueStream *st1 = &sq->streams[i];
231 int64_t ts;
232
233 if (st == st1 || st1->finished ||
234 (st1->head_ts != AV_NOPTS_VALUE &&
235 av_compare_ts(tail_ts, st->tb, st1->head_ts, st1->tb) <= 0))
236 continue;
237
238 ts = av_rescale_q(tail_ts, st->tb, st1->tb);
239 if (st1->head_ts != AV_NOPTS_VALUE)
240 ts = FFMAX(st1->head_ts + 1, ts);
241
242 stream_update_ts(sq, i, ts);
243 }
244
245 return 1;
246}
247
248int sq_send(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame)
249{
250 SyncQueueStream *st;
251 SyncQueueFrame dst;
252 int64_t ts;
253 int ret;
254
255 av_assert0(stream_idx < sq->nb_streams);
256 st = &sq->streams[stream_idx];
257
258 av_assert0(st->tb.num > 0 && st->tb.den > 0);
259
260 if (frame_null(sq, frame)) {
261 finish_stream(sq, stream_idx);
262 return 0;
263 }
264 if (st->finished)
265 return AVERROR_EOF;
266
267 ret = objpool_get(sq->pool, (void**)&dst);
268 if (ret < 0)
269 return ret;
270
271 frame_move(sq, dst, frame);
272
273 ts = frame_ts(sq, dst);
274
275 ret = av_fifo_write(st->fifo, &dst, 1);
276 if (ret < 0) {
277 frame_move(sq, frame, dst);
278 objpool_release(sq->pool, (void**)&dst);
279 return ret;
280 }
281
282 stream_update_ts(sq, stream_idx, ts);
283
284 st->frames_sent++;
285 if (st->frames_sent >= st->frames_max)
286 finish_stream(sq, stream_idx);
287
288 return 0;
289}
290
291static int receive_for_stream(SyncQueue *sq, unsigned int stream_idx,
292 SyncQueueFrame frame)
293{
294 SyncQueueStream *st_head = sq->head_stream >= 0 ?
295 &sq->streams[sq->head_stream] : NULL;
296 SyncQueueStream *st;
297
298 av_assert0(stream_idx < sq->nb_streams);
299 st = &sq->streams[stream_idx];
300
301 if (av_fifo_can_read(st->fifo)) {
302 SyncQueueFrame peek;
303 int64_t ts;
304 int cmp = 1;
305
306 av_fifo_peek(st->fifo, &peek, 1, 0);
307 ts = frame_ts(sq, peek);
308
309 /* check if this stream's tail timestamp does not overtake
310 * the overall queue head */
311 if (ts != AV_NOPTS_VALUE && st_head)
312 cmp = av_compare_ts(ts, st->tb, st_head->head_ts, st_head->tb);
313
314 /* We can release frames that do not end after the queue head.
315 * Frames with no timestamps are just passed through with no conditions.
316 */
317 if (cmp <= 0 || ts == AV_NOPTS_VALUE) {
318 frame_move(sq, frame, peek);
319 objpool_release(sq->pool, (void**)&peek);
320 av_fifo_drain2(st->fifo, 1);
321 return 0;
322 }
323 }
324
325 return (sq->finished || (st->finished && !av_fifo_can_read(st->fifo))) ?
326 AVERROR_EOF : AVERROR(EAGAIN);
327}
328
329static int receive_internal(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
330{
331 int nb_eof = 0;
332 int ret;
333
334 /* read a frame for a specific stream */
335 if (stream_idx >= 0) {
336 ret = receive_for_stream(sq, stream_idx, frame);
337 return (ret < 0) ? ret : stream_idx;
338 }
339
340 /* read a frame for any stream with available output */
341 for (unsigned int i = 0; i < sq->nb_streams; i++) {
342 ret = receive_for_stream(sq, i, frame);
343 if (ret == AVERROR_EOF || ret == AVERROR(EAGAIN)) {
344 nb_eof += (ret == AVERROR_EOF);
345 continue;
346 }
347 return (ret < 0) ? ret : i;
348 }
349
350 return (nb_eof == sq->nb_streams) ? AVERROR_EOF : AVERROR(EAGAIN);
351}
352
353int sq_receive(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
354{
355 int ret = receive_internal(sq, stream_idx, frame);
356
357 /* try again if the queue overflowed and triggered a fake heartbeat
358 * for lagging streams */
359 if (ret == AVERROR(EAGAIN) && overflow_heartbeat(sq, stream_idx))
360 ret = receive_internal(sq, stream_idx, frame);
361
362 return ret;
363}
364
365int sq_add_stream(SyncQueue *sq, int limiting)
366{
367 SyncQueueStream *tmp, *st;
368
369 tmp = av_realloc_array(sq->streams, sq->nb_streams + 1, sizeof(*sq->streams));
370 if (!tmp)
371 return AVERROR(ENOMEM);
372 sq->streams = tmp;
373
374 st = &sq->streams[sq->nb_streams];
375 memset(st, 0, sizeof(*st));
376
377 st->fifo = av_fifo_alloc2(1, sizeof(SyncQueueFrame), AV_FIFO_FLAG_AUTO_GROW);
378 if (!st->fifo)
379 return AVERROR(ENOMEM);
380
381 /* we set a valid default, so that a pathological stream that never
382 * receives even a real timebase (and no frames) won't stall all other
383 * streams forever; cf. overflow_heartbeat() */
384 st->tb = (AVRational){ 1, 1 };
385 st->head_ts = AV_NOPTS_VALUE;
386 st->frames_max = UINT64_MAX;
387 st->limiting = limiting;
388
389 return sq->nb_streams++;
390}
391
392void sq_set_tb(SyncQueue *sq, unsigned int stream_idx, AVRational tb)
393{
394 SyncQueueStream *st;
395
396 av_assert0(stream_idx < sq->nb_streams);
397 st = &sq->streams[stream_idx];
398
399 av_assert0(!av_fifo_can_read(st->fifo));
400
401 if (st->head_ts != AV_NOPTS_VALUE)
402 st->head_ts = av_rescale_q(st->head_ts, st->tb, tb);
403
404 st->tb = tb;
405}
406
407void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames)
408{
409 SyncQueueStream *st;
410
411 av_assert0(stream_idx < sq->nb_streams);
412 st = &sq->streams[stream_idx];
413
414 st->frames_max = frames;
415 if (st->frames_sent >= st->frames_max)
416 finish_stream(sq, stream_idx);
417}
418
419SyncQueue *sq_alloc(enum SyncQueueType type, int64_t buf_size_us)
420{
421 SyncQueue *sq = av_mallocz(sizeof(*sq));
422
423 if (!sq)
424 return NULL;
425
426 sq->type = type;
427 sq->buf_size_us = buf_size_us;
428
429 sq->head_stream = -1;
430 sq->head_finished_stream = -1;
431
432 sq->pool = (type == SYNC_QUEUE_PACKETS) ? objpool_alloc_packets() :
434 if (!sq->pool) {
435 av_freep(&sq);
436 return NULL;
437 }
438
439 return sq;
440}
441
443{
444 SyncQueue *sq = *psq;
445
446 if (!sq)
447 return;
448
449 for (unsigned int i = 0; i < sq->nb_streams; i++) {
450 SyncQueueFrame frame;
451 while (av_fifo_read(sq->streams[i].fifo, &frame, 1) >= 0)
452 objpool_release(sq->pool, (void**)&frame);
453
454 av_fifo_freep2(&sq->streams[i].fifo);
455 }
456
457 av_freep(&sq->streams);
458
459 objpool_free(&sq->pool);
460
461 av_freep(psq);
462}
__thread int nb_streams
ObjPool * objpool_alloc_packets(void)
ObjPool * objpool_alloc_frames(void)
void objpool_release(ObjPool *op, void **obj)
void objpool_free(ObjPool **pop)
int objpool_get(ObjPool *op, void **obj)
static int64_t frame_ts(const SyncQueue *sq, SyncQueueFrame frame)
SyncQueue * sq_alloc(enum SyncQueueType type, int64_t buf_size_us)
static void queue_head_update(SyncQueue *sq)
static void finish_stream(SyncQueue *sq, unsigned int stream_idx)
int sq_send(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame)
static int receive_for_stream(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame)
void sq_set_tb(SyncQueue *sq, unsigned int stream_idx, AVRational tb)
static void stream_update_ts(SyncQueue *sq, unsigned int stream_idx, int64_t ts)
static int frame_null(const SyncQueue *sq, SyncQueueFrame frame)
void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames)
int sq_receive(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
static int overflow_heartbeat(SyncQueue *sq, int stream_idx)
static void frame_move(const SyncQueue *sq, SyncQueueFrame dst, SyncQueueFrame src)
static int receive_internal(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
void sq_free(SyncQueue **psq)
int sq_add_stream(SyncQueue *sq, int limiting)
SyncQueueType
@ SYNC_QUEUE_PACKETS
SyncQueueStream * streams
enum SyncQueueType type
int64_t buf_size_us
unsigned int nb_streams