many_cpus/

processor_set_builder.rs

use std::collections::VecDeque;
use std::fmt::Debug;
use std::num::NonZero;

use foldhash::{HashMap, HashMapExt, HashSet, HashSetExt};
use itertools::Itertools;
use nonempty::NonEmpty;
use rand::prelude::*;
use rand::rng;

use crate::pal::Platform;
use crate::{
    EfficiencyClass, MemoryRegionId, Processor, ProcessorId, ProcessorSet, SystemHardware,
};

/// Builds a [`ProcessorSet`] based on specified criteria.
///
/// You can obtain an instance by calling [`ProcessorSet::to_builder()`] on an existing [`ProcessorSet`],
/// typically either [`SystemHardware::processors()`] or [`SystemHardware::all_processors()`].
/// # External constraints
///
/// The operating system may define constraints that prohibit the application from using all
/// the available processors (e.g. when the app is containerized and provided limited
/// hardware resources).
///
/// This package treats platform constraints as follows:
///
/// * Hard limits on which processors are allowed are respected - forbidden processors are mostly
///   ignored by this package and cannot be used to spawn threads, though such processors are still
///   accounted for when inspecting hardware information such as "max processor ID".
///   The mechanisms for defining such limits are cgroups on Linux and job objects on Windows.
///   See `examples/obey_job_affinity_limits_windows.rs` for a Windows-specific example.
/// * Soft limits on which processors are allowed are ignored by default - specifying a processor
///   affinity via `taskset` on Linux, `start.exe /affinity 0xff` on Windows or similar mechanisms
///   does not affect the set of processors this package will use by default, though you can opt in to
///   this via [`.where_available_for_current_thread()`][crate::ProcessorSetBuilder::where_available_for_current_thread].
/// * Any operating system enforced processor time quota is taken as the upper bound for the processor
///   count of the processor set returned by [`SystemHardware::processors()`].
/// * Any other processor set can be opt-in quota-limited when building the processor set. For example, by calling `SystemHardware::current().all_processors().to_builder().enforce_resource_quota().take_all()`.
///
/// See `examples/obey_job_resource_quota_limits_windows.rs` for a Windows-specific example of processor
/// time quota enforcement.
///
/// # Avoiding operating system quota penalties
///
/// If a process exceeds the processor time limit, the operating system will delay executing the
/// process further until the "debt is paid off". This is undesirable for most workloads because:
///
/// 1. There will be random latency spikes from when the operating system decides to apply a delay.
/// 1. The delay may not be evenly applied across all threads of the process, leading to unbalanced
///    load between worker threads.
///
/// For predictable behavior that does not suffer from delay side-effects, it is important that the
/// process does not exceed the processor time limit. To keep out of trouble,
/// follow these guidelines:
///
/// * Ensure that all your concurrently executing thread pools are derived from the same processor
///   set, so there is a single set of processors (up to the resource quota) that all work of the
///   process will be executed on. Any new processor sets you create should be subsets of this set,
///   thereby ensuring that all worker threads combined do not exceed the quota.
/// * Ensure that the original processor set is constructed while obeying the resource quota (which is
///   enabled by default).
///
/// If your resource constraints are already applied on process startup, you can use
/// `SystemHardware::current().processors()` as the master set from which all other
/// processor sets are derived using either `.take()` or `.to_builder()`. This will ensure the
/// processor time quota is obeyed because `processors()` is size-limited to the quota.
///
/// ```rust
/// # use many_cpus::SystemHardware;
/// # use new_zealand::nz;
/// let hw = SystemHardware::current();
///
/// // By taking both senders and receivers from the same original processor set, we
/// // guarantee that all worker threads combined cannot exceed the processor time quota.
/// let mail_senders = hw
///     .processors()
///     .take(nz!(2))
///     .expect("need at least 2 processors for mail workers")
///     .spawn_threads(|_| send_mail());
///
/// let mail_receivers = hw
///     .processors()
///     .take(nz!(2))
///     .expect("need at least 2 processors for mail workers")
///     .spawn_threads(|_| receive_mail());
/// # fn send_mail() {}
/// # fn receive_mail() {}
/// ```
///
/// # Inheriting processor affinity from current thread
///
/// By default, the processor affinity of the current thread is ignored when building a processor
/// set, as this type may be used from a thread with a different processor affinity than the threads
/// one wants to configure.
///
/// However, if you do wish to inherit the processor affinity from the current thread, you may do
/// so by calling [`.where_available_for_current_thread()`] on the builder. This filters out all
/// processors that the current thread is not configured to execute on.
///
/// [`.where_available_for_current_thread()`]: ProcessorSetBuilder::where_available_for_current_thread
/// [`SystemHardware::processors()`]: crate::SystemHardware::processors
/// [`SystemHardware::all_processors()`]: crate::SystemHardware::all_processors
#[derive(Clone, Debug)]
pub struct ProcessorSetBuilder {
    processor_type_selector: ProcessorTypeSelector,
    memory_region_selector: MemoryRegionSelector,

    except_indexes: HashSet<ProcessorId>,

    obey_resource_quota: bool,

    /// When set, only processors with IDs in this set are considered as candidates.
    /// When `None`, all platform processors are considered.
    source_processor_ids: Option<HashSet<ProcessorId>>,

    /// The hardware instance to use for pin status tracking and platform access.
    /// Passed to any processor set we create.
    hardware: SystemHardware,
}

impl ProcessorSetBuilder {
    #[must_use]
    pub(crate) fn with_internals(hardware: SystemHardware) -> Self {
        Self {
            processor_type_selector: ProcessorTypeSelector::Any,
            memory_region_selector: MemoryRegionSelector::Any,
            except_indexes: HashSet::new(),
            obey_resource_quota: false,
            source_processor_ids: None,
            hardware,
        }
    }

    /// Restricts the builder to only consider processors from the given set.
    /// This is used internally when building from an existing `ProcessorSet`.
    #[must_use]
    pub(crate) fn source_processors(mut self, processors: &NonEmpty<Processor>) -> Self {
        let ids: HashSet<ProcessorId> = processors.iter().map(Processor::id).collect();
        self.source_processor_ids = Some(ids);
        self
    }

