filter: add functions to suspend filtering temporarily

Filtering is integrated into an event loop, which is something the
filter API user provides. To make interacting with the event loop
easier, and in particular to avoid filtering to block event handling,
add functions the event loop code can suspend filtering.

While we cannot actually suspend a single filter, it's pretty easy to
suspend the filter graph run loop itself, which is responsible for
selecting which filter to run next.

This commit shouldn't change behavior at all, but the functions will be
used in later commits.

1 files changed, 41 insertions, 1 deletions

diff --git a/filters/filter.c b/filters/filter.c
index dfe6df953a..a600b7dec4 100644
--- a/filters/filter.c
+++ b/filters/filter.c
@@ -1,8 +1,11 @@
+#include <math.h>
 #include <pthread.h>
 
 #include "common/common.h"
 #include "common/global.h"
 #include "common/msg.h"
+#include "osdep/atomic.h"
+#include "osdep/timer.h"
 #include "video/hwdec.h"
 
 #include "filter.h"
@@ -65,6 +68,9 @@ struct filter_runner {
 
     struct mp_filter *root_filter;
 
+    double max_run_time;
+    atomic_bool interrupt_flag;
+
     // If we're currently running the filter graph (for avoiding recursion).
     bool filtering;
 
@@ -177,6 +183,10 @@ bool mp_filter_run(struct mp_filter *filter)
 {
     struct filter_runner *r = filter->in->runner;
 
+    int64_t end_time = 0;
+    if (isfinite(r->max_run_time))
+        end_time = mp_add_timeout(mp_time_us(), MPMAX(r->max_run_time, 0));
+
     // (could happen with separate filter graphs calling each other, for now
     // ignore this issue as we don't use such a setup anywhere)
     assert(!r->filtering);
@@ -187,13 +197,30 @@ bool mp_filter_run(struct mp_filter *filter)
     //       to queue a wakeup again later. So do not call this in the loop.
     flush_async_notifications(r);
 
-    while (r->num_pending) {
+    while (1) {
+        if (atomic_exchange_explicit(&r->interrupt_flag, false,
+                                     memory_order_acq_rel))
+        {
+            pthread_mutex_lock(&r->async_lock);
+            if (!r->async_wakeup_sent && r->wakeup_cb)
+                r->wakeup_cb(r->wakeup_ctx);
+            r->async_wakeup_sent = true;
+            pthread_mutex_unlock(&r->async_lock);
+            break;
+        }
+
+        if (!r->num_pending)
+            break;
+
         struct mp_filter *next = r->pending[r->num_pending - 1];
         r->num_pending -= 1;
         next->in->pending = false;
 
         if (next->in->info->process)
             next->in->info->process(next);
+
+        if (end_time && mp_time_us() >= end_time)
+            mp_filter_graph_interrupt(r->root_filter);
     }
 
     r->filtering = false;
@@ -644,6 +671,18 @@ void mp_filter_mark_async_progress(struct mp_filter *f)
     filter_wakeup(f, true);
 }
 
+void mp_filter_graph_set_max_run_time(struct mp_filter *f, double seconds)
+{
+    struct filter_runner *r = f->in->runner;
+    r->max_run_time = seconds;
+}
+
+void mp_filter_graph_interrupt(struct mp_filter *f)
+{
+    struct filter_runner *r = f->in->runner;
+    atomic_store(&r->interrupt_flag, true);
+}
+
 void mp_filter_free_children(struct mp_filter *f)
 {
     while(f->in->num_children)
@@ -717,6 +756,7 @@ struct mp_filter *mp_filter_create_with_params(struct mp_filter_params *params)
         *f->in->runner = (struct filter_runner){
             .global = params->global,
             .root_filter = f,
+            .max_run_time = INFINITY,
         };
         pthread_mutex_init(&f->in->runner->async_lock, NULL);
     }