ffmpeg: reduce differences with avconv for in/out scheduling.

Rework the transcode() function and its immediate annexes to have
the same structure as in avconv, while still maintaining proper
scheduling to avoid accumulation.

Using -filter_complex without inputs now works.

ffmpeg.c

@@ -950,16 +950,20 @@ static void do_video_stats(AVFormatContext *os, OutputStream *ost,
     }
 }
 
-/* check for new output on any of the filtergraphs */
+/**
-static int poll_filters(void)
+ * Get and encode new output from any of the filtergraphs, without causing
+ * activity.
+ *
+ * @return  0 for success, <0 for severe errors
+ */
+static int reap_filters(void)
 {
     AVFilterBufferRef *picref;
     AVFrame *filtered_frame = NULL;
-    int i, ret, ret_all;
+    int i;
-    unsigned nb_success = 1, av_uninit(nb_eof);
     int64_t frame_pts;
 
-    while (1) {
+    /* TODO reindent */
         /* Reap all buffers present in the buffer sinks */
         for (i = 0; i < nb_output_streams; i++) {
             OutputStream *ost = output_streams[i];
@@ -1029,27 +1033,8 @@ static int poll_filters(void)
                 avfilter_unref_buffer(picref);
             }
         }
-        if (!nb_success) /* from last round */
+
-            break;
+    return 0;
-        /* Request frames through all the graphs */
-        ret_all = nb_success = nb_eof = 0;
-        for (i = 0; i < nb_filtergraphs; i++) {
-            ret = avfilter_graph_request_oldest(filtergraphs[i]->graph);
-            if (!ret) {
-                nb_success++;
-            } else if (ret == AVERROR_EOF) {
-                nb_eof++;
-            } else if (ret != AVERROR(EAGAIN)) {
-                char buf[256];
-                av_strerror(ret, buf, sizeof(buf));
-                av_log(NULL, AV_LOG_WARNING,
-                       "Error in request_frame(): %s\n", buf);
-                ret_all = ret;
-            }
-        }
-        /* Try again if anything succeeded */
-    }
-    return nb_eof == nb_filtergraphs ? AVERROR_EOF : ret_all;
 }
 
 static void print_report(int is_last_report, int64_t timer_start, int64_t cur_time)
@@ -1429,7 +1414,7 @@ static int decode_audio(InputStream *ist, AVPacket *pkt, int *got_output)
 
     if (ret >= 0 && avctx->sample_rate <= 0) {
         av_log(avctx, AV_LOG_ERROR, "Sample rate %d invalid\n", avctx->sample_rate);
-        return AVERROR_INVALIDDATA;
+        ret = AVERROR_INVALIDDATA;
     }
 
     if (!*got_output || ret < 0) {
@@ -2422,84 +2407,27 @@ static int need_output(void)
     return 0;
 }
 
-static int input_acceptable(InputStream *ist)
-{
-    av_assert1(!ist->discard);
-    return !input_files[ist->file_index]->eagain &&
-           !input_files[ist->file_index]->eof_reached;
-}
-
-static int find_graph_input(FilterGraph *graph)
-{
-    int i, nb_req_max = 0, file_index = -1;
-
-    for (i = 0; i < graph->nb_inputs; i++) {
-        int nb_req = av_buffersrc_get_nb_failed_requests(graph->inputs[i]->filter);
-        if (nb_req > nb_req_max) {
-            InputStream *ist = graph->inputs[i]->ist;
-            if (input_acceptable(ist)) {
-                nb_req_max = nb_req;
-                file_index = ist->file_index;
-            }
-        }
-    }
-
-    return file_index;
-}
-
 /**
- * Select the input file to read from.
+ * Select the output stream to process.
  *
- * @return  >=0 index of the input file to use;
+ * @return  selected output stream, or NULL if none available
- *          -1  if no file is acceptable;
- *          -2  to read from filters without reading from a file
  */
