1use std::{
2 sync::atomic::{AtomicBool, Ordering},
3 thread::{ThreadId, current},
4 time::{Duration, Instant},
5};
6
7use anyhow::Context;
8use util::ResultExt;
9use windows::{
10 System::Threading::{
11 ThreadPool, ThreadPoolTimer, TimerElapsedHandler, WorkItemHandler, WorkItemPriority,
12 },
13 Win32::{
14 Foundation::{LPARAM, WPARAM},
15 Media::{timeBeginPeriod, timeEndPeriod},
16 System::Threading::{
17 GetCurrentThread, HIGH_PRIORITY_CLASS, SetPriorityClass, SetThreadPriority,
18 THREAD_PRIORITY_TIME_CRITICAL,
19 },
20 UI::WindowsAndMessaging::PostMessageW,
21 },
22};
23
24use crate::{
25 GLOBAL_THREAD_TIMINGS, HWND, PlatformDispatcher, Priority, PriorityQueueSender,
26 RunnableVariant, SafeHwnd, THREAD_TIMINGS, TaskTiming, ThreadTaskTimings, TimerResolutionGuard,
27 WM_GPUI_TASK_DISPATCHED_ON_MAIN_THREAD, profiler,
28};
29
30pub(crate) struct WindowsDispatcher {
31 pub(crate) wake_posted: AtomicBool,
32 main_sender: PriorityQueueSender<RunnableVariant>,
33 main_thread_id: ThreadId,
34 pub(crate) platform_window_handle: SafeHwnd,
35 validation_number: usize,
36}
37
38impl WindowsDispatcher {
39 pub(crate) fn new(
40 main_sender: PriorityQueueSender<RunnableVariant>,
41 platform_window_handle: HWND,
42 validation_number: usize,
43 ) -> Self {
44 let main_thread_id = current().id();
45 let platform_window_handle = platform_window_handle.into();
46
47 WindowsDispatcher {
48 main_sender,
49 main_thread_id,
50 platform_window_handle,
51 validation_number,
52 wake_posted: AtomicBool::new(false),
53 }
54 }
55
56 fn dispatch_on_threadpool(&self, priority: WorkItemPriority, runnable: RunnableVariant) {
57 let handler = {
58 let mut task_wrapper = Some(runnable);
59 WorkItemHandler::new(move |_| {
60 let runnable = task_wrapper.take().unwrap();
61 // Check if the executor that spawned this task was closed
62 if runnable.metadata().is_closed() {
63 return Ok(());
64 }
65 Self::execute_runnable(runnable);
66 Ok(())
67 })
68 };
69
70 ThreadPool::RunWithPriorityAsync(&handler, priority).log_err();
71 }
72
73 fn dispatch_on_threadpool_after(&self, runnable: RunnableVariant, duration: Duration) {
74 let handler = {
75 let mut task_wrapper = Some(runnable);
76 TimerElapsedHandler::new(move |_| {
77 let runnable = task_wrapper.take().unwrap();
78 // Check if the executor that spawned this task was closed
79 if runnable.metadata().is_closed() {
80 return Ok(());
81 }
82 Self::execute_runnable(runnable);
83 Ok(())
84 })
85 };
86 ThreadPoolTimer::CreateTimer(&handler, duration.into()).log_err();
87 }
88
89 #[inline(always)]
90 pub(crate) fn execute_runnable(runnable: RunnableVariant) {
91 let start = Instant::now();
92
93 let location = runnable.metadata().location;
94 let mut timing = TaskTiming {
95 location,
96 start,
97 end: None,
98 };
99 profiler::add_task_timing(timing);
100
101 runnable.run();
102
103 let end = Instant::now();
104 timing.end = Some(end);
105
106 profiler::add_task_timing(timing);
107 }
108}
109
110impl PlatformDispatcher for WindowsDispatcher {
111 fn get_all_timings(&self) -> Vec<ThreadTaskTimings> {
112 let global_thread_timings = GLOBAL_THREAD_TIMINGS.lock();
113 ThreadTaskTimings::convert(&global_thread_timings)
114 }
115
116 fn get_current_thread_timings(&self) -> Vec<crate::TaskTiming> {
117 THREAD_TIMINGS.with(|timings| {
118 let timings = timings.lock();
119 let timings = &timings.timings;
120
121 let mut vec = Vec::with_capacity(timings.len());
122
123 let (s1, s2) = timings.as_slices();
124 vec.extend_from_slice(s1);
125 vec.extend_from_slice(s2);
126 vec
127 })
128 }
129
130 fn is_main_thread(&self) -> bool {
131 current().id() == self.main_thread_id
132 }
133
134 fn dispatch(&self, runnable: RunnableVariant, priority: Priority) {
135 let priority = match priority {
136 Priority::RealtimeAudio => {
137 panic!("RealtimeAudio priority should use spawn_realtime, not dispatch")
138 }
139 Priority::High => WorkItemPriority::High,
140 Priority::Medium => WorkItemPriority::Normal,
141 Priority::Low => WorkItemPriority::Low,
142 };
143 self.dispatch_on_threadpool(priority, runnable);
144 }
145
146 fn dispatch_on_main_thread(&self, runnable: RunnableVariant, priority: Priority) {
147 match self.main_sender.send(priority, runnable) {
148 Ok(_) => {
149 if !self.wake_posted.swap(true, Ordering::AcqRel) {
150 unsafe {
151 PostMessageW(
152 Some(self.platform_window_handle.as_raw()),
153 WM_GPUI_TASK_DISPATCHED_ON_MAIN_THREAD,
154 WPARAM(self.validation_number),
155 LPARAM(0),
156 )
157 .log_err();
158 }
159 }
160 }
161 Err(runnable) => {
162 // NOTE: Runnable may wrap a Future that is !Send.
163 //
164 // This is usually safe because we only poll it on the main thread.
165 // However if the send fails, we know that:
166 // 1. main_receiver has been dropped (which implies the app is shutting down)
167 // 2. we are on a background thread.
168 // It is not safe to drop something !Send on the wrong thread, and
169 // the app will exit soon anyway, so we must forget the runnable.
170 std::mem::forget(runnable);
171 }
172 }
173 }
174
175 fn dispatch_after(&self, duration: Duration, runnable: RunnableVariant) {
176 self.dispatch_on_threadpool_after(runnable, duration);
177 }
178
179 fn spawn_realtime(&self, f: Box<dyn FnOnce() + Send>) {
180 std::thread::spawn(move || {
181 // SAFETY: always safe to call
182 let thread_handle = unsafe { GetCurrentThread() };
183
184 // SAFETY: thread_handle is a valid handle to a thread
185 unsafe { SetPriorityClass(thread_handle, HIGH_PRIORITY_CLASS) }
186 .context("thread priority class")
187 .log_err();
188
189 // SAFETY: thread_handle is a valid handle to a thread
190 unsafe { SetThreadPriority(thread_handle, THREAD_PRIORITY_TIME_CRITICAL) }
191 .context("thread priority")
192 .log_err();
193
194 f();
195 });
196 }
197
198 fn increase_timer_resolution(&self) -> TimerResolutionGuard {
199 unsafe {
200 timeBeginPeriod(1);
201 }
202 TimerResolutionGuard {
203 cleanup: Some(Box::new(|| unsafe {
204 timeEndPeriod(1);
205 })),
206 }
207 }
208}