From 0b9b1df0064396708a5e5ca27fd010ae3ad3a305 Mon Sep 17 00:00:00 2001 From: "Celina G. Val" Date: Wed, 11 Jun 2025 11:12:32 -0700 Subject: Fix format and tidy for code moved from rayon --- compiler/rustc_thread_pool/src/spawn/mod.rs | 17 +- compiler/rustc_thread_pool/src/spawn/test.rs | 255 -------------------------- compiler/rustc_thread_pool/src/spawn/tests.rs | 246 +++++++++++++++++++++++++ 3 files changed, 255 insertions(+), 263 deletions(-) delete mode 100644 compiler/rustc_thread_pool/src/spawn/test.rs create mode 100644 compiler/rustc_thread_pool/src/spawn/tests.rs (limited to 'compiler/rustc_thread_pool/src/spawn') diff --git a/compiler/rustc_thread_pool/src/spawn/mod.rs b/compiler/rustc_thread_pool/src/spawn/mod.rs index 034df30dcfb..f1679a98234 100644 --- a/compiler/rustc_thread_pool/src/spawn/mod.rs +++ b/compiler/rustc_thread_pool/src/spawn/mod.rs @@ -1,9 +1,10 @@ +use std::mem; +use std::sync::Arc; + use crate::job::*; use crate::registry::Registry; use crate::tlv::Tlv; use crate::unwind; -use std::mem; -use std::sync::Arc; /// Puts the task into the Rayon threadpool's job queue in the "static" /// or "global" scope. Just like a standard thread, this task is not @@ -28,9 +29,9 @@ use std::sync::Arc; /// other threads may steal tasks at any time. However, they are /// generally prioritized in a LIFO order on the thread from which /// they were spawned. Other threads always steal from the other end of -/// the deque, like FIFO order. The idea is that "recent" tasks are +/// the deque, like FIFO order. The idea is that "recent" tasks are /// most likely to be fresh in the local CPU's cache, while other -/// threads can steal older "stale" tasks. For an alternate approach, +/// threads can steal older "stale" tasks. For an alternate approach, /// consider [`spawn_fifo()`] instead. /// /// [`spawn_fifo()`]: fn.spawn_fifo.html @@ -39,7 +40,7 @@ use std::sync::Arc; /// /// If this closure should panic, the resulting panic will be /// propagated to the panic handler registered in the `ThreadPoolBuilder`, -/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more +/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more /// details. /// /// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler @@ -103,7 +104,7 @@ where } /// Fires off a task into the Rayon threadpool in the "static" or -/// "global" scope. Just like a standard thread, this task is not +/// "global" scope. Just like a standard thread, this task is not /// tied to the current stack frame, and hence it cannot hold any /// references other than those with `'static` lifetime. If you want /// to spawn a task that references stack data, use [the `scope_fifo()` @@ -124,7 +125,7 @@ where /// /// If this closure should panic, the resulting panic will be /// propagated to the panic handler registered in the `ThreadPoolBuilder`, -/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more +/// if any. See [`ThreadPoolBuilder::panic_handler()`][ph] for more /// details. /// /// [ph]: struct.ThreadPoolBuilder.html#method.panic_handler @@ -152,7 +153,7 @@ where let job_ref = spawn_job(func, registry); // If we're in the pool, use our thread's private fifo for this thread to execute - // in a locally-FIFO order. Otherwise, just use the pool's global injector. + // in a locally-FIFO order. Otherwise, just use the pool's global injector. match registry.current_thread() { Some(worker) => worker.push_fifo(job_ref), None => registry.inject(job_ref), diff --git a/compiler/rustc_thread_pool/src/spawn/test.rs b/compiler/rustc_thread_pool/src/spawn/test.rs deleted file mode 100644 index b7a0535aabf..00000000000 --- a/compiler/rustc_thread_pool/src/spawn/test.rs +++ /dev/null @@ -1,255 +0,0 @@ -use crate::scope; -use std::any::Any; -use std::sync::mpsc::channel; -use std::sync::Mutex; - -use super::{spawn, spawn_fifo}; -use crate::ThreadPoolBuilder; - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_then_join_in_worker() { - let (tx, rx) = channel(); - scope(move |_| { - spawn(move || tx.send(22).unwrap()); - }); - assert_eq!(22, rx.recv().unwrap()); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn spawn_then_join_outside_worker() { - let (tx, rx) = channel(); - spawn(move || tx.send(22).unwrap()); - assert_eq!(22, rx.recv().unwrap()); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn panic_fwd() { - let (tx, rx) = channel(); - - let tx = Mutex::new(tx); - let panic_handler = move |err: Box| { - let tx = tx.lock().unwrap(); - if let Some(&msg) = err.downcast_ref::<&str>() { - if msg == "Hello, world!" { - tx.send(1).unwrap(); - } else { - tx.send(2).unwrap(); - } - } else { - tx.send(3).unwrap(); - } - }; - - let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); - - builder - .build() - .unwrap() - .spawn(move || panic!("Hello, world!")); - - assert_eq!(1, rx.recv().unwrap()); -} - -/// Test what happens when the thread-pool is dropped but there are -/// still active asynchronous tasks. We expect the thread-pool to stay -/// alive and executing until those threads are complete. -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn termination_while_things_are_executing() { - let (tx0, rx0) = channel(); - let (tx1, rx1) = channel(); - - // Create a thread-pool and spawn some code in it, but then drop - // our reference to it. - { - let thread_pool = ThreadPoolBuilder::new().build().unwrap(); - thread_pool.spawn(move || { - let data = rx0.recv().unwrap(); - - // At this point, we know the "main" reference to the - // `ThreadPool` has been dropped, but there are still - // active threads. Launch one more. - spawn(move || { - tx1.send(data).unwrap(); - }); - }); - } - - tx0.send(22).unwrap(); - let v = rx1.recv().unwrap(); - assert_eq!(v, 22); -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn custom_panic_handler_and_spawn() { - let (tx, rx) = channel(); - - // Create a parallel closure that will send panics on the - // channel; since the closure is potentially executed in parallel - // with itself, we have to wrap `tx` in a mutex. - let tx = Mutex::new(tx); - let panic_handler = move |e: Box| { - tx.lock().unwrap().send(e).unwrap(); - }; - - // Execute an async that will panic. - let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); - builder.build().unwrap().spawn(move || { - panic!("Hello, world!"); - }); - - // Check that we got back the panic we expected. - let error = rx.recv().unwrap(); - if let Some(&msg) = error.downcast_ref::<&str>() { - assert_eq!(msg, "Hello, world!"); - } else { - panic!("did not receive a string from panic handler"); - } -} - -#[test] -#[cfg_attr(not(panic = "unwind"), ignore)] -fn custom_panic_handler_and_nested_spawn() { - let (tx, rx) = channel(); - - // Create a parallel closure that will send panics on the - // channel; since the closure is potentially executed in parallel - // with itself, we have to wrap `tx` in a mutex. - let tx = Mutex::new(tx); - let panic_handler = move |e| { - tx.lock().unwrap().send(e).unwrap(); - }; - - // Execute an async that will (eventually) panic. - const PANICS: usize = 3; - let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); - builder.build().unwrap().spawn(move || { - // launch 3 nested spawn-asyncs; these should be in the same - // thread-pool and hence inherit the same panic handler - for _ in 0..PANICS { - spawn(move || { - panic!("Hello, world!"); - }); - } - }); - - // Check that we get back the panics we expected. - for _ in 0..PANICS { - let error = rx.recv().unwrap(); - if let Some(&msg) = error.downcast_ref::<&str>() { - assert_eq!(msg, "Hello, world!"); - } else { - panic!("did not receive a string from panic handler"); - } - } -} - -macro_rules! test_order { - ($outer_spawn:ident, $inner_spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - let (tx, rx) = channel(); - pool.install(move || { - for i in 0..10 { - let tx = tx.clone(); - $outer_spawn(move || { - for j in 0..10 { - let tx = tx.clone(); - $inner_spawn(move || { - tx.send(i * 10 + j).unwrap(); - }); - } - }); - } - }); - rx.iter().collect::>() - }}; -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn lifo_order() { - // In the absence of stealing, `spawn()` jobs on a thread will run in LIFO order. - let vec = test_order!(spawn, spawn); - let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn fifo_order() { - // In the absence of stealing, `spawn_fifo()` jobs on a thread will run in FIFO order. - let vec = test_order!(spawn_fifo, spawn_fifo); - let expected: Vec = (0..100).collect(); // FIFO -> natural order - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn lifo_fifo_order() { - // LIFO on the outside, FIFO on the inside - let vec = test_order!(spawn, spawn_fifo); - let expected: Vec = (0..10) - .rev() - .flat_map(|i| (0..10).map(move |j| i * 10 + j)) - .collect(); - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn fifo_lifo_order() { - // FIFO on the outside, LIFO on the inside - let vec = test_order!(spawn_fifo, spawn); - let expected: Vec = (0..10) - .flat_map(|i| (0..10).rev().map(move |j| i * 10 + j)) - .collect(); - assert_eq!(vec, expected); -} - -macro_rules! spawn_send { - ($spawn:ident, $tx:ident, $i:expr) => {{ - let tx = $tx.clone(); - $spawn(move || tx.send($i).unwrap()); - }}; -} - -/// Test mixed spawns pushing a series of numbers, interleaved such -/// such that negative values are using the second kind of spawn. -macro_rules! test_mixed_order { - ($pos_spawn:ident, $neg_spawn:ident) => {{ - let builder = ThreadPoolBuilder::new().num_threads(1); - let pool = builder.build().unwrap(); - let (tx, rx) = channel(); - pool.install(move || { - spawn_send!($pos_spawn, tx, 0); - spawn_send!($neg_spawn, tx, -1); - spawn_send!($pos_spawn, tx, 1); - spawn_send!($neg_spawn, tx, -2); - spawn_send!($pos_spawn, tx, 2); - spawn_send!($neg_spawn, tx, -3); - spawn_send!($pos_spawn, tx, 3); - }); - rx.iter().collect::>() - }}; -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn mixed_lifo_fifo_order() { - let vec = test_mixed_order!(spawn, spawn_fifo); - let expected = vec![3, -1, 2, -2, 1, -3, 0]; - assert_eq!(vec, expected); -} - -#[test] -#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] -fn mixed_fifo_lifo_order() { - let vec = test_mixed_order!(spawn_fifo, spawn); - let expected = vec![0, -3, 1, -2, 2, -1, 3]; - assert_eq!(vec, expected); -} diff --git a/compiler/rustc_thread_pool/src/spawn/tests.rs b/compiler/rustc_thread_pool/src/spawn/tests.rs new file mode 100644 index 00000000000..8a70d2faf9c --- /dev/null +++ b/compiler/rustc_thread_pool/src/spawn/tests.rs @@ -0,0 +1,246 @@ +use std::any::Any; +use std::sync::Mutex; +use std::sync::mpsc::channel; + +use super::{spawn, spawn_fifo}; +use crate::{ThreadPoolBuilder, scope}; + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_then_join_in_worker() { + let (tx, rx) = channel(); + scope(move |_| { + spawn(move || tx.send(22).unwrap()); + }); + assert_eq!(22, rx.recv().unwrap()); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn spawn_then_join_outside_worker() { + let (tx, rx) = channel(); + spawn(move || tx.send(22).unwrap()); + assert_eq!(22, rx.recv().unwrap()); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn panic_fwd() { + let (tx, rx) = channel(); + + let tx = Mutex::new(tx); + let panic_handler = move |err: Box| { + let tx = tx.lock().unwrap(); + if let Some(&msg) = err.downcast_ref::<&str>() { + if msg == "Hello, world!" { + tx.send(1).unwrap(); + } else { + tx.send(2).unwrap(); + } + } else { + tx.send(3).unwrap(); + } + }; + + let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); + + builder.build().unwrap().spawn(move || panic!("Hello, world!")); + + assert_eq!(1, rx.recv().unwrap()); +} + +/// Test what happens when the thread-pool is dropped but there are +/// still active asynchronous tasks. We expect the thread-pool to stay +/// alive and executing until those threads are complete. +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn termination_while_things_are_executing() { + let (tx0, rx0) = channel(); + let (tx1, rx1) = channel(); + + // Create a thread-pool and spawn some code in it, but then drop + // our reference to it. + { + let thread_pool = ThreadPoolBuilder::new().build().unwrap(); + thread_pool.spawn(move || { + let data = rx0.recv().unwrap(); + + // At this point, we know the "main" reference to the + // `ThreadPool` has been dropped, but there are still + // active threads. Launch one more. + spawn(move || { + tx1.send(data).unwrap(); + }); + }); + } + + tx0.send(22).unwrap(); + let v = rx1.recv().unwrap(); + assert_eq!(v, 22); +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn custom_panic_handler_and_spawn() { + let (tx, rx) = channel(); + + // Create a parallel closure that will send panics on the + // channel; since the closure is potentially executed in parallel + // with itself, we have to wrap `tx` in a mutex. + let tx = Mutex::new(tx); + let panic_handler = move |e: Box| { + tx.lock().unwrap().send(e).unwrap(); + }; + + // Execute an async that will panic. + let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); + builder.build().unwrap().spawn(move || { + panic!("Hello, world!"); + }); + + // Check that we got back the panic we expected. + let error = rx.recv().unwrap(); + if let Some(&msg) = error.downcast_ref::<&str>() { + assert_eq!(msg, "Hello, world!"); + } else { + panic!("did not receive a string from panic handler"); + } +} + +#[test] +#[cfg_attr(not(panic = "unwind"), ignore)] +fn custom_panic_handler_and_nested_spawn() { + let (tx, rx) = channel(); + + // Create a parallel closure that will send panics on the + // channel; since the closure is potentially executed in parallel + // with itself, we have to wrap `tx` in a mutex. + let tx = Mutex::new(tx); + let panic_handler = move |e| { + tx.lock().unwrap().send(e).unwrap(); + }; + + // Execute an async that will (eventually) panic. + const PANICS: usize = 3; + let builder = ThreadPoolBuilder::new().panic_handler(panic_handler); + builder.build().unwrap().spawn(move || { + // launch 3 nested spawn-asyncs; these should be in the same + // thread-pool and hence inherit the same panic handler + for _ in 0..PANICS { + spawn(move || { + panic!("Hello, world!"); + }); + } + }); + + // Check that we get back the panics we expected. + for _ in 0..PANICS { + let error = rx.recv().unwrap(); + if let Some(&msg) = error.downcast_ref::<&str>() { + assert_eq!(msg, "Hello, world!"); + } else { + panic!("did not receive a string from panic handler"); + } + } +} + +macro_rules! test_order { + ($outer_spawn:ident, $inner_spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + let (tx, rx) = channel(); + pool.install(move || { + for i in 0..10 { + let tx = tx.clone(); + $outer_spawn(move || { + for j in 0..10 { + let tx = tx.clone(); + $inner_spawn(move || { + tx.send(i * 10 + j).unwrap(); + }); + } + }); + } + }); + rx.iter().collect::>() + }}; +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn lifo_order() { + // In the absence of stealing, `spawn()` jobs on a thread will run in LIFO order. + let vec = test_order!(spawn, spawn); + let expected: Vec = (0..100).rev().collect(); // LIFO -> reversed + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn fifo_order() { + // In the absence of stealing, `spawn_fifo()` jobs on a thread will run in FIFO order. + let vec = test_order!(spawn_fifo, spawn_fifo); + let expected: Vec = (0..100).collect(); // FIFO -> natural order + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn lifo_fifo_order() { + // LIFO on the outside, FIFO on the inside + let vec = test_order!(spawn, spawn_fifo); + let expected: Vec = (0..10).rev().flat_map(|i| (0..10).map(move |j| i * 10 + j)).collect(); + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn fifo_lifo_order() { + // FIFO on the outside, LIFO on the inside + let vec = test_order!(spawn_fifo, spawn); + let expected: Vec = (0..10).flat_map(|i| (0..10).rev().map(move |j| i * 10 + j)).collect(); + assert_eq!(vec, expected); +} + +macro_rules! spawn_send { + ($spawn:ident, $tx:ident, $i:expr) => {{ + let tx = $tx.clone(); + $spawn(move || tx.send($i).unwrap()); + }}; +} + +/// Test mixed spawns pushing a series of numbers, interleaved such +/// such that negative values are using the second kind of spawn. +macro_rules! test_mixed_order { + ($pos_spawn:ident, $neg_spawn:ident) => {{ + let builder = ThreadPoolBuilder::new().num_threads(1); + let pool = builder.build().unwrap(); + let (tx, rx) = channel(); + pool.install(move || { + spawn_send!($pos_spawn, tx, 0); + spawn_send!($neg_spawn, tx, -1); + spawn_send!($pos_spawn, tx, 1); + spawn_send!($neg_spawn, tx, -2); + spawn_send!($pos_spawn, tx, 2); + spawn_send!($neg_spawn, tx, -3); + spawn_send!($pos_spawn, tx, 3); + }); + rx.iter().collect::>() + }}; +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn mixed_lifo_fifo_order() { + let vec = test_mixed_order!(spawn, spawn_fifo); + let expected = vec![3, -1, 2, -2, 1, -3, 0]; + assert_eq!(vec, expected); +} + +#[test] +#[cfg_attr(any(target_os = "emscripten", target_family = "wasm"), ignore)] +fn mixed_fifo_lifo_order() { + let vec = test_mixed_order!(spawn_fifo, spawn); + let expected = vec![0, -3, 1, -2, 2, -1, 3]; + assert_eq!(vec, expected); +} -- cgit 1.4.1-3-g733a5