webgpu/
poll_thread.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
4
5//! Data and main loop of WGPU poll thread.
6//!
7//! This is roughly based on <https://github.com/LucentFlux/wgpu-async/blob/1322c7e3fcdfc1865a472c7bbbf0e2e06dcf4da8/src/wgpu_future.rs>
8
9use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
10use std::sync::{Arc, Mutex, MutexGuard};
11use std::thread::JoinHandle;
12
13use log::warn;
14use wgpu_core::global::Global;
15
16/// Polls devices while there is something to poll.
17///
18/// This objects corresponds to a thread that parks itself when there is no work,
19/// waiting on it, and then calls `poll_all_devices` repeatedly to block.
20///
21/// The thread dies when this object is dropped, and all work in submission is done.
22///
23/// ## Example
24/// ```no_run
25/// let token = self.poller.token(); // create a new token
26/// let callback = SubmittedWorkDoneClosure::from_rust(Box::from(move || {
27///    drop(token); // drop token as closure has been fired
28///    // ...
29/// }));
30/// let result = gfx_select!(queue_id => global.queue_on_submitted_work_done(queue_id, callback));
31/// self.poller.wake(); // wake poller thread to actually poll
32/// ```
33#[derive(Debug)]
34pub(crate) struct Poller {
35    /// The number of closures that still needs to be fired.
36    /// When this is 0, the thread can park itself.
37    work_count: Arc<AtomicUsize>,
38    /// True if thread should die after all work in submission is done
39    is_done: Arc<AtomicBool>,
40    /// Handle to the WGPU poller thread (to be used for unparking the thread)
41    handle: Option<JoinHandle<()>>,
42    /// Lock for device maintain calls (in poll_all_devices and queue_submit)
43    ///
44    /// This is workaround for wgpu deadlocks: <https://github.com/gfx-rs/wgpu/issues/5572>
45    lock: Arc<Mutex<()>>,
46}
47
48#[inline]
49fn poll_all_devices(
50    global: &Arc<Global>,
51    more_work: &mut bool,
52    force_wait: bool,
53    lock: &Mutex<()>,
54) {
55    let _guard = lock.lock().unwrap();
56    match global.poll_all_devices(force_wait) {
57        Ok(all_queue_empty) => *more_work = !all_queue_empty,
58        Err(e) => warn!("Poller thread got `{e}` on poll_all_devices."),
59    }
60    // drop guard
61}
62
63impl Poller {
64    pub(crate) fn new(global: Arc<Global>) -> Self {
65        let work_count = Arc::new(AtomicUsize::new(0));
66        let is_done = Arc::new(AtomicBool::new(false));
67        let work = work_count.clone();
68        let done = is_done.clone();
69        let lock = Arc::new(Mutex::new(()));
70        Self {
71            work_count,
72            is_done,
73            lock: Arc::clone(&lock),
74            handle: Some(
75                std::thread::Builder::new()
76                    .name("WGPU poller".into())
77                    .spawn(move || {
78                        while !done.load(Ordering::Acquire) {
79                            let mut more_work = false;
80                            // Do non-blocking poll unconditionally
81                            // so every `ẁake` (even spurious) will do at least one poll.
82                            // this is mostly useful for stuff that is deferred
83                            // to maintain calls in wgpu (device resource destruction)
84                            poll_all_devices(&global, &mut more_work, false, &lock);
85                            while more_work || work.load(Ordering::Acquire) != 0 {
86                                poll_all_devices(&global, &mut more_work, true, &lock);
87                            }
88                            std::thread::park(); // TODO: should we use timeout here
89                        }
90                    })
91                    .expect("Spawning thread should not fail"),
92            ),
93        }
94    }
95
96    /// Creates a token of work
97    pub(crate) fn token(&self) -> WorkToken {
98        let prev = self.work_count.fetch_add(1, Ordering::AcqRel);
99        debug_assert!(
100            prev < usize::MAX,
101            "cannot have more than `usize::MAX` outstanding operations on the GPU"
102        );
103        WorkToken {
104            work_count: Arc::clone(&self.work_count),
105        }
106    }
107
108    /// Wakes the poller thread to start polling.
109    pub(crate) fn wake(&self) {
110        self.handle
111            .as_ref()
112            .expect("Poller thread does not exist!")
113            .thread()
114            .unpark();
115    }
116
117    /// Lock for device maintain calls (in poll_all_devices and queue_submit)
118    pub(crate) fn lock(&self) -> MutexGuard<'_, ()> {
119        self.lock.lock().unwrap()
120    }
121}
122
123impl Drop for Poller {
124    fn drop(&mut self) {
125        self.is_done.store(true, Ordering::Release);
126
127        let handle = self.handle.take().expect("Poller dropped twice");
128        handle.thread().unpark();
129        handle.join().expect("Poller thread panicked");
130    }
131}
132
133/// RAII indicating that there is some work enqueued (closure to be fired),
134/// while this token is held.
135pub(crate) struct WorkToken {
136    work_count: Arc<AtomicUsize>,
137}
138
139impl Drop for WorkToken {
140    fn drop(&mut self) {
141        self.work_count.fetch_sub(1, Ordering::AcqRel);
142    }
143}
webgpu/poll_thread.rs

webgpu/
poll_thread.rs