    /// Requires that all processors in the set be marked as [performance processors][1].
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::SystemHardware;
    /// use new_zealand::nz;
    ///
    /// // Get up to 4 performance processors (or fewer if not available)
    /// let performance_processors = SystemHardware::current()
    ///     .processors()
    ///     .to_builder()
    ///     .performance_processors_only()
    ///     .take(nz!(4));
    ///
    /// if let Some(processors) = performance_processors {
    ///     println!("Found {} performance processors", processors.len());
    /// } else {
    ///     println!("Could not find 4 performance processors");
    /// }
    /// ```
    ///
    /// [1]: EfficiencyClass::Performance
    #[must_use]
    pub fn performance_processors_only(mut self) -> Self {
        self.processor_type_selector = ProcessorTypeSelector::Performance;
        self
    }

    /// Requires that all processors in the set be marked as [efficiency processors][1].
    ///
    /// # Example
    ///
    /// ```
    /// use std::num::NonZero;
    ///
    /// use many_cpus::SystemHardware;
    ///
    /// // Get all available efficiency processors for background tasks
    /// let efficiency_processors = SystemHardware::current()
    ///     .processors()
    ///     .to_builder()
    ///     .efficiency_processors_only()
    ///     .take_all();
    ///
    /// if let Some(processors) = efficiency_processors {
    ///     println!(
    ///         "Using {} efficiency processors for background work",
    ///         processors.len()
    ///     );
    ///
    ///     // Spawn threads on efficiency processors to handle background tasks
    ///     let threads = processors.spawn_threads(|processor| {
    ///         println!(
    ///             "Background worker started on efficiency processor {}",
    ///             processor.id()
    ///         );
    ///         // Background work here...
    ///     });
    /// # for thread in threads { thread.join().unwrap(); }
    /// }
    /// ```
    ///
    /// [1]: EfficiencyClass::Efficiency
    #[must_use]
    pub fn efficiency_processors_only(mut self) -> Self {
        self.processor_type_selector = ProcessorTypeSelector::Efficiency;
        self
    }

    /// Requires that all processors in the set be from different memory regions, selecting a
    /// maximum of 1 processor from each memory region.
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::SystemHardware;
    /// use new_zealand::nz;
    ///
    /// // Get one processor from each memory region for distributed processing
    /// let distributed_processors = SystemHardware::current()
    ///     .processors()
    ///     .to_builder()
    ///     .different_memory_regions()
    ///     .take(nz!(4));
    ///
    /// if let Some(processors) = distributed_processors {
    ///     println!(
    ///         "Selected {} processors from different memory regions",
    ///         processors.len()
    ///     );
    ///
    ///     // Each processor will be in a different memory region,
    ///     // ideal for parallel work on separate data sets
    ///     for processor in &processors {
    ///         println!(
    ///             "Processor {} in memory region {}",
    ///             processor.id(),
    ///             processor.memory_region_id()
    ///         );
    ///     }
    /// }
    /// ```
    #[must_use]
    pub fn different_memory_regions(mut self) -> Self {
        self.memory_region_selector = MemoryRegionSelector::RequireDifferent;
        self
    }

    /// Requires that all processors in the set be from the same memory region.
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::SystemHardware;
    /// use new_zealand::nz;
    ///
    /// // Get processors from the same memory region for data locality
    /// let local_processors = SystemHardware::current()
    ///     .processors()
    ///     .to_builder()
    ///     .same_memory_region()
    ///     .take(nz!(3));
    ///
    /// if let Some(processors) = local_processors {
    ///     println!(
    ///         "Selected {} processors from the same memory region",
    ///         processors.len()
    ///     );
    ///
    ///     // All processors share the same memory region for optimal data sharing
    ///     let memory_region = processors.processors().first().memory_region_id();
    ///     println!("All processors are in memory region {}", memory_region);
    ///
    ///     // Ideal for cooperative processing of shared data
    ///     let threads = processors.spawn_threads(|processor| {
    ///         println!(
    ///             "Worker on processor {} accessing shared memory region {}",
    ///             processor.id(),
    ///             processor.memory_region_id()
    ///         );
    ///         // Process shared data here...
    ///     });
    /// # for thread in threads { thread.join().unwrap(); }
    /// }
    /// ```
    #[must_use]
    pub fn same_memory_region(mut self) -> Self {
        self.memory_region_selector = MemoryRegionSelector::RequireSame;
        self
    }

    /// Declares a preference that all processors in the set be from different memory regions,
    /// though will select multiple processors from the same memory region as needed to satisfy
    /// the requested processor count (while keeping the spread maximal).
    #[must_use]
    pub fn prefer_different_memory_regions(mut self) -> Self {
        self.memory_region_selector = MemoryRegionSelector::PreferDifferent;
        self
    }

    /// Declares a preference that all processors in the set be from the same memory region,
    /// though will select processors from different memory regions as needed to satisfy the
    /// requested processor count (while keeping the spread minimal).
    #[must_use]
    pub fn prefer_same_memory_region(mut self) -> Self {
        self.memory_region_selector = MemoryRegionSelector::PreferSame;
        self
    }

    /// Uses a predicate to identify processors that are valid candidates for building the
    /// processor set, with a return value of `bool` indicating that a processor is a valid
    /// candidate for selection into the set.
    ///
    /// The candidates are passed to this function without necessarily first considering all other
    /// conditions - even if this predicate returns `true`, the processor may end up being filtered
    /// out by other conditions. Conversely, some candidates may already be filtered out before
    /// being passed to this predicate.
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::{EfficiencyClass, SystemHardware};
    /// use new_zealand::nz;
    ///
    /// // Select only even-numbered performance processors
    /// let filtered_processors = SystemHardware::current()
    ///     .processors()
    ///     .to_builder()
    ///     .filter(|p| p.efficiency_class() == EfficiencyClass::Performance && p.id() % 2 == 0)
    ///     .take(nz!(2));
    ///
    /// if let Some(processors) = filtered_processors {
    ///     for processor in &processors {
    ///         println!(
    ///             "Selected processor {} (performance, even ID)",
    ///             processor.id()
    ///         );
    ///     }
    /// }
    /// ```
    #[must_use]
    pub fn filter(mut self, predicate: impl Fn(&Processor) -> bool) -> Self {
        // We invoke the filters immediately because the API gets really annoying if the
        // predicate has to borrow some stuff because it would need to be 'static and that
        // is cumbersome (since we do not return a generic-lifetimed thing back to the caller).
        for processor in self.candidate_processors() {
            if !predicate(&processor) {
                self.except_indexes.insert(processor.id());
            }
        }

        self
    }

