PR #24174 opened by mkver URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/24174 Patch URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/24174.patch
This reverts commit 03dfac563018e6e8b81e331ebae0732d8edfe754. Said commit led to numerous regressions, namely: a) It can lead to wrong output interleavement (issue #23731 and (presumably) #24162). b) It can lead to streams stopping prematurely (issue #24008 and (presumably) #23988 and #24135). c) It can lead to extreme memory usage even when the input files are fine, especially mkvmerge files (see issue #24168). >From 7a749574a3c409f2df1d3cc0ab278a4176ed7a20 Mon Sep 17 00:00:00 2001 From: Andreas Rheinhardt <[email protected]> Date: Mon, 17 Aug 2026 00:15:36 +0200 Subject: [PATCH] Revert "fftools/ffmpeg_sched: allow throttling decoder outputs" This reverts commit 03dfac563018e6e8b81e331ebae0732d8edfe754. Said commit led to numerous regressions, namely: a) It can lead to wrong output interleavement (issue #23731 and (presumably) #24162). b) It can lead to streams stopping prematurely (issue #24008 and (presumably) #23988 and #24135). c) It can lead to extreme memory usage even when the input files are fine, especially mkvmerge files (see issue #24168). Signed-off-by: Andreas Rheinhardt <[email protected]> --- fftools/ffmpeg_sched.c | 70 +++++------------------------------------- 1 file changed, 7 insertions(+), 63 deletions(-) diff --git a/fftools/ffmpeg_sched.c b/fftools/ffmpeg_sched.c index 88b3bf3177..e9b57ed62b 100644 --- a/fftools/ffmpeg_sched.c +++ b/fftools/ffmpeg_sched.c @@ -32,7 +32,6 @@ #include "libavcodec/packet.h" #include "libavutil/avassert.h" -#include "libavutil/container_fifo.h" #include "libavutil/error.h" #include "libavutil/fifo.h" #include "libavutil/frame.h" @@ -87,7 +86,6 @@ typedef struct SchDec { unsigned nb_outputs; SchTask task; - SchWaiter waiter; // Queue for receiving input packets, one stream. ThreadQueue *queue; @@ -97,9 +95,6 @@ typedef struct SchDec { // temporary storage used by sch_dec_send() AVFrame *send_frame; - - // internal queue of undecoded packets used by sch_dec_receive() - AVContainerFifo *overflow; } SchDec; typedef struct SchSyncQueue { @@ -553,7 +548,6 @@ void sch_free(Scheduler **psch) tq_free(&dec->queue); av_thread_message_queue_free(&dec->queue_end_ts); - av_container_fifo_free(&dec->overflow); for (unsigned j = 0; j < dec->nb_outputs; j++) { SchDecOutput *o = &dec->outputs[j]; @@ -565,8 +559,6 @@ void sch_free(Scheduler **psch) av_freep(&dec->outputs); av_frame_free(&dec->send_frame); - - waiter_uninit(&dec->waiter); } av_freep(&sch->dec); @@ -817,14 +809,6 @@ int sch_add_dec(Scheduler *sch, SchThreadFunc func, void *ctx, int send_end_ts) return ret; } - dec->overflow = av_container_fifo_alloc_avpacket(0); - if (!dec->overflow) - return AVERROR(ENOMEM); - - ret = waiter_init(&dec->waiter); - if (ret < 0) - return ret; - return idx; } @@ -1321,9 +1305,8 @@ int sch_mux_sub_heartbeat_add(Scheduler *sch, unsigned mux_idx, unsigned stream_ enum { UNCHOKE_DEMUX = (1 << 0), UNCHOKE_FILTER = (1 << 1), - UNCHOKE_DECODE = (1 << 2), - UNCHOKE_ALL = UNCHOKE_DEMUX | UNCHOKE_FILTER | UNCHOKE_DECODE, + UNCHOKE_ALL = UNCHOKE_DEMUX | UNCHOKE_FILTER, }; static void unchoke_for_stream(Scheduler *sch, SchedulerNode src, int flags); @@ -1338,10 +1321,8 @@ static void unchoke_downstream(Scheduler *sch, SchedulerNode *dst) switch (dst->type) { case SCH_NODE_TYPE_DEC: dec = &sch->dec[dst->idx]; - if (!dec->waiter.choked_next) { - for (int i = 0; i < dec->nb_outputs; i++) - unchoke_downstream(sch, dec->outputs[i].dst); - } + for (int i = 0; i < dec->nb_outputs; i++) + unchoke_downstream(sch, dec->outputs[i].dst); break; case SCH_NODE_TYPE_ENC: enc = &sch->enc[dst->idx]; @@ -1372,7 +1353,6 @@ static void unchoke_for_stream(Scheduler *sch, SchedulerNode src, int flags) while (1) { SchFilterGraph *fg; SchDemux *demux; - SchDec *dec; switch (src.type) { case SCH_NODE_TYPE_DEMUX: // fed directly by a demuxer (i.e. not through a filtergraph) @@ -1386,11 +1366,7 @@ static void unchoke_for_stream(Scheduler *sch, SchedulerNode src, int flags) } return; case SCH_NODE_TYPE_DEC: - dec = &sch->dec[src.idx]; - if (!(flags & UNCHOKE_DECODE)) - return; - dec->waiter.choked_next = 0; - src = dec->src; + src = sch->dec[src.idx].src; continue; case SCH_NODE_TYPE_ENC: src = sch->enc[src.idx].src; @@ -1469,7 +1445,6 @@ static void schedule_update_locked(Scheduler *sch) RESET_WAITER(demux); RESET_WAITER(filters); - RESET_WAITER(dec); // figure out the sources that are allowed to proceed for (unsigned i = 0; i < sch->nb_mux; i++) { @@ -1480,11 +1455,8 @@ static void schedule_update_locked(Scheduler *sch) // unblock sources for output streams that are not finished // and not too far ahead of the trailing stream - if (ms->source_finished) { - // still allow decoders to drain - unchoke_for_stream(sch, ms->src, UNCHOKE_DECODE); + if (ms->source_finished) continue; - } if (dts == AV_NOPTS_VALUE && ms->last_dts != AV_NOPTS_VALUE) continue; if (dts != AV_NOPTS_VALUE && ms->last_dts - dts >= SCHEDULE_TOLERANCE) @@ -1538,7 +1510,6 @@ static void schedule_update_locked(Scheduler *sch) UPDATE_WAITER(demux); UPDATE_WAITER(filters); - UPDATE_WAITER(dec); } enum { @@ -2341,12 +2312,6 @@ int sch_dec_receive(Scheduler *sch, unsigned dec_idx, AVPacket *pkt) av_assert0(dec_idx < sch->nb_dec); dec = &sch->dec[dec_idx]; -retry: - // Pull a packet from the overflow FIFO while unchoked or expecting EOF ts - if (av_container_fifo_can_read(dec->overflow) && - (!atomic_load(&dec->waiter.choked) || dec->expect_end_ts)) - return av_container_fifo_read(dec->overflow, pkt, 0); - // the decoder should have given us post-flush end timestamp in pkt if (dec->expect_end_ts) { Timestamp ts = (Timestamp){ .ts = pkt->pts, .tb = pkt->time_base }; @@ -2360,29 +2325,11 @@ retry: ret = tq_receive(dec->queue, &dummy, pkt, 0); av_assert0(dummy <= 0); - // drain packets from overflow queue before returning EOF - if (ret == AVERROR_EOF && av_container_fifo_can_read(dec->overflow)) { - int terminate = waiter_wait(sch, &dec->waiter); - if (terminate) - return ret; - return av_container_fifo_read(dec->overflow, pkt, 0); - } else if (ret < 0) - return ret; - // got a flush packet, on the next call to this function the decoder - // should give us post-flush end timestamp (after draining overflow fifo) - if (!pkt->data && !pkt->side_data_elems && dec->queue_end_ts) + // will give us post-flush end timestamp + if (ret >= 0 && !pkt->data && !pkt->side_data_elems && dec->queue_end_ts) dec->expect_end_ts = 1; - // we got a packet, but we're currently choked or have existing overflow - // packets; so push it to the FIFO first - if (atomic_load(&dec->waiter.choked) || av_container_fifo_can_read(dec->overflow)) { - ret = av_container_fifo_write(dec->overflow, pkt, 0); - if (ret < 0) - return ret; - goto retry; - } - return ret; } @@ -2841,9 +2788,6 @@ int sch_stop(Scheduler *sch, int64_t *finish_ts) choke_demux(sch, i, 0); // unfreeze to allow draining } - for (unsigned i = 0; i < sch->nb_dec; i++) - waiter_set(&sch->dec[i].waiter, 0); // unfreeze to allow draining - pthread_mutex_unlock(&sch->schedule_lock); for (unsigned i = 0; i < sch->nb_demux; i++) { -- 2.52.0 _______________________________________________ ffmpeg-devel mailing list -- [email protected] To unsubscribe send an email to [email protected]
