fftools/ffmpeg_demux: switch from AVThreadMessageQueue to ThreadQueue

* the code is made shorter and simpler
* avoids constantly allocating and freeing AVPackets, thanks to
  ThreadQueue integration with ObjPool
* is consistent with decoding/filtering/muxing
* reduces the diff in the future switch to thread-aware scheduling

This makes ifile_get_packet() always block. Any potential issues caused
by this will be resolved by the switch to thread-aware scheduling in
future commits.

1 files changed, 16 insertions, 16 deletions

diff --git a/fftools/ffmpeg.c b/fftools/ffmpeg.c
index 61fcda2526..cf8a50bffc 100644
--- a/fftools/ffmpeg.c
+++ b/fftools/ffmpeg.c
@@ -1043,9 +1043,6 @@ static int check_keyboard_interaction(int64_t cur_time)
 
 static void reset_eagain(void)
 {
-    int i;
-    for (i = 0; i < nb_input_files; i++)
-        input_files[i]->eagain = 0;
     for (OutputStream *ost = ost_iter(NULL); ost; ost = ost_iter(ost))
         ost->unavailable = 0;
 }
@@ -1069,19 +1066,14 @@ static void decode_flush(InputFile *ifile)
  *   this function should be called again
  * - AVERROR_EOF -- this function should not be called again
  */
-static int process_input(int file_index)
+static int process_input(int file_index, AVPacket *pkt)
 {
     InputFile *ifile = input_files[file_index];
     InputStream *ist;
-    AVPacket *pkt;
     int ret, i;
 
-    ret = ifile_get_packet(ifile, &pkt);
+    ret = ifile_get_packet(ifile, pkt);
 
-    if (ret == AVERROR(EAGAIN)) {
-        ifile->eagain = 1;
-        return ret;
-    }
     if (ret == 1) {
         /* the input file is looped: flush the decoders */
         decode_flush(ifile);
@@ -1128,7 +1120,7 @@ static int process_input(int file_index)
 
     ret = process_input_packet(ist, pkt, 0);
 
-    av_packet_free(&pkt);
+    av_packet_unref(pkt);
 
     return ret < 0 ? ret : 0;
 }
@@ -1138,7 +1130,7 @@ static int process_input(int file_index)
  *
  * @return  0 for success, <0 for error
  */
-static int transcode_step(OutputStream *ost)
+static int transcode_step(OutputStream *ost, AVPacket *demux_pkt)
 {
     InputStream  *ist = NULL;
     int ret;
@@ -1153,10 +1145,8 @@ static int transcode_step(OutputStream *ost)
         av_assert0(ist);
     }
 
-    ret = process_input(ist->file_index);
+    ret = process_input(ist->file_index, demux_pkt);
     if (ret == AVERROR(EAGAIN)) {
-        if (input_files[ist->file_index]->eagain)
-            ost->unavailable = 1;
         return 0;
     }
 
@@ -1182,12 +1172,19 @@ static int transcode(int *err_rate_exceeded)
     int ret = 0, i;
     InputStream *ist;
     int64_t timer_start;
+    AVPacket *demux_pkt = NULL;
 
     print_stream_maps();
 
     *err_rate_exceeded = 0;
     atomic_store(&transcode_init_done, 1);
 
+    demux_pkt = av_packet_alloc();
+    if (!demux_pkt) {
+        ret = AVERROR(ENOMEM);
+        goto fail;
+    }
+
     if (stdin_interaction) {
         av_log(NULL, AV_LOG_INFO, "Press [q] to stop, [?] for help\n");
     }
@@ -1215,7 +1212,7 @@ static int transcode(int *err_rate_exceeded)
             break;
         }
 
-        ret = transcode_step(ost);
+        ret = transcode_step(ost, demux_pkt);
         if (ret < 0 && ret != AVERROR_EOF) {
             av_log(NULL, AV_LOG_ERROR, "Error while filtering: %s\n", av_err2str(ret));
             break;
@@ -1256,6 +1253,9 @@ static int transcode(int *err_rate_exceeded)
     /* dump report by using the first video and audio streams */
     print_report(1, timer_start, av_gettime_relative());
 
+fail:
+    av_packet_free(&demux_pkt);
+
     return ret;
 }