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 5EDAA1FF09C for ; Mon, 19 Oct 2026 16:24:51 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id B930C2156D; Mon, 19 Oct 2026 16:24:38 +0200 (CEST) From: Hannes Laimer To: pve-devel@lists.proxmox.com Subject: [PATCH proxmox-ebpf v4 2/5] subsystem: add the shared tc program handling Date: Mon, 19 Oct 2026 16:24:09 +0200 Message-ID: <20261019142413.2255644-3-h.laimer@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20261019142413.2255644-1-h.laimer@proxmox.com> References: <20261019142413.2255644-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: 1792419861451 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.487 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: EP4LOPSUMV7BFUMACO7SVDM4LT7SQ6ZM X-Message-ID-Hash: EP4LOPSUMV7BFUMACO7SVDM4LT7SQ6ZM 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 of its consumer, and re-running never interrupts traffic. Changes run under one lock per subsystem, taken before the change reads its input, so they reach the kernel in the order they read it. A subsystem describes its programs with a macro, from its name, the program names, the maps and the schema version. The object is found by the subsystem's name, and its fingerprint is derived from its bytes. Signed-off-by: Hannes Laimer --- src/lib.rs | 1 + src/subsystem.rs | 829 +++++++++++++++++++++++++++++++++++++++++++++++ src/tc.rs | 12 +- 3 files changed, 836 insertions(+), 6 deletions(-) create mode 100644 src/subsystem.rs diff --git a/src/lib.rs b/src/lib.rs index 8d4420a..572d861 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +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..c740fa8 --- /dev/null +++ b/src/subsystem.rs @@ -0,0 +1,829 @@ +//! 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 +//! only when what is pinned is not this build's, or links still sit under another schema. Every +//! other step works on what is already pinned. Each subsystem owns one [`TcPrograms`]. +//! +//! Every change runs under the subsystem's apply lock, a full apply exclusively and a +//! single-interface step shared. The holder reads the input of its change under the lock and +//! writes before it lets go, so a full apply and every step reach the kernel in the order they +//! read their input. Two steps of one interface do not order each other. +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"; + +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 +} + +/// A subsystem's [`TcPrograms`] from its name. The object is the build's `.bpf.o`, the +/// fingerprint follows from the bytes. +#[macro_export] +macro_rules! tc_programs { + ( + name: $name:literal, + programs: $programs:expr, + maps: $maps:expr, + schema_version: $schema:expr $(,)? + ) => { + // built at compile time wherever the macro is used, a let binding included + const { + const OBJ: &[u8] = + ::aya::include_bytes_aligned!(concat!(env!("OUT_DIR"), "/", $name, ".bpf.o")); + $crate::subsystem::TcPrograms::new($name, OBJ, $programs, $maps, $schema) + } + }; +} + +const LOCK_WAIT: Duration = Duration::from_secs(10); +const LOCK_POLL: Duration = Duration::from_millis(10); + +// flock has no timeout, so the wait retries a nonblocking flock with a sleep in between. +// +// not proxmox-sys lock because: blocks and gives up on a timer signal, and under pve-rs any signal +// perl handles cuts the blocked wait short the same way, since perl installs its handlers without +// restart +fn flock_within( + path: &Path, + exclusive: bool, + what: &str, + subsystem: &str, +) -> anyhow::Result> { + let mut file = File::options() + .create(true) + .truncate(false) + .write(true) + .open(path)?; + let arg = if exclusive { + FlockArg::LockExclusiveNonblock + } else { + FlockArg::LockSharedNonblock + }; + let deadline = Instant::now() + LOCK_WAIT; + loop { + match Flock::lock(file, arg) { + 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. +#[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 {} + +/// A subsystem's description of its tc programs, written with the `tc_programs!` macro. The +/// loaded programs live pinned in the kernel and are reached through those pins. +pub struct TcPrograms { + /// The subsystem's name, its directory under the pin root. + name: &'static str, + /// The embedded BPF object. + obj: &'static [u8], + /// FNV-1a of `obj`, the programs are pinned under it. + fingerprint: u64, + /// The directions the object carries a program for, each with that program's name in the + /// object. One entry per direction. + programs: &'static [(Direction, &'static str)], + /// The object's pinned maps. A missing one means the state is not current. + maps: &'static [&'static str], + /// Bumped with any change to a map's key or value layout. A pinned map is reused by name + /// without any check, only the Rust accessor checks its key and value sizes. + schema_version: u32, +} + +impl TcPrograms { + pub const fn new( + name: &'static str, + obj: &'static [u8], + programs: &'static [(Direction, &'static str)], + maps: &'static [&'static str], + schema_version: u32, + ) -> Self { + Self { + name, + obj, + fingerprint: obj_fingerprint(obj), + programs, + maps, + schema_version, + } + } + + 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 { + schema_links_dir(&self.schema_root()) + } + fn link_pin_path(&self, pin: &tc::LinkPin) -> PathBuf { + self.links_dir().join(pin.filename()) + } + fn progs_dir(&self) -> PathBuf { + schema_progs_dir(&self.schema_root()) + } + 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 program_name(&self, dir: Direction) -> Option<&'static str> { + self.programs + .iter() + .find(|&&(d, _)| d == dir) + .map(|&(_, name)| name) + } + fn run_dir(&self) -> PathBuf { + PathBuf::from(RUN_ROOT).join(self.name) + } + 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, [`NotLoaded`] when it is not pinned. + 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())) + } + + /// The apply lock under `/run`. A full apply, an install and a teardown take it exclusively, + /// a single-interface step shared. A wait times out instead of hanging. Shared holders do not + /// yield to a waiting exclusive one, so a node attaching interfaces without pause can keep a + /// full apply waiting until it times out. Every write to the pinned state is + /// a method of the returned guard, so a step cannot run outside the lock. + fn lock(&self, exclusive: bool) -> anyhow::Result> { + let dir = self.run_dir(); + std::fs::create_dir_all(&dir)?; + let lock = flock_within(&dir.join("lock"), exclusive, "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 full apply. 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| links_pinned_under(&others)); + elsewhere + || self + .live_links() + .map_or(true, |links| links.iter().any(|pin| pin.name == name)) + } + + /// Take the apply lock shared, for a single-interface step. + pub fn lock_shared(&self) -> anyhow::Result> { + self.lock(false) + } + + /// Take the apply lock exclusively, for a full apply, an install or a teardown. + pub fn lock_exclusive(&self) -> anyhow::Result> { + self.lock(true) + } + + 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 some of the + // links there. This build is not current until the install below has taken them over + let links_elsewhere = links_pinned_under(&others); + // programs pinned under a newer schema mean a newer build runs here and this process is + // the stale one, so it leaves the state alone. A newer schema directory without programs + // is what a rejected install of that build left behind + if let Some(newer) = self.newer_schema_installed(&others) { + 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 here is left over from a run that died between its install and the + // removal, its links moved over already + for (dir, schema) in others { + if (schema.is_some_and(|s| s < self.schema_version) + || !any_program_pinned(&schema_progs_dir(&dir))) + && 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 make that half-done state pass 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. + // It stays in place through the install and goes afterward + 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. Without program pins the directory does not pass for state of that build + let old_links: Vec = others + .iter() + .map(|(dir, _)| schema_links_dir(dir)) + .collect(); + self.install(&old_links).context("install")?; + for (dir, _) in &others { + // a link whose pin could not be moved is still live there + if any_links(&schema_links_dir(dir)) { + continue; + } + if let Err(e) = std::fs::remove_dir_all(dir) { + log::warn!( + "{}: remove old pinned state {}: {e:#}", + self.name, + dir.display() + ); + } + } + Ok(()) + } + + fn is_current(&self) -> bool { + self.programs + .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 directories under the pin root other than this build's schema. One whose name is not + // a version counts as well, so its links are moved over too + 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 newer_schema_installed(&self, others: &[(PathBuf, Option)]) -> Option { + others + .iter() + .filter(|(dir, _)| any_program_pinned(&schema_progs_dir(dir))) + .filter_map(|(_, schema)| *schema) + .find(|&s| s > self.schema_version) + } + + 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 this build's object and move every link onto its programs, this schema's and those + /// under `old_links`, then pin the programs and drop those of other builds. The verifier runs + /// here and nowhere else, so a rejection leaves what served before untouched. The program + /// pins mark this build as current and go last, so after a failure the previous build stays + /// current and the next run installs again. A link that cannot be swapped is unpinned, and + /// the next full apply attaches it again if it is still wanted. + fn install(&self, old_links: &[PathBuf]) -> anyhow::Result<()> { + std::fs::create_dir_all(self.links_dir())?; + 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 &(_, name) in self.programs { + let prog: &mut SchedClassifier = bpf + .program_mut(name) + .with_context(|| format!("{}: program {name} not found in object", self.name))? + .try_into()?; + prog.load()?; + } + self.remove_foreign_pins()?; + + let live = self.live_links()?; + for &(dir, _) in self.programs { + 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, so + // their interfaces never run without a program and keep their 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 + 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.program_name(pin.dir).is_some() + && !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.programs { + 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() != 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 name = self + .program_name(dir) + .expect("only directions with a program are loaded"); + let prog = bpf + .program_mut(name) + .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(); + let ours = name + .to_str() + .and_then(tc::LinkPin::parse) + .is_some_and(|pin| self.program_name(pin.dir).is_some()); + if ours { + continue; + } + let name = name.to_string_lossy(); + 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(()) + } + + 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 link. The pin file goes even when the kernel object behind it cannot be opened + /// anymore. + 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(tc::Error::PinGone) => { + 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, so a missing program fails the full apply before it + // changes anything + let mut progs = Vec::new(); + for &(dir, _) in self.programs { + 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 now belongs to another interface links nothing wanted + 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(tc::Error::IfaceGone) => { + 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.programs { + 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(tc::Error::PinGone) => { + 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(tc::Error::IfaceGone) => { + log::info!("{}: {ifindex} vanished before the attach", self.name); + } + Err(tc::Error::AlreadyAttached) => { + log::debug!("{}: {} attached concurrently", self.name, pin.filename()); + } + Err(e) => { + return Err(e) + .with_context(|| format!("{}: attach {}", self.name, pin.filename())); + } + } + } + Ok(()) + } +} + +fn schema_links_dir(schema_root: &Path) -> PathBuf { + schema_root.join("links") +} + +fn schema_progs_dir(schema_root: &Path) -> PathBuf { + schema_root.join("prog") +} + +fn links_pinned_under(schemas: &[(PathBuf, Option)]) -> bool { + schemas + .iter() + .any(|(dir, _)| any_links(&schema_links_dir(dir))) +} + +// whether link pins remain in a links directory. Debris of another kind does not count, it 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. A run killed between creating that +// directory and pinning leaves it empty. 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, + } +} + +/// 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 matching +/// the links to the desired set 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 map entries that stand on their own, each write atomic in the kernel. A +/// write that depends on what was read before it takes the exclusive lock, or the kernel's +/// conditional write, insert only if absent and update only if present. A shared holder cannot +/// upgrade, one that finds nothing current leaves the install to a full apply under 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. Valid while the programs are pinned, an exclusive + /// holder runs [`ensure_loaded`](ApplyLock::ensure_loaded) first. + pub fn hash_map( + &self, + name: &str, + ) -> anyhow::Result> { + self.programs.hash_map(name) + } + + /// Attach the programs to a single interface, leaving the others alone. + /// + /// If a pin already exists, swap the program in place with no traffic gap. A pin that cannot + /// be swapped is reclaimed and the interface attached fresh. Two concurrent attaches of the + /// same interface run under the shared lock, the one that loses the race finds the other's + /// link in place and is done. A failed attach returns the error, [`NotLoaded`] when nothing + /// is pinned yet, and the caller decides what an interface without the program means for it. + pub fn attach_iface(&self, ifindex: u32) -> anyhow::Result<()> { + self.programs.attach_iface(ifindex) + } + + /// Detach the programs from the interface of that name, even when the interface is gone + /// already. A pin for which `keep` returns true stays, such as one of a newer interface under + /// the same name. Returns the indexes the detached links had. + 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 attach may add to it meanwhile. The name is not unique. A leftover pin of an + /// interface that went away and the pin of its successor under the name share it, only the + /// index tells them apart. + pub fn attached(&self) -> anyhow::Result> { + self.programs.live_links() + } +} + +impl ApplyLock<'_, Exclusive> { + /// Make sure this build's programs and maps are pinned, loading only when they are not. Maps + /// keep their entries across a load unless the schema changed, then they start empty, and a + /// lost map pin comes back empty. The caller syncs the maps in the same apply. + pub fn ensure_loaded(&self) -> anyhow::Result<()> { + self.programs.ensure_loaded() + } + + /// Make the attached set match `desired`. 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. What may go is decided from the kernel's state. + /// + /// An interface that failed to attach fails the apply once every other was tried, since the + /// subsystem does not work on an interface without its program. 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. The lock under `/run` stays, + /// unlinking it under a waiter would let that waiter run unserialized. + pub fn clear(&self) -> anyhow::Result<()> { + self.programs.tear_down() + } +} diff --git a/src/tc.rs b/src/tc.rs index 05d823e..63d3fbf 100644 --- a/src/tc.rs +++ b/src/tc.rs @@ -17,7 +17,7 @@ use aya::{ /// Why a link step did not go through, with the kernel's own answer where it gives one, so a /// caller can tell a lost race or a vanished interface from a real failure. #[derive(Debug)] -pub enum Error { +pub(crate) enum Error { /// A link of the program is in place already, another attach of the interface won the race. AlreadyAttached, /// The interface is gone, its index names no netdev anymore. @@ -176,7 +176,7 @@ impl LinkPin { /// 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( +pub(crate) fn attach_and_pin( prog: &mut SchedClassifier, pin: &LinkPin, pin_path: &Path, @@ -201,7 +201,7 @@ pub fn attach_and_pin( } /// 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) -> Result<(), Error> { +pub(crate) fn swap_pinned_link(prog: &mut SchedClassifier, pin_path: &Path) -> Result<(), Error> { let pinned = PinnedLink::from_pin(pin_path)?; let fd_link: FdLink = pinned.into(); let link = fd_link.try_into()?; @@ -211,7 +211,7 @@ pub fn swap_pinned_link(prog: &mut SchedClassifier, pin_path: &Path) -> Result<( Ok(()) } -pub fn detach_pinned_link(pin_path: &Path) -> Result<(), Error> { +pub(crate) fn detach_pinned_link(pin_path: &Path) -> Result<(), Error> { let pinned = PinnedLink::from_pin(pin_path)?; let _fd_link = pinned.unpin()?; Ok(()) @@ -219,7 +219,7 @@ pub fn detach_pinned_link(pin_path: &Path) -> Result<(), Error> { /// 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> { +pub(crate) 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()) }; @@ -237,7 +237,7 @@ pub fn iface_name(ifindex: u32) -> anyhow::Result> { /// The pins in `links_dir`, see [`LinkPin`], none for a missing directory. A file that is not a /// pin is skipped with a warning. -pub fn read_pinned_links(links_dir: &Path) -> anyhow::Result> { +pub(crate) 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, -- 2.47.3