History

fftools/ffmpeg_sched: make sure to always run task cleanup

Even in cases where sch_start() failed. This ensures all links are properly closed and no tasks are left hanging. Fixes #10916.

Anton Khirnov committed 2 years ago

Commit 24b9f291 parent af817881 file · +44 −24Browse files

Unified · Split

fftools/ffmpeg_sched.c+44−24
@@ -260,6 +260,12 @@ typedef struct SchFilterGraph { · 259 unmodified lines
260260 int task_exited;
261261 } SchFilterGraph;
262262
263+enum SchedulerState {
264+ SCH_STATE_UNINIT,
265+ SCH_STATE_STARTED,
266+ SCH_STATE_STOPPED,
267+};
268+
263269 struct Scheduler {
264270 const AVClass *class;
265271
@@ -292,7 +298,7 @@ struct Scheduler { · 26 unmodified lines
292298 char *sdp_filename;
293299 int sdp_auto;
294300
295− int transcode_started;
301+ enum SchedulerState state;
296302 atomic_int terminate;
297303 atomic_int task_failed;
298304
@@ -1144,7 +1150,8 @@ int sch_mux_stream_ready(Scheduler *sch, unsigned mux_idx, unsigned stream_idx) · 845 unmodified lines
11441150
11451151 // this may be called during initialization - do not start
11461152 // threads before sch_start() is called
1147− if (++mux->nb_streams_ready == mux->nb_streams && sch->transcode_started)
1153+ if (++mux->nb_streams_ready == mux->nb_streams &&
1154+ sch->state >= SCH_STATE_STARTED)
11481155 ret = mux_init(sch, mux);
11491156
11501157 pthread_mutex_unlock(&sch->mux_ready_lock);
@@ -1514,7 +1521,8 @@ int sch_start(Scheduler *sch) · 363 unmodified lines
15141521 if (ret < 0)
15151522 return ret;
15161523
1517− sch->transcode_started = 1;
1524+ av_assert0(sch->state == SCH_STATE_UNINIT);
1525+ sch->state = SCH_STATE_STARTED;
15181526
15191527 for (unsigned i = 0; i < sch->nb_mux; i++) {
15201528 SchMux *mux = &sch->mux[i];
@@ -1522,7 +1530,7 @@ int sch_start(Scheduler *sch) · 1 unmodified line
15221530 if (mux->nb_streams_ready == mux->nb_streams) {
15231531 ret = mux_init(sch, mux);
15241532 if (ret < 0)
1525− return ret;
1533+ goto fail;
15261534 }
15271535 }
15281536
@@ -1531,7 +1539,7 @@ int sch_start(Scheduler *sch) · 2 unmodified lines
15311539
15321540 ret = task_start(&enc->task);
15331541 if (ret < 0)
1534− return ret;
1542+ goto fail;
15351543 }
15361544
15371545 for (unsigned i = 0; i < sch->nb_filters; i++) {
@@ -1539,7 +1547,7 @@ int sch_start(Scheduler *sch) · 1 unmodified line
15391547
15401548 ret = task_start(&fg->task);
15411549 if (ret < 0)
1542− return ret;
1550+ goto fail;
15431551 }
15441552
15451553 for (unsigned i = 0; i < sch->nb_dec; i++) {
@@ -1547,7 +1555,7 @@ int sch_start(Scheduler *sch) · 1 unmodified line
15471555
15481556 ret = task_start(&dec->task);
15491557 if (ret < 0)
1550− return ret;
1558+ goto fail;
15511559 }
15521560
15531561 for (unsigned i = 0; i < sch->nb_demux; i++) {
@@ -1558,7 +1566,7 @@ int sch_start(Scheduler *sch) · 4 unmodified lines
15581566
15591567 ret = task_start(&d->task);
15601568 if (ret < 0)
1561− return ret;
1569+ goto fail;
15621570 }
15631571
15641572 pthread_mutex_lock(&sch->schedule_lock);
@@ -1566,6 +1574,9 @@ int sch_start(Scheduler *sch) · 1 unmodified line
15661574 pthread_mutex_unlock(&sch->schedule_lock);
15671575
15681576 return 0;
1577+fail:
1578+ sch_stop(sch, NULL);
1579+ return ret;
15691580 }
15701581
15711582 int sch_wait(Scheduler *sch, uint64_t timeout_us, int64_t *transcode_ts)
@@ -2412,6 +2423,18 @@ int sch_filter_command(Scheduler *sch, unsigned fg_idx, AVFrame *frame) · 840 unmodified lines
24122423 return send_to_filter(sch, fg, fg->nb_inputs, frame);
24132424 }
24142425
2426+static int task_cleanup(Scheduler *sch, SchedulerNode node)
2427+{
2428+ switch (node.type) {
2429+ case SCH_NODE_TYPE_DEMUX: return demux_done (sch, node.idx);
2430+ case SCH_NODE_TYPE_MUX: return mux_done (sch, node.idx);
2431+ case SCH_NODE_TYPE_DEC: return dec_done (sch, node.idx);
2432+ case SCH_NODE_TYPE_ENC: return enc_done (sch, node.idx);
2433+ case SCH_NODE_TYPE_FILTER_IN: return filter_done(sch, node.idx);
2434+ default: av_assert0(0);
2435+ }
2436+}
2437+
24152438 static void *task_wrapper(void *arg)
24162439 {
24172440 SchTask *task = arg;
@@ -2424,15 +2447,7 @@ static void *task_wrapper(void *arg) · 6 unmodified lines
24242447 av_log(task->func_arg, AV_LOG_ERROR,
24252448 "Task finished with error code: %d (%s)\n", ret, av_err2str(ret));
24262449
2427− switch (task->node.type) {
2428− case SCH_NODE_TYPE_DEMUX: err = demux_done (sch, task->node.idx); break;
2429− case SCH_NODE_TYPE_MUX: err = mux_done (sch, task->node.idx); break;
2430− case SCH_NODE_TYPE_DEC: err = dec_done (sch, task->node.idx); break;
2431− case SCH_NODE_TYPE_ENC: err = enc_done (sch, task->node.idx); break;
2432− case SCH_NODE_TYPE_FILTER_IN: err = filter_done(sch, task->node.idx); break;
2433− default: av_assert0(0);
2434− }
2435−
2450+ err = task_cleanup(sch, task->node);
24362451 ret = err_merge(ret, err);
24372452
24382453 // EOF is considered normal termination
@@ -2448,13 +2463,13 @@ static void *task_wrapper(void *arg) · 9 unmodified lines
24482463 return (void*)(intptr_t)ret;
24492464 }
24502465
2451−static int task_stop(SchTask *task)
2466+static int task_stop(Scheduler *sch, SchTask *task)
24522467 {
24532468 int ret;
24542469 void *thread_ret;
24552470
24562471 if (!task->thread_running)
2457− return 0;
2472+ return task_cleanup(sch, task->node);
24582473
24592474 ret = pthread_join(task->thread, &thread_ret);
24602475 av_assert0(ret == 0);
@@ -2468,6 +2483,9 @@ int sch_stop(Scheduler *sch, int64_t *finish_ts) · 7 unmodified lines
24682483 {
24692484 int ret = 0, err;
24702485
2486+ if (sch->state != SCH_STATE_STARTED)
2487+ return 0;
2488+
24712489 atomic_store(&sch->terminate, 1);
24722490
24732491 for (unsigned type = 0; type < 2; type++)
@@ -2479,40 +2497,42 @@ int sch_stop(Scheduler *sch, int64_t *finish_ts) · 5 unmodified lines
24792497 for (unsigned i = 0; i < sch->nb_demux; i++) {
24802498 SchDemux *d = &sch->demux[i];
24812499
2482− err = task_stop(&d->task);
2500+ err = task_stop(sch, &d->task);
24832501 ret = err_merge(ret, err);
24842502 }
24852503
24862504 for (unsigned i = 0; i < sch->nb_dec; i++) {
24872505 SchDec *dec = &sch->dec[i];
24882506
2489− err = task_stop(&dec->task);
2507+ err = task_stop(sch, &dec->task);
24902508 ret = err_merge(ret, err);
24912509 }
24922510
24932511 for (unsigned i = 0; i < sch->nb_filters; i++) {
24942512 SchFilterGraph *fg = &sch->filters[i];
24952513
2496− err = task_stop(&fg->task);
2514+ err = task_stop(sch, &fg->task);
24972515 ret = err_merge(ret, err);
24982516 }
24992517
25002518 for (unsigned i = 0; i < sch->nb_enc; i++) {
25012519 SchEnc *enc = &sch->enc[i];
25022520
2503− err = task_stop(&enc->task);
2521+ err = task_stop(sch, &enc->task);
25042522 ret = err_merge(ret, err);
25052523 }
25062524
25072525 for (unsigned i = 0; i < sch->nb_mux; i++) {
25082526 SchMux *mux = &sch->mux[i];
25092527
2510− err = task_stop(&mux->task);
2528+ err = task_stop(sch, &mux->task);
25112529 ret = err_merge(ret, err);
25122530 }
25132531
25142532 if (finish_ts)
25152533 *finish_ts = trailing_dts(sch, 1);
25162534
2535+ sch->state = SCH_STATE_STOPPED;
2536+
25172537 return ret;
25182538 }

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t