compiletest/

executor.rs

//! This module contains a reimplementation of the subset of libtest
//! functionality needed by compiletest.
//!
//! FIXME(Zalathar): Much of this code was originally designed to mimic libtest
//! as closely as possible, for ease of migration. Now that libtest is no longer
//! used, we can potentially redesign things to be a better fit for compiletest.

use std::borrow::Cow;
use std::collections::HashMap;
use std::hash::{BuildHasherDefault, DefaultHasher};
use std::num::NonZero;
use std::sync::{Arc, Mutex, mpsc};
use std::{env, hint, io, mem, panic, thread};

use crate::common::{Config, TestPaths};
use crate::output_capture::{self, ConsoleOut};
use crate::panic_hook;

mod deadline;
mod json;

pub(crate) fn run_tests(config: &Config, tests: Vec<CollectedTest>) -> bool {
    let tests_len = tests.len();
    let filtered = filter_tests(config, tests);
    // Iterator yielding tests that haven't been started yet.
    let mut fresh_tests = (0..).map(TestId).zip(&filtered);

    let concurrency = get_concurrency();
    assert!(concurrency > 0);
    let concurrent_capacity = concurrency.min(filtered.len());

    let mut listener = json::Listener::new();
    let mut running_tests = HashMap::with_capacity_and_hasher(
        concurrent_capacity,
        BuildHasherDefault::<DefaultHasher>::new(),
    );
    let mut deadline_queue = deadline::DeadlineQueue::with_capacity(concurrent_capacity);

    let num_filtered_out = tests_len - filtered.len();
    listener.suite_started(filtered.len(), num_filtered_out);

    // Channel used by test threads to report the test outcome when done.
    let (completion_tx, completion_rx) = mpsc::channel::<TestCompletion>();

    // Unlike libtest, we don't have a separate code path for concurrency=1.
    // In that case, the tests will effectively be run serially anyway.
    loop {
        // Spawn new test threads, up to the concurrency limit.
        while running_tests.len() < concurrency
            && let Some((id, test)) = fresh_tests.next()
        {
            listener.test_started(test);
            deadline_queue.push(id, test);
            let join_handle = spawn_test_thread(id, test, completion_tx.clone());
            running_tests.insert(id, RunningTest { test, join_handle });
        }

        // If all running tests have finished, and there weren't any unstarted
        // tests to spawn, then we're done.
        if running_tests.is_empty() {
            break;
        }

        let completion = deadline_queue
            .read_channel_while_checking_deadlines(
                &completion_rx,
                |id| running_tests.contains_key(&id),
                |_id, test| listener.test_timed_out(test),
            )
            .expect("receive channel should never be closed early");

        let RunningTest { test, join_handle } = running_tests.remove(&completion.id).unwrap();
        if let Some(join_handle) = join_handle {
            join_handle.join().unwrap_or_else(|_| {
                panic!("thread for `{}` panicked after reporting completion", test.desc.name)
            });
        }

        listener.test_finished(test, &completion);

        if completion.outcome.is_failed() && config.fail_fast {
            // Prevent any other in-flight threads from panicking when they
            // write to the completion channel.
            mem::forget(completion_rx);
            break;
        }
    }

    let suite_passed = listener.suite_finished();
    suite_passed
}

/// Spawns a thread to run a single test, and returns the thread's join handle.
///
/// Returns `None` if the test was ignored, so no thread was spawned.
fn spawn_test_thread(
    id: TestId,
    test: &CollectedTest,
    completion_tx: mpsc::Sender<TestCompletion>,
) -> Option<thread::JoinHandle<()>> {
    if test.desc.ignore && !test.config.run_ignored {
        completion_tx
            .send(TestCompletion { id, outcome: TestOutcome::Ignored, stdout: None })
            .unwrap();
        return None;
    }

    let runnable_test = RunnableTest::new(test);
    let should_panic = test.desc.should_panic;
    let run_test = move || run_test_inner(id, should_panic, runnable_test, completion_tx);

    let thread_builder = thread::Builder::new().name(test.desc.name.clone());
    let join_handle = thread_builder.spawn(run_test).unwrap();
    Some(join_handle)
}