    /// Removes specific processors from the set of candidates.
    ///
    /// # Example
    ///
    /// ```
    /// use std::num::NonZero;
    ///
    /// use many_cpus::SystemHardware;
    ///
    /// // Get the default set and remove the first processor
    /// let all_processors = SystemHardware::current().processors();
    /// let first_processor = all_processors.processors().first().clone();
    ///
    /// let remaining_processors = all_processors
    ///     .to_builder()
    ///     .except([&first_processor])
    ///     .take_all();
    ///
    /// if let Some(processors) = remaining_processors {
    ///     println!(
    ///         "Using {} processors (excluding processor {})",
    ///         processors.len(),
    ///         first_processor.id()
    ///     );
    /// }
    /// ```
    #[must_use]
    pub fn except<'a, I>(mut self, processors: I) -> Self
    where
        I: IntoIterator<Item = &'a Processor>,
    {
        for processor in processors {
            self.except_indexes.insert(processor.id());
        }

        self
    }

    /// Removes processors from the set of candidates if they are not available for use by the
    /// current thread.
    ///
    /// This is a convenient way to identify the set of processors the platform prefers a process
    /// to use when called from a non-customized thread such as the `main()` entrypoint.
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::SystemHardware;
    ///
    /// // Get processors available to the current thread (respects OS affinity settings)
    /// let available_processors = SystemHardware::current()
    ///     .processors()
    ///     .to_builder()
    ///     .where_available_for_current_thread()
    ///     .take_all()
    ///     .expect("current thread must be running on at least one processor");
    ///
    /// println!(
    ///     "Current thread can use {} processors",
    ///     available_processors.len()
    /// );
    ///
    /// // Compare with all processors on the system
    /// let all_processors = SystemHardware::current().all_processors();
    ///
    /// if available_processors.len() < all_processors.len() {
    ///     println!("Thread affinity is restricting processor usage");
    ///     println!("  Total system processors: {}", all_processors.len());
    ///     println!("  Available to this thread: {}", available_processors.len());
    /// }
    /// ```
    #[must_use]
    pub fn where_available_for_current_thread(mut self) -> Self {
        let current_thread_processors = self.hardware.platform().current_thread_processors();

        for processor in self.candidate_processors() {
            if !current_thread_processors.contains(&processor.id()) {
                self.except_indexes.insert(processor.id());
            }
        }

        self
    }

    /// Enforces the process resource quota when determining the maximum number of processors
    /// that can be included in the created processor set.
    ///
    /// By default, the builder does not enforce the resource quota. Call this method to limit
    /// the processor set to the quota. See the type-level documentation for more details on
    /// resource quota handling best practices.
    ///
    /// Note that [`SystemHardware::processors()`] already enforces the resource quota, so you
    /// do not need to call this method when deriving from that processor set.
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::SystemHardware;
    /// use new_zealand::nz;
    ///
    /// // Get all processors and explicitly enforce resource quota.
    /// let quota_processors = SystemHardware::current()
    ///     .all_processors()
    ///     .to_builder()
    ///     .enforce_resource_quota()
    ///     .take(nz!(4));
    /// ```
    ///
    /// [`SystemHardware::processors()`]: crate::SystemHardware::processors
    #[must_use]
    pub fn enforce_resource_quota(mut self) -> Self {
        self.obey_resource_quota = true;
        self
    }

    /// Creates a processor set with a specific number of processors that match the
    /// configured criteria.
    ///
    /// If multiple candidate sets are a match, returns an arbitrary one of them. For example, if
    /// there are six valid candidate processors then `take(4)` may return any four of them.
    ///
    /// Returns `None` if there were not enough candidate processors to satisfy the request.
    ///
    /// # Resource quota
    ///
    /// By default, the resource quota is not enforced. Call [`enforce_resource_quota()`][1]
    /// to limit results to the quota. See the type-level documentation for more details on
    /// resource quota handling best practices.
    ///
    /// [1]: ProcessorSetBuilder::enforce_resource_quota
    #[must_use]
    pub fn take(self, count: NonZero<usize>) -> Option<ProcessorSet> {
        if let Some(max_count) = self.resource_quota_processor_count_limit()
            && count.get() > max_count
        {
            // We cannot satisfy the request.
            return None;
        }

        let candidates = self.candidates_by_memory_region();

        if candidates.is_empty() {
            // No candidates to choose from - everything was filtered out.
            return None;
        }

        let processors = match self.memory_region_selector {
            MemoryRegionSelector::Any => {
                // We do not care about memory regions, so merge into one big happy family and
                // pick a random `count` processors from it. As long as there is enough!
                let all_processors = candidates
                    .values()
                    .flat_map(|x| x.iter().cloned())
                    .collect::<Vec<_>>();

                if all_processors.len() < count.get() {
                    // Not enough processors to satisfy request.
                    return None;
                }

                all_processors
                    .choose_multiple(&mut rng(), count.get())
                    .cloned()
                    .collect_vec()
            }
            MemoryRegionSelector::PreferSame => {
                // We will start decrementing it to zero.
                let count = count.get();

                // We shuffle the memory regions and sort them by size, so if there are memory
                // regions with different numbers of candidates we will prefer the ones with more,
                // although we will consider all memory regions with at least 'count' candidates
                // as equal in sort order to avoid needlessly preferring giant memory regions.
                let mut remaining_memory_regions = candidates.keys().copied().collect_vec();
                remaining_memory_regions.shuffle(&mut rng());
                remaining_memory_regions.sort_unstable_by_key(|x| {
                    candidates
                        .get(x)
                        .expect("region must exist - we just got it from there")
                        .len()
                        // Clamp the length to `count` to treat all larger regions equally.
                        .min(count)
                });
                // We want to start with the largest memory regions first.
                remaining_memory_regions.reverse();

                let mut remaining_memory_regions = VecDeque::from(remaining_memory_regions);

                let mut processors: Vec<Processor> = Vec::with_capacity(count);

                while processors.len() < count {
                    let memory_region = remaining_memory_regions.pop_front()?;

                    let processors_in_region = candidates.get(&memory_region).expect(
                        "we picked an existing key from an existing HashSet - the values must exist",
                    );

                    // There might not be enough to fill the request, which is fine.
                    let choose_count = count.min(processors_in_region.len());

                    let region_processors = processors_in_region
                        .choose_multiple(&mut rng(), choose_count)
                        .cloned();

                    processors.extend(region_processors);
                }

                processors
            }
            MemoryRegionSelector::RequireSame => {
                // We filter out memory regions that do not have enough candidates and pick a
                // random one from the remaining, then picking a random `count` processors.
                let qualifying_memory_regions = candidates
                    .iter()
                    .filter_map(|(region, processors)| {
                        if processors.len() < count.get() {
                            return None;
                        }

                        Some(region)
                    })
                    .collect_vec();

                let memory_region = qualifying_memory_regions.choose(&mut rng())?;

                let processors = candidates.get(memory_region).expect(
                    "we picked an existing key for an existing HashSet - the values must exist",
                );

                processors
                    .choose_multiple(&mut rng(), count.get())
                    .cloned()
                    .collect_vec()
            }
            MemoryRegionSelector::PreferDifferent => {
                // We iterate through the memory regions and prefer one from each, looping through
                // memory regions that still have processors until we have as many as requested.

                // We will start removing processors are memory regions that are used up.
                let mut candidates = candidates;

                let mut processors = Vec::with_capacity(count.get());

                while processors.len() < count.get() {
                    if candidates.is_empty() {
                        // Not enough candidates remaining to satisfy request.
                        return None;
                    }

                    for remaining_processors in candidates.values_mut() {
                        let (index, processor) =
                            remaining_processors.iter().enumerate().choose(&mut rng())?;

                        let processor = processor.clone();

                        remaining_processors.remove(index);

                        processors.push(processor);

                        if processors.len() == count.get() {
                            break;
                        }
                    }

                    // Remove any memory regions that have been depleted.
                    candidates.retain(|_, remaining_processors| !remaining_processors.is_empty());
                }

                processors
            }
            MemoryRegionSelector::RequireDifferent => {
                // We pick random `count` memory regions and a random processor from each.

                if candidates.len() < count.get() {
                    // Not enough memory regions to satisfy request.
                    return None;
                }

                candidates
                    .iter()
                    .choose_multiple(&mut rng(), count.get())
                    .into_iter()
                    .map(|(_, processors)| {
                        processors.iter().choose(&mut rng()).cloned().expect(
                            "we are picking one item from a non-empty list - item must exist",
                        )
                    })
                    .collect_vec()
            }
        };

        Some(ProcessorSet::new(
            NonEmpty::from_vec(processors)?,
            self.hardware,
        ))
    }

