gpui: Expose threaded dispatcher for tests (#62096)
Some tests need real OS threading and timers rather than the deterministic, single-threaded behavior of `TestDispatcher`. The existing multithreaded dispatcher provides those production-like scheduling semantics, but it is named and gated as a benchmark-only facility, which unnecessarily pulls in Criterion and histogram dependencies. This renames `BenchDispatcher` to `ThreadedDispatcher` and exposes it through GPUI's `test-support` feature. Benchmark contexts continue using the same dispatcher, while Criterion and benchmark reporting remain isolated behind the `bench` feature. ### Testing - `cargo check -p gpui --features test-support` - `cargo check -p gpui --features bench` - `cargo nextest run -p gpui 'threaded_dispatcher::tests'` - `cargo fmt --all --check` Release Notes: - N/A
This commit is contained in:
1 parent
2758986c73
commit
077311366d
4 files changed
+52
-53
No files matched your search
@@ -10,17 +10,17 @@ use anyhow::{Result, anyhow};
|
||||
use hdrhistogram::Histogram;
|
||||
|
||||
use crate::{
|
||||
AnyView, AnyWindowHandle, App, AppCell, AppContext, BackgroundExecutor, BenchDispatcher,
|
||||
Bounds, Context, Empty, Entity, EntityId, Focusable, ForegroundExecutor, Global, Platform,
|
||||
PlatformHeadlessRenderer, PlatformTextSystem, Render, Reservation, Task, TestPlatform,
|
||||
VisualContext, Window, WindowBounds, WindowHandle, WindowOptions,
|
||||
AnyView, AnyWindowHandle, App, AppCell, AppContext, BackgroundExecutor, Bounds, Context, Empty,
|
||||
Entity, EntityId, Focusable, ForegroundExecutor, Global, Platform, PlatformHeadlessRenderer,
|
||||
PlatformTextSystem, Render, Reservation, Task, TestPlatform, ThreadedDispatcher, VisualContext,
|
||||
Window, WindowBounds, WindowHandle, WindowOptions,
|
||||
app::GpuiBorrow,
|
||||
profiler::{self, FrameTiming, FrameTimingCollector},
|
||||
};
|
||||
|
||||
/// Returns a benchmark platform backed by this thread's shared dispatcher.
|
||||
///
|
||||
/// The platform uses this thread's shared multithreaded [`BenchDispatcher`], so
|
||||
/// The platform uses this thread's shared multithreaded [`ThreadedDispatcher`], so
|
||||
/// background work runs with production concurrency in real time. The dispatcher
|
||||
/// is cached per thread and reused across benchmark invocations so worker and
|
||||
/// timer threads persist for the whole process instead of being recreated for
|
||||
@@ -42,10 +42,10 @@ pub fn bench_platform(
|
||||
text_system: Arc<dyn PlatformTextSystem>,
|
||||
) -> Rc<dyn Platform> {
|
||||
thread_local! {
|
||||
static DISPATCHER: OnceCell<Arc<BenchDispatcher>> = const { OnceCell::new() };
|
||||
static DISPATCHER: OnceCell<Arc<ThreadedDispatcher>> = const { OnceCell::new() };
|
||||
}
|
||||
let dispatcher = DISPATCHER.with(|cell| {
|
||||
cell.get_or_init(|| Arc::new(BenchDispatcher::new()))
|
||||
cell.get_or_init(|| Arc::new(ThreadedDispatcher::new()))
|
||||
.clone()
|
||||
});
|
||||
let background_executor = BackgroundExecutor::new(dispatcher.clone());
|
||||
@@ -276,7 +276,7 @@ pub struct BenchAppContext<'a, 'measurement> {
|
||||
impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
|
||||
/// Creates a new benchmark app context backed by the provided platform.
|
||||
///
|
||||
/// The platform's executors must be backed by a [`BenchDispatcher`]
|
||||
/// The platform's executors must be backed by a [`ThreadedDispatcher`]
|
||||
/// (see [`bench_platform`]) so the context can drain foreground work via
|
||||
/// [`Self::run_until_idle`]; panics otherwise.
|
||||
pub fn new(
|
||||
@@ -289,7 +289,7 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
|
||||
|
||||
/// Creates a new benchmark app context backed by the provided platform.
|
||||
///
|
||||
/// The platform's executors must be backed by a [`BenchDispatcher`]
|
||||
/// The platform's executors must be backed by a [`ThreadedDispatcher`]
|
||||
/// (see [`bench_platform`]) so the context can drain foreground work via
|
||||
/// [`Self::run_until_idle`]; panics otherwise.
|
||||
#[doc(hidden)]
|
||||
@@ -312,9 +312,9 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
|
||||
// Validate up front so misconfiguration fails at construction with a
|
||||
// clear message instead of deep inside `run_until_idle`.
|
||||
assert!(
|
||||
background_executor.dispatcher().as_bench().is_some(),
|
||||
background_executor.dispatcher().as_threaded().is_some(),
|
||||
"BenchAppContext requires a platform whose executors are backed by a \
|
||||
BenchDispatcher; construct one with gpui::bench_platform"
|
||||
ThreadedDispatcher; construct one with gpui::bench_platform"
|
||||
);
|
||||
let foreground_executor = platform.foreground_executor();
|
||||
let asset_source = Arc::new(());
|
||||
@@ -360,11 +360,11 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
|
||||
|
||||
/// Runs queued foreground tasks on this thread and waits for in flight
|
||||
/// background work to finish. Timers that aren't due yet are not waited
|
||||
/// for (see [`BenchDispatcher::run_until_idle`]).
|
||||
/// for (see [`ThreadedDispatcher::run_until_idle`]).
|
||||
pub fn run_until_idle(&self) {
|
||||
self.background_executor
|
||||
.dispatcher()
|
||||
.as_bench()
|
||||
.as_threaded()
|
||||
.expect("validated in BenchAppContext::build")
|
||||
.run_until_idle();
|
||||
}
|
||||
@@ -414,7 +414,7 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
|
||||
|
||||
let mut benchmark = || {
|
||||
dispatcher
|
||||
.as_bench()
|
||||
.as_threaded()
|
||||
.expect("validated in BenchAppContext::build")
|
||||
.run_ready_main_tasks();
|
||||
self.with_window(view.entity_id(), |window, cx| {
|
||||
@@ -496,7 +496,7 @@ impl<'a, 'measurement> BenchAppContext<'a, 'measurement> {
|
||||
|
||||
let dispatcher = self.background_executor.dispatcher();
|
||||
let dispatcher = dispatcher
|
||||
.as_bench()
|
||||
.as_threaded()
|
||||
.expect("validated in BenchAppContext::build");
|
||||
|
||||
drop(self.app);
|
||||
|
||||
@@ -41,7 +41,7 @@ pub mod profiler;
|
||||
target_os = "windows",
|
||||
target_os = "linux",
|
||||
target_family = "wasm",
|
||||
feature = "bench"
|
||||
feature = "test-support"
|
||||
))]
|
||||
#[expect(missing_docs)]
|
||||
pub mod queue;
|
||||
|
||||
@@ -9,8 +9,8 @@ pub mod layer_shell;
|
||||
/// Types for configuring parent-anchored popup windows such as menus, dropdowns and tooltips.
|
||||
pub mod popup;
|
||||
|
||||
#[cfg(any(test, feature = "bench"))]
|
||||
mod bench_dispatcher;
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
mod threaded_dispatcher;
|
||||
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
mod test;
|
||||
@@ -83,8 +83,8 @@ pub(crate) use test::*;
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
pub use test::{TestDispatcher, TestScreenCaptureSource, TestScreenCaptureStream};
|
||||
|
||||
#[cfg(any(test, feature = "bench"))]
|
||||
pub use bench_dispatcher::BenchDispatcher;
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
pub use threaded_dispatcher::ThreadedDispatcher;
|
||||
|
||||
#[cfg(all(target_os = "macos", any(test, feature = "test-support")))]
|
||||
pub use visual_test::VisualTestPlatform;
|
||||
@@ -1041,10 +1041,10 @@ pub trait PlatformDispatcher: Send + Sync {
|
||||
None
|
||||
}
|
||||
|
||||
// This cfg must match the `bench_dispatcher` module's, which implements
|
||||
// This cfg must match the `threaded_dispatcher` module's, which implements
|
||||
// this method whenever it compiles.
|
||||
#[cfg(any(test, feature = "bench"))]
|
||||
fn as_bench(&self) -> Option<&BenchDispatcher> {
|
||||
#[cfg(any(test, feature = "test-support"))]
|
||||
fn as_threaded(&self) -> Option<&ThreadedDispatcher> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
+29
-30
@@ -14,19 +14,18 @@ use crate::{
|
||||
|
||||
const MIN_THREADS: usize = 2;
|
||||
|
||||
/// A multithreaded [`PlatformDispatcher`] for benchmarks.
|
||||
/// A multithreaded [`PlatformDispatcher`] for tests and benchmarks.
|
||||
///
|
||||
/// Background tasks run in parallel on a pool of worker threads and timers fire
|
||||
/// in real time on a dedicated timer thread, mirroring the production
|
||||
/// dispatchers (see `LinuxDispatcher`). Main-thread tasks are queued until the
|
||||
/// benchmark thread drains them via [`Self::run_until_idle`], since there is no
|
||||
/// creating thread drains them via [`Self::run_until_idle`], since there is no
|
||||
/// platform run loop pumping them.
|
||||
///
|
||||
/// Unlike [`TestDispatcher`](crate::TestDispatcher), which runs everything on a
|
||||
/// single thread with a virtual clock, work dispatched through this dispatcher
|
||||
/// executes with production concurrency, so wall-clock measurements reflect
|
||||
/// real parallelism.
|
||||
pub struct BenchDispatcher {
|
||||
/// executes with production concurrency.
|
||||
pub struct ThreadedDispatcher {
|
||||
background_sender: PriorityQueueSender<RunnableVariant>,
|
||||
main_sender: PriorityQueueSender<RunnableVariant>,
|
||||
main_receiver: Mutex<PriorityQueueReceiver<RunnableVariant>>,
|
||||
@@ -36,7 +35,7 @@ pub struct BenchDispatcher {
|
||||
}
|
||||
|
||||
/// Tracks how many background and timer runnables are queued or running so
|
||||
/// [`BenchDispatcher::run_until_idle`] knows when to stop waiting.
|
||||
/// [`ThreadedDispatcher::run_until_idle`] knows when to stop waiting.
|
||||
#[derive(Default)]
|
||||
struct IdleTracker {
|
||||
inflight: Mutex<usize>,
|
||||
@@ -112,17 +111,17 @@ impl Ord for TimerEntry {
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for BenchDispatcher {
|
||||
impl Default for ThreadedDispatcher {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl BenchDispatcher {
|
||||
impl ThreadedDispatcher {
|
||||
/// Creates a dispatcher whose main thread is the calling thread.
|
||||
///
|
||||
/// Worker and timer threads live for the lifetime of the process; the
|
||||
/// dispatcher is expected to be created once and reused across benchmarks.
|
||||
/// dispatcher is expected to be created once and reused.
|
||||
pub fn new() -> Self {
|
||||
let (background_sender, background_receiver) = PriorityQueueReceiver::new();
|
||||
let (main_sender, main_receiver) = PriorityQueueReceiver::new();
|
||||
@@ -134,7 +133,7 @@ impl BenchDispatcher {
|
||||
let mut receiver: PriorityQueueReceiver<RunnableVariant> = background_receiver.clone();
|
||||
let idle = idle.clone();
|
||||
thread::Builder::new()
|
||||
.name(format!("BenchWorker-{i}"))
|
||||
.name(format!("ThreadedDispatcherWorker-{i}"))
|
||||
.spawn(move || {
|
||||
while let Ok(runnable) = receiver.pop() {
|
||||
let _decrement = idle.decrement_on_drop();
|
||||
@@ -145,7 +144,7 @@ impl BenchDispatcher {
|
||||
profiler::save_task_timing();
|
||||
}
|
||||
})
|
||||
.expect("failed to spawn benchmark worker thread");
|
||||
.expect("failed to spawn threaded dispatcher worker");
|
||||
}
|
||||
drop(background_receiver);
|
||||
|
||||
@@ -160,7 +159,7 @@ impl BenchDispatcher {
|
||||
let timers = timers.clone();
|
||||
let idle = idle.clone();
|
||||
thread::Builder::new()
|
||||
.name("BenchTimer".to_owned())
|
||||
.name("ThreadedDispatcherTimer".to_owned())
|
||||
.spawn(move || {
|
||||
let mut state = timers.state.lock();
|
||||
loop {
|
||||
@@ -196,7 +195,7 @@ impl BenchDispatcher {
|
||||
state = timers.state.lock();
|
||||
}
|
||||
})
|
||||
.expect("failed to spawn benchmark timer thread");
|
||||
.expect("failed to spawn threaded dispatcher timer");
|
||||
}
|
||||
|
||||
Self {
|
||||
@@ -221,7 +220,7 @@ impl BenchDispatcher {
|
||||
pub fn run_until_idle(&self) {
|
||||
assert!(
|
||||
self.is_main_thread(),
|
||||
"run_until_idle must be called on the benchmark main thread"
|
||||
"run_until_idle must be called on the threaded dispatcher's main thread"
|
||||
);
|
||||
loop {
|
||||
if self.drain_main_queue() {
|
||||
@@ -263,13 +262,13 @@ impl BenchDispatcher {
|
||||
pub fn run_ready_main_tasks(&self) -> bool {
|
||||
assert!(
|
||||
self.is_main_thread(),
|
||||
"run_ready_main_tasks must be called on the benchmark main thread"
|
||||
"run_ready_main_tasks must be called on the threaded dispatcher's main thread"
|
||||
);
|
||||
self.drain_main_queue()
|
||||
}
|
||||
|
||||
/// Cancels all pending timers so timers armed by one benchmark can't fire
|
||||
/// during a later benchmark sharing this process-lifetime dispatcher.
|
||||
/// Cancels all pending timers so timers armed by one workload can't fire
|
||||
/// during a later workload sharing this process-lifetime dispatcher.
|
||||
///
|
||||
/// Dropping a timer runnable drops its completion sender, waking the task
|
||||
/// awaiting the timer. Call [`Self::run_until_idle`] after this method to
|
||||
@@ -287,13 +286,13 @@ impl BenchDispatcher {
|
||||
}
|
||||
|
||||
/// Describes the dispatcher's idle-tracking state, for diagnosing
|
||||
/// benchmarks that fail to reach quiescence.
|
||||
/// workloads that fail to reach quiescence.
|
||||
pub fn debug_state(&self) -> String {
|
||||
let inflight = *self.idle.inflight.lock();
|
||||
let timers = self.timers.state.lock().heap.len();
|
||||
let main_queue_has_work = self.main_queue_has_work();
|
||||
format!(
|
||||
"BenchDispatcher {{ inflight: {inflight}, pending_timers: {timers}, \
|
||||
"ThreadedDispatcher {{ inflight: {inflight}, pending_timers: {timers}, \
|
||||
main_queue_has_work: {main_queue_has_work} }}"
|
||||
)
|
||||
}
|
||||
@@ -331,7 +330,7 @@ impl BenchDispatcher {
|
||||
}
|
||||
}
|
||||
|
||||
impl PlatformDispatcher for BenchDispatcher {
|
||||
impl PlatformDispatcher for ThreadedDispatcher {
|
||||
fn is_main_thread(&self) -> bool {
|
||||
thread::current().id() == self.main_thread_id
|
||||
}
|
||||
@@ -340,7 +339,7 @@ impl PlatformDispatcher for BenchDispatcher {
|
||||
self.idle.increment();
|
||||
self.background_sender
|
||||
.send(priority, runnable)
|
||||
.unwrap_or_else(|_| panic!("benchmark worker threads are no longer running"));
|
||||
.unwrap_or_else(|_| panic!("threaded dispatcher workers are no longer running"));
|
||||
}
|
||||
|
||||
fn dispatch_on_main_thread(&self, runnable: RunnableVariant, priority: Priority) {
|
||||
@@ -369,15 +368,15 @@ impl PlatformDispatcher for BenchDispatcher {
|
||||
}
|
||||
|
||||
fn spawn_realtime(&self, f: Box<dyn FnOnce() + Send>) {
|
||||
// Benchmarks don't need realtime scheduling priority; a plain thread
|
||||
// keeps this portable.
|
||||
// This dispatcher does not need realtime scheduling priority; a plain
|
||||
// thread keeps it portable.
|
||||
thread::Builder::new()
|
||||
.name("BenchRealtime".to_owned())
|
||||
.name("ThreadedDispatcherRealtime".to_owned())
|
||||
.spawn(f)
|
||||
.expect("failed to spawn benchmark realtime thread");
|
||||
.expect("failed to spawn threaded dispatcher realtime thread");
|
||||
}
|
||||
|
||||
fn as_bench(&self) -> Option<&BenchDispatcher> {
|
||||
fn as_threaded(&self) -> Option<&ThreadedDispatcher> {
|
||||
Some(self)
|
||||
}
|
||||
}
|
||||
@@ -391,7 +390,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn run_ready_main_tasks_does_not_wait_for_background_handoffs() {
|
||||
let dispatcher = Arc::new(BenchDispatcher::new());
|
||||
let dispatcher = Arc::new(ThreadedDispatcher::new());
|
||||
let background = BackgroundExecutor::new(dispatcher.clone());
|
||||
let foreground = ForegroundExecutor::new(dispatcher.clone());
|
||||
|
||||
@@ -423,7 +422,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn run_until_idle_completes_background_to_main_handoffs() {
|
||||
let dispatcher = Arc::new(BenchDispatcher::new());
|
||||
let dispatcher = Arc::new(ThreadedDispatcher::new());
|
||||
let background = BackgroundExecutor::new(dispatcher.clone());
|
||||
let foreground = ForegroundExecutor::new(dispatcher.clone());
|
||||
|
||||
@@ -452,7 +451,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn timers_fire_in_real_time() {
|
||||
let dispatcher = Arc::new(BenchDispatcher::new());
|
||||
let dispatcher = Arc::new(ThreadedDispatcher::new());
|
||||
let background = BackgroundExecutor::new(dispatcher);
|
||||
|
||||
let fired = Arc::new(AtomicBool::new(false));
|
||||
@@ -476,7 +475,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn cancel_pending_timers_wakes_waiters_without_waiting_for_deadline() {
|
||||
let dispatcher = Arc::new(BenchDispatcher::new());
|
||||
let dispatcher = Arc::new(ThreadedDispatcher::new());
|
||||
let background = BackgroundExecutor::new(dispatcher.clone());
|
||||
|
||||
let fired = Arc::new(AtomicBool::new(false));
|
||||
Reference in new issue
Block a user