fftools/ffmpeg_demux: implement -bsf for input

Previously bitstream filters could only be applied right before muxing,
this allows to apply them right after demuxing.

1 files changed, 124 insertions, 15 deletions

diff --git a/fftools/ffmpeg_demux.c b/fftools/ffmpeg_demux.c
index eae1f0bde5..16d4f67e59 100644
--- a/fftools/ffmpeg_demux.c
+++ b/fftools/ffmpeg_demux.c
@@ -34,6 +34,7 @@
 #include "libavutil/time.h"
 #include "libavutil/timestamp.h"
 
+#include "libavcodec/bsf.h"
 #include "libavcodec/packet.h"
 
 #include "libavformat/avformat.h"
@@ -71,6 +72,8 @@ typedef struct DemuxStream {
 
     const AVCodecDescriptor *codec_desc;
 
+    AVBSFContext *bsf;
+
     /* number of packets successfully read for this stream */
     uint64_t nb_packets;
     // combined size of all the packets read
@@ -118,6 +121,8 @@ typedef struct Demuxer {
 typedef struct DemuxThreadContext {
     // packet used for reading from the demuxer
     AVPacket *pkt_demux;
+    // packet for reading from BSFs
+    AVPacket *pkt_bsf;
 } DemuxThreadContext;
 
 static DemuxStream *ds_from_ist(InputStream *ist)
@@ -513,13 +518,17 @@ static int do_send(Demuxer *d, DemuxStream *ds, AVPacket *pkt, unsigned flags,
     return 0;
 }
 
-static int demux_send(Demuxer *d, DemuxStream *ds, AVPacket *pkt, unsigned flags)
+static int demux_send(Demuxer *d, DemuxThreadContext *dt, DemuxStream *ds,
+                      AVPacket *pkt, unsigned flags)
 {
     InputFile  *f = &d->f;
     int ret;
 
+    // pkt can be NULL only when flushing BSFs
+    av_assert0(ds->bsf || pkt);
+
     // send heartbeat for sub2video streams
-    if (d->pkt_heartbeat && pkt->pts != AV_NOPTS_VALUE) {
+    if (d->pkt_heartbeat && pkt && pkt->pts != AV_NOPTS_VALUE) {
         for (int i = 0; i < f->nb_streams; i++) {
             DemuxStream *ds1 = ds_from_ist(f->streams[i]);
 
@@ -537,10 +546,69 @@ static int demux_send(Demuxer *d, DemuxStream *ds, AVPacket *pkt, unsigned flags
         }
     }
 
-    ret = do_send(d, ds, pkt, flags, "demuxed");
-    if (ret < 0)
-        return ret;
+    if (ds->bsf) {
+        if (pkt)
+            av_packet_rescale_ts(pkt, pkt->time_base, ds->bsf->time_base_in);
+
+        ret = av_bsf_send_packet(ds->bsf, pkt);
+        if (ret < 0) {
+            if (pkt)
+                av_packet_unref(pkt);
+            av_log(ds, AV_LOG_ERROR, "Error submitting a packet for filtering: %s\n",
+                   av_err2str(ret));
+            return ret;
+        }
 
+        while (1) {
+            ret = av_bsf_receive_packet(ds->bsf, dt->pkt_bsf);
+            if (ret == AVERROR(EAGAIN))
+                return 0;
+            else if (ret < 0) {
+                if (ret != AVERROR_EOF)
+                    av_log(ds, AV_LOG_ERROR,
+                           "Error applying bitstream filters to a packet: %s\n",
+                           av_err2str(ret));
+                return ret;
+            }
+
+            dt->pkt_bsf->time_base = ds->bsf->time_base_out;
+
+            ret = do_send(d, ds, dt->pkt_bsf, 0, "filtered");
+            if (ret < 0) {
+                av_packet_unref(dt->pkt_bsf);
+                return ret;
+            }
+        }
+    } else {
+        ret = do_send(d, ds, pkt, flags, "demuxed");
+        if (ret < 0)
+            return ret;
+    }
+
+    return 0;
+}
+
+static int demux_bsf_flush(Demuxer *d, DemuxThreadContext *dt)
+{
+    InputFile *f = &d->f;
+    int ret;
+
+    for (unsigned i = 0; i < f->nb_streams; i++) {
+        DemuxStream *ds = ds_from_ist(f->streams[i]);
+
+        if (!ds->bsf)
+            continue;
+
+        ret = demux_send(d, dt, ds, NULL, 0);
+        ret = (ret == AVERROR_EOF) ? 0 : (ret < 0) ? ret : AVERROR_BUG;
+        if (ret < 0) {
+            av_log(ds, AV_LOG_ERROR, "Error flushing BSFs: %s\n",
+                   av_err2str(ret));
+            return ret;
+        }
+
+        av_bsf_flush(ds->bsf);
+    }
 
     return 0;
 }
@@ -573,6 +641,7 @@ static void thread_set_name(InputFile *f)
 static void demux_thread_uninit(DemuxThreadContext *dt)
 {
     av_packet_free(&dt->pkt_demux);
+    av_packet_free(&dt->pkt_bsf);
 
     memset(dt, 0, sizeof(*dt));
 }
@@ -585,6 +654,10 @@ static int demux_thread_init(DemuxThreadContext *dt)
     if (!dt->pkt_demux)
         return AVERROR(ENOMEM);
 
+    dt->pkt_bsf = av_packet_alloc();
+    if (!dt->pkt_bsf)
+        return AVERROR(ENOMEM);
+
     return 0;
 }
 
@@ -619,10 +692,22 @@ static void *input_thread(void *arg)
             continue;
         }
         if (ret < 0) {
+            int ret_bsf;
+
+            if (ret == AVERROR_EOF)
+                av_log(d, AV_LOG_VERBOSE, "EOF while reading input\n");
+            else {
+                av_log(d, AV_LOG_ERROR, "Error during demuxing: %s\n",
+                       av_err2str(ret));
+                ret = exit_on_error ? ret : 0;
+            }
+
+            ret_bsf = demux_bsf_flush(d, &dt);
+            ret = err_merge(ret == AVERROR_EOF ? 0 : ret, ret_bsf);
+
             if (d->loop) {
                 /* signal looping to our consumers */
                 dt.pkt_demux->stream_index = -1;
-
                 ret = sch_demux_send(d->sch, f->index, dt.pkt_demux, 0);
                 if (ret >= 0)
                     ret = seek_to_start(d, (Timestamp){ .ts = dt.pkt_demux->pts,
@@ -633,14 +718,6 @@ static void *input_thread(void *arg)
                 /* fallthrough to the error path */
             }
 
-            if (ret == AVERROR_EOF)
-                av_log(d, AV_LOG_VERBOSE, "EOF while reading input\n");
-            else {
-                av_log(d, AV_LOG_ERROR, "Error during demuxing: %s\n",
-                       av_err2str(ret));
-                ret = exit_on_error ? ret : 0;
-            }
-
             break;
         }
 
@@ -677,7 +754,7 @@ static void *input_thread(void *arg)
         if (d->readrate)
             readrate_sleep(d);
 
-        ret = demux_send(d, ds, dt.pkt_demux, send_flags);
+        ret = demux_send(d, &dt, ds, dt.pkt_demux, send_flags);
         if (ret < 0)
             break;
     }
@@ -735,9 +812,11 @@ static void demux_final_stats(Demuxer *d)
 static void ist_free(InputStream **pist)
 {
     InputStream *ist = *pist;
+    DemuxStream *ds;
 
     if (!ist)
         return;
+    ds = ds_from_ist(ist);
 
     dec_free(&ist->decoder);
 
@@ -749,6 +828,8 @@ static void ist_free(InputStream **pist)
     avcodec_free_context(&ist->dec_ctx);
     avcodec_parameters_free(&ist->par);
 
+    av_bsf_free(&ds->bsf);
+
     av_freep(pist);
 }
 
@@ -1039,6 +1120,7 @@ static int ist_add(const OptionsContext *o, Demuxer *d, AVStream *st)
     const char *hwaccel = NULL;
     char *hwaccel_output_format = NULL;
     char *codec_tag = NULL;
+    char *bsfs = NULL;
     char *next;
     char *discard_str = NULL;
     int ret;
@@ -1258,6 +1340,33 @@ static int ist_add(const OptionsContext *o, Demuxer *d, AVStream *st)
         return ret;
     }
 
+    MATCH_PER_STREAM_OPT(bitstream_filters, str, bsfs, ic, st);
+    if (bsfs) {
+        ret = av_bsf_list_parse_str(bsfs, &ds->bsf);
+        if (ret < 0) {
+            av_log(ist, AV_LOG_ERROR,
+                   "Error parsing bitstream filter sequence '%s': %s\n",
+                   bsfs, av_err2str(ret));
+            return ret;
+        }
+
+        ret = avcodec_parameters_copy(ds->bsf->par_in, ist->par);
+        if (ret < 0)
+            return ret;
+        ds->bsf->time_base_in = ist->st->time_base;
+
+        ret = av_bsf_init(ds->bsf);
+        if (ret < 0) {
+            av_log(ist, AV_LOG_ERROR, "Error initializing bitstream filters: %s\n",
+                   av_err2str(ret));
+            return ret;
+        }
+
+        ret = avcodec_parameters_copy(ist->par, ds->bsf->par_out);
+        if (ret < 0)
+            return ret;
+    }
+
     ds->codec_desc = avcodec_descriptor_get(ist->par->codec_id);
 
     return 0;