    /// Returns a processor set with all processors that match the configured criteria.
    ///
    /// If multiple alternative non-empty sets are a match, returns an arbitrary one of them.
    /// For example, if specifying only a "same memory region" constraint, it will return all
    /// the processors in an arbitrary memory region with at least one qualifying processor.
    ///
    /// Returns `None` if there were no matching processors to satisfy the request.
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::SystemHardware;
    ///
    /// // Try to get all efficiency processors, but exclude the first two
    /// let all_processors = SystemHardware::current().processors();
    /// let first_two: Vec<_> = all_processors.processors().iter().take(2).collect();
    ///
    /// let filtered_processors = SystemHardware::current()
    ///     .processors()
    ///     .to_builder()
    ///     .efficiency_processors_only()
    ///     .except(first_two)
    ///     .take_all();
    ///
    /// match filtered_processors {
    ///     Some(processors) => {
    ///         println!(
    ///             "Found {} efficiency processors (excluding first two)",
    ///             processors.len()
    ///         );
    ///
    ///         // Use remaining efficiency processors for background work
    ///         let threads = processors.spawn_threads(|processor| {
    ///             println!("Background worker on processor {}", processor.id());
    ///             // Background processing here...
    ///         });
    /// # for thread in threads { thread.join().unwrap(); }
    ///     }
    ///     None => {
    ///         // This can happen if all efficiency processors were excluded by the filter
    ///         println!("No efficiency processors remaining after filtering");
    ///     }
    /// }
    /// ```
    ///
    /// # Resource quota
    ///
    /// By default, the resource quota is not enforced. Call [`enforce_resource_quota()`][1]
    /// to limit results to the quota. See the type-level documentation for more details on
    /// resource quota handling best practices.
    ///
    /// [1]: ProcessorSetBuilder::enforce_resource_quota
    #[must_use]
    #[cfg_attr(test, mutants::skip)] // Hangs due to recursive access of OnceLock.
    pub fn take_all(self) -> Option<ProcessorSet> {
        let candidates = self.candidates_by_memory_region();

        if candidates.is_empty() {
            // No candidates to choose from - everything was filtered out.
            return None;
        }

        let processors = match self.memory_region_selector {
            MemoryRegionSelector::Any
            | MemoryRegionSelector::PreferSame
            | MemoryRegionSelector::PreferDifferent => {
                // We return all processors in all memory regions because we have no strong
                // filtering criterion we must follow - all are fine, so we return all.
                candidates
                    .values()
                    .flat_map(|x| x.iter().cloned())
                    .collect()
            }
            MemoryRegionSelector::RequireSame => {
                // We return all processors in a random memory region.
                // The candidate set only contains memory regions with at least 1 processor, so
                // we know that all candidate memory regions are valid and we were not given a
                // count, so even 1 processor is enough to satisfy the "all" criterion.
                let memory_region = candidates
                    .keys()
                    .choose(&mut rng())
                    .expect("we picked a random existing index - element must exist");

                let processors = candidates.get(memory_region).expect(
                    "we picked an existing key for an existing HashSet - the values must exist",
                );

                processors.clone()
            }
            MemoryRegionSelector::RequireDifferent => {
                // We return a random processor from each memory region.
                // The candidate set only contains memory regions with at least 1 processor, so
                // we know that all candidate memory regions have enough to satisfy our needs.
                let processors = candidates.values().map(|processors| {
                    processors
                        .choose(&mut rng())
                        .cloned()
                        .expect("we picked a random item from a non-empty list - item must exist")
                });

                processors.collect()
            }
        };

        let processors = self.reduce_processors_until_under_quota(processors);

        Some(ProcessorSet::new(
            NonEmpty::from_vec(processors)?,
            self.hardware,
        ))
    }

    fn reduce_processors_until_under_quota(&self, processors: Vec<Processor>) -> Vec<Processor> {
        let Some(max_count) = self.resource_quota_processor_count_limit() else {
            return processors;
        };

        let mut processors = processors;

        // If we picked too many, reduce until we are under quota.
        while processors.len() > max_count {
            processors
                .pop()
                .expect("guarded by len-check in loop condition");
        }

        processors
    }

