fftools/ffmpeg: rework -shortest implementation

The -shortest option (which finishes the output file at the time the
shortest stream ends) is currently implemented by faking the -t option
when an output stream ends. This approach is fragile, since it depends
on the frames/packets being processed in a specific order. E.g. there
are currently some situations in which the output file length will
depend unpredictably on unrelated factors like encoder delay. More
importantly, the present work aiming at splitting various ffmpeg
components into different threads will make this approach completely
unworkable, since the frames/packets will arrive in effectively random
order.

This commit introduces a "sync queue", which is essentially a collection
of FIFOs, one per stream. Frames/packets are submitted to these FIFOs
and are then released for further processing (encoding or muxing) when
it is ensured that the frame in question will not cause its stream to
get ahead of the other streams (the logic is similar to libavformat's
interleaving queue).

These sync queues are then used for encoding and/or muxing when the
-shortest option is specified.

A new option – -shortest_buf_duration – controls the maximum number of
queued packets, to avoid runaway memory usage.

This commit changes the results of the following tests:
- copy-shortest[12]: the last audio frame is now gone. This is
  correct, since it actually outlasts the last video frame.
- shortest-sub: the video packets following the last subtitle packet are
  now gone. This is also correct.

1 files changed, 61 insertions, 6 deletions

diff --git a/fftools/ffmpeg_mux.c b/fftools/ffmpeg_mux.c
index a3350a73e9..453ccac912 100644
--- a/fftools/ffmpeg_mux.c
+++ b/fftools/ffmpeg_mux.c
@@ -20,6 +20,7 @@
 #include <string.h>
 
 #include "ffmpeg.h"
+#include "sync_queue.h"
 
 #include "libavutil/fifo.h"
 #include "libavutil/intreadwrite.h"
@@ -56,6 +57,8 @@ struct Muxer {
     int64_t limit_filesize;
     int64_t final_filesize;
     int header_written;
+
+    AVPacket *sq_pkt;
 };
 
 static int want_sdp = 1;
@@ -72,13 +75,14 @@ static void close_all_output_streams(OutputStream *ost, OSTFinished this_stream,
 static int queue_packet(OutputFile *of, OutputStream *ost, AVPacket *pkt)
 {
     MuxStream *ms = &of->mux->streams[ost->index];
-    AVPacket *tmp_pkt;
+    AVPacket *tmp_pkt = NULL;
     int ret;
 
     if (!av_fifo_can_write(ms->muxing_queue)) {
         size_t cur_size = av_fifo_can_read(ms->muxing_queue);
+        size_t pkt_size = pkt ? pkt->size : 0;
         unsigned int are_we_over_size =
-            (ms->muxing_queue_data_size + pkt->size) > ost->muxing_queue_data_threshold;
+            (ms->muxing_queue_data_size + pkt_size) > ost->muxing_queue_data_threshold;
         size_t limit    = are_we_over_size ? ost->max_muxing_queue_size : SIZE_MAX;
         size_t new_size = FFMIN(2 * cur_size, limit);
 
@@ -93,6 +97,7 @@ static int queue_packet(OutputFile *of, OutputStream *ost, AVPacket *pkt)
             return ret;
     }
 
+    if (pkt) {
     ret = av_packet_make_refcounted(pkt);
     if (ret < 0)
         return ret;
@@ -103,6 +108,7 @@ static int queue_packet(OutputFile *of, OutputStream *ost, AVPacket *pkt)
 
     av_packet_move_ref(tmp_pkt, pkt);
     ms->muxing_queue_data_size += tmp_pkt->size;
+    }
     av_fifo_write(ms->muxing_queue, &tmp_pkt, 1);
 
     return 0;
@@ -192,11 +198,44 @@ static void write_packet(OutputFile *of, OutputStream *ost, AVPacket *pkt)
     }
 }
 
+static void submit_packet(OutputFile *of, OutputStream *ost, AVPacket *pkt)
+{
+    if (ost->sq_idx_mux >= 0) {
+        int ret = sq_send(of->sq_mux, ost->sq_idx_mux, SQPKT(pkt));
+        if (ret < 0) {
+            if (pkt)
+                av_packet_unref(pkt);
+            if (ret == AVERROR_EOF) {
+                ost->finished |= MUXER_FINISHED;
+                return;
+            } else
+                exit_program(1);
+        }
+
+        while (1) {
+            ret = sq_receive(of->sq_mux, -1, SQPKT(of->mux->sq_pkt));
+            if (ret == AVERROR_EOF || ret == AVERROR(EAGAIN))
+                return;
+            else if (ret < 0)
+                exit_program(1);
+
+            write_packet(of, output_streams[of->ost_index + ret],
+                         of->mux->sq_pkt);
+        }
+    } else {
+        if (pkt)
+            write_packet(of, ost, pkt);
+        else
+            ost->finished |= MUXER_FINISHED;
+    }
+}
+
 void of_submit_packet(OutputFile *of, AVPacket *pkt, OutputStream *ost)
 {
     AVStream *st = ost->st;
     int ret;
 
+    if (pkt) {
     /*
      * Audio encoders may split the packets --  #frames in != #packets out.
      * But there is no reordering, so we can limit the number of output packets
@@ -211,9 +250,10 @@ void of_submit_packet(OutputFile *of, AVPacket *pkt, OutputStream *ost)
         }
         ost->frame_number++;
     }
+    }
 
     if (of->mux->header_written) {
-        write_packet(of, ost, pkt);
+        submit_packet(of, ost, pkt);
     } else {
         /* the muxer is not initialized yet, buffer the packet */
         ret = queue_packet(of, ost, pkt);
@@ -321,9 +361,11 @@ int of_check_init(OutputFile *of)
             ost->mux_timebase = ost->st->time_base;
 
         while (av_fifo_read(ms->muxing_queue, &pkt, 1) >= 0) {
-            ms->muxing_queue_data_size -= pkt->size;
-            write_packet(of, ost, pkt);
-            av_packet_free(&pkt);
+            submit_packet(of, ost, pkt);
+            if (pkt) {
+                ms->muxing_queue_data_size -= pkt->size;
+                av_packet_free(&pkt);
+            }
         }
     }
 
@@ -383,6 +425,8 @@ static void mux_free(Muxer **pmux, int nb_streams)
     av_freep(&mux->streams);
     av_dict_free(&mux->opts);
 
+    av_packet_free(&mux->sq_pkt);
+
     av_freep(pmux);
 }
 
@@ -394,6 +438,9 @@ void of_close(OutputFile **pof)
     if (!of)
         return;
 
+    sq_free(&of->sq_encode);
+    sq_free(&of->sq_mux);
+
     s = of->ctx;
 
     mux_free(&of->mux, s ? s->nb_streams : 0);
@@ -437,6 +484,14 @@ int of_muxer_init(OutputFile *of, AVDictionary *opts, int64_t limit_filesize)
     if (strcmp(of->format->name, "rtp"))
         want_sdp = 0;
 
+    if (of->sq_mux) {
+        mux->sq_pkt = av_packet_alloc();
+        if (!mux->sq_pkt) {
+            ret = AVERROR(ENOMEM);
+            goto fail;
+        }
+    }
+
     /* write the header for files with no streams */
     if (of->format->flags & AVFMT_NOSTREAMS && of->ctx->nb_streams == 0) {
         ret = of_check_init(of);