retina_core/runtime/
online.rs

1use crate::config::{ConnTrackConfig, OnlineConfig, RuntimeConfig};
2use crate::dpdk;
3use crate::filter::Filter;
4use crate::lcore::monitor::Monitor;
5use crate::lcore::rx_core::RxCore;
6use crate::lcore::{CoreId, SocketId};
7use crate::memory::mempool::Mempool;
8use crate::port::*;
9use crate::subscription::*;
10
11use std::collections::BTreeMap;
12use std::os::raw::{c_uint, c_void};
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::sync::Arc;
15use std::time::Instant;
16
17pub(crate) struct OnlineRuntime<'a, S>
18where
19    S: Subscribable,
20{
21    ports: BTreeMap<PortId, Port>,
22    rx_cores: BTreeMap<CoreId, RxCore<'a, S>>,
23    monitor: Monitor,
24    filter: Filter,
25    options: OnlineOptions,
26}
27
28impl<'a, S> OnlineRuntime<'a, S>
29where
30    S: Subscribable,
31{
32    pub(crate) fn new(
33        config: &RuntimeConfig,
34        options: OnlineOptions,
35        mempools: &mut BTreeMap<SocketId, Mempool>,
36        filter: Filter,
37        subscription: Arc<Subscription<'a, S>>,
38    ) -> Self {
39        // Set up signal handler
40        let is_running = Arc::new(AtomicBool::new(true));
41        let r = Arc::clone(&is_running);
42        ctrlc::set_handler(move || {
43            r.store(false, Ordering::Relaxed);
44        })
45        .expect("Error setting Ctrl-C handler");
46
47        log::info!("Initializing Ports...");
48        let mut ports: BTreeMap<PortId, Port> = BTreeMap::new();
49        for port_map in options.online.ports.iter() {
50            let port = Port::new(port_map);
51            let socket_id = port.id.socket_id();
52            mempools.entry(socket_id).or_insert_with(|| {
53                // Create a local mempool if user is not polling the port
54                // from the same socket.
55                let mtu = if let Some(online) = &config.online {
56                    online.mtu
57                } else {
58                    Mempool::default_mtu()
59                };
60                Mempool::new(&config.mempool, socket_id, mtu)
61                    .expect("Unable to initialize local mempool")
62            });
63            port.init(
64                mempools,
65                options.online.nb_rxd,
66                options.online.mtu,
67                options.online.promiscuous,
68            )
69            .expect("Failed to initialize port.");
70            ports.insert(port.id, port);
71        }
72
73        log::info!("Initializing RX Cores...");
74        let mut rx_cores: BTreeMap<CoreId, RxCore<S>> = BTreeMap::new();
75        let mut core_map: BTreeMap<CoreId, Vec<RxQueue>> = BTreeMap::new();
76        for (_port_id, port) in ports.iter() {
77            for (rxqueue, core_id) in port.queue_map.iter() {
78                core_map.entry(*core_id).or_default().push(*rxqueue);
79            }
80        }
81        for (core_id, rxqueues) in core_map.into_iter() {
82            let rx_core = RxCore::new(
83                core_id,
84                rxqueues,
85                filter.clone(),
86                options.conntrack.clone(),
87                Arc::clone(&subscription),
88                Arc::clone(&is_running),
89            );
90            rx_cores.insert(core_id, rx_core);
91        }
92
93        let monitor = Monitor::new(config, &ports, Arc::clone(&is_running));
94
95        OnlineRuntime {
96            ports,
97            rx_cores,
98            monitor,
99            filter,
100            options,
101        }
102    }
103
104    pub(crate) fn run(&mut self) {
105        self.start_ports();
106
107        log::info!("Launching RX cores...");
108        for (core_id, _rx_core) in self.rx_cores.iter() {
109            let role = unsafe { dpdk::rte_eal_lcore_role(core_id.raw()) };
110            if role != dpdk::rte_lcore_role_t_ROLE_RTE {
111                log::error!("Attempted to launch non-DPDK core");
112                panic!();
113            }
114
115            let arg = &self.rx_cores as *const _ as *mut c_void;
116            let ret = unsafe {
117                dpdk::rte_eal_remote_launch(Some(launch_rx::<S>), arg, core_id.raw() as c_uint)
118            };
119            if ret != 0 {
120                log::error!("RX Core {} busy, launch failed.", core_id);
121                panic!();
122            }
123        }
124
125        // run main thread
126        self.run_main();
127        unsafe { dpdk::rte_eal_mp_wait_lcore() };
128
129        log::info!("Exiting loop...");
130        self.stop_ports();
131    }
132
133    fn run_main(&mut self) {
134        let id = unsafe { dpdk::rte_lcore_id() };
135        log::info!("Running main on Core {}", id);
136        let start = Instant::now();
137        self.monitor.run();
138        println!("Main done. Ran for {:?}", start.elapsed());
139    }
140
141    fn start_ports(&self) {
142        log::info!("Starting ports...");
143        for port in self.ports.values() {
144            port.start();
145
146            if self.options.online.hardware_assist {
147                log::info!("Applying hardware filters...");
148                let res = self.filter.set_hardware_filter(port);
149                match res {
150                    Ok(_) => (),
151                    Err(error) => {
152                        log::warn!("Failed to apply some patterns, passing all traffic through Port {}. Reason: {}", port.id, error);
153                    }
154                }
155            } else {
156                log::info!("No hardware assist configured for port {}, passing all traffic through device.", port.id);
157            }
158        }
159    }
160
161    fn stop_ports(&self) {
162        log::info!("Stopping ports...");
163        for port in self.ports.values() {
164            port.stop();
165        }
166    }
167}
168
169/// Read-only runtime options for the offline core
170#[derive(Debug)]
171pub(crate) struct OnlineOptions {
172    pub(crate) online: OnlineConfig,
173    pub(crate) conntrack: ConnTrackConfig,
174}
175
176extern "C" fn launch_rx<S>(arg: *mut c_void) -> i32
177where
178    S: Subscribable,
179{
180    // enforce that workers cores cannot mutate runtime
181    // TODO: make this *const and use Mutex for interior mutability
182    let rx_cores = arg as *const BTreeMap<CoreId, RxCore<S>>;
183    let rx_cores = unsafe { &*rx_cores };
184
185    let core_id = CoreId(unsafe { dpdk::rte_lcore_id() } as u32);
186    let rx_core = rx_cores.get(&core_id).expect("Invalid Core");
187
188    // TODO: slight optimization: even if filter is nonempty, if the hardware takes care of the
189    // whole thing we can also run_rx with no filter
190    rx_core.rx_loop();
191    0
192}
retina_core/runtime/online.rs

retina_core/runtime/
online.rs
