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        self.cell.set(computed);
142        let v = self.cell.get().expect("INITIALIZED after set or read-back");
143        self.ring_sidecar.push_op(
144            crate::sidecar_ops::async_pointer::OP_GET_OR_FETCH,
145            if was_resolved { 0 } else { 1 }, // cold-fetch path
146        );
147        v
148    }
149
150    /// Speculative resolution: spawn `n` worker threads that all
151    /// independently compute `f()` and race to publish the result.
152    /// First publisher wins; losers discard their results. Returns
153    /// the canonical value (the winner's).
154    ///
155    /// All N workers share the same `f` (closure must be Clone +
156    /// Send + Sync). Use `get_or_speculative_with` when each worker
157    /// needs a different closure (e.g., different backends).
158    pub fn get_or_speculative<F>(&self, n: usize, f: F) -> T
159    where F: Fn() -> T + Send + Sync + 'static + Clone,
160    {
161        if let Some(v) = self.cell.get() { return v; }
162        assert!(n >= 1, "speculative race needs at least 1 worker");
163        let mut handles = Vec::with_capacity(n);
164        for _ in 0..n {
165            let cell = self.cell.clone();
166            let f = f.clone();
167            handles.push(thread::spawn(move || {
168                // Short-circuit: if cell already filled (another
169                // worker won before we even started), skip the work.
170                if cell.is_initialized() { return; }
171                let v = f();
172                // Try to publish; if we lose, our v is silently dropped.
173                // Winner=true, loser=false; either way cell ends initialized.
174                cell.set(v);
175            }));
176        }
177        for h in handles { h.join().ok(); }
178        self.cell.get().expect("at least one worker should publish")
179    }
180
181    /// Speculative resolution with per-worker closures. Each closure
182    /// in `fs` is dispatched to one worker; first publisher wins.
183    pub fn get_or_speculative_with<I, F>(&self, fs: I) -> T
184    where I: IntoIterator<Item = F>, F: FnOnce() -> T + Send + 'static,
185    {
186        if let Some(v) = self.cell.get() { return v; }
187        let mut handles = vec![];
188        for f in fs {
189            let cell = self.cell.clone();
190            handles.push(thread::spawn(move || {
191                if cell.is_initialized() { return; }
192                let v = f();
193                // Winner=true, loser=false; either way cell ends initialized.
194                cell.set(v);
195            }));
196        }
197        assert!(!handles.is_empty(), "speculative race needs at least 1 closure");
198        for h in handles { h.join().ok(); }
199        self.cell.get().expect("at least one worker should publish")
200    }
201
202    /// Speculative resolution that tolerates worker panics: closures
203    /// that panic do not propagate; the race continues among
204    /// survivors. Returns `Err(AllWorkersDied)` if every worker
205    /// panicked AND no value was published.
206    pub fn get_or_speculative_resilient<F>(&self, n: usize, f: F) -> Result<T, SharedAsyncError>
207    where F: Fn() -> T + Send + Sync + 'static + Clone,
208    {
209        if let Some(v) = self.cell.get() { return Ok(v); }
210        assert!(n >= 1);
211        let mut handles = Vec::with_capacity(n);
212        for _ in 0..n {
213            let cell = self.cell.clone();
214            let f = f.clone();
215            handles.push(thread::spawn(move || {
216                if cell.is_initialized() { return; }
217                let v = f();
218                // Winner=true, loser=false; either way cell ends initialized.
219                cell.set(v);
220            }));
221        }
222        // Wait for all; ignore panics.
223        for h in handles { h.join().ok(); }
224        self.cell.get().ok_or(SharedAsyncError::AllWorkersDied)
225    }
226
227    /// Sync the underlying cell to disk.
228    pub fn flush(&self) -> Result<(), SharedAsyncError> {
229        Ok(self.cell.flush()?)
230    }
231
232    /// Non-blocking flush: schedules a writeback via the OS.
233    /// Note: Windows is only partially async (sync to page cache,
234    /// not to disk).
235    pub fn flush_async(&self) -> Result<(), SharedAsyncError> {
236        Ok(self.cell.flush_async()?)
237    }
238}
239
240#[cfg(test)]
241mod tests {
242    use super::*;
243    use std::sync::atomic::{AtomicU32, Ordering};
244    use std::time::Duration;
245
246    fn tmp(name: &str) -> std::path::PathBuf {
247        let mut p = std::env::temp_dir();
248        let pid = std::process::id();
249        p.push(format!("subetha-async-{name}-{pid}.bin"));
250        p
251    }
252
253    #[test]
254    fn resolved_strategy_returns_immediately() {
255        let p = tmp("resolved");
256        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
257        assert!(!sap.is_resolved());
258        sap.set_resolved(42);
259        assert!(sap.is_resolved());
260        assert_eq!(sap.try_get(), Some(42));
261        std::fs::remove_file(&p).ok();
262    }
263
264    #[test]
265    fn lazy_runs_closure_exactly_once_across_threads() {
266        let p = tmp("lazy-once");
267        let sap: Arc<SharedAsyncPointer<u64>>
268            = Arc::new(SharedAsyncPointer::create(&p).unwrap());
269        let counter = Arc::new(AtomicU32::new(0));
270        let mut handles = vec![];
271        for _ in 0..8 {
272            let sap = sap.clone();
273            let counter = counter.clone();
274            handles.push(thread::spawn(move || {
275                sap.get_or_lazy(|| {
276                    counter.fetch_add(1, Ordering::AcqRel);
277                    thread::sleep(Duration::from_millis(2));
278                    777
279                })
280            }));
281        }
282        let results: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect();
283        // All threads return the canonical value.
284        assert!(results.iter().all(|v| *v == 777));
285        // Closure ran at most once (in practice exactly once due to set CAS).
286        let runs = counter.load(Ordering::Acquire);
287        assert!((1..=8).contains(&runs),
288                "closure runs should be between 1 and 8 (one per non-fast-pathed thread); got {runs}");
289        // Actually we expect exactly 1 because get_or_lazy() ALWAYS runs
290        // the closure on the first call; subsequent threads see the
291        // cell filled before running. With 8 threads racing the
292        // closure could run multiple times (if all check is_init
293        // before any has filled), but at most once per thread. The
294        // CAS publish ensures only one value is canonical.
295        std::fs::remove_file(&p).ok();
296    }
297
298    #[test]
299    fn speculative_race_returns_one_published_result() {
300        let p = tmp("speculative");
301        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
302        let runs = Arc::new(AtomicU32::new(0));
303        let runs_clone = runs.clone();
304        let result = sap.get_or_speculative(4, move || {
305            runs_clone.fetch_add(1, Ordering::AcqRel);
306            // small sleep so workers actually overlap
307            thread::sleep(Duration::from_millis(5));
308            123u64
309        });
310        assert_eq!(result, 123);
311        assert!(runs.load(Ordering::Acquire) >= 1,
312                "at least one worker must run");
313        // is_resolved must be true after race
314        assert!(sap.is_resolved());
315        std::fs::remove_file(&p).ok();
316    }
317
318    #[test]
319    fn speculative_first_finisher_wins() {
320        let p = tmp("first-wins");
321        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
322        // Two closures: one fast (returns 100), one slow (returns 999).
323        // Fast should win the publish race.
324        let result = sap.get_or_speculative_with([
325            Box::new(|| {
326                thread::sleep(Duration::from_millis(100));
327                999u64
328            }) as Box<dyn FnOnce() -> u64 + Send>,
329            Box::new(|| {
330                thread::sleep(Duration::from_millis(2));
331                100u64
332            }) as Box<dyn FnOnce() -> u64 + Send>,
333        ]);
334        assert_eq!(result, 100, "fast closure should win");
335        std::fs::remove_file(&p).ok();
336    }
337
338    #[test]
339    fn speculative_resilient_tolerates_panicking_workers() {
340        let p = tmp("resilient");
341        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
342        let attempt = Arc::new(AtomicU32::new(0));
343        let attempt_clone = attempt.clone();
344        // Closure panics on even attempts, succeeds on odd.
345        let result = sap.get_or_speculative_resilient(8, move || {
346            let n = attempt_clone.fetch_add(1, Ordering::AcqRel);
347            if n.is_multiple_of(2) {
348                panic!("simulated worker death on attempt {n}");
349            }
350            42u64
351        }).expect("at least one survivor publishes");
352        assert_eq!(result, 42);
353        std::fs::remove_file(&p).ok();
354    }
355
356    #[test]
357    fn second_call_after_resolution_returns_cached_value() {
358        let p = tmp("cached");
359        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
360        let _r1 = sap.get_or_lazy(|| 5);
361        // Second call must NOT run a new closure; verify by ensuring
362        // the closure body would change the value.
363        let r2 = sap.get_or_lazy(|| panic!("must not run after resolution"));
364        assert_eq!(r2, 5);
365        std::fs::remove_file(&p).ok();
366    }
367
368    #[test]
369    fn cross_handle_speculative_race_shares_one_winner() {
370        let p = tmp("cross-handle-spec");
371        let sap_a = SharedAsyncPointer::<u64>::create(&p).unwrap();
372        let sap_b = SharedAsyncPointer::<u64>::open(&p).unwrap();
373        // Process A speculatively resolves.
374        let r_a = sap_a.get_or_speculative(2, || 9999u64);
375        assert_eq!(r_a, 9999);
376        // Process B sees the same value without running anything.
377        let r_b = sap_b.get_or_lazy(|| panic!("must not run on already-resolved cell"));
378        assert_eq!(r_b, 9999);
379        std::fs::remove_file(&p).ok();
380    }
381
382    #[test]
383    fn try_get_does_not_force_resolution() {
384        let p = tmp("try-get");
385        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
386        assert_eq!(sap.try_get(), None);
387        assert!(!sap.is_resolved());
388        sap.set_resolved(100);
389        assert_eq!(sap.try_get(), Some(100));
390        std::fs::remove_file(&p).ok();
391    }
392
393    #[test]
394    fn speculative_with_one_worker_is_just_lazy() {
395        let p = tmp("spec-1");
396        let sap: SharedAsyncPointer<u64> = SharedAsyncPointer::create(&p).unwrap();
397        let runs = Arc::new(AtomicU32::new(0));
398        let runs_clone = runs.clone();
399        let r = sap.get_or_speculative(1, move || {
400            runs_clone.fetch_add(1, Ordering::AcqRel);
401            17u64
402        });
403        assert_eq!(r, 17);
404        assert_eq!(runs.load(Ordering::Acquire), 1);
405        std::fs::remove_file(&p).ok();
406    }
407}
subetha_cxc/shared_async_pointer.rs

subetha_cxc/
shared_async_pointer.rs
