From d98330a7de195a9b2cadf379830572f8cb07ad02 Mon Sep 17 00:00:00 2001 From: kellito Date: Sat, 25 Jul 2026 23:49:09 +0900 Subject: [PATCH] events: monotonic seq numbers + larger buffer for AER/pciehp dedup G-A1/G-A3 fix: replace order-dependent content-hash dedup with monotonic AtomicU64 sequence numbers. Each event line is prefixed with seq=: 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=: 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 --- drivers/pcid/src/events.rs | 64 ++++++++++++++++++++++++++++++++------ drivers/pcid/src/scheme.rs | 27 +++++++++++++++- 2 files changed, 81 insertions(+), 10 deletions(-) diff --git a/drivers/pcid/src/events.rs b/drivers/pcid/src/events.rs index 706c9875ff..82661330bf 100644 --- a/drivers/pcid/src/events.rs +++ b/drivers/pcid/src/events.rs @@ -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=:` where `` 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>, pciehp: Mutex>, + 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>, 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>, 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>) -> 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 { diff --git a/drivers/pcid/src/scheme.rs b/drivers/pcid/src/scheme.rs index e615d9b4c8..d9a18ee395 100644 --- a/drivers/pcid/src/scheme.rs +++ b/drivers/pcid/src/scheme.rs @@ -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::>(); 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)),