flowlog_runtime/
intern.rs

1//! Thread-safe string interning via `lasso::ThreadedRodeo`.
2
3use std::sync::LazyLock;
4use std::sync::OnceLock;
5use std::thread;
6
7use lasso::Key;
8use lasso::Spur;
9use lasso::ThreadedRodeo;
10use rustc_hash::FxBuildHasher;
11
12/// Global string interner shared across all FlowLog engines in the process.
13///
14/// Uses `FxBuildHasher` instead of lasso's default SipHash: interner keys are
15/// program-controlled (`.dl` literals + input facts), not adversarial, so
16/// SipHash's HashDoS resistance is pure per-byte overhead on every intern and
17/// resolve.
18///
19/// **Limitation**: this is a process-local pool. In a distributed DD
20/// deployment (multiple machines), each process gets its own independent
21/// `INTERNER`, so `Spur` values are NOT comparable across machines.
22/// Distributed support would require a global interning protocol or
23/// switching back to `String`-keyed collections.
24pub static INTERNER: LazyLock<ThreadedRodeo<Spur, FxBuildHasher>> =
25    LazyLock::new(|| ThreadedRodeo::with_hasher(FxBuildHasher));
26
27const MAX_RETRIES: usize = 1024;
28
29/// Intern a string, returning its [`Spur`] handle.
30#[inline(always)]
31pub fn intern(s: &str) -> Spur {
32    for _ in 0..MAX_RETRIES {
33        match INTERNER.try_get_or_intern(s) {
34            Ok(key) => return key,
35            Err(_) => thread::yield_now(),
36        }
37    }
38    panic!("string interner failed after {MAX_RETRIES} attempts for {s:?}");
39}
40
41/// Resolve a [`Spur`] handle back to a `&'static str`.
42#[inline(always)]
43pub fn resolve(key: Spur) -> &'static str {
44    INTERNER.resolve(&key)
45}
46
47/// Flat snapshot of the interner (`Spur::into_usize()` → string) used for
48/// O(1) resolution at output/drain time. `Spur` keys are dense in
49/// `[0, len)`, so a plain `Vec` index replaces the concurrent
50/// [`ThreadedRodeo::resolve`] path (which hashes the key and takes a
51/// `DashMap` read lock on every call).
52static RESOLVED: OnceLock<Box<[&'static str]>> = OnceLock::new();
53
54/// Build the flat snapshot from the current interner contents.
55///
56/// `INTERNER` is borrowed from a `static`, so its strings are genuinely
57/// `'static`; the dense `Spur` keying lets us address them by index.
58fn build_snapshot() -> Box<[&'static str]> {
59    let mut table: Vec<&'static str> = vec![""; INTERNER.len()];
60    for (key, string) in INTERNER.iter() {
61        table[key.into_usize()] = string;
62    }
63    table.into_boxed_slice()
64}
65
66/// Resolve a [`Spur`] at output time via a flat index instead of the
67/// concurrent `DashMap` path taken by [`resolve`].
68///
69/// Built lazily on first use. Output runs after fixpoint, so the snapshot is
70/// complete in batch mode. Keys interned later (e.g. new epochs in incremental
71/// mode) fall outside its range and fall back to [`resolve`], staying correct
72/// without a rebuild.
73#[inline]
74pub fn resolve_out(key: Spur) -> &'static str {
75    let table = RESOLVED.get_or_init(build_snapshot);
76    match table.get(key.into_usize()) {
77        Some(&string) => string,
78        None => resolve(key),
79    }
80}
flowlog_runtime/intern.rs

flowlog_runtime/
intern.rs
