sysutils/ffmpeg
1
0
Fork 0
mirror of https://git.ffmpeg.org/ffmpeg.git synced 2026-10-24 02:09:03 +03:00
Code Activity

Revert "fftools/ffmpeg_sched: allow throttling decoder outputs"

Browse Source 
This reverts commit 03dfac5630.
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 <ffmpeg@haasn.dev>
Signed-off-by: Andreas Rheinhardt <andreas.rheinhardt@outlook.com>
This commit is contained in:
Andreas Rheinhardt
2026-08-17 00:15:36 +02:00
parent 4ec050211a
commit 43284b61a3
1 changed files with 7 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
fftools/ffmpeg_sched.c
View File

@@ -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++) {