oak-team/oak-gpui
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Revert "Multiple priority scheduler" (#44637)

Browse Source 
Reverts zed-industries/zed#44575
This commit is contained in:
Yara 🏳️‍⚧️
2025-12-11 16:16:43 +01:00
committed by GitHub
parent 95dbc0efc2
commit ecb8d3d4dd
24 changed files with 154 additions and 1111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/gpui/src/platform/linux/dispatcher.rs
+45 -284
View File
@@ -1,10 +1,9 @@
use crate::{
GLOBAL_THREAD_TIMINGS, PlatformDispatcher, Priority, PriorityQueueReceiver,
PriorityQueueSender, RealtimePriority, RunnableVariant, THREAD_TIMINGS, TaskLabel, TaskTiming,
ThreadTaskTimings, profiler,
GLOBAL_THREAD_TIMINGS, PlatformDispatcher, RunnableVariant, THREAD_TIMINGS, TaskLabel,
TaskTiming, ThreadTaskTimings,
};
use calloop::{
EventLoop, PostAction,
EventLoop,
channel::{self, Sender},
timer::TimeoutAction,
};
@@ -20,9 +19,9 @@ struct TimerAfter {
}
pub(crate) struct LinuxDispatcher {
main_sender: PriorityQueueCalloopSender<RunnableVariant>,
main_sender: Sender<RunnableVariant>,
timer_sender: Sender<TimerAfter>,
background_sender: PriorityQueueSender<RunnableVariant>,
background_sender: flume::Sender<RunnableVariant>,
_background_threads: Vec<thread::JoinHandle<()>>,
main_thread_id: thread::ThreadId,
}
@@ -30,20 +29,18 @@ pub(crate) struct LinuxDispatcher {
const MIN_THREADS: usize = 2;
impl LinuxDispatcher {
pub fn new(main_sender: PriorityQueueCalloopSender<RunnableVariant>) -> Self {
let (background_sender, background_receiver) = PriorityQueueReceiver::new();
pub fn new(main_sender: Sender<RunnableVariant>) -> Self {
let (background_sender, background_receiver) = flume::unbounded::<RunnableVariant>();
let thread_count =
std::thread::available_parallelism().map_or(MIN_THREADS, |i| i.get().max(MIN_THREADS));
// These thread should really be lower prio then the foreground
// executor
let mut background_threads = (0..thread_count)
.map(|i| {
let mut receiver = background_receiver.clone();
let receiver = background_receiver.clone();
std::thread::Builder::new()
.name(format!("Worker-{i}"))
.spawn(move || {
for runnable in receiver.iter() {
for runnable in receiver {
let start = Instant::now();
let mut location = match runnable {
@@ -54,7 +51,7 @@ impl LinuxDispatcher {
start,
end: None,
};
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
runnable.run();
timing
@@ -66,7 +63,7 @@ impl LinuxDispatcher {
start,
end: None,
};
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
runnable.run();
timing
@@ -75,7 +72,7 @@ impl LinuxDispatcher {
let end = Instant::now();
location.end = Some(end);
profiler::add_task_timing(location);
Self::add_task_timing(location);
log::trace!(
"background thread {}: ran runnable. took: {:?}",
@@ -116,7 +113,7 @@ impl LinuxDispatcher {
start,
end: None,
};
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
runnable.run();
timing
@@ -127,7 +124,7 @@ impl LinuxDispatcher {
start,
end: None,
};
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
runnable.run();
timing
@@ -136,7 +133,7 @@ impl LinuxDispatcher {
let end = Instant::now();
timing.end = Some(end);
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
}
TimeoutAction::Drop
},
@@ -160,6 +157,22 @@ impl LinuxDispatcher {
main_thread_id: thread::current().id(),
}
}
pub(crate) fn add_task_timing(timing: TaskTiming) {
THREAD_TIMINGS.with(|timings| {
let mut timings = timings.lock();
let timings = &mut timings.timings;
if let Some(last_timing) = timings.iter_mut().rev().next() {
if last_timing.location == timing.location {
last_timing.end = timing.end;
return;
}
}
timings.push_back(timing);
});
}
}
impl PlatformDispatcher for LinuxDispatcher {
@@ -186,26 +199,22 @@ impl PlatformDispatcher for LinuxDispatcher {
thread::current().id() == self.main_thread_id
}
fn dispatch(&self, runnable: RunnableVariant, _: Option<TaskLabel>, priority: Priority) {
self.background_sender
.send(priority, runnable)
.unwrap_or_else(|_| panic!("blocking sender returned without value"));
fn dispatch(&self, runnable: RunnableVariant, _: Option<TaskLabel>) {
self.background_sender.send(runnable).unwrap();
}
fn dispatch_on_main_thread(&self, runnable: RunnableVariant, priority: Priority) {
self.main_sender
.send(priority, runnable)
.unwrap_or_else(|runnable| {
// NOTE: Runnable may wrap a Future that is !Send.
//
// This is usually safe because we only poll it on the main thread.
// However if the send fails, we know that:
// 1. main_receiver has been dropped (which implies the app is shutting down)
// 2. we are on a background thread.
// It is not safe to drop something !Send on the wrong thread, and
// the app will exit soon anyway, so we must forget the runnable.
std::mem::forget(runnable);
});
fn dispatch_on_main_thread(&self, runnable: RunnableVariant) {
self.main_sender.send(runnable).unwrap_or_else(|runnable| {
// NOTE: Runnable may wrap a Future that is !Send.
//
// This is usually safe because we only poll it on the main thread.
// However if the send fails, we know that:
// 1. main_receiver has been dropped (which implies the app is shutting down)
// 2. we are on a background thread.
// It is not safe to drop something !Send on the wrong thread, and
// the app will exit soon anyway, so we must forget the runnable.
std::mem::forget(runnable);
});
}
fn dispatch_after(&self, duration: Duration, runnable: RunnableVariant) {
@@ -213,252 +222,4 @@ impl PlatformDispatcher for LinuxDispatcher {
.send(TimerAfter { duration, runnable })
.ok();
}
fn spawn_realtime(&self, priority: RealtimePriority, f: Box<dyn FnOnce() + Send>) {
std::thread::spawn(move || {
// SAFETY: always safe to call
let thread_id = unsafe { libc::pthread_self() };
let policy = match priority {
RealtimePriority::Audio => libc::SCHED_FIFO,
RealtimePriority::Other => libc::SCHED_RR,
};
let sched_priority = match priority {
RealtimePriority::Audio => 65,
RealtimePriority::Other => 45,
};
let sched_param = libc::sched_param { sched_priority };
// SAFETY: sched_param is a valid initialized structure
let result = unsafe { libc::pthread_setschedparam(thread_id, policy, &sched_param) };
if result != 0 {
log::warn!("failed to set realtime thread priority to {:?}", priority);
}
f();
});
}
}
pub struct PriorityQueueCalloopSender<T> {
sender: PriorityQueueSender<T>,
ping: calloop::ping::Ping,
}
impl<T> PriorityQueueCalloopSender<T> {
fn new(tx: PriorityQueueSender<T>, ping: calloop::ping::Ping) -> Self {
Self { sender: tx, ping }
}
fn send(&self, priority: Priority, item: T) -> Result<(), crate::queue::SendError<T>> {
let res = self.sender.send(priority, item);
if res.is_ok() {
self.ping.ping();
}
res
}
}
impl<T> Drop for PriorityQueueCalloopSender<T> {
fn drop(&mut self) {
self.ping.ping();
}
}
pub struct PriorityQueueCalloopReceiver<T> {
receiver: PriorityQueueReceiver<T>,
source: calloop::ping::PingSource,
ping: calloop::ping::Ping,
}
impl<T> PriorityQueueCalloopReceiver<T> {
pub fn new() -> (PriorityQueueCalloopSender<T>, Self) {
let (ping, source) = calloop::ping::make_ping().expect("Failed to create a Ping.");
let (tx, rx) = PriorityQueueReceiver::new();
(
PriorityQueueCalloopSender::new(tx, ping.clone()),
Self {
receiver: rx,
source,
ping,
},
)
}
}
use calloop::channel::Event;
#[derive(Debug)]
pub struct ChannelError(calloop::ping::PingError);
impl std::fmt::Display for ChannelError {
#[cfg_attr(feature = "nightly_coverage", coverage(off))]
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Display::fmt(&self.0, f)
}
}
impl std::error::Error for ChannelError {
#[cfg_attr(feature = "nightly_coverage", coverage(off))]
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(&self.0)
}
}
impl<T> calloop::EventSource for PriorityQueueCalloopReceiver<T> {
type Event = Event<T>;
type Metadata = ();
type Ret = ();
type Error = ChannelError;
fn process_events<F>(
&mut self,
readiness: calloop::Readiness,
token: calloop::Token,
mut callback: F,
) -> Result<calloop::PostAction, Self::Error>
where
F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
{
let mut clear_readiness = false;
let mut disconnected = false;
let action = self
.source
.process_events(readiness, token, |(), &mut ()| {
let mut is_empty = true;
let mut receiver = self.receiver.clone();
for runnable in receiver.try_iter() {
match runnable {
Ok(r) => {
callback(Event::Msg(r), &mut ());
is_empty = false;
}
Err(_) => {
disconnected = true;
}
}
}
if disconnected {
callback(Event::Closed, &mut ());
}
if is_empty {
clear_readiness = true;
}
})
.map_err(ChannelError)?;
if disconnected {
Ok(PostAction::Remove)
} else if clear_readiness {
Ok(action)
} else {
// Re-notify the ping source so we can try again.
self.ping.ping();
Ok(PostAction::Continue)
}
}
fn register(
&mut self,
poll: &mut calloop::Poll,
token_factory: &mut calloop::TokenFactory,
) -> calloop::Result<()> {
self.source.register(poll, token_factory)
}
fn reregister(
&mut self,
poll: &mut calloop::Poll,
token_factory: &mut calloop::TokenFactory,
) -> calloop::Result<()> {
self.source.reregister(poll, token_factory)
}
fn unregister(&mut self, poll: &mut calloop::Poll) -> calloop::Result<()> {
self.source.unregister(poll)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn calloop_works() {
let mut event_loop = calloop::EventLoop::try_new().unwrap();
let handle = event_loop.handle();
let (tx, rx) = PriorityQueueCalloopReceiver::new();
struct Data {
got_msg: bool,
got_closed: bool,
}
let mut data = Data {
got_msg: false,
got_closed: false,
};
let _channel_token = handle
.insert_source(rx, move |evt, &mut (), data: &mut Data| match evt {
Event::Msg(()) => {
data.got_msg = true;
}
Event::Closed => {
data.got_closed = true;
}
})
.unwrap();
// nothing is sent, nothing is received
event_loop
.dispatch(Some(::std::time::Duration::ZERO), &mut data)
.unwrap();
assert!(!data.got_msg);
assert!(!data.got_closed);
// a message is send
tx.send(Priority::Medium, ()).unwrap();
event_loop
.dispatch(Some(::std::time::Duration::ZERO), &mut data)
.unwrap();
assert!(data.got_msg);
assert!(!data.got_closed);
// the sender is dropped
drop(tx);
event_loop
.dispatch(Some(::std::time::Duration::ZERO), &mut data)
.unwrap();
assert!(data.got_msg);
assert!(data.got_closed);
}
}
// running 1 test
// test platform::linux::dispatcher::tests::tomato ... FAILED
// failures:
// ---- platform::linux::dispatcher::tests::tomato stdout ----
// [crates/gpui/src/platform/linux/dispatcher.rs:262:9]
// returning 1 tasks to process
// [crates/gpui/src/platform/linux/dispatcher.rs:480:75] evt = Msg(
// (),
// )
// returning 0 tasks to process
// thread 'platform::linux::dispatcher::tests::tomato' (478301) panicked at crates/gpui/src/platform/linux/dispatcher.rs:515:9:
// assertion failed: data.got_closed
// note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
crates/gpui/src/platform/linux/platform.rs
+5 -5
View File
@@ -14,7 +14,7 @@ use std::{
};
use anyhow::{Context as _, anyhow};
use calloop::LoopSignal;
use calloop::{LoopSignal, channel::Channel};
use futures::channel::oneshot;
use util::ResultExt as _;
use util::command::{new_smol_command, new_std_command};
@@ -25,8 +25,8 @@ use crate::{
Action, AnyWindowHandle, BackgroundExecutor, ClipboardItem, CursorStyle, DisplayId,
ForegroundExecutor, Keymap, LinuxDispatcher, Menu, MenuItem, OwnedMenu, PathPromptOptions,
Pixels, Platform, PlatformDisplay, PlatformKeyboardLayout, PlatformKeyboardMapper,
PlatformTextSystem, PlatformWindow, Point, PriorityQueueCalloopReceiver, Result,
RunnableVariant, Task, WindowAppearance, WindowParams, px,
PlatformTextSystem, PlatformWindow, Point, Result, RunnableVariant, Task, WindowAppearance,
WindowParams, px,
};
#[cfg(any(feature = "wayland", feature = "x11"))]
@@ -149,8 +149,8 @@ pub(crate) struct LinuxCommon {
}
impl LinuxCommon {
pub fn new(signal: LoopSignal) -> (Self, PriorityQueueCalloopReceiver<RunnableVariant>) {
let (main_sender, main_receiver) = PriorityQueueCalloopReceiver::new();
pub fn new(signal: LoopSignal) -> (Self, Channel<RunnableVariant>) {
let (main_sender, main_receiver) = calloop::channel::channel::<RunnableVariant>();
#[cfg(any(feature = "wayland", feature = "x11"))]
let text_system = Arc::new(crate::CosmicTextSystem::new());
crates/gpui/src/platform/linux/wayland/client.rs
+5 -5
View File
@@ -77,10 +77,10 @@ use crate::{
LinuxKeyboardLayout, Modifiers, ModifiersChangedEvent, MouseButton, MouseDownEvent,
MouseExitEvent, MouseMoveEvent, MouseUpEvent, NavigationDirection, Pixels, PlatformDisplay,
PlatformInput, PlatformKeyboardLayout, Point, ResultExt as _, SCROLL_LINES, ScrollDelta,
ScrollWheelEvent, Size, TouchPhase, WindowParams, point, profiler, px, size,
ScrollWheelEvent, Size, TouchPhase, WindowParams, point, px, size,
};
use crate::{
RunnableVariant, TaskTiming,
LinuxDispatcher, RunnableVariant, TaskTiming,
platform::{PlatformWindow, blade::BladeContext},
};
use crate::{
@@ -503,7 +503,7 @@ impl WaylandClient {
start,
end: None,
};
profiler::add_task_timing(timing);
LinuxDispatcher::add_task_timing(timing);
runnable.run();
timing
@@ -515,7 +515,7 @@ impl WaylandClient {
start,
end: None,
};
profiler::add_task_timing(timing);
LinuxDispatcher::add_task_timing(timing);
runnable.run();
timing
@@ -524,7 +524,7 @@ impl WaylandClient {
let end = Instant::now();
timing.end = Some(end);
profiler::add_task_timing(timing);
LinuxDispatcher::add_task_timing(timing);
});
}
}
crates/gpui/src/platform/linux/x11/client.rs
+4 -4
View File
@@ -1,4 +1,4 @@
use crate::{Capslock, ResultExt as _, RunnableVariant, TaskTiming, profiler, xcb_flush};
use crate::{Capslock, LinuxDispatcher, ResultExt as _, RunnableVariant, TaskTiming, xcb_flush};
use anyhow::{Context as _, anyhow};
use ashpd::WindowIdentifier;
use calloop::{
@@ -322,7 +322,7 @@ impl X11Client {
start,
end: None,
};
profiler::add_task_timing(timing);
LinuxDispatcher::add_task_timing(timing);
runnable.run();
timing
@@ -334,7 +334,7 @@ impl X11Client {
start,
end: None,
};
profiler::add_task_timing(timing);
LinuxDispatcher::add_task_timing(timing);
runnable.run();
timing
@@ -343,7 +343,7 @@ impl X11Client {
let end = Instant::now();
timing.end = Some(end);
profiler::add_task_timing(timing);
LinuxDispatcher::add_task_timing(timing);
});
}
}
crates/gpui/src/platform/mac/dispatcher.rs
+5 -140
View File
@@ -3,22 +3,11 @@
#![allow(non_snake_case)]
use crate::{
GLOBAL_THREAD_TIMINGS, PlatformDispatcher, Priority, RealtimePriority, RunnableMeta,
RunnableVariant, THREAD_TIMINGS, TaskLabel, TaskTiming, ThreadTaskTimings,
GLOBAL_THREAD_TIMINGS, PlatformDispatcher, RunnableMeta, RunnableVariant, THREAD_TIMINGS,
TaskLabel, TaskTiming, ThreadTaskTimings,
};
use anyhow::Context;
use async_task::Runnable;
use mach2::{
kern_return::KERN_SUCCESS,
mach_time::mach_timebase_info_data_t,
thread_policy::{
THREAD_EXTENDED_POLICY, THREAD_EXTENDED_POLICY_COUNT, THREAD_PRECEDENCE_POLICY,
THREAD_PRECEDENCE_POLICY_COUNT, THREAD_TIME_CONSTRAINT_POLICY,
THREAD_TIME_CONSTRAINT_POLICY_COUNT, thread_extended_policy_data_t,
thread_precedence_policy_data_t, thread_time_constraint_policy_data_t,
},
};
use objc::{
class, msg_send,
runtime::{BOOL, YES},
@@ -26,11 +15,9 @@ use objc::{
};
use std::{
ffi::c_void,
mem::MaybeUninit,
ptr::{NonNull, addr_of},
time::{Duration, Instant},
};
use util::ResultExt;
/// All items in the generated file are marked as pub, so we're gonna wrap it in a separate mod to prevent
/// these pub items from leaking into public API.
@@ -69,7 +56,7 @@ impl PlatformDispatcher for MacDispatcher {
is_main_thread == YES
}
fn dispatch(&self, runnable: RunnableVariant, _: Option<TaskLabel>, priority: Priority) {
fn dispatch(&self, runnable: RunnableVariant, _: Option<TaskLabel>) {
let (context, trampoline) = match runnable {
RunnableVariant::Meta(runnable) => (
runnable.into_raw().as_ptr() as *mut c_void,
@@ -80,24 +67,16 @@ impl PlatformDispatcher for MacDispatcher {
Some(trampoline_compat as unsafe extern "C" fn(*mut c_void)),
),
};
let queue_priority = match priority {
Priority::Realtime(_) => unreachable!(),
Priority::High => DISPATCH_QUEUE_PRIORITY_HIGH,
Priority::Medium => DISPATCH_QUEUE_PRIORITY_DEFAULT,
Priority::Low => DISPATCH_QUEUE_PRIORITY_LOW as u32,
};
unsafe {
dispatch_async_f(
dispatch_get_global_queue(queue_priority.try_into().unwrap(), 0),
dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH.try_into().unwrap(), 0),
context,
trampoline,
);
}
}
fn dispatch_on_main_thread(&self, runnable: RunnableVariant, _priority: Priority) {
fn dispatch_on_main_thread(&self, runnable: RunnableVariant) {
let (context, trampoline) = match runnable {
RunnableVariant::Meta(runnable) => (
runnable.into_raw().as_ptr() as *mut c_void,
@@ -131,120 +110,6 @@ impl PlatformDispatcher for MacDispatcher {
dispatch_after_f(when, queue, context, trampoline);
}
}
fn spawn_realtime(&self, priority: RealtimePriority, f: Box<dyn FnOnce() + Send>) {
std::thread::spawn(move || {
match priority {
RealtimePriority::Audio => set_audio_thread_priority(),
RealtimePriority::Other => set_high_thread_priority(),
}
.context(format!("for priority {:?}", priority))
.log_err();
f();
});
}
}
fn set_high_thread_priority() -> anyhow::Result<()> {
// SAFETY: always safe to call
let thread_id = unsafe { libc::pthread_self() };
// SAFETY: all sched_param members are valid when initialized to zero.
let mut sched_param = unsafe { MaybeUninit::<libc::sched_param>::zeroed().assume_init() };
sched_param.sched_priority = 45;
let result = unsafe { libc::pthread_setschedparam(thread_id, libc::SCHED_FIFO, &sched_param) };
if result != 0 {
anyhow::bail!("failed to set realtime thread priority")
}
Ok(())
}
fn set_audio_thread_priority() -> anyhow::Result<()> {
// https://chromium.googlesource.com/chromium/chromium/+/master/base/threading/platform_thread_mac.mm#93
// SAFETY: always safe to call
let thread_id = unsafe { libc::pthread_self() };
// SAFETY: thread_id is a valid thread id
let thread_id = unsafe { libc::pthread_mach_thread_np(thread_id) };
// Fixed priority thread
let mut policy = thread_extended_policy_data_t { timeshare: 0 };
// SAFETY: thread_id is a valid thread id
// SAFETY: thread_extended_policy_data_t is passed as THREAD_EXTENDED_POLICY
let result = unsafe {
mach2::thread_policy::thread_policy_set(
thread_id,
THREAD_EXTENDED_POLICY,
&mut policy as *mut _ as *mut _,
THREAD_EXTENDED_POLICY_COUNT,
)
};
if result != KERN_SUCCESS {
anyhow::bail!("failed to set thread extended policy");
}
// relatively high priority
let mut precedence = thread_precedence_policy_data_t { importance: 63 };
// SAFETY: thread_id is a valid thread id
// SAFETY: thread_precedence_policy_data_t is passed as THREAD_PRECEDENCE_POLICY
let result = unsafe {
mach2::thread_policy::thread_policy_set(
thread_id,
THREAD_PRECEDENCE_POLICY,
&mut precedence as *mut _ as *mut _,
THREAD_PRECEDENCE_POLICY_COUNT,
)
};
if result != KERN_SUCCESS {
anyhow::bail!("failed to set thread precedence policy");
}
const GUARANTEED_AUDIO_DUTY_CYCLE: f32 = 0.75;
const MAX_AUDIO_DUTY_CYCLE: f32 = 0.85;
// ~128 frames @ 44.1KHz
const TIME_QUANTUM: f32 = 2.9;
const AUDIO_TIME_NEEDED: f32 = GUARANTEED_AUDIO_DUTY_CYCLE * TIME_QUANTUM;
const MAX_TIME_ALLOWED: f32 = MAX_AUDIO_DUTY_CYCLE * TIME_QUANTUM;
let mut timebase_info = mach_timebase_info_data_t { numer: 0, denom: 0 };
// SAFETY: timebase_info is a valid pointer to a mach_timebase_info_data_t struct
unsafe { mach2::mach_time::mach_timebase_info(&mut timebase_info) };
let ms_to_abs_time = ((timebase_info.denom as f32) / (timebase_info.numer as f32)) * 1000000f32;
let mut time_constraints = thread_time_constraint_policy_data_t {
period: (TIME_QUANTUM * ms_to_abs_time) as u32,
computation: (AUDIO_TIME_NEEDED * ms_to_abs_time) as u32,
constraint: (MAX_TIME_ALLOWED * ms_to_abs_time) as u32,
preemptible: 0,
};
// SAFETY: thread_id is a valid thread id
// SAFETY: thread_precedence_pthread_time_constraint_policy_data_t is passed as THREAD_TIME_CONSTRAINT_POLICY
let result = unsafe {
mach2::thread_policy::thread_policy_set(
thread_id,
THREAD_TIME_CONSTRAINT_POLICY,
&mut time_constraints as *mut _ as *mut _,
THREAD_TIME_CONSTRAINT_POLICY_COUNT,
)
};
if result != KERN_SUCCESS {
anyhow::bail!("failed to set thread time constraint policy");
}
Ok(())
}
extern "C" fn trampoline(runnable: *mut c_void) {
crates/gpui/src/platform/test/dispatcher.rs
+3 -9
View File
@@ -1,4 +1,4 @@
use crate::{PlatformDispatcher, Priority, RunnableVariant, TaskLabel};
use crate::{PlatformDispatcher, RunnableVariant, TaskLabel};
use backtrace::Backtrace;
use collections::{HashMap, HashSet, VecDeque};
use parking::Unparker;
@@ -284,7 +284,7 @@ impl PlatformDispatcher for TestDispatcher {
state.start_time + state.time
}
fn dispatch(&self, runnable: RunnableVariant, label: Option<TaskLabel>, _priority: Priority) {
fn dispatch(&self, runnable: RunnableVariant, label: Option<TaskLabel>) {
{
let mut state = self.state.lock();
if label.is_some_and(|label| state.deprioritized_task_labels.contains(&label)) {
@@ -296,7 +296,7 @@ impl PlatformDispatcher for TestDispatcher {
self.unpark_all();
}
fn dispatch_on_main_thread(&self, runnable: RunnableVariant, _priority: Priority) {
fn dispatch_on_main_thread(&self, runnable: RunnableVariant) {
self.state
.lock()
.foreground
@@ -318,10 +318,4 @@ impl PlatformDispatcher for TestDispatcher {
fn as_test(&self) -> Option<&TestDispatcher> {
Some(self)
}
fn spawn_realtime(&self, _priority: crate::RealtimePriority, f: Box<dyn FnOnce() + Send>) {
std::thread::spawn(move || {
f();
});
}
}
crates/gpui/src/platform/windows/dispatcher.rs
+31 -54
View File
@@ -4,31 +4,24 @@ use std::{
time::{Duration, Instant},
};
use anyhow::Context;
use flume::Sender;
use util::ResultExt;
use windows::{
System::Threading::{
ThreadPool, ThreadPoolTimer, TimerElapsedHandler, WorkItemHandler, WorkItemPriority,
},
System::Threading::{ThreadPool, ThreadPoolTimer, TimerElapsedHandler, WorkItemHandler},
Win32::{
Foundation::{LPARAM, WPARAM},
System::Threading::{
GetCurrentThread, HIGH_PRIORITY_CLASS, SetPriorityClass, SetThreadPriority,
THREAD_PRIORITY_HIGHEST, THREAD_PRIORITY_TIME_CRITICAL,
},
UI::WindowsAndMessaging::PostMessageW,
},
};
use crate::{
GLOBAL_THREAD_TIMINGS, HWND, PlatformDispatcher, Priority, PriorityQueueSender,
RealtimePriority, RunnableVariant, SafeHwnd, THREAD_TIMINGS, TaskLabel, TaskTiming,
ThreadTaskTimings, WM_GPUI_TASK_DISPATCHED_ON_MAIN_THREAD, profiler,
GLOBAL_THREAD_TIMINGS, HWND, PlatformDispatcher, RunnableVariant, SafeHwnd, THREAD_TIMINGS,
TaskLabel, TaskTiming, ThreadTaskTimings, WM_GPUI_TASK_DISPATCHED_ON_MAIN_THREAD,
};
pub(crate) struct WindowsDispatcher {
pub(crate) wake_posted: AtomicBool,
main_sender: PriorityQueueSender<RunnableVariant>,
main_sender: Sender<RunnableVariant>,
main_thread_id: ThreadId,
pub(crate) platform_window_handle: SafeHwnd,
validation_number: usize,
@@ -36,7 +29,7 @@ pub(crate) struct WindowsDispatcher {
impl WindowsDispatcher {
pub(crate) fn new(
main_sender: PriorityQueueSender<RunnableVariant>,
main_sender: Sender<RunnableVariant>,
platform_window_handle: HWND,
validation_number: usize,
) -> Self {
@@ -52,7 +45,7 @@ impl WindowsDispatcher {
}
}
fn dispatch_on_threadpool(&self, priority: WorkItemPriority, runnable: RunnableVariant) {
fn dispatch_on_threadpool(&self, runnable: RunnableVariant) {
let handler = {
let mut task_wrapper = Some(runnable);
WorkItemHandler::new(move |_| {
@@ -60,8 +53,7 @@ impl WindowsDispatcher {
Ok(())
})
};
ThreadPool::RunWithPriorityAsync(&handler, priority).log_err();
ThreadPool::RunAsync(&handler).log_err();
}
fn dispatch_on_threadpool_after(&self, runnable: RunnableVariant, duration: Duration) {
@@ -87,7 +79,7 @@ impl WindowsDispatcher {
start,
end: None,
};
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
runnable.run();
@@ -99,7 +91,7 @@ impl WindowsDispatcher {
start,
end: None,
};
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
runnable.run();
@@ -110,7 +102,23 @@ impl WindowsDispatcher {
let end = Instant::now();
timing.end = Some(end);
profiler::add_task_timing(timing);
Self::add_task_timing(timing);
}
pub(crate) fn add_task_timing(timing: TaskTiming) {
THREAD_TIMINGS.with(|timings| {
let mut timings = timings.lock();
let timings = &mut timings.timings;
if let Some(last_timing) = timings.iter_mut().rev().next() {
if last_timing.location == timing.location {
last_timing.end = timing.end;
return;
}
}
timings.push_back(timing);
});
}
}
@@ -138,22 +146,15 @@ impl PlatformDispatcher for WindowsDispatcher {
current().id() == self.main_thread_id
}
fn dispatch(&self, runnable: RunnableVariant, label: Option<TaskLabel>, priority: Priority) {
let priority = match priority {
Priority::Realtime(_) => unreachable!(),
Priority::High => WorkItemPriority::High,
Priority::Medium => WorkItemPriority::Normal,
Priority::Low => WorkItemPriority::Low,
};
self.dispatch_on_threadpool(priority, runnable);
fn dispatch(&self, runnable: RunnableVariant, label: Option<TaskLabel>) {
self.dispatch_on_threadpool(runnable);
if let Some(label) = label {
log::debug!("TaskLabel: {label:?}");
}
}
fn dispatch_on_main_thread(&self, runnable: RunnableVariant, priority: Priority) {
match self.main_sender.send(priority, runnable) {
fn dispatch_on_main_thread(&self, runnable: RunnableVariant) {
match self.main_sender.send(runnable) {
Ok(_) => {
if !self.wake_posted.swap(true, Ordering::AcqRel) {
unsafe {
@@ -184,28 +185,4 @@ impl PlatformDispatcher for WindowsDispatcher {
fn dispatch_after(&self, duration: Duration, runnable: RunnableVariant) {
self.dispatch_on_threadpool_after(runnable, duration);
}
fn spawn_realtime(&self, priority: RealtimePriority, f: Box<dyn FnOnce() + Send>) {
std::thread::spawn(move || {
// SAFETY: always safe to call
let thread_handle = unsafe { GetCurrentThread() };
let thread_priority = match priority {
RealtimePriority::Audio => THREAD_PRIORITY_TIME_CRITICAL,
RealtimePriority::Other => THREAD_PRIORITY_HIGHEST,
};
// SAFETY: thread_handle is a valid handle to a thread
unsafe { SetPriorityClass(thread_handle, HIGH_PRIORITY_CLASS) }
.context("thread priority class")
.log_err();
// SAFETY: thread_handle is a valid handle to a thread
unsafe { SetThreadPriority(thread_handle, thread_priority) }
.context("thread priority")
.log_err();
f();
});
}
}
crates/gpui/src/platform/windows/events.rs
+1 -2
View File
@@ -243,8 +243,7 @@ impl WindowsWindowInner {
fn handle_timer_msg(&self, handle: HWND, wparam: WPARAM) -> Option<isize> {
if wparam.0 == SIZE_MOVE_LOOP_TIMER_ID {
let mut runnables = self.main_receiver.clone().try_iter();
while let Some(Ok(runnable)) = runnables.next() {
for runnable in self.main_receiver.drain() {
WindowsDispatcher::execute_runnable(runnable);
}
self.handle_paint_msg(handle)
crates/gpui/src/platform/windows/platform.rs
+11 -13
View File
@@ -51,7 +51,7 @@ struct WindowsPlatformInner {
raw_window_handles: std::sync::Weak<RwLock<SmallVec<[SafeHwnd; 4]>>>,
// The below members will never change throughout the entire lifecycle of the app.
validation_number: usize,
main_receiver: PriorityQueueReceiver<RunnableVariant>,
main_receiver: flume::Receiver<RunnableVariant>,
dispatcher: Arc<WindowsDispatcher>,
}
@@ -98,7 +98,7 @@ impl WindowsPlatform {
OleInitialize(None).context("unable to initialize Windows OLE")?;
}
let directx_devices = DirectXDevices::new().context("Creating DirectX devices")?;
let (main_sender, main_receiver) = PriorityQueueReceiver::new();
let (main_sender, main_receiver) = flume::unbounded::<RunnableVariant>();
let validation_number = if usize::BITS == 64 {
rand::random::<u64>() as usize
} else {
@@ -857,24 +857,22 @@ impl WindowsPlatformInner {
}
break 'tasks;
}
let mut main_receiver = self.main_receiver.clone();
match main_receiver.try_pop() {
Ok(Some(runnable)) => WindowsDispatcher::execute_runnable(runnable),
_ => break 'timeout_loop,
match self.main_receiver.try_recv() {
Err(_) => break 'timeout_loop,
Ok(runnable) => WindowsDispatcher::execute_runnable(runnable),
}
}
// Someone could enqueue a Runnable here. The flag is still true, so they will not PostMessage.
// We need to check for those Runnables after we clear the flag.
self.dispatcher.wake_posted.store(false, Ordering::Release);
let mut main_receiver = self.main_receiver.clone();
match main_receiver.try_pop() {
Ok(Some(runnable)) => {
match self.main_receiver.try_recv() {
Err(_) => break 'tasks,
Ok(runnable) => {
self.dispatcher.wake_posted.store(true, Ordering::Release);
WindowsDispatcher::execute_runnable(runnable);
}
_ => break 'tasks,
}
}
@@ -936,7 +934,7 @@ pub(crate) struct WindowCreationInfo {
pub(crate) windows_version: WindowsVersion,
pub(crate) drop_target_helper: IDropTargetHelper,
pub(crate) validation_number: usize,
pub(crate) main_receiver: PriorityQueueReceiver<RunnableVariant>,
pub(crate) main_receiver: flume::Receiver<RunnableVariant>,
pub(crate) platform_window_handle: HWND,
pub(crate) disable_direct_composition: bool,
pub(crate) directx_devices: DirectXDevices,
@@ -949,8 +947,8 @@ struct PlatformWindowCreateContext {
inner: Option<Result<Rc<WindowsPlatformInner>>>,
raw_window_handles: std::sync::Weak<RwLock<SmallVec<[SafeHwnd; 4]>>>,
validation_number: usize,
main_sender: Option<PriorityQueueSender<RunnableVariant>>,
main_receiver: Option<PriorityQueueReceiver<RunnableVariant>>,
main_sender: Option<flume::Sender<RunnableVariant>>,
main_receiver: Option<flume::Receiver<RunnableVariant>>,
directx_devices: Option<DirectXDevices>,
dispatcher: Option<Arc<WindowsDispatcher>>,
}
crates/gpui/src/platform/windows/window.rs
+2 -2
View File
@@ -81,7 +81,7 @@ pub(crate) struct WindowsWindowInner {
pub(crate) executor: ForegroundExecutor,
pub(crate) windows_version: WindowsVersion,
pub(crate) validation_number: usize,
pub(crate) main_receiver: PriorityQueueReceiver<RunnableVariant>,
pub(crate) main_receiver: flume::Receiver<RunnableVariant>,
pub(crate) platform_window_handle: HWND,
}
@@ -362,7 +362,7 @@ struct WindowCreateContext {
windows_version: WindowsVersion,
drop_target_helper: IDropTargetHelper,
validation_number: usize,
main_receiver: PriorityQueueReceiver<RunnableVariant>,
main_receiver: flume::Receiver<RunnableVariant>,
platform_window_handle: HWND,
appearance: WindowAppearance,
disable_direct_composition: bool,