ax_net/queue_runtime/
mod.rs

1//! Queue-level, fixed-CPU network poll runtime.
2//!
3//! Every physical IRQ source is assigned to an affinity domain.  A domain's
4//! hard callbacks and queue processing run on one owner CPU; only move-only DMA
5//! tokens cross the SPSC boundary to the single protocol executor.
6
7mod executor;
8mod notify;
9mod spsc;
10mod state;
11#[cfg(test)]
12mod tests;
13
14use alloc::{boxed::Box, collections::VecDeque, format, string::String, sync::Arc, vec, vec::Vec};
15use core::{
16    num::NonZeroUsize,
17    sync::atomic::{AtomicBool, AtomicU8, Ordering},
18};
19
20use ax_sync::SpinLock;
21use ax_task::{sched::CpuSet, sync::WaitQueue};
22use irq_framework::IrqId;
23use rd_net::{
24    NetError, NetHardIrqEndpoint, NetHardIrqResult, NetIrqSourceId, PreparedNetDevice,
25    WifiLinkPolicy, WifiTransaction,
26};
27
28pub use self::state::NetQueueStats;
29use self::{executor::*, notify::QueueNotification, spsc::*, state::PollGroupState};
30use crate::device::{EthernetFramePort, EthernetFramePortList};
31
32const QUEUE_BUDGET: usize = 64;
33const CPU_ROUND_BUDGET: usize = 256;
34const WIFI_CONTROL_QUEUE_CAPACITY: usize = 8;
35
36const STATE_IDLE: u8 = 0;
37const STATE_SCHEDULED: u8 = 1;
38const STATE_POLLING: u8 = 2;
39const STATE_DISABLED: u8 = 3;
40const STATE_MASK: u8 = 0x0f;
41const STATE_MISSED: u8 = 0x80;
42
43const COMMAND_WAIT: u8 = 0;
44const COMMAND_START: u8 = 1;
45const COMMAND_STOP: u8 = 2;
46const COMMAND_QUARANTINE: u8 = 3;
47const COMMAND_RUN: u8 = 4;
48const COMMAND_PRUNE: u8 = 5;
49
50const STATUS_PENDING: u8 = 0;
51const STATUS_READY: u8 = 1;
52const STATUS_FAILED: u8 = 2;
53const STATUS_EMPTY: u8 = 3;
54
55struct WifiCommandCompletion {
56    result: SpinLock<Option<Result<(), NetError>>>,
57    wait: WaitQueue,
58}
59
60impl WifiCommandCompletion {
61    fn new() -> Self {
62        Self {
63            result: SpinLock::new(None),
64            wait: WaitQueue::new(),
65        }
66    }
67
68    fn complete(&self, result: Result<(), NetError>) {
69        *self.result.lock_irqsave() = Some(result);
70        self.wait.notify_all();
71    }
72
73    fn wait(&self) -> Result<(), NetError> {
74        self.wait
75            .wait_until(|| self.result.lock_irqsave().is_some());
76        self.result
77            .lock_irqsave()
78            .take()
79            .expect("Wi-Fi completion was published without a result")
80    }
81}
82
83struct WifiControlRequest {
84    transaction: WifiTransaction,
85    completion: Arc<WifiCommandCompletion>,
86}
87
88struct WifiControlQueue {
89    requests: SpinLock<VecDeque<WifiControlRequest>>,
90    stopped: AtomicBool,
91}
92
93impl WifiControlQueue {
94    fn new() -> Self {
95        Self {
96            requests: SpinLock::new(VecDeque::with_capacity(WIFI_CONTROL_QUEUE_CAPACITY)),
97            stopped: AtomicBool::new(false),
98        }
99    }
100
101    fn submit(
102        &self,
103        transaction: WifiTransaction,
104        notify: &QueueNotification,
105    ) -> Result<(), NetError> {
106        if self.stopped.load(Ordering::Acquire) {
107            return Err(NetError::Stopped);
108        }
109        let completion = Arc::new(WifiCommandCompletion::new());
110        {
111            let mut requests = self.requests.lock_irqsave();
112            if self.stopped.load(Ordering::Acquire) {
113                return Err(NetError::Stopped);
114            }
115            if requests.len() == WIFI_CONTROL_QUEUE_CAPACITY {
116                return Err(NetError::Retry);
117            }
118            requests.push_back(WifiControlRequest {
119                transaction,
120                completion: Arc::clone(&completion),
121            });
122        }
123        notify.notify();
124        completion.wait()
125    }
126
127    fn try_pop(&self) -> Option<WifiControlRequest> {
128        self.requests.lock_irqsave().pop_front()
129    }
130
131    fn has_pending(&self) -> bool {
132        !self.requests.lock_irqsave().is_empty()
133    }
134
135    fn stop(&self) {
136        self.stopped.store(true, Ordering::Release);
137        let pending = core::mem::take(&mut *self.requests.lock_irqsave());
138        for request in pending {
139            request.completion.complete(Err(NetError::Stopped));
140        }
141    }
142}
143
144#[derive(Clone)]
145pub(crate) struct WifiRuntimeHandle {
146    device_index: usize,
147    owner_cpu: usize,
148    queue: Arc<WifiControlQueue>,
149    notify: Arc<QueueNotification>,
150    startup_group: Arc<PollGroupState>,
151}
152
153impl WifiRuntimeHandle {
154    pub(crate) const fn device_index(&self) -> usize {
155        self.device_index
156    }
157
158    pub(crate) const fn owner_cpu(&self) -> usize {
159        self.owner_cpu
160    }
161
162    pub(crate) fn submit(&self, transaction: WifiTransaction) -> Result<(), NetError> {
163        self.queue.submit(transaction, &self.notify)
164    }
165}
166
167/// Runtime initialization or lifecycle error.
168#[derive(Debug, thiserror::Error)]
169pub enum NetworkRuntimeError {
170    #[error("network device parts are inconsistent with their IRQ bindings")]
171    InvalidTopology,
172    #[error("network queue executor could not be pinned to CPU {0}")]
173    WorkerAffinity(usize),
174    #[error("network queue executor for CPU {cpu} could not be spawned: {source}")]
175    WorkerSpawn {
176        cpu: usize,
177        #[source]
178        source: ax_task::thread::TaskError,
179    },
180    #[error("network queue initialization failed: {0}")]
181    QueueInit(NetError),
182    #[error("network IRQ registration failed: {0}")]
183    IrqRegistration(#[from] PinnedNetIrqError),
184    #[error("network DMA setup failed: {0}")]
185    Device(#[from] NetError),
186    #[error("secure Wi-Fi startup entropy failed: {0}")]
187    StartupEntropy(#[from] crate::NetError),
188}
189
190/// Resolved driver source-id to physical IRQ mapping.
191#[derive(Clone, Copy, Debug, Eq, PartialEq)]
192pub struct ResolvedNetIrqSource {
193    pub source_id: NetIrqSourceId,
194    pub irq: IrqId,
195}
196
197/// One prepared device and its complete, resolved IRQ source map.
198pub struct NetworkDeviceInput {
199    pub name: String,
200    pub device: PreparedNetDevice,
201    pub irq_sources: Vec<ResolvedNetIrqSource>,
202    pub tx_queue_discipline: TxQueueDiscipline,
203}
204
205/// Protocol-side transmit queue policy for one network device.
206#[derive(Clone, Copy, Debug, Eq, PartialEq)]
207pub enum TxQueueDiscipline {
208    /// Submit directly to the device and return `Again` when it is busy.
209    NoQueue,
210    /// Retain frames in submission order while the device is busy.
211    Fifo { max_frames: NonZeroUsize },
212}
213
214/// Result of a bounded hard-IRQ callback.
215#[derive(Clone, Copy, Debug, Eq, PartialEq)]
216pub enum PinnedNetIrqOutcome {
217    Unhandled,
218    Handled,
219    Wake,
220}
221
222/// Move-only callback installed by the OS IRQ adapter.
223pub struct PinnedNetIrqAction {
224    handler: Box<dyn FnMut() -> PinnedNetIrqOutcome + Send>,
225}
226
227impl PinnedNetIrqAction {
228    pub fn new(handler: impl FnMut() -> PinnedNetIrqOutcome + Send + 'static) -> Self {
229        Self {
230            handler: Box::new(handler),
231        }
232    }
233
234    pub fn run(&mut self) -> PinnedNetIrqOutcome {
235        (self.handler)()
236    }
237}
238
239/// OS-specific fixed-affinity IRQ registration error.
240#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
241pub enum PinnedNetIrqError {
242    #[error("invalid network IRQ or owner CPU")]
243    Invalid,
244    #[error("network IRQ affinity conflicts with an existing shared action")]
245    AffinityConflict,
246    #[error("fixed network IRQ routing is unsupported")]
247    Unsupported,
248    #[error("network IRQ operation failed")]
249    Other,
250}
251
252/// Move-only registration lease.  It is created disabled.
253pub trait PinnedNetIrqRegistration: Send + Sync + 'static {
254    fn owner_cpu(&self) -> usize;
255    fn enable(&self) -> Result<(), PinnedNetIrqError>;
256    fn disable_and_synchronize(&self) -> Result<(), PinnedNetIrqError>;
257}
258
259/// OS adapter that accepts fixed affinity only.
260pub trait PinnedNetIrqRegistrar: Sync {
261    fn register(
262        &self,
263        name: String,
264        irq: IrqId,
265        owner_cpu: usize,
266        action: PinnedNetIrqAction,
267    ) -> Result<Box<dyn PinnedNetIrqRegistration>, PinnedNetIrqError>;
268}
269
270struct EndpointToRegister {
271    name: String,
272    irq: IrqId,
273    owner_cpu: usize,
274    endpoint: NetHardIrqEndpoint,
275    shared: Arc<PollGroupState>,
276}
277
278struct RegisteredEndpoint {
279    registration: Box<dyn PinnedNetIrqRegistration>,
280    shared: Arc<PollGroupState>,
281}
282
283/// Live queue runtime.  Dropping it masks IRQs before stopping executors.
284pub struct NetworkQueueRuntime {
285    registrations: Vec<Box<dyn PinnedNetIrqRegistration>>,
286    executors: Vec<ExecutorLease>,
287    group_states: Vec<Arc<PollGroupState>>,
288    _controls: Vec<Box<dyn rd_net::NetControlEndpoint>>,
289    wifi_handles: Vec<WifiRuntimeHandle>,
290    initial_wifi_policies: Vec<(usize, WifiLinkPolicy)>,
291    device_index_map: Vec<Option<usize>>,
292    protocol_owner_cpu: usize,
293}
294
295impl NetworkQueueRuntime {
296    pub(crate) fn discovery_order(&self, device_index: usize) -> usize {
297        self.device_index_map
298            .iter()
299            .position(|&index| index == Some(device_index))
300            .expect("published network port must have a discovery order")
301    }
302
303    pub fn protocol_owner_cpu(&self) -> usize {
304        self.protocol_owner_cpu
305    }
306
307    pub fn stats(&self) -> Vec<NetQueueStats> {
308        self.group_states
309            .iter()
310            .map(|state| state.stats.snapshot(state.owner_cpu))
311            .collect()
312    }
313
314    pub(crate) fn wifi_handle(&self, device_index: usize) -> Option<WifiRuntimeHandle> {
315        self.wifi_handles
316            .iter()
317            .find(|handle| handle.device_index() == device_index)
318            .cloned()
319    }
320
321    pub(crate) fn initial_wifi_policy(&self, device_index: usize) -> Option<WifiLinkPolicy> {
322        self.initial_wifi_policies
323            .iter()
324            .find_map(|(index, policy)| (*index == device_index).then_some(*policy))
325    }
326}
327
328impl Drop for NetworkQueueRuntime {
329    fn drop(&mut self) {
330        for handle in self.wifi_handles.iter().rev() {
331            handle.queue.stop();
332            handle.notify.notify();
333        }
334        let registrations = core::mem::take(&mut self.registrations);
335        let irq_synchronized = release_registrations(registrations);
336        let runtime_side_resources = (
337            core::mem::take(&mut self.group_states),
338            core::mem::take(&mut self._controls),
339            core::mem::take(&mut self.wifi_handles),
340        );
341        stop_executors(&mut self.executors, irq_synchronized);
342        release_runtime_side_resources(runtime_side_resources, irq_synchronized);
343    }
344}
345
346/// Builder for an all-at-once network queue runtime.
347pub struct NetworkRuntimeBuilder<'a> {
348    devices: Vec<NetworkDeviceInput>,
349    registrar: &'a dyn PinnedNetIrqRegistrar,
350    active_cpus: CpuSet,
351}
352
353impl<'a> NetworkRuntimeBuilder<'a> {
354    pub fn new(
355        devices: Vec<NetworkDeviceInput>,
356        registrar: &'a dyn PinnedNetIrqRegistrar,
357        active_cpus: CpuSet,
358    ) -> Self {
359        Self {
360            devices,
361            registrar,
362            active_cpus,
363        }
364    }
365
366    pub fn build(
367        self,
368    ) -> Result<(NetworkQueueRuntime, EthernetFramePortList), NetworkRuntimeError> {
369        let topology_len = self.active_cpus.topology_len();
370        let active_cpus = self
371            .active_cpus
372            .iter()
373            .map(ax_task::sched::CpuId::as_usize)
374            .collect::<Vec<_>>();
375        if topology_len == 0 || active_cpus.is_empty() {
376            // No owner context exists in which a driver can prove DMA has
377            // stopped, so prepared device backing must not be dropped.
378            release_or_quarantine(self.devices, false);
379            return Err(NetworkRuntimeError::InvalidTopology);
380        }
381
382        let group_irq_sets = match validate_and_collect_irq_sets(&self.devices) {
383            Ok(sets) => sets,
384            Err(error) => {
385                // The portable parts may already contain hardware-visible
386                // descriptor rings, but no owner CPU exists for lifecycle
387                // control yet. Preserve them as an early-init quarantine.
388                core::mem::forget(self.devices);
389                return Err(error);
390            }
391        };
392        let group_owners = assign_affinity_domains(&group_irq_sets, &active_cpus);
393        let mut groups_by_cpu = (0..topology_len)
394            .map(|_| Vec::new())
395            .collect::<Vec<Vec<QueueGroupExecutor>>>();
396        let mut wifi_by_cpu = (0..topology_len)
397            .map(|_| Vec::new())
398            .collect::<Vec<Vec<WifiExecutorSlot>>>();
399        let cpu_notifies = (0..topology_len)
400            .map(|_| Arc::new(QueueNotification::new()))
401            .collect::<Vec<_>>();
402        let mut endpoints = Vec::new();
403        let mut ports = Vec::with_capacity(self.devices.len());
404        let mut controls = Vec::new();
405        let mut port_macs = Vec::new();
406        let mut wifi_handles = Vec::new();
407        let mut startup_transactions = Vec::new();
408        let mut group_states = Vec::new();
409        let mut flat_group = 0;
410
411        for (device_index, input) in self.devices.into_iter().enumerate() {
412            let port_name = input.name.clone();
413            let PreparedNetDevice {
414                info,
415                control,
416                wifi_control,
417                poll_groups,
418            } = input.device;
419            let mut protocol_groups = Vec::with_capacity(poll_groups.len());
420            let mut checksum_capabilities = None;
421            let mut wifi_target = None;
422            let mut device_group_locations = Vec::with_capacity(poll_groups.len());
423            for mut group in poll_groups {
424                checksum_capabilities = Some(checksum_capabilities.map_or(
425                    group.tx.checksum_capabilities(),
426                    |current: rd_net::TxChecksumCapabilities| {
427                        current.intersection(group.tx.checksum_capabilities())
428                    },
429                ));
430                let owner_cpu = group_owners[flat_group];
431                let owner_group_index = groups_by_cpu[owner_cpu].len();
432                let shared = Arc::new(PollGroupState::new(
433                    owner_cpu,
434                    Arc::clone(&cpu_notifies[owner_cpu]),
435                ));
436                let rx_capacity = group.rx.capacity();
437                let (rx_ready_tx, rx_ready_rx) = spsc_ring(rx_capacity);
438                let (rx_recycle_tx, rx_recycle_rx) = spsc_ring(rx_capacity);
439                let (tx_ready_tx, tx_ready_rx) = spsc_ring(group.tx.capacity());
440                let (tx_free_tx, tx_free_rx) = spsc_ring(group.tx.capacity());
441                let rx_recycler = Arc::new(RxRecycler::new(
442                    rx_recycle_tx,
443                    Arc::clone(&shared),
444                    rx_capacity,
445                ));
446
447                for endpoint in group.irq_endpoints.drain(..) {
448                    let irq = resolve_endpoint_irq(&input.irq_sources, endpoint.source_id())
449                        .expect("network IRQ topology was validated before ownership transfer");
450                    endpoints.push(EndpointToRegister {
451                        name: format!(
452                            "{}-g{}-s{}",
453                            input.name,
454                            group.id.get(),
455                            endpoint.source_id().get()
456                        ),
457                        irq,
458                        owner_cpu,
459                        endpoint,
460                        shared: Arc::clone(&shared),
461                    });
462                }
463
464                protocol_groups.push(ProtocolGroupPort {
465                    rx_ready: rx_ready_rx,
466                    rx_recycler: Arc::clone(&rx_recycler),
467                    tx_ready: tx_ready_tx,
468                    tx_free: tx_free_rx,
469                    tx_spares: Vec::with_capacity(group.tx.capacity()),
470                    shared: Arc::clone(&shared),
471                });
472                groups_by_cpu[owner_cpu].push(QueueGroupExecutor {
473                    wifi_startup_group: None,
474                    group,
475                    rx_ready: rx_ready_tx,
476                    rx_recycle: rx_recycle_rx,
477                    rx_recycler,
478                    rx_spares: Vec::with_capacity(rx_capacity.max(QUEUE_BUDGET)),
479                    rx_extra_buffers: 0,
480                    tx_ready: tx_ready_rx,
481                    tx_free: tx_free_tx,
482                    pending_rx: None,
483                    pending_rx_refill: VecDeque::with_capacity(rx_capacity),
484                    pending_tx: None,
485                    pending_tx_free: None,
486                    retry_at: None,
487                    shared: Arc::clone(&shared),
488                });
489                wifi_target.get_or_insert((owner_cpu, owner_group_index, Arc::clone(&shared)));
490                device_group_locations.push((owner_cpu, owner_group_index));
491                group_states.push(shared);
492                flat_group += 1;
493            }
494            if let Some(wifi_control) = wifi_control {
495                let (owner_cpu, group_index, startup_group) =
496                    wifi_target.ok_or(NetworkRuntimeError::InvalidTopology)?;
497                for (group_cpu, local_index) in device_group_locations {
498                    groups_by_cpu[group_cpu][local_index].wifi_startup_group =
499                        Some(Arc::clone(&startup_group));
500                }
501                let queue = Arc::new(WifiControlQueue::new());
502                let handle = WifiRuntimeHandle {
503                    device_index,
504                    owner_cpu,
505                    queue: Arc::clone(&queue),
506                    notify: Arc::clone(&cpu_notifies[owner_cpu]),
507                    startup_group,
508                };
509                if let Some(transaction) = wifi_control.startup_transaction() {
510                    startup_transactions.push((handle.clone(), transaction));
511                }
512                wifi_by_cpu[owner_cpu].push(WifiExecutorSlot {
513                    group_index,
514                    control: wifi_control,
515                    queue,
516                    active: None,
517                });
518                wifi_handles.push(handle);
519            }
520            controls.push(control);
521            let port_mac = Arc::new(SpinLock::new(info.mac_address));
522            port_macs.push(Arc::clone(&port_mac));
523            ports.push(QueueFramePort {
524                name: port_name,
525                mac: port_mac,
526                groups: protocol_groups,
527                tx_queue_discipline: input.tx_queue_discipline,
528                pending_tx: VecDeque::new(),
529                next_rx: 0,
530                next_tx: 0,
531                checksum_capabilities: checksum_capabilities
532                    .unwrap_or(rd_net::TxChecksumCapabilities::NONE),
533            });
534        }
535
536        let mut executors = Vec::new();
537        for (owner_cpu, (groups, wifi)) in groups_by_cpu.into_iter().zip(wifi_by_cpu).enumerate() {
538            if groups.is_empty() {
539                continue;
540            }
541            let control = Arc::new(ExecutorControl {
542                owner_cpu,
543                command: AtomicU8::new(COMMAND_WAIT),
544                affinity_status: AtomicU8::new(STATUS_PENDING),
545                startup_status: AtomicU8::new(STATUS_PENDING),
546                prune_status: AtomicU8::new(STATUS_PENDING),
547                publication_status: AtomicU8::new(STATUS_PENDING),
548                startup_error: SpinLock::new(None),
549                notify: Arc::clone(&cpu_notifies[owner_cpu]),
550            });
551            let mut affinity = CpuSet::empty(topology_len);
552            if !affinity.insert(ax_task::sched::CpuId::new(owner_cpu as u32)) {
553                stop_executors(&mut executors, true);
554                return Err(NetworkRuntimeError::InvalidTopology);
555            }
556            let task_control = Arc::clone(&control);
557            let task =
558                match ax_task::thread::ThreadBuilder::new(format!("net-queue-cpu{owner_cpu}"))
559                    .affinity(affinity)
560                    .spawn(move || queue_executor_main(groups, wifi, task_control))
561                {
562                    Ok(task) => task,
563                    Err(source) => {
564                        stop_executors(&mut executors, true);
565                        return Err(NetworkRuntimeError::WorkerSpawn {
566                            cpu: owner_cpu,
567                            source,
568                        });
569                    }
570                };
571            executors.push(ExecutorLease { control, task });
572        }
573        let failed_owner = executors.iter().find_map(|executor| {
574            wait_status(&executor.control.affinity_status);
575            (executor.control.affinity_status.load(Ordering::Acquire) != STATUS_READY)
576                .then_some(executor.control.owner_cpu)
577        });
578        if let Some(owner_cpu) = failed_owner {
579            stop_executors(&mut executors, true);
580            return Err(NetworkRuntimeError::WorkerAffinity(owner_cpu));
581        }
582
583        let mut registrations = Vec::new();
584        let mut endpoint_iter = endpoints.into_iter();
585        while let Some(mut endpoint) = endpoint_iter.next() {
586            let shared = Arc::clone(&endpoint.shared);
587            let registration_state = Arc::clone(&endpoint.shared);
588            let owner_cpu = endpoint.owner_cpu;
589            let action = PinnedNetIrqAction::new(move || match endpoint.endpoint.handle_irq() {
590                NetHardIrqResult::Spurious => {
591                    shared.stats.spurious.fetch_add(1, Ordering::Relaxed);
592                    PinnedNetIrqOutcome::Unhandled
593                }
594                NetHardIrqResult::Schedule(_snapshot) => {
595                    shared.schedule_irq();
596                    PinnedNetIrqOutcome::Wake
597                }
598                NetHardIrqResult::ProbeDeferred => {
599                    shared.stats.probe_deferred.fetch_add(1, Ordering::Relaxed);
600                    shared.schedule_irq();
601                    PinnedNetIrqOutcome::Wake
602                }
603            });
604            let registration =
605                match self
606                    .registrar
607                    .register(endpoint.name, endpoint.irq, owner_cpu, action)
608                {
609                    Ok(registration) if registration.owner_cpu() == owner_cpu => registration,
610                    Ok(registration) => {
611                        registrations.push(RegisteredEndpoint {
612                            registration,
613                            shared: registration_state,
614                        });
615                        let irq_synchronized = release_registered_endpoints(registrations);
616                        stop_executors(&mut executors, irq_synchronized);
617                        release_runtime_side_resources(
618                            (
619                                controls,
620                                ports,
621                                port_macs,
622                                wifi_handles,
623                                startup_transactions,
624                                group_states,
625                                cpu_notifies,
626                                endpoint_iter,
627                            ),
628                            irq_synchronized,
629                        );
630                        return Err(NetworkRuntimeError::InvalidTopology);
631                    }
632                    Err(error) => {
633                        let irq_synchronized = release_registered_endpoints(registrations);
634                        stop_executors(&mut executors, irq_synchronized);
635                        release_runtime_side_resources(
636                            (
637                                controls,
638                                ports,
639                                port_macs,
640                                wifi_handles,
641                                startup_transactions,
642                                group_states,
643                                cpu_notifies,
644                                endpoint_iter,
645                            ),
646                            irq_synchronized,
647                        );
648                        return Err(error.into());
649                    }
650                };
651            registrations.push(RegisteredEndpoint {
652                registration,
653                shared: registration_state,
654            });
655        }
656        drop(endpoint_iter);
657
658        for registration in &registrations {
659            if let Err(error) = registration.registration.enable() {
660                let irq_synchronized = release_registered_endpoints(registrations);
661                stop_executors(&mut executors, irq_synchronized);
662                release_runtime_side_resources(
663                    (
664                        controls,
665                        ports,
666                        port_macs,
667                        wifi_handles,
668                        startup_transactions,
669                        group_states,
670                        cpu_notifies,
671                    ),
672                    irq_synchronized,
673                );
674                return Err(error.into());
675            }
676        }
677
678        for command in [COMMAND_START, COMMAND_PRUNE] {
679            for executor in &executors {
680                executor.control.command.store(command, Ordering::Release);
681                executor.control.notify.notify();
682            }
683            for executor in &executors {
684                let status = if command == COMMAND_START {
685                    &executor.control.startup_status
686                } else {
687                    &executor.control.prune_status
688                };
689                wait_status(status);
690                if status.load(Ordering::Acquire) != STATUS_READY {
691                    let error = executor
692                        .control
693                        .startup_error
694                        .lock_irqsave()
695                        .take()
696                        .unwrap_or(NetError::InvalidParts);
697                    let irq_synchronized = release_registered_endpoints(registrations);
698                    stop_executors(&mut executors, irq_synchronized);
699                    release_runtime_side_resources(
700                        (
701                            controls,
702                            ports,
703                            port_macs,
704                            wifi_handles,
705                            startup_transactions,
706                            group_states,
707                            cpu_notifies,
708                        ),
709                        irq_synchronized,
710                    );
711                    return Err(NetworkRuntimeError::QueueInit(error));
712                }
713            }
714        }
715
716        let registrations = match prune_absent_irq_registrations(registrations) {
717            Ok(registrations) => registrations,
718            Err(error) => {
719                stop_executors(&mut executors, false);
720                release_runtime_side_resources(
721                    (
722                        controls,
723                        ports,
724                        port_macs,
725                        wifi_handles,
726                        startup_transactions,
727                        group_states,
728                        cpu_notifies,
729                    ),
730                    false,
731                );
732                return Err(error.into());
733            }
734        };
735
736        publish_executors(&mut executors);
737
738        let (started_ports, device_index_map) = retain_started_ports(ports);
739        group_states.retain(|state| !state.startup_absent());
740        let controls = controls
741            .into_iter()
742            .enumerate()
743            .filter_map(|(index, control)| device_index_map[index].map(|_| control))
744            .collect::<Vec<_>>();
745        let port_macs = port_macs
746            .into_iter()
747            .enumerate()
748            .filter_map(|(index, mac)| device_index_map[index].map(|_| mac))
749            .collect::<Vec<_>>();
750        let wifi_handles = wifi_handles
751            .into_iter()
752            .filter_map(|mut handle| {
753                if handle.startup_group.startup_absent() {
754                    return None;
755                }
756                handle.device_index = device_index_map[handle.device_index]?;
757                Some(handle)
758            })
759            .collect::<Vec<_>>();
760        let startup_transactions = startup_transactions
761            .into_iter()
762            .filter_map(|(mut handle, transaction)| {
763                if handle.startup_group.startup_absent() {
764                    return None;
765                }
766                handle.device_index = device_index_map[handle.device_index]?;
767                Some((handle, transaction))
768            })
769            .collect::<Vec<_>>();
770        let active_group_owners = group_states
771            .iter()
772            .map(|state| state.owner_cpu)
773            .collect::<Vec<_>>();
774        let protocol_owner_cpu = select_protocol_owner(&active_group_owners, &active_cpus);
775        let mut runtime = NetworkQueueRuntime {
776            registrations,
777            executors,
778            group_states,
779            _controls: controls,
780            wifi_handles,
781            initial_wifi_policies: Vec::new(),
782            device_index_map,
783            protocol_owner_cpu,
784        };
785        for (handle, transaction) in startup_transactions {
786            let transaction =
787                prepare_startup_transaction(transaction, super::next_wifi_connection_entropy)?;
788            let policy = transaction.link_policy();
789            handle.submit(transaction)?;
790            if let Some(policy) = policy {
791                runtime
792                    .initial_wifi_policies
793                    .push((handle.device_index(), policy));
794            }
795        }
796        for (control, mac) in runtime._controls.iter_mut().zip(port_macs) {
797            let address = control.mac_address()?;
798            *mac.lock_irqsave() = address;
799        }
800        Ok((runtime, started_ports))
801    }
802}
803
804fn retain_started_ports(ports: Vec<QueueFramePort>) -> (EthernetFramePortList, Vec<Option<usize>>) {
805    let mut device_index_map = vec![None; ports.len()];
806    let mut started_ports = Vec::with_capacity(ports.len());
807    for (device_index, mut port) in ports.into_iter().enumerate() {
808        if port.retain_started_groups() {
809            device_index_map[device_index] = Some(started_ports.len());
810            started_ports.push(Box::new(port) as Box<dyn EthernetFramePort>);
811        } else {
812            log::warn!(
813                "network device {} is not present after owner startup; skipping it",
814                port.name
815            );
816        }
817    }
818    (started_ports, device_index_map)
819}
820
821fn prepare_startup_transaction(
822    mut transaction: WifiTransaction,
823    next_entropy: impl FnOnce() -> Result<[u8; 32], crate::NetError>,
824) -> Result<WifiTransaction, crate::NetError> {
825    if transaction.needs_connect_entropy() {
826        transaction.provide_connect_entropy(next_entropy()?);
827        log::info!("[wifi] secure startup connection entropy prepared");
828    }
829    Ok(transaction)
830}
831
832fn validate_and_collect_irq_sets(
833    devices: &[NetworkDeviceInput],
834) -> Result<Vec<Vec<IrqId>>, NetworkRuntimeError> {
835    let mut sets = Vec::new();
836    for input in devices {
837        if input.device.poll_groups.is_empty() || input.irq_sources.is_empty() {
838            return Err(NetworkRuntimeError::InvalidTopology);
839        }
840        for group in &input.device.poll_groups {
841            if group.irq_endpoints.is_empty() {
842                return Err(NetworkRuntimeError::InvalidTopology);
843            }
844            let mut irqs = Vec::new();
845            for endpoint in &group.irq_endpoints {
846                let irq = resolve_endpoint_irq(&input.irq_sources, endpoint.source_id())?;
847                if !irqs.contains(&irq) {
848                    irqs.push(irq);
849                }
850            }
851            sets.push(irqs);
852        }
853        for source in &input.irq_sources {
854            let used = input.device.poll_groups.iter().any(|group| {
855                group
856                    .irq_endpoints
857                    .iter()
858                    .any(|endpoint| endpoint.source_id() == source.source_id)
859            });
860            if !used {
861                return Err(NetworkRuntimeError::InvalidTopology);
862            }
863        }
864    }
865    Ok(sets)
866}
867
868fn resolve_endpoint_irq(
869    sources: &[ResolvedNetIrqSource],
870    source_id: NetIrqSourceId,
871) -> Result<IrqId, NetworkRuntimeError> {
872    let mut matches = sources
873        .iter()
874        .filter(|source| source.source_id == source_id)
875        .map(|source| source.irq);
876    let irq = matches.next().ok_or(NetworkRuntimeError::InvalidTopology)?;
877    if matches.next().is_some() {
878        return Err(NetworkRuntimeError::InvalidTopology);
879    }
880    Ok(irq)
881}
882
883fn assign_affinity_domains(irq_sets: &[Vec<IrqId>], active_cpus: &[usize]) -> Vec<usize> {
884    let mut parents = (0..irq_sets.len()).collect::<Vec<_>>();
885    for left in 0..irq_sets.len() {
886        for right in (left + 1)..irq_sets.len() {
887            if irq_sets[left]
888                .iter()
889                .any(|irq| irq_sets[right].contains(irq))
890            {
891                union(&mut parents, left, right);
892            }
893        }
894    }
895    let mut roots = Vec::new();
896    let mut owners = Vec::with_capacity(irq_sets.len());
897    for index in 0..irq_sets.len() {
898        let root = find(&mut parents, index);
899        let domain_index = match roots.iter().position(|candidate| *candidate == root) {
900            Some(index) => index,
901            None => {
902                roots.push(root);
903                roots.len() - 1
904            }
905        };
906        owners.push(active_cpus[domain_index % active_cpus.len()]);
907    }
908    owners
909}
910
911fn find(parents: &mut [usize], mut index: usize) -> usize {
912    while parents[index] != index {
913        let grandparent = parents[parents[index]];
914        parents[index] = grandparent;
915        index = grandparent;
916    }
917    index
918}
919
920fn union(parents: &mut [usize], left: usize, right: usize) {
921    let left_root = find(parents, left);
922    let right_root = find(parents, right);
923    if left_root != right_root {
924        let (first, second) = if left_root < right_root {
925            (left_root, right_root)
926        } else {
927            (right_root, left_root)
928        };
929        parents[second] = first;
930    }
931}
932
933fn select_protocol_owner(group_owners: &[usize], active_cpus: &[usize]) -> usize {
934    active_cpus
935        .iter()
936        .copied()
937        .min_by_key(|cpu| {
938            (
939                group_owners.iter().filter(|owner| **owner == *cpu).count(),
940                *cpu,
941            )
942        })
943        .unwrap_or(0)
944}
945
946fn wait_status(status: &AtomicU8) {
947    while status.load(Ordering::Acquire) == STATUS_PENDING {
948        crate::yield_network_thread();
949    }
950}
951
952fn release_registered_endpoints(registrations: Vec<RegisteredEndpoint>) -> bool {
953    release_registrations(
954        registrations
955            .into_iter()
956            .map(|registered| registered.registration)
957            .collect(),
958    )
959}
960
961fn prune_absent_irq_registrations(
962    registrations: Vec<RegisteredEndpoint>,
963) -> Result<Vec<Box<dyn PinnedNetIrqRegistration>>, PinnedNetIrqError> {
964    let mut started = Vec::new();
965    let mut absent = Vec::new();
966    for registered in registrations {
967        if registered.shared.startup_absent() {
968            absent.push(registered.registration);
969        } else {
970            started.push(registered.registration);
971        }
972    }
973    if release_registrations(absent) {
974        Ok(started)
975    } else {
976        let _ = release_registrations(started);
977        Err(PinnedNetIrqError::Other)
978    }
979}
980
981fn disable_registrations(registrations: &[Box<dyn PinnedNetIrqRegistration>]) -> bool {
982    let mut synchronized = true;
983    for registration in registrations.iter().rev() {
984        if registration.disable_and_synchronize().is_err() {
985            synchronized = false;
986        }
987    }
988    synchronized
989}
990
991fn release_registrations(registrations: Vec<Box<dyn PinnedNetIrqRegistration>>) -> bool {
992    let synchronized = disable_registrations(&registrations);
993    if synchronized {
994        drop(registrations);
995    } else {
996        log::warn!(
997            "quarantining {} network IRQ registrations because callback synchronization failed",
998            registrations.len()
999        );
1000        core::mem::forget(registrations);
1001    }
1002    synchronized
1003}
1004
1005fn release_runtime_side_resources<T>(resource: T, irq_synchronized: bool) {
1006    release_or_quarantine(resource, irq_synchronized);
1007}
1008
1009fn stop_executors(executors: &mut Vec<ExecutorLease>, irq_synchronized: bool) {
1010    for executor in executors.iter().rev() {
1011        executor.stop(irq_synchronized);
1012    }
1013    while let Some(executor) = executors.pop() {
1014        executor.join();
1015    }
1016}
1017
1018fn publish_executors(executors: &mut Vec<ExecutorLease>) {
1019    for executor in executors.iter() {
1020        executor
1021            .control
1022            .command
1023            .store(COMMAND_RUN, Ordering::Release);
1024        executor.control.notify.notify();
1025    }
1026    let mut index = 0;
1027    while index < executors.len() {
1028        let status = &executors[index].control.publication_status;
1029        wait_status(status);
1030        if status.load(Ordering::Acquire) == STATUS_EMPTY {
1031            executors.swap_remove(index).join();
1032        } else {
1033            index += 1;
1034        }
1035    }
1036}
ax_net/queue_runtime/mod.rs

ax_net/queue_runtime/
mod.rs
