Split/6 rvnic

Cargo.toml

@@ -15,6 +15,7 @@ default-run = "rattan"
 [workspace]
 resolver = "2"
 members = [".", "rattan-core", "rattan-env", "rattan-log"]
+exclude = ["rvnic"]
 
 [workspace.package]
 version = "0.1.1"
@@ -67,6 +68,7 @@ parking_lot = "0.12.5"
 pcap-file = { version = "2.0.0" }
 rand = "0.10.0"
 rtnetlink = "0.20.0"
+rvnic.path = "./rvnic/librvnic/"
 serde = { version = "1.0.160", features = ["derive"] }
 serde_json = "1.0.149"
 tempfile = "3"
@@ -88,6 +90,11 @@ tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 name = "rattan"
 path = "src/main.rs"
 
+[[bin]]
+name = "rattan-rv"
+path = "src/main-rv.rs"
+required-features = ["rvnic"]
+
 [[bin]]
 name = "rattan-xdp"
 path = "src/main-xdp.rs"
@@ -131,12 +138,14 @@ shadow-rs = "1.7.0"
 default = ["first-packet"]
 drift-stat = ["rattan-core/drift-stat"]
 drift-log = ["rattan-core/drift-log"]
+drop-stat = ["rattan-env/drop-stat"]
 first-packet = ["rattan-core/first-packet"]
 first-payload = ["rattan-core/first-payload"]
 http = ["rattan-core/http"]
 nat = ["dep:mnl", "dep:nftnl"]
 packet-dump = ["rattan-core/packet-dump"]
 xdp = ["rattan-env/xdp"]
+rvnic = ["rattan-env/rvnic"]
 
 # required for cargo flamegraph
 [profile.dev]

rattan-core/src/cells/external.rs

@@ -30,7 +30,8 @@ use crate::{
     radix::{PacketLogMode, BASE_TS, PKT_LOG_MODE},
 };
 
-const SEND_CHANNEL_PACKETS: usize = 16384;
+const SEND_CHANNEL_PACKETS: usize = 32768;
+const RECV_CHANNEL_PACKETS: usize = 32768;
 
 #[derive(Debug, Clone, Copy)]
 #[repr(transparent)]
@@ -119,7 +120,7 @@ where
                 tokio::spawn(Self::send(rx, s.clone()));
                 senders.push(tx);
             }