/// Runs a single test, within the dedicated thread spawned by the caller.
fn run_test_inner(
    id: TestId,
    should_panic: ShouldPanic,
    runnable_test: RunnableTest,
    completion_sender: mpsc::Sender<TestCompletion>,
) {
    let capture = CaptureKind::for_config(&runnable_test.config);

    // Install a panic-capture buffer for use by the custom panic hook.
    if capture.should_set_panic_hook() {
        panic_hook::set_capture_buf(Default::default());
    }

    if let CaptureKind::Old { ref buf } = capture {
        io::set_output_capture(Some(Arc::clone(buf)));
    }

    let stdout = capture.stdout();
    let stderr = capture.stderr();

    let panic_payload = panic::catch_unwind(move || runnable_test.run(stdout, stderr)).err();

    if let Some(panic_buf) = panic_hook::take_capture_buf() {
        let panic_buf = panic_buf.lock().unwrap_or_else(|e| e.into_inner());
        // Forward any captured panic message to (captured) stderr.
        write!(stderr, "{panic_buf}");
    }
    if matches!(capture, CaptureKind::Old { .. }) {
        io::set_output_capture(None);
    }

    let outcome = match (should_panic, panic_payload) {
        (ShouldPanic::No, None) | (ShouldPanic::Yes, Some(_)) => TestOutcome::Succeeded,
        (ShouldPanic::No, Some(_)) => TestOutcome::Failed { message: None },
        (ShouldPanic::Yes, None) => {
            TestOutcome::Failed { message: Some("test did not panic as expected") }
        }
    };

    let stdout = capture.into_inner();
    completion_sender.send(TestCompletion { id, outcome, stdout }).unwrap();
}

enum CaptureKind {
    /// Do not capture test-runner output, for `--no-capture`.
    ///
    /// (This does not affect `rustc` and other subprocesses spawned by test
    /// runners, whose output is always captured.)
    None,

    /// Use the old output-capture implementation, which relies on the unstable
    /// library feature `#![feature(internal_output_capture)]`.
    Old { buf: Arc<Mutex<Vec<u8>>> },

    /// Use the new output-capture implementation, which only uses stable Rust.
    New { buf: output_capture::CaptureBuf },
}

impl CaptureKind {
    fn for_config(config: &Config) -> Self {
        if config.nocapture {
            Self::None
        } else if config.new_output_capture {
            Self::New { buf: output_capture::CaptureBuf::new() }
        } else {
            // Create a capure buffer for `io::set_output_capture`.
            Self::Old { buf: Default::default() }
        }
    }

    fn should_set_panic_hook(&self) -> bool {
        match self {
            Self::None => false,
            Self::Old { .. } => true,
            Self::New { .. } => true,
        }
    }

    fn stdout(&self) -> &dyn ConsoleOut {
        self.capture_buf_or(&output_capture::Stdout)
    }

    fn stderr(&self) -> &dyn ConsoleOut {
        self.capture_buf_or(&output_capture::Stderr)
    }

    fn capture_buf_or<'a>(&'a self, fallback: &'a dyn ConsoleOut) -> &'a dyn ConsoleOut {
        match self {
            Self::None | Self::Old { .. } => fallback,
            Self::New { buf } => buf,
        }
    }

