vasilito/RedBear-OS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

events: monotonic seq numbers + larger buffer for AER/pciehp dedup

Browse Source 
G-A1/G-A3 fix: replace order-dependent content-hash dedup with
monotonic AtomicU64 sequence numbers. Each event line is prefixed
with seq=<n>: so consumers can track last_seq and skip already-
processed events. MAX_EVENTS raised from 64 to 256 for headroom.

G-A5 support: add /scheme/pci/aer_seq and /scheme/pci/pciehp_seq
endpoints returning latest_seq() as UTF-8, so consumers can query
'what is the latest?' atomically without parsing the whole log.

Changes:
- events.rs: AtomicU64 seq_counter + high_water_mark in EventLog
- events.rs: next_seq() allocates monotonic seq, CAS-updates HWM
- events.rs: push() prepends seq=<n>: to every line
- events.rs: latest_seq() returns high_water_mark
- events.rs: MAX_EVENTS 64 -> 256
- scheme.rs: Handle::AerSeq, Handle::PciehpSeq variants
- scheme.rs: aer_seq/pciehp_seq path, fstat, read, getdents wiring
This commit is contained in:
kellito
2026-07-25 23:49:09 +09:00
parent 29166263bb
commit d98330a7de
2 changed files with 81 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drivers/pcid/src/events.rs
+55 -9
View File
@@ -9,12 +9,24 @@
//! button, MRL sensor, power fault), then W1C-clears them.
//!
//! Events accumulate in capped in-memory logs served read-only as
//! `/scheme/pci/aer` and `/scheme/pci/pciehp`. Consumers (driver-manager's
//! unified listener) hash lines to deduplicate across polls, matching the
//! polling-fallback model in the migration plan. Proper MSI-based delivery
//! is the follow-up once pcid grows an IRQ path for AER/hotplug.
//! `/scheme/pci/aer` and `/scheme/pci/pciehp`.
//!
//! # Sequence numbers (dedup protocol)
//!
//! Every emitted event line is prefixed with `seq=<n>:` where `<n>` is a
//! process-global monotonic counter (`AtomicU64`, lock-free). Consumers use
//! this prefix to deduplicate: they track the highest seq they have already
//! processed and only emit events whose seq is strictly greater. This
//! replaces the old content-hash approach which was order-dependent and
//! silently dropped events whose hash fell below the running maximum.
//!
//! The `high_water_mark` tracks the highest seq ever issued, even after the
//! corresponding line has been evicted from the ring buffer. This lets
//! consumers query `/scheme/pci/aer_seq` / `/scheme/pci/pciehp_seq` for an
//! atomic "what's the latest?" check without parsing the whole log.
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
@@ -24,7 +36,7 @@ use pci_types::{ConfigRegionAccess, PciAddress};
use crate::cfg_access::Pcie;
const MAX_EVENTS: usize = 64;
const MAX_EVENTS: usize = 256;
const POLL_INTERVAL_MS: u64 = 500;
const AER_CAP_ID: u16 = 0x0001;
@@ -50,6 +62,8 @@ fn scheme_name(addr: &PciAddress) -> String {
pub struct EventLog {
aer: Mutex<VecDeque<String>>,
pciehp: Mutex<VecDeque<String>>,
seq_counter: AtomicU64,
high_water_mark: AtomicU64,
}
impl EventLog {
@@ -57,15 +71,47 @@ impl EventLog {
Self {
aer: Mutex::new(VecDeque::new()),
pciehp: Mutex::new(VecDeque::new()),
seq_counter: AtomicU64::new(0),
high_water_mark: AtomicU64::new(0),
}
}
fn push(log: &Mutex<VecDeque<String>>, line: String) {
/// Allocate the next monotonic sequence number and advance
/// `high_water_mark` if necessary. The counter starts at 1 so seq 0
/// is never a valid event (consumers can safely treat seq 0 as
/// "nothing seen yet").
fn next_seq(&self) -> u64 {
let seq = self.seq_counter.fetch_add(1, Ordering::Relaxed) + 1;
let mut current = self.high_water_mark.load(Ordering::Relaxed);
while seq > current {
match self.high_water_mark.compare_exchange_weak(
current,
seq,
Ordering::Relaxed,
Ordering::Relaxed,
) {
Ok(_) => break,
Err(actual) => current = actual,
}
}
seq
}
/// Returns the highest sequence number ever issued by this `EventLog`,
/// even if the corresponding event line has been evicted from the ring
/// buffer. Returns 0 when no events have been emitted yet.
pub fn latest_seq(&self) -> u64 {
self.high_water_mark.load(Ordering::Relaxed)
}
fn push(&self, log: &Mutex<VecDeque<String>>, line: String) {
let seq = self.next_seq();
let prefixed = format!("seq={seq}:{line}");
let mut queue = log.lock().unwrap_or_else(|e| e.into_inner());
if queue.len() >= MAX_EVENTS {
queue.pop_front();
}
queue.push_back(line);
queue.push_back(prefixed);
}
fn snapshot(log: &Mutex<VecDeque<String>>) -> String {
@@ -74,11 +120,11 @@ impl EventLog {
}
pub fn push_aer(&self, line: String) {
Self::push(&self.aer, line);
self.push(&self.aer, line);
}
pub fn push_pciehp(&self, line: String) {
Self::push(&self.pciehp, line);
self.push(&self.pciehp, line);
}
pub fn aer_snapshot(&self) -> String {
drivers/pcid/src/scheme.rs
+26 -1
View File
@@ -26,6 +26,8 @@ enum Handle {
Channel { addr: PciAddress, st: ChannelState },
Aer,
Pciehp,
AerSeq,
PciehpSeq,
SchemeRoot,
}
struct HandleWrapper {
@@ -41,6 +43,8 @@ impl Handle {
| Self::Channel { .. }
| Self::Aer
| Self::Pciehp
| Self::AerSeq
| Self::PciehpSeq
)
}
fn is_dir(&self) -> bool {
@@ -109,14 +113,20 @@ impl SchemeSync for PciScheme {
.map(|(addr, _)| format!("{}", addr).replace(':', "--"))
.collect::<Vec<_>>();
entries.push("aer".to_string());
entries.push("aer_seq".to_string());
entries.push("pciehp".to_string());
entries.push("pciehp_seq".to_string());
Handle::TopLevel { entries }
} else if path == "access" {
Handle::Access
} else if path == "aer" {
Handle::Aer
} else if path == "aer_seq" {
Handle::AerSeq
} else if path == "pciehp" {
Handle::Pciehp
} else if path == "pciehp_seq" {
Handle::PciehpSeq
} else {
let idx = path.find('/').unwrap_or(path.len());
let (addr_str, after) = path.split_at(idx);
@@ -156,6 +166,8 @@ impl SchemeSync for PciScheme {
Handle::Access | Handle::Channel { .. } => (0, MODE_CHR | 0o600),
Handle::Aer => (self.events.aer_snapshot().len(), MODE_FILE | 0o444),
Handle::Pciehp => (self.events.pciehp_snapshot().len(), MODE_FILE | 0o444),
Handle::AerSeq => (self.events.latest_seq().to_string().len(), MODE_FILE | 0o444),
Handle::PciehpSeq => (self.events.latest_seq().to_string().len(), MODE_FILE | 0o444),
Handle::SchemeRoot => return Err(Error::new(EBADF)),
};
stat.st_size = len as u64;
@@ -222,6 +234,17 @@ impl SchemeSync for PciScheme {
buf[..count].copy_from_slice(&bytes[offset..offset + count]);
Ok(count)
}
Handle::AerSeq | Handle::PciehpSeq => {
let data = self.events.latest_seq().to_string();
let bytes = data.as_bytes();
let offset = usize::try_from(_offset).map_err(|_| Error::new(EINVAL))?;
if offset >= bytes.len() {
return Ok(0);
}
let count = (bytes.len() - offset).min(buf.len());
buf[..count].copy_from_slice(&bytes[offset..offset + count]);
Ok(count)
}
Handle::SchemeRoot => Err(Error::new(EBADF)),
_ => Err(Error::new(EBADF)),
}
@@ -259,7 +282,9 @@ impl SchemeSync for PciScheme {
| Handle::Config { .. }
| Handle::Channel { .. }
| Handle::Aer
| Handle::Pciehp => {
| Handle::Pciehp
| Handle::AerSeq
| Handle::PciehpSeq => {
return Err(Error::new(ENOTDIR));
}
Handle::SchemeRoot => return Err(Error::new(EBADF)),