-static int select_input_file(void)
+static OutputStream *choose_output(void)
 {
-    int i, ret, nb_active_out = nb_output_streams, ost_index = -1;
+    int i;
-    int64_t opts_min;
+    int64_t opts_min = INT64_MAX;
-    OutputStream *ost;
+    OutputStream *ost_min = NULL;
-    AVFilterBufferRef *dummy;
 
-    for (i = 0; i < nb_output_streams; i++)
+    for (i = 0; i < nb_output_streams; i++) {
-        nb_active_out -= output_streams[i]->unavailable =
+        OutputStream *ost = output_streams[i];
-            output_streams[i]->finished;
+        int64_t opts = av_rescale_q(ost->st->cur_dts, ost->st->time_base,
-    while (nb_active_out) {
+                                    AV_TIME_BASE_Q);
-        opts_min = INT64_MAX;
+        if (!ost->unavailable && !ost->finished && opts < opts_min) {
-        ost_index = -1;
+            opts_min = opts;
-        for (i = 0; i < nb_output_streams; i++) {
+            ost_min  = ost;
-            OutputStream *ost = output_streams[i];
-            int64_t opts = av_rescale_q(ost->st->cur_dts, ost->st->time_base,
-                                        AV_TIME_BASE_Q);
-            if (!ost->unavailable && opts < opts_min) {
-                opts_min  = opts;
-                ost_index = i;
-            }
         }
-        if (ost_index < 0)
-            return -1;
-
-        ost = output_streams[ost_index];
-        if (ost->source_index >= 0) {
-            /* ost is directly connected to an input */
-            InputStream *ist = input_streams[ost->source_index];
-            if (input_acceptable(ist))
-                return ist->file_index;
-        } else {
-            /* ost is connected to a complex filtergraph */
-            av_assert1(ost->filter);
-            ret = av_buffersink_get_buffer_ref(ost->filter->filter, &dummy,
-                                               AV_BUFFERSINK_FLAG_PEEK);
-            if (ret >= 0)
-                return -2;
-            ret = find_graph_input(ost->filter->graph);
-            if (ret >= 0)
-                return ret;
-        }
-        ost->unavailable = 1;
-        nb_active_out--;
     }
-    return -1;
+    return ost_min;
 }
 
 static int check_keyboard_interaction(int64_t cur_time)
@@ -2717,8 +2645,8 @@ static int get_input_packet(InputFile *f, AVPacket *pkt)
 static int got_eagain(void)
 {
     int i;
-    for (i = 0; i < nb_input_files; i++)
+    for (i = 0; i < nb_output_streams; i++)
-        if (input_files[i]->eagain)
+        if (output_streams[i]->unavailable)
             return 1;
     return 0;
 }
@@ -2728,6 +2656,20 @@ static void reset_eagain(void)
     int i;
     for (i = 0; i < nb_input_files; i++)
         input_files[i]->eagain = 0;
+    for (i = 0; i < nb_output_streams; i++)
+        output_streams[i]->unavailable = 0;
+}
+
+static void close_output_stream(OutputStream *ost)
+{
+    OutputFile *of = output_files[ost->file_index];
+
+    ost->finished = 1;
+    if (of->shortest) {
+        int i;
+        for (i = 0; i < of->ctx->nb_streams; i++)
+            output_streams[of->ost_index + i]->finished = 1;
+    }
 }
 
 /**
@@ -2737,32 +2679,13 @@ static void reset_eagain(void)
  *   this function should be called again
  * - AVERROR_EOF -- this function should not be called again
  */
-static int process_input(void)
+static int process_input(int file_index)
 {
-    InputFile *ifile;
+    InputFile *ifile = input_files[file_index];
     AVFormatContext *is;
     InputStream *ist;
     AVPacket pkt;
     int ret, i, j;
-    int file_index;
-
-    /* select the stream that we must read now */
-    file_index = select_input_file();
-    /* if none, if is finished */
-    if (file_index == -2) {
-        poll_filters() ;
-        return AVERROR(EAGAIN);
-    }
-    if (file_index < 0) {
-        if (got_eagain()) {
-            reset_eagain();
-            av_usleep(10000);
-            return AVERROR(EAGAIN);
-        }
-        av_log(NULL, AV_LOG_VERBOSE, "No more inputs to read from, finishing.\n");
-        return AVERROR_EOF;
-    }
-    ifile = input_files[file_index];
 
     is  = ifile->ctx;
     ret = get_input_packet(ifile, &pkt);
@@ -2783,19 +2706,14 @@ static int process_input(void)
             ist = input_streams[ifile->ist_index + i];
             if (ist->decoding_needed)
                 output_packet(ist, NULL);
-            poll_filters();
-        }
 
-        for (i = 0; i < nb_output_streams; i++) {
+            /* mark all outputs that don't go through lavfi as finished */
-            OutputStream *ost    = output_streams[i];
+            for (j = 0; j < nb_output_streams; j++) {
-            OutputFile *of       = output_files[ost->file_index];
+                OutputStream *ost = output_streams[j];
-            AVFormatContext *os  = output_files[ost->file_index]->ctx;
 
-            if (of->shortest) {
+                if (ost->source_index == ifile->ist_index + i &&
-                int j;
+                    (ost->stream_copy || ost->enc->type == AVMEDIA_TYPE_SUBTITLE))
-                for (j = 0; j < of->ctx->nb_streams; j++)
+                    close_output_stream(ost);
-                    output_streams[of->ost_index + j]->finished = 1;
-                continue;
             }
         }
 
@@ -2893,16 +2811,14 @@ static int process_input(void)
 
     sub2video_heartbeat(ist, pkt.pts);
 
-    if ((ret = output_packet(ist, &pkt)) < 0 ||
+    ret = output_packet(ist, &pkt);
-        ((ret = poll_filters()) < 0 && ret != AVERROR_EOF)) {
+    if (ret < 0) {
         char buf[128];
         av_strerror(ret, buf, sizeof(buf));
         av_log(NULL, AV_LOG_ERROR, "Error while decoding stream #%d:%d: %s\n",
                 ist->file_index, ist->st->index, buf);
         if (exit_on_error)
             exit_program(1);
-        av_free_packet(&pkt);
-        return AVERROR(EAGAIN);
     }
 
 discard_packet:
@@ -2911,6 +2827,98 @@ discard_packet:
     return 0;
 }
 
+/**
+ * Perform a step of transcoding for the specified filter graph.
+ *
+ * @param[in]  graph     filter graph to consider
+ * @param[out] best_ist  input stream where a frame would allow to continue
+ * @return  0 for success, <0 for error
+ */
+static int transcode_from_filter(FilterGraph *graph, InputStream **best_ist)
+{
+    int i, ret;
+    int nb_requests, nb_requests_max = 0;
+    InputFilter *ifilter;
+    InputStream *ist;
+
+    *best_ist = NULL;
+    ret = avfilter_graph_request_oldest(graph->graph);
+    if (ret >= 0)
+        return reap_filters();
+
+    if (ret == AVERROR_EOF) {
+        ret = reap_filters();
+        for (i = 0; i < graph->nb_outputs; i++)
+            close_output_stream(graph->outputs[i]->ost);
+        return ret;
+    }
+    if (ret != AVERROR(EAGAIN))
+        return ret;
+
+    for (i = 0; i < graph->nb_inputs; i++) {
+        ifilter = graph->inputs[i];
+        ist = ifilter->ist;
+        if (input_files[ist->file_index]->eagain ||
+            input_files[ist->file_index]->eof_reached)
+            continue;
+        nb_requests = av_buffersrc_get_nb_failed_requests(ifilter->filter);
+        if (nb_requests > nb_requests_max) {
+            nb_requests_max = nb_requests;
+            *best_ist = ist;
+        }
+    }
+
+    if (!*best_ist)
+        for (i = 0; i < graph->nb_outputs; i++)
+            graph->outputs[i]->ost->unavailable = 1;
+
+    return 0;
+}
+
+/**
+ * Run a single step of transcoding.
+ *
+ * @return  0 for success, <0 for error
+ */
+static int transcode_step(void)
+{
+    OutputStream *ost;
+    InputStream  *ist;
+    int ret;
+
+    ost = choose_output();
+    if (!ost) {
+        if (got_eagain()) {
+            reset_eagain();
+            av_usleep(10000);
+            return 0;
+        }
+        av_log(NULL, AV_LOG_VERBOSE, "No more inputs to read from, finishing.\n");
+        return AVERROR_EOF;
+    }
+
+    if (ost->filter) {
+        if ((ret = transcode_from_filter(ost->filter->graph, &ist)) < 0)
+            return ret;
+        if (!ist)
+            return 0;
+    } else {
+        av_assert0(ost->source_index >= 0);
+        ist = input_streams[ost->source_index];
+    }
+
+    ret = process_input(ist->file_index);
+    if (ret == AVERROR(EAGAIN)) {
+        if (input_files[ist->file_index]->eagain)
+            ost->unavailable = 1;
+        return 0;
+    }
+    if (ret < 0)
+        return ret == AVERROR_EOF ? 0 : ret;
+
+    return reap_filters();
+}
+
 /*
  * The following code is the main loop of the file converter
  */
@@ -2951,12 +2959,11 @@ static int transcode(void)
             break;
         }
 
-        ret = process_input();
+        ret = transcode_step();
         if (ret < 0) {
-            if (ret == AVERROR(EAGAIN))
+            if (ret == AVERROR_EOF || ret == AVERROR(EAGAIN))
                 continue;
-            if (ret == AVERROR_EOF)
+
-                break;
             av_log(NULL, AV_LOG_ERROR, "Error while filtering.\n");
             break;
         }
@@ -2975,7 +2982,6 @@ static int transcode(void)
             output_packet(ist, NULL);
         }
     }
-    poll_filters();
     flush_encoders();
 
     term_exit();