[PR] Revert "fftools/ffmpeg_sched: allow throttling decoder outputs" (PR #24194)

PR #24194 opened by mkver
URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/24194
Patch URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/24194.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).

(cherry picked from commit 43284b61a39a7d35633ebbc637db50ff4e97c57e)



>From 92e1f4211ee16bcab7bad2581b2d29342f24eeb9 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).

Reviewed-by: Niklas Haas <[email protected]>
Signed-off-by: Andreas Rheinhardt <[email protected]>
(cherry picked from commit 43284b61a39a7d35633ebbc637db50ff4e97c57e)
---
 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 8ec3bdaf80..dddd1d2c67 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;
 }
 
@@ -2838,9 +2785,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]