PR #24337 opened by Niklas Haas (haasn) URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/24337 Patch URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/24337.patch
ThreadQueue as written was very poorly optimized for the case of multiple streams. Not only was it possible for one stream to just completely fill the FIFO, depriving other streams a chance to write anything at all, but it also had two other inefficiencies: 1. The writers were only woken up when the FIFO was completely empty, leading to a ping/pong behavior; possibly intended as a mitigation for.. 2. Any wake on the single shared condition variable immediately woke up every single writer, leading to most of them just going right back to sleep (thundering herd problem) This series addresses all three, and then dramatically reduces the number of buffered packets - which due to the lack of inter-thread scheduling on filter graph inputs actually translated directly to a number of buffered *frames*. Verified empirically that this allows much lower settings of -filter_buffured_frames (for multi-input, shared demuxer scenarios). >From f94ba2aa48d4a9f37f7e699647cc4ae3e866a064 Mon Sep 17 00:00:00 2001 From: Niklas Haas <[email protected]> Date: Mon, 31 Aug 2026 16:21:57 +0200 Subject: [PATCH 1/5] fftools/ffmpeg_opt: remove stale -thread_queue_size input option The rewrite in d119ae2fd82a494d9430ff4d4fc262961a68c598 silently turned this option into a no-op on the input side. The documentation wording is also still very much out of date, as FFmpeg is now always threaded. We could hook this up to the demuxer queue size again, but given that the caveats no longer apply, and this option is a bit pointless outside of debugging, I have opted to just remove the corresponding section and mark it as output-only. Sponsored-by: nxtedition AB Signed-off-by: Niklas Haas <[email protected]> --- doc/ffmpeg.texi | 12 +++--------- fftools/ffmpeg_opt.c | 4 ++-- 2 files changed, 5 insertions(+), 11 deletions(-) diff --git a/doc/ffmpeg.texi b/doc/ffmpeg.texi index 1cbb1ccbdd..2cb26a3e8f 100644 --- a/doc/ffmpeg.texi +++ b/doc/ffmpeg.texi @@ -2789,15 +2789,9 @@ to the @option{-ss} option is considered an actual timestamp, and is not offset by the start time of the file. This matters only for files which do not start from timestamp 0, such as transport streams. -@item -thread_queue_size @var{size} (@emph{input/output}) -For input, this option sets the maximum number of queued packets when reading -from the file or device. With low latency / high rate live streams, packets may -be discarded if they are not read in a timely manner; setting this value can -force ffmpeg to use a separate input thread and read packets as soon as they -arrive. By default ffmpeg only does this if multiple inputs are specified. - -For output, this option specified the maximum number of packets that may be -queued to each muxing thread. +@item -thread_queue_size @var{size} (@emph{output}) +This option specifies the maximum number of packets that may be queued to each +muxing thread. @item -sdp_file @var{file} (@emph{global}) Print sdp information for an output stream to @var{file}. diff --git a/fftools/ffmpeg_opt.c b/fftools/ffmpeg_opt.c index fdbc0fdc13..f93be37558 100644 --- a/fftools/ffmpeg_opt.c +++ b/fftools/ffmpeg_opt.c @@ -1802,9 +1802,9 @@ const OptionDef options[] = { { "disposition", OPT_TYPE_STRING, OPT_PERSTREAM | OPT_OUTPUT | OPT_EXPERT, { .off = OFFSET(disposition) }, "disposition", "" }, - { "thread_queue_size", OPT_TYPE_INT, OPT_OFFSET | OPT_EXPERT | OPT_INPUT | OPT_OUTPUT, + { "thread_queue_size", OPT_TYPE_INT, OPT_OFFSET | OPT_EXPERT | OPT_OUTPUT, { .off = OFFSET(thread_queue_size) }, - "set the maximum number of queued packets from the demuxer" }, + "set the maximum number of queued packets on the muxer" }, { "find_stream_info", OPT_TYPE_BOOL, OPT_INPUT | OPT_EXPERT | OPT_OFFSET, { .off = OFFSET(find_stream_info) }, "read and decode the streams to fill missing information with heuristics" }, -- 2.52.0 >From 57d98ef0964e7231edc611efecd59c0786aebe74 Mon Sep 17 00:00:00 2001 From: Niklas Haas <[email protected]> Date: Mon, 31 Aug 2026 16:29:04 +0200 Subject: [PATCH 2/5] fftools/thread_queue: apply queue size per stream This preserves FIFO properties, but allows a queue with multiple streams (e.g. a filter graph with many inputs) to have each stream contend for only its own space, rather than fighting with possibly arbitrarily many other streams for space inside a single 2-element frame queue. This does slightly change the semantics of the -thread_queue_size option, but this is already marked as an expert/debug option and was partially broken for several years, so I consider the risk of downstream annoyance to be rather minimal. (Especially as the new behavior results in a larger queue, which only wastes memory, rather than a smaller queue, which may starve) Sponsored-by: nxtedition AB Signed-off-by: Niklas Haas <[email protected]> --- doc/ffmpeg.texi | 4 ++-- fftools/ffmpeg_opt.c | 2 +- fftools/thread_queue.c | 24 +++++++++++++++++++++--- fftools/thread_queue.h | 2 +- 4 files changed, 25 insertions(+), 7 deletions(-) diff --git a/doc/ffmpeg.texi b/doc/ffmpeg.texi index 2cb26a3e8f..593f4a9968 100644 --- a/doc/ffmpeg.texi +++ b/doc/ffmpeg.texi @@ -2790,8 +2790,8 @@ offset by the start time of the file. This matters only for files which do not start from timestamp 0, such as transport streams. @item -thread_queue_size @var{size} (@emph{output}) -This option specifies the maximum number of packets that may be queued to each -muxing thread. +This option specifies the maximum number of packets per stream that may be +queued to each muxing thread. @item -sdp_file @var{file} (@emph{global}) Print sdp information for an output stream to @var{file}. diff --git a/fftools/ffmpeg_opt.c b/fftools/ffmpeg_opt.c index f93be37558..1611784ff4 100644 --- a/fftools/ffmpeg_opt.c +++ b/fftools/ffmpeg_opt.c @@ -1804,7 +1804,7 @@ const OptionDef options[] = { "disposition", "" }, { "thread_queue_size", OPT_TYPE_INT, OPT_OFFSET | OPT_EXPERT | OPT_OUTPUT, { .off = OFFSET(thread_queue_size) }, - "set the maximum number of queued packets on the muxer" }, + "set the maximum number of queued packets per stream on the muxer" }, { "find_stream_info", OPT_TYPE_BOOL, OPT_INPUT | OPT_EXPERT | OPT_OFFSET, { .off = OFFSET(find_stream_info) }, "read and decode the streams to fill missing information with heuristics" }, diff --git a/fftools/thread_queue.c b/fftools/thread_queue.c index 9fa8a46cac..64f06ebfb0 100644 --- a/fftools/thread_queue.c +++ b/fftools/thread_queue.c @@ -41,11 +41,13 @@ struct ThreadQueue { int choked; int *finished; unsigned int nb_streams; + size_t queue_size; enum ThreadQueueType type; AVContainerFifo *fifo; AVFifo *fifo_stream_index; + size_t *stream_count; pthread_mutex_t lock; pthread_cond_t cond; @@ -60,6 +62,7 @@ void tq_free(ThreadQueue **ptq) av_container_fifo_free(&tq->fifo); av_fifo_freep2(&tq->fifo_stream_index); + av_freep(&tq->stream_count); av_freep(&tq->finished); @@ -96,7 +99,7 @@ ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, if (!tq->finished) goto fail; tq->nb_streams = nb_streams; - + tq->queue_size = queue_size; tq->type = type; tq->fifo = (type == THREAD_QUEUE_FRAMES) ? @@ -104,16 +107,29 @@ ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, if (!tq->fifo) goto fail; - tq->fifo_stream_index = av_fifo_alloc2(queue_size, sizeof(unsigned), 0); + av_assert0(queue_size); + if (nb_streams > SIZE_MAX / queue_size) + goto fail; // treat like OOM + + tq->fifo_stream_index = av_fifo_alloc2(queue_size * nb_streams, sizeof(unsigned), 0); if (!tq->fifo_stream_index) goto fail; + tq->stream_count = av_calloc(nb_streams, sizeof(*tq->stream_count)); + if (!tq->stream_count) + goto fail; + return tq; fail: tq_free(&tq); return NULL; } +static int can_write(ThreadQueue *tq, unsigned int stream_idx) +{ + return tq->stream_count[stream_idx] < tq->queue_size; +} + int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) { int *finished; @@ -129,7 +145,7 @@ int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) goto finish; } - while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo_stream_index)) + while (!(*finished & FINISHED_RECV) && !can_write(tq, stream_idx)) pthread_cond_wait(&tq->cond, &tq->lock); if (*finished & FINISHED_RECV) { @@ -144,6 +160,7 @@ int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) if (ret < 0) goto finish; + tq->stream_count[stream_idx]++; pthread_cond_broadcast(&tq->cond); } @@ -167,6 +184,7 @@ static int receive_locked(ThreadQueue *tq, int *stream_idx, ret = av_fifo_read(tq->fifo_stream_index, &idx, 1); av_assert0(ret >= 0); + tq->stream_count[idx]--; if (tq->finished[idx] & FINISHED_RECV) { (tq->type == THREAD_QUEUE_FRAMES) ? av_frame_unref(data) : av_packet_unref(data); diff --git a/fftools/thread_queue.h b/fftools/thread_queue.h index 756545df44..ba54af0947 100644 --- a/fftools/thread_queue.h +++ b/fftools/thread_queue.h @@ -40,7 +40,7 @@ typedef struct ThreadQueue ThreadQueue; * @param nb_streams number of streams for which a distinct EOF state is * maintained * @param queue_size number of items that can be stored in the queue without - * blocking + * blocking (per stream) */ ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, enum ThreadQueueType type); -- 2.52.0 >From 1f789e86cadb7d32502f2cc099bfd7aec6c064e4 Mon Sep 17 00:00:00 2001 From: Niklas Haas <[email protected]> Date: Mon, 31 Aug 2026 17:26:33 +0200 Subject: [PATCH 3/5] fftools/thread_queue: split apart cond var per index Instead of multiple producers and consumers all waiting on a single cond var, resulting in a thundering herd problem, use a separate cond var for signalling the "queue full" and "queue empty" conditions. Since the previous commit splits up the universal limit into sublimits per stream, we can actually go a step further and only wake up the writer whose stream we just read from. In practice, this will result in a nice round-robin behavior, without spending countless context switches waking up idle decoders who will immediately go back to sleep. Results in 67x less sys time and 11x fewer context switches in a synthetic test where 16 streams are feeding into a single filter graph. Sponsored-by: nxtedition AB Signed-off-by: Niklas Haas <[email protected]> --- fftools/thread_queue.c | 47 +++++++++++++++++++++++++++--------------- 1 file changed, 30 insertions(+), 17 deletions(-) diff --git a/fftools/thread_queue.c b/fftools/thread_queue.c index 64f06ebfb0..ed70ce582a 100644 --- a/fftools/thread_queue.c +++ b/fftools/thread_queue.c @@ -50,7 +50,9 @@ struct ThreadQueue { size_t *stream_count; pthread_mutex_t lock; - pthread_cond_t cond; + pthread_cond_t cond_read; + pthread_cond_t *cond_write; + unsigned int nb_cond_write; }; void tq_free(ThreadQueue **ptq) @@ -66,7 +68,11 @@ void tq_free(ThreadQueue **ptq) av_freep(&tq->finished); - pthread_cond_destroy(&tq->cond); + for (unsigned i = 0; i < tq->nb_cond_write; i++) + pthread_cond_destroy(&tq->cond_write[i]); + av_freep(&tq->cond_write); + + pthread_cond_destroy(&tq->cond_read); pthread_mutex_destroy(&tq->lock); av_freep(ptq); @@ -82,7 +88,7 @@ ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, if (!tq) return NULL; - ret = pthread_cond_init(&tq->cond, NULL); + ret = pthread_cond_init(&tq->cond_read, NULL); if (ret) { av_freep(&tq); return NULL; @@ -90,11 +96,20 @@ ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, ret = pthread_mutex_init(&tq->lock, NULL); if (ret) { - pthread_cond_destroy(&tq->cond); + pthread_cond_destroy(&tq->cond_read); av_freep(&tq); return NULL; } + tq->cond_write = av_calloc(nb_streams, sizeof(*tq->cond_write)); + if (!tq->cond_write) + goto fail; + for (tq->nb_cond_write = 0; tq->nb_cond_write < nb_streams; tq->nb_cond_write++) { + ret = pthread_cond_init(&tq->cond_write[tq->nb_cond_write], NULL); + if (ret) + goto fail; + } + tq->finished = av_calloc(nb_streams, sizeof(*tq->finished)); if (!tq->finished) goto fail; @@ -146,7 +161,7 @@ int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) } while (!(*finished & FINISHED_RECV) && !can_write(tq, stream_idx)) - pthread_cond_wait(&tq->cond, &tq->lock); + pthread_cond_wait(&tq->cond_write[stream_idx], &tq->lock); if (*finished & FINISHED_RECV) { ret = AVERROR_EOF; @@ -161,7 +176,7 @@ int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) goto finish; tq->stream_count[stream_idx]++; - pthread_cond_broadcast(&tq->cond); + pthread_cond_broadcast(&tq->cond_read); // signal downstream } finish: @@ -184,7 +199,11 @@ static int receive_locked(ThreadQueue *tq, int *stream_idx, ret = av_fifo_read(tq->fifo_stream_index, &idx, 1); av_assert0(ret >= 0); - tq->stream_count[idx]--; + + // signal upstream if the fifo is empty + if (--tq->stream_count[idx] == 0) + pthread_cond_broadcast(&tq->cond_write[idx]); + if (tq->finished[idx] & FINISHED_RECV) { (tq->type == THREAD_QUEUE_FRAMES) ? av_frame_unref(data) : av_packet_unref(data); @@ -221,16 +240,10 @@ int tq_receive(ThreadQueue *tq, int *stream_idx, void *data, int flags) pthread_mutex_lock(&tq->lock); while (1) { - size_t can_read = av_container_fifo_can_read(tq->fifo); - ret = receive_locked(tq, stream_idx, data); - // signal other threads if the fifo state changed - if (can_read != av_container_fifo_can_read(tq->fifo)) - pthread_cond_broadcast(&tq->cond); - if (ret == AVERROR(EAGAIN) && !(flags & THREAD_QUEUE_FLAG_NO_BLOCK)) { - pthread_cond_wait(&tq->cond, &tq->lock); + pthread_cond_wait(&tq->cond_read, &tq->lock); continue; } @@ -253,7 +266,7 @@ void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx) * an EOF and recv-finished flag will be set */ tq->finished[stream_idx] |= FINISHED_SEND; tq->choked = 0; - pthread_cond_broadcast(&tq->cond); + pthread_cond_broadcast(&tq->cond_read); pthread_mutex_unlock(&tq->lock); } @@ -268,7 +281,7 @@ void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx) * next time the producer thread tries to send for this stream, it will * get an EOF and send-finished flag will be set */ tq->finished[stream_idx] |= FINISHED_RECV; - pthread_cond_broadcast(&tq->cond); + pthread_cond_broadcast(&tq->cond_write[stream_idx]); pthread_mutex_unlock(&tq->lock); } @@ -280,7 +293,7 @@ void tq_choke(ThreadQueue *tq, int choked) int prev_choked = tq->choked; tq->choked = choked; if (choked != prev_choked) - pthread_cond_broadcast(&tq->cond); + pthread_cond_broadcast(&tq->cond_read); pthread_mutex_unlock(&tq->lock); } -- 2.52.0 >From 926fe29776561314853c45e426fc94932e8cc6a0 Mon Sep 17 00:00:00 2001 From: Niklas Haas <[email protected]> Date: Mon, 31 Aug 2026 17:41:04 +0200 Subject: [PATCH 4/5] fftools/thread_queue: unblock writers as soon as FIFO is no longer full With the thundering herd problem resolved, there's no more reason to wait until the FIFO is empty before unblocking the writers. In practice, this means that in a downstream-throttled scenario, we will always wake up exactly the decoder that we just read from. Sponsored-by: nxtedition AB Signed-off-by: Niklas Haas <[email protected]> --- fftools/thread_queue.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/fftools/thread_queue.c b/fftools/thread_queue.c index ed70ce582a..af112810c6 100644 --- a/fftools/thread_queue.c +++ b/fftools/thread_queue.c @@ -200,8 +200,8 @@ static int receive_locked(ThreadQueue *tq, int *stream_idx, ret = av_fifo_read(tq->fifo_stream_index, &idx, 1); av_assert0(ret >= 0); - // signal upstream if the fifo is empty - if (--tq->stream_count[idx] == 0) + // signal upstream if the fifo is no longer full + if (tq->stream_count[idx]-- == tq->queue_size) pthread_cond_broadcast(&tq->cond_write[idx]); if (tq->finished[idx] & FINISHED_RECV) { -- 2.52.0 >From 71010824db7ccaa87b6678b4f7e043317f7ceeea Mon Sep 17 00:00:00 2001 From: Niklas Haas <[email protected]> Date: Mon, 31 Aug 2026 18:19:37 +0200 Subject: [PATCH 5/5] fftools/ffmpeg_sched: drop packet thread queue size The previous series of commits makes this limit apply per-stream, instead of per-packet, so there's no real reason to have it set to such a high limit anymore. Buffering 8 packets is really the same as buffering 8 frames, if the decoders are faster than the encoder (usually the case) and immediately drain it into a shared filter graph while that filter graph is stuck waiting for a different input (from the same demuxer). I also doubt the methodology of e0da916b8 still holds up; a lot of moving parts have changed since then. I was not able to reproduce any speed difference at all acress any number of packets buffered; even reducing this to 1 gave me no measurable downside. The only scenario where I can measure a difference is if the fifo size is smaller than the number of streams, but that is now impossible. I'm tempted to just lower this all the way to 1, but I kept 2 for the time being as a hedge against any pathological edge cases. In the worst case, users can still use -thread_queue_size to increase this limit for testing / performance tuning. Sponsored-by: nxtedition AB Signed-off-by: Niklas Haas <[email protected]> --- fftools/ffmpeg_sched.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fftools/ffmpeg_sched.h b/fftools/ffmpeg_sched.h index 2cf3034437..952e56e7b5 100644 --- a/fftools/ffmpeg_sched.h +++ b/fftools/ffmpeg_sched.h @@ -254,7 +254,7 @@ int sch_add_mux(Scheduler *sch, SchThreadFunc func, int (*init)(void *), * Default size of a packet thread queue. For muxing this can be overridden by * the thread_queue_size option as passed to a call to sch_add_mux(). */ -#define DEFAULT_PACKET_THREAD_QUEUE_SIZE 8 +#define DEFAULT_PACKET_THREAD_QUEUE_SIZE 2 /** * Default size of a frame thread queue. -- 2.52.0 _______________________________________________ ffmpeg-devel mailing list -- [email protected] To unsubscribe send an email to [email protected]
