mod executor;
mod notify;
mod spsc;
mod state;
#[cfg(test)]
mod tests;
use alloc::{boxed::Box, collections::VecDeque, format, string::String, sync::Arc, vec, vec::Vec};
use core::{
num::NonZeroUsize,
sync::atomic::{AtomicBool, AtomicU8, Ordering},
};
use ax_sync::SpinLock;
use ax_task::{sched::CpuSet, sync::WaitQueue};
use irq_framework::IrqId;
use rd_net::{
NetError, NetHardIrqEndpoint, NetHardIrqResult, NetIrqSourceId, PreparedNetDevice,
WifiLinkPolicy, WifiTransaction,
};
pub use self::state::NetQueueStats;
use self::{executor::*, notify::QueueNotification, spsc::*, state::PollGroupState};
use crate::device::{EthernetFramePort, EthernetFramePortList};
const QUEUE_BUDGET: usize = 64;
const CPU_ROUND_BUDGET: usize = 256;
const WIFI_CONTROL_QUEUE_CAPACITY: usize = 8;
const STATE_IDLE: u8 = 0;
const STATE_SCHEDULED: u8 = 1;
const STATE_POLLING: u8 = 2;
const STATE_DISABLED: u8 = 3;
const STATE_MASK: u8 = 0x0f;
const STATE_MISSED: u8 = 0x80;
const COMMAND_WAIT: u8 = 0;
const COMMAND_START: u8 = 1;
const COMMAND_STOP: u8 = 2;
const COMMAND_QUARANTINE: u8 = 3;
const COMMAND_RUN: u8 = 4;
const COMMAND_PRUNE: u8 = 5;
const STATUS_PENDING: u8 = 0;
const STATUS_READY: u8 = 1;
const STATUS_FAILED: u8 = 2;
const STATUS_EMPTY: u8 = 3;
struct WifiCommandCompletion {
result: SpinLock<Option<Result<(), NetError>>>,
wait: WaitQueue,
}
impl WifiCommandCompletion {
fn new() -> Self {
Self {
result: SpinLock::new(None),
wait: WaitQueue::new(),
}
}
fn complete(&self, result: Result<(), NetError>) {
*self.result.lock_irqsave() = Some(result);
self.wait.notify_all();
}
fn wait(&self) -> Result<(), NetError> {
self.wait
.wait_until(|| self.result.lock_irqsave().is_some());
self.result
.lock_irqsave()
.take()
.expect("Wi-Fi completion was published without a result")
}
}
struct WifiControlRequest {
transaction: WifiTransaction,
completion: Arc<WifiCommandCompletion>,
}
struct WifiControlQueue {
requests: SpinLock<VecDeque<WifiControlRequest>>,
stopped: AtomicBool,
}
impl WifiControlQueue {
fn new() -> Self {
Self {
requests: SpinLock::new(VecDeque::with_capacity(WIFI_CONTROL_QUEUE_CAPACITY)),
stopped: AtomicBool::new(false),
}
}
fn submit(
&self,
transaction: WifiTransaction,
notify: &QueueNotification,
) -> Result<(), NetError> {
if self.stopped.load(Ordering::Acquire) {
return Err(NetError::Stopped);
}
let completion = Arc::new(WifiCommandCompletion::new());
{
let mut requests = self.requests.lock_irqsave();
if self.stopped.load(Ordering::Acquire) {
return Err(NetError::Stopped);
}
if requests.len() == WIFI_CONTROL_QUEUE_CAPACITY {
return Err(NetError::Retry);
}
requests.push_back(WifiControlRequest {
transaction,
completion: Arc::clone(&completion),
});
}
notify.notify();
completion.wait()
}
fn try_pop(&self) -> Option<WifiControlRequest> {
self.requests.lock_irqsave().pop_front()
}
fn has_pending(&self) -> bool {
!self.requests.lock_irqsave().is_empty()
}
fn stop(&self) {
self.stopped.store(true, Ordering::Release);
let pending = core::mem::take(&mut *self.requests.lock_irqsave());
for request in pending {
request.completion.complete(Err(NetError::Stopped));
}
}
}
#[derive(Clone)]
pub(crate) struct WifiRuntimeHandle {
device_index: usize,
owner_cpu: usize,
queue: Arc<WifiControlQueue>,
notify: Arc<QueueNotification>,
startup_group: Arc<PollGroupState>,
}
impl WifiRuntimeHandle {
pub(crate) const fn device_index(&self) -> usize {
self.device_index
}
pub(crate) const fn owner_cpu(&self) -> usize {
self.owner_cpu
}
pub(crate) fn submit(&self, transaction: WifiTransaction) -> Result<(), NetError> {
self.queue.submit(transaction, &self.notify)
}
}
#[derive(Debug, thiserror::Error)]
pub enum NetworkRuntimeError {
#[error("network device parts are inconsistent with their IRQ bindings")]
InvalidTopology,
#[error("network queue executor could not be pinned to CPU {0}")]
WorkerAffinity(usize),
#[error("network queue executor for CPU {cpu} could not be spawned: {source}")]
WorkerSpawn {
cpu: usize,
#[source]
source: ax_task::thread::TaskError,
},
#[error("network queue initialization failed: {0}")]
QueueInit(NetError),
#[error("network IRQ registration failed: {0}")]
IrqRegistration(#[from] PinnedNetIrqError),
#[error("network DMA setup failed: {0}")]
Device(#[from] NetError),
#[error("secure Wi-Fi startup entropy failed: {0}")]
StartupEntropy(#[from] crate::NetError),
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ResolvedNetIrqSource {
pub source_id: NetIrqSourceId,
pub irq: IrqId,
}
pub struct NetworkDeviceInput {
pub name: String,
pub device: PreparedNetDevice,
pub irq_sources: Vec<ResolvedNetIrqSource>,
pub tx_queue_discipline: TxQueueDiscipline,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TxQueueDiscipline {
NoQueue,
Fifo { max_frames: NonZeroUsize },
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PinnedNetIrqOutcome {
Unhandled,
Handled,
Wake,
}
pub struct PinnedNetIrqAction {
handler: Box<dyn FnMut() -> PinnedNetIrqOutcome + Send>,
}
impl PinnedNetIrqAction {
pub fn new(handler: impl FnMut() -> PinnedNetIrqOutcome + Send + 'static) -> Self {
Self {
handler: Box::new(handler),
}
}
pub fn run(&mut self) -> PinnedNetIrqOutcome {
(self.handler)()
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
pub enum PinnedNetIrqError {
#[error("invalid network IRQ or owner CPU")]
Invalid,
#[error("network IRQ affinity conflicts with an existing shared action")]
AffinityConflict,
#[error("fixed network IRQ routing is unsupported")]
Unsupported,
#[error("network IRQ operation failed")]
Other,
}
pub trait PinnedNetIrqRegistration: Send + Sync + 'static {
fn owner_cpu(&self) -> usize;
fn enable(&self) -> Result<(), PinnedNetIrqError>;
fn disable_and_synchronize(&self) -> Result<(), PinnedNetIrqError>;
}
pub trait PinnedNetIrqRegistrar: Sync {
fn register(
&self,
name: String,
irq: IrqId,
owner_cpu: usize,
action: PinnedNetIrqAction,
) -> Result<Box<dyn PinnedNetIrqRegistration>, PinnedNetIrqError>;
}
struct EndpointToRegister {
name: String,
irq: IrqId,
owner_cpu: usize,
endpoint: NetHardIrqEndpoint,
shared: Arc<PollGroupState>,
}
struct RegisteredEndpoint {
registration: Box<dyn PinnedNetIrqRegistration>,
shared: Arc<PollGroupState>,
}
pub struct NetworkQueueRuntime {
registrations: Vec<Box<dyn PinnedNetIrqRegistration>>,
executors: Vec<ExecutorLease>,
group_states: Vec<Arc<PollGroupState>>,
_controls: Vec<Box<dyn rd_net::NetControlEndpoint>>,
wifi_handles: Vec<WifiRuntimeHandle>,
initial_wifi_policies: Vec<(usize, WifiLinkPolicy)>,
device_index_map: Vec<Option<usize>>,
protocol_owner_cpu: usize,
}
impl NetworkQueueRuntime {
pub(crate) fn discovery_order(&self, device_index: usize) -> usize {
self.device_index_map
.iter()
.position(|&index| index == Some(device_index))
.expect("published network port must have a discovery order")
}
pub fn protocol_owner_cpu(&self) -> usize {
self.protocol_owner_cpu
}
pub fn stats(&self) -> Vec<NetQueueStats> {
self.group_states
.iter()
.map(|state| state.stats.snapshot(state.owner_cpu))
.collect()
}
pub(crate) fn wifi_handle(&self, device_index: usize) -> Option<WifiRuntimeHandle> {
self.wifi_handles
.iter()
.find(|handle| handle.device_index() == device_index)
.cloned()
}
pub(crate) fn initial_wifi_policy(&self, device_index: usize) -> Option<WifiLinkPolicy> {
self.initial_wifi_policies
.iter()
.find_map(|(index, policy)| (*index == device_index).then_some(*policy))
}
}
impl Drop for NetworkQueueRuntime {
fn drop(&mut self) {
for handle in self.wifi_handles.iter().rev() {
handle.queue.stop();
handle.notify.notify();
}
let registrations = core::mem::take(&mut self.registrations);
let irq_synchronized = release_registrations(registrations);
let runtime_side_resources = (
core::mem::take(&mut self.group_states),
core::mem::take(&mut self._controls),
core::mem::take(&mut self.wifi_handles),
);
stop_executors(&mut self.executors, irq_synchronized);
release_runtime_side_resources(runtime_side_resources, irq_synchronized);
}
}
pub struct NetworkRuntimeBuilder<'a> {
devices: Vec<NetworkDeviceInput>,
registrar: &'a dyn PinnedNetIrqRegistrar,
active_cpus: CpuSet,
}
impl<'a> NetworkRuntimeBuilder<'a> {
pub fn new(
devices: Vec<NetworkDeviceInput>,
registrar: &'a dyn PinnedNetIrqRegistrar,
active_cpus: CpuSet,
) -> Self {
Self {
devices,
registrar,
active_cpus,
}
}
pub fn build(
self,
) -> Result<(NetworkQueueRuntime, EthernetFramePortList), NetworkRuntimeError> {
let topology_len = self.active_cpus.topology_len();
let active_cpus = self
.active_cpus
.iter()
.map(ax_task::sched::CpuId::as_usize)
.collect::<Vec<_>>();
if topology_len == 0 || active_cpus.is_empty() {
release_or_quarantine(self.devices, false);
return Err(NetworkRuntimeError::InvalidTopology);
}
let group_irq_sets = match validate_and_collect_irq_sets(&self.devices) {
Ok(sets) => sets,
Err(error) => {
core::mem::forget(self.devices);
return Err(error);
}
};
let group_owners = assign_affinity_domains(&group_irq_sets, &active_cpus);
let mut groups_by_cpu = (0..topology_len)
.map(|_| Vec::new())
.collect::<Vec<Vec<QueueGroupExecutor>>>();
let mut wifi_by_cpu = (0..topology_len)
.map(|_| Vec::new())
.collect::<Vec<Vec<WifiExecutorSlot>>>();
let cpu_notifies = (0..topology_len)
.map(|_| Arc::new(QueueNotification::new()))
.collect::<Vec<_>>();
let mut endpoints = Vec::new();
let mut ports = Vec::with_capacity(self.devices.len());
let mut controls = Vec::new();
let mut port_macs = Vec::new();
let mut wifi_handles = Vec::new();
let mut startup_transactions = Vec::new();
let mut group_states = Vec::new();
let mut flat_group = 0;
for (device_index, input) in self.devices.into_iter().enumerate() {
let port_name = input.name.clone();
let PreparedNetDevice {
info,
control,
wifi_control,
poll_groups,
} = input.device;
let mut protocol_groups = Vec::with_capacity(poll_groups.len());
let mut checksum_capabilities = None;
let mut wifi_target = None;
let mut device_group_locations = Vec::with_capacity(poll_groups.len());
for mut group in poll_groups {
checksum_capabilities = Some(checksum_capabilities.map_or(
group.tx.checksum_capabilities(),
|current: rd_net::TxChecksumCapabilities| {
current.intersection(group.tx.checksum_capabilities())
},
));
let owner_cpu = group_owners[flat_group];
let owner_group_index = groups_by_cpu[owner_cpu].len();
let shared = Arc::new(PollGroupState::new(
owner_cpu,
Arc::clone(&cpu_notifies[owner_cpu]),
));
let rx_capacity = group.rx.capacity();
let (rx_ready_tx, rx_ready_rx) = spsc_ring(rx_capacity);
let (rx_recycle_tx, rx_recycle_rx) = spsc_ring(rx_capacity);
let (tx_ready_tx, tx_ready_rx) = spsc_ring(group.tx.capacity());
let (tx_free_tx, tx_free_rx) = spsc_ring(group.tx.capacity());
let rx_recycler = Arc::new(RxRecycler::new(
rx_recycle_tx,
Arc::clone(&shared),
rx_capacity,
));
for endpoint in group.irq_endpoints.drain(..) {
let irq = resolve_endpoint_irq(&input.irq_sources, endpoint.source_id())
.expect("network IRQ topology was validated before ownership transfer");
endpoints.push(EndpointToRegister {
name: format!(
"{}-g{}-s{}",
input.name,
group.id.get(),
endpoint.source_id().get()
),
irq,
owner_cpu,
endpoint,
shared: Arc::clone(&shared),
});
}
protocol_groups.push(ProtocolGroupPort {
rx_ready: rx_ready_rx,
rx_recycler: Arc::clone(&rx_recycler),
tx_ready: tx_ready_tx,
tx_free: tx_free_rx,
tx_spares: Vec::with_capacity(group.tx.capacity()),
shared: Arc::clone(&shared),
});
groups_by_cpu[owner_cpu].push(QueueGroupExecutor {
wifi_startup_group: None,
group,
rx_ready: rx_ready_tx,
rx_recycle: rx_recycle_rx,
rx_recycler,
rx_spares: Vec::with_capacity(rx_capacity.max(QUEUE_BUDGET)),
rx_extra_buffers: 0,
tx_ready: tx_ready_rx,
tx_free: tx_free_tx,
pending_rx: None,
pending_rx_refill: VecDeque::with_capacity(rx_capacity),
pending_tx: None,
pending_tx_free: None,
retry_at: None,
shared: Arc::clone(&shared),
});
wifi_target.get_or_insert((owner_cpu, owner_group_index, Arc::clone(&shared)));
device_group_locations.push((owner_cpu, owner_group_index));
group_states.push(shared);
flat_group += 1;
}
if let Some(wifi_control) = wifi_control {
let (owner_cpu, group_index, startup_group) =
wifi_target.ok_or(NetworkRuntimeError::InvalidTopology)?;
for (group_cpu, local_index) in device_group_locations {
groups_by_cpu[group_cpu][local_index].wifi_startup_group =
Some(Arc::clone(&startup_group));
}
let queue = Arc::new(WifiControlQueue::new());
let handle = WifiRuntimeHandle {
device_index,
owner_cpu,
queue: Arc::clone(&queue),
notify: Arc::clone(&cpu_notifies[owner_cpu]),
startup_group,
};
if let Some(transaction) = wifi_control.startup_transaction() {
startup_transactions.push((handle.clone(), transaction));
}
wifi_by_cpu[owner_cpu].push(WifiExecutorSlot {
group_index,
control: wifi_control,
queue,
active: None,
});
wifi_handles.push(handle);
}
controls.push(control);
let port_mac = Arc::new(SpinLock::new(info.mac_address));
port_macs.push(Arc::clone(&port_mac));
ports.push(QueueFramePort {
name: port_name,
mac: port_mac,
groups: protocol_groups,
tx_queue_discipline: input.tx_queue_discipline,
pending_tx: VecDeque::new(),
next_rx: 0,
next_tx: 0,
checksum_capabilities: checksum_capabilities
.unwrap_or(rd_net::TxChecksumCapabilities::NONE),
});
}
let mut executors = Vec::new();
for (owner_cpu, (groups, wifi)) in groups_by_cpu.into_iter().zip(wifi_by_cpu).enumerate() {
if groups.is_empty() {
continue;
}
let control = Arc::new(ExecutorControl {
owner_cpu,
command: AtomicU8::new(COMMAND_WAIT),
affinity_status: AtomicU8::new(STATUS_PENDING),
startup_status: AtomicU8::new(STATUS_PENDING),
prune_status: AtomicU8::new(STATUS_PENDING),
publication_status: AtomicU8::new(STATUS_PENDING),
startup_error: SpinLock::new(None),
notify: Arc::clone(&cpu_notifies[owner_cpu]),
});
let mut affinity = CpuSet::empty(topology_len);
if !affinity.insert(ax_task::sched::CpuId::new(owner_cpu as u32)) {
stop_executors(&mut executors, true);
return Err(NetworkRuntimeError::InvalidTopology);
}
let task_control = Arc::clone(&control);
let task =
match ax_task::thread::ThreadBuilder::new(format!("net-queue-cpu{owner_cpu}"))
.affinity(affinity)
.spawn(move || queue_executor_main(groups, wifi, task_control))
{
Ok(task) => task,
Err(source) => {
stop_executors(&mut executors, true);
return Err(NetworkRuntimeError::WorkerSpawn {
cpu: owner_cpu,
source,
});
}
};
executors.push(ExecutorLease { control, task });
}
let failed_owner = executors.iter().find_map(|executor| {
wait_status(&executor.control.affinity_status);
(executor.control.affinity_status.load(Ordering::Acquire) != STATUS_READY)
.then_some(executor.control.owner_cpu)
});
if let Some(owner_cpu) = failed_owner {
stop_executors(&mut executors, true);
return Err(NetworkRuntimeError::WorkerAffinity(owner_cpu));
}
let mut registrations = Vec::new();
let mut endpoint_iter = endpoints.into_iter();
while let Some(mut endpoint) = endpoint_iter.next() {
let shared = Arc::clone(&endpoint.shared);
let registration_state = Arc::clone(&endpoint.shared);
let owner_cpu = endpoint.owner_cpu;
let action = PinnedNetIrqAction::new(move || match endpoint.endpoint.handle_irq() {
NetHardIrqResult::Spurious => {
shared.stats.spurious.fetch_add(1, Ordering::Relaxed);
PinnedNetIrqOutcome::Unhandled
}
NetHardIrqResult::Schedule(_snapshot) => {
shared.schedule_irq();
PinnedNetIrqOutcome::Wake
}
NetHardIrqResult::ProbeDeferred => {
shared.stats.probe_deferred.fetch_add(1, Ordering::Relaxed);
shared.schedule_irq();
PinnedNetIrqOutcome::Wake
}
});
let registration =
match self
.registrar
.register(endpoint.name, endpoint.irq, owner_cpu, action)
{
Ok(registration) if registration.owner_cpu() == owner_cpu => registration,
Ok(registration) => {
registrations.push(RegisteredEndpoint {
registration,
shared: registration_state,
});
let irq_synchronized = release_registered_endpoints(registrations);
stop_executors(&mut executors, irq_synchronized);
release_runtime_side_resources(
(
controls,
ports,
port_macs,
wifi_handles,
startup_transactions,
group_states,
cpu_notifies,
endpoint_iter,
),
irq_synchronized,
);
return Err(NetworkRuntimeError::InvalidTopology);
}
Err(error) => {
let irq_synchronized = release_registered_endpoints(registrations);
stop_executors(&mut executors, irq_synchronized);
release_runtime_side_resources(
(
controls,
ports,
port_macs,
wifi_handles,
startup_transactions,
group_states,
cpu_notifies,
endpoint_iter,
),
irq_synchronized,
);
return Err(error.into());
}
};
registrations.push(RegisteredEndpoint {
registration,
shared: registration_state,
});
}
drop(endpoint_iter);
for registration in ®istrations {
if let Err(error) = registration.registration.enable() {
let irq_synchronized = release_registered_endpoints(registrations);
stop_executors(&mut executors, irq_synchronized);
release_runtime_side_resources(
(
controls,
ports,
port_macs,
wifi_handles,
startup_transactions,
group_states,
cpu_notifies,
),
irq_synchronized,
);
return Err(error.into());
}
}
for command in [COMMAND_START, COMMAND_PRUNE] {
for executor in &executors {
executor.control.command.store(command, Ordering::Release);
executor.control.notify.notify();
}
for executor in &executors {
let status = if command == COMMAND_START {
&executor.control.startup_status
} else {
&executor.control.prune_status
};
wait_status(status);
if status.load(Ordering::Acquire) != STATUS_READY {
let error = executor
.control
.startup_error
.lock_irqsave()
.take()
.unwrap_or(NetError::InvalidParts);
let irq_synchronized = release_registered_endpoints(registrations);
stop_executors(&mut executors, irq_synchronized);
release_runtime_side_resources(
(
controls,
ports,
port_macs,
wifi_handles,
startup_transactions,
group_states,
cpu_notifies,
),
irq_synchronized,
);
return Err(NetworkRuntimeError::QueueInit(error));
}
}
}
let registrations = match prune_absent_irq_registrations(registrations) {
Ok(registrations) => registrations,
Err(error) => {
stop_executors(&mut executors, false);
release_runtime_side_resources(
(
controls,
ports,
port_macs,
wifi_handles,
startup_transactions,
group_states,
cpu_notifies,
),
false,
);
return Err(error.into());
}
};
publish_executors(&mut executors);
let (started_ports, device_index_map) = retain_started_ports(ports);
group_states.retain(|state| !state.startup_absent());
let controls = controls
.into_iter()
.enumerate()
.filter_map(|(index, control)| device_index_map[index].map(|_| control))
.collect::<Vec<_>>();
let port_macs = port_macs
.into_iter()
.enumerate()
.filter_map(|(index, mac)| device_index_map[index].map(|_| mac))
.collect::<Vec<_>>();
let wifi_handles = wifi_handles
.into_iter()
.filter_map(|mut handle| {
if handle.startup_group.startup_absent() {
return None;
}
handle.device_index = device_index_map[handle.device_index]?;
Some(handle)
})
.collect::<Vec<_>>();
let startup_transactions = startup_transactions
.into_iter()
.filter_map(|(mut handle, transaction)| {
if handle.startup_group.startup_absent() {
return None;
}
handle.device_index = device_index_map[handle.device_index]?;
Some((handle, transaction))
})
.collect::<Vec<_>>();
let active_group_owners = group_states
.iter()
.map(|state| state.owner_cpu)
.collect::<Vec<_>>();
let protocol_owner_cpu = select_protocol_owner(&active_group_owners, &active_cpus);
let mut runtime = NetworkQueueRuntime {
registrations,
executors,
group_states,
_controls: controls,
wifi_handles,
initial_wifi_policies: Vec::new(),
device_index_map,
protocol_owner_cpu,
};
for (handle, transaction) in startup_transactions {
let transaction =
prepare_startup_transaction(transaction, super::next_wifi_connection_entropy)?;
let policy = transaction.link_policy();
handle.submit(transaction)?;
if let Some(policy) = policy {
runtime
.initial_wifi_policies
.push((handle.device_index(), policy));
}
}
for (control, mac) in runtime._controls.iter_mut().zip(port_macs) {
let address = control.mac_address()?;
*mac.lock_irqsave() = address;
}
Ok((runtime, started_ports))
}
}
fn retain_started_ports(ports: Vec<QueueFramePort>) -> (EthernetFramePortList, Vec<Option<usize>>) {
let mut device_index_map = vec![None; ports.len()];
let mut started_ports = Vec::with_capacity(ports.len());
for (device_index, mut port) in ports.into_iter().enumerate() {
if port.retain_started_groups() {
device_index_map[device_index] = Some(started_ports.len());
started_ports.push(Box::new(port) as Box<dyn EthernetFramePort>);
} else {
log::warn!(
"network device {} is not present after owner startup; skipping it",
port.name
);
}
}
(started_ports, device_index_map)
}
fn prepare_startup_transaction(
mut transaction: WifiTransaction,
next_entropy: impl FnOnce() -> Result<[u8; 32], crate::NetError>,
) -> Result<WifiTransaction, crate::NetError> {
if transaction.needs_connect_entropy() {
transaction.provide_connect_entropy(next_entropy()?);
log::info!("[wifi] secure startup connection entropy prepared");
}
Ok(transaction)
}
fn validate_and_collect_irq_sets(
devices: &[NetworkDeviceInput],
) -> Result<Vec<Vec<IrqId>>, NetworkRuntimeError> {
let mut sets = Vec::new();
for input in devices {
if input.device.poll_groups.is_empty() || input.irq_sources.is_empty() {
return Err(NetworkRuntimeError::InvalidTopology);
}
for group in &input.device.poll_groups {
if group.irq_endpoints.is_empty() {
return Err(NetworkRuntimeError::InvalidTopology);
}
let mut irqs = Vec::new();
for endpoint in &group.irq_endpoints {
let irq = resolve_endpoint_irq(&input.irq_sources, endpoint.source_id())?;
if !irqs.contains(&irq) {
irqs.push(irq);
}
}
sets.push(irqs);
}
for source in &input.irq_sources {
let used = input.device.poll_groups.iter().any(|group| {
group
.irq_endpoints
.iter()
.any(|endpoint| endpoint.source_id() == source.source_id)
});
if !used {
return Err(NetworkRuntimeError::InvalidTopology);
}
}
}
Ok(sets)
}
fn resolve_endpoint_irq(
sources: &[ResolvedNetIrqSource],
source_id: NetIrqSourceId,
) -> Result<IrqId, NetworkRuntimeError> {
let mut matches = sources
.iter()
.filter(|source| source.source_id == source_id)
.map(|source| source.irq);
let irq = matches.next().ok_or(NetworkRuntimeError::InvalidTopology)?;
if matches.next().is_some() {
return Err(NetworkRuntimeError::InvalidTopology);
}
Ok(irq)
}
fn assign_affinity_domains(irq_sets: &[Vec<IrqId>], active_cpus: &[usize]) -> Vec<usize> {
let mut parents = (0..irq_sets.len()).collect::<Vec<_>>();
for left in 0..irq_sets.len() {
for right in (left + 1)..irq_sets.len() {
if irq_sets[left]
.iter()
.any(|irq| irq_sets[right].contains(irq))
{
union(&mut parents, left, right);
}
}
}
let mut roots = Vec::new();
let mut owners = Vec::with_capacity(irq_sets.len());
for index in 0..irq_sets.len() {
let root = find(&mut parents, index);
let domain_index = match roots.iter().position(|candidate| *candidate == root) {
Some(index) => index,
None => {
roots.push(root);
roots.len() - 1
}
};
owners.push(active_cpus[domain_index % active_cpus.len()]);
}
owners
}
fn find(parents: &mut [usize], mut index: usize) -> usize {
while parents[index] != index {
let grandparent = parents[parents[index]];
parents[index] = grandparent;
index = grandparent;
}
index
}
fn union(parents: &mut [usize], left: usize, right: usize) {
let left_root = find(parents, left);
let right_root = find(parents, right);
if left_root != right_root {
let (first, second) = if left_root < right_root {
(left_root, right_root)
} else {
(right_root, left_root)
};
parents[second] = first;
}
}
fn select_protocol_owner(group_owners: &[usize], active_cpus: &[usize]) -> usize {
active_cpus
.iter()
.copied()
.min_by_key(|cpu| {
(
group_owners.iter().filter(|owner| **owner == *cpu).count(),
*cpu,
)
})
.unwrap_or(0)
}
fn wait_status(status: &AtomicU8) {
while status.load(Ordering::Acquire) == STATUS_PENDING {
crate::yield_network_thread();
}
}
fn release_registered_endpoints(registrations: Vec<RegisteredEndpoint>) -> bool {
release_registrations(
registrations
.into_iter()
.map(|registered| registered.registration)
.collect(),
)
}
fn prune_absent_irq_registrations(
registrations: Vec<RegisteredEndpoint>,
) -> Result<Vec<Box<dyn PinnedNetIrqRegistration>>, PinnedNetIrqError> {
let mut started = Vec::new();
let mut absent = Vec::new();
for registered in registrations {
if registered.shared.startup_absent() {
absent.push(registered.registration);
} else {
started.push(registered.registration);
}
}
if release_registrations(absent) {
Ok(started)
} else {
let _ = release_registrations(started);
Err(PinnedNetIrqError::Other)
}
}
fn disable_registrations(registrations: &[Box<dyn PinnedNetIrqRegistration>]) -> bool {
let mut synchronized = true;
for registration in registrations.iter().rev() {
if registration.disable_and_synchronize().is_err() {
synchronized = false;
}
}
synchronized
}
fn release_registrations(registrations: Vec<Box<dyn PinnedNetIrqRegistration>>) -> bool {
let synchronized = disable_registrations(®istrations);
if synchronized {
drop(registrations);
} else {
log::warn!(
"quarantining {} network IRQ registrations because callback synchronization failed",
registrations.len()
);
core::mem::forget(registrations);
}
synchronized
}
fn release_runtime_side_resources<T>(resource: T, irq_synchronized: bool) {
release_or_quarantine(resource, irq_synchronized);
}
fn stop_executors(executors: &mut Vec<ExecutorLease>, irq_synchronized: bool) {
for executor in executors.iter().rev() {
executor.stop(irq_synchronized);
}
while let Some(executor) = executors.pop() {
executor.join();
}
}
fn publish_executors(executors: &mut Vec<ExecutorLease>) {
for executor in executors.iter() {
executor
.control
.command
.store(COMMAND_RUN, Ordering::Release);
executor.control.notify.notify();
}
let mut index = 0;
while index < executors.len() {
let status = &executors[index].control.publication_status;
wait_status(status);
if status.load(Ordering::Acquire) == STATUS_EMPTY {
executors.swap_remove(index).join();
} else {
index += 1;
}
}
}