1 files changed, 27 insertions, 6 deletions

diff --git a/filters/filter.c b/filters/filter.c
index 4134aff88b..cddeea0a3f 100644
--- a/filters/filter.c
+++ b/filters/filter.c
@@ -110,13 +110,13 @@ struct mp_filter_internal {
     struct mp_filter *error_handler;
 
     char *name;
+    bool high_priority;
 
     bool pending;
     bool async_pending;
     bool failed;
 };
 
-
 // Called when new work needs to be done on a pin belonging to the filter:
 //  - new data was requested
 //  - new data has been queued
@@ -133,7 +133,11 @@ static void add_pending(struct mp_filter *f)
     // This should probably really be some sort of priority queue, but for now
     // something naive and dumb does the job too.
     f->in->pending = true;
-    MP_TARRAY_APPEND(r, r->pending, r->num_pending, f);
+    if (f->in->high_priority) {
+        MP_TARRAY_INSERT_AT(r, r->pending, r->num_pending, 0, f);
+    } else {
+        MP_TARRAY_APPEND(r, r->pending, r->num_pending, f);
+    }
 }
 
 static void add_pending_pin(struct mp_pin *p)
@@ -217,6 +221,8 @@ bool mp_filter_graph_run(struct mp_filter *filter)
 
     flush_async_notifications(r);
 
+    bool exit_req = false;
+
     while (1) {
         if (atomic_exchange_explicit(&r->interrupt_flag, false,
                                      memory_order_acq_rel))
@@ -226,7 +232,7 @@ bool mp_filter_graph_run(struct mp_filter *filter)
                 r->wakeup_cb(r->wakeup_ctx);
             r->async_wakeup_sent = true;
             pthread_mutex_unlock(&r->async_lock);
-            break;
+            exit_req = true;
         }
 
         if (!r->num_pending) {
@@ -235,10 +241,20 @@ bool mp_filter_graph_run(struct mp_filter *filter)
                 break;
         }
 
-        struct mp_filter *next = r->pending[r->num_pending - 1];
-        r->num_pending -= 1;
-        next->in->pending = false;
+        struct mp_filter *next = NULL;
+
+        if (r->pending[0]->in->high_priority) {
+            next = r->pending[0];
+            MP_TARRAY_REMOVE_AT(r->pending, r->num_pending, 0);
+        } else if (!exit_req) {
+            next = r->pending[r->num_pending - 1];
+            r->num_pending -= 1;
+        }
+
+        if (!next)
+            break;
 
+        next->in->pending = false;
         if (next->in->info->process)
             next->in->info->process(next);
 
@@ -517,6 +533,11 @@ const struct mp_filter_info *mp_filter_get_info(struct mp_filter *f)
     return f->in->info;
 }
 
+void mp_filter_set_high_priority(struct mp_filter *f, bool pri)
+{
+    f->in->high_priority = pri;
+}
+
 void mp_filter_set_name(struct mp_filter *f, const char *name)
 {
     talloc_free(f->in->name);