subetha_cxc/
shared_graph.rs

1//! `SharedGraph<N, E>` - cross-process directed graph with
2//! arbitrary out-degree.
3//!
4//! Nodes carry `N` values; edges carry `E` values + destination
5//! index. Adjacency stored as per-node linked lists of edges (each
6//! edge has a `next_in_src_list` link to the next edge from the
7//! same source).
8//!
9//! # Files
10//!
11//! - `<base>.nodes.bin` - `SharedRegion<GraphNode<N>>`
12//! - `<base>.edges.bin` - `SharedRegion<GraphEdge<E>>`
13//!
14//! # Concurrency
15//!
16//! SINGLE-WRITER, MULTI-READER. Reads (neighbors, node_value,
17//! edge_value, iter) are lock-free. Writes (add_node, add_edge,
18//! remove_edge) require external serialisation.
19//!
20//! # Safety
21//!
22//! - Bounded capacity at create (both regions).
23//! - SharedRegion's ABA-safe free list backs slot reuse.
24//! - No spin loops, no Drop guards, no atomic underflow risk.
25
26use std::marker::PhantomData;
27use std::path::{Path, PathBuf};
28
29use crate::shared_region::{OffsetPtr, RegionError, SharedRegion};
30
31pub const NIL_INDEX: u32 = u32::MAX;
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
34#[repr(C)]
35pub struct NodeIndex<N> {
36    pub index: u32,
37    _phantom: PhantomData<N>,
38}
39
40impl<N> NodeIndex<N> {
41    pub const NIL: Self = Self { index: NIL_INDEX, _phantom: PhantomData };
42    #[inline]
43    pub fn new(index: u32) -> Self { Self { index, _phantom: PhantomData } }
44    #[inline]
45    pub fn is_nil(self) -> bool { self.index == NIL_INDEX }
46}
47
48#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
49#[repr(C)]
50pub struct EdgeIndex<E> {
51    pub index: u32,
52    _phantom: PhantomData<E>,
53}
54
55impl<E> EdgeIndex<E> {
56    pub const NIL: Self = Self { index: NIL_INDEX, _phantom: PhantomData };
57    #[inline]
58    pub fn new(index: u32) -> Self { Self { index, _phantom: PhantomData } }
59    #[inline]
60    pub fn is_nil(self) -> bool { self.index == NIL_INDEX }
61}
62
63#[repr(C)]
64#[derive(Clone, Copy)]
65pub struct GraphNode<N: Copy + Default + 'static> {
66    pub value: N,
67    pub first_out_edge: u32,
68    pub n_out_edges: u32,
69}
70
71#[repr(C)]
72#[derive(Clone, Copy)]
73pub struct GraphEdge<E: Copy + Default + 'static> {
74    pub value: E,
75    pub dst: u32,
76    pub next_in_src_list: u32,
77}
78
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
80pub enum GraphError {
81    Region(RegionError),
82    InvalidNode,
83    InvalidEdge,
84    IoError(std::io::ErrorKind),
85}
86
87impl From<RegionError> for GraphError {
88    fn from(e: RegionError) -> Self { Self::Region(e) }
89}
90impl From<std::io::Error> for GraphError {
91    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
92}
93
94fn nodes_path(base: &Path) -> PathBuf {
95    let mut p = base.to_path_buf();
96    let stem = p.file_name().unwrap().to_string_lossy().to_string();
97    p.set_file_name(format!("{stem}.nodes.bin"));
98    p
99}
100fn edges_path(base: &Path) -> PathBuf {
101    let mut p = base.to_path_buf();
102    let stem = p.file_name().unwrap().to_string_lossy().to_string();
103    p.set_file_name(format!("{stem}.edges.bin"));
104    p
105}
106
107pub struct SharedGraph<
108    N: Copy + Default + 'static,
109    E: Copy + Default + 'static,
110> {
111    nodes: SharedRegion<GraphNode<N>>,
112    edges: SharedRegion<GraphEdge<E>>,
113    _phantom: PhantomData<(N, E)>,
114    header_sidecar: subetha_core::HandshakeHeader,
115    ring_sidecar: Box<subetha_core::ObservationRing>,
116}
117
118impl<
119    N: Copy + Default + Send + Sync + 'static,
120    E: Copy + Default + Send + Sync + 'static,
121> subetha_sidecar::AdaptiveInstance for SharedGraph<N, E> {
122    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
123    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
124    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
125        Box::new(subetha_sidecar::NoMigrationPolicy)
126    }
127}
128
129impl<N: Copy + Default + 'static, E: Copy + Default + 'static>
130    SharedGraph<N, E>
131{
132    pub fn create(
133        base_path: impl AsRef<Path>,
134        max_nodes: usize,
135        max_edges: usize,
136    ) -> Result<Self, GraphError> {
137        let base = base_path.as_ref();
138        let nodes = SharedRegion::create(nodes_path(base), max_nodes)?;
139        let edges = SharedRegion::create(edges_path(base), max_edges)?;
140        Ok(Self {
141            nodes, edges, _phantom: PhantomData,
142            header_sidecar: subetha_core::HandshakeHeader::new(),
143            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
144        })
145    }
146
147    pub fn open(
148        base_path: impl AsRef<Path>,
149        max_nodes: usize,
150        max_edges: usize,
151    ) -> Result<Self, GraphError> {
152        let base = base_path.as_ref();
153        let nodes = SharedRegion::open(nodes_path(base), max_nodes)?;
154        let edges = SharedRegion::open(edges_path(base), max_edges)?;
155        Ok(Self {
156            nodes, edges, _phantom: PhantomData,
157            header_sidecar: subetha_core::HandshakeHeader::new(),
158            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
159        })
160    }
161
162    pub fn node_count(&self) -> usize { self.nodes.len() }
163    pub fn edge_count(&self) -> usize { self.edges.len() }
164    pub fn max_nodes(&self) -> usize { self.nodes.capacity() }
165    pub fn max_edges(&self) -> usize { self.edges.capacity() }
166
167    /// Add a new node with `value`. Returns its index.
168    pub fn add_node(&self, value: N) -> Result<NodeIndex<N>, GraphError> {
169        let r = self.nodes.allocate(GraphNode {
170            value,
171            first_out_edge: NIL_INDEX,
172            n_out_edges: 0,
173        });
174        self.ring_sidecar.push_op(
175            crate::sidecar_ops::graph::OP_ADD_NODE,
176            if r.is_err() { 1 } else { 0 },
177        );
178        Ok(NodeIndex::new(r?.index))
179    }
180
181    /// Add an edge from `src` to `dst` carrying `value`. Returns
182    /// its index. Single-writer per src node (the linked-list head
183    /// update isn't synchronised internally).
184    pub fn add_edge(
185        &self, src: NodeIndex<N>, dst: NodeIndex<N>, value: E,
186    ) -> Result<EdgeIndex<E>, GraphError> {
187        let r = self.add_edge_inner(src, dst, value);
188        self.ring_sidecar.push_op(
189            crate::sidecar_ops::graph::OP_ADD_EDGE,
190            if r.is_err() { 1 } else { 0 },
191        );
192        r
193    }
194
195    fn add_edge_inner(
196        &self, src: NodeIndex<N>, dst: NodeIndex<N>, value: E,
197    ) -> Result<EdgeIndex<E>, GraphError> {
198        if src.is_nil() || dst.is_nil()
199            || src.index as usize >= self.nodes.capacity()
200            || dst.index as usize >= self.nodes.capacity()
201        {
202            return Err(GraphError::InvalidNode);
203        }
204        let mut src_node = self.nodes.get(OffsetPtr::new(src.index))?;
205        let edge_ptr = self.edges.allocate(GraphEdge {
206            value,
207            dst: dst.index,
208            next_in_src_list: src_node.first_out_edge,
209        })?;
210        src_node.first_out_edge = edge_ptr.index;
211        src_node.n_out_edges = src_node.n_out_edges.wrapping_add(1);
212        self.nodes.set(OffsetPtr::new(src.index), src_node)?;
213        Ok(EdgeIndex::new(edge_ptr.index))
214    }
215
216    /// Read the value at a node index.
217    pub fn node_value(&self, idx: NodeIndex<N>) -> Option<N> {
218        if idx.is_nil() { return None; }
219        self.nodes.get(OffsetPtr::new(idx.index)).ok().map(|n| n.value)
220    }
221
222    /// Read the (value, dst) for an edge index.
223    pub fn edge_endpoints(&self, idx: EdgeIndex<E>) -> Option<(E, NodeIndex<N>)> {
224        if idx.is_nil() { return None; }
225        let e = self.edges.get(OffsetPtr::new(idx.index)).ok()?;
226        Some((e.value, NodeIndex::new(e.dst)))
227    }
228
229    /// Out-degree of a node.
230    pub fn out_degree(&self, src: NodeIndex<N>) -> Option<u32> {
231        if src.is_nil() { return None; }
232        self.nodes.get(OffsetPtr::new(src.index)).ok().map(|n| n.n_out_edges)
233    }
234
235    /// Enumerate outgoing edges from `src` as (EdgeIndex, dst, edge_value).
236    /// Snapshot at call time; not stable under concurrent writes to src.
237    pub fn neighbors(&self, src: NodeIndex<N>) -> Vec<(EdgeIndex<E>, NodeIndex<N>, E)> {
238        if src.is_nil() {
239            self.ring_sidecar
240                .push_op(crate::sidecar_ops::graph::OP_NEIGHBORS, 2);
241            return Vec::new();
242        }
243        let src_node = match self.nodes.get(OffsetPtr::new(src.index)) {
244            Ok(n) => n,
245            Err(_) => {
246                self.ring_sidecar
247                    .push_op(crate::sidecar_ops::graph::OP_NEIGHBORS, 2);
248                return Vec::new();
249            }
250        };
251        let mut out = Vec::with_capacity(src_node.n_out_edges as usize);
252        let mut cur = src_node.first_out_edge;
253        let mut visited = 0u32;
254        let max_iter = self.edges.capacity() as u32;
255        while cur != NIL_INDEX && visited < max_iter {
256            let e = match self.edges.get(OffsetPtr::new(cur)) {
257                Ok(e) => e, Err(_) => break,
258            };
259            out.push((EdgeIndex::new(cur), NodeIndex::new(e.dst), e.value));
260            cur = e.next_in_src_list;
261            visited += 1;
262        }
263        self.ring_sidecar
264            .push_op(crate::sidecar_ops::graph::OP_NEIGHBORS, 0);
265        out
266    }
267
268    /// Remove an edge from `src`'s out-list. Returns the removed
269    /// edge's value if found.
270    pub fn remove_edge(
271        &self, src: NodeIndex<N>, edge_idx: EdgeIndex<E>,
272    ) -> Option<E> {
273        let r = self.remove_edge_inner(src, edge_idx);
274        self.ring_sidecar.push_op(
275            crate::sidecar_ops::graph::OP_REMOVE_EDGE,
276            if r.is_none() { 2 } else { 0 },
277        );
278        r
279    }
280
281    fn remove_edge_inner(
282        &self, src: NodeIndex<N>, edge_idx: EdgeIndex<E>,
283    ) -> Option<E> {
284        if src.is_nil() || edge_idx.is_nil() { return None; }
285        let mut src_node = self.nodes.get(OffsetPtr::new(src.index)).ok()?;
286        let target_value;
287        let target_next;
288        // Find and unlink.
289        if src_node.first_out_edge == edge_idx.index {
290            let e = self.edges.get(OffsetPtr::new(edge_idx.index)).ok()?;
291            target_value = e.value;
292            target_next = e.next_in_src_list;
293            src_node.first_out_edge = target_next;
294        } else {
295            let mut prev_idx = src_node.first_out_edge;
296            loop {
297                if prev_idx == NIL_INDEX { return None; }
298                let mut prev_edge = self.edges.get(OffsetPtr::new(prev_idx)).ok()?;
299                if prev_edge.next_in_src_list == edge_idx.index {
300                    let removed = self.edges.get(OffsetPtr::new(edge_idx.index)).ok()?;
301                    target_value = removed.value;
302                    target_next = removed.next_in_src_list;
303                    prev_edge.next_in_src_list = target_next;
304                    self.edges.set(OffsetPtr::new(prev_idx), prev_edge).ok()?;
305                    break;
306                }
307                prev_idx = prev_edge.next_in_src_list;
308            }
309        }
310        src_node.n_out_edges = src_node.n_out_edges.saturating_sub(1);
311        self.nodes.set(OffsetPtr::new(src.index), src_node).ok()?;
312        // Free returns the prior value; we don't need it here.
313        self.edges.free(OffsetPtr::new(edge_idx.index)).ok();
314        Some(target_value)
315    }
316
317    pub fn flush(&self) -> Result<(), GraphError> {
318        self.nodes.flush()?;
319        self.edges.flush()?;
320        Ok(())
321    }
322    pub fn flush_async(&self) -> Result<(), GraphError> {
323        self.nodes.flush_async()?;
324        self.edges.flush_async()?;
325        Ok(())
326    }
327}
328
329#[cfg(test)]
330mod tests {
331    use super::*;
332
333    fn tmp_base(name: &str) -> PathBuf {
334        let mut p = std::env::temp_dir();
335        let pid = std::process::id();
336        p.push(format!("subetha-graph-{name}-{pid}"));
337        p
338    }
339
340    fn cleanup(base: &Path) {
341        std::fs::remove_file(nodes_path(base)).ok();
342        std::fs::remove_file(edges_path(base)).ok();
343    }
344
345    #[test]
346    fn create_initial_state_is_empty() {
347        let base = tmp_base("init");
348        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 16, 32).unwrap();
349        assert_eq!(g.node_count(), 0);
350        assert_eq!(g.edge_count(), 0);
351        cleanup(&base);
352    }
353
354    #[test]
355    fn add_node_returns_distinct_indices() {
356        let base = tmp_base("add-node");
357        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 16, 32).unwrap();
358        let a = g.add_node(10).unwrap();
359        let b = g.add_node(20).unwrap();
360        let c = g.add_node(30).unwrap();
361        assert_ne!(a, b);
362        assert_ne!(b, c);
363        assert_eq!(g.node_value(a), Some(10));
364        assert_eq!(g.node_value(b), Some(20));
365        assert_eq!(g.node_value(c), Some(30));
366        assert_eq!(g.node_count(), 3);
367        cleanup(&base);
368    }
369
370    #[test]
371    fn add_edge_links_correctly() {
372        let base = tmp_base("add-edge");
373        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 16, 32).unwrap();
374        let a = g.add_node(1).unwrap();
375        let b = g.add_node(2).unwrap();
376        let c = g.add_node(3).unwrap();
377        let _e1 = g.add_edge(a, b, 100).unwrap();
378        let _e2 = g.add_edge(a, c, 200).unwrap();
379        assert_eq!(g.out_degree(a), Some(2));
380        assert_eq!(g.out_degree(b), Some(0));
381        let nbrs = g.neighbors(a);
382        let mut dsts: Vec<u32> = nbrs.iter().map(|(_, d, _)| d.index).collect();
383        let mut vals: Vec<u32> = nbrs.iter().map(|(_, _, v)| *v).collect();
384        dsts.sort();
385        vals.sort();
386        assert_eq!(dsts, vec![b.index, c.index]);
387        assert_eq!(vals, vec![100, 200]);
388        cleanup(&base);
389    }
390
391    #[test]
392    fn multiple_edges_from_one_source() {
393        let base = tmp_base("multi-edge");
394        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 16, 32).unwrap();
395        let src = g.add_node(0).unwrap();
396        let dsts: Vec<NodeIndex<u32>> = (1..=5).map(|i| g.add_node(i).unwrap()).collect();
397        for (i, &d) in dsts.iter().enumerate() {
398            g.add_edge(src, d, (i as u32) * 10).unwrap();
399        }
400        assert_eq!(g.out_degree(src), Some(5));
401        let nbrs = g.neighbors(src);
402        assert_eq!(nbrs.len(), 5);
403        cleanup(&base);
404    }
405
406    #[test]
407    fn remove_edge_unlinks_head() {
408        let base = tmp_base("remove-head");
409        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 16, 32).unwrap();
410        let a = g.add_node(0).unwrap();
411        let b = g.add_node(1).unwrap();
412        let c = g.add_node(2).unwrap();
413        let e1 = g.add_edge(a, b, 100).unwrap();
414        let e2 = g.add_edge(a, c, 200).unwrap();
415        // e2 was added last so it's at head of the linked list.
416        let removed = g.remove_edge(a, e2);
417        assert_eq!(removed, Some(200));
418        assert_eq!(g.out_degree(a), Some(1));
419        let nbrs = g.neighbors(a);
420        assert_eq!(nbrs.len(), 1);
421        assert_eq!(nbrs[0].0, e1);
422        cleanup(&base);
423    }
424
425    #[test]
426    fn remove_edge_unlinks_middle() {
427        let base = tmp_base("remove-middle");
428        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 16, 32).unwrap();
429        let src = g.add_node(0).unwrap();
430        let dsts: Vec<NodeIndex<u32>> = (1..=4).map(|i| g.add_node(i).unwrap()).collect();
431        let edges: Vec<EdgeIndex<u32>> = dsts.iter().enumerate()
432            .map(|(i, &d)| g.add_edge(src, d, (i as u32) * 10).unwrap())
433            .collect();
434        // Remove edges[1] (an arbitrary middle one).
435        let removed = g.remove_edge(src, edges[1]);
436        assert_eq!(removed, Some(10));
437        assert_eq!(g.out_degree(src), Some(3));
438        let nbrs = g.neighbors(src);
439        assert_eq!(nbrs.len(), 3);
440        // The removed edge index isn't in the neighbor list anymore.
441        let edge_idxs: Vec<EdgeIndex<u32>> = nbrs.iter().map(|(e, _, _)| *e).collect();
442        assert!(!edge_idxs.contains(&edges[1]));
443        cleanup(&base);
444    }
445
446    #[test]
447    fn invalid_node_index_rejected_on_add_edge() {
448        let base = tmp_base("invalid-node");
449        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 4, 8).unwrap();
450        let a = g.add_node(0).unwrap();
451        let bogus = NodeIndex::<u32>::new(999);
452        assert_eq!(
453            g.add_edge(a, bogus, 1).err(),
454            Some(GraphError::InvalidNode)
455        );
456        assert_eq!(g.add_edge(NodeIndex::NIL, a, 1).err(), Some(GraphError::InvalidNode));
457        cleanup(&base);
458    }
459
460    #[test]
461    fn cross_handle_visibility() {
462        let base = tmp_base("cross-handle");
463        let w: SharedGraph<u32, u32> = SharedGraph::create(&base, 8, 16).unwrap();
464        let r: SharedGraph<u32, u32> = SharedGraph::open(&base, 8, 16).unwrap();
465        let a = w.add_node(1).unwrap();
466        let b = w.add_node(2).unwrap();
467        let e = w.add_edge(a, b, 42).unwrap();
468        // Reader sees the same graph.
469        assert_eq!(r.node_value(a), Some(1));
470        assert_eq!(r.node_value(b), Some(2));
471        let nbrs = r.neighbors(a);
472        assert_eq!(nbrs.len(), 1);
473        assert_eq!(nbrs[0].0, e);
474        cleanup(&base);
475    }
476
477    #[test]
478    fn struct_payload_round_trip() {
479        #[derive(Clone, Copy, Debug, PartialEq, Default)]
480        #[repr(C)]
481        struct Page { url_hash: u64, depth: u32 }
482        #[derive(Clone, Copy, Debug, PartialEq, Default)]
483        #[repr(C)]
484        struct Link { weight: f32, rel: u32 }
485        let base = tmp_base("struct");
486        let g: SharedGraph<Page, Link> = SharedGraph::create(&base, 8, 16).unwrap();
487        let p1 = Page { url_hash: 0xAAAA, depth: 0 };
488        let p2 = Page { url_hash: 0xBBBB, depth: 1 };
489        let a = g.add_node(p1).unwrap();
490        let b = g.add_node(p2).unwrap();
491        let _e = g.add_edge(a, b, Link { weight: 0.75, rel: 1 }).unwrap();
492        assert_eq!(g.node_value(a), Some(p1));
493        let nbrs = g.neighbors(a);
494        assert_eq!(nbrs.len(), 1);
495        assert_eq!(nbrs[0].1, b);
496        assert_eq!(nbrs[0].2.weight, 0.75);
497        cleanup(&base);
498    }
499
500    #[test]
501    fn capacity_exhaustion_returns_error() {
502        let base = tmp_base("exhaust");
503        let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 3, 2).unwrap();
504        let _a = g.add_node(0).unwrap();
505        let _b = g.add_node(1).unwrap();
506        let _c = g.add_node(2).unwrap();
507        // Node region full.
508        assert!(g.add_node(3).is_err());
509        cleanup(&base);
510    }
511
512    #[test]
513    fn disk_persistence_survives_reopen() {
514        let base = tmp_base("disk");
515        let saved_a;
516        let saved_b;
517        {
518            let g: SharedGraph<u32, u32> = SharedGraph::create(&base, 8, 16).unwrap();
519            saved_a = g.add_node(100).unwrap();
520            saved_b = g.add_node(200).unwrap();
521            g.add_edge(saved_a, saved_b, 42).unwrap();
522            g.flush().unwrap();
523        }
524        let g2: SharedGraph<u32, u32> = SharedGraph::open(&base, 8, 16).unwrap();
525        assert_eq!(g2.node_count(), 2);
526        assert_eq!(g2.node_value(saved_a), Some(100));
527        let nbrs = g2.neighbors(saved_a);
528        assert_eq!(nbrs.len(), 1);
529        assert_eq!(nbrs[0].2, 42);
530        cleanup(&base);
531    }
532}
subetha_cxc/shared_graph.rs

subetha_cxc/
shared_graph.rs
