FFmpeg coverage

Directory: ../../../ffmpeg/
File: src/fftools/sync_queue.c
Date: 2023-06-04 16:45:34
Exec Total Coverage
Lines: 312 329 94.8%
Functions: 21 21 100.0%
Branches: 227 266 85.3%
Line Branch Exec Source
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/channel_layout.h"
24 #include "libavutil/cpu.h"
25 #include "libavutil/error.h"
26 #include "libavutil/fifo.h"
27 #include "libavutil/mathematics.h"
28 #include "libavutil/mem.h"
29 #include "libavutil/samplefmt.h"
30 #include "libavutil/timestamp.h"
31
32 #include "objpool.h"
33 #include "sync_queue.h"
34
35 /*
36 * How this works:
37 * --------------
38 * time: 0 1 2 3 4 5 6 7 8 9 10 11 12 13
39 * -------------------------------------------------------------------
40 * | | | | | | | | | | | | | |
41 * | ┌───┐┌────────┐┌───┐┌─────────────┐
42 * stream 0| │d=1││ d=2 ││d=1││ d=3 │
43 * | └───┘└────────┘└───┘└─────────────┘
44 * ┌───┐ ┌───────────────────────┐
45 * stream 1│d=1│ │ d=5 │
46 * └───┘ └───────────────────────┘
47 * | ┌───┐┌───┐┌───┐┌───┐
48 * stream 2| │d=1││d=1││d=1││d=1│ <- stream 2 is the head stream of the queue
49 * | └───┘└───┘└───┘└───┘
50 * ^ ^
51 * [stream 2 tail] [stream 2 head]
52 *
53 * We have N streams (N=3 in the diagram), each stream is a FIFO. The *tail* of
54 * each FIFO is the frame with smallest end time, the *head* is the frame with
55 * the largest end time. Frames submitted to the queue with sq_send() are placed
56 * after the head, frames returned to the caller with sq_receive() are taken
57 * from the tail.
58 *
59 * The head stream of the whole queue (SyncQueue.head_stream) is the limiting
60 * stream with the *smallest* head timestamp, i.e. the stream whose source lags
61 * furthest behind all other streams. It determines which frames can be output
62 * from the queue.
63 *
64 * In the diagram, the head stream is 2, because it head time is t=5, while
65 * streams 0 and 1 end at t=8 and t=9 respectively. All frames that _end_ at
66 * or before t=5 can be output, i.e. the first 3 frames from stream 0, first
67 * frame from stream 1, and all 4 frames from stream 2.
68 */
69
70 typedef struct SyncQueueStream {
71 AVFifo *fifo;
72 AVRational tb;
73
74 /* number of audio samples in fifo */
75 uint64_t samples_queued;
76 /* stream head: largest timestamp seen */
77 int64_t head_ts;
78 int limiting;
79 /* no more frames will be sent for this stream */
80 int finished;
81
82 uint64_t frames_sent;
83 uint64_t samples_sent;
84 uint64_t frames_max;
85 int frame_samples;
86 } SyncQueueStream;
87
88 struct SyncQueue {
89 enum SyncQueueType type;
90
91 void *logctx;
92
93 /* no more frames will be sent for any stream */
94 int finished;
95 /* sync head: the stream with the _smallest_ head timestamp
96 * this stream determines which frames can be output */
97 int head_stream;
98 /* the finished stream with the smallest finish timestamp or -1 */
99 int head_finished_stream;
100
101 // maximum buffering duration in microseconds
102 int64_t buf_size_us;
103
104 SyncQueueStream *streams;
105 unsigned int nb_streams;
106
107 // pool of preallocated frames to avoid constant allocations
108 ObjPool *pool;
109
110 int have_limiting;
111
112 uintptr_t align_mask;
113 };
114
115 79591 static void frame_move(const SyncQueue *sq, SyncQueueFrame dst,
116 SyncQueueFrame src)
117 {
118
2/2
✓ Branch 0 taken 1528 times.
✓ Branch 1 taken 78063 times.
 79591 if (sq->type == SYNC_QUEUE_PACKETS)
119 1528 av_packet_move_ref(dst.p, src.p);
120 else
121 78063 av_frame_move_ref(dst.f, src.f);
122 79591 }
123
124 /**
125 * Compute the end timestamp of a frame. If nb_samples is provided, consider
126 * the frame to have this number of audio samples, otherwise use frame duration.
127 */
128 260897 static int64_t frame_end(const SyncQueue *sq, SyncQueueFrame frame, int nb_samples)
129 {
130
2/2
✓ Branch 0 taken 70247 times.
✓ Branch 1 taken 190650 times.
 260897 if (nb_samples) {
131 70247 int64_t d = av_rescale_q(nb_samples, (AVRational){ 1, frame.f->sample_rate},
132 70247 frame.f->time_base);
133 70247 return frame.f->pts + d;
134 }
135
136 190650 return (sq->type == SYNC_QUEUE_PACKETS) ?
137
2/2
✓ Branch 0 taken 4053 times.
✓ Branch 1 taken 186597 times.
 377247 frame.p->pts + frame.p->duration :
138 186597 frame.f->pts + frame.f->duration;
139 }
140
141 99853 static int frame_samples(const SyncQueue *sq, SyncQueueFrame frame)
142 {
143
2/2
✓ Branch 0 taken 97585 times.
✓ Branch 1 taken 2268 times.
 99853 return (sq->type == SYNC_QUEUE_PACKETS) ? 0 : frame.f->nb_samples;
144 }
145
146 70410 static int frame_null(const SyncQueue *sq, SyncQueueFrame frame)
147 {
148
2/2
✓ Branch 0 taken 838 times.
✓ Branch 1 taken 69572 times.
 70410 return (sq->type == SYNC_QUEUE_PACKETS) ? (frame.p == NULL) : (frame.f == NULL);
149 }
150
151 59329 static void tb_update(const SyncQueue *sq, SyncQueueStream *st,
152 const SyncQueueFrame frame)
153 {
154 118658 AVRational tb = (sq->type == SYNC_QUEUE_PACKETS) ?
155
2/2
✓ Branch 0 taken 788 times.
✓ Branch 1 taken 58541 times.
 59329 frame.p->time_base : frame.f->time_base;
156
157
2/4
✓ Branch 0 taken 59329 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 59329 times.
 59329 av_assert0(tb.num > 0 && tb.den > 0);
158
159
4/4
✓ Branch 0 taken 59214 times.
✓ Branch 1 taken 115 times.
✓ Branch 2 taken 56551 times.
✓ Branch 3 taken 2663 times.
 59329 if (tb.num == st->tb.num && tb.den == st->tb.den)
160 56551 return;
161
162 // timebase should not change after the first frame
163
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2778 times.
 2778 av_assert0(!av_fifo_can_read(st->fifo));
164
165
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 2777 times.
 2778 if (st->head_ts != AV_NOPTS_VALUE)
166 1 st->head_ts = av_rescale_q(st->head_ts, st->tb, tb);
167
168 2778 st->tb = tb;
169 }
170
171 13688 static void finish_stream(SyncQueue *sq, unsigned int stream_idx)
172 {
173 13688 SyncQueueStream *st = &sq->streams[stream_idx];
174
175
2/2
✓ Branch 0 taken 2789 times.
✓ Branch 1 taken 10899 times.
 13688 if (!st->finished)
176 2789 av_log(sq->logctx, AV_LOG_DEBUG,
177 "sq: finish %u; head ts %s\n", stream_idx,
178 2789 av_ts2timestr(st->head_ts, &st->tb));
179
180 13688 st->finished = 1;
181
182
4/4
✓ Branch 0 taken 13157 times.
✓ Branch 1 taken 531 times.
✓ Branch 2 taken 13143 times.
✓ Branch 3 taken 14 times.
 13688 if (st->limiting && st->head_ts != AV_NOPTS_VALUE) {
183 /* check if this stream is the new finished head */
184
4/4
✓ Branch 0 taken 10509 times.
✓ Branch 1 taken 2634 times.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 10508 times.
 23652 if (sq->head_finished_stream < 0 ||
185 10509 av_compare_ts(st->head_ts, st->tb,
186 10509 sq->streams[sq->head_finished_stream].head_ts,
187 10509 sq->streams[sq->head_finished_stream].tb) < 0) {
188 2635 sq->head_finished_stream = stream_idx;
189 }
190
191 /* mark as finished all streams that should no longer receive new frames,
192 * due to them being ahead of some finished stream */
193 13143 st = &sq->streams[sq->head_finished_stream];
194
2/2
✓ Branch 0 taken 13170 times.
✓ Branch 1 taken 13143 times.
 26313 for (unsigned int i = 0; i < sq->nb_streams; i++) {
195 13170 SyncQueueStream *st1 = &sq->streams[i];
196
5/6
✓ Branch 0 taken 27 times.
✓ Branch 1 taken 13143 times.
✓ Branch 2 taken 27 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 21 times.
✓ Branch 5 taken 6 times.
 13197 if (st != st1 && st1->head_ts != AV_NOPTS_VALUE &&
197 27 av_compare_ts(st->head_ts, st->tb, st1->head_ts, st1->tb) <= 0) {
198
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 18 times.
 21 if (!st1->finished)
199 3 av_log(sq->logctx, AV_LOG_DEBUG,
200 "sq: finish secondary %u; head ts %s\n", i,
201 3 av_ts2timestr(st1->head_ts, &st1->tb));
202
203 21 st1->finished = 1;
204 }
205 }
206 }
207
208 /* mark the whole queue as finished if all streams are finished */
209
2/2
✓ Branch 0 taken 13899 times.
✓ Branch 1 taken 13661 times.
 27560 for (unsigned int i = 0; i < sq->nb_streams; i++) {
210
2/2
✓ Branch 0 taken 27 times.
✓ Branch 1 taken 13872 times.
 13899 if (!sq->streams[i].finished)
211 27 return;
212 }
213 13661 sq->finished = 1;
214
215 13661 av_log(sq->logctx, AV_LOG_DEBUG, "sq: finish queue\n");
216 }
217
218 11027 static void queue_head_update(SyncQueue *sq)
219 {
220
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 11027 times.
 11027 av_assert0(sq->have_limiting);
221
222
2/2
✓ Branch 0 taken 2680 times.
✓ Branch 1 taken 8347 times.
 11027 if (sq->head_stream < 0) {
223 2680 unsigned first_limiting = UINT_MAX;
224
225 /* wait for one timestamp in each stream before determining
226 * the queue head */
227
2/2
✓ Branch 0 taken 2734 times.
✓ Branch 1 taken 2634 times.
 5368 for (unsigned int i = 0; i < sq->nb_streams; i++) {
228 2734 SyncQueueStream *st = &sq->streams[i];
229
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 2730 times.
 2734 if (!st->limiting)
230 4 continue;
231
2/2
✓ Branch 0 taken 46 times.
✓ Branch 1 taken 2684 times.
 2730 if (st->head_ts == AV_NOPTS_VALUE)
232 46 return;
233
2/2
✓ Branch 0 taken 2680 times.
✓ Branch 1 taken 4 times.
 2684 if (first_limiting == UINT_MAX)
234 2680 first_limiting = i;
235 }
236
237 // placeholder value, correct one will be found below
238
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2634 times.
 2634 av_assert0(first_limiting < UINT_MAX);
239 2634 sq->head_stream = first_limiting;
240 }
241
242
2/2
✓ Branch 0 taken 11372 times.
✓ Branch 1 taken 10981 times.
 22353 for (unsigned int i = 0; i < sq->nb_streams; i++) {
243 11372 SyncQueueStream *st_head = &sq->streams[sq->head_stream];
244 11372 SyncQueueStream *st_other = &sq->streams[i];
245
5/6
✓ Branch 0 taken 11338 times.
✓ Branch 1 taken 34 times.
✓ Branch 2 taken 11338 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 35 times.
✓ Branch 5 taken 11303 times.
 22710 if (st_other->limiting && st_other->head_ts != AV_NOPTS_VALUE &&
246 11338 av_compare_ts(st_other->head_ts, st_other->tb,
247 st_head->head_ts, st_head->tb) < 0)
248 35 sq->head_stream = i;
249 }
250 }
251
252 /* update this stream's head timestamp */
253 59423 static void stream_update_ts(SyncQueue *sq, unsigned int stream_idx, int64_t ts)
254 {
255 59423 SyncQueueStream *st = &sq->streams[stream_idx];
256
257
2/2
✓ Branch 0 taken 59420 times.
✓ Branch 1 taken 3 times.
 59423 if (ts == AV_NOPTS_VALUE ||
258
4/4
✓ Branch 0 taken 56645 times.
✓ Branch 1 taken 2775 times.
✓ Branch 2 taken 88 times.
✓ Branch 3 taken 56557 times.
 59420 (st->head_ts != AV_NOPTS_VALUE && st->head_ts >= ts))
259 91 return;
260
261 59332 st->head_ts = ts;
262
263 /* if this stream is now ahead of some finished stream, then
264 * this stream is also finished */
265
4/4
✓ Branch 0 taken 33 times.
✓ Branch 1 taken 59299 times.
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 29 times.
 59365 if (sq->head_finished_stream >= 0 &&
266 33 av_compare_ts(sq->streams[sq->head_finished_stream].head_ts,
267 33 sq->streams[sq->head_finished_stream].tb,
268 ts, st->tb) <= 0)
269 4 finish_stream(sq, stream_idx);
270
271 /* update the overall head timestamp if it could have changed */
272
2/2
✓ Branch 0 taken 11516 times.
✓ Branch 1 taken 47816 times.
 59332 if (st->limiting &&
273
4/4
✓ Branch 0 taken 8836 times.
✓ Branch 1 taken 2680 times.
✓ Branch 2 taken 8347 times.
✓ Branch 3 taken 489 times.
 11516 (sq->head_stream < 0 || sq->head_stream == stream_idx))
274 11027 queue_head_update(sq);
275 }
276
277 /* If the queue for the given stream (or all streams when stream_idx=-1)
278 * is overflowing, trigger a fake heartbeat on lagging streams.
279 *
280 * @return 1 if heartbeat triggered, 0 otherwise
281 */
282 56802 static int overflow_heartbeat(SyncQueue *sq, int stream_idx)
283 {
284 SyncQueueStream *st;
285 SyncQueueFrame frame;
286 56802 int64_t tail_ts = AV_NOPTS_VALUE;
287
288 /* if no stream specified, pick the one that is most ahead */
289
2/2
✓ Branch 0 taken 885 times.
✓ Branch 1 taken 55917 times.
 56802 if (stream_idx < 0) {
290 885 int64_t ts = AV_NOPTS_VALUE;
291
292
2/2
✓ Branch 0 taken 1797 times.
✓ Branch 1 taken 885 times.
 2682 for (int i = 0; i < sq->nb_streams; i++) {
293 1797 st = &sq->streams[i];
294
4/4
✓ Branch 0 taken 1696 times.
✓ Branch 1 taken 101 times.
✓ Branch 2 taken 812 times.
✓ Branch 3 taken 884 times.
 1797 if (st->head_ts != AV_NOPTS_VALUE &&
295
2/2
✓ Branch 0 taken 52 times.
✓ Branch 1 taken 760 times.
 812 (ts == AV_NOPTS_VALUE ||
296 812 av_compare_ts(ts, sq->streams[stream_idx].tb,
297 st->head_ts, st->tb) < 0)) {
298 936 ts = st->head_ts;
299 936 stream_idx = i;
300 }
301 }
302 /* no stream has a timestamp yet -> nothing to do */
303
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 884 times.
 885 if (stream_idx < 0)
304 1 return 0;
305 }
306
307 56801 st = &sq->streams[stream_idx];
308
309 /* get the chosen stream's tail timestamp */
310
4/4
✓ Branch 0 taken 56801 times.
✓ Branch 1 taken 36786 times.
✓ Branch 2 taken 36786 times.
✓ Branch 3 taken 20015 times.
 150388 for (size_t i = 0; tail_ts == AV_NOPTS_VALUE &&
311 93587 av_fifo_peek(st->fifo, &frame, 1, i) >= 0; i++)
312 36786 tail_ts = frame_end(sq, frame, 0);
313
314 /* overflow triggers when the tail is over specified duration behind the head */
315
4/4
✓ Branch 0 taken 36786 times.
✓ Branch 1 taken 20015 times.
✓ Branch 2 taken 17121 times.
✓ Branch 3 taken 19665 times.
 56801 if (tail_ts == AV_NOPTS_VALUE || tail_ts >= st->head_ts ||
316
2/2
✓ Branch 0 taken 17027 times.
✓ Branch 1 taken 94 times.
 17121 av_rescale_q(st->head_ts - tail_ts, st->tb, AV_TIME_BASE_Q) < sq->buf_size_us)
317 56707 return 0;
318
319 /* signal a fake timestamp for all streams that prevent tail_ts from being output */
320 94 tail_ts++;
321
2/2
✓ Branch 0 taken 188 times.
✓ Branch 1 taken 94 times.
 282 for (unsigned int i = 0; i < sq->nb_streams; i++) {
322 188 SyncQueueStream *st1 = &sq->streams[i];
323 int64_t ts;
324
325
3/4
✓ Branch 0 taken 94 times.
✓ Branch 1 taken 94 times.
✓ Branch 2 taken 94 times.
✗ Branch 3 not taken.
 188 if (st == st1 || st1->finished ||
326
3/4
✓ Branch 0 taken 93 times.
✓ Branch 1 taken 1 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 93 times.
 187 (st1->head_ts != AV_NOPTS_VALUE &&
327 93 av_compare_ts(tail_ts, st->tb, st1->head_ts, st1->tb) <= 0))
328 94 continue;
329
330 94 ts = av_rescale_q(tail_ts, st->tb, st1->tb);
331
2/2
✓ Branch 0 taken 93 times.
✓ Branch 1 taken 1 times.
 94 if (st1->head_ts != AV_NOPTS_VALUE)
332
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 93 times.
 93 ts = FFMAX(st1->head_ts + 1, ts);
333
334 94 av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u overflow heardbeat %s -> %s\n",
335 94 i, av_ts2timestr(st1->head_ts, &st1->tb), av_ts2timestr(ts, &st1->tb));
336
337 94 stream_update_ts(sq, i, ts);
338 }
339
340 94 return 1;
341 }
342
343 70410 int sq_send(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame)
344 {
345 SyncQueueStream *st;
346 SyncQueueFrame dst;
347 int64_t ts;
348 int ret, nb_samples;
349
350
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 70410 times.
 70410 av_assert0(stream_idx < sq->nb_streams);
351 70410 st = &sq->streams[stream_idx];
352
353
2/2
✓ Branch 1 taken 11044 times.
✓ Branch 2 taken 59366 times.
 70410 if (frame_null(sq, frame)) {
354 11044 av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u EOF\n", stream_idx);
355 11044 finish_stream(sq, stream_idx);
356 11044 return 0;
357 }
358
2/2
✓ Branch 0 taken 37 times.
✓ Branch 1 taken 59329 times.
 59366 if (st->finished)
359 37 return AVERROR_EOF;
360
361 59329 tb_update(sq, st, frame);
362
363 59329 ret = objpool_get(sq->pool, (void**)&dst);
364
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 59329 times.
 59329 if (ret < 0)
365 ✗ return ret;
366
367 59329 frame_move(sq, dst, frame);
368
369 59329 nb_samples = frame_samples(sq, dst);
370 // make sure frame duration is consistent with sample count
371
2/2
✓ Branch 0 taken 51761 times.
✓ Branch 1 taken 7568 times.
 59329 if (nb_samples) {
372
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 51761 times.
 51761 av_assert0(dst.f->sample_rate > 0);
373 51761 dst.f->duration = av_rescale_q(nb_samples, (AVRational){ 1, dst.f->sample_rate },
374 51761 dst.f->time_base);
375 }
376
377 59329 ts = frame_end(sq, dst, 0);
378
379 59329 av_log(sq->logctx, AV_LOG_DEBUG, "sq: send %u ts %s\n", stream_idx,
380 59329 av_ts2timestr(ts, &st->tb));
381
382 59329 ret = av_fifo_write(st->fifo, &dst, 1);
383
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 59329 times.
 59329 if (ret < 0) {
384 ✗ frame_move(sq, frame, dst);
385 ✗ objpool_release(sq->pool, (void**)&dst);
386 ✗ return ret;
387 }
388
389 59329 stream_update_ts(sq, stream_idx, ts);
390
391 59329 st->samples_queued += nb_samples;
392 59329 st->samples_sent += nb_samples;
393
394
2/2
✓ Branch 0 taken 47643 times.
✓ Branch 1 taken 11686 times.
 59329 if (st->frame_samples)
395 47643 st->frames_sent = st->samples_sent / st->frame_samples;
396 else
397 11686 st->frames_sent++;
398
399
2/2
✓ Branch 0 taken 2626 times.
✓ Branch 1 taken 56703 times.
 59329 if (st->frames_sent >= st->frames_max) {
400 2626 av_log(sq->logctx, AV_LOG_DEBUG, "sq: %u frames_max %"PRIu64" reached\n",
401 stream_idx, st->frames_max);
402
403 2626 finish_stream(sq, stream_idx);
404 }
405
406 59329 return 0;
407 }
408
409 58406 static void offset_audio(AVFrame *f, int nb_samples)
410 {
411 58406 const int planar = av_sample_fmt_is_planar(f->format);
412
2/2
✓ Branch 0 taken 25044 times.
✓ Branch 1 taken 33362 times.
 58406 const int planes = planar ? f->ch_layout.nb_channels : 1;
413 58406 const int bps = av_get_bytes_per_sample(f->format);
414
2/2
✓ Branch 0 taken 33362 times.
✓ Branch 1 taken 25044 times.
 58406 const int offset = nb_samples * bps * (planar ? 1 : f->ch_layout.nb_channels);
415
416
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 58406 times.
 58406 av_assert0(bps > 0);
417
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 58406 times.
 58406 av_assert0(nb_samples < f->nb_samples);
418
419
2/2
✓ Branch 0 taken 83392 times.
✓ Branch 1 taken 58406 times.
 141798 for (int i = 0; i < planes; i++) {
420 83392 f->extended_data[i] += offset;
421
1/2
✓ Branch 0 taken 83392 times.
✗ Branch 1 not taken.
 83392 if (i < FF_ARRAY_ELEMS(f->data))
422 83392 f->data[i] = f->extended_data[i];
423 }
424 58406 f->linesize[0] -= offset;
425 58406 f->nb_samples -= nb_samples;
426 58406 f->duration = av_rescale_q(f->nb_samples, (AVRational){ 1, f->sample_rate },
427 f->time_base);
428 58406 f->pts += av_rescale_q(nb_samples, (AVRational){ 1, f->sample_rate },
429 f->time_base);
430 58406 }
431
432 50757 static int frame_is_aligned(const SyncQueue *sq, const AVFrame *frame)
433 {
434 // only checks linesize[0], so only works for audio
435
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 50757 times.
 50757 av_assert0(frame->nb_samples > 0);
436
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 50757 times.
 50757 av_assert0(sq->align_mask);
437
438 // only check data[0], because we always offset all data pointers
439 // by the same offset, so if one is aligned, all are
440
2/2
✓ Branch 0 taken 50598 times.
✓ Branch 1 taken 159 times.
 50757 if (!((uintptr_t)frame->data[0] & sq->align_mask) &&
441
1/2
✓ Branch 0 taken 50598 times.
✗ Branch 1 not taken.
 50598 !(frame->linesize[0] & sq->align_mask) &&
442
1/2
✓ Branch 0 taken 50598 times.
✗ Branch 1 not taken.
 50598 frame->linesize[0] > sq->align_mask)
443 50598 return 1;
444
445 159 return 0;
446 }
447
448 61609 static int receive_samples(SyncQueue *sq, SyncQueueStream *st,
449 AVFrame *dst, int nb_samples)
450 {
451 SyncQueueFrame src;
452 int ret;
453
454
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 61609 times.
 61609 av_assert0(st->samples_queued >= nb_samples);
455
456 61609 ret = av_fifo_peek(st->fifo, &src, 1, 0);
457
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 61609 times.
 61609 av_assert0(ret >= 0);
458
459 // peeked frame has enough samples and its data is aligned
460 // -> we can just make a reference and limit its sample count
461
4/4
✓ Branch 0 taken 42132 times.
✓ Branch 1 taken 19477 times.
✓ Branch 3 taken 41974 times.
✓ Branch 4 taken 158 times.
 61609 if (src.f->nb_samples > nb_samples && frame_is_aligned(sq, src.f)) {
462 41974 ret = av_frame_ref(dst, src.f);
463
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 41974 times.
 41974 if (ret < 0)
464 ✗ return ret;
465
466 41974 dst->nb_samples = nb_samples;
467 41974 offset_audio(src.f, nb_samples);
468 41974 st->samples_queued -= nb_samples;
469
470 41974 goto finish;
471 }
472
473 // otherwise allocate a new frame and copy the data
474 19635 ret = av_channel_layout_copy(&dst->ch_layout, &src.f->ch_layout);
475
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19635 times.
 19635 if (ret < 0)
476 ✗ return ret;
477
478 19635 dst->format = src.f->format;
479 19635 dst->nb_samples = nb_samples;
480
481 19635 ret = av_frame_get_buffer(dst, 0);
482
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19635 times.
 19635 if (ret < 0)
483 ✗ goto fail;
484
485 19635 ret = av_frame_copy_props(dst, src.f);
486
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19635 times.
 19635 if (ret < 0)
487 ✗ goto fail;
488
489 19635 dst->nb_samples = 0;
490
2/2
✓ Branch 0 taken 55449 times.
✓ Branch 1 taken 19635 times.
 75084 while (dst->nb_samples < nb_samples) {
491 int to_copy;
492
493 55449 ret = av_fifo_peek(st->fifo, &src, 1, 0);
494
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 55449 times.
 55449 av_assert0(ret >= 0);
495
496 55449 to_copy = FFMIN(nb_samples - dst->nb_samples, src.f->nb_samples);
497
498 55449 av_samples_copy(dst->extended_data, src.f->extended_data, dst->nb_samples,
499 55449 0, to_copy, dst->ch_layout.nb_channels, dst->format);
500
501
2/2
✓ Branch 0 taken 16432 times.
✓ Branch 1 taken 39017 times.
 55449 if (to_copy < src.f->nb_samples)
502 16432 offset_audio(src.f, to_copy);
503 else {
504 39017 av_frame_unref(src.f);
505 39017 objpool_release(sq->pool, (void**)&src);
506 39017 av_fifo_drain2(st->fifo, 1);
507 }
508 55449 st->samples_queued -= to_copy;
509
510 55449 dst->nb_samples += to_copy;
511 }
512
513 19635 finish:
514 61609 dst->duration = av_rescale_q(nb_samples, (AVRational){ 1, dst->sample_rate },
515 dst->time_base);
516
517 61609 return 0;
518
519 ✗ fail:
520 ✗ av_frame_unref(dst);
521 ✗ return ret;
522 }
523
524 145216 static int receive_for_stream(SyncQueue *sq, unsigned int stream_idx,
525 SyncQueueFrame frame)
526 {
527 290432 SyncQueueStream *st_head = sq->head_stream >= 0 ?
528
2/2
✓ Branch 0 taken 26675 times.
✓ Branch 1 taken 118541 times.
 145216 &sq->streams[sq->head_stream] : NULL;
529 SyncQueueStream *st;
530
531
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 145216 times.
 145216 av_assert0(stream_idx < sq->nb_streams);
532 145216 st = &sq->streams[stream_idx];
533
534
2/2
✓ Branch 1 taken 118826 times.
✓ Branch 2 taken 26390 times.
 145216 if (av_fifo_can_read(st->fifo) &&
535
4/4
✓ Branch 0 taken 36022 times.
✓ Branch 1 taken 82804 times.
✓ Branch 2 taken 107 times.
✓ Branch 3 taken 35915 times.
 118826 (st->frame_samples <= st->samples_queued || st->finished)) {
536 82911 int nb_samples = st->frame_samples;
537 SyncQueueFrame peek;
538 int64_t ts;
539 82911 int cmp = 1;
540
541
2/2
✓ Branch 0 taken 2834 times.
✓ Branch 1 taken 80077 times.
 82911 if (st->finished)
542 2834 nb_samples = FFMIN(nb_samples, st->samples_queued);
543
544 82911 av_fifo_peek(st->fifo, &peek, 1, 0);
545 82911 ts = frame_end(sq, peek, nb_samples);
546
547 /* check if this stream's tail timestamp does not overtake
548 * the overall queue head */
549
4/4
✓ Branch 0 taken 82908 times.
✓ Branch 1 taken 3 times.
✓ Branch 2 taken 12423 times.
✓ Branch 3 taken 70485 times.
 82911 if (ts != AV_NOPTS_VALUE && st_head)
550 12423 cmp = av_compare_ts(ts, st->tb, st_head->head_ts, st_head->tb);
551
552 /* We can release frames that do not end after the queue head.
553 * Frames with no timestamps are just passed through with no conditions.
554 * Frames are also passed through when there are no limiting streams.
555 */
556
6/6
✓ Branch 0 taken 71480 times.
✓ Branch 1 taken 11431 times.
✓ Branch 2 taken 71477 times.
✓ Branch 3 taken 3 times.
✓ Branch 4 taken 70437 times.
✓ Branch 5 taken 1040 times.
 82911 if (cmp <= 0 || ts == AV_NOPTS_VALUE || !sq->have_limiting) {
557
2/2
✓ Branch 0 taken 70233 times.
✓ Branch 1 taken 11638 times.
 81871 if (nb_samples &&
558
4/4
✓ Branch 0 taken 8625 times.
✓ Branch 1 taken 61608 times.
✓ Branch 3 taken 1 times.
✓ Branch 4 taken 8624 times.
 131842 (nb_samples != peek.f->nb_samples || !frame_is_aligned(sq, peek.f))) {
559 61609 int ret = receive_samples(sq, st, frame.f, nb_samples);
560
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 61609 times.
 61609 if (ret < 0)
561 ✗ return ret;
562 } else {
563 20262 frame_move(sq, frame, peek);
564 20262 objpool_release(sq->pool, (void**)&peek);
565 20262 av_fifo_drain2(st->fifo, 1);
566
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 20262 times.
 20262 av_assert0(st->samples_queued >= frame_samples(sq, frame));
567 20262 st->samples_queued -= frame_samples(sq, frame);
568 }
569
570
2/2
✓ Branch 0 taken 11431 times.
✓ Branch 1 taken 70440 times.
 175173 av_log(sq->logctx, AV_LOG_DEBUG,
571 "sq: receive %u ts %s queue head %d ts %s\n", stream_idx,
572 81871 av_ts2timestr(frame_end(sq, frame, 0), &st->tb),
573 sq->head_stream,
574 11431 st_head ? av_ts2timestr(st_head->head_ts, &st_head->tb) : "N/A");
575
576 81871 return 0;
577 }
578 }
579
580
4/4
✓ Branch 0 taken 85 times.
✓ Branch 1 taken 57855 times.
✓ Branch 3 taken 24 times.
✓ Branch 4 taken 61 times.
 57940 return (sq->finished || (st->finished && !av_fifo_can_read(st->fifo))) ?
581
2/2
✓ Branch 0 taken 57940 times.
✓ Branch 1 taken 5405 times.
 121285 AVERROR_EOF : AVERROR(EAGAIN);
582 }
583
584 144067 static int receive_internal(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
585 {
586 144067 int nb_eof = 0;
587 int ret;
588
589 /* read a frame for a specific stream */
590
2/2
✓ Branch 0 taken 142431 times.
✓ Branch 1 taken 1636 times.
 144067 if (stream_idx >= 0) {
591 142431 ret = receive_for_stream(sq, stream_idx, frame);
592
2/2
✓ Branch 0 taken 61300 times.
✓ Branch 1 taken 81131 times.
 142431 return (ret < 0) ? ret : stream_idx;
593 }
594
595 /* read a frame for any stream with available output */
596
2/2
✓ Branch 0 taken 2785 times.
✓ Branch 1 taken 896 times.
 3681 for (unsigned int i = 0; i < sq->nb_streams; i++) {
597 2785 ret = receive_for_stream(sq, i, frame);
598
4/4
✓ Branch 0 taken 2739 times.
✓ Branch 1 taken 46 times.
✓ Branch 2 taken 1999 times.
✓ Branch 3 taken 740 times.
 2785 if (ret == AVERROR_EOF || ret == AVERROR(EAGAIN)) {
599 2045 nb_eof += (ret == AVERROR_EOF);
600 2045 continue;
601 }
602
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 740 times.
 740 return (ret < 0) ? ret : i;
603 }
604
605
2/2
✓ Branch 0 taken 11 times.
✓ Branch 1 taken 885 times.
 896 return (nb_eof == sq->nb_streams) ? AVERROR_EOF : AVERROR(EAGAIN);
606 }
607
608 143973 int sq_receive(SyncQueue *sq, int stream_idx, SyncQueueFrame frame)
609 {
610 143973 int ret = receive_internal(sq, stream_idx, frame);
611
612 /* try again if the queue overflowed and triggered a fake heartbeat
613 * for lagging streams */
614
4/4
✓ Branch 0 taken 56802 times.
✓ Branch 1 taken 87171 times.
✓ Branch 3 taken 94 times.
✓ Branch 4 taken 56708 times.
 143973 if (ret == AVERROR(EAGAIN) && overflow_heartbeat(sq, stream_idx))
615 94 ret = receive_internal(sq, stream_idx, frame);
616
617 143973 return ret;
618 }
619
620 2792 int sq_add_stream(SyncQueue *sq, int limiting)
621 {
622 SyncQueueStream *tmp, *st;
623
624 2792 tmp = av_realloc_array(sq->streams, sq->nb_streams + 1, sizeof(*sq->streams));
625
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2792 times.
 2792 if (!tmp)
626 ✗ return AVERROR(ENOMEM);
627 2792 sq->streams = tmp;
628
629 2792 st = &sq->streams[sq->nb_streams];
630 2792 memset(st, 0, sizeof(*st));
631
632 2792 st->fifo = av_fifo_alloc2(1, sizeof(SyncQueueFrame), AV_FIFO_FLAG_AUTO_GROW);
633
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2792 times.
 2792 if (!st->fifo)
634 ✗ return AVERROR(ENOMEM);
635
636 /* we set a valid default, so that a pathological stream that never
637 * receives even a real timebase (and no frames) won't stall all other
638 * streams forever; cf. overflow_heartbeat() */
639 2792 st->tb = (AVRational){ 1, 1 };
640 2792 st->head_ts = AV_NOPTS_VALUE;
641 2792 st->frames_max = UINT64_MAX;
642 2792 st->limiting = limiting;
643
644 2792 sq->have_limiting |= limiting;
645
646 2792 return sq->nb_streams++;
647 }
648
649 2642 void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames)
650 {
651 SyncQueueStream *st;
652
653
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2642 times.
 2642 av_assert0(stream_idx < sq->nb_streams);
654 2642 st = &sq->streams[stream_idx];
655
656 2642 st->frames_max = frames;
657
2/2
✓ Branch 0 taken 14 times.
✓ Branch 1 taken 2628 times.
 2642 if (st->frames_sent >= st->frames_max)
658 14 finish_stream(sq, stream_idx);
659 2642 }
660
661 120 void sq_frame_samples(SyncQueue *sq, unsigned int stream_idx,
662 int frame_samples)
663 {
664 SyncQueueStream *st;
665
666
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 120 times.
 120 av_assert0(sq->type == SYNC_QUEUE_FRAMES);
667
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 120 times.
 120 av_assert0(stream_idx < sq->nb_streams);
668 120 st = &sq->streams[stream_idx];
669
670 120 st->frame_samples = frame_samples;
671
672 120 sq->align_mask = av_cpu_max_align() - 1;
673 120 }
674
675 2753 SyncQueue *sq_alloc(enum SyncQueueType type, int64_t buf_size_us, void *logctx)
676 {
677 2753 SyncQueue *sq = av_mallocz(sizeof(*sq));
678
679
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2753 times.
 2753 if (!sq)
680 ✗ return NULL;
681
682 2753 sq->type = type;
683 2753 sq->buf_size_us = buf_size_us;
684 2753 sq->logctx = logctx;
685
686 2753 sq->head_stream = -1;
687 2753 sq->head_finished_stream = -1;
688
689
2/2
✓ Branch 0 taken 12 times.
✓ Branch 1 taken 2741 times.
 2753 sq->pool = (type == SYNC_QUEUE_PACKETS) ? objpool_alloc_packets() :
690 2741 objpool_alloc_frames();
691
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2753 times.
 2753 if (!sq->pool) {
692 ✗ av_freep(&sq);
693 ✗ return NULL;
694 }
695
696 2753 return sq;
697 }
698
699 13132 void sq_free(SyncQueue **psq)
700 {
701 13132 SyncQueue *sq = *psq;
702
703
2/2
✓ Branch 0 taken 10379 times.
✓ Branch 1 taken 2753 times.
 13132 if (!sq)
704 10379 return;
705
706
2/2
✓ Branch 0 taken 2792 times.
✓ Branch 1 taken 2753 times.
 5545 for (unsigned int i = 0; i < sq->nb_streams; i++) {
707 SyncQueueFrame frame;
708
2/2
✓ Branch 1 taken 50 times.
✓ Branch 2 taken 2792 times.
 2842 while (av_fifo_read(sq->streams[i].fifo, &frame, 1) >= 0)
709 50 objpool_release(sq->pool, (void**)&frame);
710
711 2792 av_fifo_freep2(&sq->streams[i].fifo);
712 }
713
714 2753 av_freep(&sq->streams);
715
716 2753 objpool_free(&sq->pool);
717
718 2753 av_freep(psq);
719 }
720