Run subtests before entirely new tests

Author: oli-obk

src/core.rs

@@ -94,14 +94,17 @@ pub enum CrateType {
 
 /// A generic multithreaded runner that has a thread for producing work,
 /// a thread for collecting work, and `num_threads` threads for doing the work.
-pub fn run_and_collect<SUBMISSION: Send, RESULT: Send>(
+pub fn run_and_collect<const N: usize, SUBMISSION: Send, RESULT: Send>(
     num_threads: NonZeroUsize,
-    submitter: impl FnOnce(Sender<SUBMISSION>) + Send,
-    runner: impl Sync + Fn(&Receiver<SUBMISSION>, Sender<RESULT>) -> Result<()>,
+    submitter: impl FnOnce([Sender<SUBMISSION>; N]) + Send,
+    runner: impl Sync + Fn(&[Receiver<SUBMISSION>; N], Sender<RESULT>) -> Result<()>,
     collector: impl FnOnce(Receiver<RESULT>) + Send,
 ) -> Result<()> {
     // A channel for files to process
-    let (submit, receive) = unbounded();
+    let (submit, receive): (Vec<_>, Vec<_>) = std::iter::repeat_with(unbounded).take(N).unzip();
+    let receive = receive[..].try_into().unwrap();
+    let mut submit = submit.into_iter();
+    let submit = std::array::from_fn(|_| submit.next().unwrap());
 
     thread::scope(|s| {
         // Create a thread that is in charge of walking the directory and submitting jobs.
@@ -119,7 +122,7 @@ pub fn run_and_collect<SUBMISSION: Send, RESULT: Send>(
         // Create N worker threads that receive files to test.
         for _ in 0..num_threads.get() {
             let finished_files_sender = finished_files_sender.clone();
-            threads.push(s.spawn(|| runner(&receive, finished_files_sender)));
+            threads.push(s.spawn(|| runner(receive, finished_files_sender)));
         }
 
         for thread in threads {

src/lib.rs

@@ -17,6 +17,7 @@ pub use color_eyre::eyre::Result;
 pub use core::run_and_collect;
 pub use core::CrateType;
 use crossbeam_channel::Sender;
+use crossbeam_channel::TryRecvError;
 pub use filter::Match;
 use per_test_config::TestConfig;
 use spanned::Spanned;
@@ -198,12 +199,12 @@ pub fn run_tests_generic(
     let mut filtered = 0;
     core::run_and_collect(
         num_threads,
-        |submit| {
+        |[submit, priority_submit]| {
             let mut todo = VecDeque::new();
 
             let configs: Vec<_> = configs
                 .into_iter()
-                .map(|config| Arc::new(BuildManager::new(config, submit.clone())))
+                .map(|config| Arc::new(BuildManager::new(config, priority_submit.clone())))
                 .collect();
             for build_manager in &configs {
                 todo.push_back((
@@ -287,11 +288,24 @@ pub fn run_tests_generic(
                 }
             }
         },
-        |receive, finished_files_sender| -> Result<()> {
-            for closure in receive {
-                closure(&finished_files_sender)?;
+        |[receive, priority_receive], finished_files_sender| -> Result<()> {
+            loop {
+                for closure in priority_receive.try_iter() {
+                    closure(&finished_files_sender)?;
+                }
+                match receive.try_recv() {
+                    Ok(closure) => {
+                        closure(&finished_files_sender)?;
+                    }
+                    Err(TryRecvError::Empty) => {}
+                    Err(TryRecvError::Disconnected) => {
+                        for closure in priority_receive {
+                            closure(&finished_files_sender)?;
+                        }
+                        return Ok(());
+                    }
+                }
             }
-            Ok(())
         },
         |finished_files_recv| {
             for run in finished_files_recv {