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::{HWND, SafeHwnd, WM_GPUI_TASK_DISPATCHED_ON_MAIN_THREAD};
25use gpui::{
26 GLOBAL_THREAD_TIMINGS, PlatformDispatcher, Priority, PriorityQueueSender, RunnableVariant,
27 THREAD_TIMINGS, TaskTiming, ThreadTaskTimings, TimerResolutionGuard,
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 Self::execute_runnable(runnable);
62 Ok(())
63 })
64 };
65
66 ThreadPool::RunWithPriorityAsync(&handler, priority).log_err();
67 }
68
69 fn dispatch_on_threadpool_after(&self, runnable: RunnableVariant, duration: Duration) {
70 let handler = {
71 let mut task_wrapper = Some(runnable);
72 TimerElapsedHandler::new(move |_| {
73 let runnable = task_wrapper.take().unwrap();
74 Self::execute_runnable(runnable);
75 Ok(())
76 })
77 };
78 ThreadPoolTimer::CreateTimer(&handler, duration.into()).log_err();
79 }
80
81 #[inline(always)]
82 pub(crate) fn execute_runnable(runnable: RunnableVariant) {
83 let start = Instant::now();
84
85 let location = runnable.metadata().location;
86 let mut timing = TaskTiming {
87 location,
88 start,
89 end: None,
90 };
91 gpui::profiler::add_task_timing(timing);
92
93 runnable.run();
94
95 let end = Instant::now();
96 timing.end = Some(end);
97
98 gpui::profiler::add_task_timing(timing);
99 }
100}
101
102impl PlatformDispatcher for WindowsDispatcher {
103 fn get_all_timings(&self) -> Vec<ThreadTaskTimings> {
104 let global_thread_timings = GLOBAL_THREAD_TIMINGS.lock();
105 ThreadTaskTimings::convert(&global_thread_timings)
106 }
107
108 fn get_current_thread_timings(&self) -> gpui::ThreadTaskTimings {
109 THREAD_TIMINGS.with(|timings| {
110 let timings = timings.lock();
111 let thread_name = timings.thread_name.clone();
112 let total_pushed = timings.total_pushed;
113 let timings = &timings.timings;
114
115 let mut vec = Vec::with_capacity(timings.len());
116
117 let (s1, s2) = timings.as_slices();
118 vec.extend_from_slice(s1);
119 vec.extend_from_slice(s2);
120
121 gpui::ThreadTaskTimings {
122 thread_name,
123 thread_id: std::thread::current().id(),
124 timings: vec,
125 total_pushed,
126 }
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 util::defer(Box::new(|| unsafe {
203 timeEndPeriod(1);
204 }))
205 }
206}