    fn into_inner(self) -> Option<Vec<u8>> {
        match self {
            Self::None => None,
            Self::Old { buf } => Some(buf.lock().unwrap_or_else(|e| e.into_inner()).to_vec()),
            Self::New { buf } => Some(buf.into_inner().into()),
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
struct TestId(usize);

struct RunnableTest {
    config: Arc<Config>,
    testpaths: TestPaths,
    revision: Option<String>,
}

impl RunnableTest {
    fn new(test: &CollectedTest) -> Self {
        let config = Arc::clone(&test.config);
        let testpaths = test.testpaths.clone();
        let revision = test.revision.clone();
        Self { config, testpaths, revision }
    }

    fn run(&self, stdout: &dyn ConsoleOut, stderr: &dyn ConsoleOut) {
        __rust_begin_short_backtrace(|| {
            crate::runtest::run(
                Arc::clone(&self.config),
                stdout,
                stderr,
                &self.testpaths,
                self.revision.as_deref(),
            );
        });
    }
}

/// Fixed frame used to clean the backtrace with `RUST_BACKTRACE=1`.
#[inline(never)]
fn __rust_begin_short_backtrace<T, F: FnOnce() -> T>(f: F) -> T {
    let result = f();

    // prevent this frame from being tail-call optimised away
    hint::black_box(result)
}

struct RunningTest<'a> {
    test: &'a CollectedTest,
    join_handle: Option<thread::JoinHandle<()>>,
}

/// Test completion message sent by individual test threads when their test
/// finishes (successfully or unsuccessfully).
struct TestCompletion {
    id: TestId,
    outcome: TestOutcome,
    stdout: Option<Vec<u8>>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum TestOutcome {
    Succeeded,
    Failed { message: Option<&'static str> },
    Ignored,
}

impl TestOutcome {
    fn is_failed(&self) -> bool {
        matches!(self, Self::Failed { .. })
    }
}

/// Applies command-line arguments for filtering/skipping tests by name.
///
/// Adapted from `filter_tests` in libtest.
///
/// FIXME(#139660): Now that libtest has been removed, redesign the whole filtering system to
/// do a better job of understanding and filtering _paths_, instead of being tied to libtest's
/// substring/exact matching behaviour.
fn filter_tests(opts: &Config, tests: Vec<CollectedTest>) -> Vec<CollectedTest> {
    let mut filtered = tests;

    let matches_filter = |test: &CollectedTest, filter_str: &str| {
        let test_name = &test.desc.name;
        if opts.filter_exact { test_name == filter_str } else { test_name.contains(filter_str) }
    };

    // Remove tests that don't match the test filter
    if !opts.filters.is_empty() {
        filtered.retain(|test| opts.filters.iter().any(|filter| matches_filter(test, filter)));
    }

    // Skip tests that match any of the skip filters
    if !opts.skip.is_empty() {
        filtered.retain(|test| !opts.skip.iter().any(|sf| matches_filter(test, sf)));
    }

    filtered
}

/// Determines the number of tests to run concurrently.
///
/// Copied from `get_concurrency` in libtest.
///
/// FIXME(#139660): After the libtest dependency is removed, consider making bootstrap specify the
/// number of threads on the command-line, instead of propagating the `RUST_TEST_THREADS`
/// environment variable.
fn get_concurrency() -> usize {
    if let Ok(value) = env::var("RUST_TEST_THREADS") {
        match value.parse::<NonZero<usize>>().ok() {
            Some(n) => n.get(),
            _ => panic!("RUST_TEST_THREADS is `{value}`, should be a positive integer."),
        }
    } else {
        thread::available_parallelism().map(|n| n.get()).unwrap_or(1)
    }
}

/// Information that was historically needed to create a libtest `TestDescAndFn`.
pub(crate) struct CollectedTest {
    pub(crate) desc: CollectedTestDesc,
    pub(crate) config: Arc<Config>,
    pub(crate) testpaths: TestPaths,
    pub(crate) revision: Option<String>,
}

/// Information that was historically needed to create a libtest `TestDesc`.
pub(crate) struct CollectedTestDesc {
    pub(crate) name: String,
    pub(crate) ignore: bool,
    pub(crate) ignore_message: Option<Cow<'static, str>>,
    pub(crate) should_panic: ShouldPanic,
}

/// Whether console output should be colored or not.
#[derive(Copy, Clone, Default, Debug)]
pub enum ColorConfig {
    #[default]
    AutoColor,
    AlwaysColor,
    NeverColor,
}

/// Whether test is expected to panic or not.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub(crate) enum ShouldPanic {
    No,
    Yes,
}