This is an automated email from the git hooks/post-receive script. Git pushed a commit to branch master in repository ffmpeg.
commit 98bbcf138d9519a1df7db5064a0fdc2acf5d054d Author: Niklas Haas <[email protected]> AuthorDate: Mon Aug 31 16:29:04 2026 +0200 Commit: Niklas Haas <[email protected]> CommitDate: Mon Sep 21 14:23:20 2026 +0000 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 1e7f280996..221d9dae0a 100644 --- a/doc/ffmpeg.texi +++ b/doc/ffmpeg.texi @@ -2832,8 +2832,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 f87f1fef2d..265bfba35f 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); -- 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]