    /// Takes the exact processors specified from the candidate set, returning a new
    /// [`ProcessorSet`] containing only those processors.
    ///
    /// # Panics
    ///
    /// Panics if any of the specified processors are not present in the candidate set.
    ///
    /// # Example
    ///
    /// ```
    /// use many_cpus::SystemHardware;
    /// use nonempty::nonempty;
    ///
    /// let hw = SystemHardware::current();
    /// let candidates = hw.processors();
    ///
    /// // Get the first processor from the set.
    /// let first_processor = candidates.processors().first().clone();
    ///
    /// // Create a new set containing exactly that processor.
    /// let single_processor_set = candidates
    ///     .to_builder()
    ///     .take_exact(nonempty![first_processor]);
    ///
    /// assert_eq!(single_processor_set.len(), 1);
    /// ```
    #[must_use]
    pub fn take_exact(self, processors: NonEmpty<Processor>) -> ProcessorSet {
        let candidates = self.candidates_by_memory_region();
        let candidate_ids: HashSet<_> = candidates
            .values()
            .flat_map(|v| v.iter().map(Processor::id))
            .collect();

        for processor in &processors {
            assert!(
                candidate_ids.contains(&processor.id()),
                "processor {} is not in the candidate set",
                processor.id()
            );
        }

        ProcessorSet::new(processors, self.hardware)
    }

    /// Executes the first stage filters to kick out processors purely based on their individual
    /// characteristics. Whatever pass this filter are valid candidates for selection as long
    /// as the next stage of filtering (the memory region logic) permits it.
    ///
    /// Returns candidates grouped by memory region, with each returned memory region having at
    /// least one candidate processor.
    fn candidates_by_memory_region(&self) -> HashMap<MemoryRegionId, Vec<Processor>> {
        let candidates_iter = self
            .candidate_processors()
            .into_iter()
            .filter_map(move |p| {
                if self.except_indexes.contains(&p.id()) {
                    return None;
                }

                let is_acceptable_type = match self.processor_type_selector {
                    ProcessorTypeSelector::Any => true,
                    ProcessorTypeSelector::Performance => {
                        p.efficiency_class() == EfficiencyClass::Performance
                    }
                    ProcessorTypeSelector::Efficiency => {
                        p.efficiency_class() == EfficiencyClass::Efficiency
                    }
                };

                if !is_acceptable_type {
                    return None;
                }

                Some((p.memory_region_id(), p))
            });

        let mut candidates = HashMap::new();
        for (region, processor) in candidates_iter {
            candidates
                .entry(region)
                .or_insert_with(Vec::new)
                .push(processor);
        }

        candidates
    }

    /// Returns the processors to consider as candidates. If a source processor set was provided
    /// (via `source_processors`), only those processors are returned. Otherwise, all
    /// platform processors are returned.
    fn candidate_processors(&self) -> Vec<Processor> {
        let all = self.all_processors();

        match &self.source_processor_ids {
            Some(source_ids) => all
                .into_iter()
                .filter(|p| source_ids.contains(&p.id()))
                .collect(),
            None => all.into_iter().collect(),
        }
    }

    fn all_processors(&self) -> NonEmpty<Processor> {
        // Cheap conversion, reasonable to do it inline since we do not expect
        // processor set logic to be on the hot path anyway.
        self.hardware
            .platform()
            .get_all_processors()
            .map(Processor::new)
    }

    fn resource_quota_processor_count_limit(&self) -> Option<usize> {
        if self.obey_resource_quota {
            let max_processor_time = self.hardware.platform().max_processor_time();

            // We round down the quota to get a whole number of processors.
            // We specifically round down because our goal with the resource quota is to never
            // exceed it, even by a fraction, as that would cause quality of service degradation.
            #[expect(clippy::cast_sign_loss, reason = "quota cannot be negative")]
            #[expect(
                clippy::cast_possible_truncation,
                reason = "we are correctly rounding to avoid the problem"
            )]
            let max_processor_count = max_processor_time.floor() as usize;

            // We can never restrict to less than 1 processor by quota because that would be
            // nonsense - there is always some available processor time, so at least one
            // processor must be usable. Therefore, we round below 1, and round down above 1.
            Some(max_processor_count.max(1))
        } else {
            None
        }
    }
}

#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
enum MemoryRegionSelector {
    /// The default - memory regions are not considered in processor selection.
    #[default]
    Any,

    /// Processors are all from the same memory region. If there are not enough processors in any
    /// memory region, the request will fail.
    RequireSame,

    /// Processors are all from the different memory regions. If there are not enough memory
    /// regions, the request will fail.
    RequireDifferent,

    /// Processors are ideally all from the same memory region. If there are not enough processors
    /// in a single memory region, more memory regions will be added to the candidate set as needed,
    /// but still keeping it to as few as possible.
    PreferSame,

    /// Processors are ideally all from the different memory regions. If there are not enough memory
    /// regions, multiple processors from the same memory region will be returned, but still keeping
    /// to as many different memory regions as possible to spread the processors out.
    PreferDifferent,
}

#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
enum ProcessorTypeSelector {
    /// The default - all processors are valid candidates.
    #[default]
    Any,

    /// Only performance processors (faster, energy-hungry) are valid candidates.
    ///
    /// Every system is guaranteed to have at least one performance processor.
    /// If all processors are of the same performance class,
    /// they are all considered to be performance processors.
    Performance,

    /// Only efficiency processors (slower, energy-efficient) are valid candidates.
    ///
    /// There is no guarantee that any efficiency processors are present on the system.
    Efficiency,
}

