FFmpeg coverage

Directory: ../../../ffmpeg/
File: src/fftools/ffmpeg_sched.c
Date: 2026-08-23 18:45:53
Exec Total Coverage
Lines: 1171 1323 88.5%
Functions: 71 73 97.3%
Branches: 635 854 74.4%
Line Branch Exec Source
1 /*
2 * Inter-thread scheduling/synchronization.
3 * Copyright (c) 2023 Anton Khirnov
4 *
5 * This file is part of FFmpeg.
6 *
7 * FFmpeg is free software; you can redistribute it and/or
8 * modify it under the terms of the GNU Lesser General Public
9 * License as published by the Free Software Foundation; either
10 * version 2.1 of the License, or (at your option) any later version.
11 *
12 * FFmpeg is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with FFmpeg; if not, write to the Free Software
19 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
20 */
21
22 #include <stdatomic.h>
23 #include <stddef.h>
24 #include <stdint.h>
25
26 #include "cmdutils.h"
27 #include "ffmpeg_sched.h"
28 #include "ffmpeg_utils.h"
29 #include "sync_queue.h"
30 #include "thread_queue.h"
31
32 #include "libavcodec/packet.h"
33
34 #include "libavutil/avassert.h"
35 #include "libavutil/error.h"
36 #include "libavutil/fifo.h"
37 #include "libavutil/frame.h"
38 #include "libavutil/mem.h"
39 #include "libavutil/thread.h"
40 #include "libavutil/threadmessage.h"
41 #include "libavutil/time.h"
42
43 // 100 ms
44 // FIXME: some other value? make this dynamic?
45 #define SCHEDULE_TOLERANCE (100 * 1000)
46
47 enum QueueType {
48 QUEUE_PACKETS,
49 QUEUE_FRAMES,
50 };
51
52 typedef struct SchWaiter {
53 pthread_mutex_t lock;
54 pthread_cond_t cond;
55 atomic_int choked;
56
57 // the following are internal state of schedule_update_locked() and must not
58 // be accessed outside of it
59 int choked_prev;
60 int choked_next;
61 } SchWaiter;
62
63 typedef struct SchTask {
64 Scheduler *parent;
65 SchedulerNode node;
66
67 SchThreadFunc func;
68 void *func_arg;
69
70 pthread_t thread;
71 int thread_running;
72 } SchTask;
73
74 typedef struct SchDecOutput {
75 SchedulerNode *dst;
76 uint8_t *dst_finished;
77 unsigned nb_dst;
78 } SchDecOutput;
79
80 typedef struct SchDec {
81 const AVClass *class;
82
83 SchedulerNode src;
84
85 SchDecOutput *outputs;
86 unsigned nb_outputs;
87
88 SchTask task;
89 // Queue for receiving input packets, one stream.
90 ThreadQueue *queue;
91
92 // Queue for sending post-flush end timestamps back to the source
93 AVThreadMessageQueue *queue_end_ts;
94 int expect_end_ts;
95
96 // temporary storage used by sch_dec_send()
97 AVFrame *send_frame;
98 } SchDec;
99
100 typedef struct SchSyncQueue {
101 SyncQueue *sq;
102 AVFrame *frame;
103 pthread_mutex_t lock;
104
105 unsigned *enc_idx;
106 unsigned nb_enc_idx;
107 } SchSyncQueue;
108
109 typedef struct SchEnc {
110 const AVClass *class;
111
112 SchedulerNode src;
113 SchedulerNode *dst;
114 uint8_t *dst_finished;
115 unsigned nb_dst;
116
117 // [0] - index of the sync queue in Scheduler.sq_enc,
118 // [1] - index of this encoder in the sq
119 int sq_idx[2];
120
121 /* Opening encoders is somewhat nontrivial due to their interaction with
122 * sync queues, which are (among other things) responsible for maintaining
123 * constant audio frame size, when it is required by the encoder.
124 *
125 * Opening the encoder requires stream parameters, obtained from the first
126 * frame. However, that frame cannot be properly chunked by the sync queue
127 * without knowing the required frame size, which is only available after
128 * opening the encoder.
129 *
130 * This apparent circular dependency is resolved in the following way:
131 * - the caller creating the encoder gives us a callback which opens the
132 * encoder and returns the required frame size (if any)
133 * - when the first frame is sent to the encoder, the sending thread
134 * - calls this callback, opening the encoder
135 * - passes the returned frame size to the sync queue
136 */
137 int (*open_cb)(void *opaque, const AVFrame *frame);
138 int opened;
139
140 SchTask task;
141 // Queue for receiving input frames, one stream.
142 ThreadQueue *queue;
143 // tq_send() to queue returned EOF
144 int in_finished;
145
146 // temporary storage used by sch_enc_send()
147 AVPacket *send_pkt;
148 } SchEnc;
149
150 typedef struct SchDemuxStream {
151 SchedulerNode *dst;
152 uint8_t *dst_finished;
153 unsigned nb_dst;
154 } SchDemuxStream;
155
156 typedef struct SchDemux {
157 const AVClass *class;
158
159 SchDemuxStream *streams;
160 unsigned nb_streams;
161
162 SchTask task;
163 SchWaiter waiter;
164
165 // temporary storage used by sch_demux_send()
166 AVPacket *send_pkt;
167
168 // protected by schedule_lock
169 int task_exited;
170 } SchDemux;
171
172 typedef struct PreMuxQueue {
173 /**
174 * Queue for buffering the packets before the muxer task can be started.
175 */
176 AVFifo *fifo;
177 /**
178 * Maximum number of packets in fifo.
179 */
180 int max_packets;
181 /*
182 * The size of the AVPackets' buffers in queue.
183 * Updated when a packet is either pushed or pulled from the queue.
184 */
185 size_t data_size;
186 /* Threshold after which max_packets will be in effect */
187 size_t data_threshold;
188 } PreMuxQueue;
189
190 typedef struct SchMuxStream {
191 SchedulerNode src;
192
193 unsigned *sub_heartbeat_dst;
194 unsigned nb_sub_heartbeat_dst;
195
196 PreMuxQueue pre_mux_queue;
197
198 // an EOF was generated while flushing the pre-mux queue
199 int init_eof;
200
201 ////////////////////////////////////////////////////////////
202 // The following are protected by Scheduler.schedule_lock //
203
204 /* dts+duration of the last packet sent to this stream
205 in AV_TIME_BASE_Q */
206 int64_t last_dts;
207 // this stream no longer accepts input
208 int source_finished;
209 ////////////////////////////////////////////////////////////
210 } SchMuxStream;
211
212 typedef struct SchMux {
213 const AVClass *class;
214
215 SchMuxStream *streams;
216 unsigned nb_streams;
217 unsigned nb_streams_ready;
218
219 int (*init)(void *arg);
220
221 SchTask task;
222 /**
223 * Set to 1 after starting the muxer task and flushing the
224 * pre-muxing queues.
225 * Set either before any tasks have started, or with
226 * Scheduler.mux_ready_lock held.
227 */
228 atomic_int mux_started;
229 ThreadQueue *queue;
230 unsigned queue_size;
231
232 AVPacket *sub_heartbeat_pkt;
233 } SchMux;
234
235 typedef struct SchFilterIn {
236 SchedulerNode src;
237 int send_finished;
238 int receive_finished;
239 } SchFilterIn;
240
241 typedef struct SchFilterOut {
242 SchedulerNode dst;
243 } SchFilterOut;
244
245 typedef struct SchFilterGraph {
246 const AVClass *class;
247
248 SchFilterIn *inputs;
249 unsigned nb_inputs;
250 unsigned nb_inputs_finished_send;
251 unsigned nb_inputs_finished_receive;
252
253 SchFilterOut *outputs;
254 unsigned nb_outputs;
255
256 SchTask task;
257 // input queue, nb_inputs+1 streams
258 // last stream is control
259 ThreadQueue *queue;
260 SchWaiter waiter;
261
262 // protected by schedule_lock
263 unsigned best_input;
264 int task_exited;
265 } SchFilterGraph;
266
267 enum SchedulerState {
268 SCH_STATE_UNINIT,
269 SCH_STATE_STARTED,
270 SCH_STATE_STOPPED,
271 };
272
273 struct Scheduler {
274 const AVClass *class;
275
276 SchDemux *demux;
277 unsigned nb_demux;
278
279 SchMux *mux;
280 unsigned nb_mux;
281
282 unsigned nb_mux_ready;
283 pthread_mutex_t mux_ready_lock;
284
285 unsigned nb_mux_done;
286 unsigned task_failed;
287 pthread_mutex_t finish_lock;
288 pthread_cond_t finish_cond;
289
290
291 SchDec *dec;
292 unsigned nb_dec;
293
294 SchEnc *enc;
295 unsigned nb_enc;
296
297 SchSyncQueue *sq_enc;
298 unsigned nb_sq_enc;
299
300 SchFilterGraph *filters;
301 unsigned nb_filters;
302
303 char *sdp_filename;
304 int sdp_auto;
305
306 enum SchedulerState state;
307 atomic_int terminate;
308
309 pthread_mutex_t schedule_lock;
310
311 atomic_int_least64_t last_dts;
312 };
313
314 /**
315 * Wait until this task is allowed to proceed.
316 *
317 * @retval 0 the caller should proceed
318 * @retval 1 the caller should terminate
319 */
320 554424 static int waiter_wait(Scheduler *sch, SchWaiter *w)
321 {
322 int terminate;
323
324
2/2
✓ Branch 0 taken 551518 times.
✓ Branch 1 taken 2906 times.
 554424 if (!atomic_load(&w->choked))
325 551518 return 0;
326
327 2906 pthread_mutex_lock(&w->lock);
328
329
4/4
✓ Branch 0 taken 2933 times.
✓ Branch 1 taken 2481 times.
✓ Branch 2 taken 2508 times.
✓ Branch 3 taken 425 times.
 5414 while (atomic_load(&w->choked) && !atomic_load(&sch->terminate))
330 2508 pthread_cond_wait(&w->cond, &w->lock);
331
332 2906 terminate = atomic_load(&sch->terminate);
333
334 2906 pthread_mutex_unlock(&w->lock);
335
336 2906 return terminate;
337 }
338
339 61689 static void waiter_set(SchWaiter *w, int choked)
340 {
341 61689 pthread_mutex_lock(&w->lock);
342
343 61689 atomic_store(&w->choked, choked);
344 61689 pthread_cond_signal(&w->cond);
345
346 61689 pthread_mutex_unlock(&w->lock);
347 61689 }
348
349 15849 static int waiter_init(SchWaiter *w)
350 {
351 int ret;
352
353 15849 atomic_init(&w->choked, 0);
354
355 15849 ret = pthread_mutex_init(&w->lock, NULL);
356
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 15849 times.
 15849 if (ret)
357 ✗ return AVERROR(ret);
358
359 15849 ret = pthread_cond_init(&w->cond, NULL);
360
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 15849 times.
 15849 if (ret)
361 ✗ return AVERROR(ret);
362
363 15849 return 0;
364 }
365
366 15849 static void waiter_uninit(SchWaiter *w)
367 {
368 15849 pthread_mutex_destroy(&w->lock);
369 15849 pthread_cond_destroy(&w->cond);
370 15849 }
371
372 32428 static int queue_alloc(ThreadQueue **ptq, unsigned nb_streams, unsigned queue_size,
373 enum QueueType type)
374 {
375 ThreadQueue *tq;
376
377
1/2
✓ Branch 0 taken 32428 times.
✗ Branch 1 not taken.
 32428 if (queue_size <= 0) {
378
2/2
✓ Branch 0 taken 16612 times.
✓ Branch 1 taken 15816 times.
 32428 if (type == QUEUE_FRAMES)
379 16612 queue_size = DEFAULT_FRAME_THREAD_QUEUE_SIZE;
380 else
381 15816 queue_size = DEFAULT_PACKET_THREAD_QUEUE_SIZE;
382 }
383
384
2/2
✓ Branch 0 taken 16612 times.
✓ Branch 1 taken 15816 times.
 32428 if (type == QUEUE_FRAMES) {
385 // This queue length is used in the decoder code to ensure that
386 // there are enough entries in fixed-size frame pools to account
387 // for frames held in queues inside the ffmpeg utility. If this
388 // can ever dynamically change then the corresponding decode
389 // code needs to be updated as well.
390
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 16612 times.
 16612 av_assert0(queue_size <= DEFAULT_FRAME_THREAD_QUEUE_SIZE);
391 }
392
393 32428 tq = tq_alloc(nb_streams, queue_size,
394 (type == QUEUE_PACKETS) ? THREAD_QUEUE_PACKETS : THREAD_QUEUE_FRAMES);
395
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 32428 times.
 32428 if (!tq)
396 ✗ return AVERROR(ENOMEM);
397
398 32428 *ptq = tq;
399 32428 return 0;
400 }
401
402 static void *task_wrapper(void *arg);
403
404 40031 static int task_start(SchTask *task)
405 {
406 int ret;
407
408
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 40028 times.
 40031 if (!task->parent)
409 3 return 0;
410
411 40028 av_log(task->func_arg, AV_LOG_VERBOSE, "Starting thread...\n");
412
413
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 40028 times.
 40028 av_assert0(!task->thread_running);
414
415 40028 ret = pthread_create(&task->thread, NULL, task_wrapper, task);
416
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 40028 times.
 40028 if (ret) {
417 ✗ av_log(task->func_arg, AV_LOG_ERROR, "pthread_create() failed: %s\n",
418 strerror(ret));
419 ✗ return AVERROR(ret);
420 }
421
422 40028 task->thread_running = 1;
423 40028 return 0;
424 }
425
426 40062 static void task_init(Scheduler *sch, SchTask *task, enum SchedulerNodeType type, unsigned idx,
427 SchThreadFunc func, void *func_arg)
428 {
429 40062 task->parent = sch;
430
431 40062 task->node.type = type;
432 40062 task->node.idx = idx;
433
434 40062 task->func = func;
435 40062 task->func_arg = func_arg;
436 40062 }
437
438 654596 static int64_t trailing_dts(const Scheduler *sch)
439 {
440 654596 int64_t min_dts = INT64_MAX;
441
442
2/2
✓ Branch 0 taken 655665 times.
✓ Branch 1 taken 626057 times.
 1281722 for (unsigned i = 0; i < sch->nb_mux; i++) {
443 655665 const SchMux *mux = &sch->mux[i];
444
445
2/2
✓ Branch 0 taken 722312 times.
✓ Branch 1 taken 627126 times.
 1349438 for (unsigned j = 0; j < mux->nb_streams; j++) {
446 722312 const SchMuxStream *ms = &mux->streams[j];
447
448
2/2
✓ Branch 0 taken 38784 times.
✓ Branch 1 taken 683528 times.
 722312 if (ms->source_finished)
449 38784 continue;
450
2/2
✓ Branch 0 taken 28539 times.
✓ Branch 1 taken 654989 times.
 683528 if (ms->last_dts == AV_NOPTS_VALUE)
451 28539 return AV_NOPTS_VALUE;
452
453 654989 min_dts = FFMIN(min_dts, ms->last_dts);
454 }
455 }
456
457
2/2
✓ Branch 0 taken 596412 times.
✓ Branch 1 taken 29645 times.
 626057 return min_dts == INT64_MAX ? AV_NOPTS_VALUE : min_dts;
458 }
459
460 663314 static int64_t progressing_dts(const Scheduler *sch, int count_finished)
461 {
462 663314 int64_t max_dts = INT64_MIN;
463
464
2/2
✓ Branch 0 taken 664669 times.
✓ Branch 1 taken 663314 times.
 1327983 for (unsigned i = 0; i < sch->nb_mux; i++) {
465 664669 const SchMux *mux = &sch->mux[i];
466
467
2/2
✓ Branch 0 taken 740493 times.
✓ Branch 1 taken 664669 times.
 1405162 for (unsigned j = 0; j < mux->nb_streams; j++) {
468 740493 const SchMuxStream *ms = &mux->streams[j];
469
470
4/4
✓ Branch 0 taken 48150 times.
✓ Branch 1 taken 692343 times.
✓ Branch 2 taken 38922 times.
✓ Branch 3 taken 9228 times.
 740493 if (ms->source_finished && !count_finished)
471 38922 continue;
472
2/2
✓ Branch 0 taken 666796 times.
✓ Branch 1 taken 34775 times.
 701571 if (ms->last_dts != AV_NOPTS_VALUE)
473 666796 max_dts = FFMAX(max_dts, ms->last_dts);
474 }
475 }
476
477 663314 return max_dts == INT64_MIN ? AV_NOPTS_VALUE : max_dts;
478 }
479
480 3 void sch_remove_filtergraph(Scheduler *sch, int idx)
481 {
482 3 SchFilterGraph *fg = &sch->filters[idx];
483
484
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3 av_assert0(!fg->task.thread_running);
485 3 memset(&fg->task, 0, sizeof(fg->task));
486
487 3 tq_free(&fg->queue);
488
489 3 av_freep(&fg->inputs);
490 3 fg->nb_inputs = 0;
491 3 av_freep(&fg->outputs);
492 3 fg->nb_outputs = 0;
493
494 3 fg->task_exited = 1;
495 3 }
496
497 8727 void sch_free(Scheduler **psch)
498 {
499 8727 Scheduler *sch = *psch;
500
501
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8727 times.
 8727 if (!sch)
502 ✗ return;
503
504 8727 sch_stop(sch, NULL);
505
506
2/2
✓ Branch 0 taken 7630 times.
✓ Branch 1 taken 8727 times.
 16357 for (unsigned i = 0; i < sch->nb_demux; i++) {
507 7630 SchDemux *d = &sch->demux[i];
508
509
2/2
✓ Branch 0 taken 7906 times.
✓ Branch 1 taken 7630 times.
 15536 for (unsigned j = 0; j < d->nb_streams; j++) {
510 7906 SchDemuxStream *ds = &d->streams[j];
511 7906 av_freep(&ds->dst);
512 7906 av_freep(&ds->dst_finished);
513 }
514 7630 av_freep(&d->streams);
515
516 7630 av_packet_free(&d->send_pkt);
517
518 7630 waiter_uninit(&d->waiter);
519 }
520 8727 av_freep(&sch->demux);
521
522
2/2
✓ Branch 0 taken 8726 times.
✓ Branch 1 taken 8727 times.
 17453 for (unsigned i = 0; i < sch->nb_mux; i++) {
523 8726 SchMux *mux = &sch->mux[i];
524
525
2/2
✓ Branch 0 taken 9232 times.
✓ Branch 1 taken 8726 times.
 17958 for (unsigned j = 0; j < mux->nb_streams; j++) {
526 9232 SchMuxStream *ms = &mux->streams[j];
527
528
1/2
✓ Branch 0 taken 9232 times.
✗ Branch 1 not taken.
 9232 if (ms->pre_mux_queue.fifo) {
529 AVPacket *pkt;
530
2/2
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 9232 times.
 9233 while (av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1) >= 0)
531 1 av_packet_free(&pkt);
532 9232 av_fifo_freep2(&ms->pre_mux_queue.fifo);
533 }
534
535 9232 av_freep(&ms->sub_heartbeat_dst);
536 }
537 8726 av_freep(&mux->streams);
538
539 8726 av_packet_free(&mux->sub_heartbeat_pkt);
540
541 8726 tq_free(&mux->queue);
542 }
543 8727 av_freep(&sch->mux);
544
545
2/2
✓ Branch 0 taken 7094 times.
✓ Branch 1 taken 8727 times.
 15821 for (unsigned i = 0; i < sch->nb_dec; i++) {
546 7094 SchDec *dec = &sch->dec[i];
547
548 7094 tq_free(&dec->queue);
549
550 7094 av_thread_message_queue_free(&dec->queue_end_ts);
551
552
2/2
✓ Branch 0 taken 7100 times.
✓ Branch 1 taken 7094 times.
 14194 for (unsigned j = 0; j < dec->nb_outputs; j++) {
553 7100 SchDecOutput *o = &dec->outputs[j];
554
555 7100 av_freep(&o->dst);
556 7100 av_freep(&o->dst_finished);
557 }
558
559 7094 av_freep(&dec->outputs);
560
561 7094 av_frame_free(&dec->send_frame);
562 }
563 8727 av_freep(&sch->dec);
564
565
2/2
✓ Branch 0 taken 8393 times.
✓ Branch 1 taken 8727 times.
 17120 for (unsigned i = 0; i < sch->nb_enc; i++) {
566 8393 SchEnc *enc = &sch->enc[i];
567
568 8393 tq_free(&enc->queue);
569
570 8393 av_packet_free(&enc->send_pkt);
571
572 8393 av_freep(&enc->dst);
573 8393 av_freep(&enc->dst_finished);
574 }
575 8727 av_freep(&sch->enc);
576
577
2/2
✓ Branch 0 taken 3243 times.
✓ Branch 1 taken 8727 times.
 11970 for (unsigned i = 0; i < sch->nb_sq_enc; i++) {
578 3243 SchSyncQueue *sq = &sch->sq_enc[i];
579 3243 sq_free(&sq->sq);
580 3243 av_frame_free(&sq->frame);
581 3243 pthread_mutex_destroy(&sq->lock);
582 3243 av_freep(&sq->enc_idx);
583 }
584 8727 av_freep(&sch->sq_enc);
585
586
2/2
✓ Branch 0 taken 8219 times.
✓ Branch 1 taken 8727 times.
 16946 for (unsigned i = 0; i < sch->nb_filters; i++) {
587 8219 SchFilterGraph *fg = &sch->filters[i];
588
589 8219 tq_free(&fg->queue);
590
591 8219 av_freep(&fg->inputs);
592 8219 av_freep(&fg->outputs);
593
594 8219 waiter_uninit(&fg->waiter);
595 }
596 8727 av_freep(&sch->filters);
597
598 8727 av_freep(&sch->sdp_filename);
599
600 8727 pthread_mutex_destroy(&sch->schedule_lock);
601
602 8727 pthread_mutex_destroy(&sch->mux_ready_lock);
603
604 8727 pthread_mutex_destroy(&sch->finish_lock);
605 8727 pthread_cond_destroy(&sch->finish_cond);
606
607 8727 av_freep(psch);
608 }
609
610 static const AVClass scheduler_class = {
611 .class_name = "Scheduler",
612 .version = LIBAVUTIL_VERSION_INT,
613 };
614
615 8727 Scheduler *sch_alloc(void)
616 {
617 Scheduler *sch;
618 int ret;
619
620 8727 sch = av_mallocz(sizeof(*sch));
621
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8727 times.
 8727 if (!sch)
622 ✗ return NULL;
623
624 8727 sch->class = &scheduler_class;
625 8727 sch->sdp_auto = 1;
626
627 8727 ret = pthread_mutex_init(&sch->schedule_lock, NULL);
628
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8727 times.
 8727 if (ret)
629 ✗ goto fail;
630
631 8727 ret = pthread_mutex_init(&sch->mux_ready_lock, NULL);
632
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8727 times.
 8727 if (ret)
633 ✗ goto fail;
634
635 8727 ret = pthread_mutex_init(&sch->finish_lock, NULL);
636
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8727 times.
 8727 if (ret)
637 ✗ goto fail;
638
639 8727 ret = pthread_cond_init(&sch->finish_cond, NULL);
640
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8727 times.
 8727 if (ret)
641 ✗ goto fail;
642
643 8727 return sch;
644 ✗ fail:
645 ✗ sch_free(&sch);
646 ✗ return NULL;
647 }
648
649 ✗ int sch_sdp_filename(Scheduler *sch, const char *sdp_filename)
650 {
651 ✗ av_freep(&sch->sdp_filename);
652 ✗ sch->sdp_filename = av_strdup(sdp_filename);
653 ✗ return sch->sdp_filename ? 0 : AVERROR(ENOMEM);
654 }
655
656 static const AVClass sch_mux_class = {
657 .class_name = "SchMux",
658 .version = LIBAVUTIL_VERSION_INT,
659 .parent_log_context_offset = offsetof(SchMux, task.func_arg),
660 };
661
662 8726 int sch_add_mux(Scheduler *sch, SchThreadFunc func, int (*init)(void *),
663 void *arg, int sdp_auto, unsigned thread_queue_size)
664 {
665 8726 const unsigned idx = sch->nb_mux;
666
667 SchMux *mux;
668 int ret;
669
670 8726 ret = GROW_ARRAY(sch->mux, sch->nb_mux);
671
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8726 times.
 8726 if (ret < 0)
672 ✗ return ret;
673
674 8726 mux = &sch->mux[idx];
675 8726 mux->class = &sch_mux_class;
676 8726 mux->init = init;
677 8726 mux->queue_size = thread_queue_size;
678
679 8726 task_init(sch, &mux->task, SCH_NODE_TYPE_MUX, idx, func, arg);
680
681 8726 sch->sdp_auto &= sdp_auto;
682
683 8726 return idx;
684 }
685
686 9232 int sch_add_mux_stream(Scheduler *sch, unsigned mux_idx)
687 {
688 SchMux *mux;
689 SchMuxStream *ms;
690 unsigned stream_idx;
691 int ret;
692
693
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9232 times.
 9232 av_assert0(mux_idx < sch->nb_mux);
694 9232 mux = &sch->mux[mux_idx];
695
696 9232 ret = GROW_ARRAY(mux->streams, mux->nb_streams);
697
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9232 times.
 9232 if (ret < 0)
698 ✗ return ret;
699 9232 stream_idx = mux->nb_streams - 1;
700
701 9232 ms = &mux->streams[stream_idx];
702
703 9232 ms->pre_mux_queue.fifo = av_fifo_alloc2(8, sizeof(AVPacket*), 0);
704
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9232 times.
 9232 if (!ms->pre_mux_queue.fifo)
705 ✗ return AVERROR(ENOMEM);
706
707 9232 ms->last_dts = AV_NOPTS_VALUE;
708
709 9232 return stream_idx;
710 }
711
712 static const AVClass sch_demux_class = {
713 .class_name = "SchDemux",
714 .version = LIBAVUTIL_VERSION_INT,
715 .parent_log_context_offset = offsetof(SchDemux, task.func_arg),
716 };
717
718 7630 int sch_add_demux(Scheduler *sch, SchThreadFunc func, void *ctx)
719 {
720 7630 const unsigned idx = sch->nb_demux;
721
722 SchDemux *d;
723 int ret;
724
725 7630 ret = GROW_ARRAY(sch->demux, sch->nb_demux);
726
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7630 times.
 7630 if (ret < 0)
727 ✗ return ret;
728
729 7630 d = &sch->demux[idx];
730
731 7630 task_init(sch, &d->task, SCH_NODE_TYPE_DEMUX, idx, func, ctx);
732
733 7630 d->class = &sch_demux_class;
734 7630 d->send_pkt = av_packet_alloc();
735
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7630 times.
 7630 if (!d->send_pkt)
736 ✗ return AVERROR(ENOMEM);
737
738 7630 ret = waiter_init(&d->waiter);
739
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7630 times.
 7630 if (ret < 0)
740 ✗ return ret;
741
742 7630 return idx;
743 }
744
745 7906 int sch_add_demux_stream(Scheduler *sch, unsigned demux_idx)
746 {
747 SchDemux *d;
748 int ret;
749
750
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7906 times.
 7906 av_assert0(demux_idx < sch->nb_demux);
751 7906 d = &sch->demux[demux_idx];
752
753 7906 ret = GROW_ARRAY(d->streams, d->nb_streams);
754
1/2
✓ Branch 0 taken 7906 times.
✗ Branch 1 not taken.
 7906 return ret < 0 ? ret : d->nb_streams - 1;
755 }
756
757 7100 int sch_add_dec_output(Scheduler *sch, unsigned dec_idx)
758 {
759 SchDec *dec;
760 int ret;
761
762
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7100 times.
 7100 av_assert0(dec_idx < sch->nb_dec);
763 7100 dec = &sch->dec[dec_idx];
764
765 7100 ret = GROW_ARRAY(dec->outputs, dec->nb_outputs);
766
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7100 times.
 7100 if (ret < 0)
767 ✗ return ret;
768
769 7100 return dec->nb_outputs - 1;
770 }
771
772 static const AVClass sch_dec_class = {
773 .class_name = "SchDec",
774 .version = LIBAVUTIL_VERSION_INT,
775 .parent_log_context_offset = offsetof(SchDec, task.func_arg),
776 };
777
778 7094 int sch_add_dec(Scheduler *sch, SchThreadFunc func, void *ctx, int send_end_ts)
779 {
780 7094 const unsigned idx = sch->nb_dec;
781
782 SchDec *dec;
783 int ret;
784
785 7094 ret = GROW_ARRAY(sch->dec, sch->nb_dec);
786
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
 7094 if (ret < 0)
787 ✗ return ret;
788
789 7094 dec = &sch->dec[idx];
790
791 7094 task_init(sch, &dec->task, SCH_NODE_TYPE_DEC, idx, func, ctx);
792
793 7094 dec->class = &sch_dec_class;
794 7094 dec->send_frame = av_frame_alloc();
795
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
 7094 if (!dec->send_frame)
796 ✗ return AVERROR(ENOMEM);
797
798 7094 ret = sch_add_dec_output(sch, idx);
799
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
 7094 if (ret < 0)
800 ✗ return ret;
801
802 7094 ret = queue_alloc(&dec->queue, 1, 0, QUEUE_PACKETS);
803
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
 7094 if (ret < 0)
804 ✗ return ret;
805
806
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 7093 times.
 7094 if (send_end_ts) {
807 1 ret = av_thread_message_queue_alloc(&dec->queue_end_ts, 1, sizeof(Timestamp));
808
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (ret < 0)
809 ✗ return ret;
810 }
811
812 7094 return idx;
813 }
814
815 static const AVClass sch_enc_class = {
816 .class_name = "SchEnc",
817 .version = LIBAVUTIL_VERSION_INT,
818 .parent_log_context_offset = offsetof(SchEnc, task.func_arg),
819 };
820
821 8393 int sch_add_enc(Scheduler *sch, SchThreadFunc func, void *ctx,
822 int (*open_cb)(void *opaque, const AVFrame *frame))
823 {
824 8393 const unsigned idx = sch->nb_enc;
825
826 SchEnc *enc;
827 int ret;
828
829 8393 ret = GROW_ARRAY(sch->enc, sch->nb_enc);
830
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8393 times.
 8393 if (ret < 0)
831 ✗ return ret;
832
833 8393 enc = &sch->enc[idx];
834
835 8393 enc->class = &sch_enc_class;
836 8393 enc->open_cb = open_cb;
837 8393 enc->sq_idx[0] = -1;
838 8393 enc->sq_idx[1] = -1;
839
840 8393 task_init(sch, &enc->task, SCH_NODE_TYPE_ENC, idx, func, ctx);
841
842 8393 enc->send_pkt = av_packet_alloc();
843
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8393 times.
 8393 if (!enc->send_pkt)
844 ✗ return AVERROR(ENOMEM);
845
846 8393 ret = queue_alloc(&enc->queue, 1, 0, QUEUE_FRAMES);
847
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8393 times.
 8393 if (ret < 0)
848 ✗ return ret;
849
850 8393 return idx;
851 }
852
853 static const AVClass sch_fg_class = {
854 .class_name = "SchFilterGraph",
855 .version = LIBAVUTIL_VERSION_INT,
856 .parent_log_context_offset = offsetof(SchFilterGraph, task.func_arg),
857 };
858
859 8219 int sch_add_filtergraph(Scheduler *sch, unsigned nb_inputs, unsigned nb_outputs,
860 SchThreadFunc func, void *ctx)
861 {
862 8219 const unsigned idx = sch->nb_filters;
863
864 SchFilterGraph *fg;
865 int ret;
866
867 8219 ret = GROW_ARRAY(sch->filters, sch->nb_filters);
868
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8219 times.
 8219 if (ret < 0)
869 ✗ return ret;
870 8219 fg = &sch->filters[idx];
871
872 8219 fg->class = &sch_fg_class;
873
874 8219 task_init(sch, &fg->task, SCH_NODE_TYPE_FILTER_IN, idx, func, ctx);
875
876
2/2
✓ Branch 0 taken 7022 times.
✓ Branch 1 taken 1197 times.
 8219 if (nb_inputs) {
877 7022 fg->inputs = av_calloc(nb_inputs, sizeof(*fg->inputs));
878
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7022 times.
 7022 if (!fg->inputs)
879 ✗ return AVERROR(ENOMEM);
880 7022 fg->nb_inputs = nb_inputs;
881 }
882
883
1/2
✓ Branch 0 taken 8219 times.
✗ Branch 1 not taken.
 8219 if (nb_outputs) {
884 8219 fg->outputs = av_calloc(nb_outputs, sizeof(*fg->outputs));
885
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8219 times.
 8219 if (!fg->outputs)
886 ✗ return AVERROR(ENOMEM);
887 8219 fg->nb_outputs = nb_outputs;
888 }
889
890 8219 ret = waiter_init(&fg->waiter);
891
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8219 times.
 8219 if (ret < 0)
892 ✗ return ret;
893
894 8219 ret = queue_alloc(&fg->queue, fg->nb_inputs + 1, 0, QUEUE_FRAMES);
895
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8219 times.
 8219 if (ret < 0)
896 ✗ return ret;
897
898 8219 return idx;
899 }
900
901 3243 int sch_add_sq_enc(Scheduler *sch, uint64_t buf_size_us, void *logctx)
902 {
903 SchSyncQueue *sq;
904 int ret;
905
906 3243 ret = GROW_ARRAY(sch->sq_enc, sch->nb_sq_enc);
907
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3243 times.
 3243 if (ret < 0)
908 ✗ return ret;
909 3243 sq = &sch->sq_enc[sch->nb_sq_enc - 1];
910
911 3243 sq->sq = sq_alloc(SYNC_QUEUE_FRAMES, buf_size_us, logctx);
912
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3243 times.
 3243 if (!sq->sq)
913 ✗ return AVERROR(ENOMEM);
914
915 3243 sq->frame = av_frame_alloc();
916
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3243 times.
 3243 if (!sq->frame)
917 ✗ return AVERROR(ENOMEM);
918
919 3243 ret = pthread_mutex_init(&sq->lock, NULL);
920
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3243 times.
 3243 if (ret)
921 ✗ return AVERROR(ret);
922
923 3243 return sq - sch->sq_enc;
924 }
925
926 3308 int sch_sq_add_enc(Scheduler *sch, unsigned sq_idx, unsigned enc_idx,
927 int limiting, uint64_t max_frames)
928 {
929 SchSyncQueue *sq;
930 SchEnc *enc;
931 int ret;
932
933
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3308 times.
 3308 av_assert0(sq_idx < sch->nb_sq_enc);
934 3308 sq = &sch->sq_enc[sq_idx];
935
936
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3308 times.
 3308 av_assert0(enc_idx < sch->nb_enc);
937 3308 enc = &sch->enc[enc_idx];
938
939 3308 ret = GROW_ARRAY(sq->enc_idx, sq->nb_enc_idx);
940
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3308 times.
 3308 if (ret < 0)
941 ✗ return ret;
942 3308 sq->enc_idx[sq->nb_enc_idx - 1] = enc_idx;
943
944 3308 ret = sq_add_stream(sq->sq, limiting);
945
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3308 times.
 3308 if (ret < 0)
946 ✗ return ret;
947
948 3308 enc->sq_idx[0] = sq_idx;
949 3308 enc->sq_idx[1] = ret;
950
951
2/2
✓ Branch 0 taken 3083 times.
✓ Branch 1 taken 225 times.
 3308 if (max_frames != INT64_MAX)
952 3083 sq_limit_frames(sq->sq, enc->sq_idx[1], max_frames);
953
954 3308 return 0;
955 }
956
957 31840 int sch_connect(Scheduler *sch, SchedulerNode src, SchedulerNode dst)
958 {
959 int ret;
960
961
4/5
✓ Branch 0 taken 7932 times.
✓ Branch 1 taken 7164 times.
✓ Branch 2 taken 8351 times.
✓ Branch 3 taken 8393 times.
✗ Branch 4 not taken.
 31840 switch (src.type) {
962 7932 case SCH_NODE_TYPE_DEMUX: {
963 SchDemuxStream *ds;
964
965
2/4
✓ Branch 0 taken 7932 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 7932 times.
 7932 av_assert0(src.idx < sch->nb_demux &&
966 src.idx_stream < sch->demux[src.idx].nb_streams);
967 7932 ds = &sch->demux[src.idx].streams[src.idx_stream];
968
969 7932 ret = GROW_ARRAY(ds->dst, ds->nb_dst);
970
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7932 times.
 7932 if (ret < 0)
971 ✗ return ret;
972
973 7932 ds->dst[ds->nb_dst - 1] = dst;
974
975 // demuxed packets go to decoding or streamcopy
976
2/3
✓ Branch 0 taken 7093 times.
✓ Branch 1 taken 839 times.
✗ Branch 2 not taken.
 7932 switch (dst.type) {
977 7093 case SCH_NODE_TYPE_DEC: {
978 SchDec *dec;
979
980
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7093 times.
 7093 av_assert0(dst.idx < sch->nb_dec);
981 7093 dec = &sch->dec[dst.idx];
982
983
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7093 times.
 7093 av_assert0(!dec->src.type);
984 7093 dec->src = src;
985 7093 break;
986 }
987 839 case SCH_NODE_TYPE_MUX: {
988 SchMuxStream *ms;
989
990
2/4
✓ Branch 0 taken 839 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 839 times.
 839 av_assert0(dst.idx < sch->nb_mux &&
991 dst.idx_stream < sch->mux[dst.idx].nb_streams);
992 839 ms = &sch->mux[dst.idx].streams[dst.idx_stream];
993
994
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 839 times.
 839 av_assert0(!ms->src.type);
995 839 ms->src = src;
996
997 839 break;
998 }
999 ✗ default: av_assert0(0);
1000 }
1001
1002 7932 break;
1003 }
1004 7164 case SCH_NODE_TYPE_DEC: {
1005 SchDec *dec;
1006 SchDecOutput *o;
1007
1008
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7164 times.
 7164 av_assert0(src.idx < sch->nb_dec);
1009 7164 dec = &sch->dec[src.idx];
1010
1011
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7164 times.
 7164 av_assert0(src.idx_stream < dec->nb_outputs);
1012 7164 o = &dec->outputs[src.idx_stream];
1013
1014 7164 ret = GROW_ARRAY(o->dst, o->nb_dst);
1015
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7164 times.
 7164 if (ret < 0)
1016 ✗ return ret;
1017
1018 7164 o->dst[o->nb_dst - 1] = dst;
1019
1020 // decoded frames go to filters or encoding
1021
2/3
✓ Branch 0 taken 7122 times.
✓ Branch 1 taken 42 times.
✗ Branch 2 not taken.
 7164 switch (dst.type) {
1022 7122 case SCH_NODE_TYPE_FILTER_IN: {
1023 SchFilterIn *fi;
1024
1025
2/4
✓ Branch 0 taken 7122 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 7122 times.
 7122 av_assert0(dst.idx < sch->nb_filters &&
1026 dst.idx_stream < sch->filters[dst.idx].nb_inputs);
1027 7122 fi = &sch->filters[dst.idx].inputs[dst.idx_stream];
1028
1029
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7122 times.
 7122 av_assert0(!fi->src.type);
1030 7122 fi->src = src;
1031 7122 break;
1032 }
1033 42 case SCH_NODE_TYPE_ENC: {
1034 SchEnc *enc;
1035
1036
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 42 times.
 42 av_assert0(dst.idx < sch->nb_enc);
1037 42 enc = &sch->enc[dst.idx];
1038
1039
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 42 times.
 42 av_assert0(!enc->src.type);
1040 42 enc->src = src;
1041 42 break;
1042 }
1043 ✗ default: av_assert0(0);
1044 }
1045
1046 7164 break;
1047 }
1048 8351 case SCH_NODE_TYPE_FILTER_OUT: {
1049 SchFilterOut *fo;
1050
1051
2/4
✓ Branch 0 taken 8351 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 8351 times.
 8351 av_assert0(src.idx < sch->nb_filters &&
1052 src.idx_stream < sch->filters[src.idx].nb_outputs);
1053 8351 fo = &sch->filters[src.idx].outputs[src.idx_stream];
1054
1055
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8351 times.
 8351 av_assert0(!fo->dst.type);
1056 8351 fo->dst = dst;
1057
1058 // filtered frames go to encoding or another filtergraph
1059
2/3
✓ Branch 0 taken 8350 times.
✓ Branch 1 taken 1 times.
✗ Branch 2 not taken.
 8351 switch (dst.type) {
1060 8350 case SCH_NODE_TYPE_ENC: {
1061 SchEnc *enc;
1062
1063
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8350 times.
 8350 av_assert0(dst.idx < sch->nb_enc);
1064 8350 enc = &sch->enc[dst.idx];
1065
1066
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8350 times.
 8350 av_assert0(!enc->src.type);
1067 8350 enc->src = src;
1068 8350 break;
1069 }
1070 1 case SCH_NODE_TYPE_FILTER_IN: {
1071 SchFilterIn *fi;
1072
1073
2/4
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 1 times.
 1 av_assert0(dst.idx < sch->nb_filters &&
1074 dst.idx_stream < sch->filters[dst.idx].nb_inputs);
1075 1 fi = &sch->filters[dst.idx].inputs[dst.idx_stream];
1076
1077
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 av_assert0(!fi->src.type);
1078 1 fi->src = src;
1079 1 break;
1080 }
1081 ✗ default: av_assert0(0);
1082 }
1083
1084
1085 8351 break;
1086 }
1087 8393 case SCH_NODE_TYPE_ENC: {
1088 SchEnc *enc;
1089
1090
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8393 times.
 8393 av_assert0(src.idx < sch->nb_enc);
1091 8393 enc = &sch->enc[src.idx];
1092
1093 8393 ret = GROW_ARRAY(enc->dst, enc->nb_dst);
1094
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8393 times.
 8393 if (ret < 0)
1095 ✗ return ret;
1096
1097 8393 enc->dst[enc->nb_dst - 1] = dst;
1098
1099 // encoding packets go to muxing or decoding
1100
2/3
✓ Branch 0 taken 8392 times.
✓ Branch 1 taken 1 times.
✗ Branch 2 not taken.
 8393 switch (dst.type) {
1101 8392 case SCH_NODE_TYPE_MUX: {
1102 SchMuxStream *ms;
1103
1104
2/4
✓ Branch 0 taken 8392 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 8392 times.
 8392 av_assert0(dst.idx < sch->nb_mux &&
1105 dst.idx_stream < sch->mux[dst.idx].nb_streams);
1106 8392 ms = &sch->mux[dst.idx].streams[dst.idx_stream];
1107
1108
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8392 times.
 8392 av_assert0(!ms->src.type);
1109 8392 ms->src = src;
1110
1111 8392 break;
1112 }
1113 1 case SCH_NODE_TYPE_DEC: {
1114 SchDec *dec;
1115
1116
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 av_assert0(dst.idx < sch->nb_dec);
1117 1 dec = &sch->dec[dst.idx];
1118
1119
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 av_assert0(!dec->src.type);
1120 1 dec->src = src;
1121
1122 1 break;
1123 }
1124 ✗ default: av_assert0(0);
1125 }
1126
1127 8393 break;
1128 }
1129 ✗ default: av_assert0(0);
1130 }
1131
1132 31840 return 0;
1133 }
1134
1135 8721 static int mux_task_start(SchMux *mux)
1136 {
1137 8721 int ret = 0;
1138
1139 8721 ret = task_start(&mux->task);
1140
1/2
✓ Branch 0 taken 8721 times.
✗ Branch 1 not taken.
 8721 if (ret < 0)
1141 ✗ return ret;
1142
1143 /* flush the pre-muxing queues */
1144 3828 while (1) {
1145 12549 int min_stream = -1;
1146 12549 Timestamp min_ts = { .ts = AV_NOPTS_VALUE };
1147
1148 AVPacket *pkt;
1149
1150 // find the stream with the earliest dts or EOF in pre-muxing queue
1151
2/2
✓ Branch 0 taken 22299 times.
✓ Branch 1 taken 12493 times.
 34792 for (unsigned i = 0; i < mux->nb_streams; i++) {
1152 22299 SchMuxStream *ms = &mux->streams[i];
1153
1154
2/2
✓ Branch 1 taken 16626 times.
✓ Branch 2 taken 5673 times.
 22299 if (av_fifo_peek(ms->pre_mux_queue.fifo, &pkt, 1, 0) < 0)
1155 16626 continue;
1156
1157
4/4
✓ Branch 0 taken 5630 times.
✓ Branch 1 taken 43 times.
✓ Branch 2 taken 13 times.
✓ Branch 3 taken 5617 times.
 5673 if (!pkt || pkt->dts == AV_NOPTS_VALUE) {
1158 56 min_stream = i;
1159 56 break;
1160 }
1161
1162
4/4
✓ Branch 0 taken 1819 times.
✓ Branch 1 taken 3798 times.
✓ Branch 2 taken 26 times.
✓ Branch 3 taken 1793 times.
 7436 if (min_ts.ts == AV_NOPTS_VALUE ||
1163 1819 av_compare_ts(min_ts.ts, min_ts.tb, pkt->dts, pkt->time_base) > 0) {
1164 3824 min_stream = i;
1165 3824 min_ts = (Timestamp){ .ts = pkt->dts, .tb = pkt->time_base };
1166 }
1167 }
1168
1169
2/2
✓ Branch 0 taken 3828 times.
✓ Branch 1 taken 8721 times.
 12549 if (min_stream >= 0) {
1170 3828 SchMuxStream *ms = &mux->streams[min_stream];
1171
1172 3828 ret = av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1);
1173
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3828 times.
 3828 av_assert0(ret >= 0);
1174
1175
2/2
✓ Branch 0 taken 3785 times.
✓ Branch 1 taken 43 times.
 3828 if (pkt) {
1176
2/2
✓ Branch 0 taken 2644 times.
✓ Branch 1 taken 1141 times.
 3785 if (!ms->init_eof)
1177 2644 ret = tq_send(mux->queue, min_stream, pkt);
1178 3785 av_packet_free(&pkt);
1179
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 3783 times.
 3785 if (ret == AVERROR_EOF)
1180 2 ms->init_eof = 1;
1181
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3783 times.
 3783 else if (ret < 0)
1182 ✗ return ret;
1183 } else
1184 43 tq_send_finish(mux->queue, min_stream);
1185
1186 3828 continue;
1187 }
1188
1189 8721 break;
1190 }
1191
1192 8721 atomic_store(&mux->mux_started, 1);
1193
1194 8721 return 0;
1195 }
1196
1197 int print_sdp(const char *filename);
1198
1199 8721 static int mux_init(Scheduler *sch, SchMux *mux)
1200 {
1201 int ret;
1202
1203 8721 ret = mux->init(mux->task.func_arg);
1204
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8721 times.
 8721 if (ret < 0)
1205 ✗ return ret;
1206
1207 8721 sch->nb_mux_ready++;
1208
1209
2/4
✓ Branch 0 taken 8721 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 8721 times.
 8721 if (sch->sdp_filename || sch->sdp_auto) {
1210 ✗ if (sch->nb_mux_ready < sch->nb_mux)
1211 ✗ return 0;
1212
1213 ✗ ret = print_sdp(sch->sdp_filename);
1214 ✗ if (ret < 0) {
1215 ✗ av_log(sch, AV_LOG_ERROR, "Error writing the SDP.\n");
1216 ✗ return ret;
1217 }
1218
1219 /* SDP is written only after all the muxers are ready, so now we
1220 * start ALL the threads */
1221 ✗ for (unsigned i = 0; i < sch->nb_mux; i++) {
1222 ✗ ret = mux_task_start(&sch->mux[i]);
1223 ✗ if (ret < 0)
1224 ✗ return ret;
1225 }
1226 } else {
1227 8721 ret = mux_task_start(mux);
1228
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8721 times.
 8721 if (ret < 0)
1229 ✗ return ret;
1230 }
1231
1232 8721 return 0;
1233 }
1234
1235 9232 void sch_mux_stream_buffering(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
1236 size_t data_threshold, int max_packets)
1237 {
1238 SchMux *mux;
1239 SchMuxStream *ms;
1240
1241
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9232 times.
 9232 av_assert0(mux_idx < sch->nb_mux);
1242 9232 mux = &sch->mux[mux_idx];
1243
1244
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9232 times.
 9232 av_assert0(stream_idx < mux->nb_streams);
1245 9232 ms = &mux->streams[stream_idx];
1246
1247 9232 ms->pre_mux_queue.max_packets = max_packets;
1248 9232 ms->pre_mux_queue.data_threshold = data_threshold;
1249 9232 }
1250
1251 9227 int sch_mux_stream_ready(Scheduler *sch, unsigned mux_idx, unsigned stream_idx)
1252 {
1253 SchMux *mux;
1254 9227 int ret = 0;
1255
1256
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9227 times.
 9227 av_assert0(mux_idx < sch->nb_mux);
1257 9227 mux = &sch->mux[mux_idx];
1258
1259
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9227 times.
 9227 av_assert0(stream_idx < mux->nb_streams);
1260
1261 9227 pthread_mutex_lock(&sch->mux_ready_lock);
1262
1263
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9227 times.
 9227 av_assert0(mux->nb_streams_ready < mux->nb_streams);
1264
1265 // this may be called during initialization - do not start
1266 // threads before sch_start() is called
1267
2/2
✓ Branch 0 taken 8721 times.
✓ Branch 1 taken 506 times.
 9227 if (++mux->nb_streams_ready == mux->nb_streams &&
1268
2/2
✓ Branch 0 taken 8138 times.
✓ Branch 1 taken 583 times.
 8721 sch->state >= SCH_STATE_STARTED)
1269 8138 ret = mux_init(sch, mux);
1270
1271 9227 pthread_mutex_unlock(&sch->mux_ready_lock);
1272
1273 9227 return ret;
1274 }
1275
1276 1 int sch_mux_sub_heartbeat_add(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
1277 unsigned dec_idx)
1278 {
1279 SchMux *mux;
1280 SchMuxStream *ms;
1281 1 int ret = 0;
1282
1283
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 av_assert0(mux_idx < sch->nb_mux);
1284 1 mux = &sch->mux[mux_idx];
1285
1286
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 av_assert0(stream_idx < mux->nb_streams);
1287 1 ms = &mux->streams[stream_idx];
1288
1289 1 ret = GROW_ARRAY(ms->sub_heartbeat_dst, ms->nb_sub_heartbeat_dst);
1290
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (ret < 0)
1291 ✗ return ret;
1292
1293
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 av_assert0(dec_idx < sch->nb_dec);
1294 1 ms->sub_heartbeat_dst[ms->nb_sub_heartbeat_dst - 1] = dec_idx;
1295
1296
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 if (!mux->sub_heartbeat_pkt) {
1297 1 mux->sub_heartbeat_pkt = av_packet_alloc();
1298
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (!mux->sub_heartbeat_pkt)
1299 ✗ return AVERROR(ENOMEM);
1300 }
1301
1302 1 return 0;
1303 }
1304
1305 enum {
1306 UNCHOKE_DEMUX = (1 << 0),
1307 UNCHOKE_FILTER = (1 << 1),
1308
1309 UNCHOKE_ALL = UNCHOKE_DEMUX | UNCHOKE_FILTER,
1310 };
1311
1312 static void unchoke_for_stream(Scheduler *sch, SchedulerNode src, int flags);
1313
1314 // Unchoke any filter graphs that are downstream of this node, to prevent it
1315 // from getting stuck trying to push data to a full queue
1316 1147192 static void unchoke_downstream(Scheduler *sch, SchedulerNode *dst)
1317 {
1318 SchFilterGraph *fg;
1319 SchDec *dec;
1320 SchEnc *enc;
1321
4/5
✓ Branch 0 taken 522321 times.
✓ Branch 1 taken 1275 times.
✓ Branch 2 taken 101895 times.
✓ Branch 3 taken 521701 times.
✗ Branch 4 not taken.
 1147192 switch (dst->type) {
1322 522321 case SCH_NODE_TYPE_DEC:
1323 522321 dec = &sch->dec[dst->idx];
1324
2/2
✓ Branch 0 taken 522976 times.
✓ Branch 1 taken 522321 times.
 1045297 for (int i = 0; i < dec->nb_outputs; i++)
1325 522976 unchoke_downstream(sch, dec->outputs[i].dst);
1326 522321 break;
1327 1275 case SCH_NODE_TYPE_ENC:
1328 1275 enc = &sch->enc[dst->idx];
1329
2/2
✓ Branch 0 taken 1275 times.
✓ Branch 1 taken 1275 times.
 2550 for (int i = 0; i < enc->nb_dst; i++)
1330 1275 unchoke_downstream(sch, &enc->dst[i]);
1331 1275 break;
1332 101895 case SCH_NODE_TYPE_MUX:
1333 // muxers are never choked
1334 101895 break;
1335 521701 case SCH_NODE_TYPE_FILTER_IN:
1336 521701 fg = &sch->filters[dst->idx];
1337
2/2
✓ Branch 0 taken 455 times.
✓ Branch 1 taken 521246 times.
 521701 if (fg->best_input == fg->nb_inputs) {
1338 455 fg->waiter.choked_next = 0;
1339 } else {
1340 // ensure that this filter graph is not stuck waiting for
1341 // input from a different upstream source
1342 521246 unchoke_for_stream(sch, fg->inputs[fg->best_input].src, UNCHOKE_ALL);
1343 }
1344 521701 break;
1345 ✗ default:
1346 ✗ av_unreachable("Invalid destination node type?");
1347 break;
1348 }
1349 1147192 }
1350
1351 1188150 static void unchoke_for_stream(Scheduler *sch, SchedulerNode src, int flags)
1352 {
1353 2157369 while (1) {
1354 SchFilterGraph *fg;
1355 SchDemux *demux;
1356
4/5
✓ Branch 0 taken 1139772 times.
✓ Branch 1 taken 1051625 times.
✓ Branch 2 taken 577618 times.
✓ Branch 3 taken 576504 times.
✗ Branch 4 not taken.
 3345519 switch (src.type) {
1357 1139772 case SCH_NODE_TYPE_DEMUX:
1358 // fed directly by a demuxer (i.e. not through a filtergraph)
1359 1139772 demux = &sch->demux[src.idx];
1360
2/2
✓ Branch 0 taken 556959 times.
✓ Branch 1 taken 582813 times.
 1139772 if (demux->waiter.choked_next == 0)
1361 556959 return; // prevent infinite loop
1362
1/2
✓ Branch 0 taken 582813 times.
✗ Branch 1 not taken.
 582813 if (flags & UNCHOKE_DEMUX) {
1363 582813 demux->waiter.choked_next = 0;
1364
2/2
✓ Branch 0 taken 622941 times.
✓ Branch 1 taken 582813 times.
 1205754 for (int i = 0; i < demux->nb_streams; i++)
1365 622941 unchoke_downstream(sch, demux->streams[i].dst);
1366 }
1367 582813 return;
1368 1051625 case SCH_NODE_TYPE_DEC:
1369 1051625 src = sch->dec[src.idx].src;
1370 1051625 continue;
1371 577618 case SCH_NODE_TYPE_ENC:
1372 577618 src = sch->enc[src.idx].src;
1373 577618 continue;
1374 576504 case SCH_NODE_TYPE_FILTER_OUT:
1375 576504 fg = &sch->filters[src.idx];
1376 // the filtergraph contains internal sources and
1377 // requested to be scheduled directly
1378
2/2
✓ Branch 0 taken 48378 times.
✓ Branch 1 taken 528126 times.
 576504 if (fg->best_input == fg->nb_inputs) {
1379
1/2
✓ Branch 0 taken 48378 times.
✗ Branch 1 not taken.
 48378 if (flags & UNCHOKE_FILTER)
1380 48378 fg->waiter.choked_next = 0;
1381 48378 return;
1382 }
1383 528126 src = fg->inputs[fg->best_input].src;
1384 528126 continue;
1385 ✗ default:
1386 ✗ av_unreachable("Invalid source node type?");
1387 return;
1388 }
1389 }
1390 }
1391
1392 28339 static void choke_demux(const Scheduler *sch, int demux_id, int choked)
1393 {
1394 av_assert1(demux_id < sch->nb_demux);
1395 28339 SchDemux *demux = &sch->demux[demux_id];
1396
1397
2/2
✓ Branch 0 taken 29112 times.
✓ Branch 1 taken 28339 times.
 57451 for (int i = 0; i < demux->nb_streams; i++) {
1398 29112 SchedulerNode *dst = demux->streams[i].dst;
1399 SchFilterGraph *fg;
1400
1401
2/5
✓ Branch 0 taken 27276 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1836 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
 29112 switch (dst->type) {
1402 27276 case SCH_NODE_TYPE_DEC:
1403 27276 tq_choke(sch->dec[dst->idx].queue, choked);
1404 27276 break;
1405 ✗ case SCH_NODE_TYPE_ENC:
1406 ✗ tq_choke(sch->enc[dst->idx].queue, choked);
1407 ✗ break;
1408 1836 case SCH_NODE_TYPE_MUX:
1409 1836 break;
1410 ✗ case SCH_NODE_TYPE_FILTER_IN:
1411 ✗ fg = &sch->filters[dst->idx];
1412 ✗ if (fg->nb_inputs == 1)
1413 ✗ tq_choke(fg->queue, choked);
1414 ✗ break;
1415 ✗ default:
1416 ✗ av_unreachable("Invalid destination node type?");
1417 break;
1418 }
1419 }
1420 28339 }
1421
1422 669711 static void schedule_update_locked(Scheduler *sch)
1423 {
1424 int64_t dts;
1425 669711 int have_unchoked = 0;
1426
1427 // on termination request all waiters are choked,
1428 // we are not to unchoke them
1429
2/2
✓ Branch 0 taken 15115 times.
✓ Branch 1 taken 654596 times.
 669711 if (atomic_load(&sch->terminate))
1430 15115 return;
1431
1432 654596 dts = trailing_dts(sch);
1433
1434 654596 atomic_store(&sch->last_dts, progressing_dts(sch, 0));
1435
1436 // initialize our internal state
1437 #define RESET_WAITER(field) \
1438 do { \
1439 for (unsigned i = 0; i < sch->nb_##field; i++) { \
1440 SchWaiter *w = &sch->field[i].waiter; \
1441 w->choked_prev = atomic_load(&w->choked); \
1442 w->choked_next = 1; \
1443 } \
1444 } while (0)
1445
1446
2/2
✓ Branch 0 taken 630837 times.
✓ Branch 1 taken 654596 times.
 1285433 RESET_WAITER(demux);
1447
2/2
✓ Branch 0 taken 595829 times.
✓ Branch 1 taken 654596 times.
 1250425 RESET_WAITER(filters);
1448
1449 // figure out the sources that are allowed to proceed
1450
2/2
✓ Branch 0 taken 655947 times.
✓ Branch 1 taken 654596 times.
 1310543 for (unsigned i = 0; i < sch->nb_mux; i++) {
1451 655947 SchMux *mux = &sch->mux[i];
1452
1453
2/2
✓ Branch 0 taken 731265 times.
✓ Branch 1 taken 655947 times.
 1387212 for (unsigned j = 0; j < mux->nb_streams; j++) {
1454 731265 SchMuxStream *ms = &mux->streams[j];
1455
1456 // unblock sources for output streams that are not finished
1457 // and not too far ahead of the trailing stream
1458
2/2
✓ Branch 0 taken 38922 times.
✓ Branch 1 taken 692343 times.
 731265 if (ms->source_finished)
1459 38922 continue;
1460
4/4
✓ Branch 0 taken 43966 times.
✓ Branch 1 taken 648377 times.
✓ Branch 2 taken 9345 times.
✓ Branch 3 taken 34621 times.
 692343 if (dts == AV_NOPTS_VALUE && ms->last_dts != AV_NOPTS_VALUE)
1461 9345 continue;
1462
4/4
✓ Branch 0 taken 648377 times.
✓ Branch 1 taken 34621 times.
✓ Branch 2 taken 17233 times.
✓ Branch 3 taken 631144 times.
 682998 if (dts != AV_NOPTS_VALUE && ms->last_dts - dts >= SCHEDULE_TOLERANCE)
1463 17233 continue;
1464
1465 // resolve the source to unchoke
1466 665765 unchoke_for_stream(sch, ms->src, UNCHOKE_ALL);
1467 665765 have_unchoked = 1;
1468 }
1469 }
1470
1471 // also unchoke any sources feeding into closed filter graph inputs, so
1472 // that they can observe the downstream EOF
1473
2/2
✓ Branch 0 taken 595829 times.
✓ Branch 1 taken 654596 times.
 1250425 for (unsigned i = 0; i < sch->nb_filters; i++) {
1474 595829 SchFilterGraph *fg = &sch->filters[i];
1475
1476
2/2
✓ Branch 0 taken 583526 times.
✓ Branch 1 taken 595829 times.
 1179355 for (unsigned j = 0; j < fg->nb_inputs; j++) {
1477 583526 SchFilterIn *fi = &fg->inputs[j];
1478
4/4
✓ Branch 0 taken 6091 times.
✓ Branch 1 taken 577435 times.
✓ Branch 2 taken 1139 times.
✓ Branch 3 taken 4952 times.
 583526 if (fi->receive_finished && !fi->send_finished)
1479 1139 unchoke_for_stream(sch, fi->src, UNCHOKE_ALL);
1480 }
1481 }
1482
1483 // make sure to unchoke at least one source, if still available
1484 #define UNCHOKE_ONCE(field) \
1485 do { \
1486 for (unsigned i = 0; !have_unchoked && i < sch->nb_##field; i++) { \
1487 SchWaiter *w = &sch->field[i].waiter; \
1488 if (!sch->field[i].task_exited) { \
1489 w->choked_next = 0; \
1490 have_unchoked = 1; \
1491 break; \
1492 } \
1493 } \
1494 } while (0)
1495
1496
6/6
✓ Branch 0 taken 15767 times.
✓ Branch 1 taken 11890 times.
✓ Branch 2 taken 41535 times.
✓ Branch 3 taken 624951 times.
✓ Branch 4 taken 27657 times.
✓ Branch 5 taken 13878 times.
 666486 UNCHOKE_ONCE(demux);
1497
6/6
✓ Branch 0 taken 6639 times.
✓ Branch 1 taken 6253 times.
✓ Branch 2 taken 20131 times.
✓ Branch 3 taken 640718 times.
✓ Branch 4 taken 12892 times.
✓ Branch 5 taken 7239 times.
 660849 UNCHOKE_ONCE(filters);
1498
1499 #define UPDATE_WAITER(field) \
1500 do { \
1501 for (unsigned i = 0; i < sch->nb_##field; i++) { \
1502 SchWaiter *w = &sch->field[i].waiter; \
1503 if (w->choked_prev != w->choked_next) { \
1504 waiter_set(w, w->choked_next); \
1505 if (offsetof(Scheduler, field) == offsetof(Scheduler, demux)) \
1506 choke_demux(sch, i, w->choked_next); \
1507 } \
1508 } \
1509 } while (0)
1510
1511
4/4
✓ Branch 0 taken 20716 times.
✓ Branch 1 taken 610121 times.
✓ Branch 4 taken 630837 times.
✓ Branch 5 taken 654596 times.
 1285433 UPDATE_WAITER(demux);
1512
4/4
✓ Branch 0 taken 25132 times.
✓ Branch 1 taken 570697 times.
✓ Branch 3 taken 595829 times.
✓ Branch 4 taken 654596 times.
 1250425 UPDATE_WAITER(filters);
1513 }
1514
1515 enum {
1516 CYCLE_NODE_NEW = 0,
1517 CYCLE_NODE_STARTED,
1518 CYCLE_NODE_DONE,
1519 };
1520
1521 // Finds the filtergraph or muxer upstream of a scheduler node
1522 7127 static SchedulerNode src_filtergraph(const Scheduler *sch, SchedulerNode src)
1523 {
1524 while (1) {
1525
3/4
✓ Branch 0 taken 7127 times.
✓ Branch 1 taken 7126 times.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
 14254 switch (src.type) {
1526 7127 case SCH_NODE_TYPE_DEMUX:
1527 case SCH_NODE_TYPE_FILTER_OUT:
1528 7127 return src;
1529 7126 case SCH_NODE_TYPE_DEC:
1530 7126 src = sch->dec[src.idx].src;
1531 7127 continue;
1532 1 case SCH_NODE_TYPE_ENC:
1533 1 src = sch->enc[src.idx].src;
1534 1 continue;
1535 ✗ default:
1536 ✗ av_unreachable("Invalid source node type?");
1537 return (SchedulerNode) {0};
1538 }
1539 }
1540 }
1541
1542 static int
1543 8218 check_acyclic_for_output(const Scheduler *sch, SchedulerNode src,
1544 uint8_t *filters_visited, SchedulerNode *filters_stack)
1545 {
1546 8218 unsigned nb_filters_stack = 0;
1547
1548 8218 memset(filters_visited, 0, sch->nb_filters * sizeof(*filters_visited));
1549
1550 7129 while (1) {
1551 15347 const SchFilterGraph *fg = &sch->filters[src.idx];
1552
1553 15347 filters_visited[src.idx] = CYCLE_NODE_STARTED;
1554
1555 // descend into every input, depth first
1556
2/2
✓ Branch 0 taken 7127 times.
✓ Branch 1 taken 8220 times.
 15347 if (src.idx_stream < fg->nb_inputs) {
1557 7127 const SchFilterIn *fi = &fg->inputs[src.idx_stream++];
1558 7127 SchedulerNode node = src_filtergraph(sch, fi->src);
1559
1560 // connected to demuxer, no cycles possible
1561
2/2
✓ Branch 0 taken 7125 times.
✓ Branch 1 taken 2 times.
 7127 if (node.type == SCH_NODE_TYPE_DEMUX)
1562 7127 continue;
1563
1564 // otherwise connected to another filtergraph
1565
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 av_assert0(node.type == SCH_NODE_TYPE_FILTER_OUT);
1566
1567 // found a cycle
1568
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 if (filters_visited[node.idx] == CYCLE_NODE_STARTED)
1569 ✗ return AVERROR(EINVAL);
1570
1571 // place current position on stack and descend
1572
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 av_assert0(nb_filters_stack < sch->nb_filters);
1573 2 filters_stack[nb_filters_stack++] = src;
1574 2 src = (SchedulerNode){ .idx = node.idx, .idx_stream = 0 };
1575 2 continue;
1576 }
1577
1578 8220 filters_visited[src.idx] = CYCLE_NODE_DONE;
1579
1580 // previous search finished,
1581
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 8218 times.
 8220 if (nb_filters_stack) {
1582 2 src = filters_stack[--nb_filters_stack];
1583 2 continue;
1584 }
1585 8218 return 0;
1586 }
1587 }
1588
1589 8718 static int check_acyclic(Scheduler *sch)
1590 {
1591 8718 uint8_t *filters_visited = NULL;
1592 8718 SchedulerNode *filters_stack = NULL;
1593
1594 8718 int ret = 0;
1595
1596
2/2
✓ Branch 0 taken 617 times.
✓ Branch 1 taken 8101 times.
 8718 if (!sch->nb_filters)
1597 617 return 0;
1598
1599 8101 filters_visited = av_malloc_array(sch->nb_filters, sizeof(*filters_visited));
1600
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8101 times.
 8101 if (!filters_visited)
1601 ✗ return AVERROR(ENOMEM);
1602
1603 8101 filters_stack = av_malloc_array(sch->nb_filters, sizeof(*filters_stack));
1604
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8101 times.
 8101 if (!filters_stack) {
1605 ✗ ret = AVERROR(ENOMEM);
1606 ✗ goto fail;
1607 }
1608
1609 // trace the transcoding graph upstream from every filtegraph
1610
2/2
✓ Branch 0 taken 8218 times.
✓ Branch 1 taken 8101 times.
 16319 for (unsigned i = 0; i < sch->nb_filters; i++) {
1611 8218 ret = check_acyclic_for_output(sch, (SchedulerNode){ .idx = i },
1612 filters_visited, filters_stack);
1613
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8218 times.
 8218 if (ret < 0) {
1614 ✗ av_log(&sch->filters[i], AV_LOG_ERROR, "Transcoding graph has a cycle\n");
1615 ✗ goto fail;
1616 }
1617 }
1618
1619 8101 fail:
1620 8101 av_freep(&filters_visited);
1621 8101 av_freep(&filters_stack);
1622 8101 return ret;
1623 }
1624
1625 8718 static int start_prepare(Scheduler *sch)
1626 {
1627 int ret;
1628
1629
2/2
✓ Branch 0 taken 7623 times.
✓ Branch 1 taken 8718 times.
 16341 for (unsigned i = 0; i < sch->nb_demux; i++) {
1630 7623 SchDemux *d = &sch->demux[i];
1631
1632
2/2
✓ Branch 0 taken 7903 times.
✓ Branch 1 taken 7623 times.
 15526 for (unsigned j = 0; j < d->nb_streams; j++) {
1633 7903 SchDemuxStream *ds = &d->streams[j];
1634
1635
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7903 times.
 7903 if (!ds->nb_dst) {
1636 ✗ av_log(d, AV_LOG_ERROR,
1637 "Demuxer stream %u not connected to any sink\n", j);
1638 ✗ return AVERROR(EINVAL);
1639 }
1640
1641 7903 ds->dst_finished = av_calloc(ds->nb_dst, sizeof(*ds->dst_finished));
1642
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7903 times.
 7903 if (!ds->dst_finished)
1643 ✗ return AVERROR(ENOMEM);
1644 }
1645 }
1646
1647
2/2
✓ Branch 0 taken 7093 times.
✓ Branch 1 taken 8718 times.
 15811 for (unsigned i = 0; i < sch->nb_dec; i++) {
1648 7093 SchDec *dec = &sch->dec[i];
1649
1650
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7093 times.
 7093 if (!dec->src.type) {
1651 ✗ av_log(dec, AV_LOG_ERROR,
1652 "Decoder not connected to a source\n");
1653 ✗ return AVERROR(EINVAL);
1654 }
1655
1656
2/2
✓ Branch 0 taken 7099 times.
✓ Branch 1 taken 7093 times.
 14192 for (unsigned j = 0; j < dec->nb_outputs; j++) {
1657 7099 SchDecOutput *o = &dec->outputs[j];
1658
1659
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7099 times.
 7099 if (!o->nb_dst) {
1660 ✗ av_log(dec, AV_LOG_ERROR,
1661 "Decoder output %u not connected to any sink\n", j);
1662 ✗ return AVERROR(EINVAL);
1663 }
1664
1665 7099 o->dst_finished = av_calloc(o->nb_dst, sizeof(*o->dst_finished));
1666
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7099 times.
 7099 if (!o->dst_finished)
1667 ✗ return AVERROR(ENOMEM);
1668 }
1669 }
1670
1671
2/2
✓ Branch 0 taken 8391 times.
✓ Branch 1 taken 8718 times.
 17109 for (unsigned i = 0; i < sch->nb_enc; i++) {
1672 8391 SchEnc *enc = &sch->enc[i];
1673
1674
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8391 times.
 8391 if (!enc->src.type) {
1675 ✗ av_log(enc, AV_LOG_ERROR,
1676 "Encoder not connected to a source\n");
1677 ✗ return AVERROR(EINVAL);
1678 }
1679
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8391 times.
 8391 if (!enc->nb_dst) {
1680 ✗ av_log(enc, AV_LOG_ERROR,
1681 "Encoder not connected to any sink\n");
1682 ✗ return AVERROR(EINVAL);
1683 }
1684
1685 8391 enc->dst_finished = av_calloc(enc->nb_dst, sizeof(*enc->dst_finished));
1686
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8391 times.
 8391 if (!enc->dst_finished)
1687 ✗ return AVERROR(ENOMEM);
1688 }
1689
1690
2/2
✓ Branch 0 taken 8722 times.
✓ Branch 1 taken 8718 times.
 17440 for (unsigned i = 0; i < sch->nb_mux; i++) {
1691 8722 SchMux *mux = &sch->mux[i];
1692
1693
2/2
✓ Branch 0 taken 9228 times.
✓ Branch 1 taken 8722 times.
 17950 for (unsigned j = 0; j < mux->nb_streams; j++) {
1694 9228 SchMuxStream *ms = &mux->streams[j];
1695
1696
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9228 times.
 9228 if (!ms->src.type) {
1697 ✗ av_log(mux, AV_LOG_ERROR,
1698 "Muxer stream #%u not connected to a source\n", j);
1699 ✗ return AVERROR(EINVAL);
1700 }
1701 }
1702
1703 8722 ret = queue_alloc(&mux->queue, mux->nb_streams, mux->queue_size,
1704 QUEUE_PACKETS);
1705
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8722 times.
 8722 if (ret < 0)
1706 ✗ return ret;
1707 }
1708
1709
2/2
✓ Branch 0 taken 8218 times.
✓ Branch 1 taken 8718 times.
 16936 for (unsigned i = 0; i < sch->nb_filters; i++) {
1710 8218 SchFilterGraph *fg = &sch->filters[i];
1711
1712
2/2
✓ Branch 0 taken 7122 times.
✓ Branch 1 taken 8218 times.
 15340 for (unsigned j = 0; j < fg->nb_inputs; j++) {
1713 7122 SchFilterIn *fi = &fg->inputs[j];
1714
1715
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7122 times.
 7122 if (!fi->src.type) {
1716 ✗ av_log(fg, AV_LOG_ERROR,
1717 "Filtergraph input %u not connected to a source\n", j);
1718 ✗ return AVERROR(EINVAL);
1719 }
1720 }
1721
1722
2/2
✓ Branch 0 taken 8350 times.
✓ Branch 1 taken 8218 times.
 16568 for (unsigned j = 0; j < fg->nb_outputs; j++) {
1723 8350 SchFilterOut *fo = &fg->outputs[j];
1724
1725
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8350 times.
 8350 if (!fo->dst.type) {
1726 ✗ av_log(fg, AV_LOG_ERROR,
1727 "Filtergraph %u output %u not connected to a sink\n", i, j);
1728 ✗ return AVERROR(EINVAL);
1729 }
1730 }
1731 }
1732
1733 // Check that the transcoding graph has no cycles.
1734 8718 ret = check_acyclic(sch);
1735
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8718 times.
 8718 if (ret < 0)
1736 ✗ return ret;
1737
1738 8718 return 0;
1739 }
1740
1741 8718 int sch_start(Scheduler *sch)
1742 {
1743 int ret;
1744
1745 8718 ret = start_prepare(sch);
1746
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8718 times.
 8718 if (ret < 0)
1747 ✗ return ret;
1748
1749
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8718 times.
 8718 av_assert0(sch->state == SCH_STATE_UNINIT);
1750 8718 sch->state = SCH_STATE_STARTED;
1751
1752
2/2
✓ Branch 0 taken 8722 times.
✓ Branch 1 taken 8718 times.
 17440 for (unsigned i = 0; i < sch->nb_mux; i++) {
1753 8722 SchMux *mux = &sch->mux[i];
1754
1755
2/2
✓ Branch 0 taken 583 times.
✓ Branch 1 taken 8139 times.
 8722 if (mux->nb_streams_ready == mux->nb_streams) {
1756 583 ret = mux_init(sch, mux);
1757
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 583 times.
 583 if (ret < 0)
1758 ✗ goto fail;
1759 }
1760 }
1761
1762
2/2
✓ Branch 0 taken 8391 times.
✓ Branch 1 taken 8718 times.
 17109 for (unsigned i = 0; i < sch->nb_enc; i++) {
1763 8391 SchEnc *enc = &sch->enc[i];
1764
1765 8391 ret = task_start(&enc->task);
1766
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8391 times.
 8391 if (ret < 0)
1767 ✗ goto fail;
1768 }
1769
1770
2/2
✓ Branch 0 taken 8218 times.
✓ Branch 1 taken 8718 times.
 16936 for (unsigned i = 0; i < sch->nb_filters; i++) {
1771 8218 SchFilterGraph *fg = &sch->filters[i];
1772
1773 8218 ret = task_start(&fg->task);
1774
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8218 times.
 8218 if (ret < 0)
1775 ✗ goto fail;
1776 }
1777
1778
2/2
✓ Branch 0 taken 7093 times.
✓ Branch 1 taken 8718 times.
 15811 for (unsigned i = 0; i < sch->nb_dec; i++) {
1779 7093 SchDec *dec = &sch->dec[i];
1780
1781 7093 ret = task_start(&dec->task);
1782
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7093 times.
 7093 if (ret < 0)
1783 ✗ goto fail;
1784 }
1785
1786
2/2
✓ Branch 0 taken 7623 times.
✓ Branch 1 taken 8718 times.
 16341 for (unsigned i = 0; i < sch->nb_demux; i++) {
1787 7623 SchDemux *d = &sch->demux[i];
1788
1789
2/2
✓ Branch 0 taken 15 times.
✓ Branch 1 taken 7608 times.
 7623 if (!d->nb_streams)
1790 15 continue;
1791
1792 7608 ret = task_start(&d->task);
1793
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7608 times.
 7608 if (ret < 0)
1794 ✗ goto fail;
1795 }
1796
1797 8718 pthread_mutex_lock(&sch->schedule_lock);
1798 8718 schedule_update_locked(sch);
1799 8718 pthread_mutex_unlock(&sch->schedule_lock);
1800
1801 8718 return 0;
1802 ✗ fail:
1803 ✗ sch_stop(sch, NULL);
1804 ✗ return ret;
1805 }
1806
1807 27510 int sch_wait(Scheduler *sch, uint64_t timeout_us, int64_t *transcode_ts)
1808 {
1809 int ret;
1810
1811 // convert delay to absolute timestamp
1812 27510 timeout_us += av_gettime();
1813
1814 27510 pthread_mutex_lock(&sch->finish_lock);
1815
1816
1/2
✓ Branch 0 taken 27510 times.
✗ Branch 1 not taken.
 27510 if (sch->nb_mux_done < sch->nb_mux) {
1817 27510 struct timespec tv = { .tv_sec = timeout_us / 1000000,
1818 27510 .tv_nsec = (timeout_us % 1000000) * 1000 };
1819 27510 pthread_cond_timedwait(&sch->finish_cond, &sch->finish_lock, &tv);
1820 }
1821
1822 // abort transcoding if any task failed
1823
4/4
✓ Branch 0 taken 18795 times.
✓ Branch 1 taken 8715 times.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 18793 times.
 27510 ret = sch->nb_mux_done == sch->nb_mux || sch->task_failed;
1824
1825 27510 pthread_mutex_unlock(&sch->finish_lock);
1826
1827 27510 *transcode_ts = atomic_load(&sch->last_dts);
1828
1829 27510 return ret;
1830 }
1831
1832 8348 static int enc_open(Scheduler *sch, SchEnc *enc, const AVFrame *frame)
1833 {
1834 int ret;
1835
1836 8348 ret = enc->open_cb(enc->task.func_arg, frame);
1837
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8348 times.
 8348 if (ret < 0)
1838 ✗ return ret;
1839
1840 // ret>0 signals audio frame size, which means sync queue must
1841 // have been enabled during encoder creation
1842
2/2
✓ Branch 0 taken 218 times.
✓ Branch 1 taken 8130 times.
 8348 if (ret > 0) {
1843 SchSyncQueue *sq;
1844
1845
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 218 times.
 218 av_assert0(enc->sq_idx[0] >= 0);
1846 218 sq = &sch->sq_enc[enc->sq_idx[0]];
1847
1848 218 pthread_mutex_lock(&sq->lock);
1849
1850 218 sq_frame_samples(sq->sq, enc->sq_idx[1], ret);
1851
1852 218 pthread_mutex_unlock(&sq->lock);
1853 }
1854
1855 8348 return 0;
1856 }
1857
1858 554533 static int send_to_enc_thread(Scheduler *sch, SchEnc *enc, AVFrame *frame)
1859 {
1860 int ret;
1861
1862
2/2
✓ Branch 0 taken 20382 times.
✓ Branch 1 taken 534151 times.
 554533 if (!frame) {
1863 20382 tq_send_finish(enc->queue, 0);
1864 20382 return 0;
1865 }
1866
1867
2/2
✓ Branch 0 taken 22 times.
✓ Branch 1 taken 534129 times.
 534151 if (enc->in_finished)
1868 22 return AVERROR_EOF;
1869
1870 534129 ret = tq_send(enc->queue, 0, frame);
1871
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 534125 times.
 534129 if (ret < 0)
1872 4 enc->in_finished = 1;
1873
1874 534129 return ret;
1875 }
1876
1877 41349 static int send_to_enc_sq(Scheduler *sch, SchEnc *enc, AVFrame *frame)
1878 {
1879 41349 SchSyncQueue *sq = &sch->sq_enc[enc->sq_idx[0]];
1880 41349 int ret = 0;
1881
1882 // inform the scheduling code that no more input will arrive along this path;
1883 // this is necessary because the sync queue may not send an EOF downstream
1884 // until other streams finish
1885 // TODO: consider a cleaner way of passing this information through
1886 // the pipeline
1887
2/2
✓ Branch 0 taken 6777 times.
✓ Branch 1 taken 34572 times.
 41349 if (!frame) {
1888
2/2
✓ Branch 0 taken 6777 times.
✓ Branch 1 taken 6777 times.
 13554 for (unsigned i = 0; i < enc->nb_dst; i++) {
1889 SchMux *mux;
1890 SchMuxStream *ms;
1891
1892
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6777 times.
 6777 if (enc->dst[i].type != SCH_NODE_TYPE_MUX)
1893 ✗ continue;
1894
1895 6777 mux = &sch->mux[enc->dst[i].idx];
1896 6777 ms = &mux->streams[enc->dst[i].idx_stream];
1897
1898 6777 pthread_mutex_lock(&sch->schedule_lock);
1899
1900 6777 ms->source_finished = 1;
1901 6777 schedule_update_locked(sch);
1902
1903 6777 pthread_mutex_unlock(&sch->schedule_lock);
1904 }
1905 }
1906
1907 41349 pthread_mutex_lock(&sq->lock);
1908
1909 41349 ret = sq_send(sq->sq, enc->sq_idx[1], SQFRAME(frame));
1910
2/2
✓ Branch 0 taken 41348 times.
✓ Branch 1 taken 1 times.
 41349 if (ret < 0)
1911 1 goto finish;
1912
1913 117990 while (1) {
1914 SchEnc *enc;
1915
1916 // TODO: the SQ API should be extended to allow returning EOF
1917 // for individual streams
1918 159338 ret = sq_receive(sq->sq, -1, SQFRAME(sq->frame));
1919
2/2
✓ Branch 0 taken 41348 times.
✓ Branch 1 taken 117990 times.
 159338 if (ret < 0) {
1920
2/2
✓ Branch 0 taken 9779 times.
✓ Branch 1 taken 31569 times.
 41348 ret = (ret == AVERROR(EAGAIN)) ? 0 : ret;
1921 41348 break;
1922 }
1923
1924 117990 enc = &sch->enc[sq->enc_idx[ret]];
1925 117990 ret = send_to_enc_thread(sch, enc, sq->frame);
1926
2/2
✓ Branch 0 taken 23 times.
✓ Branch 1 taken 117967 times.
 117990 if (ret < 0) {
1927 23 av_frame_unref(sq->frame);
1928
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 23 times.
 23 if (ret != AVERROR_EOF)
1929 ✗ break;
1930
1931 23 sq_send(sq->sq, enc->sq_idx[1], SQFRAME(NULL));
1932 23 continue;
1933 }
1934 }
1935
1936
2/2
✓ Branch 0 taken 31569 times.
✓ Branch 1 taken 9779 times.
 41348 if (ret < 0) {
1937 // close all encoders fed from this sync queue
1938
2/2
✓ Branch 0 taken 10259 times.
✓ Branch 1 taken 9779 times.
 20038 for (unsigned i = 0; i < sq->nb_enc_idx; i++) {
1939 10259 int err = send_to_enc_thread(sch, &sch->enc[sq->enc_idx[i]], NULL);
1940
1941 // if the sync queue error is EOF and closing the encoder
1942 // produces a more serious error, make sure to pick the latter
1943
2/4
✓ Branch 0 taken 10259 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 10259 times.
✗ Branch 3 not taken.
 10259 ret = err_merge((ret == AVERROR_EOF && err < 0) ? 0 : ret, err);
1944 }
1945 }
1946
1947 41348 finish:
1948 41349 pthread_mutex_unlock(&sq->lock);
1949
1950 41349 return ret;
1951 }
1952
1953 467636 static int send_to_enc(Scheduler *sch, SchEnc *enc, AVFrame *frame)
1954 {
1955
6/6
✓ Branch 0 taken 466551 times.
✓ Branch 1 taken 1085 times.
✓ Branch 2 taken 449693 times.
✓ Branch 3 taken 16858 times.
✓ Branch 4 taken 8348 times.
✓ Branch 5 taken 441345 times.
 467636 if (enc->open_cb && frame && !enc->opened) {
1956 8348 int ret = enc_open(sch, enc, frame);
1957
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8348 times.
 8348 if (ret < 0)
1958 ✗ return ret;
1959 8348 enc->opened = 1;
1960
1961 // discard empty frames that only carry encoder init parameters
1962
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 8345 times.
 8348 if (!frame->buf[0]) {
1963 3 av_frame_unref(frame);
1964 3 return 0;
1965 }
1966 }
1967
1968 467633 return (enc->sq_idx[0] >= 0) ?
1969
2/2
✓ Branch 0 taken 41349 times.
✓ Branch 1 taken 426284 times.
 893917 send_to_enc_sq (sch, enc, frame) :
1970 426284 send_to_enc_thread(sch, enc, frame);
1971 }
1972
1973 3829 static int mux_queue_packet(SchMux *mux, SchMuxStream *ms, AVPacket *pkt)
1974 {
1975 3829 PreMuxQueue *q = &ms->pre_mux_queue;
1976 3829 AVPacket *tmp_pkt = NULL;
1977 int ret;
1978
1979
2/2
✓ Branch 1 taken 96 times.
✓ Branch 2 taken 3733 times.
 3829 if (!av_fifo_can_write(q->fifo)) {
1980 96 size_t packets = av_fifo_can_read(q->fifo);
1981
1/2
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
 96 size_t pkt_size = pkt ? pkt->size : 0;
1982 96 int thresh_reached = (q->data_size + pkt_size) > q->data_threshold;
1983
2/2
✓ Branch 0 taken 90 times.
✓ Branch 1 taken 6 times.
 96 size_t max_packets = thresh_reached ? q->max_packets : SIZE_MAX;
1984 96 size_t new_size = FFMIN(2 * packets, max_packets);
1985
1986
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 96 times.
 96 if (new_size <= packets) {
1987 ✗ av_log(mux, AV_LOG_ERROR,
1988 "Too many packets buffered for output stream.\n");
1989 ✗ return AVERROR_BUFFER_TOO_SMALL;
1990 }
1991 96 ret = av_fifo_grow2(q->fifo, new_size - packets);
1992
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 96 times.
 96 if (ret < 0)
1993 ✗ return ret;
1994 }
1995
1996
2/2
✓ Branch 0 taken 3785 times.
✓ Branch 1 taken 44 times.
 3829 if (pkt) {
1997 3785 tmp_pkt = av_packet_alloc();
1998
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3785 times.
 3785 if (!tmp_pkt)
1999 ✗ return AVERROR(ENOMEM);
2000
2001 3785 av_packet_move_ref(tmp_pkt, pkt);
2002 3785 q->data_size += tmp_pkt->size;
2003 }
2004 3829 av_fifo_write(q->fifo, &tmp_pkt, 1);
2005
2006 3829 return 0;
2007 }
2008
2009 614262 static int send_to_mux(Scheduler *sch, SchMux *mux, unsigned stream_idx,
2010 AVPacket *pkt)
2011 {
2012 614262 SchMuxStream *ms = &mux->streams[stream_idx];
2013
2/2
✓ Branch 0 taken 596675 times.
✓ Branch 1 taken 8359 times.
 605034 int64_t dts = (pkt && pkt->dts != AV_NOPTS_VALUE) ?
2014
2/2
✓ Branch 0 taken 605034 times.
✓ Branch 1 taken 9228 times.
 1219296 av_rescale_q(pkt->dts + pkt->duration, pkt->time_base, AV_TIME_BASE_Q) :
2015 AV_NOPTS_VALUE;
2016
2017 // queue the packet if the muxer cannot be started yet
2018
2/2
✓ Branch 0 taken 3860 times.
✓ Branch 1 taken 610402 times.
 614262 if (!atomic_load(&mux->mux_started)) {
2019 3860 int queued = 0;
2020
2021 // the muxer could have started between the above atomic check and
2022 // locking the mutex, then this block falls through to normal send path
2023 3860 pthread_mutex_lock(&sch->mux_ready_lock);
2024
2025
2/2
✓ Branch 0 taken 3829 times.
✓ Branch 1 taken 31 times.
 3860 if (!atomic_load(&mux->mux_started)) {
2026 3829 int ret = mux_queue_packet(mux, ms, pkt);
2027
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3829 times.
 3829 queued = ret < 0 ? ret : 1;
2028 }
2029
2030 3860 pthread_mutex_unlock(&sch->mux_ready_lock);
2031
2032
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3860 times.
 3860 if (queued < 0)
2033 ✗ return queued;
2034
2/2
✓ Branch 0 taken 3829 times.
✓ Branch 1 taken 31 times.
 3860 else if (queued)
2035 3829 goto update_schedule;
2036 }
2037
2038
2/2
✓ Branch 0 taken 601249 times.
✓ Branch 1 taken 9184 times.
 610433 if (pkt) {
2039 int ret;
2040
2041
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 601248 times.
 601249 if (ms->init_eof)
2042 1 return AVERROR_EOF;
2043
2044 601248 ret = tq_send(mux->queue, stream_idx, pkt);
2045
2/2
✓ Branch 0 taken 83 times.
✓ Branch 1 taken 601165 times.
 601248 if (ret < 0)
2046 83 return ret;
2047 } else
2048 9184 tq_send_finish(mux->queue, stream_idx);
2049
2050 614178 update_schedule:
2051 // TODO: use atomics to check whether this changes trailing dts
2052 // to avoid locking unnecessarily
2053
4/4
✓ Branch 0 taken 17587 times.
✓ Branch 1 taken 596591 times.
✓ Branch 2 taken 9228 times.
✓ Branch 3 taken 8359 times.
 614178 if (dts != AV_NOPTS_VALUE || !pkt) {
2054 605819 pthread_mutex_lock(&sch->schedule_lock);
2055
2056
2/2
✓ Branch 0 taken 596591 times.
✓ Branch 1 taken 9228 times.
 605819 if (pkt) ms->last_dts = dts;
2057 9228 else ms->source_finished = 1;
2058
2059 605819 schedule_update_locked(sch);
2060
2061 605819 pthread_mutex_unlock(&sch->schedule_lock);
2062 }
2063
2064 614178 return 0;
2065 }
2066
2067 static int
2068 529894 demux_stream_send_to_dst(Scheduler *sch, const SchedulerNode dst,
2069 uint8_t *dst_finished, AVPacket *pkt, unsigned flags)
2070 {
2071 int ret;
2072
2073
2/2
✓ Branch 0 taken 3245 times.
✓ Branch 1 taken 526649 times.
 529894 if (*dst_finished)
2074 3245 return AVERROR_EOF;
2075
2076
4/4
✓ Branch 0 taken 521965 times.
✓ Branch 1 taken 4684 times.
✓ Branch 2 taken 77047 times.
✓ Branch 3 taken 444918 times.
 526649 if (pkt && dst.type == SCH_NODE_TYPE_MUX &&
2077
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 77045 times.
 77047 (flags & DEMUX_SEND_STREAMCOPY_EOF)) {
2078 2 av_packet_unref(pkt);
2079 2 pkt = NULL;
2080 }
2081
2082
2/2
✓ Branch 0 taken 4686 times.
✓ Branch 1 taken 521963 times.
 526649 if (!pkt)
2083 4686 goto finish;
2084
2085 1043926 ret = (dst.type == SCH_NODE_TYPE_MUX) ?
2086
2/2
✓ Branch 0 taken 77045 times.
✓ Branch 1 taken 444918 times.
 521963 send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, pkt) :
