determine NetworkDevice to use for xdp early

Author: alexpyattaev

Cargo.lock

@@ -1573,10 +1573,7 @@ impl Validator {
                 .port();
             let src_ip = match node.bind_ip_addrs.active() {
                 IpAddr::V4(ip) if !ip.is_unspecified() => Some(ip),
-                IpAddr::V4(_unspecified) => xdp_config
-                    .interface
-                    .as_ref()
-                    .and_then(|iface| master_ip_if_bonded(iface)),
+                IpAddr::V4(_unspecified) => master_ip_if_bonded(&xdp_config.network_device.if_name),
                 _ => panic!("IPv6 not supported"),
             };
             let (rtx, sender) = XdpRetransmitter::new(xdp_config, src_port, src_ip, exit.clone())

core/src/validator.rs

@@ -3,11 +3,8 @@ pub use agave_xdp::set_cpu_affinity;
 #[cfg(target_os = "linux")]
 use {
     agave_xdp::{
-        device::{NetworkDevice, QueueId},
-        load_xdp_program,
-        route::Router,
-        route_monitor::RouteMonitor,
-        tx_loop::tx_loop,
+        device::QueueId, get_network_device::NetworkDevice, load_xdp_program, route::Router,
+        route_monitor::RouteMonitor, tx_loop::tx_loop,
     },
     arc_swap::ArcSwap,
     crossbeam_channel::TryRecvError,
@@ -29,7 +26,7 @@ const ROUTE_MONITOR_UPDATE_INTERVAL: Duration = Duration::from_millis(50);
 
 #[derive(Clone, Debug)]
 pub struct XdpConfig {
-    pub interface: Option<String>,
+    pub network_device: NetworkDevice,
     pub cpus: Vec<usize>,
     pub zero_copy: bool,
     // The capacity of the channel that sits between retransmit stage and each XDP thread that
@@ -42,21 +39,10 @@ impl XdpConfig {
     const DEFAULT_RTX_CHANNEL_CAP: usize = 1_000_000;
 }
 
-impl Default for XdpConfig {
-    fn default() -> Self {
-        Self {
-            interface: None,
-            cpus: vec![],
-            zero_copy: false,
-            rtx_channel_cap: Self::DEFAULT_RTX_CHANNEL_CAP,
-        }
-    }
-}
-
 impl XdpConfig {
-    pub fn new(interface: Option<impl Into<String>>, cpus: Vec<usize>, zero_copy: bool) -> Self {
+    pub fn new(network_device: NetworkDevice, cpus: Vec<usize>, zero_copy: bool) -> Self {
         Self {
-            interface: interface.map(|s| s.into()),
+            network_device,
             cpus,
             zero_copy,
             rtx_channel_cap: XdpConfig::DEFAULT_RTX_CHANNEL_CAP,
@@ -138,19 +124,15 @@ impl XdpRetransmitter {
         };
         const DROP_CHANNEL_CAP: usize = 1_000_000;
 
+        let dev = config.network_device;
+
         // switch to higher caps while we setup XDP. We assume that an error in
         // this function is irrecoverable so we don't try to drop on errors.
         for cap in [CAP_NET_ADMIN, CAP_NET_RAW, CAP_BPF, CAP_PERFMON] {
             caps::raise(None, CapSet::Effective, cap)
                 .map_err(|e| format!("failed to raise {cap:?} capability: {e}"))?;
         }
 
-        let dev = Arc::new(if let Some(interface) = config.interface {
-            NetworkDevice::new(interface).unwrap()
-        } else {
-            NetworkDevice::new_from_default_route().unwrap()
-        });
-
         let ebpf = if config.zero_copy {
             Some(load_xdp_program(&dev).map_err(|e| format!("failed to attach xdp program: {e}"))?)
         } else {
@@ -205,7 +187,7 @@ impl XdpRetransmitter {
             .zip(config.cpus.into_iter())
             .enumerate()
         {
-            let dev = Arc::clone(&dev);
+            let dev = dev.clone();
             let drop_sender = drop_sender.clone();
             let atomic_router = Arc::clone(&atomic_router);
             threads.push(

turbine/src/xdp.rs

@@ -91,6 +91,7 @@ solana-vote-program = { workspace = true }
 symlink = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+agave-xdp ={workspace=true}
 
 [target.'cfg(not(any(target_env = "msvc", target_os = "freebsd")))'.dependencies]
 jemallocator = { workspace = true }

validator/Cargo.toml

@@ -11,6 +11,7 @@ use {
         snapshot_config::{SnapshotConfig, SnapshotUsage},
         ArchiveFormat, SnapshotInterval, SnapshotVersion,
     },
+    agave_xdp::get_network_device::get_network_device,
     clap::{crate_name, value_t, value_t_or_exit, values_t, values_t_or_exit, ArgMatches},
     crossbeam_channel::unbounded,
     log::*,
@@ -462,11 +463,10 @@ pub fn execute(
     let xdp_interface = matches.value_of("retransmit_xdp_interface");
     let xdp_zero_copy = matches.is_present("retransmit_xdp_zero_copy");
     let retransmit_xdp = matches.value_of("retransmit_xdp_cpu_cores").map(|cpus| {
-        XdpConfig::new(
-            xdp_interface,
-            parse_cpu_ranges(cpus).unwrap(),
-            xdp_zero_copy,
-        )
+        // this will only succeed on linux
+        let xdp_device =
+            get_network_device(xdp_interface).expect("XDP interface must be a valid net device");
+        XdpConfig::new(xdp_device, parse_cpu_ranges(cpus).unwrap(), xdp_zero_copy)
     });
 
     let account_paths: Vec<PathBuf> =

validator/src/commands/run/execute.rs

@@ -24,9 +24,10 @@ use {
 #[derive(Copy, Clone, Debug)]
 pub struct QueueId(pub u64);
 
+#[derive(Debug, Clone)]
 pub struct NetworkDevice {
     if_index: u32,
-    if_name: String,
+    pub if_name: String,
 }
 
 impl NetworkDevice {

xdp/src/device.rs

@@ -36,6 +36,9 @@ pub mod umem;
 
 #[cfg(target_os = "linux")]
 pub use program::load_xdp_program;
+
+pub mod get_network_device;
+
 use std::io;
 
 #[cfg(target_os = "linux")]