#[cfg(not(miri))] // Miri cannot call platform APIs.
#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests_real {
    use new_zealand::nz;

    use crate::SystemHardware;

    #[test]
    fn spawn_on_any_processor() {
        let set = SystemHardware::current().processors();
        let result = set.spawn_thread(move |_| 1234).join().unwrap();

        assert_eq!(result, 1234);
    }

    #[test]
    fn spawn_on_every_processor() {
        let set = SystemHardware::current().processors();
        let processor_count = set.len();

        let join_handles = set.spawn_threads(move |_| 4567);

        assert_eq!(join_handles.len(), processor_count);

        for handle in join_handles {
            let result = handle.join().unwrap();
            assert_eq!(result, 4567);
        }
    }

    #[test]
    fn filter_by_memory_region() {
        let hw = SystemHardware::current();

        // We know there is at least one memory region, so these must succeed.
        hw.processors()
            .to_builder()
            .same_memory_region()
            .take_all()
            .unwrap();
        hw.processors()
            .to_builder()
            .same_memory_region()
            .take(nz!(1))
            .unwrap();
        hw.processors()
            .to_builder()
            .different_memory_regions()
            .take_all()
            .unwrap();
        hw.processors()
            .to_builder()
            .different_memory_regions()
            .take(nz!(1))
            .unwrap();
    }

    #[test]
    fn filter_by_efficiency_class() {
        let hw = SystemHardware::current();

        // There must be at least one.
        hw.processors()
            .to_builder()
            .performance_processors_only()
            .take_all()
            .unwrap();
        hw.processors()
            .to_builder()
            .performance_processors_only()
            .take(nz!(1))
            .unwrap();

        // There might not be any. We just try resolving it and ignore the result.
        // As long as it does not panic, we are good.
        drop(
            hw.processors()
                .to_builder()
                .efficiency_processors_only()
                .take_all(),
        );
        drop(
            hw.processors()
                .to_builder()
                .efficiency_processors_only()
                .take(nz!(1)),
        );
    }

    #[test]
    fn filter_in_all() {
        let hw = SystemHardware::current();

        // Ensure we use a constant starting set, in case we are running tests under constraints.
        let starting_set = hw.processors();

        // If we filter in all processors, we should get all of them.
        let processors = starting_set
            .to_builder()
            .filter(|_| true)
            .take_all()
            .unwrap();
        let processor_count = starting_set.len();

        assert_eq!(processors.len(), processor_count);
    }

    #[test]
    fn filter_out_all() {
        let hw = SystemHardware::current();

        // If we filter out all processors, there should be nothing left.
        assert!(
            hw.processors()
                .to_builder()
                .filter(|_| false)
                .take_all()
                .is_none()
        );
    }

    #[test]
    fn except_all() {
        let hw = SystemHardware::current();

        // Ensure we use a constant starting set, in case we are running tests under constraints.
        let starting_set = hw.processors();

        // If we exclude all processors, there should be nothing left.
        assert!(
            starting_set
                .to_builder()
                .except(starting_set.processors().iter())
                .take_all()
                .is_none()
        );
    }
}

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
    use std::panic::{RefUnwindSafe, UnwindSafe};

    use new_zealand::nz;
    use nonempty::nonempty;
    use static_assertions::assert_impl_all;

    use super::*;
    use crate::fake::{HardwareBuilder, ProcessorBuilder};

    assert_impl_all!(ProcessorSetBuilder: UnwindSafe, RefUnwindSafe);

    /// Helper to build a `ProcessorBuilder` with the given properties.
    fn proc(id: u32, memory_region: u32, efficiency_class: EfficiencyClass) -> ProcessorBuilder {
        ProcessorBuilder::new()
            .id(id)
            .memory_region(memory_region)
            .efficiency_class(efficiency_class)
    }

    #[test]
    fn smoke_test() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance)),
        );

        let set = hw.processors().to_builder().take_all().unwrap();
        assert_eq!(set.len(), 2);
    }

    #[test]
    fn efficiency_class_filter_take() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .efficiency_processors_only()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), 0);
    }

    #[test]
    fn efficiency_class_filter_take_all() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .efficiency_processors_only()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), 0);
    }

    #[test]
    fn take_n_processors() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .processor(proc(2, 0, EfficiencyClass::Efficiency)),
        );

        let set = hw.processors().to_builder().take(nz!(2)).unwrap();
        assert_eq!(set.len(), 2);
    }

    #[test]
    fn take_n_not_enough_processors() {
        // Configure only 2 processors with low quota so the request for 3 fails.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(2.0),
        );

        let set = hw.processors().to_builder().take(nz!(3));
        assert!(set.is_none());
    }

    #[test]
    fn take_n_not_enough_processor_time_quota() {
        // Configure quota of only 1.0 processor time.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(1.0),
        );

        let set = hw
            .all_processors()
            .to_builder()
            .enforce_resource_quota()
            .take(nz!(2));
        assert!(set.is_none());
    }

    #[test]
    fn take_n_quota_floors_to_limit() {
        // Configure quota of 1.5 processor time. The floor of 1.5 is 1, so requesting 2
        // processors should fail because the quota limit is less than the requested count.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(1.5),
        );

        let set = hw
            .all_processors()
            .to_builder()
            .enforce_resource_quota()
            .take(nz!(2));
        assert!(set.is_none());
    }

    #[test]
    fn take_n_exactly_at_quota_limit() {
        // Configure quota of exactly 2.0 processor time and request exactly 2 processors.
        // This tests the boundary condition where count == max_count (should succeed).
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(2.0),
        );

        let set = hw
            .all_processors()
            .to_builder()
            .enforce_resource_quota()
            .take(nz!(2));
        assert!(set.is_some());
        assert_eq!(set.unwrap().len(), 2);
    }

    #[test]
    fn take_n_under_quota_succeeds() {
        // Configure quota of 3.0 processor time and request only 2 processors.
        // This tests that the quota branch passes through when count < max_count.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .processor(proc(2, 0, EfficiencyClass::Efficiency))
                .max_processor_time(3.0),
        );

        let set = hw
            .all_processors()
            .to_builder()
            .enforce_resource_quota()
            .take(nz!(2));
        assert!(set.is_some());
        assert_eq!(set.unwrap().len(), 2);
    }

    #[test]
    fn take_n_quota_not_enforced_by_default() {
        // Configure quota of only 1.0 processor time. Since quota is not enforced by default,
        // we should still be able to get all processors via all_processors().to_builder().
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(1.0),
        );

        // Verify that hw.processors() respects the quota limit.
        let limited_set = hw.processors();
        assert_eq!(limited_set.len(), 1);

        // Verify that all_processors().to_builder() does NOT enforce quota by default.
        let set = hw.all_processors().to_builder().take(nz!(2));
        assert_eq!(set.unwrap().len(), 2);
    }

    #[test]
    fn take_n_quota_limit_min_1() {
        // Configure quota of only 0.001 processor time, which should round UP to 1.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(0.001),
        );

        let set = hw
            .all_processors()
            .to_builder()
            .enforce_resource_quota()
            .take(nz!(1));
        assert_eq!(set.unwrap().len(), 1);
    }

    #[test]
    fn take_all_rounds_down_quota() {
        // Configure quota of 1.999 which should round DOWN to 1.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(1.999),
        );

        let set = hw
            .all_processors()
            .to_builder()
            .enforce_resource_quota()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
    }

    #[test]
    fn take_all_min_1_despite_quota() {
        // Configure quota of 0.001 which should round UP to 1 (never zero).
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(0.001),
        );

        let set = hw
            .all_processors()
            .to_builder()
            .enforce_resource_quota()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
    }

    #[test]
    fn take_all_not_enough_processors() {
        // All processors are efficiency class, so performance filter should fail.
        let hw = SystemHardware::fake(HardwareBuilder::new().processor(proc(
            0,
            0,
            EfficiencyClass::Efficiency,
        )));

        let set = hw
            .processors()
            .to_builder()
            .performance_processors_only()
            .take_all();
        assert!(set.is_none());
    }

    #[test]
    fn except_filter_take() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance)),
        );

        let builder = hw.processors().to_builder();

        let except_set = builder.clone().filter(|p| p.id() == 0).take_all().unwrap();
        assert_eq!(except_set.len(), 1);

        let set = builder.except(except_set.processors()).take_all().unwrap();
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), 1);
    }

    #[test]
    fn except_filter_take_all() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance)),
        );

        let builder = hw.processors().to_builder();
        let except_set = builder.clone().filter(|p| p.id() == 0).take_all().unwrap();
        assert_eq!(except_set.len(), 1);

        let set = builder.except(except_set.processors()).take_all().unwrap();
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), 1);
    }

    #[test]
    fn custom_filter_take() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .filter(|p| p.id() == 1)
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), 1);
    }

    #[test]
    fn custom_filter_take_all() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .filter(|p| p.id() == 1)
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), 1);
    }

    #[test]
    fn same_memory_region_filter_take() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .same_memory_region()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
    }

    #[test]
    fn same_memory_region_filter_take_all() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .same_memory_region()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
    }

    #[test]
    fn different_memory_region_filter_take() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Efficiency))
                .processor(proc(2, 1, EfficiencyClass::Performance))
                .processor(proc(3, 1, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .different_memory_regions()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 2);

        assert_ne!(
            set.processors().first().memory_region_id(),
            set.processors().last().memory_region_id()
        );
    }

    #[test]
    fn different_memory_region_filter_take_all() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Efficiency))
                .processor(proc(2, 1, EfficiencyClass::Performance))
                .processor(proc(3, 1, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .different_memory_regions()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 2);

        assert_ne!(
            set.processors().first().memory_region_id(),
            set.processors().last().memory_region_id()
        );
    }

    #[test]
    fn filter_combinations() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency)),
        );

        let builder = hw.processors().to_builder();
        let except_set = builder.clone().filter(|p| p.id() == 0).take_all().unwrap();
        let set = builder
            .efficiency_processors_only()
            .except(except_set.processors())
            .different_memory_regions()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), 2);
    }

    #[test]
    fn same_memory_region_take_two_processors() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency)),
        );

        let set = hw
            .processors()
            .to_builder()
            .same_memory_region()
            .take(nz!(2))
            .unwrap();
        assert_eq!(set.len(), 2);
        assert!(set.processors().iter().any(|p| p.id() == 1));
        assert!(set.processors().iter().any(|p| p.id() == 2));
    }

    #[test]
    fn different_memory_region_and_efficiency_class_filters() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 2, EfficiencyClass::Efficiency))
                .processor(proc(3, 3, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .different_memory_regions()
            .efficiency_processors_only()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 2);
        assert!(set.processors().iter().any(|p| p.id() == 0));
        assert!(set.processors().iter().any(|p| p.id() == 2));
    }

    #[test]
    fn performance_processors_but_all_efficiency() {
        let hw = SystemHardware::fake(HardwareBuilder::new().processor(proc(
            0,
            0,
            EfficiencyClass::Efficiency,
        )));

        let set = hw
            .processors()
            .to_builder()
            .performance_processors_only()
            .take_all();
        assert!(set.is_none(), "No performance processors should be found.");
    }

    #[test]
    fn require_different_single_region() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Efficiency)),
        );

        let set = hw
            .processors()
            .to_builder()
            .different_memory_regions()
            .take(nz!(2));
        assert!(
            set.is_none(),
            "Should fail because there is not enough distinct memory regions."
        );
    }

    #[test]
    fn prefer_different_memory_regions_take_all() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency)),
        );

        let set = hw
            .processors()
            .to_builder()
            .prefer_different_memory_regions()
            .take_all()
            .unwrap();
        assert_eq!(set.len(), 3);
    }

    #[test]
    fn prefer_different_memory_regions_take_n() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency))
                .processor(proc(3, 2, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .prefer_different_memory_regions()
            .take(nz!(2))
            .unwrap();
        assert_eq!(set.len(), 2);
        let regions: HashSet<_> = set
            .processors()
            .iter()
            .map(Processor::memory_region_id)
            .collect();
        assert_eq!(regions.len(), 2);
    }

    #[test]
    fn prefer_same_memory_regions_take_n() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency))
                .processor(proc(3, 2, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .prefer_same_memory_region()
            .take(nz!(2))
            .unwrap();
        assert_eq!(set.len(), 2);
        let regions: HashSet<_> = set
            .processors()
            .iter()
            .map(Processor::memory_region_id)
            .collect();
        assert_eq!(regions.len(), 1);
    }

    #[test]
    fn prefer_different_memory_regions_take_n_not_enough() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency))
                .processor(proc(3, 2, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .prefer_different_memory_regions()
            .take(nz!(4))
            .unwrap();
        assert_eq!(set.len(), 4);
    }

    #[test]
    fn prefer_same_memory_regions_take_n_not_enough() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency))
                .processor(proc(3, 2, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .prefer_same_memory_region()
            .take(nz!(3))
            .unwrap();
        assert_eq!(set.len(), 3);
        let regions: HashSet<_> = set
            .processors()
            .iter()
            .map(Processor::memory_region_id)
            .collect();
        assert_eq!(
            2,
            regions.len(),
            "should have picked to minimize memory regions (biggest first)"
        );
    }

    #[test]
    fn prefer_same_memory_regions_take_n_picks_best_fit() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency))
                .processor(proc(3, 2, EfficiencyClass::Performance)),
        );

        let set = hw
            .processors()
            .to_builder()
            .prefer_same_memory_region()
            .take(nz!(2))
            .unwrap();
        assert_eq!(set.len(), 2);
        let regions: HashSet<_> = set
            .processors()
            .iter()
            .map(Processor::memory_region_id)
            .collect();
        assert_eq!(
            1,
            regions.len(),
            "should have picked from memory region 1 which can accommodate the preference"
        );
    }

    #[test]
    fn take_any_returns_none_when_not_enough_processors() {
        // This tests the MemoryRegionSelector::Any branch returning None when there
        // are not enough processors in the candidate set to satisfy the request.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .max_processor_time(10.0),
        );

        // Request more processors than exist (3 > 2). Quota is high so the quota check passes,
        // but there are not enough candidates in the Any branch.
        let set = hw.processors().to_builder().take(nz!(3));
        assert!(
            set.is_none(),
            "should return None when not enough processors available"
        );
    }

    #[test]
    fn take_prefer_different_returns_none_when_candidates_exhausted() {
        // This tests the MemoryRegionSelector::PreferDifferent branch returning None
        // when the round-robin through memory regions exhausts all candidates.
        // Configuration: 2 memory regions with 1 processor each. Request 3 processors.
        // The round-robin will pick 1 from each region (total 2), then have no more
        // candidates to pick from, so it returns None.
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .max_processor_time(10.0),
        );

        // Request 3 processors with prefer_different_memory_regions.
        // We only have 2 total, so after 2 rounds the candidates will be exhausted.
        let set = hw
            .processors()
            .to_builder()
            .prefer_different_memory_regions()
            .take(nz!(3));
        assert!(
            set.is_none(),
            "should return None when candidates exhausted in PreferDifferent mode"
        );
    }

    #[test]
    fn take_exact_returns_set_with_specified_processors() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency))
                .max_processor_time(10.0),
        );

        let candidates = hw.all_processors();
        let p1 = candidates.processors().first().clone();

        let set = candidates.to_builder().take_exact(nonempty![p1.clone()]);
        assert_eq!(set.len(), 1);
        assert_eq!(set.processors().first().id(), p1.id());
    }

    #[test]
    fn take_exact_with_multiple_processors() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 1, EfficiencyClass::Performance))
                .processor(proc(2, 1, EfficiencyClass::Efficiency))
                .max_processor_time(10.0),
        );

        let candidates = hw.all_processors();
        let p0 = candidates
            .processors()
            .iter()
            .find(|p| p.id() == 0)
            .cloned()
            .unwrap();
        let p2 = candidates
            .processors()
            .iter()
            .find(|p| p.id() == 2)
            .cloned()
            .unwrap();

        let set = candidates.to_builder().take_exact(nonempty![p0, p2]);
        assert_eq!(set.len(), 2);
        assert!(set.processors().iter().any(|p| p.id() == 0));
        assert!(set.processors().iter().any(|p| p.id() == 2));
    }

    #[test]
    #[should_panic]
    fn take_exact_panics_if_processor_not_in_candidates() {
        let hw = SystemHardware::fake(
            HardwareBuilder::new()
                .processor(proc(0, 0, EfficiencyClass::Efficiency))
                .processor(proc(1, 0, EfficiencyClass::Performance))
                .max_processor_time(10.0),
        );

        let candidates = hw.all_processors();

        // Explicitly find the processor with ID 0 (not relying on ordering).
        let p0 = candidates
            .processors()
            .iter()
            .find(|p| p.id() == 0)
            .cloned()
            .unwrap();

        // Filter out processor 0, then try to take_exact with it - this should panic.
        drop(
            candidates
                .to_builder()
                .filter(|p| p.id() != 0)
                .take_exact(nonempty![p0]),
        );
    }
}

