epics_base_rs/server/database/scan_index.rs
1use std::collections::BTreeSet;
2use std::sync::Arc;
3
4use crate::server::record::{PiniMode, RecordCell, ScanList, ScanType};
5
6use super::PvDatabase;
7
8/// The scan buckets themselves. `bucket` is private to THIS module, so no
9/// other file in `database` can open a bucket and write it — the only
10/// transitions are [`PvDatabase::add_to_scan_list`] and
11/// [`PvDatabase::delete_from_scan_list`], which is C's arrangement:
12/// `addToList`/`deleteFromList` are `static` in `dbScan.c` and `scanAdd` /
13/// `scanDelete` (`dbScan.c:240`/`:308` at R7.0.10) are the whole public
14/// surface.
15/// One scan list's records, and the revision that says whether they are still
16/// the ones a cursor last saw.
17///
18/// The two live together because the revision is only meaningful as a
19/// statement about these keys: [`ScanCursor`] walks the list as it stood at a
20/// revision, and the only thing that can invalidate that is a record entering,
21/// leaving, or being re-keyed within THIS bucket. Every such transition goes
22/// through [`ScanBucket::transition`], which moves the revision itself, so a
23/// bucket cannot change without saying so.
24///
25/// The revision sits OUTSIDE the mutex because that is what lets an ordinary
26/// sweep step skip the mutex entirely: while the bucket still reads the value
27/// the cursor holds, the cursor's copy of the list is the list. It is written
28/// only under the mutex and published `Release`, so a cursor that sees a new
29/// value and then takes the mutex sees the keys that produced it.
30struct ScanBucket {
31 revision: std::sync::atomic::AtomicU64,
32 entries: crate::runtime::sync::PriorityInheritanceMutex<Bucket>,
33}
34
35/// The keys themselves, and the ordered form a cursor walks.
36struct Bucket {
37 keys: BTreeSet<super::ScanKey>,
38 /// `keys` in order, materialised on the first ask after a transition and
39 /// shared with every ask until the next one. Written only by
40 /// [`ScanBucket::transition`] and [`ScanBucket::snapshot`].
41 ordered: Option<Arc<[super::ScanKey]>>,
42}
43
44impl ScanBucket {
45 fn new() -> Self {
46 Self {
47 revision: std::sync::atomic::AtomicU64::new(0),
48 entries: crate::runtime::sync::PriorityInheritanceMutex::new(Bucket {
49 keys: BTreeSet::new(),
50 ordered: None,
51 }),
52 }
53 }
54
55 /// The ONE way this bucket's keys change — C keeps `addToList` and
56 /// `deleteFromList` `static` in `dbScan.c` for the same reason.
57 ///
58 /// Moving the revision and dropping the now-stale ordered form are this
59 /// method's own writes rather than the caller's, so neither can be
60 /// forgotten and the bucket cannot announce a change while still handing
61 /// out the keys from before it. The revision moves on every call whether
62 /// or not the closure changed anything: an insert of an equal key is the
63 /// re-key case, and a cursor standing on the old key has to be told.
64 fn transition(&self, f: impl FnOnce(&mut BTreeSet<super::ScanKey>)) {
65 let mut entries = self.entries.lock();
66 f(&mut entries.keys);
67 entries.ordered = None;
68 self.revision
69 .fetch_add(1, std::sync::atomic::Ordering::Release);
70 }
71
72 /// The keys in order, and the revision they are the keys at.
73 fn snapshot(&self) -> (u64, Arc<[super::ScanKey]>) {
74 let mut entries = self.entries.lock();
75 if entries.ordered.is_none() {
76 let ordered: Arc<[super::ScanKey]> = entries.keys.iter().cloned().collect();
77 entries.ordered = Some(ordered);
78 }
79 (
80 // Read under the mutex, beside the keys it describes. A
81 // transition landing after this returns is one the cursor meets
82 // on its next step, which is where C meets it too.
83 self.revision.load(std::sync::atomic::Ordering::Relaxed),
84 entries.ordered.clone().expect("just materialised"),
85 )
86 }
87}
88
89pub(super) struct ScanIndex {
90 /// One bucket per scan list. Sized from the LOADED `menuScan`
91 /// ([`ScanList::count`]), not from a compile-time rate list, and built on
92 /// first use rather than at construction: the site's `menuScan.dbd` is
93 /// loaded after the database object exists, so sizing this eagerly would
94 /// freeze the menu before the loader could install one. C reaches the same
95 /// point by ordering — `dbLoadDatabase`, then `iocInit` → `initPeriodic`
96 /// sizes `papPeriodic`.
97 buckets: std::sync::OnceLock<Box<[ScanBucket]>>,
98 /// Cumulative over-runs per list — C `periodic_scan_list::overruns`
99 /// (`dbScan.c:95`), which `scanppl` prints beside the list it belongs to
100 /// (`dbScan.c:408-409`). It lives here for the same reason C puts it on
101 /// `periodic_scan_list`: one owner per rate holds both the list and its
102 /// over-run count, so the counter cannot drift away from the list it
103 /// counts. Only the periodic scan threads write it.
104 overruns: std::sync::OnceLock<Box<[std::sync::atomic::AtomicU64]>>,
105}
106
107impl ScanIndex {
108 pub(super) fn new() -> Self {
109 Self {
110 buckets: std::sync::OnceLock::new(),
111 overruns: std::sync::OnceLock::new(),
112 }
113 }
114
115 /// The bucket holding `list`'s records. Total — see [`ScanList::slot`].
116 fn bucket(&self, list: ScanList) -> &ScanBucket {
117 &self
118 .buckets
119 .get_or_init(|| (0..ScanList::count()).map(|_| ScanBucket::new()).collect())
120 .as_ref()[list.slot()]
121 }
122
123 /// This list's over-run counter — the same lazy sizing as [`Self::bucket`].
124 fn overrun(&self, list: ScanList) -> &std::sync::atomic::AtomicU64 {
125 &self
126 .overruns
127 .get_or_init(|| {
128 (0..ScanList::count())
129 .map(|_| std::sync::atomic::AtomicU64::new(0))
130 .collect()
131 })
132 .as_ref()[list.slot()]
133 }
134}
135
136impl PvDatabase {
137 /// C `addToList` (`dbScan.c:1074` at R7.0.10) — the ONE place a record
138 /// enters a scan bucket.
139 ///
140 /// C keeps it `static` so that `scanAdd` is the only way in; the port's
141 /// equivalent is [`ScanBucket::transition`], which this and
142 /// [`Self::delete_from_scan_list`] are the only callers of. It takes the
143 /// bucket lock itself, exactly as C takes `psl->lock`, and takes no
144 /// registration lock: L46 belongs to
145 /// whichever owner called — [`Self::update_scan_index`], which acquires it,
146 /// or `add_record`, which is already inside it. Both used to open-code the
147 /// bucket write instead, which is how a transition could bypass its owner.
148 ///
149 /// A SCAN that names no list (`Passive`, or an index outside `menuScan`)
150 /// keys no bucket — C `scanAdd` refuses the same two, the latter with
151 /// "scanAdd detected illegal SCAN value".
152 pub(super) fn add_to_scan_list(
153 &self,
154 scan: ScanType,
155 phas: i16,
156 record_type: &str,
157 load_order: u64,
158 name: &str,
159 ) {
160 let Some(list) = scan.scan_list() else {
161 return;
162 };
163 // Resolved once here, where a record enters the list, instead of once
164 // per record per sweep inside the process frame.
165 let handle = self
166 .get_record_no_resolve(name)
167 .map(|rec| std::sync::Arc::downgrade(&rec))
168 .unwrap_or_default();
169 self.inner.scan_index.bucket(list).transition(|keys| {
170 keys.insert(super::ScanKey::new(
171 phas,
172 record_type,
173 load_order,
174 name,
175 handle,
176 ));
177 });
178 }
179
180 /// C `deleteFromList` (`dbScan.c:1096` at R7.0.10) — the ONE place a
181 /// record leaves a scan bucket. Same ownership rules as
182 /// [`Self::add_to_scan_list`].
183 ///
184 /// Matches by record name alone: PHAS and load-order may be stale relative
185 /// to the entry actually present, and a stale secondary key would leave a
186 /// phantom entry behind.
187 pub(super) fn delete_from_scan_list(&self, scan: ScanType, name: &str) {
188 let Some(list) = scan.scan_list() else {
189 return;
190 };
191 self.inner
192 .scan_index
193 .bucket(list)
194 .transition(|keys| keys.retain(|k| k.name.as_ref() != name));
195 }
196
197 /// Update scan index when a record's SCAN or PHAS field changes.
198 ///
199 /// Takes `registration_mutex` so the read of
200 /// the records map (to verify the record still exists) and the
201 /// scan_index mutation are atomic vs. concurrent `remove_record`.
202 ///
203 /// The `new_scan` / `new_phas` parameters
204 /// the caller passes are advisory only. After acquiring the
205 /// mutex we read the LIVE record's current scan/phas and insert
206 /// based on those. Pre-fix a put-then-update sequence could
207 /// race a remove+re-add of the same name: the caller's
208 /// `new_scan` reflected the old (now-removed) record's value;
209 /// inserting that under the fresh record's name produced a
210 /// stale scan-index entry pointing at a wrong scan rate. The
211 /// live-read makes the index strictly reflect the record's
212 /// current state at insert time.
213 ///
214 /// **Synchronous.** It had exactly two suspension points and neither was a
215 /// real one: the `registration_mutex` acquisition (L46) and two
216 /// `scan_index` write acquisitions (L8b) — both awaited before step 4.
217 /// Both are blocking PI mutexes now, so the whole scan-index update is a
218 /// bounded critical
219 /// section — which is what lets it run *inside* the L1 record-gate window
220 /// without putting an `.await` there. C reaches `scanAdd`/`scanDelete`
221 /// (`dbScan.c:241-330`) the same way, from inside `dbPut` under
222 /// `dbScanLock`.
223 pub fn update_scan_index(
224 &self,
225 name: &str,
226 old_scan: ScanType,
227 _new_scan: ScanType,
228 old_phas: i16,
229 _new_phas: i16,
230 ) {
231 let _ = old_phas; // entry matched by name; PHAS not needed.
232 // The LIVE record's SCAN/PHAS are read under L46 so that the map
233 // check and the index mutation are one transaction against a
234 // concurrent `remove_record`. Record data is behind the record's
235 // lock set, which sits ABOVE L46, so the set is taken first — a
236 // re-entry for every caller, which holds it already (C reaches
237 // `scanAdd` from `dbPut` under `dbScanLock`) — and the map is
238 // re-read under the gate to prove the handle locked is the handle
239 // registered. A remove+re-add between the two reads is retried
240 // against the fresh record.
241 let (rec_arc, _record_gate, _gate) = loop {
242 let rec_arc = self.inner.records.read().get(name).cloned();
243 let record_gate = rec_arc.as_ref().map(|rec| self.lock_instance(rec));
244 let gate = self.lock_registration("update_scan_index");
245 let live = self.inner.records.read().get(name).cloned();
246 match (&rec_arc, &live) {
247 (Some(locked), Some(live)) if Arc::ptr_eq(locked, live) => {}
248 (None, None) => {}
249 _ => continue,
250 }
251 break (rec_arc, record_gate, gate);
252 };
253 // 1) Remove the OLD entry the caller knew about — even if
254 // remove_record already swept it.
255 self.delete_from_scan_list(old_scan, name);
256 // 2) Re-insert from the LIVE record's state. If concurrent
257 // remove+re-add replaced the Arc with a fresh one whose
258 // scan differs from the caller's `_new_scan`, we re-insert
259 // based on the fresh record's state. The fresh record's
260 // own `add_record` call also registered its scan index, so
261 // duplicate-insertion of the same (phas, name) pair into
262 // the same scan bucket is a no-op (`BTreeSet::insert`
263 // returns false on present key).
264 let Some(rec_arc) = rec_arc else {
265 return;
266 };
267 let (cur_scan, cur_phas, cur_type) = {
268 let inst = rec_arc.read();
269 (
270 inst.common.scan,
271 inst.common.phas,
272 inst.record.record_type(),
273 )
274 };
275 // Re-use the record's existing load-order sequence so the scan-index
276 // secondary key stays stable across SCAN/PHAS edits. A record loaded
277 // before should always scan before a later-loaded record at the same
278 // PHAS.
279 let seq = self.inner.load_order.load().get(name).copied().unwrap_or(0);
280 self.add_to_scan_list(cur_scan, cur_phas, cur_type, seq, name);
281 }
282
283 /// Count one over-run for `scan`'s list — C `ppsl->overruns++`
284 /// (`dbScan.c:827`). The periodic scan thread for that rate is the only
285 /// caller; a SCAN value naming no list has no counter to move.
286 pub(crate) fn record_scan_overrun(&self, scan: ScanType) {
287 if let Some(list) = scan.scan_list() {
288 self.inner
289 .scan_index
290 .overrun(list)
291 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
292 }
293 }
294
295 /// How many times `list`'s sweep has run past its deadline since boot —
296 /// what C's `scanppl` prints as `(%lu over-runs)` (`dbScan.c:408-409`).
297 pub(crate) fn scan_overruns(&self, list: crate::server::record::ScanList) -> u64 {
298 self.inner
299 .scan_index
300 .overrun(list)
301 .load(std::sync::atomic::Ordering::Relaxed)
302 }
303
304 /// A **snapshot** of a scan list, in C's order: PHAS, then DBD
305 /// record-type order, then `.db` load order — see `ScanKey`. The stable
306 /// same-PHAS FIFO is `addToList`; the order it is a FIFO over is
307 /// `buildScanLists`, and both are in the key.
308 ///
309 /// For reporting and counting only (`scanppl`, `dbstat`). A sweep that
310 /// PROCESSES the list must not use this: `dbProcess` can change the SCAN
311 /// field of an arbitrary number of records, so a snapshot processes
312 /// records the list no longer holds. `Self::scan_list_once` is the sweep.
313 ///
314 /// A SCAN value that names no list (`Passive`, or an index outside
315 /// `menuScan`) holds no records — there is no such bucket to look up.
316 pub async fn records_for_scan(&self, scan_type: ScanType) -> Vec<String> {
317 let Some(list) = scan_type.scan_list() else {
318 return Vec::new();
319 };
320 // One bucket's lock, held for the clone alone. C `scanList`
321 // (`dbScan.c:1007-1051`) also releases `psl->lock` around every
322 // `dbProcess` — but it releases it holding a cursor INTO the list and
323 // re-reads `ellNext` under the lock on the next step, so the two are
324 // not the same construction and this must not be read as parity with
325 // it. Detaching from the list is only sound because nothing here
326 // processes; [`Self::scan_list_once`] is what C's `scanList` is.
327 self.inner
328 .scan_index
329 .bucket(list)
330 .snapshot()
331 .1
332 .iter()
333 .map(|k| k.name.to_string())
334 .collect()
335 }
336
337 /// The scan key `name` would be inserted under RIGHT NOW — its live PHAS,
338 /// record type and load-order sequence. `None` if the record is gone.
339 fn live_scan_key(&self, name: &str) -> Option<super::ScanKey> {
340 let rec = self.get_record_no_resolve(name)?;
341 let (phas, record_type) = {
342 let inst = rec.read();
343 (inst.common.phas, inst.record.record_type())
344 };
345 let seq = self.inner.load_order.load().get(name).copied().unwrap_or(0);
346 Some(super::ScanKey::new(
347 phas,
348 record_type,
349 seq,
350 name,
351 std::sync::Arc::downgrade(&rec),
352 ))
353 }
354
355 /// A cursor over `list` as it stands at each step — see [`ScanCursor`].
356 pub(crate) fn scan_cursor(&self, list: ScanList) -> ScanCursor {
357 ScanCursor {
358 list,
359 snapshot: None,
360 next: 0,
361 revision: 0,
362 }
363 }
364
365 /// Sweep one scan list, processing every record still in it — C `scanList`
366 /// (`dbScan.c:998-1051`), the single owner of "walk a scan list and
367 /// process it". Every driver goes through here: the periodic threads, and
368 /// the event lists, whose `eventCallback` (`dbScan.c:459-465`) is a bare
369 /// `scanList` call.
370 pub(crate) async fn scan_list_once(&self, list: ScanList) {
371 let mut cursor = self.scan_cursor(list);
372 // One set for the whole sweep. `run_process_frame` owns the unwind and
373 // takes its own marker back out on every exit, so the set is empty
374 // again when a record's cascade returns — reusing it is the same set a
375 // fresh `HashSet::new()` would be, minus the table allocation each
376 // record was paying for its first insert.
377 let mut visited = crate::server::database::ProcStack::new();
378 while let Some((name, rec)) = cursor.next(self) {
379 let _ = match rec {
380 Some(rec) => self.process_record_with_links_resolved(name, rec, &mut visited),
381 None => self.process_record_with_links_sync(name, &mut visited),
382 };
383 debug_assert!(
384 visited.is_empty(),
385 "a returned process frame left its cycle marker behind"
386 );
387 }
388 }
389
390 /// Get all record names whose `PINI` is **exactly** `mode`.
391 ///
392 /// C matches the menu index with `!=` (`iocInit.c:598`
393 /// `if (precord->pini != pphase->pini) return;`), so each `menuPini`
394 /// choice selects a disjoint set of records driven by a *different* pass:
395 /// `YES` at `initialProcess()` (`iocInit.c:656`), `RUN`/`RUNNING`/`PAUSE`/
396 /// `PAUSED` from `piniProcessHook` (`iocInit.c:629-646`). A `PINI=RUN`
397 /// record must NOT be processed by the `YES` pass.
398 ///
399 /// Snapshot the records map under the outer
400 /// read lock, then drop it before fanning out per-record reads.
401 /// Pre-fix the outer `records.read()` lock was held across every
402 /// `rec.read().await` — under contention with a pending
403 /// `add_record` (which now takes the registration_mutex →
404 /// records.write()), startup could stall while every PINI
405 /// record was inspected serially.
406 pub async fn pini_records(&self, mode: PiniMode) -> Vec<String> {
407 let mut result = Vec::new();
408 for (name, rec) in self.records_in_load_order().await {
409 if rec.read().common.pini == mode.to_u16() as i16 {
410 result.push(name);
411 }
412 }
413 result
414 }
415
416 /// Every record, in database **load order** — the port's analogue of C's
417 /// `iterateRecords`, which walks the record-type / record-instance lists in
418 /// the order the `.db` declared them. That order is what makes two
419 /// same-`PHAS` records process deterministically; `load_order` is one of
420 /// the `scan_index` sort keys for the same reason (see `ScanKey`, which
421 /// also carries the record-type ordinal C's `iterateRecords` walks first —
422 /// this PINI sweep does not, and that is a separate gap).
423 ///
424 /// Snapshots the map under the records read lock and releases it before the
425 /// caller takes any per-record lock.
426 async fn records_in_load_order(&self) -> Vec<(String, std::sync::Arc<RecordCell>)> {
427 let snapshot: Vec<_> = {
428 let records = self.inner.records.read();
429 records
430 .iter()
431 .map(|(n, r)| (n.to_string(), r.clone()))
432 .collect()
433 };
434 let mut keyed: Vec<_> = {
435 let load_order = self.inner.load_order.load();
436 snapshot
437 .into_iter()
438 .map(|(name, rec)| {
439 (
440 load_order.get(name.as_str()).copied().unwrap_or(0),
441 name,
442 rec,
443 )
444 })
445 .collect()
446 };
447 keyed.sort_unstable_by(|a, b| (a.0, &a.1).cmp(&(b.0, &b.1)));
448 keyed
449 .into_iter()
450 .map(|(_seq, name, rec)| (name, rec))
451 .collect()
452 }
453
454 /// C `piniProcess` (`iocInit.c:608-627`) — process every record whose
455 /// `PINI` is exactly `mode`, **in ascending `PHAS` order**, each with its
456 /// full link chain.
457 ///
458 /// The single owner of "run a PINI pass": `initialProcess()` calls it with
459 /// [`PiniMode::Yes`], and the `initHook` lifecycle calls it with
460 /// [`PiniMode::Run`] / [`PiniMode::Running`]. Every driver goes through
461 /// here, so neither the pass selection nor the phase ordering can diverge
462 /// between them.
463 ///
464 /// The sweep is C's, not a pre-sorted list: each pass over the database
465 /// processes the records at the current phase and, while doing so, finds
466 /// the *next* lowest `PHAS` still ahead of it. C spells this out
467 /// (`iocInit.c:614-619`) — "PHAS fields can be changed at runtime, so we
468 /// have to look for the lowest value of PHAS each time" — so a record whose
469 /// phase is raised by an earlier PINI record's processing is still picked
470 /// up in the correct later pass, which a snapshot-and-sort would miss.
471 pub async fn pini_process(&self, mode: PiniMode) {
472 // C `dbScan.h:34-35`: MAX_PHASE = SHRT_MAX, MIN_PHASE = SHRT_MIN. The
473 // phase cursors are `int` (`phaseData_t`), so `MAX_PHASE + 1` — the
474 // "no further phase found" sentinel — does not overflow.
475 const MIN_PHASE: i32 = i16::MIN as i32;
476 const NO_NEXT_PHASE: i32 = i16::MAX as i32 + 1;
477
478 let mut next = MIN_PHASE;
479 loop {
480 let this = next;
481 next = NO_NEXT_PHASE;
482 // `doRecordPini` (`iocInit.c:592-606`) over the record list. PINI
483 // and PHAS are read at the moment the record is visited, not
484 // snapshotted up front, so a PHAS that an earlier record's
485 // processing changed is honoured — the reason C re-scans rather
486 // than sorting once.
487 // See `scan_list_once`: one set per sweep, emptied by each frame's
488 // own unwind.
489 let mut visited = crate::server::database::ProcStack::new();
490 for (name, rec) in self.records_in_load_order().await {
491 let (pini, phas) = {
492 let instance = rec.read();
493 (instance.common.pini, i32::from(instance.common.phas))
494 };
495 if pini != mode.to_u16() as i16 {
496 continue;
497 }
498 if phas == this {
499 let _ =
500 self.process_record_with_links_resolved(&name, rec.clone(), &mut visited);
501 debug_assert!(
502 visited.is_empty(),
503 "a returned process frame left its cycle marker behind"
504 );
505 } else if phas > this && phas < next {
506 next = phas;
507 }
508 }
509 if next == NO_NEXT_PHASE {
510 return;
511 }
512 }
513 }
514
515 /// Process all records with `SCAN=Event`, regardless of `EVNT`.
516 ///
517 /// Back-compat entry point for the iocsh `postEvent` command,
518 /// whose handler currently drops the numeric event argument.
519 /// Prefer [`Self::post_event_named`] for C-correct per-event
520 /// routing — see `dbScan.c:548-552` `post_event` →
521 /// `postEvent(pevent_list[event])`.
522 pub async fn post_event(&self) {
523 // C `postEvent` (`dbScan.c:536-539`): an event posted while the
524 // facility is not running queues nothing at all.
525 if !crate::server::scan::scan_is_running() {
526 return;
527 }
528 if let Some(list) = ScanType::Event.scan_list() {
529 self.scan_list_once(list).await;
530 }
531 }
532
533 /// Process only the `SCAN=Event` records whose `EVNT` resolves to
534 /// `event_name`. Mirrors C `dbScan.c` event routing: each
535 /// `event_list` (`eventNameToHandle`) holds exactly the records
536 /// whose `EVNT` matches, and `postEvent` walks only that list.
537 ///
538 /// Event-name matching follows `eventNameToHandle` (`dbScan.c:469`):
539 /// surrounding whitespace is trimmed, and a numeric string with an
540 /// integer part in `[1,255]` is normalised to its integer form so
541 /// `"5"`, `" 5 "` and `"5.0"` all name the same event.
542 pub async fn post_event_named(&self, event_name: &str) {
543 // C `postEvent` (`dbScan.c:536-539`), reached through
544 // `post_event`/`eventNameToHandle` — the same gate, applied before
545 // the name lookup because C applies it before touching the list.
546 if !crate::server::scan::scan_is_running() {
547 return;
548 }
549 let want = normalize_event_name(event_name);
550 if want.is_empty() {
551 // `eventNameToHandle` returns NULL for "0"/empty — no event.
552 return;
553 }
554 let Some(list) = ScanType::Event.scan_list() else {
555 return;
556 };
557 // Same live cursor as every other sweep: C keeps one `scan_list` per
558 // (event, priority) and `eventCallback` hands it to `scanList`
559 // (`dbScan.c:459-465`), so a record whose SCAN changes mid-sweep leaves
560 // the walk here exactly as it does on a periodic list. The port keeps
561 // one Event list and filters by EVNT at the cursor instead.
562 let mut cursor = self.scan_cursor(list);
563 // See `scan_list_once`: one set per sweep, emptied by each frame's own
564 // unwind.
565 let mut visited = crate::server::database::ProcStack::new();
566 while let Some((name, rec)) = cursor.next(self) {
567 // Read the record's EVNT and compare against the posted
568 // event name. Records that do not match are skipped — a
569 // record configured `EVNT=5` only fires on event 5.
570 let Some(rec) = rec.or_else(|| self.get_record(name)) else {
571 continue;
572 };
573 let evnt = rec.read().common.evnt.clone();
574 if normalize_event_name(&evnt) != want {
575 continue;
576 }
577 let _ = self.process_record_with_links_resolved(name, rec, &mut visited);
578 debug_assert!(
579 visited.is_empty(),
580 "a returned process frame left its cycle marker behind"
581 );
582 }
583 }
584}
585
586/// A cursor over a LIVE scan list — C `scanList`'s `pse` (`dbScan.c:998-1051`).
587///
588/// The invariant: **the sweep observes the list, it does not own a copy of it.**
589/// C re-reads `ellNext(&pse->node)` under `psl->lock` on every step and carries
590/// a cursor-repair walk that exists precisely because `dbProcess` can change
591/// the SCAN field of an arbitrary number of records mid-sweep. A snapshot taken
592/// once at the top of the tick processes records the list no longer holds.
593///
594/// The cursor does hold the list in a `snapshot`, and that does not weaken the
595/// invariant, because the bucket's revision is what the snapshot is read
596/// through: the revision moves on every transition ([`ScanBucket::transition`]),
597/// so while it still reads what the cursor holds, no record has entered, left
598/// or been re-keyed and the snapshot IS the live list. The moment it differs
599/// the snapshot is discarded and the place re-found. What that buys is the
600/// step: an index, against a bucket lock, an ordered-set lookup and a key
601/// clone per record per sweep to learn that a list nothing had touched still
602/// held what it held.
603///
604/// The port's list is an ordered set, not a linked list, so "my element left
605/// the list" has one answer instead of C's three: the next key strictly greater
606/// than where the cursor stood. That subsumes C's prev/next repair
607/// (`dbScan.c:1030-1044`) — and, because the position is a key rather than a
608/// pointer, it has no counterpart to C's "too many changes, wait till the next
609/// period" (`:1045-1048`), which is an artefact of losing the place rather than
610/// a scanning rule.
611///
612/// The step re-reads the last record's CURRENT key before advancing, so a
613/// record that moved within this list (its own processing changed its PHAS) is
614/// advanced from where it is now — C's `pse->pscan_list == psl` branch
615/// (`:1023-1029`).
616pub(crate) struct ScanCursor {
617 list: ScanList,
618 /// The list as it stood at [`Self::revision`]; `None` before the first
619 /// step, which is the one state in which there is no place to keep.
620 snapshot: Option<Arc<[super::ScanKey]>>,
621 /// Index into `snapshot` of the entry the next step hands out. The entry
622 /// before it is where the cursor stands.
623 next: usize,
624 /// The bucket revision `snapshot` was taken at.
625 revision: u64,
626}
627
628impl ScanCursor {
629 /// Where the cursor stands: the entry the last step handed out.
630 fn standing(&self) -> Option<&super::ScanKey> {
631 self.snapshot.as_ref()?.get(self.next.checked_sub(1)?)
632 }
633
634 /// Take the list again and re-find the place in it — C's cursor repair
635 /// (`dbScan.c:1023-1044`), now paid once per transition rather than once
636 /// per record.
637 ///
638 /// The record last handed out may have moved WITHIN this list, its own
639 /// processing having changed its PHAS, so its key is rebuilt from live
640 /// state before the resume point is looked up. `live_scan_key` reads the
641 /// records map and the record itself, so no bucket lock is held across it.
642 fn resync(&mut self, db: &PvDatabase) {
643 let was = self.standing().cloned();
644 let live = was.as_ref().and_then(|w| db.live_scan_key(&w.name));
645 let (revision, snapshot) = db.inner.scan_index.bucket(self.list).snapshot();
646 let resume = match live {
647 // The rebuilt key is the resume point only if THIS list is where
648 // the record landed; one whose SCAN moved it to another list
649 // resumes the sweep from where it stood.
650 Some(k) if snapshot.binary_search(&k).is_ok() => Some(k),
651 _ => was,
652 };
653 // The next key strictly greater than the resume point — C's
654 // `ellNext`, and the same answer for a record that left the list as
655 // for one that is still in it.
656 self.next = resume.map_or(0, |r| snapshot.partition_point(|k| *k <= r));
657 self.revision = revision;
658 self.snapshot = Some(snapshot);
659 }
660
661 /// The next record still in the list, or `None` at its end: the name, and
662 /// the instance the list holds beside it (`None` only for a key whose
663 /// record has since been dropped).
664 ///
665 /// Takes no lock while the list is unchanged, and the bucket lock for the
666 /// re-find alone when it is — never across processing, as C releases
667 /// `psl->lock` around every `dbProcess`.
668 #[allow(clippy::type_complexity)]
669 pub(crate) fn next(&mut self, db: &PvDatabase) -> Option<(&str, Option<Arc<RecordCell>>)> {
670 let bucket = db.inner.scan_index.bucket(self.list);
671 if self.snapshot.is_none()
672 || bucket.revision.load(std::sync::atomic::Ordering::Acquire) != self.revision
673 {
674 self.resync(db);
675 }
676 let key = self.snapshot.as_ref()?.get(self.next)?;
677 self.next += 1;
678 Some((&key.name, key.handle.upgrade()))
679 }
680}
681
682/// Normalise an EPICS event name for routing comparison.
683///
684/// Mirrors `dbScan.c::eventNameToHandle` (`dbScan.c:469-533`):
685/// * leading/trailing whitespace is stripped;
686/// * a string that parses as a number with an integer part in
687/// `[1,255]` is canonicalised to that integer's decimal form
688/// (so numeric events from calc records match symbolic "5");
689/// * `"0"` (and anything that resolves to event 0) becomes empty —
690/// C's `eventNameToHandle` returns NULL for event 0.
691pub(crate) fn normalize_event_name(name: &str) -> String {
692 let trimmed = name.trim();
693 if trimmed.is_empty() {
694 return String::new();
695 }
696 if let Ok(num) = trimmed.parse::<f64>() {
697 if num >= 0.0 && num < 256.0 {
698 let int = num as i64;
699 if int < 1 {
700 // event 0 → no event
701 return String::new();
702 }
703 return int.to_string();
704 }
705 // Numeric but outside [0,256): fall through to literal match.
706 }
707 trimmed.to_string()
708}
709
710#[cfg(test)]
711mod tests {
712 use super::PvDatabase;
713 use super::normalize_event_name;
714 use crate::server::record::ScanType;
715
716 /// The cycle guard belongs to the frame that inserted it, and a frame
717 /// that does not run must not insert one. The set is ONE per sweep
718 /// (`scan_list_once`), so a marker left behind by a name that is no
719 /// longer in the database reads as "already on this stack" for every
720 /// later entry in that sweep — silencing every forward link onto it.
721 #[test]
722 fn a_frame_that_finds_no_record_leaves_no_cycle_marker() {
723 let db = PvDatabase::new();
724 let mut visited = crate::server::database::ProcStack::new();
725
726 let result = db.process_record_with_links_sync("NO:SUCH:RECORD", &mut visited);
727
728 assert!(
729 result.is_err(),
730 "a name the database does not hold is an error"
731 );
732 assert!(
733 visited.is_empty(),
734 "the frame left its cycle marker behind: {visited:?}"
735 );
736 }
737
738 /// The ordering rule, stated as the thing that must hold rather than as
739 /// the record shape that once broke it: **`update_scan_index` takes L46
740 /// itself, so no caller may hold L46 when calling it.**
741 ///
742 /// A caller that breaks it used to park on itself forever, because
743 /// `PriorityInheritanceMutex` is not reentrant. The failure reached CI as
744 /// a 120-second timeout on whatever test happened to register a record —
745 /// a shape that reads as a load flake and hides which caller is at fault.
746 /// This pins the replacement: the violating call panics, immediately, and
747 /// names both ends.
748 ///
749 /// Deliberately expressed with a bare `lock_registration` + direct
750 /// `update_scan_index` pair and no record, no SIML and no SCAN field: the
751 /// rule belongs to the caller/owner contract, not to the one composition
752 /// (`add_loaded_record` → `rec_gbl_init_simm` → `apply_simm_scan_swap`)
753 /// that first exposed it. Any future tail added under a registration gate
754 /// trips this the same way.
755 #[test]
756 fn a_caller_holding_l46_cannot_reach_the_scan_index_owner() {
757 let db = PvDatabase::new();
758 let held = db.lock_registration("a_test_standing_in_for_a_registration_entry_point");
759
760 let violation = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
761 db.update_scan_index("ANY", ScanType::Passive, ScanType::SEC01, 0, 0);
762 }));
763
764 let payload = violation
765 .expect_err("holding L46 across update_scan_index must panic, not park the thread");
766 let msg = payload
767 .downcast_ref::<String>()
768 .map(String::as_str)
769 .or_else(|| payload.downcast_ref::<&str>().copied())
770 .unwrap_or("");
771 assert!(
772 msg.contains("not reentrant") && msg.contains("update_scan_index"),
773 "the panic must name the rule and the violating site, got: {msg}"
774 );
775 drop(held);
776 }
777
778 /// The other side of the same boundary — with no gate held, the owner
779 /// takes L46 itself and completes. Without this case the test above would
780 /// still pass if `update_scan_index` panicked unconditionally.
781 #[test]
782 fn the_scan_index_owner_takes_l46_itself_when_no_caller_holds_it() {
783 let db = PvDatabase::new();
784 db.update_scan_index("ANY", ScanType::Passive, ScanType::SEC01, 0, 0);
785 }
786
787 /// The gate is released on drop, including by unwinding out of a panic
788 /// between acquisitions. A leaked flag would make every later
789 /// registration on this thread panic and turn the tripwire into its own
790 /// outage.
791 #[test]
792 fn the_registration_gate_clears_on_drop_and_on_unwind() {
793 let db = PvDatabase::new();
794 drop(db.lock_registration("first"));
795 let _second = db.lock_registration("second");
796 drop(_second);
797
798 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
799 let _g = db.lock_registration("panics_while_held");
800 panic!("unwind with the gate live");
801 }));
802 drop(db.lock_registration("after_unwind"));
803 }
804
805 /// A cursor skips its repair walk for as long as the bucket revision it
806 /// carries still matches, so the revision has to move on every transition
807 /// that can invalidate a key it holds — including an insert of a key the
808 /// bucket already has, which is how a re-key that keeps the same PHAS
809 /// arrives.
810 #[test]
811 fn every_scan_bucket_transition_moves_the_revision() {
812 let db = PvDatabase::new();
813 let list = ScanType::SEC01.scan_list().expect("1 second names a list");
814 let revision = || {
815 db.inner
816 .scan_index
817 .bucket(list)
818 .revision
819 .load(std::sync::atomic::Ordering::Relaxed)
820 };
821
822 let empty = revision();
823 db.add_to_scan_list(ScanType::SEC01, 0, "calc", 0, "R:ONE");
824 let added = revision();
825 assert_ne!(added, empty, "an insert is a transition");
826
827 db.add_to_scan_list(ScanType::SEC01, 0, "calc", 0, "R:ONE");
828 let re_added = revision();
829 assert_ne!(
830 re_added, added,
831 "re-inserting a key the bucket already holds is a transition too"
832 );
833
834 db.delete_from_scan_list(ScanType::SEC01, "R:ONE");
835 assert_ne!(revision(), re_added, "a removal is a transition");
836 }
837
838 #[test]
839 fn event_name_numeric_normalisation() {
840 // Whitespace trimmed, numeric forms canonicalised.
841 assert_eq!(normalize_event_name(" 5 "), "5");
842 assert_eq!(normalize_event_name("5.0"), "5");
843 assert_eq!(normalize_event_name("5"), "5");
844 // Event 0 / empty → no event.
845 assert_eq!(normalize_event_name("0"), "");
846 assert_eq!(normalize_event_name(""), "");
847 assert_eq!(normalize_event_name(" "), "");
848 // Symbolic name preserved.
849 assert_eq!(normalize_event_name("myEvent"), "myEvent");
850 assert_eq!(normalize_event_name(" myEvent "), "myEvent");
851 // Numeric out of [0,256) is treated literally.
852 assert_eq!(normalize_event_name("999"), "999");
853 }
854}