2087 444918 tq_send(sch->dec[dst.idx].queue, 0, pkt);
2088
2/2
✓ Branch 0 taken 3243 times.
✓ Branch 1 taken 518720 times.
 521963 if (ret == AVERROR_EOF)
2089 3243 goto finish;
2090
2091 518720 return ret;
2092
2093 7929 finish:
2094
2/2
✓ Branch 0 taken 837 times.
✓ Branch 1 taken 7092 times.
 7929 if (dst.type == SCH_NODE_TYPE_MUX)
2095 837 send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, NULL);
2096 else
2097 7092 tq_send_finish(sch->dec[dst.idx].queue, 0);
2098
2099 7929 *dst_finished = 1;
2100 7929 return AVERROR_EOF;
2101 }
2102
2103 529425 static int demux_send_for_stream(Scheduler *sch, SchDemux *d, SchDemuxStream *ds,
2104 AVPacket *pkt, unsigned flags)
2105 {
2106 529425 unsigned nb_done = 0;
2107
2108
2/2
✓ Branch 0 taken 529894 times.
✓ Branch 1 taken 529425 times.
 1059319 for (unsigned i = 0; i < ds->nb_dst; i++) {
2109 529894 AVPacket *to_send = pkt;
2110 529894 uint8_t *finished = &ds->dst_finished[i];
2111
2112 int ret;
2113
2114 // sending a packet consumes it, so make a temporary reference if needed
2115
4/4
✓ Branch 0 taken 521965 times.
✓ Branch 1 taken 7929 times.
✓ Branch 2 taken 443 times.
✓ Branch 3 taken 521522 times.
 529894 if (pkt && i < ds->nb_dst - 1) {
2116 443 to_send = d->send_pkt;
2117
2118 443 ret = av_packet_ref(to_send, pkt);
2119
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 443 times.
 443 if (ret < 0)
2120 ✗ return ret;
2121 }
2122
2123 529894 ret = demux_stream_send_to_dst(sch, ds->dst[i], finished, to_send, flags);
2124
2/2
✓ Branch 0 taken 521965 times.
✓ Branch 1 taken 7929 times.
 529894 if (to_send)
2125 521965 av_packet_unref(to_send);
2126
2/2
✓ Branch 0 taken 11174 times.
✓ Branch 1 taken 518720 times.
 529894 if (ret == AVERROR_EOF)
2127 11174 nb_done++;
2128
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 518720 times.
 518720 else if (ret < 0)
2129 ✗ return ret;
2130 }
2131
2132
2/2
✓ Branch 0 taken 11147 times.
✓ Branch 1 taken 518278 times.
 529425 return (nb_done == ds->nb_dst) ? AVERROR_EOF : 0;
2133 }
2134
2135 11 static int demux_flush(Scheduler *sch, SchDemux *d, AVPacket *pkt)
2136 {
2137 11 Timestamp max_end_ts = (Timestamp){ .ts = AV_NOPTS_VALUE };
2138
2139
3/6
✓ Branch 0 taken 11 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 11 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 11 times.
 11 av_assert0(!pkt->buf && !pkt->data && !pkt->side_data_elems);
2140
2141
2/2
✓ Branch 0 taken 14 times.
✓ Branch 1 taken 11 times.
 25 for (unsigned i = 0; i < d->nb_streams; i++) {
2142 14 SchDemuxStream *ds = &d->streams[i];
2143
2144
2/2
✓ Branch 0 taken 14 times.
✓ Branch 1 taken 14 times.
 28 for (unsigned j = 0; j < ds->nb_dst; j++) {
2145 14 const SchedulerNode *dst = &ds->dst[j];
2146 SchDec *dec;
2147 int ret;
2148
2149
3/4
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 8 times.
✓ Branch 3 taken 6 times.
 14 if (ds->dst_finished[j] || dst->type != SCH_NODE_TYPE_DEC)
2150 8 continue;
2151
2152 6 dec = &sch->dec[dst->idx];
2153
2154 6 ret = tq_send(dec->queue, 0, pkt);
2155
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret < 0)
2156 ✗ return ret;
2157
2158
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 3 times.
 6 if (dec->queue_end_ts) {
2159 Timestamp ts;
2160 3 ret = av_thread_message_queue_recv(dec->queue_end_ts, &ts, 0);
2161
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3 if (ret < 0)
2162 ✗ return ret;
2163
2164
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3 if (max_end_ts.ts == AV_NOPTS_VALUE ||
2165 ✗ (ts.ts != AV_NOPTS_VALUE &&
2166 ✗ av_compare_ts(max_end_ts.ts, max_end_ts.tb, ts.ts, ts.tb) < 0))
2167 3 max_end_ts = ts;
2168
2169 }
2170 }
2171 }
2172
2173 11 pkt->pts = max_end_ts.ts;
2174 11 pkt->time_base = max_end_ts.tb;
2175
2176 11 return 0;
2177 }
2178
2179 521955 int sch_demux_send(Scheduler *sch, unsigned demux_idx, AVPacket *pkt,
2180 unsigned flags)
2181 {
2182 SchDemux *d;
2183 int terminate;
2184
2185
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 521955 times.
 521955 av_assert0(demux_idx < sch->nb_demux);
2186 521955 d = &sch->demux[demux_idx];
2187
2188 521955 terminate = waiter_wait(sch, &d->waiter);
2189
2/2
✓ Branch 0 taken 422 times.
✓ Branch 1 taken 521533 times.
 521955 if (terminate)
2190 422 return AVERROR_EXIT;
2191
2192 // flush the downstreams after seek
2193
2/2
✓ Branch 0 taken 11 times.
✓ Branch 1 taken 521522 times.
 521533 if (pkt->stream_index == -1)
2194 11 return demux_flush(sch, d, pkt);
2195
2196
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 521522 times.
 521522 av_assert0(pkt->stream_index < d->nb_streams);
2197
2198 521522 return demux_send_for_stream(sch, d, &d->streams[pkt->stream_index], pkt, flags);
2199 }
2200
2201 7623 static int demux_done(Scheduler *sch, unsigned demux_idx)
2202 {
2203 7623 SchDemux *d = &sch->demux[demux_idx];
2204 7623 int ret = 0;
2205
2206
2/2
✓ Branch 0 taken 7903 times.
✓ Branch 1 taken 7623 times.
 15526 for (unsigned i = 0; i < d->nb_streams; i++) {
2207 7903 int err = demux_send_for_stream(sch, d, &d->streams[i], NULL, 0);
2208
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7903 times.
 7903 if (err != AVERROR_EOF)
2209 ✗ ret = err_merge(ret, err);
2210 }
2211
2212 7623 pthread_mutex_lock(&sch->schedule_lock);
2213
2214 7623 d->task_exited = 1;
2215
2216 7623 schedule_update_locked(sch);
2217
2218 7623 pthread_mutex_unlock(&sch->schedule_lock);
2219
2220 7623 return ret;
2221 }
2222
2223 621409 int sch_mux_receive(Scheduler *sch, unsigned mux_idx, AVPacket *pkt)
2224 {
2225 SchMux *mux;
2226 int ret, stream_idx;
2227
2228
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 621409 times.
 621409 av_assert0(mux_idx < sch->nb_mux);
2229 621409 mux = &sch->mux[mux_idx];
2230
2231 621409 ret = tq_receive(mux->queue, &stream_idx, pkt, 0);
2232 621409 pkt->stream_index = stream_idx;
2233 621409 return ret;
2234 }
2235
2236 245 void sch_mux_receive_finish(Scheduler *sch, unsigned mux_idx, unsigned stream_idx)
2237 {
2238 SchMux *mux;
2239
2240
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 245 times.
 245 av_assert0(mux_idx < sch->nb_mux);
2241 245 mux = &sch->mux[mux_idx];
2242
2243
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 245 times.
 245 av_assert0(stream_idx < mux->nb_streams);
2244 245 tq_receive_finish(mux->queue, stream_idx);
2245
2246 245 pthread_mutex_lock(&sch->schedule_lock);
2247 245 mux->streams[stream_idx].source_finished = 1;
2248
2249 245 schedule_update_locked(sch);
2250
2251 245 pthread_mutex_unlock(&sch->schedule_lock);
2252 245 }
2253
2254 563173 int sch_mux_sub_heartbeat(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
2255 const AVPacket *pkt)
2256 {
2257 SchMux *mux;
2258 SchMuxStream *ms;
2259
2260
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 563173 times.
 563173 av_assert0(mux_idx < sch->nb_mux);
2261 563173 mux = &sch->mux[mux_idx];
2262
2263
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 563173 times.
 563173 av_assert0(stream_idx < mux->nb_streams);
2264 563173 ms = &mux->streams[stream_idx];
2265
2266
2/2
✓ Branch 0 taken 5 times.
✓ Branch 1 taken 563173 times.
 563178 for (unsigned i = 0; i < ms->nb_sub_heartbeat_dst; i++) {
2267 5 SchDec *dst = &sch->dec[ms->sub_heartbeat_dst[i]];
2268 int ret;
2269
2270 5 ret = av_packet_copy_props(mux->sub_heartbeat_pkt, pkt);
2271
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5 times.
 5 if (ret < 0)
2272 ✗ return ret;
2273
2274 5 tq_send(dst->queue, 0, mux->sub_heartbeat_pkt);
2275 }
2276
2277 563173 return 0;
2278 }
2279
2280 8722 static int mux_done(Scheduler *sch, unsigned mux_idx)
2281 {
2282 8722 SchMux *mux = &sch->mux[mux_idx];
2283
2284 8722 pthread_mutex_lock(&sch->schedule_lock);
2285
2286
2/2
✓ Branch 0 taken 9228 times.
✓ Branch 1 taken 8722 times.
 17950 for (unsigned i = 0; i < mux->nb_streams; i++) {
2287 9228 tq_receive_finish(mux->queue, i);
2288 9228 mux->streams[i].source_finished = 1;
2289 }
2290
2291 8722 schedule_update_locked(sch);
2292
2293 8722 pthread_mutex_unlock(&sch->schedule_lock);
2294
2295 8722 pthread_mutex_lock(&sch->finish_lock);
2296
2297
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8722 times.
 8722 av_assert0(sch->nb_mux_done < sch->nb_mux);
2298 8722 sch->nb_mux_done++;
2299
2300 8722 pthread_cond_signal(&sch->finish_cond);
2301
2302 8722 pthread_mutex_unlock(&sch->finish_lock);
2303
2304 8722 return 0;
2305 }
2306
2307 417055 int sch_dec_receive(Scheduler *sch, unsigned dec_idx, AVPacket *pkt)
2308 {
2309 SchDec *dec;
2310 int ret, dummy;
2311
2312
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 417055 times.
 417055 av_assert0(dec_idx < sch->nb_dec);
2313 417055 dec = &sch->dec[dec_idx];
2314
2315 // the decoder should have given us post-flush end timestamp in pkt
2316
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 417052 times.
 417055 if (dec->expect_end_ts) {
2317 3 Timestamp ts = (Timestamp){ .ts = pkt->pts, .tb = pkt->time_base };
2318 3 ret = av_thread_message_queue_send(dec->queue_end_ts, &ts, 0);
2319
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3 if (ret < 0)
2320 ✗ return ret;
2321
2322 3 dec->expect_end_ts = 0;
2323 }
2324
2325 417055 ret = tq_receive(dec->queue, &dummy, pkt, 0);
2326
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 417055 times.
 417055 av_assert0(dummy <= 0);
2327
2328 // got a flush packet, on the next call to this function the decoder
2329 // will give us post-flush end timestamp
2330
7/8
✓ Branch 0 taken 413555 times.
✓ Branch 1 taken 3500 times.
✓ Branch 2 taken 958 times.
✓ Branch 3 taken 412597 times.
✓ Branch 4 taken 958 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3 times.
✓ Branch 7 taken 955 times.
 417055 if (ret >= 0 && !pkt->data && !pkt->side_data_elems && dec->queue_end_ts)
2331 3 dec->expect_end_ts = 1;
2332
2333 417055 return ret;
2334 }
2335
2336 445869 static int send_to_filter(Scheduler *sch, SchFilterGraph *fg,
2337 unsigned in_idx, AVFrame *frame)
2338 {
2339
2/2
✓ Branch 0 taken 438746 times.
✓ Branch 1 taken 7123 times.
 445869 if (frame)
2340 438746 return tq_send(fg->queue, in_idx, frame);
2341
2342 7123 pthread_mutex_lock(&sch->schedule_lock);
2343
2344
2/2
✓ Branch 0 taken 7122 times.
✓ Branch 1 taken 1 times.
 7123 if (!fg->inputs[in_idx].send_finished) {
2345 7122 fg->inputs[in_idx].send_finished = 1;
2346 7122 tq_send_finish(fg->queue, in_idx);
2347
2348 // close the control stream when all actual inputs are done
2349
2/2
✓ Branch 0 taken 7018 times.
✓ Branch 1 taken 104 times.
 7122 if (++fg->nb_inputs_finished_send == fg->nb_inputs)
2350 7018 tq_send_finish(fg->queue, fg->nb_inputs);
2351
2352 7122 schedule_update_locked(sch);
2353 }
2354
2355 7123 pthread_mutex_unlock(&sch->schedule_lock);
2356 7123 return 0;
2357 }
2358
2359 450656 static int dec_send_to_dst(Scheduler *sch, const SchedulerNode dst,
2360 uint8_t *dst_finished, AVFrame *frame)
2361 {
2362 int ret;
2363
2364
2/2
✓ Branch 0 taken 7352 times.
✓ Branch 1 taken 443304 times.
 450656 if (*dst_finished)
2365 7352 return AVERROR_EOF;
2366
2367
2/2
✓ Branch 0 taken 3516 times.
✓ Branch 1 taken 439788 times.
 443304 if (!frame)
2368 3516 goto finish;
2369
2370 879576 ret = (dst.type == SCH_NODE_TYPE_FILTER_IN) ?
2371
2/2
✓ Branch 0 taken 438745 times.
✓ Branch 1 taken 1043 times.
 439788 send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, frame) :
