This is an automated email from the git hooks/post-receive script.
Git pushed a commit to branch master
in repository ffmpeg.
The following commit(s) were added to refs/heads/master by this push:
new 43284b61a3 Revert "fftools/ffmpeg_sched: allow throttling decoder
outputs"
43284b61a3 is described below
commit 43284b61a39a7d35633ebbc637db50ff4e97c57e
Author: Andreas Rheinhardt <[email protected]>
AuthorDate: Mon Aug 17 00:15:36 2026 +0200
Commit: Andreas Rheinhardt <[email protected]>
CommitDate: Mon Aug 17 18:43:51 2026 +0200
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]>
---
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++) {
--
To stop receiving notification emails like this one, please contact
[email protected].
_______________________________________________
ffmpeg-cvslog mailing list -- [email protected]
To unsubscribe send an email to [email protected]
