From 242ee7b0a23fdd116869c0556001f67102d6b049 Mon Sep 17 00:00:00 2001
From: Anton Khirnov <anton@khirnov.net>
Date: Mon, 14 Oct 2024 13:12:42 +0200
Subject: [PATCH] fftools/ffmpeg_sched: be smarter about flushing the
 pre-muxing queues

These per-stream FIFOs hold the packets before every stream is
initialized and the header can be written. Once that happens, current
code will flush each stream's queue one after the other. However, in
case we buffered a lot of data for multiple streams, this may cause the
muxer to overflow max_interleave_delta, resulting in worse interleaving
than necessary.

Change the code to interleave packets from all the queues by DTS.
---
 fftools/ffmpeg_sched.c | 40 +++++++++++++++++++++++++++++++++++-----
 1 file changed, 35 insertions(+), 5 deletions(-)

diff --git a/fftools/ffmpeg_sched.c b/fftools/ffmpeg_sched.c
index ef0b6e2897..b8ed8ae3f8 100644
--- a/fftools/ffmpeg_sched.c
+++ b/fftools/ffmpeg_sched.c
@@ -1111,22 +1111,52 @@ static int mux_task_start(SchMux *mux)
         return ret;
 
     /* flush the pre-muxing queues */
-    for (unsigned i = 0; i < mux->nb_streams; i++) {
-        SchMuxStream *ms = &mux->streams[i];
+    while (1) {
+        int       min_stream = -1;
+        Timestamp min_ts     = { .ts = AV_NOPTS_VALUE };
+
         AVPacket *pkt;
 
-        while (av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1) >= 0) {
+        // find the stream with the earliest dts or EOF in pre-muxing queue
+        for (unsigned i = 0; i < mux->nb_streams; i++) {
+            SchMuxStream *ms = &mux->streams[i];
+
+            if (av_fifo_peek(ms->pre_mux_queue.fifo, &pkt, 1, 0) < 0)
+                continue;
+
+            if (!pkt || pkt->dts == AV_NOPTS_VALUE) {
+                min_stream = i;
+                break;
+            }
+
+            if (min_ts.ts == AV_NOPTS_VALUE ||
+                av_compare_ts(min_ts.ts, min_ts.tb, pkt->dts, pkt->time_base) > 0) {
+                min_stream = i;
+                min_ts     = (Timestamp){ .ts = pkt->dts, .tb = pkt->time_base };
+            }
+        }
+
+        if (min_stream >= 0) {
+            SchMuxStream *ms = &mux->streams[min_stream];
+
+            ret = av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1);
+            av_assert0(ret >= 0);
+
             if (pkt) {
                 if (!ms->init_eof)
-                    ret = tq_send(mux->queue, i, pkt);
+                    ret = tq_send(mux->queue, min_stream, pkt);
                 av_packet_free(&pkt);
                 if (ret == AVERROR_EOF)
                     ms->init_eof = 1;
                 else if (ret < 0)
                     return ret;
             } else
-                tq_send_finish(mux->queue, i);
+                tq_send_finish(mux->queue, min_stream);
+
+            continue;
         }
+
+        break;
     }
 
     atomic_store(&mux->mux_started, 1);