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

Hello

I am sorry to reply to you, but I don't know how to unsubscribe from this
mailing list. I have tried emailing the various words/codes to various
email addresses specified on the Ffmpeg website over several months and
still seem to be getting these emails. I signed up as I had a question and
wanted to potentially hire a developer a year ago. All sorted, and I never
heard back from anyone, as far as I could tell.

If you have any advice it would be appreciated. I keep thinking about the
co2 emissions per email I am getting!

Thank you very much,
Braith




On Mon, Aug 17, 2026 at 9:53 AM mkver via ffmpeg-devel <
[email protected]> wrote:

> 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]
>
_______________________________________________
ffmpeg-devel mailing list -- [email protected]
To unsubscribe send an email to [email protected]