rusticl/queue: Only take a weak ref to the last Event

This resolves a memory leak when the application drops its last reference
to the queue, but never waits explicitly.

The problem was, that the queue was refed by QueueState::last and that ref
only gets dropped on a blocking wait. This is problematic as non user
Event objects also hold a ref on the Queue they are created on, therefore
causing a cyclic ref relation.

In order to resolve it, just use a weak reference. A failure of upgrading
the Weak ref is not an issue as in this case we'd only wait on an already
destroyed or processed event. The worker thread already makes sure
everything stays in sync.

Fixes: 5b3ff7e3f3 ("rusticl/queue: overhaul of the queue+event handling")
Signed-off-by: Karol Herbst <kherbst@redhat.com>
Reviewed-by: @LingMan <18294-LingMan@users.noreply.gitlab.freedesktop.org>
Part-of: <https://gitlab.freedesktop.org/mesa/mesa/-/merge_requests/25926>
(cherry picked from commit 9a3af6e1d8)

.pick_status.json

@@ -664,7 +664,7 @@
         "description": "rusticl/queue: Only take a weak ref to the last Event",
         "nominated": true,
         "nomination_type": 1,
-        "resolution": 0,
+        "resolution": 1,
         "main_sha": null,
         "because_sha": "5b3ff7e3f3d0e35f7bc17d9f208a8aeee3062427",
         "notes": null

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

@@ -13,12 +13,13 @@ use std::collections::HashSet;
 use std::sync::mpsc;
 use std::sync::Arc;
 use std::sync::Mutex;
+use std::sync::Weak;
 use std::thread;
 use std::thread::JoinHandle;
 
 struct QueueState {
     pending: Vec<Arc<Event>>,
-    last: Option<Arc<Event>>,
+    last: Weak<Event>,
     // `Sync` on `Sender` was stabilized in 1.72, until then, put it into our Mutex.
     // see https://github.com/rust-lang/rust/commit/5f56956b3c7edb9801585850d1f41b0aeb1888ff
     chan_in: mpsc::Sender<Vec<Arc<Event>>>,
@@ -62,7 +63,7 @@ impl Queue {
             props_v2: props_v2,
             state: Mutex::new(QueueState {
                 pending: Vec::new(),
-                last: None,
+                last: Weak::new(),
                 chan_in: tx_q,
             }),
             _thrd: thread::Builder::new()
@@ -134,7 +135,7 @@ impl Queue {
         // Update last if and only if we get new events, this prevents breaking application code
         // doing things like `clFlush(q); clFinish(q);`
         if let Some(last) = state.pending.last() {
-            state.last = Some(last.clone());
+            state.last = Arc::downgrade(last);
         }
 
         let events = state.pending.drain(0..).collect();
@@ -144,9 +145,10 @@ impl Queue {
             .send(events)
             .map_err(|_| CL_OUT_OF_HOST_MEMORY)?;
         if wait {
-            // Waiting on the last event is good enough here as the queue will process it in order,
-            // also take the value so we can release the event once we are done
-            state.last.take().map(|e| e.wait());
+            // Waiting on the last event is good enough here as the queue will process it in order
+            // It's not a problem if the weak ref is invalid as that means the work is already done
+            // and waiting isn't necessary anymore.
+            state.last.upgrade().map(|e| e.wait());
         }
         Ok(())
     }