/// Fallback PAL integration tests - these test the integration between `ProcessorSetBuilder`
/// and the fallback platform abstraction layer.
#[cfg(all(test, not(miri)))] // Miri cannot call platform APIs.
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests_fallback {
    use std::num::NonZero;

    use new_zealand::nz;

    use crate::SystemHardware;
    use crate::pal::fallback::BUILD_TARGET_PLATFORM;

    #[test]
    fn builder_smoke_test() {
        let hw = SystemHardware::fallback();
        let builder = hw.processors().to_builder();

        let set = builder.take_all().unwrap();

        assert!(set.len() >= 1);
    }

    #[test]
    fn take_respects_limit() {
        let hw = SystemHardware::fallback();
        let builder = hw.processors().to_builder();

        let set = builder.take(nz!(1));

        assert!(set.is_some());
        assert_eq!(set.unwrap().len(), 1);
    }

    #[test]
    fn take_all_returns_all() {
        let hw = SystemHardware::fallback();
        let builder = hw.processors().to_builder();

        let set = builder.take_all().unwrap();

        let expected_count = std::thread::available_parallelism()
            .map(NonZero::get)
            .unwrap_or(1);

        assert_eq!(set.len(), expected_count);
    }

    #[test]
    fn performance_only_filter() {
        // All processors on the fallback platform are Performance class.
        let hw = SystemHardware::fallback();
        let builder = hw.processors().to_builder();

        let set = builder.performance_processors_only().take_all().unwrap();

        let expected_count = std::thread::available_parallelism()
            .map(NonZero::get)
            .unwrap_or(1);

        assert_eq!(set.len(), expected_count);
    }

    #[test]
    fn same_memory_region_filter() {
        // All processors on the fallback platform are in memory region 0.
        let hw = SystemHardware::fallback();
        let builder = hw.processors().to_builder();

        let set = builder.same_memory_region().take_all().unwrap();

        let expected_count = std::thread::available_parallelism()
            .map(NonZero::get)
            .unwrap_or(1);

        assert_eq!(set.len(), expected_count);

        for processor in set.processors() {
            assert_eq!(processor.memory_region_id(), 0);
        }
    }

    #[test]
    fn except_filter() {
        let hw = SystemHardware::fallback();
        let builder = hw.processors().to_builder();

        if BUILD_TARGET_PLATFORM.processor_count() > 1 {
            let except_set = builder.clone().filter(|p| p.id() == 0).take_all().unwrap();
            assert_eq!(except_set.len(), 1);

            let set = builder.except(except_set.processors()).take_all().unwrap();

            for processor in set.processors() {
                assert_ne!(processor.id(), 0);
            }
        }
    }

    #[test]
    fn enforce_resource_quota() {
        let hw = SystemHardware::fallback();
        let builder = hw.all_processors().to_builder();

        let set = builder.enforce_resource_quota().take_all().unwrap();

        assert!(set.len() >= 1);
    }

    #[test]
    fn where_available_for_current_thread() {
        use std::thread;

        thread::spawn(|| {
            let hw = SystemHardware::fallback();

            // First pin to one processor.
            let one = hw.processors().to_builder().take(nz!(1)).unwrap();

            one.pin_current_thread_to();

            // Now build a new set inheriting the affinity.
            let set = hw
                .processors()
                .to_builder()
                .where_available_for_current_thread()
                .take_all();

            assert!(set.is_some());
            assert_eq!(set.unwrap().len(), 1);
        })
        .join()
        .unwrap();
    }

    #[test]
    fn quota_not_enforced_by_default_on_builder() {
        let hw = SystemHardware::fallback();
        let builder = hw.all_processors().to_builder();

        let set = builder.take_all().unwrap();

        // The fallback platform reports max_processor_time == processor_count.
        // Since quota is NOT enforced by default, we should get all processors.
        let expected_count = std::thread::available_parallelism()
            .map(NonZero::get)
            .unwrap_or(1);

        assert_eq!(set.len(), expected_count);
    }
}