linkrs_storage/edge/single_mutable_csr/
maintenance.rs

1//! Positional delete/revert and maintenance for `SingleMutableCsr`:
2//! position-addressed writes, reclaim probes and compaction.
3
4use linkrs_core::StorageResult;
5
6use super::super::csr_shared::{
7    can_revert_delete, decide_slot_delete, is_reclaimable_cold, DeleteSlotOutcome,
8};
9use super::super::{EdgeId, EdgePosition, Timestamp, INVALID_EDGE_ID};
10use super::{empty_slot, SingleMutableCsr};
11
12impl SingleMutableCsr {
13    /// Delete the single slot at `position` when it holds `expected`.
14    ///
15    /// Only the row-local primary slot zero is valid; every other position
16    /// is stale and refused without touching any slot.
17    pub fn delete_edge_at_position(
18        &mut self,
19        src: u32,
20        position: EdgePosition,
21        expected: EdgeId,
22        ts: Timestamp,
23    ) -> StorageResult<bool> {
24        if !matches!(position, EdgePosition::Primary { slot: 0 }) {
25            return Ok(false);
26        }
27        let src_idx = src as usize;
28        if src_idx >= self.vertex_capacity() {
29            return Ok(false);
30        }
31        let probe = match self.slot_at(src_idx) {
32            Some(probe) => probe,
33            None => return Ok(false),
34        };
35        if probe.edge_id == INVALID_EDGE_ID || probe.edge_id != expected {
36            return Ok(false);
37        }
38        if matches!(
39            decide_slot_delete(&probe, expected, ts)?,
40            DeleteSlotOutcome::AlreadyStamped
41        ) {
42            return Ok(false);
43        }
44        self.stamp_delete(src_idx, ts);
45        self.edge_count -= 1;
46        Ok(true)
47    }
48
49    /// Revert the single-slot deletion at `position` when it holds `expected`.
50    pub fn revert_delete_at_position(
51        &mut self,
52        src: u32,
53        position: EdgePosition,
54        expected: EdgeId,
55        ts: Timestamp,
56    ) -> bool {
57        if !matches!(position, EdgePosition::Primary { slot: 0 }) {
58            return false;
59        }
60        let src_idx = src as usize;
61        if src_idx >= self.vertex_capacity() {
62            return false;
63        }
64        let probe = match self.slot_at(src_idx) {
65            Some(probe) => probe,
66            None => return false,
67        };
68        if probe.edge_id == INVALID_EDGE_ID || probe.edge_id != expected {
69            return false;
70        }
71        if can_revert_delete(&probe, ts) {
72            self.clear_delete(src_idx);
73            self.edge_count += 1;
74            return true;
75        }
76        false
77    }
78
79    /// Delete the single matching entry, reporting its row-local position.
80    pub fn delete_edge_by_dst_reporting_positioned(
81        &mut self,
82        src: u32,
83        dst: super::super::VertexId,
84        ts: Timestamp,
85        on_deleted: &mut dyn FnMut(EdgeId, Option<EdgePosition>),
86    ) -> usize {
87        let src_idx = src as usize;
88        if src_idx >= self.vertex_capacity() {
89            return 0;
90        }
91        let (dst_ep, dst_rank) = match super::super::csr_shared::decode_endpoint_pair(dst) {
92            Some(pair) => pair,
93            None => return 0,
94        };
95        let probe = match self.slot_at(src_idx) {
96            Some(probe) => probe,
97            None => return 0,
98        };
99        if probe.edge_id == INVALID_EDGE_ID
100            || probe.endpoint != dst_ep
101            || probe.rank != dst_rank
102            || !self.cold_is_live_at(src_idx)
103        {
104            return 0;
105        }
106        self.stamp_delete(src_idx, ts);
107        self.edge_count -= 1;
108        on_deleted(probe.edge_id, Some(EdgePosition::Primary { slot: 0 }));
109        1
110    }
111
112    pub fn reclaimable_count(&self, vid: u32, cutoff: Timestamp) -> usize {
113        if cutoff == Timestamp::MAX {
114            return 0;
115        }
116        let hot = self
117            .hot_at(vid as usize)
118            .unwrap_or(super::super::HotNbr::dead_gap());
119        let cold = self
120            .cold_at(vid as usize)
121            .unwrap_or(super::super::ColdStamps::dead_gap());
122        if hot.edge_id != INVALID_EDGE_ID && is_reclaimable_cold(&cold, cutoff) {
123            1
124        } else {
125            0
126        }
127    }
128
129    pub fn vertex_census(&self, vid: u32) -> (usize, usize, usize) {
130        let hot = match self.hot_at(vid as usize) {
131            Some(h) => h,
132            None => return (0, 0, 0),
133        };
134        let cold = self
135            .cold_at(vid as usize)
136            .unwrap_or(super::super::ColdStamps::dead_gap());
137        if hot.edge_id == INVALID_EDGE_ID {
138            return (0, 0, 0);
139        }
140        if cold.is_live() {
141            (1, 0, 1)
142        } else {
143            (0, 1, 1)
144        }
145    }
146
147    /// Single-slot reclaim probe: `(dead, reclaimable)` in one slot read.
148    pub fn vertex_reclaim_probe(&self, vid: u32, cutoff: Timestamp) -> (usize, usize) {
149        let hot = match self.hot_at(vid as usize) {
150            Some(h) => h,
151            None => return (0, 0),
152        };
153        let cold = self
154            .cold_at(vid as usize)
155            .unwrap_or(super::super::ColdStamps::dead_gap());
156        if hot.edge_id == INVALID_EDGE_ID || cold.is_live() {
157            return (0, 0);
158        }
159        let reclaimable =
160            usize::from(cutoff != Timestamp::MAX && is_reclaimable_cold(&cold, cutoff));
161        (1, reclaimable)
162    }
163
164    pub fn compact_vertex_with_reporting(
165        &mut self,
166        vid: u32,
167        cutoff: Timestamp,
168        on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
169    ) -> usize {
170        if self.reclaimable_count(vid, cutoff) == 0 {
171            return 0;
172        }
173        let src_idx = vid as usize;
174        let edge_id = self
175            .hot_at(src_idx)
176            .map(|h| h.edge_id)
177            .unwrap_or(INVALID_EDGE_ID);
178        let delete_ts = self
179            .cold_at(src_idx)
180            .map(|c| c.delete_ts)
181            .unwrap_or(Timestamp::MAX);
182        on_edge_removed(edge_id, delete_ts);
183        self.set_slot(src_idx, empty_slot());
184        1
185    }
186
187    pub fn compact_with_ts_reporting(
188        &mut self,
189        cutoff: Timestamp,
190        on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
191    ) -> usize {
192        if cutoff == Timestamp::MAX {
193            return 0;
194        }
195        let mut reclaimable: Vec<(EdgeId, Timestamp, usize)> = Vec::new();
196        for seg_idx in 0..self.segments.len() {
197            let Some(segment) = self.segments.get(seg_idx).and_then(|s| s.as_ref()) else {
198                continue;
199            };
200            for off in 0..Self::SEGMENT_ROWS {
201                let hot = segment.hot[off];
202                let cold = segment.cold[off];
203                if hot.edge_id != INVALID_EDGE_ID && is_reclaimable_cold(&cold, cutoff) {
204                    let idx = seg_idx * Self::SEGMENT_ROWS + off;
205                    reclaimable.push((hot.edge_id, cold.delete_ts, idx));
206                }
207            }
208        }
209        let mut removed = 0usize;
210        for (edge_id, delete_ts, idx) in reclaimable {
211            on_edge_removed(edge_id, delete_ts);
212            self.set_slot(idx, empty_slot());
213            removed += 1;
214        }
215        removed
216    }
217
218    pub fn clear(&mut self) {
219        self.segments.clear();
220        self.present.clear();
221        self.vertex_capacity = 0;
222        self.edge_count = 0;
223    }
224}
linkrs_storage/edge/single_mutable_csr/maintenance.rs

linkrs_storage/edge/single_mutable_csr/
maintenance.rs
