rusticl/event: flush queues from dependencies

We have to flush all event dependencies inside clWaitForEvents and
clFinish implicitly otherwise applications might wait forever.

Closes: https://gitlab.freedesktop.org/mesa/mesa/-/issues/9052
Fixes: ebfffa1dae ("rusticl/event: wrong but non crashing impl of clWaitForEvents")
Cc: mesa-stable
Signed-off-by: Karol Herbst <git@karolherbst.de>
Part-of: <https://gitlab.freedesktop.org/mesa/mesa/-/merge_requests/23110>

src/gallium/frontends/rusticl/api/event.rs

@@ -74,13 +74,14 @@ pub fn wait_for_events(num_events: cl_uint, event_list: *const cl_event) -> CLRe
         return Err(CL_INVALID_CONTEXT);
     }
 
-    // TODO better impl
+    // find all queues we have to flush
+    for q in Event::deep_unflushed_queues(&evs) {
+        q.flush(false)?;
+    }
+
+    // now wait on all events and check if we got any errors
     let mut err = false;
     for e in evs {
-        if let Some(q) = &e.queue {
-            q.flush(false)?;
-        }
-
         err |= e.wait() < 0;
     }
 

src/gallium/frontends/rusticl/api/queue.rs

@@ -195,7 +195,13 @@ pub fn flush_queue(command_queue: cl_command_queue) -> CLResult<()> {
 
 pub fn finish_queue(command_queue: cl_command_queue) -> CLResult<()> {
     // CL_INVALID_COMMAND_QUEUE if command_queue is not a valid host command-queue.
-    command_queue.get_ref()?.flush(true)
+    let q = command_queue.get_ref()?;
+
+    for q in q.dependencies_for_pending_events() {
+        q.flush(false)?;
+    }
+
+    q.flush(true)
 }
 
 pub fn release_command_queue(command_queue: cl_command_queue) -> CLResult<()> {

src/gallium/frontends/rusticl/core/event.rs

@@ -9,6 +9,7 @@ use mesa_rust::pipe::fence::*;
 use mesa_rust_util::static_assert;
 use rusticl_opencl_gen::*;
 
+use std::collections::HashSet;
 use std::os::raw::c_void;
 use std::slice;
 use std::sync::Arc;
@@ -185,6 +186,39 @@ impl Event {
             status
         }
     }
+
+    fn deep_unflushed_deps_impl<'a>(&'a self, result: &mut HashSet<&'a Event>) {
+        if self.status() <= CL_SUBMITTED as i32 {
+            return;
+        }
+
+        // only scan dependencies if it's a new one
+        if result.insert(self) {
+            for e in &self.deps {
+                e.deep_unflushed_deps_impl(result);
+            }
+        }
+    }
+
+    /// does a deep search and returns a list of all dependencies including `events` which haven't
+    /// been flushed out yet
+    pub fn deep_unflushed_deps(events: &[Arc<Event>]) -> HashSet<&Event> {
+        let mut result = HashSet::new();
+
+        for e in events {
+            e.deep_unflushed_deps_impl(&mut result);
+        }
+
+        result
+    }
+
+    /// does a deep search and returns a list of all queues which haven't been flushed yet
+    pub fn deep_unflushed_queues(events: &[Arc<Event>]) -> HashSet<Arc<Queue>> {
+        Event::deep_unflushed_deps(events)
+            .iter()
+            .filter_map(|e| e.queue.clone())
+            .collect()
+    }
 }
 
 // TODO worker thread per device

src/gallium/frontends/rusticl/core/queue.rs

@@ -7,6 +7,7 @@ use crate::impl_cl_type_trait;
 use mesa_rust_util::properties::*;
 use rusticl_opencl_gen::*;
 
+use std::collections::HashSet;
 use std::sync::mpsc;
 use std::sync::Arc;
 use std::sync::Mutex;
@@ -93,6 +94,14 @@ impl Queue {
         }
         Ok(())
     }
+
+    pub fn dependencies_for_pending_events(&self) -> HashSet<Arc<Queue>> {
+        let p = self.pending.lock().unwrap();
+
+        let mut queues = Event::deep_unflushed_queues(&p);
+        queues.remove(self);
+        queues
+    }
 }
 
 impl Drop for Queue {