From 24b9f29ff2e0b84ae1345f51cbf7240e079d7a2b Mon Sep 17 00:00:00 2001 From: Anton Khirnov Date: Wed, 27 Mar 2024 09:49:27 +0100 Subject: [PATCH] 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. --- fftools/ffmpeg_sched.c | 68 +++++++++++++++++++++++++++--------------- 1 file changed, 44 insertions(+), 24 deletions(-) diff --git a/fftools/ffmpeg_sched.c b/fftools/ffmpeg_sched.c index 422c665c96..f8485db30b 100644 --- a/fftools/ffmpeg_sched.c +++ b/fftools/ffmpeg_sched.c @@ -260,6 +260,12 @@ typedef struct SchFilterGraph { int task_exited; } SchFilterGraph; +enum SchedulerState { + SCH_STATE_UNINIT, + SCH_STATE_STARTED, + SCH_STATE_STOPPED, +}; + struct Scheduler { const AVClass *class; @@ -292,7 +298,7 @@ struct Scheduler { char *sdp_filename; int sdp_auto; - int transcode_started; + enum SchedulerState state; atomic_int terminate; atomic_int task_failed; @@ -1144,7 +1150,8 @@ int sch_mux_stream_ready(Scheduler *sch, unsigned mux_idx, unsigned stream_idx) // this may be called during initialization - do not start // threads before sch_start() is called - if (++mux->nb_streams_ready == mux->nb_streams && sch->transcode_started) + if (++mux->nb_streams_ready == mux->nb_streams && + sch->state >= SCH_STATE_STARTED) ret = mux_init(sch, mux); pthread_mutex_unlock(&sch->mux_ready_lock); @@ -1514,7 +1521,8 @@ int sch_start(Scheduler *sch) if (ret < 0) return ret; - sch->transcode_started = 1; + av_assert0(sch->state == SCH_STATE_UNINIT); + sch->state = SCH_STATE_STARTED; for (unsigned i = 0; i < sch->nb_mux; i++) { SchMux *mux = &sch->mux[i]; @@ -1522,7 +1530,7 @@ int sch_start(Scheduler *sch) if (mux->nb_streams_ready == mux->nb_streams) { ret = mux_init(sch, mux); if (ret < 0) - return ret; + goto fail; } } @@ -1531,7 +1539,7 @@ int sch_start(Scheduler *sch) ret = task_start(&enc->task); if (ret < 0) - return ret; + goto fail; } for (unsigned i = 0; i < sch->nb_filters; i++) { @@ -1539,7 +1547,7 @@ int sch_start(Scheduler *sch) ret = task_start(&fg->task); if (ret < 0) - return ret; + goto fail; } for (unsigned i = 0; i < sch->nb_dec; i++) { @@ -1547,7 +1555,7 @@ int sch_start(Scheduler *sch) ret = task_start(&dec->task); if (ret < 0) - return ret; + goto fail; } for (unsigned i = 0; i < sch->nb_demux; i++) { @@ -1558,7 +1566,7 @@ int sch_start(Scheduler *sch) ret = task_start(&d->task); if (ret < 0) - return ret; + goto fail; } pthread_mutex_lock(&sch->schedule_lock); @@ -1566,6 +1574,9 @@ int sch_start(Scheduler *sch) pthread_mutex_unlock(&sch->schedule_lock); return 0; +fail: + sch_stop(sch, NULL); + return ret; } 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) return send_to_filter(sch, fg, fg->nb_inputs, frame); } +static int task_cleanup(Scheduler *sch, SchedulerNode node) +{ + switch (node.type) { + case SCH_NODE_TYPE_DEMUX: return demux_done (sch, node.idx); + case SCH_NODE_TYPE_MUX: return mux_done (sch, node.idx); + case SCH_NODE_TYPE_DEC: return dec_done (sch, node.idx); + case SCH_NODE_TYPE_ENC: return enc_done (sch, node.idx); + case SCH_NODE_TYPE_FILTER_IN: return filter_done(sch, node.idx); + default: av_assert0(0); + } +} + static void *task_wrapper(void *arg) { SchTask *task = arg; @@ -2424,15 +2447,7 @@ static void *task_wrapper(void *arg) av_log(task->func_arg, AV_LOG_ERROR, "Task finished with error code: %d (%s)\n", ret, av_err2str(ret)); - switch (task->node.type) { - case SCH_NODE_TYPE_DEMUX: err = demux_done (sch, task->node.idx); break; - case SCH_NODE_TYPE_MUX: err = mux_done (sch, task->node.idx); break; - case SCH_NODE_TYPE_DEC: err = dec_done (sch, task->node.idx); break; - case SCH_NODE_TYPE_ENC: err = enc_done (sch, task->node.idx); break; - case SCH_NODE_TYPE_FILTER_IN: err = filter_done(sch, task->node.idx); break; - default: av_assert0(0); - } - + err = task_cleanup(sch, task->node); ret = err_merge(ret, err); // EOF is considered normal termination @@ -2448,13 +2463,13 @@ static void *task_wrapper(void *arg) return (void*)(intptr_t)ret; } -static int task_stop(SchTask *task) +static int task_stop(Scheduler *sch, SchTask *task) { int ret; void *thread_ret; if (!task->thread_running) - return 0; + return task_cleanup(sch, task->node); ret = pthread_join(task->thread, &thread_ret); av_assert0(ret == 0); @@ -2468,6 +2483,9 @@ int sch_stop(Scheduler *sch, int64_t *finish_ts) { int ret = 0, err; + if (sch->state != SCH_STATE_STARTED) + return 0; + atomic_store(&sch->terminate, 1); for (unsigned type = 0; type < 2; type++) @@ -2479,40 +2497,42 @@ int sch_stop(Scheduler *sch, int64_t *finish_ts) for (unsigned i = 0; i < sch->nb_demux; i++) { SchDemux *d = &sch->demux[i]; - err = task_stop(&d->task); + err = task_stop(sch, &d->task); ret = err_merge(ret, err); } for (unsigned i = 0; i < sch->nb_dec; i++) { SchDec *dec = &sch->dec[i]; - err = task_stop(&dec->task); + err = task_stop(sch, &dec->task); ret = err_merge(ret, err); } for (unsigned i = 0; i < sch->nb_filters; i++) { SchFilterGraph *fg = &sch->filters[i]; - err = task_stop(&fg->task); + err = task_stop(sch, &fg->task); ret = err_merge(ret, err); } for (unsigned i = 0; i < sch->nb_enc; i++) { SchEnc *enc = &sch->enc[i]; - err = task_stop(&enc->task); + err = task_stop(sch, &enc->task); ret = err_merge(ret, err); } for (unsigned i = 0; i < sch->nb_mux; i++) { SchMux *mux = &sch->mux[i]; - err = task_stop(&mux->task); + err = task_stop(sch, &mux->task); ret = err_merge(ret, err); } if (finish_ts) *finish_ts = trailing_dts(sch, 1); + sch->state = SCH_STATE_STOPPED; + return ret; }