diff --git a/Cargo.lock b/Cargo.lock index e2c5a72bde..44cf32073f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -123,7 +123,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -134,7 +134,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1457,7 +1457,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1469,23 +1469,6 @@ dependencies = [ "arrayvec", ] -[[package]] -name = "ethtool" -version = "0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7f0218cf6f43c8207b24602757c18a43dc8cd71a24ecce4b656489a2f27bc76" -dependencies = [ - "futures", - "genetlink", - "log", - "netlink-packet-core", - "netlink-packet-generic", - "netlink-proto", - "netlink-sys", - "thiserror 1.0.69", - "tokio", -] - [[package]] name = "event-listener" version = "5.4.1" @@ -1779,20 +1762,6 @@ dependencies = [ "version_check", ] -[[package]] -name = "genetlink" -version = "0.2.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d8935531e8e0919b17043c668cc18bfac1622f2fab73125f4f018124ee330b8" -dependencies = [ - "futures", - "log", - "netlink-packet-core", - "netlink-packet-generic", - "netlink-proto", - "thiserror 1.0.69", -] - [[package]] name = "getrandom" version = "0.2.17" @@ -1886,9 +1855,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.14" +version = "0.4.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "171fefbc92fe4a4de27e0698d6a5b392d6a0e333506bc49133760b3bcf948733" +checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" dependencies = [ "atomic-waker", "bytes", @@ -2463,7 +2432,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3159,52 +3128,6 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b" -[[package]] -name = "netlink-packet-core" -version = "0.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3463cbb78394cb0141e2c926b93fc2197e473394b761986eca3b9da2c63ae0f4" -dependencies = [ - "paste", -] - -[[package]] -name = "netlink-packet-generic" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f891b2e0054cac5a684a06628f59568f841c93da4e551239da6e518f539e775" -dependencies = [ - "netlink-packet-core", -] - -[[package]] -name = "netlink-proto" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6f7398dddf5f152d2a91a2921a134c6097056e292c0d4b9906007855e7cece6" -dependencies = [ - "bytes", - "futures-channel", - "futures-util", - "log", - "netlink-packet-core", - "netlink-sys", - "thiserror 2.0.18", -] - -[[package]] -name = "netlink-sys" -version = "0.8.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cd6c30ed10fa69cc491d491b85cc971f6bdeb8e7367b7cde2ee6cc878d583fae" -dependencies = [ - "bytes", - "futures-util", - "libc", - "log", - "tokio", -] - [[package]] name = "nix" version = "0.27.1" @@ -3300,7 +3223,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4148,11 +4071,8 @@ dependencies = [ "aya", "aya-log", "aya-obj", - "ethtool", - "futures", "libc", "thiserror 2.0.18", - "tokio", "xdp", ] @@ -4649,7 +4569,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4708,7 +4628,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4837,7 +4757,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5070,7 +4990,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -5343,7 +5263,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6208,7 +6128,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6629,9 +6549,9 @@ dependencies = [ [[package]] name = "xdp" -version = "0.7.3" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21a573b5aa423fecf27e2dc9c4009db67cd2f35718c2a859d7352ee3fe0d1780" +checksum = "9efb2e28167f29f3f15e2ab4a1fcd56df6e01d33052b95db2b28a93568e88cce" [[package]] name = "yansi" diff --git a/Cargo.toml b/Cargo.toml index 2174de89ae..705a593e32 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -282,7 +282,7 @@ tracing-subscriber = "0.3" tryhard = "0.5.2" url = { version = "2.5.8", features = ["serde"] } uuid = { version = "1.19.0", default-features = false, features = ["v4"] } -xdp = "0.7.1" +xdp = "0.8.0" # Corrosion libraries. The libraries that corrosion transitively depends on that # we also need to depend on should be kept in line with the versions used by corrosion diff --git a/about.toml b/about.toml index 382ca2bdc0..ea0271c4ef 100644 --- a/about.toml +++ b/about.toml @@ -41,10 +41,13 @@ accepted = ["BSL-1.0"] [enquote] accepted = ["Unlicense"] +[minicbor] +accepted = ["BlueOak-1.0.0"] + [webpki.clarify] license = "ISC AND Apache-2.0 AND BSD-3-Clause" files = [ { path = "LICENSE", checksum = "5b698ca13897be3afdb7174256fa1574f8c6892b8bea1a66dd6469d3fe27885a", license = "ISC" }, { path = "third-party/bettertls/LICENSE", checksum = "0cec06e0e55fbc3dc5cee4fca9b607f66cb8f4e4dbcf3b3c013594dd156732e9", license = "Apache-2.0" }, { path = "third-party/chromium/LICENSE", checksum = "845022e0c1db1abb41a6ba4cd3c4b674ec290f3359d9d3c78ae558d4c0ed9308", license = "BSD-3-Clause" }, -] \ No newline at end of file +] diff --git a/crates/xdp/Cargo.toml b/crates/xdp/Cargo.toml index 59daf38bd7..2c32b7c941 100644 --- a/crates/xdp/Cargo.toml +++ b/crates/xdp/Cargo.toml @@ -16,11 +16,7 @@ aya = { git = "https://github.com/aya-rs/aya", rev = "2453204" } # Allows forwarding trace messages from eBPF programs to a global logger, note we use a git dependency here as aya has # updated many dependencies but has not made a release in over half a year aya-log = { git = "https://github.com/aya-rs/aya", rev = "2453204" } -# Used to query/set NIC queue channels (`ethtool -l`/`-L`) via netlink -ethtool.workspace = true -futures.workspace = true libc.workspace = true -tokio.workspace = true [target.'cfg(target_os = "linux")'.dev-dependencies] # Validates the committed eBPF object, must be the same revision as `aya` diff --git a/crates/xdp/src/channels.rs b/crates/xdp/src/channels.rs deleted file mode 100644 index 02098f5349..0000000000 --- a/crates/xdp/src/channels.rs +++ /dev/null @@ -1,151 +0,0 @@ -//! Queries and updates a NIC's queue channel count via the kernel's ethtool -//! netlink interface (`ethtool -l`/`ethtool -L`). - -use crate::NicIndex; - -/// Queue channel counts reported by `ETHTOOL_MSG_CHANNELS_GET` (`ethtool -l`). -#[derive(Debug, Default, Clone, Copy)] -pub struct Channels { - combined_max: u32, - rx_count: u32, - tx_count: u32, - combined_count: u32, -} - -impl Channels { - /// Queues actually in use, whichever reporting style the driver uses. - pub fn current(&self) -> u32 { - self.rx_count.max(self.tx_count).max(self.combined_count) - } - - /// Some drivers report separate RX/TX queues, others a single combined count. - pub fn uses_combined(&self) -> bool { - self.combined_max > 0 - } - - fn from_nlas(nlas: Vec) -> Self { - let mut channels = Self::default(); - for nla in nlas { - let ethtool::EthtoolAttr::Channel(attr) = nla else { - continue; - }; - match attr { - ethtool::EthtoolChannelAttr::CombinedMax(v) => channels.combined_max = v, - ethtool::EthtoolChannelAttr::RxCount(v) => channels.rx_count = v, - ethtool::EthtoolChannelAttr::TxCount(v) => channels.tx_count = v, - ethtool::EthtoolChannelAttr::CombinedCount(v) => channels.combined_count = v, - _ => {} - } - } - channels - } -} - -/// Queries `iface`'s current queue channels (`ethtool -l`). -async fn get_channels(iface: &str) -> Result { - use futures::TryStreamExt as _; - - let (connection, mut handle, _) = - ethtool::new_connection().map_err(|error| ethtool::EthtoolError::Bug(error.to_string()))?; - tokio::spawn(connection); - - let mut stream = handle.channel().get(Some(iface)).execute().await; - let mut channels = Channels::default(); - while let Some(msg) = stream.try_next().await? { - channels = Channels::from_nlas(msg.payload.nlas); - } - Ok(channels) -} - -/// Sets `iface`'s queue channels to `count`, using combined-style if -/// `uses_combined`, else separate rx/tx (`ethtool -L`). -async fn set_channel_count( - iface: &str, - uses_combined: bool, - count: u32, -) -> Result<(), ethtool::EthtoolError> { - let (connection, mut handle, _) = - ethtool::new_connection().map_err(|error| ethtool::EthtoolError::Bug(error.to_string()))?; - tokio::spawn(connection); - - let request = handle.channel().set(iface); - if uses_combined { - request.combined_count(count).execute().await - } else { - request.rx_count(count).tx_count(count).execute().await - } -} - -/// Bridges an async ethtool call into this sync-facing API; needs a -/// multi-threaded runtime, same as the rest of XDP setup. -fn block_on(fut: F) -> F::Output { - tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(fut)) -} - -/// Resolves `nic`'s interface name as UTF-8, the form the `ethtool` crate needs. -fn iface_name(nic: NicIndex) -> std::io::Result { - let name = nic - .name() - .map_err(|_err| std::io::Error::from(std::io::ErrorKind::NotFound))?; - name.as_str().map(str::to_owned).ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidData, - "interface name is not utf-8", - ) - }) -} - -/// Queries `nic`'s current queue channels (`ethtool -l`); the single source -/// of truth for queue count and reporting style. -pub fn query_channels(nic: NicIndex) -> std::io::Result { - let name = iface_name(nic)?; - block_on(get_channels(&name)).map_err(std::io::Error::other) -} - -/// Reduces `nic`'s configured queue count to `count` (`ethtool -L`). -pub fn shrink_queue_count(nic: NicIndex, uses_combined: bool, count: u32) -> std::io::Result<()> { - let name = iface_name(nic)?; - block_on(set_channel_count(&name, uses_combined, count)).map_err(std::io::Error::other) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn channels_current_takes_whichever_style_is_populated() { - let separate = Channels { - combined_max: 0, - rx_count: 4, - tx_count: 4, - combined_count: 0, - }; - assert_eq!(separate.current(), 4); - assert!(!separate.uses_combined()); - - let combined = Channels { - combined_max: 8, - rx_count: 0, - tx_count: 0, - combined_count: 2, - }; - assert_eq!(combined.current(), 2); - assert!(combined.uses_combined()); - } - - #[test] - fn channels_from_nlas_parses_relevant_attrs_and_ignores_others() { - let nlas = vec![ - ethtool::EthtoolAttr::Channel(ethtool::EthtoolChannelAttr::CombinedMax(4)), - ethtool::EthtoolAttr::Channel(ethtool::EthtoolChannelAttr::RxCount(2)), - ethtool::EthtoolAttr::Channel(ethtool::EthtoolChannelAttr::TxCount(2)), - ethtool::EthtoolAttr::Channel(ethtool::EthtoolChannelAttr::OtherCount(0)), - ]; - - let channels = Channels::from_nlas(nlas); - assert_eq!(channels.combined_max, 4); - assert_eq!(channels.rx_count, 2); - assert_eq!(channels.tx_count, 2); - assert_eq!(channels.combined_count, 0); - } -} diff --git a/crates/xdp/src/lib.rs b/crates/xdp/src/lib.rs index 92f73e80ba..d0741565a8 100644 --- a/crates/xdp/src/lib.rs +++ b/crates/xdp/src/lib.rs @@ -2,8 +2,3 @@ mod linux; #[cfg(target_os = "linux")] pub use linux::*; - -#[cfg(target_os = "linux")] -mod channels; -#[cfg(target_os = "linux")] -pub use channels::*; diff --git a/crates/xdp/src/linux.rs b/crates/xdp/src/linux.rs index a7108214cf..77129d2307 100644 --- a/crates/xdp/src/linux.rs +++ b/crates/xdp/src/linux.rs @@ -162,8 +162,10 @@ impl EbpfProgram { self.bpf.map_mut("XSK").expect("failed to retrieve XSK map"), )?; - let mut entries = Vec::with_capacity(device_caps.queue_count as _); - for i in 0..device_caps.queue_count { + let queue_count = device_caps.queues.rx_count(); + + let mut entries = Vec::with_capacity(queue_count as _); + for i in 0..queue_count { let umem = xdp::Umem::map(umem_cfg)?; let mut sb = xdp::socket::XdpSocketBuilder::new()?; let (rings, mut bind_flags) = sb.build_wakable_rings(&umem, ring_cfg)?; diff --git a/src/net/io/nic/xdp.rs b/src/net/io/nic/xdp.rs index 0320de4a36..64a33e684e 100644 --- a/src/net/io/nic/xdp.rs +++ b/src/net/io/nic/xdp.rs @@ -90,8 +90,6 @@ pub enum XdpSetupError { NicUnavailable(#[from] NicUnavailable), #[error("failed to query device capabilities for {0}: {1}")] NicQuery(NicName, #[source] std::io::Error), - #[error("failed to query queue channels for {0}: {1}")] - ChannelsQuery(NicName, #[source] std::io::Error), #[error("failed to query ip addresses for {0}: {1}")] AddressQuery(NicName, #[source] std::io::Error), #[error("`XDP_ZEROCOPY` is unavailable for {0}")] @@ -165,28 +163,26 @@ fn next_queue_count_to_try(current: u32) -> Option { fn attach_xdp_program( ebpf_prog: &mut quilkin_xdp::EbpfProgram, nic: NicIndex, - initial_channels: quilkin_xdp::Channels, -) -> Result< - (quilkin_xdp::aya::programs::xdp::XdpLinkId, u32), - quilkin_xdp::aya::programs::ProgramError, -> { + queues: &mut quilkin_xdp::xdp::nic::Queues, +) -> Result { ebpf_prog.load_into_kernel()?; let mode = quilkin_xdp::aya::programs::xdp::XdpMode::default(); - let uses_combined = initial_channels.uses_combined(); - let mut queue_count = initial_channels.current(); + // We've already synchronized the tx and rx counts if needed before this function + let mut queue_count = queues.rx_count(); loop { match ebpf_prog.attach(nic, mode) { - Ok(link) => return Ok((link, queue_count)), + Ok(link) => return Ok(link), Err(error) => { - let Some(reduced) = next_queue_count_to_try(queue_count) else { + let reduced = queue_count / 2; + if reduced < 1 { return Err(error); - }; + } + + queues.set_rx_and_tx(reduced); - if let Err(shrink_error) = - quilkin_xdp::shrink_queue_count(nic, uses_combined, reduced) - { + if let Err(shrink_error) = nic.set_queue_counts(queues) { tracing::warn!( %shrink_error, ?nic, @@ -214,7 +210,7 @@ fn attach_xdp_program( /// Binding to fewer queues is possible in the future but requires additional /// work in the `xdp` crate pub fn setup_xdp_io(config: XdpConfig<'_>) -> Result { - let nic_index = match config.nic { + let nic = match config.nic { NicConfig::Default => { let mut chosen = None; @@ -240,38 +236,31 @@ pub fn setup_xdp_io(config: XdpConfig<'_>) -> Result NicConfig::Index(index) => xdp::nic::NicIndex::new(index), }; - let name = nic_index + let name = nic .name() - .map_err(|_err| NicUnavailable::UnknownIndex(nic_index.into()))?; + .map_err(|_err| NicUnavailable::UnknownIndex(nic.into()))?; - tracing::info!(nic = ?nic_index, "selected NIC"); + tracing::info!(nic = ?nic, "selected NIC"); - let mut device_caps = nic_index + let mut device_caps = nic .query_capabilities() .map_err(|err| XdpSetupError::NicQuery(name, err))?; - // `query_capabilities` reports its own queue count, but `query_channels` - // is the richer, authoritative source (it also knows the reporting - // style), so supersede it here rather than tracking two queue counts. - let channels = quilkin_xdp::query_channels(nic_index) - .map_err(|err| XdpSetupError::ChannelsQuery(name, err))?; - device_caps.queue_count = channels.current(); - - tracing::debug!(?device_caps, nic = ?nic_index, "XDP features for device"); + tracing::info!(?device_caps, nic = ?nic, "XDP features for device"); if config.require_zero_copy && matches!(device_caps.zero_copy, xdp::nic::XdpZeroCopy::Unavailable) { - tracing::error!(?device_caps, nic = ?nic_index, "XDP features for device"); + tracing::error!(?device_caps, nic = ?nic, "XDP features for device"); return Err(XdpSetupError::ZeroCopyUnavailable(name)); } if config.require_tx_checksum && !device_caps.tx_metadata.checksum() { - tracing::error!(?device_caps, nic = ?nic_index, "XDP features for device"); + tracing::error!(?device_caps, nic = ?nic, "XDP features for device"); return Err(XdpSetupError::TxChecksumUnavailable(name)); } - let (ipv4, ipv6) = nic_index + let (ipv4, ipv6) = nic .addresses() .and_then(|(ipv4, ipv6)| { if ipv4.is_none() && ipv6.is_none() { @@ -294,8 +283,34 @@ pub fn setup_xdp_io(config: XdpConfig<'_>) -> Result // and we only need 2k since we only care about non-fragmented UDP packets const PACKET_SIZE: u64 = 2 * 1024; + // It's extremely unlikely these number will ever differ, but if they do we take the minimum + let rx_count = device_caps.queues.rx_count(); + let tx_count = device_caps.queues.tx_count(); + + if rx_count != tx_count { + tracing::warn!( + ?nic, + rx_count, + tx_count, + "the selected NIC has a different number of RX and TX queues" + ); + let qc = rx_count.min(tx_count); + + // If the tx count is higher it doesn't really matter since they just wouldn't be used, but if we have more rx than tx + // we need to reduce as otherwise we might not be able to send on some of the i/o loops + if rx_count > tx_count { + device_caps.queues.set_rx_and_tx(qc); + if let Err(error) = nic.set_queue_counts(&device_caps.queues) { + tracing::error!(%error, ?nic, rx_count, tx_count, "failed to synchronize queue counts before attach"); + return Err(XdpSetupError::NicQuery(name, error)); + } + } + } + + let queue_count = device_caps.queues.rx_count(); + let packet_count = if let Some(max) = config.maximum_packet_memory { - let bytes_per_socket = max / device_caps.queue_count as u64; + let bytes_per_socket = max / queue_count as u64; let packet_count = (bytes_per_socket / PACKET_SIZE).next_power_of_two(); if MINIMUM_UMEM_COUNT > packet_count { fn byte_units(b: u64) -> (f64, &'static str) { @@ -312,8 +327,7 @@ pub fn setup_xdp_io(config: XdpConfig<'_>) -> Result } let (max, xunit) = byte_units(max); - let (min, nunit) = - byte_units(MINIMUM_UMEM_COUNT * PACKET_SIZE * device_caps.queue_count as u64); + let (min, nunit) = byte_units(MINIMUM_UMEM_COUNT * PACKET_SIZE * queue_count as u64); return Err(XdpSetupError::MinimumMemoryRequirementsExceeded { max, @@ -321,7 +335,7 @@ pub fn setup_xdp_io(config: XdpConfig<'_>) -> Result min, nunit, nic: name, - queue_count: device_caps.queue_count, + queue_count, }); } @@ -334,8 +348,7 @@ pub fn setup_xdp_io(config: XdpConfig<'_>) -> Result // Attach before binding: some drivers (eg gve) need XDP already enabled // before a socket can bind a queue with `XDP_ZEROCOPY`. - let (xdp_link, queue_count) = attach_xdp_program(&mut ebpf_prog, nic_index, channels)?; - device_caps.queue_count = queue_count; + let xdp_link = attach_xdp_program(&mut ebpf_prog, nic, &mut device_caps.queues)?; let umem_cfg = xdp::umem::UmemCfgBuilder { frame_size: xdp::umem::FrameSize::TwoK, @@ -355,12 +368,12 @@ pub fn setup_xdp_io(config: XdpConfig<'_>) -> Result .build()?; let ring_cfg = xdp::RingConfigBuilder::default().build()?; - let workers = ebpf_prog.create_and_bind_sockets(nic_index, umem_cfg, &device_caps, ring_cfg)?; + let workers = ebpf_prog.create_and_bind_sockets(nic, umem_cfg, &device_caps, ring_cfg)?; Ok(XdpWorkers { ebpf_prog, workers, - nic: nic_index, + nic, xdp_link, external_port: config.external_port.into(), qcmp_port: config.qcmp_port.into(),