-            tokio::spawn(Self::demux(dev_rx, senders.clone()));
+            tokio::spawn(Self::demux(dev_rx, senders));
         }
 
         Self {
@@ -244,11 +245,18 @@ where
         log_tx: Option<UnboundedSender<RattanLogOp>>,
         base_ts: i64,
     ) -> Result<Self, Error> {
-        let (tx, rx) = tokio::sync::mpsc::channel(1024);
+        let (tx, rx) = tokio::sync::mpsc::channel(RECV_CHANNEL_PACKETS);
         let flow_map = Arc::new(FlowMap::new(id));
-
         for d in driver.into_iter() {
-            let notify = AsyncFd::new(d.raw_fd())?;
+            if D::use_blocking_recv() {
+                tracing::warn!("Recv in thread on Driver's request");
+                let tx_clone = tx.clone();
+                std::thread::spawn(|| D::recv_thread(d.into_receiver(), tx_clone));
+                continue;
+            }
+
+            let notify = AsyncFd::new(d.raw_fd())
+                .inspect_err(|e| tracing::error!(?e, "Failed to epoll on raw fd {}", d.raw_fd()))?;
             tokio::spawn(Self::recv(
                 flow_map.clone(),
                 notify,
@@ -278,6 +286,7 @@ where
             let mut _guard = notify.readable().await.unwrap();
 
             let Ok(packet) = _guard.try_io(|_fd| receiver.receive()) else {
+                tokio::task::yield_now().await;
                 continue; // would block
             };
 

rattan-core/src/config/mod.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, path::PathBuf};
+use std::{collections::HashMap, env, path::PathBuf};
 
 use rattan_env::env::RattanEnvConfig;
 #[cfg(feature = "serde")]
@@ -78,6 +78,31 @@ impl RattanResourceConfig {
     pub fn new() -> Self {
         Default::default()
     }
+
+    pub fn working_threads(&self) -> usize {
+        // If env `RATTAN_WORKING_THREADS` is set, use it; otherwise use cpu cores
+        if let Some(threads) = env::var("RATTAN_WORKING_THREADS")
+            .ok()
+            .and_then(|v| v.parse().ok())
+        {
+            return threads;
+        }
+        if let Some(cpu) = &self.cpu {
+            cpu.len()
+        } else {
+            4
+        }
+    }
+
+    pub fn get_cpu(&self) -> Option<Vec<u32>> {
+        // If env `RATTAN_CPU` is set, use it; otherwise use the cpu cores from the config
+        if let Ok(cpu_list) = env::var("RATTAN_CPU") {
+            // split by comma and parse each core
+            let cpu: Vec<u32> = cpu_list.split(',').filter_map(|s| s.parse().ok()).collect();
+            return Some(cpu);
+        }
+        self.cpu.clone()
+    }
 }
 
 #[cfg_attr(

rattan-core/src/radix/mod.rs

@@ -1,12 +1,16 @@
 #[cfg(feature = "http")]
 use std::net::IpAddr;
 use std::{
-    sync::{mpsc, Arc},
+    sync::{atomic::AtomicUsize, mpsc, Arc},
     thread,
     time::{SystemTime, UNIX_EPOCH},
 };
 
 use backon::{BlockingRetryable, ExponentialBuilder};
+use nix::{
+    sched::{sched_setaffinity, CpuSet},
+    unistd::Pid,
+};
 use once_cell::sync::{Lazy, OnceCell};
 use rattan_env::{env::standard::AfPacketDriver, InterfaceDriver, StdNetEnv};
 use rattan_env::{
@@ -18,10 +22,6 @@ use rattan_log::{file_logging_thread, RattanLogOp, LOGGING_TX};
 use tokio::{runtime::Runtime, time::Instant};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, span, warn, Level};
-// use nix::{
-//     sched::{sched_setaffinity, CpuSet},
-//     unistd::Pid,
-// };
 
 use crate::{
     cells::{
@@ -52,6 +52,44 @@ pub static BASE_TS: Lazy<(i64, u64, tokio::time::Instant)> = Lazy::new(|| {
     (machine_time, unix_time as u64, Instant::now())
 });
 
+// Upon worker thread startup, this counter is incremented. So that the working threads
+// are assigned unique CPU affinity from the given user-specified CPU set.
+static AFFINITY_COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+// Utility function to set CPU affinity for the current working thread.
+// If anything failes, an error is printed, and the worker thread will continue.
+fn set_cpu_affinity(available_cpus: &[u32]) {
+    let index = AFFINITY_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+    if available_cpus.is_empty() {
+        return;
+    }
+    let cpu = available_cpus[index % available_cpus.len()] as usize;
+
+    let mut cpu_set = CpuSet::new();
+    if cpu_set.set(cpu).is_err() {
+        error!(
+            target : "working_thread",
+            "Failed to set CPU affinity: CPU {} is not available", cpu
+        );
+        return;
+    }
+
+    info!(
+        target : "working_thread",
+        "Trying to set CPU affinity for working thread to CPU {}", cpu
+    );
+
+    if let Err(e) = sched_setaffinity(
+        Pid::from_raw(0), // current thread
+        &cpu_set,
+    ) {
+        error!(
+            target : "working_thread",
+            "Failed to set CPU affinity to CPU {}: {}", cpu, e
+        );
+    }
+}
+
 #[derive(Clone, Copy, Debug, clap::ValueEnum, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub enum PacketLogMode {
@@ -128,7 +166,14 @@ where
             )
             .call()?;
 
-        let running_core = config.env.get_running_cpu();
+        let running_core = config.resource.get_cpu().unwrap_or_default();
+        let worker_thread_cnt = config.resource.working_threads().max(1);
+
+        if running_core.is_empty() {
+            info!(target : "working_thread", "Try to start {} working threads without setting CPU affinity", worker_thread_cnt);
+        } else {
+            info!(target : "working_thread", "Try to start {} working threads with CPU affinity on CPUs: {:?}",  worker_thread_cnt, running_core);
+        };
 
         // Log env creation to kmsg for system-level observability
         if let Ok(mut kmesg_logger) = std::fs::OpenOptions::new().write(true).open("/dev/kmsg") {
@@ -165,15 +210,11 @@ where
 
             // TODO(enhancement): need to handle panic due to affinity setting
             let runtime = tokio::runtime::Builder::new_multi_thread()
-                .worker_threads(running_core.len())
+                .worker_threads(worker_thread_cnt)
                 .enable_all()
-                // .on_thread_start(move || {
-                //     let mut cpuset = CpuSet::new();
-                //     for core in running_core.iter() {
-                //         cpuset.set(*core as usize).unwrap();
-                //     }
-                //     sched_setaffinity(Pid::from_raw(0), &cpuset).unwrap();
-                // })
+                .on_thread_start(move || {
+                    set_cpu_affinity(running_core.as_slice());
+                })
                 .build()
                 .map(Arc::new);
 
@@ -269,9 +310,15 @@ where
             http_thread_handle,
         };
         // This could be veth pairs, or rvnic
-        radix.init_iterfaces()?;
-        radix.load_cells_config(config.cells)?;
-        radix.link_cells(config.links)?;
+        radix
+            .init_iterfaces()
+            .inspect_err(|e| tracing::error!(?e, "Failed to init interfaces"))?;
+        radix
+            .load_cells_config(config.cells)
+            .inspect_err(|e| tracing::error!(?e, "Failed to load cell config"))?;
+        radix
+            .link_cells(config.links)
+            .inspect_err(|e| tracing::error!(?e, "Failed to link cells"))?;
         Ok(radix)
     }
 
@@ -299,13 +346,16 @@ where
             drivers,
         } = interface;
 
+        let name_clone = name.clone();
+
         self.build_cell(name.clone(), move |rt| {
             let _guard = rt.enter();
             let mut id = VirtualEthernetId::new();
             id.set_ns_id(ns_id);
             id.set_veth_id_copied(veth_id);
             VirtualEthernet::<D>::new(drivers, id)
-        })?;
+        })
+        .inspect_err(|e| tracing::error!(?e, "Failed to init interface {}", name_clone))?;
 
         Ok(())
     }
@@ -316,9 +366,11 @@ where
         let interfaces = self
             .env
             .build_interfaces(self.rattan.get_runtime_handle())?;
+        info!("{} interfaces to be registered as cells.", interfaces.len());
         for interface in interfaces.into_iter() {
             self.init_interface(interface)?;
         }
+        info!("Interfaces registered as cells");
         Ok(())
     }
 

rattan-env/Cargo.toml

@@ -16,11 +16,12 @@ futures = { workspace = true }
 ipnet = { workspace = true }
 lazy_static = "1.5.0"
 libc = { workspace = true }
-nix = { workspace = true }
+nix.workspace = true
 once_cell = { workspace = true }
 parking_lot = { workspace = true }
 rand = { workspace = true }
 rtnetlink = { workspace = true }
+rvnic = { workspace = true , optional = true}
 serde = { workspace = true, optional = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
@@ -33,6 +34,8 @@ rattan-log = { path = "../rattan-log" }
 test-log = { workspace = true }
 
 [features]
-default = ["xdp"]
+default = []
+drop-stat = []
 serde = ["dep:serde"]
 xdp = ["dep:camellia-net"]
+rvnic = ["dep:rvnic"]