avfilter/af_join: fix eof timestamps and blocking in many scenarios

libavfilter/af_join.c

@@ -50,6 +50,8 @@ typedef struct JoinContext {
     char    *channel_layout_str;
     uint64_t channel_layout;
 
+    int64_t  eof_pts;
+
     int      nb_channels;
     ChannelMap *channels;
 
@@ -372,12 +374,15 @@ static int try_push_frame(AVFilterContext *ctx)
     int i, j, ret;
 
     for (i = 0; i < ctx->nb_inputs; i++) {
-        if (!s->input_frames[i])
+        if (!s->input_frames[i]) {
-            return 0;
+            nb_samples = 0;
-        nb_samples = FFMIN(nb_samples, s->input_frames[i]->nb_samples);
+            break;
+        } else {
+            nb_samples = FFMIN(nb_samples, s->input_frames[i]->nb_samples);
+        }
     }
     if (!nb_samples)
-        return 0;
+        goto eof;
 
     /* setup the output frame */
     frame = av_frame_alloc();
@@ -454,6 +459,9 @@ static int try_push_frame(AVFilterContext *ctx)
                FFMIN(FF_ARRAY_ELEMS(frame->data), s->nb_channels));
     }
 
+    s->eof_pts = frame->pts + av_rescale_q(frame->nb_samples,
+                                           av_make_q(1, outlink->sample_rate),
+                                           outlink->time_base);
     ret = ff_filter_frame(outlink, frame);
 
     for (i = 0; i < ctx->nb_inputs; i++)
@@ -464,6 +472,16 @@ static int try_push_frame(AVFilterContext *ctx)
 fail:
     av_frame_free(&frame);
     return ret;
+eof:
+    for (i = 0; i < ctx->nb_inputs; i++) {
+        if (ff_outlink_get_status(ctx->inputs[i]) &&
+            ff_inlink_queued_samples(ctx->inputs[i]) <= 0 &&
+            !s->input_frames[i]) {
+            ff_outlink_set_status(outlink, AVERROR_EOF, s->eof_pts);
+        }
+    }
+
+    return 0;
 }
 
 static int activate(AVFilterContext *ctx)
@@ -479,16 +497,13 @@ static int activate(AVFilterContext *ctx)
         ret = ff_inlink_consume_frame(ctx->inputs[0], &s->input_frames[0]);
         if (ret < 0) {
             return ret;
-        } else if (ff_inlink_acknowledge_status(ctx->inputs[0], &status, &pts)) {
+        } else if (ret == 0 && ff_inlink_acknowledge_status(ctx->inputs[0], &status, &pts)) {
-            ff_outlink_set_status(ctx->outputs[0], status, pts);
+            ff_outlink_set_status(ctx->outputs[0], status, s->eof_pts);
             return 0;
-        } else {
-            if (ff_outlink_frame_wanted(ctx->outputs[0]) && !s->input_frames[0]) {
-                ff_inlink_request_frame(ctx->inputs[0]);
-                return 0;
-            }
         }
-        if (!s->input_frames[0]) {
+
+        if (!s->input_frames[0] && ff_outlink_frame_wanted(ctx->outputs[0])) {
+            ff_inlink_request_frame(ctx->inputs[0]);
             return 0;
         }
     }
@@ -498,20 +513,17 @@ static int activate(AVFilterContext *ctx)
     for (i = 1; i < ctx->nb_inputs && nb_samples > 0; i++) {
         if (s->input_frames[i])
             continue;
+        ret = ff_inlink_consume_samples(ctx->inputs[i], nb_samples, nb_samples, &s->input_frames[i]);
+        if (ret < 0) {
+            return ret;
+        } else if (ff_inlink_acknowledge_status(ctx->inputs[i], &status, &pts)) {
+            ff_outlink_set_status(ctx->outputs[0], status, pts);
+            return 0;
+        }
 
-        if (ff_inlink_check_available_samples(ctx->inputs[i], nb_samples) > 0) {
+        if (!s->input_frames[i]) {
-            ret = ff_inlink_consume_samples(ctx->inputs[i], nb_samples, nb_samples, &s->input_frames[i]);
+            ff_inlink_request_frame(ctx->inputs[i]);
-            if (ret < 0) {
+            return 0;
-                return ret;
-            } else if (ff_inlink_acknowledge_status(ctx->inputs[i], &status, &pts)) {
-                ff_outlink_set_status(ctx->outputs[0], status, pts);
-                return 0;
-            }
-        } else {
-            if (ff_outlink_frame_wanted(ctx->outputs[0])) {
-                ff_inlink_request_frame(ctx->inputs[i]);
-                return 0;
-            }
         }
     }