From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [IPv6:2a0f:8001:1:32::40]) by lore.proxmox.com (Postfix) with ESMTPS id E16591FF0B3 for ; Wed, 09 Sep 2026 12:40:37 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 2A1AB215D4; Wed, 09 Sep 2026 12:40:33 +0200 (CEST) From: Hannes Laimer To: pve-devel@lists.proxmox.com Subject: [PATCH proxmox-ebpf v3 1/3] add the shared tc subsystem code Date: Wed, 9 Sep 2026 12:39:50 +0200 Message-ID: <20260909103952.1108084-2-h.laimer@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20260909103952.1108084-1-h.laimer@proxmox.com> References: <20260909103952.1108084-1-h.laimer@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1788950389822 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.575 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: BSSEXVNOO4IJZPC7R3UTWZ423KUKAD2P X-Message-ID-Hash: BSSEXVNOO4IJZPC7R3UTWZ423KUKAD2P X-MailFrom: h.laimer@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox VE development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: A subsystem owns tc classifier programs and their maps, attached per interface and pinned in bpffs together with an object fingerprint and a schema version. So state survives between the one-shot calls from the subsystem driving it, and re-running never interrupts traffic. Apply paths coordinate through a shared flock, loading takes it exclusively. A link is pinned under its interface's index, name and direction. So an unplug reported once the interface is gone still finds the pin, and a reused index does not pass for the old link. Changes are ordered by a counter the subsystem draws before a change reads its input. A full apply records the generation it took, and a later one under an older generation is dropped. A single change carries its generation in the entry it writes and never overwrites a newer one. Waits on the locks are bounded, a holder that never returns fails the waiter instead of hanging it. Signed-off-by: Hannes Laimer --- src/lib.rs | 3 + src/subsystem.rs | 901 +++++++++++++++++++++++++++++++++++++++++++++++ src/tc.rs | 257 ++++++++++++++ 3 files changed, 1161 insertions(+) create mode 100644 src/subsystem.rs create mode 100644 src/tc.rs diff --git a/src/lib.rs b/src/lib.rs index 244f44c..572d861 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,2 +1,5 @@ //! eBPF subsystems, one per concern, each behind a cargo feature of its name. Nothing here runs //! on its own, consumers pull in only the subsystem they drive. + +pub mod subsystem; +pub mod tc; diff --git a/src/subsystem.rs b/src/subsystem.rs new file mode 100644 index 0000000..efae754 --- /dev/null +++ b/src/subsystem.rs @@ -0,0 +1,901 @@ +//! The shared subsystem code. A subsystem has one tc program per direction it declares, with +//! their maps and links. They are pinned under `/sys/fs/bpf/proxmox-ebpf///`. The +//! programs are pinned by the fingerprint of their object, so the pinned state itself says which +//! build it belongs to. +//! The loaded BPF stays in the kernel between invocations, so [`ApplyLock::ensure_loaded`] loads +//! and verifies only when nothing current is pinned, on a schema or object change and after a +//! lost map pin. Everything else attaches links and syncs maps against what is already there. +//! Each subsystem owns one [`TcPrograms`]. +//! +//! A subsystem compiles its desired state outside the lock, so two of its changes can reach the +//! kernel in the wrong order. A generation from [`TcPrograms::next_generation`] orders them. A +//! full apply runs through [`ApplyLock::with_generation`] and is dropped when a newer one is in +//! place. A single-entry change keeps its generation in the entry it writes and compares there, +//! under [`ApplyLock::change`]. +use std::{ + collections::HashSet, + fs::File, + io::ErrorKind, + marker::PhantomData, + path::{Path, PathBuf}, + time::{Duration, Instant}, +}; + +use anyhow::Context; +use aya::{EbpfLoader, programs::SchedClassifier}; +use nix::{ + errno::Errno, + fcntl::{Flock, FlockArg}, +}; + +use crate::tc::{self, Direction}; + +const PIN_ROOT: &str = "/sys/fs/bpf/proxmox-ebpf"; +const RUN_ROOT: &str = "/run/proxmox-ebpf"; + +// a wait for the lock normally lasts one pass. The bound only catches a holder that never +// returns. flock has no timeout, so the wait asks nonblocking and sleeps in between + +const LOCK_WAIT: Duration = Duration::from_secs(10); +const LOCK_POLL: Duration = Duration::from_millis(10); + +fn flock_within( + mut file: File, + arg: FlockArg, + what: &str, + subsystem: &str, +) -> anyhow::Result> { + let nonblocking = match arg { + FlockArg::LockShared => FlockArg::LockSharedNonblock, + FlockArg::LockExclusive => FlockArg::LockExclusiveNonblock, + other => other, + }; + let deadline = Instant::now() + LOCK_WAIT; + loop { + match Flock::lock(file, nonblocking) { + Ok(lock) => return Ok(lock), + Err((returned, Errno::EWOULDBLOCK)) if Instant::now() < deadline => { + file = returned; + std::thread::sleep(LOCK_POLL); + } + Err((_, Errno::EWOULDBLOCK)) => anyhow::bail!( + "{subsystem}: {what} still held after {}s", + LOCK_WAIT.as_secs() + ), + Err((_, e)) => return Err(e.into()), + } + } +} + +/// A step that needs the programs pinned found none. Only [`ApplyLock::ensure_loaded`] under the +/// exclusive lock installs them, so a shared holder hands the work back to its caller. +#[derive(Debug)] +pub struct NotLoaded; + +impl std::fmt::Display for NotLoaded { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("programs not loaded") + } +} + +impl std::error::Error for NotLoaded {} + +/// The pinned tc programs for one subsystem. +/// +/// Holds only the immutable description. The loaded BPF lives pinned in the kernel and is reached +/// through those pins, so normal operation runs no verifier. +pub struct TcPrograms { + /// The subsystem's name, its directory under the pin root. + pub name: &'static str, + /// The embedded BPF object. + pub obj: &'static [u8], + /// [`obj_fingerprint`](Self::obj_fingerprint) of `obj`, folded into a `const`. + pub fingerprint: u64, + /// The name of the object's program for a direction. + pub prog_name: fn(Direction) -> &'static str, + /// The directions the object carries a program for. + pub directions: &'static [Direction], + /// The object's pinned maps, they count towards the state being current. + pub maps: &'static [&'static str], + /// Bumped with any change to a map's key or value layout, a pinned map is reused by name + /// with no check beyond its key and value sizes. + pub schema_version: u32, +} + +impl TcPrograms { + fn pin_root(&self) -> PathBuf { + PathBuf::from(PIN_ROOT).join(self.name) + } + fn schema_root(&self) -> PathBuf { + self.pin_root().join(self.schema_version.to_string()) + } + fn links_dir(&self) -> PathBuf { + self.schema_root().join("links") + } + fn link_pin_path(&self, pin: &tc::LinkPin) -> PathBuf { + self.links_dir().join(pin.filename()) + } + fn progs_dir(&self) -> PathBuf { + self.schema_root().join("prog") + } + fn fingerprint_dir_name(&self) -> String { + format!("{:016x}", self.fingerprint) + } + fn prog_dir(&self) -> PathBuf { + self.progs_dir().join(self.fingerprint_dir_name()) + } + fn prog_pin_path(&self, dir: Direction) -> PathBuf { + self.prog_dir().join(dir.as_str()) + } + fn run_dir(&self) -> PathBuf { + PathBuf::from(RUN_ROOT).join(self.name) + } + fn generation_path(&self) -> PathBuf { + self.run_dir().join("generation") + } + fn applied_path(&self) -> PathBuf { + self.run_dir().join("applied") + } + + // the counter and the stamp of the last full apply are plain files next to the apply lock, + // read and written under this lock. A single-entry change holds it for its own read, compare + // and write too, so changes serialize without giving up the apply lock's sharing + + fn generation_lock(&self) -> anyhow::Result> { + std::fs::create_dir_all(self.run_dir())?; + let file = File::options() + .create(true) + .truncate(false) + .write(true) + .open(self.run_dir().join("generation.lock"))?; + flock_within(file, FlockArg::LockExclusive, "generation lock", self.name) + } + + /// Draw the next generation. Draw after the input of a change is complete and before that + /// input is read, then two changes are ordered by the input they saw and not by when they + /// reached the kernel. A draw before the input was written orders nothing, a newer change may + /// then read the older state and win. Restarts with the node like everything it orders. + pub fn next_generation(&self) -> anyhow::Result { + let _lock = self.generation_lock()?; + let next = read_stamp(&self.generation_path())?.unwrap_or(0) + 1; + write_stamp(&self.generation_path(), next)?; + Ok(next) + } + + fn hash_map( + &self, + name: &str, + ) -> anyhow::Result> { + let map = aya::maps::MapData::from_pin(self.schema_root().join(name)) + .with_context(|| format!("{}: open pinned map {name}", self.name))?; + Ok(aya::maps::HashMap::try_from(aya::maps::Map::HashMap(map))?) + } + + /// Open one direction's pinned program. A plain `BPF_OBJ_GET`, no verifier, since the program + /// was checked once when [`ensure_loaded`](ApplyLock::ensure_loaded) installed it. + fn program(&self, dir: Direction) -> anyhow::Result { + let path = self.prog_pin_path(dir); + if !path.exists() { + return Err(NotLoaded.into()); + } + SchedClassifier::from_pin(path) + .with_context(|| format!("{}: open pinned program {}", self.name, dir.as_str())) + } + + /// FNV-1a over the embedded object. A `const fn`, so each subsystem folds it into a `const` at + /// compile time and the per-invocation path never rehashes a constant. + pub const fn obj_fingerprint(obj: &[u8]) -> u64 { + let mut hash = 0xcbf29ce484222325u64; + let mut i = 0; + while i < obj.len() { + hash ^= obj[i] as u64; + hash = hash.wrapping_mul(0x100000001b3); + i += 1; + } + hash + } + + /// The per-subsystem apply lock under `/run`, one file taken in two modes. A full apply and + /// any install or teardown take it [exclusively](Self::lock_exclusive), so enumerate, detach + /// and attach run as one unit. A single-interface step takes it [shared](Self::lock_shared), + /// so guest plugs run concurrently and block only behind a full apply. The kernel drops the + /// lock with a dying process. A wait is bounded, a holder that never returns fails the + /// waiter instead of hanging it. Shared holders are not held back for a waiting exclusive + /// one, a node plugging without pause could keep a full apply waiting until its bound. Every + /// write to the pinned state, and the check whether it is current, is a method of the + /// returned guard. So a step cannot run outside the lock, and the mode it needs is checked + /// at compile time. + fn lock(&self, arg: FlockArg) -> anyhow::Result> { + let dir = self.run_dir(); + std::fs::create_dir_all(&dir)?; + let file = File::options() + .create(true) + .truncate(false) + .write(true) + .open(dir.join("lock"))?; + let lock = flock_within(file, arg, "apply lock", self.name)?; + Ok(ApplyLock { + programs: self, + _lock: lock, + _mode: PhantomData, + }) + } + + /// Whether a link of the interface name is pinned, read without the lock. Most interfaces + /// never had a link and their unplug must not wait behind a pass. A miss is a snapshot, a pin + /// arriving after it is another attach's. A failed listing counts as a hit. So do links still + /// pinned under another schema, an install may be moving them over. + pub fn link_pinned(&self, name: &str) -> bool { + let elsewhere = self.other_schemas().map_or(true, |others| { + others.iter().any(|(dir, _)| any_links(&dir.join("links"))) + }); + elsewhere + || self + .live_links() + .map_or(true, |links| links.iter().any(|pin| pin.name == name)) + } + + /// Take the apply lock in shared mode, for an additive single-interface apply. + pub fn lock_shared(&self) -> anyhow::Result> { + self.lock(FlockArg::LockShared) + } + + /// Take the apply lock exclusively, for a full apply or an install/teardown. + pub fn lock_exclusive(&self) -> anyhow::Result> { + self.lock(FlockArg::LockExclusive) + } + + fn ensure_loaded(&self) -> anyhow::Result<()> { + let others = self.other_schemas()?; + // a link pinned under another schema is a live attachment to a program of that schema. + // A run that died between moving the links and pinning its programs leaves them there. + // This build is not current until the install below has taken them over + + let links_elsewhere = others.iter().any(|(dir, _)| any_links(&dir.join("links"))); + // a newer schema with programs means a newer build already runs here and this process + // is the stale one, it must not touch that. A directory of a newer schema holding no + // program is what a rejected install of that build left behind, it goes with the rest + if let Some(newer) = others + .iter() + .filter(|(dir, _)| any_program_pinned(&dir.join("prog"))) + .filter_map(|(_, schema)| *schema) + .find(|&s| s > self.schema_version) + { + anyhow::bail!( + "{}: pinned state is of schema {newer}, newer than this build's {}, clear it to \ + downgrade", + self.name, + self.schema_version + ); + } + if self.is_current() && !links_elsewhere { + // an older schema is what a run left that died between installing and removing it, + // its links moved over already. A newer schema without programs is what a rejected + // install left, its maps in the kernel with nothing using them + + for (dir, schema) in others { + if (schema.is_some_and(|s| s < self.schema_version) + || !any_program_pinned(&dir.join("prog"))) + && let Err(e) = std::fs::remove_dir_all(&dir) + { + log::warn!( + "{}: remove old pinned state {}: {e:#}", + self.name, + dir.display() + ); + } + } + return Ok(()); + } + + // a lost map pin comes back at load time, ahead of the programs. If the install then + // failed, the previous run's program pins would pass that half-done state for current. + // So they go first, the links keep the old programs running meanwhile + + if !self.maps_pinned() { + match std::fs::remove_dir_all(self.prog_dir()) { + Ok(()) => {} + Err(e) if e.kind() == ErrorKind::NotFound => {} + Err(e) => return Err(e).context("remove program pins ahead of the reinstall"), + } + } + + // any other schema pinned here is old state whose map layout this build cannot bind to. + // Its links move onto the new programs while it is still in place, so no interface runs + // without one and a verifier rejection leaves it serving. The rest goes afterwards + let schema_changed = !others.is_empty(); + if schema_changed { + log::warn!("{}: pinned state of another schema, rebuilding", self.name); + } + + // a failed install leaves its directory as it is. A link moved into it already is a live + // attachment, the next run swaps it onto its own object again. Without program pins the + // directory does not pass for state of that build + + let old_links: Vec = others.iter().map(|(dir, _)| dir.join("links")).collect(); + self.install(&old_links).context("install")?; + for (dir, _) in &others { + // a link whose pin could not be moved is still live there, the next run swaps it + // onto its programs again + if any_links(&dir.join("links")) { + continue; + } + if let Err(e) = std::fs::remove_dir_all(dir) { + log::warn!( + "{}: remove old pinned state {}: {e:#}", + self.name, + dir.display() + ); + } + } + Ok(()) + } + + // every direction's program pinned under this build's schema and fingerprint and every map + // with them + fn is_current(&self) -> bool { + self.directions + .iter() + .all(|&dir| self.prog_pin_path(dir).exists()) + && self.maps_pinned() + } + + fn maps_pinned(&self) -> bool { + self.maps + .iter() + .all(|map| self.schema_root().join(map).exists()) + } + + // the state pinned here under schema versions other than this build's. A directory that is + // no version at all is taken along as such state, its links moved over. A plain file is + // not state + fn other_schemas(&self) -> anyhow::Result)>> { + let entries = match std::fs::read_dir(self.pin_root()) { + Ok(entries) => entries, + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => return Err(e.into()), + }; + let mut others = Vec::new(); + for entry in entries { + let entry = entry?; + if !entry.file_type()?.is_dir() { + continue; + } + let schema = entry + .file_name() + .to_str() + .and_then(|s| s.parse::().ok()); + if schema != Some(self.schema_version) { + others.push((entry.path(), schema)); + } + } + Ok(others) + } + + fn tear_down(&self) -> anyhow::Result<()> { + match std::fs::remove_dir_all(self.pin_root()) { + Ok(()) => Ok(()), + Err(e) if e.kind() == ErrorKind::NotFound => Ok(()), + Err(e) => Err(e.into()), + } + } + + /// Load and verify the object. Move the pinned links onto it, this schema's and those under + /// `old_links`. Then pin its programs and drop those of other builds. The verifier runs here + /// and nowhere else, a rejection leaves what served before untouched. The programs are pinned + /// last since their pins mark this build as current. A failure before that leaves the + /// previous build current, and the next run repeats the install. Links moved over already are + /// swapped onto its object again. A link the update cannot reach has lost its netdev and is + /// unpinned, the next reconcile attaches the interface fresh if still wanted. + fn install(&self, old_links: &[PathBuf]) -> anyhow::Result<()> { + std::fs::create_dir_all(self.links_dir())?; + self.remove_foreign_pins()?; + let mut bpf = EbpfLoader::new() + .map_pin_path(self.schema_root()) + .load(self.obj)?; + // load everything before touching any link, so a verifier rejection of one direction + // leaves the old state whole + for &dir in self.directions { + let name = (self.prog_name)(dir); + let prog: &mut SchedClassifier = bpf + .program_mut(name) + .with_context(|| format!("{}: program {name} not found in object", self.name))? + .try_into()?; + prog.load()?; + } + + let live = self.live_links()?; + for &dir in self.directions { + let prog = self.loaded(&mut bpf, dir)?; + for pin in live.iter().filter(|pin| pin.dir == dir) { + let path = self.link_pin_path(pin); + if let Err(e) = tc::swap_pinned_link(prog, &path) { + log::warn!( + "{}: {} swap failed ({e:#}), reclaiming", + self.name, + pin.filename() + ); + self.reclaim_link(pin); + } + } + } + + // the links of another schema are rebound the same way and their pins moved over, the + // interface never runs without a program and keeps its place in the tcx order. One this + // build has no program for, or that already has a link here, is detached instead. A + // rebound link whose pin cannot be moved stays where it is, the next run finds it again + for links_dir in old_links { + for pin in tc::read_pinned_links(links_dir)? { + let old = links_dir.join(pin.filename()); + let new = self.link_pin_path(&pin); + let migrated = self.directions.contains(&pin.dir) + && !new.exists() + && match tc::swap_pinned_link(self.loaded(&mut bpf, pin.dir)?, &old) { + Ok(()) => true, + Err(e) => { + log::warn!( + "{}: {} swap failed ({e:#}), reclaiming", + self.name, + pin.filename() + ); + false + } + }; + if migrated { + if let Err(e) = std::fs::rename(&old, &new) { + log::warn!( + "{}: move link pin {} to {}: {e:#}", + self.name, + old.display(), + new.display() + ); + } + } else if let Err(e) = tc::detach_pinned_link(&old) { + log::warn!("{}: detach old link {}: {e:#}", self.name, old.display()); + let _ = std::fs::remove_file(&old); + } + } + } + + // a run that died after pinning some of the directions left a partial set behind + match std::fs::remove_dir_all(self.prog_dir()) { + Ok(()) => {} + Err(e) if e.kind() == ErrorKind::NotFound => {} + Err(e) => return Err(e).context("remove partial program pins"), + } + std::fs::create_dir_all(self.prog_dir())?; + for &dir in self.directions { + self.loaded(&mut bpf, dir)?.pin(self.prog_pin_path(dir))?; + } + + // the links are on this build now, the other builds' programs go with their pins + let mine = self.fingerprint_dir_name(); + for entry in std::fs::read_dir(self.progs_dir())? { + let entry = entry?; + if entry.file_name().to_string_lossy() != mine.as_str() + && let Err(e) = std::fs::remove_dir_all(entry.path()) + { + log::warn!( + "{}: remove stale program pins {}: {e:#}", + self.name, + entry.path().display() + ); + } + } + Ok(()) + } + + // one direction's program out of the loaded object, resolved once by the load loop + fn loaded<'b>( + &self, + bpf: &'b mut aya::Ebpf, + dir: Direction, + ) -> anyhow::Result<&'b mut SchedClassifier> { + let prog = bpf + .program_mut((self.prog_name)(dir)) + .expect("every program was resolved by the load loop") + .try_into()?; + Ok(prog) + } + + fn live_links(&self) -> anyhow::Result> { + tc::read_pinned_links(&self.links_dir()) + } + + // a pin this build cannot read or has no program for is not its state. It would keep a link + // running with nothing to update it, so it goes at install time + + fn remove_foreign_pins(&self) -> anyhow::Result<()> { + for entry in std::fs::read_dir(self.links_dir())? { + let entry = entry?; + let name = entry.file_name().to_string_lossy().into_owned(); + let ours = + tc::LinkPin::parse(&name).is_some_and(|pin| self.directions.contains(&pin.dir)); + if ours { + continue; + } + log::warn!("{}: removing pin {name} this build cannot use", self.name); + let path = entry.path(); + if let Err(e) = tc::detach_pinned_link(&path) { + log::warn!("{}: detach {name}: {e:#}", self.name); + let _ = std::fs::remove_file(&path); + } + } + Ok(()) + } + + // every link pinned under the interface name goes, whatever index it had, unless `keep` + // answers for it. An interface that was never attached is done. Returns the indexes the + // detached links had + + fn detach_named( + &self, + name: &str, + keep: impl Fn(&tc::LinkPin) -> bool, + ) -> anyhow::Result> { + let mut ifindexes = Vec::new(); + for pin in self.live_links()? { + if pin.name != name || keep(&pin) { + continue; + } + self.reclaim_link(&pin); + if !ifindexes.contains(&pin.ifindex) { + ifindexes.push(pin.ifindex); + } + } + Ok(ifindexes) + } + + /// Unpin a defunct link. The pin file goes even when the kernel object behind it cannot be + /// opened any more. + fn reclaim_link(&self, pin: &tc::LinkPin) { + let path = self.link_pin_path(pin); + match tc::detach_pinned_link(&path) { + Ok(()) => {} + // the other side of a lost race, another attach or unplug took the pin first + Err(e) if tc::pin_missing(&e) => { + log::debug!( + "{}: reclaim {}: pin gone already", + self.name, + pin.filename() + ); + } + Err(e) => { + log::warn!( + "{}: reclaim {}: unpin stale link: {e:#}", + self.name, + pin.filename() + ); + let _ = std::fs::remove_file(&path); + } + } + } + + fn reconcile(&self, desired: &HashSet) -> anyhow::Result<()> { + // opened before anything is detached, a missing program fails the pass whole + let mut progs = Vec::new(); + for &dir in self.directions { + progs.push((dir, self.program(dir)?)); + } + let live = self.live_links()?; + let mut attached: HashSet<(u32, Direction)> = HashSet::new(); + + for pin in &live { + // a pin whose index is not this interface's anymore is not a link of anything + // wanted, whatever its index says + + let alive = tc::iface_name(pin.ifindex)?.is_some_and(|name| name == pin.name); + if alive && desired.contains(&pin.ifindex) { + attached.insert((pin.ifindex, pin.dir)); + continue; + } + log::debug!("{}: detach {}", self.name, pin.filename()); + let path = self.link_pin_path(pin); + if let Err(e) = tc::detach_pinned_link(&path) { + log::error!("{}: detach {}: {e:#}", self.name, pin.filename()); + let _ = std::fs::remove_file(&path); + } else { + log::info!("{}: detached {}", self.name, pin.filename()); + } + } + + let mut failed = 0usize; + for (dir, prog) in &mut progs { + let dir = *dir; + for &ifindex in desired { + if attached.contains(&(ifindex, dir)) { + continue; + } + let Some(name) = tc::iface_name(ifindex)? else { + log::info!("{}: {ifindex} vanished before the attach", self.name); + continue; + }; + let pin = tc::LinkPin { ifindex, name, dir }; + match tc::attach_and_pin(prog, &pin, &self.link_pin_path(&pin)) { + Ok(()) => {} + Err(e) if e.downcast_ref::().is_some() => { + log::info!("{}: {ifindex} vanished before the attach", self.name); + } + Err(e) => { + log::error!("{}: attach {}: {e:#}", self.name, pin.filename()); + failed += 1; + } + } + } + } + if failed > 0 { + anyhow::bail!("{}: {failed} link(s) failed to attach", self.name); + } + Ok(()) + } + + fn attach_iface(&self, ifindex: u32) -> anyhow::Result<()> { + // opened for every direction ahead of the pin checks, so a missing program fails here, + // never counts as a dead link and leaves no direction attached alone + let mut progs = Vec::new(); + for &dir in self.directions { + progs.push((dir, self.program(dir)?)); + } + let Some(name) = tc::iface_name(ifindex)? else { + log::info!("{}: {ifindex} vanished before the attach", self.name); + return Ok(()); + }; + let live = self.live_links()?; + for (dir, prog) in &mut progs { + let dir = *dir; + let pin = tc::LinkPin { + ifindex, + name: name.clone(), + dir, + }; + let path = self.link_pin_path(&pin); + for stale in live.iter().filter(|p| p.ifindex == ifindex && p.dir == dir) { + if stale.name == name { + match tc::swap_pinned_link(prog, &path) { + Ok(()) => {} + // the pin went away under the swap, another plug of the same interface + // is rebuilding it and the attach below finds its link + Err(e) if tc::pin_missing(&e) => { + log::debug!( + "{}: {} pin taken by another attach", + self.name, + path.display() + ); + continue; + } + Err(e) => { + log::warn!( + "{}: {} swap failed ({e:#}), rebuilding", + self.name, + pin.filename() + ); + self.reclaim_link(stale); + continue; + } + } + } else { + // the index was reused since, the old link is not this interface's + self.reclaim_link(stale); + } + } + if path.exists() { + continue; + } + match tc::attach_and_pin(prog, &pin, &path) { + Ok(()) => {} + Err(e) if e.downcast_ref::().is_some() => { + log::info!("{}: {ifindex} vanished before the attach", self.name); + } + Err(e) if e.downcast_ref::().is_some() => { + log::debug!("{}: {} attached concurrently", self.name, pin.filename()); + } + Err(e) => { + return Err(e) + .with_context(|| format!("{}: attach {}", self.name, pin.filename())); + } + } + } + Ok(()) + } + + // the generation lock is held for the stamp check and write only, the pass itself runs + // under the exclusive apply lock, which already keeps every other writer out + fn with_generation( + &self, + generation: u64, + f: impl FnOnce() -> anyhow::Result, + ) -> anyhow::Result> { + { + let _lock = self.generation_lock()?; + if let Some(last) = read_stamp(&self.applied_path())? + && last >= generation + { + log::info!( + "{}: full apply of generation {generation} is older than the applied \ + {last}, dropping it", + self.name + ); + return Ok(None); + } + write_stamp(&self.applied_path(), generation)?; + } + Ok(Some(f()?)) + } + + fn change(&self, f: impl FnOnce(Option) -> anyhow::Result) -> anyhow::Result { + let _lock = self.generation_lock()?; + f(read_stamp(&self.applied_path())?) + } +} + +// whether link pins remain in a links directory, only those count, debris of another kind goes +// with the directory. One that cannot be read may hold anything, a missing one is empty +fn any_links(links_dir: &Path) -> bool { + match tc::read_pinned_links(links_dir) { + Ok(links) => !links.is_empty(), + Err(_) => true, + } +} + +// a program pin is a file under a fingerprint directory, an empty fingerprint directory is +// what a run killed between creating it and pinning left behind. Like the links, a directory +// that cannot be read may hold anything, a missing one is empty +fn any_program_pinned(progs_dir: &Path) -> bool { + let non_empty = |dir: &Path| match std::fs::read_dir(dir) { + Ok(mut entries) => entries.next().is_some(), + Err(e) => e.kind() != ErrorKind::NotFound, + }; + match std::fs::read_dir(progs_dir) { + Ok(fingerprints) => fingerprints.into_iter().any(|fp| match fp { + Ok(fp) => non_empty(&fp.path()), + Err(_) => true, + }), + Err(e) => e.kind() != ErrorKind::NotFound, + } +} + +fn read_stamp(path: &Path) -> anyhow::Result> { + match std::fs::read_to_string(path) { + Ok(s) => s + .trim() + .parse() + .map(Some) + .with_context(|| format!("stamp {} holds no number", path.display())), + Err(e) if e.kind() == ErrorKind::NotFound => Ok(None), + Err(e) => Err(e.into()), + } +} + +// written beside and renamed over, so a stamp is either the old one or the new one +fn write_stamp(path: &Path, stamp: u64) -> anyhow::Result<()> { + let tmp = path.with_extension("tmp"); + std::fs::write(&tmp, stamp.to_string())?; + Ok(std::fs::rename(&tmp, path)?) +} + +/// Marker for a guard holding the apply lock shared. +pub struct Shared; + +/// Marker for a guard holding the apply lock exclusively. +pub struct Exclusive; + +/// The apply lock of one subsystem, held for as long as the guard lives. Loading and the +/// reconcile of links rewrite pinned state and exist only on the [`Exclusive`] guard, so nothing +/// runs them next to shared holders. A shared holder adds or drops its own link and writes its +/// own map entries. There is no upgrade. A shared holder that finds nothing current hands off, +/// the full pass takes the exclusive lock. +pub struct ApplyLock<'a, M> { + programs: &'a TcPrograms, + _lock: Flock, + _mode: PhantomData, +} + +impl ApplyLock<'_, M> { + /// Whether the programs and maps of this build are pinned. What a shared holder checks before + /// it writes, since it cannot install. Under the lock the program and map pins do not change, + /// only links come and go, so the answer holds for the holder's steps. + pub fn is_current(&self) -> bool { + self.programs.is_current() + } + + /// Open a pinned BPF hash map by name, for the owning subsystem to sync. Valid while the + /// programs are pinned. An exclusive holder runs [`ensure_loaded`](ApplyLock::ensure_loaded) + /// first, a shared one checks [`is_current`](ApplyLock::is_current) and hands off when it is + /// not. + pub fn hash_map( + &self, + name: &str, + ) -> anyhow::Result> { + self.programs.hash_map(name) + } + + /// Attach the programs to a single interface, additively, without touching others. + /// + /// If a pin already exists, swap the program in place with no traffic gap. The swap only + /// succeeds while the link is still on a live netdev, a defunct pin is reclaimed and the + /// interface attached fresh. Two concurrent attaches of the same interface run under the + /// shared lock, the one that loses the pin race finds the other's link in place and is done. + /// A failed attach propagates so the caller can decide what an interface without the program + /// means for it, [`NotLoaded`] among them when nothing is pinned yet. + pub fn attach_iface(&self, ifindex: u32) -> anyhow::Result<()> { + self.programs.attach_iface(ifindex) + } + + /// Detach the programs from the interface of that name. The pin goes even when the interface + /// is gone already, which it is for an unplug reported after the fact. A pin `keep` answers + /// true for stays, for a later incarnation of the name that is not the one the unplug is + /// about. Returns the indexes the detached links had, for state keyed by them. + pub fn detach_named( + &self, + name: &str, + keep: impl Fn(&tc::LinkPin) -> bool, + ) -> anyhow::Result> { + self.programs.detach_named(name, keep) + } + + /// The links of this subsystem as their pins name them. A shared holder sees a snapshot, + /// another plug may add to it meanwhile. The name is not unique. The leftover pin of a + /// crashed guest and its restarted interface share it, only the index tells them apart. + pub fn attached(&self) -> anyhow::Result> { + self.programs.live_links() + } + + /// Run `f` as a single-entry change under the lock every such change takes. Its read of the + /// entry, the compare of generations and its write are then one step against another change + /// of the same entry. `f` gets the generation of the last full apply. A change older than + /// that was in the input the apply took, whether the apply went through or not. The + /// subsystem decides what such a change may still write. Keep `f` to the map accesses of one + /// change. + pub fn change(&self, f: impl FnOnce(Option) -> anyhow::Result) -> anyhow::Result { + self.programs.change(f) + } +} + +impl ApplyLock<'_, Exclusive> { + /// Run `f` as a full apply if `generation` is newer than the last one. The generation is + /// recorded before `f` runs, so a state compiled from older input than one already in place + /// is dropped and `None` returned. What single-entry changes wrote since `f` read its input is + /// `f`'s to leave alone, by the generation each entry carries. The record marks the input as + /// taken, not the state as complete. A change older than it is dropped even when `f` failed, + /// its input was this apply's and the next apply carries it. A failed apply is reported and + /// healed by the next one under a new generation. A retry under the same one is dropped as + /// applied. + pub fn with_generation( + &self, + generation: u64, + f: impl FnOnce() -> anyhow::Result, + ) -> anyhow::Result> { + self.programs.with_generation(generation, f) + } + + /// Make sure the programs are loaded and pinned. The load runs when nothing current is + /// pinned, on a schema change (a rebuild), on a lost map pin, or on an object change (a + /// refresh). Otherwise what an earlier run pinned is reused untouched. A rebuild starts with + /// empty maps. A refresh keeps every map with its entries. A lost map pin brings only that + /// map back empty. The caller syncs all of them in the same pass. + pub fn ensure_loaded(&self) -> anyhow::Result<()> { + self.programs.ensure_loaded() + } + + /// Make the attached set match `desired`. Detach what is no longer wanted, attach what is + /// missing. Refreshing links onto new code is [`ensure_loaded`](Self::ensure_loaded)'s job. + /// The live set cannot change under it, every other writer waits on the exclusive lock. Links + /// carry no generation, what may go is decided from the kernel's state. + /// + /// An interface that failed to attach fails the apply once every other was tried, one left + /// without its program is the subsystem not doing its job there. One that vanished since it + /// was named is no failure, and a failed detach only leaves a program running longer and is + /// logged. + pub fn reconcile(&self, desired: &HashSet) -> anyhow::Result<()> { + self.programs.reconcile(desired) + } + + /// Drop the pinned state of this subsystem, links included. Leaving links attached would keep + /// the programs running with nothing left to update them, so retiring the subsystem on a node + /// ends here. The lock and the generations under `/run` stay. Unlinking the lock under a waiter + /// would let that waiter run unserialized, and the generations keep ordering the changes that + /// follow. + pub fn clear(&self) -> anyhow::Result<()> { + self.programs.tear_down() + } +} diff --git a/src/tc.rs b/src/tc.rs new file mode 100644 index 0000000..71c0ac7 --- /dev/null +++ b/src/tc.rs @@ -0,0 +1,257 @@ +//! TC link plumbing shared by all subsystems. A `Direction` enum, attach/swap/detach free +//! functions over tcx links (kernel 6.6 and newer), and a uniform pin-filename layout, see +//! [`LinkPin`]. + +use std::{io::ErrorKind, path::Path, str::FromStr}; + +use anyhow::Context; +use aya::{ + pin::PinError, + programs::{ + SchedClassifier, TcAttachType, + links::{FdLink, LinkError, LinkOrder, PinnedLink}, + tc::TcAttachOptions, + }, +}; + +/// An attach that found its pin already present, so a caller racing another attach of the same +/// interface can tell that outcome from a real failure. +#[derive(Debug)] +pub struct PinExists; + +impl std::fmt::Display for PinExists { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("pin path already exists") + } +} + +impl std::error::Error for PinExists {} + +/// An attach whose interface is gone, the index names no netdev any more. +#[derive(Debug)] +pub struct IfaceGone; + +impl std::fmt::Display for IfaceGone { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("interface is gone") + } +} + +impl std::error::Error for IfaceGone {} + +#[derive(Copy, Clone, Hash, PartialEq, Eq, Debug)] +pub enum Direction { + Ingress, + Egress, +} + +impl Direction { + pub fn as_str(self) -> &'static str { + match self { + Self::Ingress => "ingress", + Self::Egress => "egress", + } + } + pub fn aya_type(self) -> TcAttachType { + match self { + Self::Ingress => TcAttachType::Ingress, + Self::Egress => TcAttachType::Egress, + } + } +} + +impl FromStr for Direction { + type Err = (); + fn from_str(s: &str) -> Result { + match s { + "ingress" => Ok(Self::Ingress), + "egress" => Ok(Self::Egress), + _ => Err(()), + } + } +} + +/// One pinned link as its pin file names it. The interface name is part of the file name, so an +/// unplug reported after the interface is gone still finds the pin and a reused index does not +/// pass for the old link. The colon separates, the kernel forbids it in interface names. It also +/// stands in for the dot inside the name, which bpffs refuses in a file name. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct LinkPin { + pub ifindex: u32, + pub name: String, + pub dir: Direction, +} + +impl LinkPin { + pub fn filename(&self) -> String { + format!( + "{}:{}:{}", + self.ifindex, + self.dir.as_str(), + self.name.replace('.', ":") + ) + } + + /// The pin a file name stands for, none when it is not one of ours. + pub fn parse(filename: &str) -> Option { + let mut parts = filename.splitn(3, ':'); + let ifindex = parts.next()?.parse().ok()?; + let dir = parts.next()?.parse().ok()?; + let name = parts.next()?.replace(':', "."); + let pin = LinkPin { ifindex, name, dir }; + // a name that parses but is not the one written for that link is not ours, and no + // interface carries what the kernel refuses in a name + let valid_name = !pin.name.is_empty() + && pin.name.len() < 16 + && pin.name != "." + && pin.name != ".." + && !pin.name.chars().any(|c| c.is_whitespace() || c == '/'); + (pin.filename() == filename && valid_name).then_some(pin) + } +} + +/// Attach `prog` in `dir` as a tcx link and pin it. tcx (kernel 6.6 and newer) runs its +/// programs on the netdev hook ahead of the interface's qdisc filters, among them the `ingress` +/// qdisc the rate limiter installs. `TC_ACT_UNSPEC` hands a packet on to the next program and +/// then to the filters. Any other verdict ends the chain right there. So a subsystem's program +/// returns `TC_ACT_UNSPEC` for whatever it leaves alone. Programs of several subsystems run in +/// attach order, each new link goes last. +pub fn attach_and_pin( + prog: &mut SchedClassifier, + pin: &LinkPin, + pin_path: &Path, +) -> anyhow::Result<()> { + let link_id = prog + .attach_with_options( + &pin.name, + pin.dir.aya_type(), + TcAttachOptions::TcxOrder(LinkOrder::last()), + ) + .map_err(|e| { + // the interface can go away between the name lookup and the attach + if matches!(iface_name(pin.ifindex), Ok(None)) { + anyhow::Error::from(IfaceGone) + } else { + e.into() + } + })?; + let link = prog.take_link(link_id)?; + let fd_link: FdLink = link.try_into()?; + // a lost race against another attach of the same interface shows up here, the unpinned link + // detaches again when its fd drops + match fd_link.pin(pin_path) { + Ok(_) => Ok(()), + Err(PinError::SyscallError(e)) if e.io_error.kind() == ErrorKind::AlreadyExists => { + Err(PinExists.into()) + } + Err(e) => Err(e.into()), + } +} + +/// Whether a swap or an unpin failed because the pin was gone by the time it was opened or +/// removed. +pub fn pin_missing(e: &anyhow::Error) -> bool { + matches!( + e.downcast_ref::(), + Some(LinkError::SyscallError(s)) if s.io_error.kind() == ErrorKind::NotFound + ) || e + .downcast_ref::() + .is_some_and(|e| e.kind() == ErrorKind::NotFound) +} + +/// Rebind a pinned link to `prog` via LINK_UPDATE. Atomic, traffic sees no detach/reattach gap. +pub fn swap_pinned_link(prog: &mut SchedClassifier, pin_path: &Path) -> anyhow::Result<()> { + let pinned = PinnedLink::from_pin(pin_path)?; + let fd_link: FdLink = pinned.into(); + let link = fd_link.try_into()?; + let new_id = prog.attach_to_link(link)?; + // take the handle out of aya's internal tracking, we have the pin + let _ = prog.take_link(new_id)?; + Ok(()) +} + +pub fn detach_pinned_link(pin_path: &Path) -> anyhow::Result<()> { + let pinned = PinnedLink::from_pin(pin_path)?; + let _fd_link = pinned.unpin()?; + Ok(()) +} + +/// The name of an interface by index, none when it is gone. Any other failure says nothing +/// about the interface and is not taken as its absence, so the lookup reads errno itself. +pub fn iface_name(ifindex: u32) -> anyhow::Result> { + let mut buf = [0u8; nix::libc::IF_NAMESIZE]; + // SAFETY: the buffer holds IF_NAMESIZE bytes, the most the call writes + let name = unsafe { nix::libc::if_indextoname(ifindex, buf.as_mut_ptr().cast()) }; + if name.is_null() { + return match nix::errno::Errno::last() { + nix::errno::Errno::ENXIO => Ok(None), + e => Err(e).with_context(|| format!("name of interface {ifindex}")), + }; + } + let name = std::ffi::CStr::from_bytes_until_nul(&buf)?; + Ok(Some(name.to_string_lossy().into_owned())) +} + +/// Read and parse every pin file in `links_dir`, see [`LinkPin`]. Unrecognized names are skipped +/// with a warning. +pub fn read_pinned_links(links_dir: &Path) -> anyhow::Result> { + let mut out = Vec::new(); + let dir = match std::fs::read_dir(links_dir) { + Ok(d) => d, + Err(e) if matches!(e.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => { + return Ok(out); + } + Err(e) => return Err(e.into()), + }; + for entry in dir { + let entry = entry?; + let name = entry.file_name(); + let name = name.to_string_lossy(); + match LinkPin::parse(&name) { + Some(pin) => out.push(pin), + None => log::warn!("unrecognized pin file {name} in {}", links_dir.display()), + } + } + Ok(out) +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn pin_names_round_trip() { + for name in ["tap100i0", "veth100i31", "vmbr0.100", "bond0.5.7"] { + let pin = LinkPin { + ifindex: 42, + name: name.into(), + dir: Direction::Ingress, + }; + assert!(!pin.filename().contains('.'), "{}", pin.filename()); + assert_eq!(LinkPin::parse(&pin.filename()), Some(pin)); + } + assert_eq!( + LinkPin::parse("7:egress:eno1").map(|p| p.name), + Some("eno1".into()) + ); + } + + #[test] + fn pin_names_are_canonical() { + for bad in [ + "042:ingress:tap0", + "+42:ingress:tap0", + "42:ingress:", + "42:ingress", + "42:Ingress:tap0", + "42-ingress", + "4294967296:ingress:tap0", + "42:ingress:tap 0", + "42:ingress::", + "42:ingress:::", + "42:ingress:sixteencharsname", + ] { + assert_eq!(LinkPin::parse(bad), None, "{bad}"); + } + } +} -- 2.47.3