FFmpeg coverage

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