2372 1043 send_to_enc(sch, &sch->enc[dst.idx], frame);
2373
2/2
✓ Branch 0 taken 3647 times.
✓ Branch 1 taken 436141 times.
 439788 if (ret == AVERROR_EOF)
2374 3647 goto finish;
2375
2376 436141 return ret;
2377
2378 7163 finish:
2379
2/2
✓ Branch 0 taken 7121 times.
✓ Branch 1 taken 42 times.
 7163 if (dst.type == SCH_NODE_TYPE_FILTER_IN)
2380 7121 send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, NULL);
2381 else
2382 42 send_to_enc(sch, &sch->enc[dst.idx], NULL);
2383
2384 7163 *dst_finished = 1;
2385
2386 7163 return AVERROR_EOF;
2387 }
2388
2389 441865 int sch_dec_send(Scheduler *sch, unsigned dec_idx,
2390 unsigned out_idx, AVFrame *frame)
2391 {
2392 SchDec *dec;
2393 SchDecOutput *o;
2394 int ret;
2395 441865 unsigned nb_done = 0;
2396
2397
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 441865 times.
 441865 av_assert0(dec_idx < sch->nb_dec);
2398 441865 dec = &sch->dec[dec_idx];
2399
2400
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 441865 times.
 441865 av_assert0(out_idx < dec->nb_outputs);
2401 441865 o = &dec->outputs[out_idx];
2402
2403
2/2
✓ Branch 0 taken 443493 times.
✓ Branch 1 taken 441865 times.
 885358 for (unsigned i = 0; i < o->nb_dst; i++) {
2404 443493 uint8_t *finished = &o->dst_finished[i];
2405 443493 AVFrame *to_send = frame;
2406
2407 // sending a frame consumes it, so make a temporary reference if needed
2408
2/2
✓ Branch 0 taken 1628 times.
✓ Branch 1 taken 441865 times.
 443493 if (i < o->nb_dst - 1) {
2409 1628 to_send = dec->send_frame;
2410
2411 // frame may sometimes contain props only,
2412 // e.g. to signal EOF timestamp
2413
2/2
✓ Branch 0 taken 1518 times.
✓ Branch 1 taken 110 times.
 1628 ret = frame->buf[0] ? av_frame_ref(to_send, frame) :
2414 110 av_frame_copy_props(to_send, frame);
2415
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1628 times.
 1628 if (ret < 0)
2416 ✗ return ret;
2417 }
2418
2419 443493 ret = dec_send_to_dst(sch, o->dst[i], finished, to_send);
2420
2/2
✓ Branch 0 taken 7352 times.
✓ Branch 1 taken 436141 times.
 443493 if (ret < 0) {
2421 7352 av_frame_unref(to_send);
2422
1/2
✓ Branch 0 taken 7352 times.
✗ Branch 1 not taken.
 7352 if (ret == AVERROR_EOF) {
2423 7352 nb_done++;
2424 7352 continue;
2425 }
2426 ✗ return ret;
2427 }
2428 }
2429
2430
2/2
✓ Branch 0 taken 7193 times.
✓ Branch 1 taken 434672 times.
 441865 return (nb_done == o->nb_dst) ? AVERROR_EOF : 0;
2431 }
2432
2433 7093 static int dec_done(Scheduler *sch, unsigned dec_idx)
2434 {
2435 7093 SchDec *dec = &sch->dec[dec_idx];
2436 7093 int ret = 0;
2437
2438 7093 tq_receive_finish(dec->queue, 0);
2439
2440 // make sure our source does not get stuck waiting for end timestamps
2441 // that will never arrive
2442
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 7092 times.
 7093 if (dec->queue_end_ts)
2443 1 av_thread_message_queue_set_err_recv(dec->queue_end_ts, AVERROR_EOF);
2444
2445
2/2
✓ Branch 0 taken 7099 times.
✓ Branch 1 taken 7093 times.
 14192 for (unsigned i = 0; i < dec->nb_outputs; i++) {
2446 7099 SchDecOutput *o = &dec->outputs[i];
2447
2448
2/2
✓ Branch 0 taken 7163 times.
✓ Branch 1 taken 7099 times.
 14262 for (unsigned j = 0; j < o->nb_dst; j++) {
2449 7163 int err = dec_send_to_dst(sch, o->dst[j], &o->dst_finished[j], NULL);
2450
2/4
✓ Branch 0 taken 7163 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 7163 times.
 7163 if (err < 0 && err != AVERROR_EOF)
2451 ✗ ret = err_merge(ret, err);
2452 }
2453 }
2454
2455 7093 return ret;
2456 }
2457
2458 542509 int sch_enc_receive(Scheduler *sch, unsigned enc_idx, AVFrame *frame)
2459 {
2460 SchEnc *enc;
2461 int ret, dummy;
2462
2463
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 542509 times.
 542509 av_assert0(enc_idx < sch->nb_enc);
2464 542509 enc = &sch->enc[enc_idx];
2465
2466 542509 ret = tq_receive(enc->queue, &dummy, frame, 0);
2467
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 542509 times.
 542509 av_assert0(dummy <= 0);
2468
2469 542509 return ret;
2470 }
2471
2472 536457 static int enc_send_to_dst(Scheduler *sch, const SchedulerNode dst,
2473 uint8_t *dst_finished, AVPacket *pkt)
2474 {
2475 int ret;
2476
2477
2/2
✓ Branch 0 taken 28 times.
✓ Branch 1 taken 536429 times.
 536457 if (*dst_finished)
2478 28 return AVERROR_EOF;
2479
2480
2/2
✓ Branch 0 taken 8390 times.
✓ Branch 1 taken 528039 times.
 536429 if (!pkt)
2481 8390 goto finish;
2482
2483 1056078 ret = (dst.type == SCH_NODE_TYPE_MUX) ?
2484
2/2
✓ Branch 0 taken 527989 times.
✓ Branch 1 taken 50 times.
 528039 send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, pkt) :
