FFmpeg coverage


Directory: ../../../ffmpeg/
File: src/fftools/ffmpeg_sched.c
Date: 2024-02-29 09:57:37
Exec Total Coverage
Lines: 909 1029 88.3%
Functions: 58 60 96.7%
Branches: 501 690 72.6%

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 SchDec {
75 const AVClass *class;
76
77 SchedulerNode src;
78 SchedulerNode *dst;
79 uint8_t *dst_finished;
80 unsigned nb_dst;
81
82 SchTask task;
83 // Queue for receiving input packets, one stream.
84 ThreadQueue *queue;
85
86 // Queue for sending post-flush end timestamps back to the source
87 AVThreadMessageQueue *queue_end_ts;
88 int expect_end_ts;
89
90 // temporary storage used by sch_dec_send()
91 AVFrame *send_frame;
92 } SchDec;
93
94 typedef struct SchSyncQueue {
95 SyncQueue *sq;
96 AVFrame *frame;
97 pthread_mutex_t lock;
98
99 unsigned *enc_idx;
100 unsigned nb_enc_idx;
101 } SchSyncQueue;
102
103 typedef struct SchEnc {
104 const AVClass *class;
105
106 SchedulerNode src;
107 SchedulerNode dst;
108
109 // [0] - index of the sync queue in Scheduler.sq_enc,
110 // [1] - index of this encoder in the sq
111 int sq_idx[2];
112
113 /* Opening encoders is somewhat nontrivial due to their interaction with
114 * sync queues, which are (among other things) responsible for maintaining
115 * constant audio frame size, when it is required by the encoder.
116 *
117 * Opening the encoder requires stream parameters, obtained from the first
118 * frame. However, that frame cannot be properly chunked by the sync queue
119 * without knowing the required frame size, which is only available after
120 * opening the encoder.
121 *
122 * This apparent circular dependency is resolved in the following way:
123 * - the caller creating the encoder gives us a callback which opens the
124 * encoder and returns the required frame size (if any)
125 * - when the first frame is sent to the encoder, the sending thread
126 * - calls this callback, opening the encoder
127 * - passes the returned frame size to the sync queue
128 */
129 int (*open_cb)(void *opaque, const AVFrame *frame);
130 int opened;
131
132 SchTask task;
133 // Queue for receiving input frames, one stream.
134 ThreadQueue *queue;
135 // tq_send() to queue returned EOF
136 int in_finished;
137 int out_finished;
138 } SchEnc;
139
140 typedef struct SchDemuxStream {
141 SchedulerNode *dst;
142 uint8_t *dst_finished;
143 unsigned nb_dst;
144 } SchDemuxStream;
145
146 typedef struct SchDemux {
147 const AVClass *class;
148
149 SchDemuxStream *streams;
150 unsigned nb_streams;
151
152 SchTask task;
153 SchWaiter waiter;
154
155 // temporary storage used by sch_demux_send()
156 AVPacket *send_pkt;
157
158 // protected by schedule_lock
159 int task_exited;
160 } SchDemux;
161
162 typedef struct PreMuxQueue {
163 /**
164 * Queue for buffering the packets before the muxer task can be started.
165 */
166 AVFifo *fifo;
167 /**
168 * Maximum number of packets in fifo.
169 */
170 int max_packets;
171 /*
172 * The size of the AVPackets' buffers in queue.
173 * Updated when a packet is either pushed or pulled from the queue.
174 */
175 size_t data_size;
176 /* Threshold after which max_packets will be in effect */
177 size_t data_threshold;
178 } PreMuxQueue;
179
180 typedef struct SchMuxStream {
181 SchedulerNode src;
182 SchedulerNode src_sched;
183
184 unsigned *sub_heartbeat_dst;
185 unsigned nb_sub_heartbeat_dst;
186
187 PreMuxQueue pre_mux_queue;
188
189 // an EOF was generated while flushing the pre-mux queue
190 int init_eof;
191
192 ////////////////////////////////////////////////////////////
193 // The following are protected by Scheduler.schedule_lock //
194
195 /* dts+duration of the last packet sent to this stream
196 in AV_TIME_BASE_Q */
197 int64_t last_dts;
198 // this stream no longer accepts input
199 int source_finished;
200 ////////////////////////////////////////////////////////////
201 } SchMuxStream;
202
203 typedef struct SchMux {
204 const AVClass *class;
205
206 SchMuxStream *streams;
207 unsigned nb_streams;
208 unsigned nb_streams_ready;
209
210 int (*init)(void *arg);
211
212 SchTask task;
213 /**
214 * Set to 1 after starting the muxer task and flushing the
215 * pre-muxing queues.
216 * Set either before any tasks have started, or with
217 * Scheduler.mux_ready_lock held.
218 */
219 atomic_int mux_started;
220 ThreadQueue *queue;
221 unsigned queue_size;
222
223 AVPacket *sub_heartbeat_pkt;
224 } SchMux;
225
226 typedef struct SchFilterIn {
227 SchedulerNode src;
228 SchedulerNode src_sched;
229 int send_finished;
230 int receive_finished;
231 } SchFilterIn;
232
233 typedef struct SchFilterOut {
234 SchedulerNode dst;
235 } SchFilterOut;
236
237 typedef struct SchFilterGraph {
238 const AVClass *class;
239
240 SchFilterIn *inputs;
241 unsigned nb_inputs;
242 atomic_uint nb_inputs_finished_send;
243 unsigned nb_inputs_finished_receive;
244
245 SchFilterOut *outputs;
246 unsigned nb_outputs;
247
248 SchTask task;
249 // input queue, nb_inputs+1 streams
250 // last stream is control
251 ThreadQueue *queue;
252 SchWaiter waiter;
253
254 // protected by schedule_lock
255 unsigned best_input;
256 int task_exited;
257 } SchFilterGraph;
258
259 struct Scheduler {
260 const AVClass *class;
261
262 SchDemux *demux;
263 unsigned nb_demux;
264
265 SchMux *mux;
266 unsigned nb_mux;
267
268 unsigned nb_mux_ready;
269 pthread_mutex_t mux_ready_lock;
270
271 unsigned nb_mux_done;
272 pthread_mutex_t mux_done_lock;
273 pthread_cond_t mux_done_cond;
274
275
276 SchDec *dec;
277 unsigned nb_dec;
278
279 SchEnc *enc;
280 unsigned nb_enc;
281
282 SchSyncQueue *sq_enc;
283 unsigned nb_sq_enc;
284
285 SchFilterGraph *filters;
286 unsigned nb_filters;
287
288 char *sdp_filename;
289 int sdp_auto;
290
291 int transcode_started;
292 atomic_int terminate;
293 atomic_int task_failed;
294
295 pthread_mutex_t schedule_lock;
296
297 atomic_int_least64_t last_dts;
298 };
299
300 /**
301 * Wait until this task is allowed to proceed.
302 *
303 * @retval 0 the caller should proceed
304 * @retval 1 the caller should terminate
305 */
306 509117 static int waiter_wait(Scheduler *sch, SchWaiter *w)
307 {
308 int terminate;
309
310
2/2
✓ Branch 0 taken 508822 times.
✓ Branch 1 taken 295 times.
509117 if (!atomic_load(&w->choked))
311 508822 return 0;
312
313 295 pthread_mutex_lock(&w->lock);
314
315
4/4
✓ Branch 0 taken 297 times.
✓ Branch 1 taken 244 times.
✓ Branch 2 taken 246 times.
✓ Branch 3 taken 51 times.
541 while (atomic_load(&w->choked) && !atomic_load(&sch->terminate))
316 246 pthread_cond_wait(&w->cond, &w->lock);
317
318 295 terminate = atomic_load(&sch->terminate);
319
320 295 pthread_mutex_unlock(&w->lock);
321
322 295 return terminate;
323 }
324
325 43708 static void waiter_set(SchWaiter *w, int choked)
326 {
327 43708 pthread_mutex_lock(&w->lock);
328
329 43708 atomic_store(&w->choked, choked);
330 43708 pthread_cond_signal(&w->cond);
331
332 43708 pthread_mutex_unlock(&w->lock);
333 43708 }
334
335 13168 static int waiter_init(SchWaiter *w)
336 {
337 int ret;
338
339 13168 atomic_init(&w->choked, 0);
340
341 13168 ret = pthread_mutex_init(&w->lock, NULL);
342
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 13168 times.
13168 if (ret)
343 return AVERROR(ret);
344
345 13168 ret = pthread_cond_init(&w->cond, NULL);
346
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 13168 times.
13168 if (ret)
347 return AVERROR(ret);
348
349 13168 return 0;
350 }
351
352 13168 static void waiter_uninit(SchWaiter *w)
353 {
354 13168 pthread_mutex_destroy(&w->lock);
355 13168 pthread_cond_destroy(&w->cond);
356 13168 }
357
358 26005 static int queue_alloc(ThreadQueue **ptq, unsigned nb_streams, unsigned queue_size,
359 enum QueueType type)
360 {
361 ThreadQueue *tq;
362 ObjPool *op;
363
364
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26005 times.
26005 queue_size = queue_size > 0 ? queue_size : 8;
365
366
2/2
✓ Branch 0 taken 13145 times.
✓ Branch 1 taken 12860 times.
26005 op = (type == QUEUE_PACKETS) ? objpool_alloc_packets() :
367 12860 objpool_alloc_frames();
368
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26005 times.
26005 if (!op)
369 return AVERROR(ENOMEM);
370
371
2/2
✓ Branch 0 taken 13145 times.
✓ Branch 1 taken 12860 times.
26005 tq = tq_alloc(nb_streams, queue_size, op,
372 (type == QUEUE_PACKETS) ? pkt_move : frame_move);
373
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26005 times.
26005 if (!tq) {
374 objpool_free(&op);
375 return AVERROR(ENOMEM);
376 }
377
378 26005 *ptq = tq;
379 26005 return 0;
380 }
381
382 static void *task_wrapper(void *arg);
383
384 65574 static int task_stop(SchTask *task)
385 {
386 int ret;
387 void *thread_ret;
388
389
2/2
✓ Branch 0 taken 32801 times.
✓ Branch 1 taken 32773 times.
65574 if (!task->thread_running)
390 32801 return 0;
391
392 32773 ret = pthread_join(task->thread, &thread_ret);
393
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 32773 times.
32773 av_assert0(ret == 0);
394
395 32773 task->thread_running = 0;
396
397 32773 return (intptr_t)thread_ret;
398 }
399
400 32773 static int task_start(SchTask *task)
401 {
402 int ret;
403
404 32773 av_log(task->func_arg, AV_LOG_VERBOSE, "Starting thread...\n");
405
406
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 32773 times.
32773 av_assert0(!task->thread_running);
407
408 32773 ret = pthread_create(&task->thread, NULL, task_wrapper, task);
409
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 32773 times.
32773 if (ret) {
410 av_log(task->func_arg, AV_LOG_ERROR, "pthread_create() failed: %s\n",
411 strerror(ret));
412 return AVERROR(ret);
413 }
414
415 32773 task->thread_running = 1;
416 32773 return 0;
417 }
418
419 32787 static void task_init(Scheduler *sch, SchTask *task, enum SchedulerNodeType type, unsigned idx,
420 SchThreadFunc func, void *func_arg)
421 {
422 32787 task->parent = sch;
423
424 32787 task->node.type = type;
425 32787 task->node.idx = idx;
426
427 32787 task->func = func;
428 32787 task->func_arg = func_arg;
429 32787 }
430
431 532723 static int64_t trailing_dts(const Scheduler *sch, int count_finished)
432 {
433 532723 int64_t min_dts = INT64_MAX;
434
435
2/2
✓ Branch 0 taken 533647 times.
✓ Branch 1 taken 518532 times.
1052179 for (unsigned i = 0; i < sch->nb_mux; i++) {
436 533647 const SchMux *mux = &sch->mux[i];
437
438
2/2
✓ Branch 0 taken 600569 times.
✓ Branch 1 taken 519456 times.
1120025 for (unsigned j = 0; j < mux->nb_streams; j++) {
439 600569 const SchMuxStream *ms = &mux->streams[j];
440
441
4/4
✓ Branch 0 taken 37148 times.
✓ Branch 1 taken 563421 times.
✓ Branch 2 taken 30090 times.
✓ Branch 3 taken 7058 times.
600569 if (ms->source_finished && !count_finished)
442 30090 continue;
443
2/2
✓ Branch 0 taken 14191 times.
✓ Branch 1 taken 556288 times.
570479 if (ms->last_dts == AV_NOPTS_VALUE)
444 14191 return AV_NOPTS_VALUE;
445
446 556288 min_dts = FFMIN(min_dts, ms->last_dts);
447 }
448 }
449
450
2/2
✓ Branch 0 taken 494053 times.
✓ Branch 1 taken 24479 times.
518532 return min_dts == INT64_MAX ? AV_NOPTS_VALUE : min_dts;
451 }
452
453 13503 int sch_stop(Scheduler *sch, int64_t *finish_ts)
454 {
455 13503 int ret = 0, err;
456
457 13503 atomic_store(&sch->terminate, 1);
458
459
2/2
✓ Branch 0 taken 27006 times.
✓ Branch 1 taken 13503 times.
40509 for (unsigned type = 0; type < 2; type++)
460
4/4
✓ Branch 0 taken 27067 times.
✓ Branch 1 taken 26275 times.
✓ Branch 2 taken 26336 times.
✓ Branch 3 taken 27006 times.
53342 for (unsigned i = 0; i < (type ? sch->nb_demux : sch->nb_filters); i++) {
461
2/2
✓ Branch 0 taken 13564 times.
✓ Branch 1 taken 12772 times.
26336 SchWaiter *w = type ? &sch->demux[i].waiter : &sch->filters[i].waiter;
462 26336 waiter_set(w, 1);
463 }
464
465
2/2
✓ Branch 0 taken 13564 times.
✓ Branch 1 taken 13503 times.
27067 for (unsigned i = 0; i < sch->nb_demux; i++) {
466 13564 SchDemux *d = &sch->demux[i];
467
468 13564 err = task_stop(&d->task);
469 13564 ret = err_merge(ret, err);
470 }
471
472
2/2
✓ Branch 0 taken 12784 times.
✓ Branch 1 taken 13503 times.
26287 for (unsigned i = 0; i < sch->nb_dec; i++) {
473 12784 SchDec *dec = &sch->dec[i];
474
475 12784 err = task_stop(&dec->task);
476 12784 ret = err_merge(ret, err);
477 }
478
479
2/2
✓ Branch 0 taken 12772 times.
✓ Branch 1 taken 13503 times.
26275 for (unsigned i = 0; i < sch->nb_filters; i++) {
480 12772 SchFilterGraph *fg = &sch->filters[i];
481
482 12772 err = task_stop(&fg->task);
483 12772 ret = err_merge(ret, err);
484 }
485
486
2/2
✓ Branch 0 taken 12948 times.
✓ Branch 1 taken 13503 times.
26451 for (unsigned i = 0; i < sch->nb_enc; i++) {
487 12948 SchEnc *enc = &sch->enc[i];
488
489 12948 err = task_stop(&enc->task);
490 12948 ret = err_merge(ret, err);
491 }
492
493
2/2
✓ Branch 0 taken 13506 times.
✓ Branch 1 taken 13503 times.
27009 for (unsigned i = 0; i < sch->nb_mux; i++) {
494 13506 SchMux *mux = &sch->mux[i];
495
496 13506 err = task_stop(&mux->task);
497 13506 ret = err_merge(ret, err);
498 }
499
500
2/2
✓ Branch 0 taken 6751 times.
✓ Branch 1 taken 6752 times.
13503 if (finish_ts)
501 6751 *finish_ts = trailing_dts(sch, 1);
502
503 13503 return ret;
504 }
505
506 6752 void sch_free(Scheduler **psch)
507 {
508 6752 Scheduler *sch = *psch;
509
510
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6752 times.
6752 if (!sch)
511 return;
512
513 6752 sch_stop(sch, NULL);
514
515
2/2
✓ Branch 0 taken 6782 times.
✓ Branch 1 taken 6752 times.
13534 for (unsigned i = 0; i < sch->nb_demux; i++) {
516 6782 SchDemux *d = &sch->demux[i];
517
518
2/2
✓ Branch 0 taken 6989 times.
✓ Branch 1 taken 6782 times.
13771 for (unsigned j = 0; j < d->nb_streams; j++) {
519 6989 SchDemuxStream *ds = &d->streams[j];
520 6989 av_freep(&ds->dst);
521 6989 av_freep(&ds->dst_finished);
522 }
523 6782 av_freep(&d->streams);
524
525 6782 av_packet_free(&d->send_pkt);
526
527 6782 waiter_uninit(&d->waiter);
528 }
529 6752 av_freep(&sch->demux);
530
531
2/2
✓ Branch 0 taken 6753 times.
✓ Branch 1 taken 6752 times.
13505 for (unsigned i = 0; i < sch->nb_mux; i++) {
532 6753 SchMux *mux = &sch->mux[i];
533
534
2/2
✓ Branch 0 taken 7094 times.
✓ Branch 1 taken 6753 times.
13847 for (unsigned j = 0; j < mux->nb_streams; j++) {
535 7094 SchMuxStream *ms = &mux->streams[j];
536
537
1/2
✓ Branch 0 taken 7094 times.
✗ Branch 1 not taken.
7094 if (ms->pre_mux_queue.fifo) {
538 AVPacket *pkt;
539
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 7094 times.
7094 while (av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1) >= 0)
540 av_packet_free(&pkt);
541 7094 av_fifo_freep2(&ms->pre_mux_queue.fifo);
542 }
543
544 7094 av_freep(&ms->sub_heartbeat_dst);
545 }
546 6753 av_freep(&mux->streams);
547
548 6753 av_packet_free(&mux->sub_heartbeat_pkt);
549
550 6753 tq_free(&mux->queue);
551 }
552 6752 av_freep(&sch->mux);
553
554
2/2
✓ Branch 0 taken 6392 times.
✓ Branch 1 taken 6752 times.
13144 for (unsigned i = 0; i < sch->nb_dec; i++) {
555 6392 SchDec *dec = &sch->dec[i];
556
557 6392 tq_free(&dec->queue);
558
559 6392 av_thread_message_queue_free(&dec->queue_end_ts);
560
561 6392 av_freep(&dec->dst);
562 6392 av_freep(&dec->dst_finished);
563
564 6392 av_frame_free(&dec->send_frame);
565 }
566 6752 av_freep(&sch->dec);
567
568
2/2
✓ Branch 0 taken 6474 times.
✓ Branch 1 taken 6752 times.
13226 for (unsigned i = 0; i < sch->nb_enc; i++) {
569 6474 SchEnc *enc = &sch->enc[i];
570
571 6474 tq_free(&enc->queue);
572 }
573 6752 av_freep(&sch->enc);
574
575
2/2
✓ Branch 0 taken 2777 times.
✓ Branch 1 taken 6752 times.
9529 for (unsigned i = 0; i < sch->nb_sq_enc; i++) {
576 2777 SchSyncQueue *sq = &sch->sq_enc[i];
577 2777 sq_free(&sq->sq);
578 2777 av_frame_free(&sq->frame);
579 2777 pthread_mutex_destroy(&sq->lock);
580 2777 av_freep(&sq->enc_idx);
581 }
582 6752 av_freep(&sch->sq_enc);
583
584
2/2
✓ Branch 0 taken 6386 times.
✓ Branch 1 taken 6752 times.
13138 for (unsigned i = 0; i < sch->nb_filters; i++) {
585 6386 SchFilterGraph *fg = &sch->filters[i];
586
587 6386 tq_free(&fg->queue);
588
589 6386 av_freep(&fg->inputs);
590 6386 av_freep(&fg->outputs);
591
592 6386 waiter_uninit(&fg->waiter);
593 }
594 6752 av_freep(&sch->filters);
595
596 6752 av_freep(&sch->sdp_filename);
597
598 6752 pthread_mutex_destroy(&sch->schedule_lock);
599
600 6752 pthread_mutex_destroy(&sch->mux_ready_lock);
601
602 6752 pthread_mutex_destroy(&sch->mux_done_lock);
603 6752 pthread_cond_destroy(&sch->mux_done_cond);
604
605 6752 av_freep(psch);
606 }
607
608 static const AVClass scheduler_class = {
609 .class_name = "Scheduler",
610 .version = LIBAVUTIL_VERSION_INT,
611 };
612
613 6752 Scheduler *sch_alloc(void)
614 {
615 Scheduler *sch;
616 int ret;
617
618 6752 sch = av_mallocz(sizeof(*sch));
619
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6752 times.
6752 if (!sch)
620 return NULL;
621
622 6752 sch->class = &scheduler_class;
623 6752 sch->sdp_auto = 1;
624
625 6752 ret = pthread_mutex_init(&sch->schedule_lock, NULL);
626
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6752 times.
6752 if (ret)
627 goto fail;
628
629 6752 ret = pthread_mutex_init(&sch->mux_ready_lock, NULL);
630
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6752 times.
6752 if (ret)
631 goto fail;
632
633 6752 ret = pthread_mutex_init(&sch->mux_done_lock, NULL);
634
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6752 times.
6752 if (ret)
635 goto fail;
636
637 6752 ret = pthread_cond_init(&sch->mux_done_cond, NULL);
638
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6752 times.
6752 if (ret)
639 goto fail;
640
641 6752 return sch;
642 fail:
643 sch_free(&sch);
644 return NULL;
645 }
646
647 int sch_sdp_filename(Scheduler *sch, const char *sdp_filename)
648 {
649 av_freep(&sch->sdp_filename);
650 sch->sdp_filename = av_strdup(sdp_filename);
651 return sch->sdp_filename ? 0 : AVERROR(ENOMEM);
652 }
653
654 static const AVClass sch_mux_class = {
655 .class_name = "SchMux",
656 .version = LIBAVUTIL_VERSION_INT,
657 .parent_log_context_offset = offsetof(SchMux, task.func_arg),
658 };
659
660 6753 int sch_add_mux(Scheduler *sch, SchThreadFunc func, int (*init)(void *),
661 void *arg, int sdp_auto, unsigned thread_queue_size)
662 {
663 6753 const unsigned idx = sch->nb_mux;
664
665 SchMux *mux;
666 int ret;
667
668 6753 ret = GROW_ARRAY(sch->mux, sch->nb_mux);
669
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6753 times.
6753 if (ret < 0)
670 return ret;
671
672 6753 mux = &sch->mux[idx];
673 6753 mux->class = &sch_mux_class;
674 6753 mux->init = init;
675 6753 mux->queue_size = thread_queue_size;
676
677 6753 task_init(sch, &mux->task, SCH_NODE_TYPE_MUX, idx, func, arg);
678
679 6753 sch->sdp_auto &= sdp_auto;
680
681 6753 return idx;
682 }
683
684 7094 int sch_add_mux_stream(Scheduler *sch, unsigned mux_idx)
685 {
686 SchMux *mux;
687 SchMuxStream *ms;
688 unsigned stream_idx;
689 int ret;
690
691
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 av_assert0(mux_idx < sch->nb_mux);
692 7094 mux = &sch->mux[mux_idx];
693
694 7094 ret = GROW_ARRAY(mux->streams, mux->nb_streams);
695
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 if (ret < 0)
696 return ret;
697 7094 stream_idx = mux->nb_streams - 1;
698
699 7094 ms = &mux->streams[stream_idx];
700
701 7094 ms->pre_mux_queue.fifo = av_fifo_alloc2(8, sizeof(AVPacket*), 0);
702
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 if (!ms->pre_mux_queue.fifo)
703 return AVERROR(ENOMEM);
704
705 7094 ms->last_dts = AV_NOPTS_VALUE;
706
707 7094 return stream_idx;
708 }
709
710 static const AVClass sch_demux_class = {
711 .class_name = "SchDemux",
712 .version = LIBAVUTIL_VERSION_INT,
713 .parent_log_context_offset = offsetof(SchDemux, task.func_arg),
714 };
715
716 6782 int sch_add_demux(Scheduler *sch, SchThreadFunc func, void *ctx)
717 {
718 6782 const unsigned idx = sch->nb_demux;
719
720 SchDemux *d;
721 int ret;
722
723 6782 ret = GROW_ARRAY(sch->demux, sch->nb_demux);
724
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6782 times.
6782 if (ret < 0)
725 return ret;
726
727 6782 d = &sch->demux[idx];
728
729 6782 task_init(sch, &d->task, SCH_NODE_TYPE_DEMUX, idx, func, ctx);
730
731 6782 d->class = &sch_demux_class;
732 6782 d->send_pkt = av_packet_alloc();
733
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6782 times.
6782 if (!d->send_pkt)
734 return AVERROR(ENOMEM);
735
736 6782 ret = waiter_init(&d->waiter);
737
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6782 times.
6782 if (ret < 0)
738 return ret;
739
740 6782 return idx;
741 }
742
743 6989 int sch_add_demux_stream(Scheduler *sch, unsigned demux_idx)
744 {
745 SchDemux *d;
746 int ret;
747
748
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6989 times.
6989 av_assert0(demux_idx < sch->nb_demux);
749 6989 d = &sch->demux[demux_idx];
750
751 6989 ret = GROW_ARRAY(d->streams, d->nb_streams);
752
1/2
✓ Branch 0 taken 6989 times.
✗ Branch 1 not taken.
6989 return ret < 0 ? ret : d->nb_streams - 1;
753 }
754
755 static const AVClass sch_dec_class = {
756 .class_name = "SchDec",
757 .version = LIBAVUTIL_VERSION_INT,
758 .parent_log_context_offset = offsetof(SchDec, task.func_arg),
759 };
760
761 6392 int sch_add_dec(Scheduler *sch, SchThreadFunc func, void *ctx,
762 int send_end_ts)
763 {
764 6392 const unsigned idx = sch->nb_dec;
765
766 SchDec *dec;
767 int ret;
768
769 6392 ret = GROW_ARRAY(sch->dec, sch->nb_dec);
770
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 if (ret < 0)
771 return ret;
772
773 6392 dec = &sch->dec[idx];
774
775 6392 task_init(sch, &dec->task, SCH_NODE_TYPE_DEC, idx, func, ctx);
776
777 6392 dec->class = &sch_dec_class;
778 6392 dec->send_frame = av_frame_alloc();
779
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 if (!dec->send_frame)
780 return AVERROR(ENOMEM);
781
782 6392 ret = queue_alloc(&dec->queue, 1, 0, QUEUE_PACKETS);
783
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 if (ret < 0)
784 return ret;
785
786
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 6391 times.
6392 if (send_end_ts) {
787 1 ret = av_thread_message_queue_alloc(&dec->queue_end_ts, 1, sizeof(Timestamp));
788
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 if (ret < 0)
789 return ret;
790 }
791
792 6392 return idx;
793 }
794
795 static const AVClass sch_enc_class = {
796 .class_name = "SchEnc",
797 .version = LIBAVUTIL_VERSION_INT,
798 .parent_log_context_offset = offsetof(SchEnc, task.func_arg),
799 };
800
801 6474 int sch_add_enc(Scheduler *sch, SchThreadFunc func, void *ctx,
802 int (*open_cb)(void *opaque, const AVFrame *frame))
803 {
804 6474 const unsigned idx = sch->nb_enc;
805
806 SchEnc *enc;
807 int ret;
808
809 6474 ret = GROW_ARRAY(sch->enc, sch->nb_enc);
810
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6474 times.
6474 if (ret < 0)
811 return ret;
812
813 6474 enc = &sch->enc[idx];
814
815 6474 enc->class = &sch_enc_class;
816 6474 enc->open_cb = open_cb;
817 6474 enc->sq_idx[0] = -1;
818 6474 enc->sq_idx[1] = -1;
819
820 6474 task_init(sch, &enc->task, SCH_NODE_TYPE_ENC, idx, func, ctx);
821
822 6474 ret = queue_alloc(&enc->queue, 1, 0, QUEUE_FRAMES);
823
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6474 times.
6474 if (ret < 0)
824 return ret;
825
826 6474 return idx;
827 }
828
829 static const AVClass sch_fg_class = {
830 .class_name = "SchFilterGraph",
831 .version = LIBAVUTIL_VERSION_INT,
832 .parent_log_context_offset = offsetof(SchFilterGraph, task.func_arg),
833 };
834
835 6386 int sch_add_filtergraph(Scheduler *sch, unsigned nb_inputs, unsigned nb_outputs,
836 SchThreadFunc func, void *ctx)
837 {
838 6386 const unsigned idx = sch->nb_filters;
839
840 SchFilterGraph *fg;
841 int ret;
842
843 6386 ret = GROW_ARRAY(sch->filters, sch->nb_filters);
844
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6386 times.
6386 if (ret < 0)
845 return ret;
846 6386 fg = &sch->filters[idx];
847
848 6386 fg->class = &sch_fg_class;
849
850 6386 task_init(sch, &fg->task, SCH_NODE_TYPE_FILTER_IN, idx, func, ctx);
851
852
2/2
✓ Branch 0 taken 6331 times.
✓ Branch 1 taken 55 times.
6386 if (nb_inputs) {
853 6331 fg->inputs = av_calloc(nb_inputs, sizeof(*fg->inputs));
854
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6331 times.
6331 if (!fg->inputs)
855 return AVERROR(ENOMEM);
856 6331 fg->nb_inputs = nb_inputs;
857 }
858
859
1/2
✓ Branch 0 taken 6386 times.
✗ Branch 1 not taken.
6386 if (nb_outputs) {
860 6386 fg->outputs = av_calloc(nb_outputs, sizeof(*fg->outputs));
861
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6386 times.
6386 if (!fg->outputs)
862 return AVERROR(ENOMEM);
863 6386 fg->nb_outputs = nb_outputs;
864 }
865
866 6386 ret = waiter_init(&fg->waiter);
867
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6386 times.
6386 if (ret < 0)
868 return ret;
869
870 6386 ret = queue_alloc(&fg->queue, fg->nb_inputs + 1, 0, QUEUE_FRAMES);
871
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6386 times.
6386 if (ret < 0)
872 return ret;
873
874 6386 return idx;
875 }
876
877 2777 int sch_add_sq_enc(Scheduler *sch, uint64_t buf_size_us, void *logctx)
878 {
879 SchSyncQueue *sq;
880 int ret;
881
882 2777 ret = GROW_ARRAY(sch->sq_enc, sch->nb_sq_enc);
883
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2777 times.
2777 if (ret < 0)
884 return ret;
885 2777 sq = &sch->sq_enc[sch->nb_sq_enc - 1];
886
887 2777 sq->sq = sq_alloc(SYNC_QUEUE_FRAMES, buf_size_us, logctx);
888
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2777 times.
2777 if (!sq->sq)
889 return AVERROR(ENOMEM);
890
891 2777 sq->frame = av_frame_alloc();
892
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2777 times.
2777 if (!sq->frame)
893 return AVERROR(ENOMEM);
894
895 2777 ret = pthread_mutex_init(&sq->lock, NULL);
896
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2777 times.
2777 if (ret)
897 return AVERROR(ret);
898
899 2777 return sq - sch->sq_enc;
900 }
901
902 2822 int sch_sq_add_enc(Scheduler *sch, unsigned sq_idx, unsigned enc_idx,
903 int limiting, uint64_t max_frames)
904 {
905 SchSyncQueue *sq;
906 SchEnc *enc;
907 int ret;
908
909
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2822 times.
2822 av_assert0(sq_idx < sch->nb_sq_enc);
910 2822 sq = &sch->sq_enc[sq_idx];
911
912
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2822 times.
2822 av_assert0(enc_idx < sch->nb_enc);
913 2822 enc = &sch->enc[enc_idx];
914
915 2822 ret = GROW_ARRAY(sq->enc_idx, sq->nb_enc_idx);
916
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2822 times.
2822 if (ret < 0)
917 return ret;
918 2822 sq->enc_idx[sq->nb_enc_idx - 1] = enc_idx;
919
920 2822 ret = sq_add_stream(sq->sq, limiting);
921
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2822 times.
2822 if (ret < 0)
922 return ret;
923
924 2822 enc->sq_idx[0] = sq_idx;
925 2822 enc->sq_idx[1] = ret;
926
927
2/2
✓ Branch 0 taken 2652 times.
✓ Branch 1 taken 170 times.
2822 if (max_frames != INT64_MAX)
928 2652 sq_limit_frames(sq->sq, enc->sq_idx[1], max_frames);
929
930 2822 return 0;
931 }
932
933 26360 int sch_connect(Scheduler *sch, SchedulerNode src, SchedulerNode dst)
934 {
935 int ret;
936
937
4/5
✓ Branch 0 taken 7012 times.
✓ Branch 1 taken 6438 times.
✓ Branch 2 taken 6436 times.
✓ Branch 3 taken 6474 times.
✗ Branch 4 not taken.
26360 switch (src.type) {
938 7012 case SCH_NODE_TYPE_DEMUX: {
939 SchDemuxStream *ds;
940
941
2/4
✓ Branch 0 taken 7012 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 7012 times.
7012 av_assert0(src.idx < sch->nb_demux &&
942 src.idx_stream < sch->demux[src.idx].nb_streams);
943 7012 ds = &sch->demux[src.idx].streams[src.idx_stream];
944
945 7012 ret = GROW_ARRAY(ds->dst, ds->nb_dst);
946
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7012 times.
7012 if (ret < 0)
947 return ret;
948
949 7012 ds->dst[ds->nb_dst - 1] = dst;
950
951 // demuxed packets go to decoding or streamcopy
952
2/3
✓ Branch 0 taken 6392 times.
✓ Branch 1 taken 620 times.
✗ Branch 2 not taken.
7012 switch (dst.type) {
953 6392 case SCH_NODE_TYPE_DEC: {
954 SchDec *dec;
955
956
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 av_assert0(dst.idx < sch->nb_dec);
957 6392 dec = &sch->dec[dst.idx];
958
959
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 av_assert0(!dec->src.type);
960 6392 dec->src = src;
961 6392 break;
962 }
963 620 case SCH_NODE_TYPE_MUX: {
964 SchMuxStream *ms;
965
966
2/4
✓ Branch 0 taken 620 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 620 times.
620 av_assert0(dst.idx < sch->nb_mux &&
967 dst.idx_stream < sch->mux[dst.idx].nb_streams);
968 620 ms = &sch->mux[dst.idx].streams[dst.idx_stream];
969
970
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 620 times.
620 av_assert0(!ms->src.type);
971 620 ms->src = src;
972
973 620 break;
974 }
975 default: av_assert0(0);
976 }
977
978 7012 break;
979 }
980 6438 case SCH_NODE_TYPE_DEC: {
981 SchDec *dec;
982
983
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6438 times.
6438 av_assert0(src.idx < sch->nb_dec);
984 6438 dec = &sch->dec[src.idx];
985
986 6438 ret = GROW_ARRAY(dec->dst, dec->nb_dst);
987
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6438 times.
6438 if (ret < 0)
988 return ret;
989
990 6438 dec->dst[dec->nb_dst - 1] = dst;
991
992 // decoded frames go to filters or encoding
993
2/3
✓ Branch 0 taken 6400 times.
✓ Branch 1 taken 38 times.
✗ Branch 2 not taken.
6438 switch (dst.type) {
994 6400 case SCH_NODE_TYPE_FILTER_IN: {
995 SchFilterIn *fi;
996
997
2/4
✓ Branch 0 taken 6400 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6400 times.
6400 av_assert0(dst.idx < sch->nb_filters &&
998 dst.idx_stream < sch->filters[dst.idx].nb_inputs);
999 6400 fi = &sch->filters[dst.idx].inputs[dst.idx_stream];
1000
1001
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6400 times.
6400 av_assert0(!fi->src.type);
1002 6400 fi->src = src;
1003 6400 break;
1004 }
1005 38 case SCH_NODE_TYPE_ENC: {
1006 SchEnc *enc;
1007
1008
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 38 times.
38 av_assert0(dst.idx < sch->nb_enc);
1009 38 enc = &sch->enc[dst.idx];
1010
1011
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 38 times.
38 av_assert0(!enc->src.type);
1012 38 enc->src = src;
1013 38 break;
1014 }
1015 default: av_assert0(0);
1016 }
1017
1018 6438 break;
1019 }
1020 6436 case SCH_NODE_TYPE_FILTER_OUT: {
1021 SchFilterOut *fo;
1022 SchEnc *enc;
1023
1024
2/4
✓ Branch 0 taken 6436 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6436 times.
6436 av_assert0(src.idx < sch->nb_filters &&
1025 src.idx_stream < sch->filters[src.idx].nb_outputs);
1026 // filtered frames go to encoding
1027
2/4
✓ Branch 0 taken 6436 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6436 times.
6436 av_assert0(dst.type == SCH_NODE_TYPE_ENC &&
1028 dst.idx < sch->nb_enc);
1029
1030 6436 fo = &sch->filters[src.idx].outputs[src.idx_stream];
1031 6436 enc = &sch->enc[dst.idx];
1032
1033
2/4
✓ Branch 0 taken 6436 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6436 times.
6436 av_assert0(!fo->dst.type && !enc->src.type);
1034 6436 fo->dst = dst;
1035 6436 enc->src = src;
1036
1037 6436 break;
1038 }
1039 6474 case SCH_NODE_TYPE_ENC: {
1040 SchEnc *enc;
1041 SchMuxStream *ms;
1042
1043
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6474 times.
6474 av_assert0(src.idx < sch->nb_enc);
1044 // encoding packets go to muxing
1045
3/6
✓ Branch 0 taken 6474 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 6474 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 6474 times.
6474 av_assert0(dst.type == SCH_NODE_TYPE_MUX &&
1046 dst.idx < sch->nb_mux &&
1047 dst.idx_stream < sch->mux[dst.idx].nb_streams);
1048 6474 enc = &sch->enc[src.idx];
1049 6474 ms = &sch->mux[dst.idx].streams[dst.idx_stream];
1050
1051
2/4
✓ Branch 0 taken 6474 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6474 times.
6474 av_assert0(!enc->dst.type && !ms->src.type);
1052 6474 enc->dst = dst;
1053 6474 ms->src = src;
1054
1055 6474 break;
1056 }
1057 default: av_assert0(0);
1058 }
1059
1060 26360 return 0;
1061 }
1062
1063 6753 static int mux_task_start(SchMux *mux)
1064 {
1065 6753 int ret = 0;
1066
1067 6753 ret = task_start(&mux->task);
1068
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6753 times.
6753 if (ret < 0)
1069 return ret;
1070
1071 /* flush the pre-muxing queues */
1072
2/2
✓ Branch 0 taken 7094 times.
✓ Branch 1 taken 6753 times.
13847 for (unsigned i = 0; i < mux->nb_streams; i++) {
1073 7094 SchMuxStream *ms = &mux->streams[i];
1074 AVPacket *pkt;
1075
1076
2/2
✓ Branch 1 taken 3193 times.
✓ Branch 2 taken 7094 times.
10287 while (av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1) >= 0) {
1077
2/2
✓ Branch 0 taken 3151 times.
✓ Branch 1 taken 42 times.
3193 if (pkt) {
1078
2/2
✓ Branch 0 taken 3142 times.
✓ Branch 1 taken 9 times.
3151 if (!ms->init_eof)
1079 3142 ret = tq_send(mux->queue, i, pkt);
1080 3151 av_packet_free(&pkt);
1081
2/2
✓ Branch 0 taken 10 times.
✓ Branch 1 taken 3141 times.
3151 if (ret == AVERROR_EOF)
1082 10 ms->init_eof = 1;
1083
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3141 times.
3141 else if (ret < 0)
1084 return ret;
1085 } else
1086 42 tq_send_finish(mux->queue, i);
1087 }
1088 }
1089
1090 6753 atomic_store(&mux->mux_started, 1);
1091
1092 6753 return 0;
1093 }
1094
1095 int print_sdp(const char *filename);
1096
1097 6753 static int mux_init(Scheduler *sch, SchMux *mux)
1098 {
1099 int ret;
1100
1101 6753 ret = mux->init(mux->task.func_arg);
1102
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6753 times.
6753 if (ret < 0)
1103 return ret;
1104
1105 6753 sch->nb_mux_ready++;
1106
1107
2/4
✓ Branch 0 taken 6753 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6753 times.
6753 if (sch->sdp_filename || sch->sdp_auto) {
1108 if (sch->nb_mux_ready < sch->nb_mux)
1109 return 0;
1110
1111 ret = print_sdp(sch->sdp_filename);
1112 if (ret < 0) {
1113 av_log(sch, AV_LOG_ERROR, "Error writing the SDP.\n");
1114 return ret;
1115 }
1116
1117 /* SDP is written only after all the muxers are ready, so now we
1118 * start ALL the threads */
1119 for (unsigned i = 0; i < sch->nb_mux; i++) {
1120 ret = mux_task_start(&sch->mux[i]);
1121 if (ret < 0)
1122 return ret;
1123 }
1124 } else {
1125 6753 ret = mux_task_start(mux);
1126
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6753 times.
6753 if (ret < 0)
1127 return ret;
1128 }
1129
1130 6753 return 0;
1131 }
1132
1133 7094 void sch_mux_stream_buffering(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
1134 size_t data_threshold, int max_packets)
1135 {
1136 SchMux *mux;
1137 SchMuxStream *ms;
1138
1139
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 av_assert0(mux_idx < sch->nb_mux);
1140 7094 mux = &sch->mux[mux_idx];
1141
1142
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 av_assert0(stream_idx < mux->nb_streams);
1143 7094 ms = &mux->streams[stream_idx];
1144
1145 7094 ms->pre_mux_queue.max_packets = max_packets;
1146 7094 ms->pre_mux_queue.data_threshold = data_threshold;
1147 7094 }
1148
1149 7094 int sch_mux_stream_ready(Scheduler *sch, unsigned mux_idx, unsigned stream_idx)
1150 {
1151 SchMux *mux;
1152 7094 int ret = 0;
1153
1154
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 av_assert0(mux_idx < sch->nb_mux);
1155 7094 mux = &sch->mux[mux_idx];
1156
1157
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 av_assert0(stream_idx < mux->nb_streams);
1158
1159 7094 pthread_mutex_lock(&sch->mux_ready_lock);
1160
1161
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7094 times.
7094 av_assert0(mux->nb_streams_ready < mux->nb_streams);
1162
1163 // this may be called during initialization - do not start
1164 // threads before sch_start() is called
1165
4/4
✓ Branch 0 taken 6753 times.
✓ Branch 1 taken 341 times.
✓ Branch 2 taken 6321 times.
✓ Branch 3 taken 432 times.
7094 if (++mux->nb_streams_ready == mux->nb_streams && sch->transcode_started)
1166 6321 ret = mux_init(sch, mux);
1167
1168 7094 pthread_mutex_unlock(&sch->mux_ready_lock);
1169
1170 7094 return ret;
1171 }
1172
1173 1 int sch_mux_sub_heartbeat_add(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
1174 unsigned dec_idx)
1175 {
1176 SchMux *mux;
1177 SchMuxStream *ms;
1178 1 int ret = 0;
1179
1180
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 av_assert0(mux_idx < sch->nb_mux);
1181 1 mux = &sch->mux[mux_idx];
1182
1183
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 av_assert0(stream_idx < mux->nb_streams);
1184 1 ms = &mux->streams[stream_idx];
1185
1186 1 ret = GROW_ARRAY(ms->sub_heartbeat_dst, ms->nb_sub_heartbeat_dst);
1187
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 if (ret < 0)
1188 return ret;
1189
1190
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 av_assert0(dec_idx < sch->nb_dec);
1191 1 ms->sub_heartbeat_dst[ms->nb_sub_heartbeat_dst - 1] = dec_idx;
1192
1193
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
1 if (!mux->sub_heartbeat_pkt) {
1194 1 mux->sub_heartbeat_pkt = av_packet_alloc();
1195
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 if (!mux->sub_heartbeat_pkt)
1196 return AVERROR(ENOMEM);
1197 }
1198
1199 1 return 0;
1200 }
1201
1202 528930 static void schedule_update_locked(Scheduler *sch)
1203 {
1204 int64_t dts;
1205 528930 int have_unchoked = 0;
1206
1207 // on termination request all waiters are choked,
1208 // we are not to unchoke them
1209
2/2
✓ Branch 0 taken 2958 times.
✓ Branch 1 taken 525972 times.
528930 if (atomic_load(&sch->terminate))
1210 2958 return;
1211
1212 525972 dts = trailing_dts(sch, 0);
1213
1214 525972 atomic_store(&sch->last_dts, dts);
1215
1216 // initialize our internal state
1217
2/2
✓ Branch 0 taken 1051944 times.
✓ Branch 1 taken 525972 times.
1577916 for (unsigned type = 0; type < 2; type++)
1218
4/4
✓ Branch 0 taken 1000898 times.
✓ Branch 1 taken 1056756 times.
✓ Branch 2 taken 1005710 times.
✓ Branch 3 taken 1051944 times.
2057654 for (unsigned i = 0; i < (type ? sch->nb_filters : sch->nb_demux); i++) {
1219
2/2
✓ Branch 0 taken 474926 times.
✓ Branch 1 taken 530784 times.
1005710 SchWaiter *w = type ? &sch->filters[i].waiter : &sch->demux[i].waiter;
1220 1005710 w->choked_prev = atomic_load(&w->choked);
1221 1005710 w->choked_next = 1;
1222 }
1223
1224 // figure out the sources that are allowed to proceed
1225
2/2
✓ Branch 0 taken 526912 times.
✓ Branch 1 taken 525972 times.
1052884 for (unsigned i = 0; i < sch->nb_mux; i++) {
1226 526912 SchMux *mux = &sch->mux[i];
1227
1228
2/2
✓ Branch 0 taken 600861 times.
✓ Branch 1 taken 526912 times.
1127773 for (unsigned j = 0; j < mux->nb_streams; j++) {
1229 600861 SchMuxStream *ms = &mux->streams[j];
1230 SchDemux *d;
1231
1232 // unblock sources for output streams that are not finished
1233 // and not too far ahead of the trailing stream
1234
2/2
✓ Branch 0 taken 30194 times.
✓ Branch 1 taken 570667 times.
600861 if (ms->source_finished)
1235 30194 continue;
1236
4/4
✓ Branch 0 taken 24362 times.
✓ Branch 1 taken 546305 times.
✓ Branch 2 taken 5717 times.
✓ Branch 3 taken 18645 times.
570667 if (dts == AV_NOPTS_VALUE && ms->last_dts != AV_NOPTS_VALUE)
1237 5717 continue;
1238
4/4
✓ Branch 0 taken 546305 times.
✓ Branch 1 taken 18645 times.
✓ Branch 2 taken 19243 times.
✓ Branch 3 taken 527062 times.
564950 if (dts != AV_NOPTS_VALUE && ms->last_dts - dts >= SCHEDULE_TOLERANCE)
1239 19243 continue;
1240
1241 // for outputs fed from filtergraphs, consider that filtergraph's
1242 // best_input information, in other cases there is a well-defined
1243 // source demuxer
1244
2/2
✓ Branch 0 taken 472946 times.
✓ Branch 1 taken 72761 times.
545707 if (ms->src_sched.type == SCH_NODE_TYPE_FILTER_OUT) {
1245 472946 SchFilterGraph *fg = &sch->filters[ms->src_sched.idx];
1246 SchFilterIn *fi;
1247
1248 // the filtergraph contains internal sources and
1249 // requested to be scheduled directly
1250
2/2
✓ Branch 0 taken 2064 times.
✓ Branch 1 taken 470882 times.
472946 if (fg->best_input == fg->nb_inputs) {
1251 2064 fg->waiter.choked_next = 0;
1252 2064 have_unchoked = 1;
1253 2064 continue;
1254 }
1255
1256 470882 fi = &fg->inputs[fg->best_input];
1257 470882 d = &sch->demux[fi->src_sched.idx];
1258 } else
1259 72761 d = &sch->demux[ms->src_sched.idx];
1260
1261 543643 d->waiter.choked_next = 0;
1262 543643 have_unchoked = 1;
1263 }
1264 }
1265
1266 // make sure to unchoke at least one source, if still available
1267
4/4
✓ Branch 0 taken 54973 times.
✓ Branch 1 taken 512941 times.
✓ Branch 2 taken 41942 times.
✓ Branch 3 taken 13031 times.
567914 for (unsigned type = 0; !have_unchoked && type < 2; type++)
1268
4/4
✓ Branch 0 taken 30045 times.
✓ Branch 1 taken 42281 times.
✓ Branch 2 taken 41832 times.
✓ Branch 3 taken 30494 times.
72326 for (unsigned i = 0; i < (type ? sch->nb_filters : sch->nb_demux); i++) {
1269
2/2
✓ Branch 0 taken 17014 times.
✓ Branch 1 taken 24818 times.
41832 int exited = type ? sch->filters[i].task_exited : sch->demux[i].task_exited;
1270
2/2
✓ Branch 0 taken 17014 times.
✓ Branch 1 taken 24818 times.
41832 SchWaiter *w = type ? &sch->filters[i].waiter : &sch->demux[i].waiter;
1271
2/2
✓ Branch 0 taken 11448 times.
✓ Branch 1 taken 30384 times.
41832 if (!exited) {
1272 11448 w->choked_next = 0;
1273 11448 have_unchoked = 1;
1274 11448 break;
1275 }
1276 }
1277
1278
1279
2/2
✓ Branch 0 taken 1051944 times.
✓ Branch 1 taken 525972 times.
1577916 for (unsigned type = 0; type < 2; type++)
1280
4/4
✓ Branch 0 taken 1000898 times.
✓ Branch 1 taken 1056756 times.
✓ Branch 2 taken 1005710 times.
✓ Branch 3 taken 1051944 times.
2057654 for (unsigned i = 0; i < (type ? sch->nb_filters : sch->nb_demux); i++) {
1281
2/2
✓ Branch 0 taken 474926 times.
✓ Branch 1 taken 530784 times.
1005710 SchWaiter *w = type ? &sch->filters[i].waiter : &sch->demux[i].waiter;
1282
2/2
✓ Branch 0 taken 17372 times.
✓ Branch 1 taken 988338 times.
1005710 if (w->choked_prev != w->choked_next)
1283 17372 waiter_set(w, w->choked_next);
1284 }
1285
1286 }
1287
1288 6751 int sch_start(Scheduler *sch)
1289 {
1290 int ret;
1291
1292 6751 sch->transcode_started = 1;
1293
1294
2/2
✓ Branch 0 taken 6753 times.
✓ Branch 1 taken 6751 times.
13504 for (unsigned i = 0; i < sch->nb_mux; i++) {
1295 6753 SchMux *mux = &sch->mux[i];
1296
1297
2/2
✓ Branch 0 taken 7094 times.
✓ Branch 1 taken 6753 times.
13847 for (unsigned j = 0; j < mux->nb_streams; j++) {
1298 7094 SchMuxStream *ms = &mux->streams[j];
1299
1300
2/3
✓ Branch 0 taken 6474 times.
✓ Branch 1 taken 620 times.
✗ Branch 2 not taken.
7094 switch (ms->src.type) {
1301 6474 case SCH_NODE_TYPE_ENC: {
1302 6474 SchEnc *enc = &sch->enc[ms->src.idx];
1303
2/2
✓ Branch 0 taken 38 times.
✓ Branch 1 taken 6436 times.
6474 if (enc->src.type == SCH_NODE_TYPE_DEC) {
1304 38 ms->src_sched = sch->dec[enc->src.idx].src;
1305
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 38 times.
38 av_assert0(ms->src_sched.type == SCH_NODE_TYPE_DEMUX);
1306 } else {
1307 6436 ms->src_sched = enc->src;
1308
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6436 times.
6436 av_assert0(ms->src_sched.type == SCH_NODE_TYPE_FILTER_OUT);
1309 }
1310 6474 break;
1311 }
1312 620 case SCH_NODE_TYPE_DEMUX:
1313 620 ms->src_sched = ms->src;
1314 620 break;
1315 default:
1316 av_log(mux, AV_LOG_ERROR,
1317 "Muxer stream #%u not connected to a source\n", j);
1318 return AVERROR(EINVAL);
1319 }
1320 }
1321
1322 6753 ret = queue_alloc(&mux->queue, mux->nb_streams, mux->queue_size,
1323 QUEUE_PACKETS);
1324
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6753 times.
6753 if (ret < 0)
1325 return ret;
1326
1327
2/2
✓ Branch 0 taken 432 times.
✓ Branch 1 taken 6321 times.
6753 if (mux->nb_streams_ready == mux->nb_streams) {
1328 432 ret = mux_init(sch, mux);
1329
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 432 times.
432 if (ret < 0)
1330 return ret;
1331 }
1332 }
1333
1334
2/2
✓ Branch 0 taken 6474 times.
✓ Branch 1 taken 6751 times.
13225 for (unsigned i = 0; i < sch->nb_enc; i++) {
1335 6474 SchEnc *enc = &sch->enc[i];
1336
1337
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6474 times.
6474 if (!enc->src.type) {
1338 av_log(enc, AV_LOG_ERROR,
1339 "Encoder not connected to a source\n");
1340 return AVERROR(EINVAL);
1341 }
1342
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6474 times.
6474 if (!enc->dst.type) {
1343 av_log(enc, AV_LOG_ERROR,
1344 "Encoder not connected to a sink\n");
1345 return AVERROR(EINVAL);
1346 }
1347
1348 6474 ret = task_start(&enc->task);
1349
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6474 times.
6474 if (ret < 0)
1350 return ret;
1351 }
1352
1353
2/2
✓ Branch 0 taken 6386 times.
✓ Branch 1 taken 6751 times.
13137 for (unsigned i = 0; i < sch->nb_filters; i++) {
1354 6386 SchFilterGraph *fg = &sch->filters[i];
1355
1356
2/2
✓ Branch 0 taken 6400 times.
✓ Branch 1 taken 6386 times.
12786 for (unsigned j = 0; j < fg->nb_inputs; j++) {
1357 6400 SchFilterIn *fi = &fg->inputs[j];
1358
1359
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6400 times.
6400 if (!fi->src.type) {
1360 av_log(fg, AV_LOG_ERROR,
1361 "Filtergraph input %u not connected to a source\n", j);
1362 return AVERROR(EINVAL);
1363 }
1364
1365 6400 fi->src_sched = sch->dec[fi->src.idx].src;
1366 }
1367
1368
2/2
✓ Branch 0 taken 6436 times.
✓ Branch 1 taken 6386 times.
12822 for (unsigned j = 0; j < fg->nb_outputs; j++) {
1369 6436 SchFilterOut *fo = &fg->outputs[j];
1370
1371
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6436 times.
6436 if (!fo->dst.type) {
1372 av_log(fg, AV_LOG_ERROR,
1373 "Filtergraph %u output %u not connected to a sink\n", i, j);
1374 return AVERROR(EINVAL);
1375 }
1376 }
1377
1378 6386 ret = task_start(&fg->task);
1379
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6386 times.
6386 if (ret < 0)
1380 return ret;
1381 }
1382
1383
2/2
✓ Branch 0 taken 6392 times.
✓ Branch 1 taken 6751 times.
13143 for (unsigned i = 0; i < sch->nb_dec; i++) {
1384 6392 SchDec *dec = &sch->dec[i];
1385
1386
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 if (!dec->src.type) {
1387 av_log(dec, AV_LOG_ERROR,
1388 "Decoder not connected to a source\n");
1389 return AVERROR(EINVAL);
1390 }
1391
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 if (!dec->nb_dst) {
1392 av_log(dec, AV_LOG_ERROR,
1393 "Decoder not connected to any sink\n");
1394 return AVERROR(EINVAL);
1395 }
1396
1397 6392 dec->dst_finished = av_calloc(dec->nb_dst, sizeof(*dec->dst_finished));
1398
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 if (!dec->dst_finished)
1399 return AVERROR(ENOMEM);
1400
1401 6392 ret = task_start(&dec->task);
1402
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6392 times.
6392 if (ret < 0)
1403 return ret;
1404 }
1405
1406
2/2
✓ Branch 0 taken 6782 times.
✓ Branch 1 taken 6751 times.
13533 for (unsigned i = 0; i < sch->nb_demux; i++) {
1407 6782 SchDemux *d = &sch->demux[i];
1408
1409
2/2
✓ Branch 0 taken 14 times.
✓ Branch 1 taken 6768 times.
6782 if (!d->nb_streams)
1410 14 continue;
1411
1412
2/2
✓ Branch 0 taken 6989 times.
✓ Branch 1 taken 6768 times.
13757 for (unsigned j = 0; j < d->nb_streams; j++) {
1413 6989 SchDemuxStream *ds = &d->streams[j];
1414
1415
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6989 times.
6989 if (!ds->nb_dst) {
1416 av_log(d, AV_LOG_ERROR,
1417 "Demuxer stream %u not connected to any sink\n", j);
1418 return AVERROR(EINVAL);
1419 }
1420
1421 6989 ds->dst_finished = av_calloc(ds->nb_dst, sizeof(*ds->dst_finished));
1422
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6989 times.
6989 if (!ds->dst_finished)
1423 return AVERROR(ENOMEM);
1424 }
1425
1426 6768 ret = task_start(&d->task);
1427
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6768 times.
6768 if (ret < 0)
1428 return ret;
1429 }
1430
1431 6751 pthread_mutex_lock(&sch->schedule_lock);
1432 6751 schedule_update_locked(sch);
1433 6751 pthread_mutex_unlock(&sch->schedule_lock);
1434
1435 6751 return 0;
1436 }
1437
1438 21235 int sch_wait(Scheduler *sch, uint64_t timeout_us, int64_t *transcode_ts)
1439 {
1440 int ret, err;
1441
1442 // convert delay to absolute timestamp
1443 21235 timeout_us += av_gettime();
1444
1445 21235 pthread_mutex_lock(&sch->mux_done_lock);
1446
1447
1/2
✓ Branch 0 taken 21235 times.
✗ Branch 1 not taken.
21235 if (sch->nb_mux_done < sch->nb_mux) {
1448 21235 struct timespec tv = { .tv_sec = timeout_us / 1000000,
1449 21235 .tv_nsec = (timeout_us % 1000000) * 1000 };
1450 21235 pthread_cond_timedwait(&sch->mux_done_cond, &sch->mux_done_lock, &tv);
1451 }
1452
1453 21235 ret = sch->nb_mux_done == sch->nb_mux;
1454
1455 21235 pthread_mutex_unlock(&sch->mux_done_lock);
1456
1457 21235 *transcode_ts = atomic_load(&sch->last_dts);
1458
1459 // abort transcoding if any task failed
1460 21235 err = atomic_load(&sch->task_failed);
1461
1462
3/4
✓ Branch 0 taken 14484 times.
✓ Branch 1 taken 6751 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 14484 times.
21235 return ret || err;
1463 }
1464
1465 6436 static int enc_open(Scheduler *sch, SchEnc *enc, const AVFrame *frame)
1466 {
1467 int ret;
1468
1469 6436 ret = enc->open_cb(enc->task.func_arg, frame);
1470
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6436 times.
6436 if (ret < 0)
1471 return ret;
1472
1473 // ret>0 signals audio frame size, which means sync queue must
1474 // have been enabled during encoder creation
1475
2/2
✓ Branch 0 taken 156 times.
✓ Branch 1 taken 6280 times.
6436 if (ret > 0) {
1476 SchSyncQueue *sq;
1477
1478
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 156 times.
156 av_assert0(enc->sq_idx[0] >= 0);
1479 156 sq = &sch->sq_enc[enc->sq_idx[0]];
1480
1481 156 pthread_mutex_lock(&sq->lock);
1482
1483 156 sq_frame_samples(sq->sq, enc->sq_idx[1], ret);
1484
1485 156 pthread_mutex_unlock(&sq->lock);
1486 }
1487
1488 6436 return 0;
1489 }
1490
1491 444147 static int send_to_enc_thread(Scheduler *sch, SchEnc *enc, AVFrame *frame)
1492 {
1493 int ret;
1494
1495
2/2
✓ Branch 0 taken 13077 times.
✓ Branch 1 taken 431070 times.
444147 if (!frame) {
1496 13077 tq_send_finish(enc->queue, 0);
1497 13077 return 0;
1498 }
1499
1500
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 431070 times.
431070 if (enc->in_finished)
1501 return AVERROR_EOF;
1502
1503 431070 ret = tq_send(enc->queue, 0, frame);
1504
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 431067 times.
431070 if (ret < 0)
1505 3 enc->in_finished = 1;
1506
1507 431070 return ret;
1508 }
1509
1510 68912 static int send_to_enc_sq(Scheduler *sch, SchEnc *enc, AVFrame *frame)
1511 {
1512 68912 SchSyncQueue *sq = &sch->sq_enc[enc->sq_idx[0]];
1513 68912 int ret = 0;
1514
1515 // inform the scheduling code that no more input will arrive along this path;
1516 // this is necessary because the sync queue may not send an EOF downstream
1517 // until other streams finish
1518 // TODO: consider a cleaner way of passing this information through
1519 // the pipeline
1520
2/2
✓ Branch 0 taken 2989 times.
✓ Branch 1 taken 65923 times.
68912 if (!frame) {
1521 2989 SchMux *mux = &sch->mux[enc->dst.idx];
1522 2989 SchMuxStream *ms = &mux->streams[enc->dst.idx_stream];
1523
1524 2989 pthread_mutex_lock(&sch->schedule_lock);
1525
1526 2989 ms->source_finished = 1;
1527 2989 schedule_update_locked(sch);
1528
1529 2989 pthread_mutex_unlock(&sch->schedule_lock);
1530 }
1531
1532 68912 pthread_mutex_lock(&sq->lock);
1533
1534 68912 ret = sq_send(sq->sq, enc->sq_idx[1], SQFRAME(frame));
1535
2/2
✓ Branch 0 taken 68910 times.
✓ Branch 1 taken 2 times.
68912 if (ret < 0)
1536 2 goto finish;
1537
1538 81835 while (1) {
1539 SchEnc *enc;
1540
1541 // TODO: the SQ API should be extended to allow returning EOF
1542 // for individual streams
1543 150745 ret = sq_receive(sq->sq, -1, SQFRAME(sq->frame));
1544
2/2
✓ Branch 0 taken 68910 times.
✓ Branch 1 taken 81835 times.
150745 if (ret < 0) {
1545
2/2
✓ Branch 0 taken 5582 times.
✓ Branch 1 taken 63328 times.
68910 ret = (ret == AVERROR(EAGAIN)) ? 0 : ret;
1546 68910 break;
1547 }
1548
1549 81835 enc = &sch->enc[sq->enc_idx[ret]];
1550 81835 ret = send_to_enc_thread(sch, enc, sq->frame);
1551
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 81833 times.
81835 if (ret < 0) {
1552 2 av_frame_unref(sq->frame);
1553
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
2 if (ret != AVERROR_EOF)
1554 break;
1555
1556 2 sq_send(sq->sq, enc->sq_idx[1], SQFRAME(NULL));
1557 2 continue;
1558 }
1559 }
1560
1561
2/2
✓ Branch 0 taken 63328 times.
✓ Branch 1 taken 5582 times.
68910 if (ret < 0) {
1562 // close all encoders fed from this sync queue
1563
2/2
✓ Branch 0 taken 5812 times.
✓ Branch 1 taken 5582 times.
11394 for (unsigned i = 0; i < sq->nb_enc_idx; i++) {
1564 5812 int err = send_to_enc_thread(sch, &sch->enc[sq->enc_idx[i]], NULL);
1565
1566 // if the sync queue error is EOF and closing the encoder
1567 // produces a more serious error, make sure to pick the latter
1568
2/4
✓ Branch 0 taken 5812 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 5812 times.
✗ Branch 3 not taken.
5812 ret = err_merge((ret == AVERROR_EOF && err < 0) ? 0 : ret, err);
1569 }
1570 }
1571
1572 68910 finish:
1573 68912 pthread_mutex_unlock(&sq->lock);
1574
1575 68912 return ret;
1576 }
1577
1578 425415 static int send_to_enc(Scheduler *sch, SchEnc *enc, AVFrame *frame)
1579 {
1580
6/6
✓ Branch 0 taken 424498 times.
✓ Branch 1 taken 917 times.
✓ Branch 2 taken 414282 times.
✓ Branch 3 taken 10216 times.
✓ Branch 4 taken 6436 times.
✓ Branch 5 taken 407846 times.
425415 if (enc->open_cb && frame && !enc->opened) {
1581 6436 int ret = enc_open(sch, enc, frame);
1582
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6436 times.
6436 if (ret < 0)
1583 return ret;
1584 6436 enc->opened = 1;
1585
1586 // discard empty frames that only carry encoder init parameters
1587
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 6433 times.
6436 if (!frame->buf[0]) {
1588 3 av_frame_unref(frame);
1589 3 return 0;
1590 }
1591 }
1592
1593 425412 return (enc->sq_idx[0] >= 0) ?
1594
2/2
✓ Branch 0 taken 68912 times.
✓ Branch 1 taken 356500 times.
781912 send_to_enc_sq (sch, enc, frame) :
1595 356500 send_to_enc_thread(sch, enc, frame);
1596 }
1597
1598 3193 static int mux_queue_packet(SchMux *mux, SchMuxStream *ms, AVPacket *pkt)
1599 {
1600 3193 PreMuxQueue *q = &ms->pre_mux_queue;
1601 3193 AVPacket *tmp_pkt = NULL;
1602 int ret;
1603
1604
2/2
✓ Branch 1 taken 145 times.
✓ Branch 2 taken 3048 times.
3193 if (!av_fifo_can_write(q->fifo)) {
1605 145 size_t packets = av_fifo_can_read(q->fifo);
1606
1/2
✓ Branch 0 taken 145 times.
✗ Branch 1 not taken.
145 size_t pkt_size = pkt ? pkt->size : 0;
1607 145 int thresh_reached = (q->data_size + pkt_size) > q->data_threshold;
1608
2/2
✓ Branch 0 taken 140 times.
✓ Branch 1 taken 5 times.
145 size_t max_packets = thresh_reached ? q->max_packets : SIZE_MAX;
1609 145 size_t new_size = FFMIN(2 * packets, max_packets);
1610
1611
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 145 times.
145 if (new_size <= packets) {
1612 av_log(mux, AV_LOG_ERROR,
1613 "Too many packets buffered for output stream.\n");
1614 return AVERROR(ENOSPC);
1615 }
1616 145 ret = av_fifo_grow2(q->fifo, new_size - packets);
1617
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 145 times.
145 if (ret < 0)
1618 return ret;
1619 }
1620
1621
2/2
✓ Branch 0 taken 3151 times.
✓ Branch 1 taken 42 times.
3193 if (pkt) {
1622 3151 tmp_pkt = av_packet_alloc();
1623
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3151 times.
3151 if (!tmp_pkt)
1624 return AVERROR(ENOMEM);
1625
1626 3151 av_packet_move_ref(tmp_pkt, pkt);
1627 3151 q->data_size += tmp_pkt->size;
1628 }
1629 3193 av_fifo_write(q->fifo, &tmp_pkt, 1);
1630
1631 3193 return 0;
1632 }
1633
1634 503608 static int send_to_mux(Scheduler *sch, SchMux *mux, unsigned stream_idx,
1635 AVPacket *pkt)
1636 {
1637 503608 SchMuxStream *ms = &mux->streams[stream_idx];
1638
2/2
✓ Branch 0 taken 487494 times.
✓ Branch 1 taken 9020 times.
496514 int64_t dts = (pkt && pkt->dts != AV_NOPTS_VALUE) ?
1639
2/2
✓ Branch 0 taken 496514 times.
✓ Branch 1 taken 7094 times.
1000122 av_rescale_q(pkt->dts + pkt->duration, pkt->time_base, AV_TIME_BASE_Q) :
1640 AV_NOPTS_VALUE;
1641
1642 // queue the packet if the muxer cannot be started yet
1643
2/2
✓ Branch 0 taken 3224 times.
✓ Branch 1 taken 500384 times.
503608 if (!atomic_load(&mux->mux_started)) {
1644 3224 int queued = 0;
1645
1646 // the muxer could have started between the above atomic check and
1647 // locking the mutex, then this block falls through to normal send path
1648 3224 pthread_mutex_lock(&sch->mux_ready_lock);
1649
1650
2/2
✓ Branch 0 taken 3193 times.
✓ Branch 1 taken 31 times.
3224 if (!atomic_load(&mux->mux_started)) {
1651 3193 int ret = mux_queue_packet(mux, ms, pkt);
1652
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3193 times.
3193 queued = ret < 0 ? ret : 1;
1653 }
1654
1655 3224 pthread_mutex_unlock(&sch->mux_ready_lock);
1656
1657
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3224 times.
3224 if (queued < 0)
1658 return queued;
1659
2/2
✓ Branch 0 taken 3193 times.
✓ Branch 1 taken 31 times.
3224 else if (queued)
1660 3193 goto update_schedule;
1661 }
1662
1663
2/2
✓ Branch 0 taken 493363 times.
✓ Branch 1 taken 7052 times.
500415 if (pkt) {
1664 int ret;
1665
1666
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 493362 times.
493363 if (ms->init_eof)
1667 1 return AVERROR_EOF;
1668
1669 493362 ret = tq_send(mux->queue, stream_idx, pkt);
1670
2/2
✓ Branch 0 taken 46 times.
✓ Branch 1 taken 493316 times.
493362 if (ret < 0)
1671 46 return ret;
1672 } else
1673 7052 tq_send_finish(mux->queue, stream_idx);
1674
1675 503561 update_schedule:
1676 // TODO: use atomics to check whether this changes trailing dts
1677 // to avoid locking unnecesarily
1678
4/4
✓ Branch 0 taken 16114 times.
✓ Branch 1 taken 487447 times.
✓ Branch 2 taken 7094 times.
✓ Branch 3 taken 9020 times.
503561 if (dts != AV_NOPTS_VALUE || !pkt) {
1679 494541 pthread_mutex_lock(&sch->schedule_lock);
1680
1681
2/2
✓ Branch 0 taken 487447 times.
✓ Branch 1 taken 7094 times.
494541 if (pkt) ms->last_dts = dts;
1682 7094 else ms->source_finished = 1;
1683
1684 494541 schedule_update_locked(sch);
1685
1686 494541 pthread_mutex_unlock(&sch->schedule_lock);
1687 }
1688
1689 503561 return 0;
1690 }
1691
1692 static int
1693 517541 demux_stream_send_to_dst(Scheduler *sch, const SchedulerNode dst,
1694 uint8_t *dst_finished, AVPacket *pkt, unsigned flags)
1695 {
1696 int ret;
1697
1698
2/2
✓ Branch 0 taken 3075 times.
✓ Branch 1 taken 514466 times.
517541 if (*dst_finished)
1699 3075 return AVERROR_EOF;
1700
1701
4/4
✓ Branch 0 taken 510529 times.
✓ Branch 1 taken 3937 times.
✓ Branch 2 taken 71601 times.
✓ Branch 3 taken 438928 times.
514466 if (pkt && dst.type == SCH_NODE_TYPE_MUX &&
1702
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 71599 times.
71601 (flags & DEMUX_SEND_STREAMCOPY_EOF)) {
1703 2 av_packet_unref(pkt);
1704 2 pkt = NULL;
1705 }
1706
1707
2/2
✓ Branch 0 taken 3939 times.
✓ Branch 1 taken 510527 times.
514466 if (!pkt)
1708 3939 goto finish;
1709
1710 1021054 ret = (dst.type == SCH_NODE_TYPE_MUX) ?
1711
2/2
✓ Branch 0 taken 71599 times.
✓ Branch 1 taken 438928 times.
510527 send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, pkt) :
1712 438928 tq_send(sch->dec[dst.idx].queue, 0, pkt);
1713
2/2
✓ Branch 0 taken 3073 times.
✓ Branch 1 taken 507454 times.
510527 if (ret == AVERROR_EOF)
1714 3073 goto finish;
1715
1716 507454 return ret;
1717
1718 7012 finish:
1719
2/2
✓ Branch 0 taken 620 times.
✓ Branch 1 taken 6392 times.
7012 if (dst.type == SCH_NODE_TYPE_MUX)
1720 620 send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, NULL);
1721 else
1722 6392 tq_send_finish(sch->dec[dst.idx].queue, 0);
1723
1724 7012 *dst_finished = 1;
1725 7012 return AVERROR_EOF;
1726 }
1727
1728 514382 static int demux_send_for_stream(Scheduler *sch, SchDemux *d, SchDemuxStream *ds,
1729 AVPacket *pkt, unsigned flags)
1730 {
1731 514382 unsigned nb_done = 0;
1732
1733
2/2
✓ Branch 0 taken 517541 times.
✓ Branch 1 taken 514382 times.
1031923 for (unsigned i = 0; i < ds->nb_dst; i++) {
1734 517541 AVPacket *to_send = pkt;
1735 517541 uint8_t *finished = &ds->dst_finished[i];
1736
1737 int ret;
1738
1739 // sending a packet consumes it, so make a temporary reference if needed
1740
4/4
✓ Branch 0 taken 510529 times.
✓ Branch 1 taken 7012 times.
✓ Branch 2 taken 3136 times.
✓ Branch 3 taken 507393 times.
517541 if (pkt && i < ds->nb_dst - 1) {
1741 3136 to_send = d->send_pkt;
1742
1743 3136 ret = av_packet_ref(to_send, pkt);
1744
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3136 times.
3136 if (ret < 0)
1745 return ret;
1746 }
1747
1748 517541 ret = demux_stream_send_to_dst(sch, ds->dst[i], finished, to_send, flags);
1749
2/2
✓ Branch 0 taken 510529 times.
✓ Branch 1 taken 7012 times.
517541 if (to_send)
1750 510529 av_packet_unref(to_send);
1751
2/2
✓ Branch 0 taken 10087 times.
✓ Branch 1 taken 507454 times.
517541 if (ret == AVERROR_EOF)
1752 10087 nb_done++;
1753
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 507454 times.
507454 else if (ret < 0)
1754 return ret;
1755 }
1756
1757
2/2
✓ Branch 0 taken 10064 times.
✓ Branch 1 taken 504318 times.
514382 return (nb_done == ds->nb_dst) ? AVERROR_EOF : 0;
1758 }
1759
1760 11 static int demux_flush(Scheduler *sch, SchDemux *d, AVPacket *pkt)
1761 {
1762 11 Timestamp max_end_ts = (Timestamp){ .ts = AV_NOPTS_VALUE };
1763
1764
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);
1765
1766
2/2
✓ Branch 0 taken 14 times.
✓ Branch 1 taken 11 times.
25 for (unsigned i = 0; i < d->nb_streams; i++) {
1767 14 SchDemuxStream *ds = &d->streams[i];
1768
1769
2/2
✓ Branch 0 taken 14 times.
✓ Branch 1 taken 14 times.
28 for (unsigned j = 0; j < ds->nb_dst; j++) {
1770 14 const SchedulerNode *dst = &ds->dst[j];
1771 SchDec *dec;
1772 int ret;
1773
1774
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)
1775 8 continue;
1776
1777 6 dec = &sch->dec[dst->idx];
1778
1779 6 ret = tq_send(dec->queue, 0, pkt);
1780
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
6 if (ret < 0)
1781 return ret;
1782
1783
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 3 times.
6 if (dec->queue_end_ts) {
1784 Timestamp ts;
1785 3 ret = av_thread_message_queue_recv(dec->queue_end_ts, &ts, 0);
1786
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
3 if (ret < 0)
1787 return ret;
1788
1789
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
3 if (max_end_ts.ts == AV_NOPTS_VALUE ||
1790 (ts.ts != AV_NOPTS_VALUE &&
1791 av_compare_ts(max_end_ts.ts, max_end_ts.tb, ts.ts, ts.tb) < 0))
1792 3 max_end_ts = ts;
1793
1794 }
1795 }
1796 }
1797
1798 11 pkt->pts = max_end_ts.ts;
1799 11 pkt->time_base = max_end_ts.tb;
1800
1801 11 return 0;
1802 }
1803
1804 507455 int sch_demux_send(Scheduler *sch, unsigned demux_idx, AVPacket *pkt,
1805 unsigned flags)
1806 {
1807 SchDemux *d;
1808 int terminate;
1809
1810
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 507455 times.
507455 av_assert0(demux_idx < sch->nb_demux);
1811 507455 d = &sch->demux[demux_idx];
1812
1813 507455 terminate = waiter_wait(sch, &d->waiter);
1814
2/2
✓ Branch 0 taken 51 times.
✓ Branch 1 taken 507404 times.
507455 if (terminate)
1815 51 return AVERROR_EXIT;
1816
1817 // flush the downstreams after seek
1818
2/2
✓ Branch 0 taken 11 times.
✓ Branch 1 taken 507393 times.
507404 if (pkt->stream_index == -1)
1819 11 return demux_flush(sch, d, pkt);
1820
1821
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 507393 times.
507393 av_assert0(pkt->stream_index < d->nb_streams);
1822
1823 507393 return demux_send_for_stream(sch, d, &d->streams[pkt->stream_index], pkt, flags);
1824 }
1825
1826 6768 static int demux_done(Scheduler *sch, unsigned demux_idx)
1827 {
1828 6768 SchDemux *d = &sch->demux[demux_idx];
1829 6768 int ret = 0;
1830
1831
2/2
✓ Branch 0 taken 6989 times.
✓ Branch 1 taken 6768 times.
13757 for (unsigned i = 0; i < d->nb_streams; i++) {
1832 6989 int err = demux_send_for_stream(sch, d, &d->streams[i], NULL, 0);
1833
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6989 times.
6989 if (err != AVERROR_EOF)
1834 ret = err_merge(ret, err);
1835 }
1836
1837 6768 pthread_mutex_lock(&sch->schedule_lock);
1838
1839 6768 d->task_exited = 1;
1840
1841 6768 schedule_update_locked(sch);
1842
1843 6768 pthread_mutex_unlock(&sch->schedule_lock);
1844
1845 6768 return ret;
1846 }
1847
1848 510080 int sch_mux_receive(Scheduler *sch, unsigned mux_idx, AVPacket *pkt)
1849 {
1850 SchMux *mux;
1851 int ret, stream_idx;
1852
1853
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 510080 times.
510080 av_assert0(mux_idx < sch->nb_mux);
1854 510080 mux = &sch->mux[mux_idx];
1855
1856 510080 ret = tq_receive(mux->queue, &stream_idx, pkt);
1857 510080 pkt->stream_index = stream_idx;
1858 510080 return ret;
1859 }
1860
1861 188 void sch_mux_receive_finish(Scheduler *sch, unsigned mux_idx, unsigned stream_idx)
1862 {
1863 SchMux *mux;
1864
1865
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 188 times.
188 av_assert0(mux_idx < sch->nb_mux);
1866 188 mux = &sch->mux[mux_idx];
1867
1868
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 188 times.
188 av_assert0(stream_idx < mux->nb_streams);
1869 188 tq_receive_finish(mux->queue, stream_idx);
1870
1871 188 pthread_mutex_lock(&sch->schedule_lock);
1872 188 mux->streams[stream_idx].source_finished = 1;
1873
1874 188 schedule_update_locked(sch);
1875
1876 188 pthread_mutex_unlock(&sch->schedule_lock);
1877 188 }
1878
1879 460183 int sch_mux_sub_heartbeat(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
1880 const AVPacket *pkt)
1881 {
1882 SchMux *mux;
1883 SchMuxStream *ms;
1884
1885
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 460183 times.
460183 av_assert0(mux_idx < sch->nb_mux);
1886 460183 mux = &sch->mux[mux_idx];
1887
1888
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 460183 times.
460183 av_assert0(stream_idx < mux->nb_streams);
1889 460183 ms = &mux->streams[stream_idx];
1890
1891
2/2
✓ Branch 0 taken 5 times.
✓ Branch 1 taken 460183 times.
460188 for (unsigned i = 0; i < ms->nb_sub_heartbeat_dst; i++) {
1892 5 SchDec *dst = &sch->dec[ms->sub_heartbeat_dst[i]];
1893 int ret;
1894
1895 5 ret = av_packet_copy_props(mux->sub_heartbeat_pkt, pkt);
1896
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5 times.
5 if (ret < 0)
1897 return ret;
1898
1899 5 tq_send(dst->queue, 0, mux->sub_heartbeat_pkt);
1900 }
1901
1902 460183 return 0;
1903 }
1904
1905 6753 static int mux_done(Scheduler *sch, unsigned mux_idx)
1906 {
1907 6753 SchMux *mux = &sch->mux[mux_idx];
1908
1909 6753 pthread_mutex_lock(&sch->schedule_lock);
1910
1911
2/2
✓ Branch 0 taken 7094 times.
✓ Branch 1 taken 6753 times.
13847 for (unsigned i = 0; i < mux->nb_streams; i++) {
1912 7094 tq_receive_finish(mux->queue, i);
1913 7094 mux->streams[i].source_finished = 1;
1914 }
1915
1916 6753 schedule_update_locked(sch);
1917
1918 6753 pthread_mutex_unlock(&sch->schedule_lock);
1919
1920 6753 pthread_mutex_lock(&sch->mux_done_lock);
1921
1922
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6753 times.
6753 av_assert0(sch->nb_mux_done < sch->nb_mux);
1923 6753 sch->nb_mux_done++;
1924
1925 6753 pthread_cond_signal(&sch->mux_done_cond);
1926
1927 6753 pthread_mutex_unlock(&sch->mux_done_lock);
1928
1929 6753 return 0;
1930 }
1931
1932 415531 int sch_dec_receive(Scheduler *sch, unsigned dec_idx, AVPacket *pkt)
1933 {
1934 SchDec *dec;
1935 int ret, dummy;
1936
1937
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 415531 times.
415531 av_assert0(dec_idx < sch->nb_dec);
1938 415531 dec = &sch->dec[dec_idx];
1939
1940 // the decoder should have given us post-flush end timestamp in pkt
1941
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 415528 times.
415531 if (dec->expect_end_ts) {
1942 3 Timestamp ts = (Timestamp){ .ts = pkt->pts, .tb = pkt->time_base };
1943 3 ret = av_thread_message_queue_send(dec->queue_end_ts, &ts, 0);
1944
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
3 if (ret < 0)
1945 return ret;
1946
1947 3 dec->expect_end_ts = 0;
1948 }
1949
1950 415531 ret = tq_receive(dec->queue, &dummy, pkt);
1951
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 415531 times.
415531 av_assert0(dummy <= 0);
1952
1953 // got a flush packet, on the next call to this function the decoder
1954 // will give us post-flush end timestamp
1955
7/8
✓ Branch 0 taken 412225 times.
✓ Branch 1 taken 3306 times.
✓ Branch 2 taken 958 times.
✓ Branch 3 taken 411267 times.
✓ Branch 4 taken 958 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3 times.
✓ Branch 7 taken 955 times.
415531 if (ret >= 0 && !pkt->data && !pkt->side_data_elems && dec->queue_end_ts)
1956 3 dec->expect_end_ts = 1;
1957
1958 415531 return ret;
1959 }
1960
1961 450165 static int send_to_filter(Scheduler *sch, SchFilterGraph *fg,
1962 unsigned in_idx, AVFrame *frame)
1963 {
1964
2/2
✓ Branch 0 taken 443765 times.
✓ Branch 1 taken 6400 times.
450165 if (frame)
1965 443765 return tq_send(fg->queue, in_idx, frame);
1966
1967
1/2
✓ Branch 0 taken 6400 times.
✗ Branch 1 not taken.
6400 if (!fg->inputs[in_idx].send_finished) {
1968 6400 fg->inputs[in_idx].send_finished = 1;
1969 6400 tq_send_finish(fg->queue, in_idx);
1970
1971 // close the control stream when all actual inputs are done
1972
2/2
✓ Branch 0 taken 6331 times.
✓ Branch 1 taken 69 times.
6400 if (atomic_fetch_add(&fg->nb_inputs_finished_send, 1) == fg->nb_inputs - 1)
1973 6331 tq_send_finish(fg->queue, fg->nb_inputs);
1974 }
1975 6400 return 0;
1976 }
1977
1978 454217 static int dec_send_to_dst(Scheduler *sch, const SchedulerNode dst,
1979 uint8_t *dst_finished, AVFrame *frame)
1980 {
1981 int ret;
1982
1983
2/2
✓ Branch 0 taken 6261 times.
✓ Branch 1 taken 447956 times.
454217 if (*dst_finished)
1984 6261 return AVERROR_EOF;
1985
1986
2/2
✓ Branch 0 taken 3312 times.
✓ Branch 1 taken 444644 times.
447956 if (!frame)
1987 3312 goto finish;
1988
1989 889288 ret = (dst.type == SCH_NODE_TYPE_FILTER_IN) ?
1990
2/2
✓ Branch 0 taken 443765 times.
✓ Branch 1 taken 879 times.
444644 send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, frame) :
1991 879 send_to_enc(sch, &sch->enc[dst.idx], frame);
1992
2/2
✓ Branch 0 taken 3126 times.
✓ Branch 1 taken 441518 times.
444644 if (ret == AVERROR_EOF)
1993 3126 goto finish;
1994
1995 441518 return ret;
1996
1997 6438 finish:
1998
2/2
✓ Branch 0 taken 6400 times.
✓ Branch 1 taken 38 times.
6438 if (dst.type == SCH_NODE_TYPE_FILTER_IN)
1999 6400 send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, NULL);
2000 else
2001 38 send_to_enc(sch, &sch->enc[dst.idx], NULL);
2002
2003 6438 *dst_finished = 1;
2004
2005 6438 return AVERROR_EOF;
2006 }
2007
2008 439879 int sch_dec_send(Scheduler *sch, unsigned dec_idx, AVFrame *frame)
2009 {
2010 SchDec *dec;
2011 439879 int ret = 0;
2012 439879 unsigned nb_done = 0;
2013
2014
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 439879 times.
439879 av_assert0(dec_idx < sch->nb_dec);
2015 439879 dec = &sch->dec[dec_idx];
2016
2017
2/2
✓ Branch 0 taken 447779 times.
✓ Branch 1 taken 439879 times.
887658 for (unsigned i = 0; i < dec->nb_dst; i++) {
2018 447779 uint8_t *finished = &dec->dst_finished[i];
2019 447779 AVFrame *to_send = frame;
2020
2021 // sending a frame consumes it, so make a temporary reference if needed
2022
2/2
✓ Branch 0 taken 7900 times.
✓ Branch 1 taken 439879 times.
447779 if (i < dec->nb_dst - 1) {
2023 7900 to_send = dec->send_frame;
2024
2025 // frame may sometimes contain props only,
2026 // e.g. to signal EOF timestamp
2027
2/2
✓ Branch 0 taken 7808 times.
✓ Branch 1 taken 92 times.
7900 ret = frame->buf[0] ? av_frame_ref(to_send, frame) :
2028 92 av_frame_copy_props(to_send, frame);
2029
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7900 times.
7900 if (ret < 0)
2030 return ret;
2031 }
2032
2033 447779 ret = dec_send_to_dst(sch, dec->dst[i], finished, to_send);
2034
2/2
✓ Branch 0 taken 6261 times.
✓ Branch 1 taken 441518 times.
447779 if (ret < 0) {
2035 6261 av_frame_unref(to_send);
2036
1/2
✓ Branch 0 taken 6261 times.
✗ Branch 1 not taken.
6261 if (ret == AVERROR_EOF) {
2037 6261 nb_done++;
2038 6261 ret = 0;
2039 6261 continue;
2040 }
2041 return ret;
2042 }
2043 }
2044
2045
2/2
✓ Branch 0 taken 6184 times.
✓ Branch 1 taken 433695 times.
439879 return (nb_done == dec->nb_dst) ? AVERROR_EOF : 0;
2046 }
2047
2048 6392 static int dec_done(Scheduler *sch, unsigned dec_idx)
2049 {
2050 6392 SchDec *dec = &sch->dec[dec_idx];
2051 6392 int ret = 0;
2052
2053 6392 tq_receive_finish(dec->queue, 0);
2054
2055 // make sure our source does not get stuck waiting for end timestamps
2056 // that will never arrive
2057
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 6391 times.
6392 if (dec->queue_end_ts)
2058 1 av_thread_message_queue_set_err_recv(dec->queue_end_ts, AVERROR_EOF);
2059
2060
2/2
✓ Branch 0 taken 6438 times.
✓ Branch 1 taken 6392 times.
12830 for (unsigned i = 0; i < dec->nb_dst; i++) {
2061 6438 int err = dec_send_to_dst(sch, dec->dst[i], &dec->dst_finished[i], NULL);
2062
2/4
✓ Branch 0 taken 6438 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6438 times.
6438 if (err < 0 && err != AVERROR_EOF)
2063 ret = err_merge(ret, err);
2064 }
2065
2066 6392 return ret;
2067 }
2068
2069 437520 int sch_enc_receive(Scheduler *sch, unsigned enc_idx, AVFrame *frame)
2070 {
2071 SchEnc *enc;
2072 int ret, dummy;
2073
2074
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 437520 times.
437520 av_assert0(enc_idx < sch->nb_enc);
2075 437520 enc = &sch->enc[enc_idx];
2076
2077 437520 ret = tq_receive(enc->queue, &dummy, frame);
2078
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 437520 times.
437520 av_assert0(dummy <= 0);
2079
2080 437520 return ret;
2081 }
2082
2083 424915 int sch_enc_send(Scheduler *sch, unsigned enc_idx, AVPacket *pkt)
2084 {
2085 SchEnc *enc;
2086 int ret;
2087
2088
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 424915 times.
424915 av_assert0(enc_idx < sch->nb_enc);
2089 424915 enc = &sch->enc[enc_idx];
2090
2091
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 424915 times.
424915 if (enc->out_finished)
2092 return pkt ? AVERROR_EOF : 0;
2093
2094 424915 ret = send_to_mux(sch, &sch->mux[enc->dst.idx], enc->dst.idx_stream, pkt);
2095
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 424913 times.
424915 if (ret < 0)
2096 2 enc->out_finished = 1;
2097
2098 424915 return ret;
2099 }
2100
2101 6474 static int enc_done(Scheduler *sch, unsigned enc_idx)
2102 {
2103 6474 SchEnc *enc = &sch->enc[enc_idx];
2104
2105 6474 tq_receive_finish(enc->queue, 0);
2106
2107 6474 return send_to_mux(sch, &sch->mux[enc->dst.idx], enc->dst.idx_stream, NULL);
2108 }
2109
2110 419619 int sch_filter_receive(Scheduler *sch, unsigned fg_idx,
2111 unsigned *in_idx, AVFrame *frame)
2112 {
2113 SchFilterGraph *fg;
2114
2115
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 419619 times.
419619 av_assert0(fg_idx < sch->nb_filters);
2116 419619 fg = &sch->filters[fg_idx];
2117
2118
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 419619 times.
419619 av_assert0(*in_idx <= fg->nb_inputs);
2119
2120 // update scheduling to account for desired input stream, if it changed
2121 //
2122 // this check needs no locking because only the filtering thread
2123 // updates this value
2124
2/2
✓ Branch 0 taken 4554 times.
✓ Branch 1 taken 415065 times.
419619 if (*in_idx != fg->best_input) {
2125 4554 pthread_mutex_lock(&sch->schedule_lock);
2126
2127 4554 fg->best_input = *in_idx;
2128 4554 schedule_update_locked(sch);
2129
2130 4554 pthread_mutex_unlock(&sch->schedule_lock);
2131 }
2132
2133
2/2
✓ Branch 0 taken 417957 times.
✓ Branch 1 taken 1662 times.
419619 if (*in_idx == fg->nb_inputs) {
2134 1662 int terminate = waiter_wait(sch, &fg->waiter);
2135
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1662 times.
1662 return terminate ? AVERROR_EOF : AVERROR(EAGAIN);
2136 }
2137
2138 10 while (1) {
2139 int ret, idx;
2140
2141 417967 ret = tq_receive(fg->queue, &idx, frame);
2142
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 417967 times.
417967 if (idx < 0)
2143 417957 return AVERROR_EOF;
2144
2/2
✓ Branch 0 taken 417957 times.
✓ Branch 1 taken 10 times.
417967 else if (ret >= 0) {
2145 417957 *in_idx = idx;
2146 417957 return 0;
2147 }
2148
2149 // disregard EOFs for specific streams - they should always be
2150 // preceded by an EOF frame
2151 }
2152 }
2153
2154 1 void sch_filter_receive_finish(Scheduler *sch, unsigned fg_idx, unsigned in_idx)
2155 {
2156 SchFilterGraph *fg;
2157 SchFilterIn *fi;
2158
2159
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 av_assert0(fg_idx < sch->nb_filters);
2160 1 fg = &sch->filters[fg_idx];
2161
2162
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 av_assert0(in_idx < fg->nb_inputs);
2163 1 fi = &fg->inputs[in_idx];
2164
2165
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
1 if (!fi->receive_finished) {
2166 1 fi->receive_finished = 1;
2167 1 tq_receive_finish(fg->queue, in_idx);
2168
2169 // close the control stream when all actual inputs are done
2170
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
1 if (++fg->nb_inputs_finished_receive == fg->nb_inputs)
2171 tq_receive_finish(fg->queue, fg->nb_inputs);
2172 }
2173 1 }
2174
2175 418062 int sch_filter_send(Scheduler *sch, unsigned fg_idx, unsigned out_idx, AVFrame *frame)
2176 {
2177 SchFilterGraph *fg;
2178
2179
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 418062 times.
418062 av_assert0(fg_idx < sch->nb_filters);
2180 418062 fg = &sch->filters[fg_idx];
2181
2182
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 418062 times.
418062 av_assert0(out_idx < fg->nb_outputs);
2183 418062 return send_to_enc(sch, &sch->enc[fg->outputs[out_idx].dst.idx], frame);
2184 }
2185
2186 6386 static int filter_done(Scheduler *sch, unsigned fg_idx)
2187 {
2188 6386 SchFilterGraph *fg = &sch->filters[fg_idx];
2189 6386 int ret = 0;
2190
2191
2/2
✓ Branch 0 taken 12786 times.
✓ Branch 1 taken 6386 times.
19172 for (unsigned i = 0; i <= fg->nb_inputs; i++)
2192 12786 tq_receive_finish(fg->queue, i);
2193
2194
2/2
✓ Branch 0 taken 6436 times.
✓ Branch 1 taken 6386 times.
12822 for (unsigned i = 0; i < fg->nb_outputs; i++) {
2195 6436 SchEnc *enc = &sch->enc[fg->outputs[i].dst.idx];
2196 6436 int err = send_to_enc(sch, enc, NULL);
2197
3/4
✓ Branch 0 taken 2805 times.
✓ Branch 1 taken 3631 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 2805 times.
6436 if (err < 0 && err != AVERROR_EOF)
2198 ret = err_merge(ret, err);
2199 }
2200
2201 6386 pthread_mutex_lock(&sch->schedule_lock);
2202
2203 6386 fg->task_exited = 1;
2204
2205 6386 schedule_update_locked(sch);
2206
2207 6386 pthread_mutex_unlock(&sch->schedule_lock);
2208
2209 6386 return ret;
2210 }
2211
2212 int sch_filter_command(Scheduler *sch, unsigned fg_idx, AVFrame *frame)
2213 {
2214 SchFilterGraph *fg;
2215
2216 av_assert0(fg_idx < sch->nb_filters);
2217 fg = &sch->filters[fg_idx];
2218
2219 return send_to_filter(sch, fg, fg->nb_inputs, frame);
2220 }
2221
2222 32773 static void *task_wrapper(void *arg)
2223 {
2224 32773 SchTask *task = arg;
2225 32773 Scheduler *sch = task->parent;
2226 int ret;
2227 32773 int err = 0;
2228
2229 32773 ret = (intptr_t)task->func(task->func_arg);
2230
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 32772 times.
32773 if (ret < 0)
2231 1 av_log(task->func_arg, AV_LOG_ERROR,
2232 1 "Task finished with error code: %d (%s)\n", ret, av_err2str(ret));
2233
2234
5/6
✓ Branch 0 taken 6768 times.
✓ Branch 1 taken 6753 times.
✓ Branch 2 taken 6392 times.
✓ Branch 3 taken 6474 times.
✓ Branch 4 taken 6386 times.
✗ Branch 5 not taken.
32773 switch (task->node.type) {
2235 6768 case SCH_NODE_TYPE_DEMUX: err = demux_done (sch, task->node.idx); break;
2236 6753 case SCH_NODE_TYPE_MUX: err = mux_done (sch, task->node.idx); break;
2237 6392 case SCH_NODE_TYPE_DEC: err = dec_done (sch, task->node.idx); break;
2238 6474 case SCH_NODE_TYPE_ENC: err = enc_done (sch, task->node.idx); break;
2239 6386 case SCH_NODE_TYPE_FILTER_IN: err = filter_done(sch, task->node.idx); break;
2240 default: av_assert0(0);
2241 }
2242
2243 32773 ret = err_merge(ret, err);
2244
2245 // EOF is considered normal termination
2246
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 32773 times.
32773 if (ret == AVERROR_EOF)
2247 ret = 0;
2248
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 32772 times.
32773 if (ret < 0)
2249 1 atomic_store(&sch->task_failed, 1);
2250
2251
4/4
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 32772 times.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 32772 times.
32774 av_log(task->func_arg, ret < 0 ? AV_LOG_ERROR : AV_LOG_VERBOSE,
2252 "Terminating thread with return code %d (%s)\n", ret,
2253 1 ret < 0 ? av_err2str(ret) : "success");
2254
2255 32773 return (void*)(intptr_t)ret;
2256 }
2257