This is an automated email from the git hooks/post-receive script. Git pushed a commit to branch master in repository ffmpeg.
commit 9048d1b0ad8a9473ac968814c2e45a5812d24c29 Author: Niklas Haas <[email protected]> AuthorDate: Mon Aug 31 17:26:33 2026 +0200 Commit: Niklas Haas <[email protected]> CommitDate: Mon Sep 21 14:23:20 2026 +0000 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); } -- 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]
