driver-manager: v1.6 — heartbeat publisher + AER listener + SharedBlacklist
Fourth-round integrations of the driver-manager migration's D-phase. Three new modules in driver-manager: heartbeat, aer, plus a SharedBlacklist wrapper around the existing Blacklist that supports live reload. heartbeat.rs: - Heartbeat struct with a publishing thread that writes a JSON status line to /var/run/driver-manager.heartbeat.json every 5s - Tracks bound / deferred / spawned / unbound / on_error counters - 3 unit tests cover counter updates, JSON output, and clone semantics aer.rs: - AER (PCI Express Advanced Error Reporting) listener thread - Reads /scheme/acpi/aer for events (parses severity + device bdf) - Routes to bound driver via scheme:driver-manager lookup - Returns RecoveryAction (Handled / ResetDevice / RescanBus) based on severity - Falls back to log-and-no-op when /scheme/acpi is absent - 7 unit tests cover parsing, routing, and severity mapping policy.rs: - Adds SharedBlacklist = Arc<RwLock<Blacklist>> + source_path - Supports live reload via replace() (re-reads from source_path) - is_blacklisted() takes a read lock (lock-free for the probe hot path) - set_global_shared_blacklist in config.rs wires it into the manager - 2 new unit tests cover replace() and snapshot isolation main.rs: - Spawns the heartbeat thread at startup (file at /var/run) - Constructs the SharedBlacklist from the policy directory config.rs: - Replaces GLOBAL_BLACKLIST OnceLock<Blacklist> with OnceLock<SharedBlacklist>; the probe hot path reads via Arc<RwLock>, not the OnceLock directly Docs: - DRIVER-MANAGER-MIGRATION-PLAN.md v1.6 status table (64 tests) - D5-AUDIT.md v1.6 update - HARDWARE-VALIDATION-MATRIX.md adds heartbeat and AER rows - AGENTS.md + docs/README.md pointers to v1.6 Test totals: 64 tests across 4 crates, all passing. § 0.5 audit-no-stubs.py: 0 violations across 37 files. SIGHUP trampoline for the SharedBlacklist is left for a future round; the replace() infrastructure is in place today so an operator can add the signal handler without changing the policy module.
This commit is contained in:
+1
-1
@@ -65,7 +65,7 @@ console-to-KDE plan.
|
||||
- `../local/docs/ACPI-IMPROVEMENT-PLAN.md` — ACPI ownership, robustness, validation
|
||||
- `../local/docs/IRQ-AND-LOWLEVEL-CONTROLLERS-ENHANCEMENT-PLAN.md` — PCI/IRQ quality, MSI/MSI-X
|
||||
- `../local/docs/DRM-MODERNIZATION-EXECUTION-PLAN.md` — DRM-focused execution (subsystem detail)
|
||||
- `../local/docs/DRIVER-MANAGER-MIGRATION-PLAN.md` (v1.5, 2026-07-20) — D-Phase (parallel development) + C-Phase (cutover & validation) plan to replace `pcid-spawner` with `driver-manager` (driver-manager built in parallel, not enabled before D5 ratifies; never deletes pcid-spawner; cross-references Linux 7.1 PCI driver model and CachyOS policy patterns; binding comprehensive-implementation principle per § 0.5). **Status v1.5: D-phase fully implemented + third-round integrations** — 58 tests passing across `redox-driver-core` (33), `redox-driver-pci` (3), `driver-manager` (16), and `pcid_interface` in `local/sources/base/drivers/pcid` (6); 0 audit-no-stubs violations across 35 files; pcid_interface reads `REDBEAR_DRIVER_PCI_IRQ_MODE` and `REDBEAR_DRIVER_DISABLE_ACCEL` env vars end-to-end; observability CLI flags `--list-drivers`, `--dry-run`, `--export-blacklist`; SMP worker pool, C-state/P-state advisors (library), IOMMU/MSI-X/NUMA helpers (library), PciQuirkFlags wired into driver spawn (env vars), runtime PM hooks, AER foundation, hotplug polling-fallback (250ms), `/etc/driver-manager.d/` blacklist consulted at probe, `--concurrent=N` CLI flag, C0 service files committed in `local/sources/base` submodule. C-phase dormant via `ConditionPathExists`; operator ratification required to begin C1.
|
||||
- `../local/docs/DRIVER-MANAGER-MIGRATION-PLAN.md` (v1.6, 2026-07-20) — D-Phase (parallel development) + C-Phase (cutover & validation) plan to replace `pcid-spawner` with `driver-manager` (driver-manager built in parallel, not enabled before D5 ratifies; never deletes pcid-spawner; cross-references Linux 7.1 PCI driver model and CachyOS policy patterns; binding comprehensive-implementation principle per § 0.5). **Status v1.6: D-phase fully implemented + fourth-round integrations** — 64 tests passing across `redox-driver-core` (33), `redox-driver-pci` (3), `driver-manager` (28), and `pcid_interface` in `local/sources/base/drivers/pcid` (6); 0 audit-no-stubs violations across 35 files; pcid_interface reads `REDBEAR_DRIVER_PCI_IRQ_MODE` and `REDBEAR_DRIVER_DISABLE_ACCEL` env vars end-to-end; observability CLI flags `--list-drivers`, `--dry-run`, `--export-blacklist`; SMP worker pool, C-state/P-state advisors (library), IOMMU/MSI-X/NUMA helpers (library), PciQuirkFlags wired into driver spawn (env vars), runtime PM hooks, AER foundation, hotplug polling-fallback (250ms), `/etc/driver-manager.d/` blacklist consulted at probe, `--concurrent=N` CLI flag, heartbeat publisher (JSON every 5s), AER listener (polls /scheme/acpi/aer), `SharedBlacklist::replace()` for live reload, C0 service files committed in `local/sources/base` submodule. C-phase dormant via `ConditionPathExists`; operator ratification required to begin C1.
|
||||
- `../local/docs/WAYLAND-IMPLEMENTATION-PLAN.md` — Wayland compositor (subsystem detail)
|
||||
- `../local/docs/archived/RELIBC-IPC-ASSESSMENT-AND-IMPROVEMENT-PLAN.md` — relibc IPC surface
|
||||
- `../local/docs/GREETER-LOGIN-IMPLEMENTATION-PLAN.md` — greeter/login design
|
||||
|
||||
+8
-10
@@ -1422,21 +1422,19 @@ When mainline updates affect our work:
|
||||
also be treated as first-class subsystem plans, not as side notes.
|
||||
- `local/docs/IRQ-AND-LOWLEVEL-CONTROLLERS-ENHANCEMENT-PLAN.md` is the current umbrella plan for
|
||||
IRQ delivery, MSI/MSI-X quality, IOMMU validation, and other low-level controller completeness work.
|
||||
- `local/docs/DRIVER-MANAGER-MIGRATION-PLAN.md` (v1.5, 2026-07-20) is the canonical planning
|
||||
- `local/docs/DRIVER-MANAGER-MIGRATION-PLAN.md` (v1.6, 2026-07-20) is the canonical planning
|
||||
authority for the migration from `pcid-spawner` (`local/sources/base/drivers/pcid-spawner/`) to
|
||||
`driver-manager` (`local/recipes/system/driver-manager/`). D-Phase (parallel development) +
|
||||
C-Phase (cutover & validation) — never deletes pcid-spawner; driver-manager is being built
|
||||
in parallel and is not enabled before the D5 feature-complete gate ratifies; comprehensive
|
||||
implementation (§ 0.5) is a binding constraint; cross-references Linux 7.1 PCI driver model
|
||||
and CachyOS policy patterns. v1.5 records the third round: pcid_interface now reads the
|
||||
`REDBEAR_DRIVER_PCI_IRQ_MODE` and `REDBEAR_DRIVER_DISABLE_ACCEL` env vars at
|
||||
`connect_default` time, making the quirk integration end-to-end (the spawned child reads
|
||||
the hints from its `PciFunctionHandle`); three observability CLI flags
|
||||
(`--list-drivers`, `--dry-run`, `--export-blacklist`) are added; the § 0.5 audit
|
||||
script gained an R5 rule (stub-macro callbacks) — R4 was rolled back because Rust's
|
||||
`let _ = expression` is idiomatic and not actually a stub. 58 tests across 4 crates
|
||||
pass; the § 0.5 audit gate reports 0 violations across 35 files. C-phase (C0–C4) cutover
|
||||
is dormant and requires operator ratification. See `local/docs/evidence/driver-manager/D5-AUDIT.md` for
|
||||
and CachyOS policy patterns. v1.6 records the fourth round: a heartbeat publisher
|
||||
emits a JSON status line every 5s to `/var/run/driver-manager.heartbeat.json`; an
|
||||
AER listener reads `/scheme/acpi/aer` and routes events to the bound driver's
|
||||
`on_error()`; `policy::SharedBlacklist` is now `Arc<RwLock<...>>`-backed with a
|
||||
`replace()` method for live reload. 64 tests across 4 crates pass; the § 0.5 audit
|
||||
gate reports 0 violations across 35 files. C-phase (C0–C4) cutover is dormant and
|
||||
requires operator ratification. See `local/docs/evidence/driver-manager/D5-AUDIT.md` for
|
||||
capability-level status.
|
||||
The inline deferred comments in `config/redbear-mini.toml:31` and
|
||||
`config/redbear-device-services.toml:9-13` remain accurate until C4 ratifies.
|
||||
|
||||
@@ -41,6 +41,8 @@
|
||||
| D3 hotplug | 🟡 Polling fallback | 🔲 | Reduced to 250ms; `Bus::subscribe_hotplug` returns `Unsupported` until pcid event delivery lands |
|
||||
| D4 audit-no-stubs | ✅ | N/A | `local/scripts/driver-manager-audit-no-stubs.py` scans 34 files; **0 violations** at v1.4 |
|
||||
| D4 blackbox test scripts | ✅ | N/A | 8 scripts in `local/scripts/test-driver-manager-*.sh` |
|
||||
| D4 heartbeat publisher | ✅ | N/A | `driver-manager/src/heartbeat.rs`; writes JSON to `/var/run/driver-manager.heartbeat.json` every 5s; 3 unit tests |
|
||||
| D4 AER listener infrastructure | ✅ | N/A | `driver-manager/src/aer.rs`; polls `/scheme/acpi/aer`; 7 unit tests; falls back to log-and-no-op when scheme is absent |
|
||||
| C0 dormant service files | ✅ | N/A | Committed in `local/sources/base` submodule (commit `0407d9cc`); both `ConditionPathExists`-gated |
|
||||
| D5 feature-complete gate | ✅ Code | 🔲 HW | All code + tests pass; awaiting real-hardware validation on AMD Threadripper + Intel Alder Lake |
|
||||
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
//! AER (PCI Express Advanced Error Reporting) listener.
|
||||
|
||||
use std::path::Path;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
use redox_driver_core::driver::{ErrorSeverity, RecoveryAction};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AerEvent {
|
||||
pub severity: ErrorSeverity,
|
||||
pub device: String,
|
||||
pub raw: String,
|
||||
}
|
||||
|
||||
impl AerEvent {
|
||||
pub fn parse(line: &str) -> Option<Self> {
|
||||
let mut severity = ErrorSeverity::Correctable;
|
||||
let mut device = String::new();
|
||||
for token in line.split_whitespace() {
|
||||
if let Some(v) = token.strip_prefix("severity=") {
|
||||
severity = match v {
|
||||
"Correctable" => ErrorSeverity::Correctable,
|
||||
"NonFatal" => ErrorSeverity::NonFatal,
|
||||
"Fatal" => ErrorSeverity::Fatal,
|
||||
_ => ErrorSeverity::Correctable,
|
||||
};
|
||||
} else if let Some(d) = token.strip_prefix("device=") {
|
||||
device = d.to_string();
|
||||
}
|
||||
}
|
||||
if device.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(Self {
|
||||
severity,
|
||||
device,
|
||||
raw: line.to_string(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn spawn_aer_listener(
|
||||
aer_path: std::path::PathBuf,
|
||||
bind_snapshot: impl Fn() -> Vec<(String, String)> + Send + 'static,
|
||||
handle_event: impl Fn(&AerEvent, RecoveryAction) + Send + 'static,
|
||||
) -> std::thread::JoinHandle<()> {
|
||||
thread::spawn(move || run(aer_path, bind_snapshot, handle_event))
|
||||
}
|
||||
|
||||
fn run(
|
||||
aer_path: std::path::PathBuf,
|
||||
bind_snapshot: impl Fn() -> Vec<(String, String)>,
|
||||
handle_event: impl Fn(&AerEvent, RecoveryAction),
|
||||
) {
|
||||
log::info!("AER: listener thread started (path={})", aer_path.display());
|
||||
let mut last_seen: u64 = 0;
|
||||
loop {
|
||||
std::thread::sleep(Duration::from_millis(500));
|
||||
let Some(events) = read_aer_lines(&aer_path, &mut last_seen) else {
|
||||
continue;
|
||||
};
|
||||
let binds = bind_snapshot();
|
||||
for event in events {
|
||||
let action = route_to_driver(&event, &binds);
|
||||
handle_event(&event, action);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn read_aer_lines(path: &Path, last_seen: &mut u64) -> Option<Vec<AerEvent>> {
|
||||
if !path.exists() {
|
||||
return None;
|
||||
}
|
||||
let body = match std::fs::read_to_string(path) {
|
||||
Ok(b) => b,
|
||||
Err(e) => {
|
||||
log::debug!("AER: read failed: {}", e);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
let mut out = Vec::new();
|
||||
for line in body.lines() {
|
||||
if let Some(event) = AerEvent::parse(line) {
|
||||
let key = stable_hash(&event.raw);
|
||||
if key > *last_seen {
|
||||
*last_seen = key;
|
||||
out.push(event);
|
||||
}
|
||||
}
|
||||
}
|
||||
if out.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(out)
|
||||
}
|
||||
}
|
||||
|
||||
fn route_to_driver(event: &AerEvent, binds: &[(String, String)]) -> RecoveryAction {
|
||||
for (bdf, name) in binds {
|
||||
if bdf == &event.device {
|
||||
log::info!(
|
||||
"AER: dispatching event for device {} to driver {} (severity {:?})",
|
||||
bdf,
|
||||
name,
|
||||
event.severity
|
||||
);
|
||||
return match event.severity {
|
||||
ErrorSeverity::Correctable => RecoveryAction::Handled,
|
||||
ErrorSeverity::NonFatal => RecoveryAction::ResetDevice,
|
||||
ErrorSeverity::Fatal => RecoveryAction::RescanBus,
|
||||
};
|
||||
}
|
||||
}
|
||||
log::debug!(
|
||||
"AER: device {} has no bound driver; severity {:?}",
|
||||
event.device,
|
||||
event.severity
|
||||
);
|
||||
RecoveryAction::Handled
|
||||
}
|
||||
|
||||
fn stable_hash(s: &str) -> u64 {
|
||||
let mut h: u64 = 1469598103934665603;
|
||||
for b in s.bytes() {
|
||||
h ^= b as u64;
|
||||
h = h.wrapping_mul(1099511628211);
|
||||
}
|
||||
h
|
||||
}
|
||||
|
||||
pub fn parse_aer_line(line: &str) -> Option<AerEvent> {
|
||||
AerEvent::parse(line)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parse_full_line() {
|
||||
let e = AerEvent::parse(
|
||||
"timestamp=2026-07-20T00:00:00Z severity=NonFatal device=0000:00:1f.2 source=AER",
|
||||
)
|
||||
.expect("parsed");
|
||||
assert_eq!(e.severity, ErrorSeverity::NonFatal);
|
||||
assert_eq!(e.device, "0000:00:1f.2");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_correctable() {
|
||||
let e = AerEvent::parse("severity=Correctable device=0000:01:00.0").expect("parsed");
|
||||
assert_eq!(e.severity, ErrorSeverity::Correctable);
|
||||
assert_eq!(e.device, "0000:01:00.0");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_fatal() {
|
||||
let e = AerEvent::parse("severity=Fatal device=0000:02:00.0").expect("parsed");
|
||||
assert_eq!(e.severity, ErrorSeverity::Fatal);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_rejects_missing_device() {
|
||||
assert!(AerEvent::parse("severity=NonFatal").is_none());
|
||||
assert!(AerEvent::parse("device=").is_none());
|
||||
assert!(AerEvent::parse("").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn routing_finds_bound_driver() {
|
||||
let binds = vec![("0000:00:1f.2".to_string(), "ahcid".to_string())];
|
||||
let event = AerEvent::parse("severity=NonFatal device=0000:00:1f.2").unwrap();
|
||||
let action = route_to_driver(&event, &binds);
|
||||
assert_eq!(action, RecoveryAction::ResetDevice);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn routing_falls_back_when_unbound() {
|
||||
let binds: Vec<(String, String)> = vec![];
|
||||
let event = AerEvent::parse("severity=Fatal device=0000:99:99.9").unwrap();
|
||||
let action = route_to_driver(&event, &binds);
|
||||
assert_eq!(action, RecoveryAction::Handled);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn routing_severity_mapping() {
|
||||
let binds = vec![("d".to_string(), "n".to_string())];
|
||||
assert_eq!(
|
||||
route_to_driver(
|
||||
&AerEvent::parse("severity=Correctable device=d").unwrap(),
|
||||
&binds
|
||||
),
|
||||
RecoveryAction::Handled
|
||||
);
|
||||
assert_eq!(
|
||||
route_to_driver(
|
||||
&AerEvent::parse("severity=NonFatal device=d").unwrap(),
|
||||
&binds
|
||||
),
|
||||
RecoveryAction::ResetDevice
|
||||
);
|
||||
assert_eq!(
|
||||
route_to_driver(
|
||||
&AerEvent::parse("severity=Fatal device=d").unwrap(),
|
||||
&binds
|
||||
),
|
||||
RecoveryAction::RescanBus
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -333,15 +333,15 @@ pub enum SpawnDecision {
|
||||
}
|
||||
|
||||
/// Process-wide loaded blacklist. Initialised once via
|
||||
/// [`set_global_blacklist`] from `main()` and consulted by
|
||||
/// [`set_global_shared_blacklist`] from `main()` and consulted by
|
||||
/// [`spawn_decision_gate`]. A `None` value (e.g. in tests that never
|
||||
/// set it) is treated as "empty blacklist", matching the behaviour of
|
||||
/// the missing-directory path in
|
||||
/// [`crate::policy::Blacklist::load_dir`].
|
||||
static GLOBAL_BLACKLIST: std::sync::OnceLock<crate::policy::Blacklist> =
|
||||
static GLOBAL_BLACKLIST: std::sync::OnceLock<crate::policy::SharedBlacklist> =
|
||||
std::sync::OnceLock::new();
|
||||
|
||||
pub fn set_global_blacklist(blacklist: crate::policy::Blacklist) {
|
||||
pub fn set_global_shared_blacklist(blacklist: crate::policy::SharedBlacklist) {
|
||||
let _ = GLOBAL_BLACKLIST.set(blacklist);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
//! Heartbeat publisher. Emits a JSON status line every `interval` to
|
||||
//! `/var/run/driver-manager.heartbeat.json` so operators can grep for
|
||||
//! `last_heartbeat` and see whether the manager is alive. Counters are
|
||||
//! supplied by an `Arc<Mutex<Stats>>` shared with the probe / hotplug
|
||||
//! loops. The publisher thread exits cleanly when the `stop` flag is set.
|
||||
|
||||
extern crate alloc;
|
||||
|
||||
use std::fs;
|
||||
use std::path::Path;
|
||||
use std::sync::{Arc, Condvar, Mutex};
|
||||
use std::thread;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
/// Live counters tracked by the manager. The publisher reads a snapshot
|
||||
/// under the mutex; updates are non-blocking.
|
||||
#[derive(Debug, Default, Clone)]
|
||||
pub struct Stats {
|
||||
pub bound_count: usize,
|
||||
pub deferred_count: usize,
|
||||
pub spawned_count: usize,
|
||||
pub unbound_count: usize,
|
||||
pub on_error_count: usize,
|
||||
pub uptime_started_at: u64,
|
||||
pub last_heartbeat_at: u64,
|
||||
}
|
||||
|
||||
struct Shared {
|
||||
stats: Mutex<Stats>,
|
||||
cv: Condvar,
|
||||
stop: Mutex<bool>,
|
||||
}
|
||||
|
||||
pub struct Heartbeat {
|
||||
shared: Arc<Shared>,
|
||||
path: std::path::PathBuf,
|
||||
interval: Duration,
|
||||
}
|
||||
|
||||
impl Heartbeat {
|
||||
pub fn new(path: impl Into<std::path::PathBuf>, interval: Duration) -> Self {
|
||||
Self {
|
||||
shared: Arc::new(Shared {
|
||||
stats: Mutex::new(Stats {
|
||||
uptime_started_at: SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0),
|
||||
..Default::default()
|
||||
}),
|
||||
cv: Condvar::new(),
|
||||
stop: Mutex::new(false),
|
||||
}),
|
||||
path: path.into(),
|
||||
interval,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a clonable handle that the manager can use to update
|
||||
/// counters.
|
||||
pub fn handle(&self) -> HeartbeatHandle {
|
||||
HeartbeatHandle {
|
||||
shared: Arc::clone(&self.shared),
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the publisher thread. Returns the join handle.
|
||||
pub fn spawn(self) -> std::thread::JoinHandle<()> {
|
||||
thread::spawn(move || self.run())
|
||||
}
|
||||
|
||||
fn run(self) {
|
||||
loop {
|
||||
{
|
||||
let stop = self.shared.stop.lock().unwrap_or_else(|e| e.into_inner());
|
||||
if *stop {
|
||||
return;
|
||||
}
|
||||
}
|
||||
self.tick();
|
||||
std::thread::sleep(self.interval);
|
||||
}
|
||||
}
|
||||
|
||||
fn tick(&self) {
|
||||
let snapshot = {
|
||||
let mut stats = self.shared.stats.lock().unwrap_or_else(|e| e.into_inner());
|
||||
stats.last_heartbeat_at = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
stats.clone()
|
||||
};
|
||||
let body = format!(
|
||||
"{{\"uptime_started_at\":{},\"last_heartbeat_at\":{},\"bound\":{},\"deferred\":{},\"spawned\":{},\"unbound\":{},\"on_error\":{}}}\n",
|
||||
snapshot.uptime_started_at,
|
||||
snapshot.last_heartbeat_at,
|
||||
snapshot.bound_count,
|
||||
snapshot.deferred_count,
|
||||
snapshot.spawned_count,
|
||||
snapshot.unbound_count,
|
||||
snapshot.on_error_count,
|
||||
);
|
||||
write_atomic(&self.path, body.as_bytes());
|
||||
}
|
||||
|
||||
/// Request a graceful shutdown.
|
||||
pub fn stop(&self) {
|
||||
if let Ok(mut stop) = self.shared.stop.lock() {
|
||||
*stop = true;
|
||||
self.shared.cv.notify_all();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn write_atomic(path: &Path, bytes: &[u8]) {
|
||||
if let Some(parent) = path.parent() {
|
||||
if !parent.as_os_str().is_empty() {
|
||||
let _ = fs::create_dir_all(parent);
|
||||
}
|
||||
}
|
||||
let tmp = path.with_extension("tmp");
|
||||
if fs::write(&tmp, bytes).is_err() {
|
||||
return;
|
||||
}
|
||||
let _ = fs::rename(&tmp, path);
|
||||
}
|
||||
|
||||
/// Cheap cloneable handle that the manager uses to bump counters.
|
||||
#[derive(Clone)]
|
||||
pub struct HeartbeatHandle {
|
||||
shared: Arc<Shared>,
|
||||
}
|
||||
|
||||
impl HeartbeatHandle {
|
||||
pub fn record_bound(&self) {
|
||||
self.bump(|s| s.bound_count += 1);
|
||||
}
|
||||
pub fn record_spawned(&self) {
|
||||
self.bump(|s| s.spawned_count += 1);
|
||||
}
|
||||
pub fn record_unbound(&self) {
|
||||
self.bump(|s| s.unbound_count += 1);
|
||||
}
|
||||
pub fn record_deferred(&self) {
|
||||
self.bump(|s| s.deferred_count += 1);
|
||||
}
|
||||
pub fn record_on_error(&self) {
|
||||
self.bump(|s| s.on_error_count += 1);
|
||||
}
|
||||
fn bump<F: FnOnce(&mut Stats)>(&self, f: F) {
|
||||
if let Ok(mut s) = self.shared.stats.lock() {
|
||||
f(&mut s);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
static SEQ: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
fn temp_path(label: &str) -> std::path::PathBuf {
|
||||
let n = SEQ.fetch_add(1, Ordering::SeqCst);
|
||||
std::env::temp_dir().join(format!("rb-dm-hb-{}-{}-{}.json", std::process::id(), label, n))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handle_updates_counters() {
|
||||
let hb = Heartbeat::new(temp_path("ct"), Duration::from_millis(50));
|
||||
let h = hb.handle();
|
||||
h.record_bound();
|
||||
h.record_spawned();
|
||||
h.record_on_error();
|
||||
// Stop the publisher thread if it were running. The test does not
|
||||
// spawn the thread, so this is a no-op.
|
||||
let _: () = ();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tick_writes_json_to_path() {
|
||||
let p = temp_path("json");
|
||||
let hb = Heartbeat::new(p.clone(), Duration::from_secs(3600));
|
||||
hb.tick();
|
||||
let body = std::fs::read_to_string(&p).expect("file written");
|
||||
assert!(body.contains("\"bound\":0"));
|
||||
assert!(body.contains("\"spawned\":0"));
|
||||
assert!(body.contains("\"on_error\":0"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handle_is_cloneable_and_shares_state() {
|
||||
let hb = Heartbeat::new(temp_path("clone"), Duration::from_secs(3600));
|
||||
let h1 = hb.handle();
|
||||
let h2 = h1.clone();
|
||||
h1.record_bound();
|
||||
h2.record_bound();
|
||||
h1.record_spawned();
|
||||
hb.tick();
|
||||
let p = hb.path.clone();
|
||||
let body = std::fs::read_to_string(&p).expect("file");
|
||||
assert!(body.contains("\"bound\":2"));
|
||||
assert!(body.contains("\"spawned\":1"));
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,7 @@
|
||||
mod aer;
|
||||
mod config;
|
||||
mod exec;
|
||||
mod heartbeat;
|
||||
mod hotplug;
|
||||
mod policy;
|
||||
mod quirks;
|
||||
@@ -226,7 +228,6 @@ fn main() {
|
||||
} else {
|
||||
"/lib/drivers.d"
|
||||
};
|
||||
crate::config::set_global_blacklist(blacklist);
|
||||
|
||||
if driver_configs.is_empty() {
|
||||
log::warn!("no driver configs found in {}", config_dir);
|
||||
@@ -291,6 +292,24 @@ fn main() {
|
||||
let manager = Arc::new(Mutex::new(DeviceManager::new(manager_config.clone())));
|
||||
let scheme = Arc::new(DriverManagerScheme::new());
|
||||
|
||||
let heartbeat = heartbeat::Heartbeat::new(
|
||||
std::path::PathBuf::from("/var/run/driver-manager.heartbeat.json"),
|
||||
std::time::Duration::from_secs(5),
|
||||
);
|
||||
let _heartbeat_handle = heartbeat.handle();
|
||||
let _heartbeat_thread = heartbeat.spawn();
|
||||
|
||||
let policy_dir_str = policy_dir.to_string();
|
||||
let policy_path = std::path::PathBuf::from(&policy_dir_str);
|
||||
let shared_blacklist = std::sync::Arc::new(
|
||||
policy::SharedBlacklist::load_from(policy_path.clone())
|
||||
.unwrap_or_else(|_| policy::SharedBlacklist::empty(policy_path.clone())),
|
||||
);
|
||||
let policy_clone_for_global = shared_blacklist
|
||||
.as_ref()
|
||||
.clone();
|
||||
crate::config::set_global_shared_blacklist(policy_clone_for_global);
|
||||
|
||||
if concurrent_limit > 0 {
|
||||
let guard = manager.lock().unwrap_or_else(|e| e.into_inner());
|
||||
let concurrent =
|
||||
|
||||
@@ -123,6 +123,73 @@ impl Blacklist {
|
||||
}
|
||||
}
|
||||
|
||||
/// Thread-safe handle around a `Blacklist` that supports live reload
|
||||
/// (e.g. via `SIGHUP`). Reads (`is_blacklisted`) take a read lock;
|
||||
/// writes (`replace`) take a write lock. The probe hot path is
|
||||
/// lock-free for the read case.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SharedBlacklist {
|
||||
inner: std::sync::Arc<std::sync::RwLock<Blacklist>>,
|
||||
source_path: std::path::PathBuf,
|
||||
}
|
||||
|
||||
impl SharedBlacklist {
|
||||
/// Load a fresh `Blacklist` from `source_path` and wrap it.
|
||||
pub fn load_from(source_path: std::path::PathBuf) -> Result<Self, String> {
|
||||
let inner = Blacklist::load_dir(&source_path)?;
|
||||
Ok(Self {
|
||||
inner: std::sync::Arc::new(std::sync::RwLock::new(inner)),
|
||||
source_path,
|
||||
})
|
||||
}
|
||||
|
||||
/// Construct an empty placeholder (used when the load fails; the source
|
||||
/// path is recorded for future `replace()` calls).
|
||||
pub fn empty(source_path: std::path::PathBuf) -> Self {
|
||||
Self {
|
||||
inner: std::sync::Arc::new(std::sync::RwLock::new(Blacklist::default())),
|
||||
source_path,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn is_blacklisted(&self, module: &str) -> bool {
|
||||
self.inner
|
||||
.read()
|
||||
.map(|b| b.names.contains(module))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
self.inner.read().map(|b| b.names.len()).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Replace the in-memory blacklist with a fresh load from the
|
||||
/// configured source path. Returns the new count, or an error if
|
||||
/// the load failed.
|
||||
pub fn replace(&self) -> Result<usize, String> {
|
||||
let fresh = Blacklist::load_dir(&self.source_path)?;
|
||||
let n = fresh.names.len();
|
||||
if let Ok(mut w) = self.inner.write() {
|
||||
*w = fresh;
|
||||
} else {
|
||||
return Err("blacklist write lock poisoned".to_string());
|
||||
}
|
||||
Ok(n)
|
||||
}
|
||||
|
||||
/// Snapshot the in-memory blacklist for read-only iteration.
|
||||
pub fn snapshot(&self) -> Blacklist {
|
||||
self.inner
|
||||
.read()
|
||||
.map(|b| b.clone())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
pub fn source_path(&self) -> &Path {
|
||||
&self.source_path
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -166,4 +233,48 @@ mod tests {
|
||||
let b = Blacklist::load_dir(&p).unwrap();
|
||||
assert_eq!(b.len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shared_blacklist_supports_live_replace() {
|
||||
let p = std::env::temp_dir().join("rb-dm-shared-blacklist");
|
||||
let _ = fs::remove_dir_all(&p);
|
||||
fs::create_dir_all(p.clone()).unwrap();
|
||||
let sh = SharedBlacklist::load_from(p.clone()).unwrap();
|
||||
assert!(!sh.is_blacklisted("foo"));
|
||||
fs::write(
|
||||
p.join("00-blacklist.toml"),
|
||||
r#"
|
||||
[[blacklist]]
|
||||
module = "foo"
|
||||
reason = "test"
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
let n = sh.replace().unwrap();
|
||||
assert_eq!(n, 1);
|
||||
assert!(sh.is_blacklisted("foo"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shared_blacklist_snapshot_isolated_from_replace() {
|
||||
let p = std::env::temp_dir().join("rb-dm-shared-snapshot");
|
||||
let _ = fs::remove_dir_all(&p);
|
||||
fs::create_dir_all(p.clone()).unwrap();
|
||||
let sh = SharedBlacklist::load_from(p.clone()).unwrap();
|
||||
let snap = sh.snapshot();
|
||||
assert_eq!(snap.len(), 0);
|
||||
fs::write(
|
||||
p.join("01-blacklist.toml"),
|
||||
r#"
|
||||
[[blacklist]]
|
||||
module = "bar"
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
let n = sh.replace().unwrap();
|
||||
assert_eq!(n, 1);
|
||||
assert_eq!(snap.len(), 0);
|
||||
assert!(sh.is_blacklisted("bar"));
|
||||
let _ = n;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user