video: better pipelining with vf_vapoursynth

Most of this is explained in the code comments. This change should
improve performance with vapoursynth, especially if concurrent requests
are used.

This should change nothing if vf_vapoursynth is not in the filter chain,
since non-threaded filters obviously can not asynchronously finish
filtering of frames.

3 files changed, 73 insertions, 8 deletions

diff --git a/video/filter/vf.c b/video/filter/vf.c
index 8886c64ccb..7a55f72cc5 100644
--- a/video/filter/vf.c
+++ b/video/filter/vf.c
@@ -415,14 +415,12 @@ int vf_filter_frame(struct vf_chain *c, struct mp_image *img)
     return vf_do_filter(c->first, img);
 }
 
-// Output the next queued image (if any) from the full filter chain.
-// The frame can be retrieved with vf_read_output_frame().
-//  eof: if set, assume there's no more input i.e. vf_filter_frame() will
-//       not be called (until reset) - flush all internally delayed frames
-//  returns: -1: error, 0: no output, 1: output available
-int vf_output_frame(struct vf_chain *c, bool eof)
+// Similar to vf_output_frame(), but only ensure that the filter "until" has
+// output, instead of the end of the filter chain.
+static int vf_output_frame_until(struct vf_chain *c, struct vf_instance *until,
+                                 bool eof)
 {
-    if (c->last->num_out_queued)
+    if (until->num_out_queued)
         return 1;
     if (c->initialized < 1)
         return -1;
@@ -438,10 +436,12 @@ int vf_output_frame(struct vf_chain *c, bool eof)
             }
             if (vf_has_output_frame(cur))
                 last = cur;
+            if (cur == until)
+                break;
         }
         if (!last)
             return 0;
-        if (!last->next)
+        if (last == until)
             return 1;
         int r = vf_do_filter(last->next, vf_dequeue_output_frame(last));
         if (r < 0)
@@ -449,6 +449,16 @@ int vf_output_frame(struct vf_chain *c, bool eof)
     }
 }
 
+// Output the next queued image (if any) from the full filter chain.
+// The frame can be retrieved with vf_read_output_frame().
+//  eof: if set, assume there's no more input i.e. vf_filter_frame() will
+//       not be called (until reset) - flush all internally delayed frames
+//  returns: -1: error, 0: no output, 1: output available
+int vf_output_frame(struct vf_chain *c, bool eof)
+{
+    return vf_output_frame_until(c, c->last, eof);
+}
+
 struct mp_image *vf_read_output_frame(struct vf_chain *c)
 {
     if (!c->last->num_out_queued)
@@ -456,6 +466,34 @@ struct mp_image *vf_read_output_frame(struct vf_chain *c)
     return vf_dequeue_output_frame(c->last);
 }
 
+// Some filters (vf_vapoursynth) filter on separate threads, and may need new
+// input from the decoder, even though the core does not need a new output image
+// yet (this is required to get proper pipelining in the filter). If the filter
+// needs new data, it will call c->wakeup_callback, which in turn causes the
+// core to recheck the filter chain, calling this function. Each filter is asked
+// whether it needs a frame (with vf->needs_input), and if so, it will try to
+// feed it a new frame. If this fails, it will request a new frame from the
+// core by returning 1.
+// returns -1: error, 0: nothing needed, 1: add new frame with vf_filter_frame()
+int vf_needs_input(struct vf_chain *c)
+{
+    struct vf_instance *prev = c->first;
+    for (struct vf_instance *cur = c->first; cur; cur = cur->next) {
+        while (cur->needs_input && cur->needs_input(cur)) {
+            // Get frames from preceding filters, or if there are none,
+            // request new frames from decoder.
+            int r = vf_output_frame_until(c, prev, false);
+            if (r < 1)
+                return r < 0 ? -1 : 1;
+            r = vf_do_filter(cur, vf_dequeue_output_frame(prev));
+            if (r < 0)
+                return r;
+        }
+        prev = cur;
+    }
+    return 0;
+}
+
 static void vf_forget_frames(struct vf_instance *vf)
 {
     for (int n = 0; n < vf->num_out_queued; n++)
diff --git a/video/filter/vf.h b/video/filter/vf.h
index e7e53bc3ce..4ff8398229 100644
--- a/video/filter/vf.h
+++ b/video/filter/vf.h
@@ -84,6 +84,12 @@ typedef struct vf_instance {
     // May be called multiple times, even if the filter gives no output.
     int (*filter_out)(struct vf_instance *vf);
 
+    // Optional function that checks whether the filter needs additional
+    // input. This is for filters with asynchronous behavior: they filter
+    // frames in the background, and to get good pipelining behavior, new
+    // data should be fed, even if the playback core doesn't need any yet.
+    bool (*needs_input)(struct vf_instance *vf);
+
     void (*uninit)(struct vf_instance *vf);
 
     char *label;
@@ -121,6 +127,12 @@ struct vf_chain {
     struct MPOpts *opts;
     struct mpv_global *global;
     struct mp_hwdec_info *hwdec;
+
+    // Call when the filter chain wants new processing (for filters with
+    // asynchronous behavior) - must be immutable once filters are created,
+    // since they are supposed to call it from foreign threads.
+    void (*wakeup_callback)(void *ctx);
+    void *wakeup_callback_ctx;
 };
 
 typedef struct vf_seteq {
@@ -150,6 +162,7 @@ int vf_control_any(struct vf_chain *c, int cmd, void *arg);
 int vf_control_by_label(struct vf_chain *c, int cmd, void *arg, bstr label);
 int vf_filter_frame(struct vf_chain *c, struct mp_image *img);
 int vf_output_frame(struct vf_chain *c, bool eof);
+int vf_needs_input(struct vf_chain *c);
 struct mp_image *vf_read_output_frame(struct vf_chain *c);
 void vf_seek_reset(struct vf_chain *c);
 struct vf_instance *vf_append_filter(struct vf_chain *c, const char *name,
diff --git a/video/filter/vf_vapoursynth.c b/video/filter/vf_vapoursynth.c
index ba630f8df1..30fd992071 100644
--- a/video/filter/vf_vapoursynth.c
+++ b/video/filter/vf_vapoursynth.c
@@ -381,6 +381,17 @@ static int filter_out(struct vf_instance *vf)
     return ret;
 }
 
+static bool needs_input(struct vf_instance *vf)
+{
+    struct vf_priv_s *p = vf->priv;
+    bool r = false;
+    pthread_mutex_lock(&p->lock);
+    locked_read_output(vf);
+    r = vf->num_out_queued < p->max_requests && locked_need_input(vf);
+    pthread_mutex_unlock(&p->lock);
+    return r;
+}
+
 static void VS_CC infiltInit(VSMap *in, VSMap *out, void **instanceData,
                              VSNode *node, VSCore *core, const VSAPI *vsapi)
 {
@@ -451,6 +462,8 @@ static const VSFrameRef *VS_CC infiltGetFrame(int frameno, int activationReason,
             if (p->num_buffered) {
                 drain_oldest_buffered_frame(p);
                 pthread_cond_broadcast(&p->wakeup);
+                if (vf->chain->wakeup_callback)
+                    vf->chain->wakeup_callback(vf->chain->wakeup_callback_ctx);
                 continue;
             }
         }
@@ -682,6 +695,7 @@ static int vf_open(vf_instance_t *vf)
     vf->config = config;
     vf->filter_ext = filter_ext;
     vf->filter_out = filter_out;
+    vf->needs_input = needs_input;
     vf->query_format = query_format;
     vf->control = control;
     vf->uninit = uninit;