use std::collections::BTreeSet;
use crate::server::record::{PiniMode, RecordInstance, ScanList, ScanType};
use super::PvDatabase;
pub(super) struct ScanIndex {
buckets: std::sync::OnceLock<
Box<[crate::runtime::sync::PriorityInheritanceMutex<BTreeSet<super::ScanKey>>]>,
>,
overruns: std::sync::OnceLock<Box<[std::sync::atomic::AtomicU64]>>,
}
impl ScanIndex {
pub(super) fn new() -> Self {
Self {
buckets: std::sync::OnceLock::new(),
overruns: std::sync::OnceLock::new(),
}
}
fn bucket(
&self,
list: ScanList,
) -> &crate::runtime::sync::PriorityInheritanceMutex<BTreeSet<super::ScanKey>> {
&self
.buckets
.get_or_init(|| {
(0..ScanList::count())
.map(|_| crate::runtime::sync::PriorityInheritanceMutex::new(BTreeSet::new()))
.collect()
})
.as_ref()[list.slot()]
}
fn overrun(&self, list: ScanList) -> &std::sync::atomic::AtomicU64 {
&self
.overruns
.get_or_init(|| {
(0..ScanList::count())
.map(|_| std::sync::atomic::AtomicU64::new(0))
.collect()
})
.as_ref()[list.slot()]
}
}
impl PvDatabase {
pub(super) fn add_to_scan_list(
&self,
scan: ScanType,
phas: i16,
record_type: &str,
load_order: u64,
name: &str,
) {
let Some(list) = scan.scan_list() else {
return;
};
self.inner
.scan_index
.bucket(list)
.lock()
.insert(super::ScanKey::new(phas, record_type, load_order, name));
}
pub(super) fn delete_from_scan_list(&self, scan: ScanType, name: &str) {
let Some(list) = scan.scan_list() else {
return;
};
self.inner
.scan_index
.bucket(list)
.lock()
.retain(|k| k.name != name);
}
pub fn update_scan_index(
&self,
name: &str,
old_scan: ScanType,
_new_scan: ScanType,
old_phas: i16,
_new_phas: i16,
) {
let _gate = self.lock_registration("update_scan_index");
let _ = old_phas; self.delete_from_scan_list(old_scan, name);
let rec_arc = match self.inner.records.read().get(name).cloned() {
Some(r) => r,
None => return,
};
let (cur_scan, cur_phas, cur_type) = {
let inst = rec_arc.read();
(
inst.common.scan,
inst.common.phas,
inst.record.record_type(),
)
};
let seq = self.inner.load_order.load().get(name).copied().unwrap_or(0);
self.add_to_scan_list(cur_scan, cur_phas, cur_type, seq, name);
}
pub(crate) fn record_scan_overrun(&self, scan: ScanType) {
if let Some(list) = scan.scan_list() {
self.inner
.scan_index
.overrun(list)
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
}
pub(crate) fn scan_overruns(&self, list: crate::server::record::ScanList) -> u64 {
self.inner
.scan_index
.overrun(list)
.load(std::sync::atomic::Ordering::Relaxed)
}
pub async fn records_for_scan(&self, scan_type: ScanType) -> Vec<String> {
let Some(list) = scan_type.scan_list() else {
return Vec::new();
};
self.inner
.scan_index
.bucket(list)
.lock()
.iter()
.map(|k| k.name.clone())
.collect()
}
fn live_scan_key(&self, name: &str) -> Option<super::ScanKey> {
let rec = self.get_record_no_resolve(name)?;
let (phas, record_type) = {
let inst = rec.read();
(inst.common.phas, inst.record.record_type())
};
let seq = self.inner.load_order.load().get(name).copied().unwrap_or(0);
Some(super::ScanKey::new(phas, record_type, seq, name))
}
pub(crate) fn scan_cursor(&self, list: ScanList) -> ScanCursor {
ScanCursor { list, at: None }
}
pub(crate) async fn scan_list_once(&self, list: ScanList) {
let mut cursor = self.scan_cursor(list);
while let Some(name) = cursor.next(self) {
let mut visited = std::collections::HashSet::new();
let _ = self.process_record_with_links(&name, &mut visited, 0).await;
}
}
pub async fn pini_records(&self, mode: PiniMode) -> Vec<String> {
let mut result = Vec::new();
for (name, rec) in self.records_in_load_order().await {
if rec.read().common.pini == mode.to_u16() as i16 {
result.push(name);
}
}
result
}
async fn records_in_load_order(
&self,
) -> Vec<(String, std::sync::Arc<parking_lot::RwLock<RecordInstance>>)> {
let snapshot: Vec<_> = {
let records = self.inner.records.read();
records
.iter()
.map(|(n, r)| (n.clone(), r.clone()))
.collect()
};
let mut keyed: Vec<_> = {
let load_order = self.inner.load_order.load();
snapshot
.into_iter()
.map(|(name, rec)| (load_order.get(&name).copied().unwrap_or(0), name, rec))
.collect()
};
keyed.sort_unstable_by(|a, b| (a.0, &a.1).cmp(&(b.0, &b.1)));
keyed
.into_iter()
.map(|(_seq, name, rec)| (name, rec))
.collect()
}
pub async fn pini_process(&self, mode: PiniMode) {
const MIN_PHASE: i32 = i16::MIN as i32;
const NO_NEXT_PHASE: i32 = i16::MAX as i32 + 1;
let mut next = MIN_PHASE;
loop {
let this = next;
next = NO_NEXT_PHASE;
for (name, rec) in self.records_in_load_order().await {
let (pini, phas) = {
let instance = rec.read();
(instance.common.pini, i32::from(instance.common.phas))
};
if pini != mode.to_u16() as i16 {
continue;
}
if phas == this {
let mut visited = std::collections::HashSet::new();
let _ = self.process_record_with_links(&name, &mut visited, 0).await;
} else if phas > this && phas < next {
next = phas;
}
}
if next == NO_NEXT_PHASE {
return;
}
}
}
pub async fn post_event(&self) {
if !crate::server::scan::scan_is_running() {
return;
}
if let Some(list) = ScanType::Event.scan_list() {
self.scan_list_once(list).await;
}
}
pub async fn post_event_named(&self, event_name: &str) {
if !crate::server::scan::scan_is_running() {
return;
}
let want = normalize_event_name(event_name);
if want.is_empty() {
return;
}
let Some(list) = ScanType::Event.scan_list() else {
return;
};
let mut cursor = self.scan_cursor(list);
while let Some(name) = cursor.next(self) {
let evnt = match self.get_record(&name) {
Some(rec) => rec.read().common.evnt.clone(),
None => continue,
};
if normalize_event_name(&evnt) != want {
continue;
}
let mut visited = std::collections::HashSet::new();
let _ = self.process_record_with_links(&name, &mut visited, 0).await;
}
}
}
pub(crate) struct ScanCursor {
list: ScanList,
at: Option<super::ScanKey>,
}
impl ScanCursor {
pub(crate) fn next(&mut self, db: &PvDatabase) -> Option<String> {
use std::ops::Bound;
let resume = match self.at.take() {
None => None,
Some(was) => {
let live = db.live_scan_key(&was.name);
let moved_but_present = match &live {
Some(k) => db.inner.scan_index.bucket(self.list).lock().contains(k),
None => false,
};
Some(if moved_but_present {
live.expect("checked present")
} else {
was
})
}
};
let next = {
let bucket = db.inner.scan_index.bucket(self.list).lock();
match &resume {
None => bucket.iter().next().cloned(),
Some(at) => bucket
.range((Bound::Excluded(at.clone()), Bound::Unbounded))
.next()
.cloned(),
}
};
self.at = next.clone();
next.map(|k| k.name)
}
}
pub(crate) fn normalize_event_name(name: &str) -> String {
let trimmed = name.trim();
if trimmed.is_empty() {
return String::new();
}
if let Ok(num) = trimmed.parse::<f64>() {
if num >= 0.0 && num < 256.0 {
let int = num as i64;
if int < 1 {
return String::new();
}
return int.to_string();
}
}
trimmed.to_string()
}
#[cfg(test)]
mod tests {
use super::PvDatabase;
use super::normalize_event_name;
use crate::server::record::ScanType;
#[test]
fn a_caller_holding_l46_cannot_reach_the_scan_index_owner() {
let db = PvDatabase::new();
let held = db.lock_registration("a_test_standing_in_for_a_registration_entry_point");
let violation = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
db.update_scan_index("ANY", ScanType::Passive, ScanType::SEC01, 0, 0);
}));
let payload = violation
.expect_err("holding L46 across update_scan_index must panic, not park the thread");
let msg = payload
.downcast_ref::<String>()
.map(String::as_str)
.or_else(|| payload.downcast_ref::<&str>().copied())
.unwrap_or("");
assert!(
msg.contains("not reentrant") && msg.contains("update_scan_index"),
"the panic must name the rule and the violating site, got: {msg}"
);
drop(held);
}
#[test]
fn the_scan_index_owner_takes_l46_itself_when_no_caller_holds_it() {
let db = PvDatabase::new();
db.update_scan_index("ANY", ScanType::Passive, ScanType::SEC01, 0, 0);
}
#[test]
fn the_registration_gate_clears_on_drop_and_on_unwind() {
let db = PvDatabase::new();
drop(db.lock_registration("first"));
let _second = db.lock_registration("second");
drop(_second);
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _g = db.lock_registration("panics_while_held");
panic!("unwind with the gate live");
}));
drop(db.lock_registration("after_unwind"));
}
#[test]
fn event_name_numeric_normalisation() {
assert_eq!(normalize_event_name(" 5 "), "5");
assert_eq!(normalize_event_name("5.0"), "5");
assert_eq!(normalize_event_name("5"), "5");
assert_eq!(normalize_event_name("0"), "");
assert_eq!(normalize_event_name(""), "");
assert_eq!(normalize_event_name(" "), "");
assert_eq!(normalize_event_name("myEvent"), "myEvent");
assert_eq!(normalize_event_name(" myEvent "), "myEvent");
assert_eq!(normalize_event_name("999"), "999");
}
}