subetha_cxc/
shared_async_pointer.rs

1//! `SharedAsyncPointer<T>` - cross-process lazy / speculative
2//! resolution wrapping a `SharedOnceCell<T>`.
3//!
4//! Three resolution strategies; all converge to a single canonical
5//! value in the underlying `SharedOnceCell`:
6//!
7//! | Strategy   | Race width | Failover                                |
8//! |------------|-----------|-----------------------------------------|
9//! | Resolved   | n/a       | n/a (value pre-set)                     |
10//! | Lazy       | 1         | first caller to set wins; others read   |
11//! | Speculative| N (workers)| first-publisher-wins; losers discard    |
12//!
13//! # The Speculative race
14//!
15//! `get_or_speculative(n, f)` spawns N worker threads (or in the
16//! cross-process variant, dispatches N Passes via the
17//! BackgroundScheduler). Each worker independently computes `f()`
18//! and CAS-attempts to publish the result via the underlying
19//! SharedOnceCell. The first to win the CAS becomes the canonical
20//! result; losers see the cell already filled and DISCARD their
21//! result.
22//!
23//! This is the architectural novelty: redundant cross-process
24//! compute with first-publisher-wins. No existing Rust async runtime
25//! provides this primitive. It's useful for:
26//!
27//! - Latency hedging: race 2-3 backend lookups, take the fastest
28//! - Survivability: race N solvers across processes; any one
29//!   surviving suffices
30//! - Fault-tolerant fetch: if one resolver dies, others continue
31//!
32//! Failover within 1 epoch: if a resolver dies mid-compute, the
33//! others are unaffected; the first survivor publishes. No
34//! coordinator needed - the CAS protocol IS the coordination.
35
36use std::path::Path;
37use std::sync::Arc;
38use std::thread;
39
40use crate::shared_once_cell::{SharedOnceCell, SharedOnceError};
41
42/// Strategy tag observed by the substrate.
43pub mod strategy {
44    pub const RESOLVED: u32 = 0;
45    pub const LAZY: u32 = 1;
46    pub const SPECULATIVE: u32 = 2;
47}
48
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum SharedAsyncError {
51    Once(SharedOnceError),
52    AllWorkersDied,
53}
54
55impl From<SharedOnceError> for SharedAsyncError {
56    fn from(e: SharedOnceError) -> Self { Self::Once(e) }
57}
58
59pub struct SharedAsyncPointer<T: Copy + Send + Sync + 'static> {
60    cell: Arc<SharedOnceCell<T>>,
61    header_sidecar: subetha_core::HandshakeHeader,
62    ring_sidecar: Box<subetha_core::ObservationRing>,
63}
64
65impl<T: Copy + Send + Sync + 'static>
66    subetha_sidecar::AdaptiveInstance for SharedAsyncPointer<T>
67{
68    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
69    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
70    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
71        Box::new(subetha_sidecar::NoMigrationPolicy)
72    }
73}
74
75impl<T: Copy + Send + Sync + 'static> SharedAsyncPointer<T> {
76    /// Direction signature of `SharedAsyncPointer<T>`. Engages the
77    /// `K_async` axis (future / async-state stored at slot for
78    /// cross-process await on a value that may not yet exist).
79    pub const SIGNATURE: subetha_core::AxisMask = subetha_core::AxisMask::from_axes(
80        &[subetha_core::Axis::Async],
81    );
82
83    /// Create a new shared async pointer backed by an MMF cell at
84    /// `path`. The cell starts EMPTY.
85    pub fn create(path: impl AsRef<Path>) -> Result<Self, SharedAsyncError> {
86        let cell = Arc::new(SharedOnceCell::create(path)?);
87        Ok(Self {
88            cell,
89            header_sidecar: subetha_core::HandshakeHeader::new(),
90            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
91        })
92    }
93
94    /// Open an existing shared async pointer.
95    pub fn open(path: impl AsRef<Path>) -> Result<Self, SharedAsyncError> {
96        let cell = Arc::new(SharedOnceCell::open(path)?);
97        Ok(Self {
98            cell,
99            header_sidecar: subetha_core::HandshakeHeader::new(),
100            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
101        })
102    }
103
104    /// True when the underlying cell is initialised.
105    pub fn is_resolved(&self) -> bool {
106        self.cell.is_initialized()
107    }
108
109    /// Non-blocking peek. Returns the canonical value if any.
110    pub fn try_get(&self) -> Option<T> {
111        let r = self.cell.get();
112        self.ring_sidecar.push_op(
113            crate::sidecar_ops::async_pointer::OP_TRY_GET,
114            if r.is_none() { 2 } else { 0 },
115        );
116        r
117    }
118
119    /// Pre-resolve by setting the value. Returns `true` if this
120    /// caller won the init race, `false` if the cell was already
121    /// initialised.
122    pub fn set_resolved(&self, value: T) -> bool {
123        self.cell.set(value)
124    }
125
126    /// Lazy resolution: if the cell is initialised, return its
127    /// value. Otherwise, the caller runs `f` once and attempts to
128    /// publish the result. If another concurrent caller wins the
129    /// publish race, the caller still returns the canonical value
130    /// (theirs may be discarded silently).
131    pub fn get_or_lazy<F>(&self, f: F) -> T
132    where F: FnOnce() -> T,
133    {
134        let was_resolved = self.cell.is_initialized();
135        if let Some(v) = self.cell.get() {
136            self.ring_sidecar
137                .push_op(crate::sidecar_ops::async_pointer::OP_GET_OR_FETCH, 0);
138            return v;
139        }
140        let computed = f();
141        if !self.cell.set(computed) {
142            // `set` reports false both when the cell is already published and
143            // when another caller holds it mid-init, where the value is not yet
144            // readable. Wait for the winner to publish before reading.
145            while !self.cell.is_initialized() {
146                std::hint::spin_loop();
147            }
148        }
149        let v = self.cell.get().expect("INITIALIZED after set or read-back");
150        self.ring_sidecar.push_op(
151            crate::sidecar_ops::async_pointer::OP_GET_OR_FETCH,
152            if was_resolved { 0 } else { 1 }, // cold-fetch path
153        );
154        v
155    }
156
157    /// Speculative resolution: spawn `n` worker threads that all
158    /// independently compute `f()` and race to publish the result.
159    /// First publisher wins; losers discard their results. Returns
160    /// the canonical value (the winner's).
161    ///
162    /// All N workers share the same `f` (closure must be Clone +
163    /// Send + Sync). Use `get_or_speculative_with` when each worker
164    /// needs a different closure (e.g., different backends).
165    pub fn get_or_speculative<F>(&self, n: usize, f: F) -> T
166    where F: Fn() -> T + Send + Sync + 'static + Clone,
167    {
168        if let Some(v) = self.cell.get() { return v; }
169        assert!(n >= 1, "speculative race needs at least 1 worker");
170        let mut handles = Vec::with_capacity(n);
171        for _ in 0..n {
172            let cell = self.cell.clone();
173            let f = f.clone();
174            handles.push(thread::spawn(move || {
175                // Short-circuit: if cell already filled (another
176                // worker won before we even started), skip the work.
177                if cell.is_initialized() { return; }
178                let v = f();
179                // Try to publish; if we lose, our v is silently dropped.
180                // Winner=true, loser=false; either way cell ends initialized.
181                cell.set(v);
182            }));
183        }
184        for h in handles { h.join().ok(); }
185        self.cell.get().expect("at least one worker should publish")
186    }
187
188    /// Speculative resolution with per-worker closures. Each closure
189    /// in `fs` is dispatched to one worker; first publisher wins.
190    pub fn get_or_speculative_with<I, F>(&self, fs: I) -> T
191    where I: IntoIterator<Item = F>, F: FnOnce() -> T + Send + 'static,
192    {
193        if let Some(v) = self.cell.get() { return v; }
194        let mut handles = vec![];
195        for f in fs {
196            let cell = self.cell.clone();
197            handles.push(thread::spawn(move || {
198                if cell.is_initialized() { return; }
199                let v = f();
200                // Winner=true, loser=false; either way cell ends initialized.
201                cell.set(v);
202            }));
203        }
204        assert!(!handles.is_empty(), "speculative race needs at least 1 closure");
205        for h in handles { h.join().ok(); }
206        self.cell.get().expect("at least one worker should publish")
207    }
208
209    /// Speculative resolution that tolerates worker panics: closures
210    /// that panic do not propagate; the race continues among
211    /// survivors. Returns `Err(AllWorkersDied)` if every worker
212    /// panicked AND no value was published.
213    pub fn get_or_speculative_resilient<F>(&self, n: usize, f: F) -> Result<T, SharedAsyncError>
214    where F: Fn() -> T + Send + Sync + 'static + Clone,
215    {
216        if let Some(v) = self.cell.get() { return Ok(v); }
217        assert!(n >= 1);
218        let mut handles = Vec::with_capacity(n);
219        for _ in 0..n {
220            let cell = self.cell.clone();
221            let f = f.clone();
222            handles.push(thread::spawn(move || {
223                if cell.is_initialized() { return; }
224                let v = f();
225                // Winner=true, loser=false; either way cell ends initialized.
226                cell.set(v);
227            }));
228        }
229        // Wait for all; ignore panics.
230        for h in handles { h.join().ok(); }
231        self.cell.get().ok_or(SharedAsyncError::AllWorkersDied)
232    }
233
234    /// Sync the underlying cell to disk.
235    pub fn flush(&self) -> Result<(), SharedAsyncError> {
236        Ok(self.cell.flush()?)
237    }
238
239    /// Non-blocking flush: schedules a writeback via the OS.
240    /// Note: Windows is only partially async (sync to page cache,
241    /// not to disk).
242    pub fn flush_async(&self) -> Result<(), SharedAsyncError> {
243        Ok(self.cell.flush_async()?)
244    }
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250    use std::sync::atomic::{AtomicU32, Ordering};
251    use std::time::Duration;
252
253    fn tmp(name: &str) -> std::path::PathBuf {
254        let mut p = std::env::temp_dir();
255        let pid = std::process::id();
256        p.push(format!("subetha-async-{name}-{pid}.bin"));
257        p
258    }
259
260    #[test]
261    fn resolved_strategy_returns_immediately() {
262        let p = tmp("resolved");
263        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
264        assert!(!sap.is_resolved());
265        sap.set_resolved(42);
266        assert!(sap.is_resolved());
267        assert_eq!(sap.try_get(), Some(42));
268        std::fs::remove_file(&p).ok();
269    }
270
271    #[test]
272    fn lazy_runs_closure_exactly_once_across_threads() {
273        let p = tmp("lazy-once");
274        let sap: Arc<SharedAsyncPointer<u64>>
275            = Arc::new(SharedAsyncPointer::create(&p).unwrap());
276        let counter = Arc::new(AtomicU32::new(0));
277        let mut handles = vec![];
278        for _ in 0..8 {
279            let sap = sap.clone();
280            let counter = counter.clone();
281            handles.push(thread::spawn(move || {
282                sap.get_or_lazy(|| {
283                    counter.fetch_add(1, Ordering::AcqRel);
284                    thread::sleep(Duration::from_millis(2));
285                    777
286                })
287            }));
288        }
289        let results: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect();
290        // All threads return the canonical value.
291        assert!(results.iter().all(|v| *v == 777));
292        // Closure ran at most once (in practice exactly once due to set CAS).
293        let runs = counter.load(Ordering::Acquire);
294        assert!((1..=8).contains(&runs),
295                "closure runs should be between 1 and 8 (one per non-fast-pathed thread); got {runs}");
296        // Actually we expect exactly 1 because get_or_lazy() ALWAYS runs
297        // the closure on the first call; subsequent threads see the
298        // cell filled before running. With 8 threads racing the
299        // closure could run multiple times (if all check is_init
300        // before any has filled), but at most once per thread. The
301        // CAS publish ensures only one value is canonical.
302        std::fs::remove_file(&p).ok();
303    }
304
305    #[test]
306    fn speculative_race_returns_one_published_result() {
307        let p = tmp("speculative");
308        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
309        let runs = Arc::new(AtomicU32::new(0));
310        let runs_clone = runs.clone();
311        let result = sap.get_or_speculative(4, move || {
312            runs_clone.fetch_add(1, Ordering::AcqRel);
313            // small sleep so workers actually overlap
314            thread::sleep(Duration::from_millis(5));
315            123u64
316        });
317        assert_eq!(result, 123);
318        assert!(runs.load(Ordering::Acquire) >= 1,
319                "at least one worker must run");
320        // is_resolved must be true after race
321        assert!(sap.is_resolved());
322        std::fs::remove_file(&p).ok();
323    }
324
325    #[test]
326    fn speculative_first_finisher_wins() {
327        let p = tmp("first-wins");
328        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
329        // Two closures: one fast (returns 100), one slow (returns 999).
330        // Fast should win the publish race.
331        let result = sap.get_or_speculative_with([
332            Box::new(|| {
333                thread::sleep(Duration::from_millis(100));
334                999u64
335            }) as Box<dyn FnOnce() -> u64 + Send>,
336            Box::new(|| {
337                thread::sleep(Duration::from_millis(2));
338                100u64
339            }) as Box<dyn FnOnce() -> u64 + Send>,
340        ]);
341        assert_eq!(result, 100, "fast closure should win");
342        std::fs::remove_file(&p).ok();
343    }
344
345    #[test]
346    fn speculative_resilient_tolerates_panicking_workers() {
347        let p = tmp("resilient");
348        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
349        let attempt = Arc::new(AtomicU32::new(0));
350        let attempt_clone = attempt.clone();
351        // Closure panics on even attempts, succeeds on odd.
352        let result = sap.get_or_speculative_resilient(8, move || {
353            let n = attempt_clone.fetch_add(1, Ordering::AcqRel);
354            if n.is_multiple_of(2) {
355                panic!("simulated worker death on attempt {n}");
356            }
357            42u64
358        }).expect("at least one survivor publishes");
359        assert_eq!(result, 42);
360        std::fs::remove_file(&p).ok();
361    }
362
363    #[test]
364    fn second_call_after_resolution_returns_cached_value() {
365        let p = tmp("cached");
366        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
367        let _r1 = sap.get_or_lazy(|| 5);
368        // Second call must NOT run a new closure; verify by ensuring
369        // the closure body would change the value.
370        let r2 = sap.get_or_lazy(|| panic!("must not run after resolution"));
371        assert_eq!(r2, 5);
372        std::fs::remove_file(&p).ok();
373    }
374
375    #[test]
376    fn cross_handle_speculative_race_shares_one_winner() {
377        let p = tmp("cross-handle-spec");
378        let sap_a = SharedAsyncPointer::<u64>::create(&p).unwrap();
379        let sap_b = SharedAsyncPointer::<u64>::open(&p).unwrap();
380        // Process A speculatively resolves.
381        let r_a = sap_a.get_or_speculative(2, || 9999u64);
382        assert_eq!(r_a, 9999);
383        // Process B sees the same value without running anything.
384        let r_b = sap_b.get_or_lazy(|| panic!("must not run on already-resolved cell"));
385        assert_eq!(r_b, 9999);
386        std::fs::remove_file(&p).ok();
387    }
388
389    #[test]
390    fn try_get_does_not_force_resolution() {
391        let p = tmp("try-get");
392        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
393        assert_eq!(sap.try_get(), None);
394        assert!(!sap.is_resolved());
395        sap.set_resolved(100);
396        assert_eq!(sap.try_get(), Some(100));
397        std::fs::remove_file(&p).ok();
398    }
399
400    #[test]
401    fn speculative_with_one_worker_is_just_lazy() {
402        let p = tmp("spec-1");
403        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
404        let runs = Arc::new(AtomicU32::new(0));
405        let runs_clone = runs.clone();
406        let r = sap.get_or_speculative(1, move || {
407            runs_clone.fetch_add(1, Ordering::AcqRel);
408            17u64
409        });
410        assert_eq!(r, 17);
411        assert_eq!(runs.load(Ordering::Acquire), 1);
412        std::fs::remove_file(&p).ok();
413    }
414
415    /// The signature catalog lets `MmfDispatcher` route by
416    /// `provided.satisfies(required)` containment, which a pointer
417    /// declaring no axis at all would silently drop out of.
418    #[test]
419    fn signature_engages_the_async_axis() {
420        let sig = SharedAsyncPointer::<u8>::SIGNATURE;
421        assert_ne!(sig, subetha_core::AxisMask::EMPTY);
422        assert!(sig.contains(subetha_core::Axis::Async));
423    }
424}
subetha_cxc/shared_async_pointer.rs

subetha_cxc/
shared_async_pointer.rs
