fftools/ffmpeg: add a live mux modelive

Track the wallclock time at which each input packet is demuxed and
propagate it through decoding and encoding.
When the live mux option is used, drop all packets demuxed before the
muxer is opened. This is intended to avoid latency when opening the
muxer takes a long time.

5 files changed, 53 insertions, 1 deletions

diff --git a/fftools/ffmpeg.c b/fftools/ffmpeg.c
index 0682a6fcc5..4da1f94c03 100644
--- a/fftools/ffmpeg.c
+++ b/fftools/ffmpeg.c
@@ -925,6 +925,7 @@ static int encode_frame(OutputFile *of, OutputStream *ost, AVFrame *frame)
                    av_ts2str(frame->pts), av_ts2timestr(frame->pts, &enc->time_base),
                    enc->time_base.num, enc->time_base.den);
         }
+        frame->reordered_opaque = (int64_t)frame->opaque;
     }
 
     update_benchmark(NULL);
@@ -956,6 +957,11 @@ static int encode_frame(OutputFile *of, OutputStream *ost, AVFrame *frame)
             return ret;
         }
 
+        if (enc->codec->capabilities & AV_CODEC_CAP_ENCODER_REORDERED_OPAQUE)
+            pkt->opaque = (void*)enc->reordered_opaque;
+        else if (frame)
+            pkt->opaque = frame->opaque;
+
         if (debug_ts) {
             av_log(NULL, AV_LOG_INFO, "encoder -> type:%s "
                    "pkt_pts:%s pkt_pts_time:%s pkt_dts:%s pkt_dts_time:%s "
@@ -2076,6 +2082,7 @@ static int decode(AVCodecContext *avctx, AVFrame *frame, int *got_frame, AVPacke
     *got_frame = 0;
 
     if (pkt) {
+        avctx->reordered_opaque = (int64_t)pkt->opaque;
         ret = avcodec_send_packet(avctx, pkt);
         // In particular, we don't expect AVERROR(EAGAIN), because we read all
         // decoded frames with avcodec_receive_frame() until done.
@@ -2086,8 +2093,10 @@ static int decode(AVCodecContext *avctx, AVFrame *frame, int *got_frame, AVPacke
     ret = avcodec_receive_frame(avctx, frame);
     if (ret < 0 && ret != AVERROR(EAGAIN))
         return ret;
-    if (ret >= 0)
+    if (ret >= 0) {
         *got_frame = 1;
+        frame->opaque = (void*)frame->reordered_opaque;
+    }
 
     return 0;
 }
diff --git a/fftools/ffmpeg.h b/fftools/ffmpeg.h
index 2ac7cbe522..3a7527aee9 100644
--- a/fftools/ffmpeg.h
+++ b/fftools/ffmpeg.h
@@ -163,6 +163,7 @@ typedef struct OptionsContext {
     float shortest_buf_duration;
     int shortest;
     int bitexact;
+    int live;
 
     int video_disable;
     int audio_disable;
@@ -610,6 +611,7 @@ typedef struct OutputFile {
 
     int shortest;
     int bitexact;
+    int live;
 } OutputFile;
 
 extern InputStream **input_streams;
diff --git a/fftools/ffmpeg_demux.c b/fftools/ffmpeg_demux.c
index d15cee614d..01fd19f199 100644
--- a/fftools/ffmpeg_demux.c
+++ b/fftools/ffmpeg_demux.c
@@ -176,6 +176,8 @@ static void *input_thread(void *arg)
             }
         }
 
+        pkt->opaque = (void*)av_gettime_relative();
+
         msg.pkt = av_packet_alloc();
         if (!msg.pkt) {
             av_packet_unref(pkt);
diff --git a/fftools/ffmpeg_mux.c b/fftools/ffmpeg_mux.c
index 08a76f0066..24f638d0e5 100644
--- a/fftools/ffmpeg_mux.c
+++ b/fftools/ffmpeg_mux.c
@@ -29,6 +29,7 @@
 #include "libavutil/intreadwrite.h"
 #include "libavutil/log.h"
 #include "libavutil/mem.h"
+#include "libavutil/time.h"
 #include "libavutil/timestamp.h"
 #include "libavutil/thread.h"
 
@@ -69,6 +70,10 @@ struct Muxer {
     atomic_int_least64_t last_filesize;
     int header_written;
 
+    int live_started;
+    int64_t header_write_wallclock;
+    int64_t live_ts_offset;
+
     AVPacket *sq_pkt;
 };
 
@@ -100,6 +105,37 @@ static int write_packet(OutputFile *of, OutputStream *ost, AVPacket *pkt)
     if (fs >= of->mux->limit_filesize)
         return AVERROR_EOF;
 
+    if (of->live) {
+        if (!of->mux->live_started) {
+            int64_t ts = av_rescale_q(pkt->dts, ost->mux_timebase, AV_TIME_BASE_Q);
+
+            of->mux->live_ts_offset = FFMAX(of->mux->live_ts_offset, ts);
+
+            // drop all packets that were demuxed before avformat_write_header()
+            if ((int64_t)pkt->opaque < of->mux->header_write_wallclock) {
+                fprintf(stderr, "live drop %s pts %g\n",
+                        av_get_media_type_string(st->codecpar->codec_type),
+                        pkt->pts * av_q2d(ost->mux_timebase));
+                av_packet_unref(pkt);
+                return 0;
+            }
+
+            of->mux->live_started = 1;
+            fprintf(stderr, "live started, ts offset %g\n", of->mux->live_ts_offset / (double)AV_TIME_BASE);
+        }
+
+        pkt->pts -= av_rescale_q(of->mux->live_ts_offset, AV_TIME_BASE_Q, ost->mux_timebase);
+        pkt->dts -= av_rescale_q(of->mux->live_ts_offset, AV_TIME_BASE_Q, ost->mux_timebase);
+
+        if (pkt->pts < 0) {
+            fprintf(stderr, "live drop %s pts %g\n",
+                    av_get_media_type_string(st->codecpar->codec_type),
+                    pkt->pts * av_q2d(ost->mux_timebase));
+            av_packet_unref(pkt);
+            return 0;
+        }
+    }
+
     if ((st->codecpar->codec_type == AVMEDIA_TYPE_VIDEO && ost->vsync_method == VSYNC_DROP) ||
         (st->codecpar->codec_type == AVMEDIA_TYPE_AUDIO && audio_sync_method < 0))
         pkt->pts = pkt->dts = AV_NOPTS_VALUE;
@@ -478,6 +514,7 @@ int of_check_init(OutputFile *of)
     }
     //assert_avoptions(of->opts);
     of->mux->header_written = 1;
+    of->mux->header_write_wallclock = av_gettime_relative();
 
     av_dump_format(fc, of->index, fc->url, 1);
     nb_output_dumped++;
diff --git a/fftools/ffmpeg_opt.c b/fftools/ffmpeg_opt.c
index 97f14b2a5b..151b3bad07 100644
--- a/fftools/ffmpeg_opt.c
+++ b/fftools/ffmpeg_opt.c
@@ -2892,6 +2892,7 @@ static int open_output_file(OptionsContext *o, const char *filename)
     of->recording_time = o->recording_time;
     of->start_time     = o->start_time;
     of->shortest       = o->shortest;
+    of->live           = o->live;
     av_dict_copy(&format_opts, o->g->format_opts, 0);
 
     if (!strcmp(filename, "-"))
@@ -3969,6 +3970,7 @@ const OptionDef options[] = {
     { "bits_per_raw_sample", OPT_INT | HAS_ARG | OPT_EXPERT | OPT_SPEC | OPT_OUTPUT,
         { .off = OFFSET(bits_per_raw_sample) },
         "set the number of bits per raw sample", "number" },
+    { "live", OPT_BOOL | OPT_OUTPUT | OPT_OFFSET, { .off = OFFSET(live) }, "live output" },
 
     /* video options */
     { "vframes",      OPT_VIDEO | HAS_ARG  | OPT_PERFILE | OPT_OUTPUT,           { .func_arg = opt_video_frames },