use thiserror::Error;
#[derive(Debug, Error)]
pub enum CpuAffinityError {
#[error("sched_setaffinity failed: {0}")]
SetAffinity(#[from] zenith_linux::AffinityError),
#[error("CPU {cpu} out of range (max: {max})")]
CpuOutOfRange {
cpu: usize,
max: usize,
},
#[error("NUMA node {node} not found")]
NumaNodeNotFound {
node: u32,
},
#[error("worker {worker_id} not found in affinity configs (count: {config_count})")]
WorkerNotFound {
worker_id: u32,
config_count: usize,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CpuId {
pub id: usize,
pub node: u32,
}
impl CpuId {
pub fn new(id: usize, node: u32) -> Self {
Self { id, node }
}
}
pub fn set_thread_affinity(cpu: usize) -> Result<(), CpuAffinityError> {
zenith_linux::set_thread_affinity(cpu).map_err(CpuAffinityError::SetAffinity)
}
pub fn set_thread_affinity_range(cpus: &[usize]) -> Result<(), CpuAffinityError> {
zenith_linux::set_thread_affinity_range(cpus).map_err(CpuAffinityError::SetAffinity)
}
pub fn get_online_cpus() -> usize {
zenith_linux::online_cpu_count()
}
#[derive(Debug, Clone)]
pub struct CpuAffinityConfig {
pub worker_id: u32,
pub cpu_ids: Vec<usize>,
pub numa_node: u32,
}
impl CpuAffinityConfig {
pub fn new(worker_id: u32, cpu_ids: Vec<usize>, numa_node: u32) -> Self {
Self { worker_id, cpu_ids, numa_node }
}
pub fn apply(&self) -> Result<(), CpuAffinityError> {
let online = get_online_cpus();
for &cpu in &self.cpu_ids {
if cpu >= online {
return Err(CpuAffinityError::CpuOutOfRange { cpu, max: online });
}
}
if self.cpu_ids.len() == 1 {
set_thread_affinity(self.cpu_ids[0])
} else {
set_thread_affinity_range(&self.cpu_ids)
}
}
}
#[derive(Debug)]
pub struct AffinityManager {
configs: Vec<CpuAffinityConfig>,
}
impl AffinityManager {
pub fn new() -> Self {
Self { configs: Vec::new() }
}
pub fn add_config(&mut self, config: CpuAffinityConfig) {
self.configs.push(config);
}
pub fn apply_all(&self) -> Vec<(u32, Result<(), CpuAffinityError>)> {
self.configs
.iter()
.map(|c| {
let result = c.apply();
(c.worker_id, result)
})
.collect()
}
pub fn apply_for_worker(&self, worker_id: u32) -> Result<(), CpuAffinityError> {
let config = self
.configs
.iter()
.find(|c| c.worker_id == worker_id)
.ok_or(CpuAffinityError::WorkerNotFound {
worker_id,
config_count: self.configs.len(),
})?;
config.apply()
}
pub fn configs(&self) -> &[CpuAffinityConfig] {
&self.configs
}
}
impl Default for AffinityManager {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_cpu_id() {
let cpu = CpuId::new(3, 0);
assert_eq!(cpu.id, 3);
assert_eq!(cpu.node, 0);
}
#[test]
fn test_affinity_config() {
let config = CpuAffinityConfig::new(1, vec![2, 3], 0);
assert_eq!(config.worker_id, 1);
assert_eq!(config.cpu_ids, vec![2, 3]);
assert_eq!(config.numa_node, 0);
}
#[test]
fn test_affinity_manager() {
let mut mgr = AffinityManager::new();
mgr.add_config(CpuAffinityConfig::new(0, vec![0], 0));
mgr.add_config(CpuAffinityConfig::new(1, vec![1], 0));
assert_eq!(mgr.configs().len(), 2);
}
#[test]
fn test_apply_validates_cpu_bounds() {
let max = get_online_cpus().max(1);
let config = CpuAffinityConfig::new(1, vec![max + 100], 0);
let err = config.apply().unwrap_err();
assert!(matches!(err, CpuAffinityError::CpuOutOfRange { .. }));
}
#[test]
fn test_get_online_cpus() {
let cpus = get_online_cpus();
assert!(cpus >= 1);
}
#[cfg(target_os = "linux")]
#[test]
fn test_set_affinity_valid_cpu() {
let max = get_online_cpus();
if max > 0 {
let result = set_thread_affinity(0);
assert!(result.is_ok());
}
}
}