2485 50 tq_send(sch->dec[dst.idx].queue, 0, pkt);
2486
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 528037 times.
 528039 if (ret == AVERROR_EOF)
2487 2 goto finish;
2488
2489 528037 return ret;
2490
2491 8392 finish:
2492
2/2
✓ Branch 0 taken 8391 times.
✓ Branch 1 taken 1 times.
 8392 if (dst.type == SCH_NODE_TYPE_MUX)
2493 8391 send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, NULL);
2494 else
2495 1 tq_send_finish(sch->dec[dst.idx].queue, 0);
2496
2497 8392 *dst_finished = 1;
2498
2499 8392 return AVERROR_EOF;
2500 }
2501
2502 528015 int sch_enc_send(Scheduler *sch, unsigned enc_idx, AVPacket *pkt)
2503 {
2504 SchEnc *enc;
2505 int ret;
2506
2507
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 528015 times.
 528015 av_assert0(enc_idx < sch->nb_enc);
2508 528015 enc = &sch->enc[enc_idx];
2509
2510
2/2
✓ Branch 0 taken 528065 times.
✓ Branch 1 taken 528015 times.
 1056080 for (unsigned i = 0; i < enc->nb_dst; i++) {
2511 528065 uint8_t *finished = &enc->dst_finished[i];
2512 528065 AVPacket *to_send = pkt;
2513
2514 // sending a packet consumes it, so make a temporary reference if needed
2515
2/2
✓ Branch 0 taken 50 times.
✓ Branch 1 taken 528015 times.
 528065 if (i < enc->nb_dst - 1) {
2516 50 to_send = enc->send_pkt;
2517
2518 50 ret = av_packet_ref(to_send, pkt);
2519
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 50 times.
 50 if (ret < 0)
2520 ✗ return ret;
2521 }
2522
2523 528065 ret = enc_send_to_dst(sch, enc->dst[i], finished, to_send);
2524
2/2
✓ Branch 0 taken 28 times.
✓ Branch 1 taken 528037 times.
 528065 if (ret < 0) {
2525 28 av_packet_unref(to_send);
2526
1/2
✓ Branch 0 taken 28 times.
✗ Branch 1 not taken.
 28 if (ret == AVERROR_EOF)
2527 28 continue;
2528 ✗ return ret;
2529 }
2530 }
2531
2532 528015 return 0;
2533 }
2534
2535 8391 static int enc_done(Scheduler *sch, unsigned enc_idx)
2536 {
2537 8391 SchEnc *enc = &sch->enc[enc_idx];
2538 8391 int ret = 0;
2539
2540 8391 tq_receive_finish(enc->queue, 0);
2541
2542
2/2
✓ Branch 0 taken 8392 times.
✓ Branch 1 taken 8391 times.
 16783 for (unsigned i = 0; i < enc->nb_dst; i++) {
2543 8392 int err = enc_send_to_dst(sch, enc->dst[i], &enc->dst_finished[i], NULL);
2544
2/4
✓ Branch 0 taken 8392 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 8392 times.
 8392 if (err < 0 && err != AVERROR_EOF)
2545 ✗ ret = err_merge(ret, err);
2546 }
2547
2548 8391 return ret;
2549 }
2550
2551 460723 int sch_filter_receive(Scheduler *sch, unsigned fg_idx,
2552 unsigned *in_idx, AVFrame *frame)
2553 {
2554 SchFilterGraph *fg;
2555 int ret, idx;
2556
2557
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 460723 times.
 460723 av_assert0(fg_idx < sch->nb_filters);
2558 460723 fg = &sch->filters[fg_idx];
2559
2560
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 460723 times.
 460723 av_assert0(*in_idx <= fg->nb_inputs);
2561
2562 // update scheduling to account for desired input stream, if it changed
2563 //
2564 // this check needs no locking because only the filtering thread
2565 // updates this value
2566
2/2
✓ Branch 0 taken 9049 times.
✓ Branch 1 taken 451674 times.
 460723 if (*in_idx != fg->best_input) {
2567 9049 pthread_mutex_lock(&sch->schedule_lock);
2568
2569 9049 fg->best_input = *in_idx;
2570 9049 schedule_update_locked(sch);
2571
2572 9049 pthread_mutex_unlock(&sch->schedule_lock);
2573 }
2574
2575
2/2
✓ Branch 0 taken 427979 times.
✓ Branch 1 taken 32744 times.
 460723 if (*in_idx == fg->nb_inputs) {
2576 // drain incoming frames before waiting, to avoid blocking downstream
2577 32744 ret = tq_receive(fg->queue, &idx, frame, THREAD_QUEUE_FLAG_NO_BLOCK);
2578
2/2
✓ Branch 0 taken 275 times.
✓ Branch 1 taken 32469 times.
 32744 if (ret >= 0) {
2579
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 275 times.
 275 av_assert0(idx >= 0);
2580 275 *in_idx = idx;
2581 275 return 0;
2582 }
2583
2584 32469 int terminate = waiter_wait(sch, &fg->waiter);
2585
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 32466 times.
 32469 return terminate ? AVERROR_EOF : AVERROR(EAGAIN);
2586 }
2587
2588 while (1) {
2589 428007 ret = tq_receive(fg->queue, &idx, frame, 0);
2590
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 427999 times.
 428007 if (idx < 0)
2591 8 return AVERROR_EOF;
2592
2/2
✓ Branch 0 taken 427971 times.
✓ Branch 1 taken 28 times.
 427999 else if (ret >= 0) {
2593 427971 *in_idx = idx;
2594 427971 return 0;
2595 }
2596
2597 // disregard EOFs for specific streams - they should always be
2598 // preceded by an EOF frame
2599 }
2600 }
2601
2602 408 void sch_filter_receive_finish(Scheduler *sch, unsigned fg_idx, unsigned in_idx)
2603 {
2604 SchFilterGraph *fg;
2605 SchFilterIn *fi;
2606
2607
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 408 times.
 408 av_assert0(fg_idx < sch->nb_filters);
2608 408 fg = &sch->filters[fg_idx];
2609
2610
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 408 times.
 408 av_assert0(in_idx < fg->nb_inputs);
2611 408 fi = &fg->inputs[in_idx];
2612
2613 408 pthread_mutex_lock(&sch->schedule_lock);
2614
2615
1/2
✓ Branch 0 taken 408 times.
✗ Branch 1 not taken.
 408 if (!fi->receive_finished) {
2616 408 fi->receive_finished = 1;
2617 408 tq_receive_finish(fg->queue, in_idx);
2618
2619 // close the control stream when all actual inputs are done
2620
2/2
✓ Branch 0 taken 315 times.
✓ Branch 1 taken 93 times.
 408 if (++fg->nb_inputs_finished_receive == fg->nb_inputs)
2621 315 tq_receive_finish(fg->queue, fg->nb_inputs);
2622
2623 408 schedule_update_locked(sch);
2624 }
2625
2626 408 pthread_mutex_unlock(&sch->schedule_lock);
2627 408 }
2628
2629 454957 int sch_filter_send(Scheduler *sch, unsigned fg_idx, unsigned out_idx, AVFrame *frame)
2630 {
2631 SchFilterGraph *fg;
2632 SchedulerNode dst;
2633 int ret;
2634
2635
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 454957 times.
 454957 av_assert0(fg_idx < sch->nb_filters);
2636 454957 fg = &sch->filters[fg_idx];
2637
2638
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 454957 times.
 454957 av_assert0(out_idx < fg->nb_outputs);
2639 454957 dst = fg->outputs[out_idx].dst;
2640
2641
2/2
✓ Branch 0 taken 454955 times.
✓ Branch 1 taken 2 times.
 454957 if (dst.type == SCH_NODE_TYPE_ENC) {
2642 454955 ret = send_to_enc(sch, &sch->enc[dst.idx], frame);
2643
2/2
✓ Branch 0 taken 3247 times.
✓ Branch 1 taken 451708 times.
 454955 if (ret == AVERROR_EOF)
2644 3247 send_to_enc(sch, &sch->enc[dst.idx], NULL);
2645 } else {
2646 2 ret = send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, frame);
2647
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 if (ret == AVERROR_EOF)
2648 ✗ send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, NULL);
2649 }
2650 454957 return ret;
2651 }
2652
2653 8215 static int filter_done(Scheduler *sch, unsigned fg_idx)
2654 {
2655 8215 SchFilterGraph *fg = &sch->filters[fg_idx];
2656 8215 int ret = 0;
2657
2658
2/2
✓ Branch 0 taken 15337 times.
✓ Branch 1 taken 8215 times.
 23552 for (unsigned i = 0; i <= fg->nb_inputs; i++)
2659 15337 tq_receive_finish(fg->queue, i);
2660
2661
2/2
✓ Branch 0 taken 8350 times.
✓ Branch 1 taken 8215 times.
 16565 for (unsigned i = 0; i < fg->nb_outputs; i++) {
2662 8350 SchedulerNode dst = fg->outputs[i].dst;
2663 16700 int err = (dst.type == SCH_NODE_TYPE_ENC) ?
2664
2/2
✓ Branch 0 taken 8349 times.
✓ Branch 1 taken 1 times.
 8350 send_to_enc (sch, &sch->enc[dst.idx], NULL) :
2665 1 send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, NULL);
2666
2667
3/4
✓ Branch 0 taken 3292 times.
✓ Branch 1 taken 5058 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 3292 times.
 8350 if (err < 0 && err != AVERROR_EOF)
2668 ✗ ret = err_merge(ret, err);
2669 }
2670
2671 8215 pthread_mutex_lock(&sch->schedule_lock);
2672
2673 8215 fg->task_exited = 1;
2674
2675 8215 schedule_update_locked(sch);
2676
2677 8215 pthread_mutex_unlock(&sch->schedule_lock);
2678
2679 8215 return ret;
2680 }
2681
2682 ✗ int sch_filter_command(Scheduler *sch, unsigned fg_idx, AVFrame *frame)
2683 {
2684 SchFilterGraph *fg;
2685
2686 ✗ av_assert0(fg_idx < sch->nb_filters);
2687 ✗ fg = &sch->filters[fg_idx];
2688
2689 ✗ return send_to_filter(sch, fg, fg->nb_inputs, frame);
2690 }
2691
2692 7013 void sch_filter_choke_inputs(Scheduler *sch, unsigned fg_idx)
2693 {
2694 SchFilterGraph *fg;
2695
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7013 times.
 7013 av_assert0(fg_idx < sch->nb_filters);
2696 7013 fg = &sch->filters[fg_idx];
2697
2698 7013 pthread_mutex_lock(&sch->schedule_lock);
2699 7013 fg->best_input = fg->nb_inputs;
2700 7013 schedule_update_locked(sch);
2701 7013 pthread_mutex_unlock(&sch->schedule_lock);
2702 7013 }
2703
2704 40044 static int task_cleanup(Scheduler *sch, SchedulerNode node)
2705 {
2706
5/6
✓ Branch 0 taken 7623 times.
✓ Branch 1 taken 8722 times.
✓ Branch 2 taken 7093 times.
✓ Branch 3 taken 8391 times.
✓ Branch 4 taken 8215 times.
✗ Branch 5 not taken.
 40044 switch (node.type) {
2707 7623 case SCH_NODE_TYPE_DEMUX: return demux_done (sch, node.idx);
2708 8722 case SCH_NODE_TYPE_MUX: return mux_done (sch, node.idx);
2709 7093 case SCH_NODE_TYPE_DEC: return dec_done (sch, node.idx);
2710 8391 case SCH_NODE_TYPE_ENC: return enc_done (sch, node.idx);
2711 8215 case SCH_NODE_TYPE_FILTER_IN: return filter_done(sch, node.idx);
2712 ✗ default: av_unreachable("Invalid node type?");
2713 }
2714 }
2715
2716 40028 static void *task_wrapper(void *arg)
2717 {
2718 40028 SchTask *task = arg;
2719 40028 Scheduler *sch = task->parent;
2720 int ret;
2721 40028 int err = 0;
2722
2723 40028 ret = task->func(task->func_arg);
2724
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 40025 times.
 40028 if (ret < 0)
2725 3 av_log(task->func_arg, AV_LOG_ERROR,
2726 3 "Task finished with error: %s\n", av_err2str(ret));
2727
2728 40028 err = task_cleanup(sch, task->node);
2729 40028 ret = err_merge(ret, err);
2730
2731 // EOF is considered normal termination
2732
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 40028 times.
 40028 if (ret == AVERROR_EOF)
2733 ✗ ret = 0;
2734
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 40025 times.
 40028 if (ret < 0) {
2735 3 pthread_mutex_lock(&sch->finish_lock);
2736 3 sch->task_failed = 1;
2737 3 pthread_cond_signal(&sch->finish_cond);
2738 3 pthread_mutex_unlock(&sch->finish_lock);
2739 }
2740
2741
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 40025 times.
 40028 if (ret < 0)
2742 3 av_log(task->func_arg, AV_LOG_ERROR,
2743 3 "Terminating thread with error: %s\n", av_err2str(ret));
2744 else
2745 40025 av_log(task->func_arg, AV_LOG_VERBOSE,
2746 "Terminating thread with success\n");
2747
2748 40028 return (void*)(intptr_t)ret;
2749 }
2750
2751 40047 static int task_stop(Scheduler *sch, SchTask *task)
2752 {
2753 int ret;
2754 void *thread_ret;
2755
2756
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 40044 times.
 40047 if (!task->parent)
2757 3 return 0;
2758
2759
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 40028 times.
 40044 if (!task->thread_running)
2760 16 return task_cleanup(sch, task->node);
2761
2762 40028 ret = pthread_join(task->thread, &thread_ret);
2763
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 40028 times.
 40028 av_assert0(ret == 0);
2764
2765 40028 task->thread_running = 0;
2766
2767 40028 return (intptr_t)thread_ret;
2768 }
2769
2770 17445 int sch_stop(Scheduler *sch, int64_t *finish_ts)
2771 {
2772 17445 int ret = 0, err;
2773
2774
2/2
✓ Branch 0 taken 8727 times.
✓ Branch 1 taken 8718 times.
 17445 if (sch->state != SCH_STATE_STARTED)
2775 8727 return 0;
2776
2777 8718 atomic_store(&sch->terminate, 1);
2778
2779 // Ensure no other thread is currently in schedule_update_locked while
2780 // we are choking all demuxers
2781 8718 pthread_mutex_lock(&sch->schedule_lock);
2782
2783
2/2
✓ Branch 0 taken 17436 times.
✓ Branch 1 taken 8718 times.
 26154 for (unsigned type = 0; type < 2; type++)
2784
4/4
✓ Branch 0 taken 16341 times.
✓ Branch 1 taken 16936 times.
✓ Branch 2 taken 15841 times.
✓ Branch 3 taken 17436 times.
 33277 for (unsigned i = 0; i < (type ? sch->nb_demux : sch->nb_filters); i++) {
2785
2/2
✓ Branch 0 taken 7623 times.
✓ Branch 1 taken 8218 times.
 15841 SchWaiter *w = type ? &sch->demux[i].waiter : &sch->filters[i].waiter;
2786 15841 waiter_set(w, 1);
2787
2/2
✓ Branch 0 taken 7623 times.
✓ Branch 1 taken 8218 times.
 15841 if (type)
2788 7623 choke_demux(sch, i, 0); // unfreeze to allow draining
2789 }
2790
2791 8718 pthread_mutex_unlock(&sch->schedule_lock);
2792
2793
2/2
✓ Branch 0 taken 7623 times.
✓ Branch 1 taken 8718 times.
 16341 for (unsigned i = 0; i < sch->nb_demux; i++) {
2794 7623 SchDemux *d = &sch->demux[i];
2795
2796 7623 err = task_stop(sch, &d->task);
2797 7623 ret = err_merge(ret, err);
2798 }
2799
2800
2/2
✓ Branch 0 taken 7093 times.
✓ Branch 1 taken 8718 times.
 15811 for (unsigned i = 0; i < sch->nb_dec; i++) {
2801 7093 SchDec *dec = &sch->dec[i];
2802
2803 7093 err = task_stop(sch, &dec->task);
2804 7093 ret = err_merge(ret, err);
2805 }
2806
2807
2/2
✓ Branch 0 taken 8218 times.
✓ Branch 1 taken 8718 times.
 16936 for (unsigned i = 0; i < sch->nb_filters; i++) {
2808 8218 SchFilterGraph *fg = &sch->filters[i];
2809
2810 8218 err = task_stop(sch, &fg->task);
2811 8218 ret = err_merge(ret, err);
2812 }
2813
2814
2/2
✓ Branch 0 taken 8391 times.
✓ Branch 1 taken 8718 times.
 17109 for (unsigned i = 0; i < sch->nb_enc; i++) {
2815 8391 SchEnc *enc = &sch->enc[i];
2816
2817 8391 err = task_stop(sch, &enc->task);
2818 8391 ret = err_merge(ret, err);
2819 }
2820
2821
2/2
✓ Branch 0 taken 8722 times.
✓ Branch 1 taken 8718 times.
 17440 for (unsigned i = 0; i < sch->nb_mux; i++) {
2822 8722 SchMux *mux = &sch->mux[i];
2823
2824 8722 err = task_stop(sch, &mux->task);
2825 8722 ret = err_merge(ret, err);
2826 }
2827
2828
1/2
✓ Branch 0 taken 8718 times.
✗ Branch 1 not taken.
 8718 if (finish_ts)
2829 8718 *finish_ts = progressing_dts(sch, 1);
2830
2831 8718 sch->state = SCH_STATE_STOPPED;
2832
2833 8718 return ret;
2834 }
2835