vyre_primitives/graph/csr_queue_delta/
mod.rs

1//! Queue-to-queue sparse CSR expansion for delta fixpoint waves.
2//!
3//! A full frontier bitset scan is the wrong shape once a dataflow pipeline has
4//! already compacted the active wave. This primitive consumes only queued
5//! sources, updates a resident accumulator bitset, and appends first-time
6//! discoveries directly into the next active queue.
7
8use vyre_foundation::ir::{DataType, Program};
9
10use crate::graph::csr_frontier_step::{
11    csr_queue_step_program, CsrQueueEmit, CsrQueueInputs, CsrQueueLanes, CsrQueueRowPlan,
12    CsrQueueStepSpec,
13};
14
15mod strided;
16
17pub use strided::{
18    csr_queue_delta_strided_dispatch_grid, csr_queue_delta_strided_enqueue,
19    csr_queue_delta_strided_enqueue_with, csr_queue_delta_strided_logical_lanes_per_launch,
20    csr_queue_delta_strided_source_slots_per_launch,
21    CSR_QUEUE_DELTA_STRIDED_CAPPED_LAUNCH_MIN_CAPACITY, CSR_QUEUE_DELTA_STRIDED_ENQUEUE_OP_ID,
22    CSR_QUEUE_DELTA_STRIDED_LANES_PER_SOURCE, CSR_QUEUE_DELTA_STRIDED_MAX_SOURCE_SLOTS_PER_LAUNCH,
23};
24
25/// Canonical op id for queue-to-queue delta CSR expansion.
26pub const CSR_QUEUE_DELTA_ENQUEUE_OP_ID: &str = "vyre-primitives::graph::csr_queue_delta_enqueue";
27
28/// Default workgroup size for queue-to-queue delta expansion.
29pub const CSR_QUEUE_DELTA_ENQUEUE_WORKGROUP_SIZE: [u32; 3] = [256, 1, 1];
30
31/// Positional inputs shared by [`csr_queue_delta_enqueue`] and
32/// [`csr_queue_delta_strided_enqueue`].
33#[derive(Clone, Copy, Debug)]
34pub struct CsrQueueDeltaEnqueueParams<'a> {
35    /// Compacted queue of active source nodes.
36    pub active_queue: &'a str,
37    /// Single-element resident length of `active_queue`.
38    pub active_len: &'a str,
39    /// CSR row pointers, `node_count + 1` entries.
40    pub edge_offsets: &'a str,
41    /// CSR edge destinations.
42    pub edge_targets: &'a str,
43    /// Per-edge kind bits tested against `allow_mask`.
44    pub edge_kind_mask: &'a str,
45    /// Monotone reachability bitset each reached destination is ORed into.
46    pub accumulator: &'a str,
47    /// Queue first-time discoveries are appended to.
48    pub next_queue: &'a str,
49    /// Single-element observed next length, which may exceed the capacity.
50    pub next_len: &'a str,
51    /// Node count the CSR row pointers and destination bounds are sized by.
52    pub node_count: u32,
53    /// Logical edge count the edge-slot bound check uses.
54    pub edge_count: u32,
55    /// Static capacity of `active_queue`.
56    pub active_queue_capacity: u32,
57    /// Static capacity of `next_queue`.
58    pub next_queue_capacity: u32,
59    /// Edge kinds this traversal is allowed to follow.
60    pub allow_mask: u32,
61}
62
63/// Build a GPU program that expands queued CSR rows and enqueues only new nodes.
64///
65/// `accumulator` is the monotone reachability bitset. When an allowed edge
66/// reaches a destination whose bit was absent, the destination is appended to
67/// `next_queue` and `next_len` is incremented. The observed next length can
68/// exceed `next_queue_capacity`; stores are clamped so callers can detect
69/// overflow pressure without corrupting resident memory.
70#[must_use]
71#[allow(clippy::too_many_arguments)]
72pub fn csr_queue_delta_enqueue(
73    active_queue: &str,
74    active_len: &str,
75    edge_offsets: &str,
76    edge_targets: &str,
77    edge_kind_mask: &str,
78    accumulator: &str,
79    next_queue: &str,
80    next_len: &str,
81    node_count: u32,
82    edge_count: u32,
83    active_queue_capacity: u32,
84    next_queue_capacity: u32,
85    allow_mask: u32,
86) -> Program {
87    csr_queue_delta_enqueue_with(CsrQueueDeltaEnqueueParams {
88        active_queue,
89        active_len,
90        edge_offsets,
91        edge_targets,
92        edge_kind_mask,
93        accumulator,
94        next_queue,
95        next_len,
96        node_count,
97        edge_count,
98        active_queue_capacity,
99        next_queue_capacity,
100        allow_mask,
101    })
102}
103
104/// Build a GPU program that expands queued CSR rows and enqueues only new nodes.
105#[must_use]
106pub fn csr_queue_delta_enqueue_with(params: CsrQueueDeltaEnqueueParams<'_>) -> Program {
107    let node_count = params.node_count;
108    let active_queue_capacity = params.active_queue_capacity;
109    let next_queue_capacity = params.next_queue_capacity;
110    if node_count == 0 || active_queue_capacity == 0 || next_queue_capacity == 0 {
111        return crate::invalid_output_program(CSR_QUEUE_DELTA_ENQUEUE_OP_ID,
112        params.next_len,
113        DataType::U32,
114        format!(
115            "Fix: csr_queue_delta_enqueue requires node_count > 0 and non-zero queue capacities, got node_count={node_count} active_queue_capacity={active_queue_capacity} next_queue_capacity={next_queue_capacity}."
116        ),);
117    }
118    csr_queue_step_program(&params.spec(
119        CSR_QUEUE_DELTA_ENQUEUE_OP_ID,
120        "csr_queue_delta_enqueue",
121        "qd",
122        CsrQueueLanes::Scalar,
123    ))
124}
125
126impl<'a> CsrQueueDeltaEnqueueParams<'a> {
127    /// Point these inputs at the shared queue-step builder. Both delta entry
128    /// points differ only in op id, variable prefix, and lane assignment.
129    fn spec(
130        &self,
131        op_id: &'static str,
132        builder_name: &'static str,
133        prefix: &'a str,
134        lanes: CsrQueueLanes,
135    ) -> CsrQueueStepSpec<'a> {
136        CsrQueueStepSpec {
137            op_id,
138            builder_name,
139            prefix,
140            workgroup_size: CSR_QUEUE_DELTA_ENQUEUE_WORKGROUP_SIZE,
141            inputs: CsrQueueInputs {
142                active_queue: self.active_queue,
143                queue_len: self.active_len,
144                edge_offsets: self.edge_offsets,
145                edge_targets: self.edge_targets,
146                edge_kind_mask: self.edge_kind_mask,
147            },
148            lanes,
149            row_plan: CsrQueueRowPlan::ExpandAll,
150            emit: CsrQueueEmit::Delta {
151                accumulator: self.accumulator,
152                next_queue: self.next_queue,
153                next_len: self.next_len,
154                next_queue_capacity: self.next_queue_capacity,
155            },
156            node_count: self.node_count,
157            edge_count: self.edge_count,
158            queue_capacity: self.active_queue_capacity,
159            allow_mask: self.allow_mask,
160        }
161    }
162}
163
164/// CPU reference for queue-to-queue delta expansion.
165#[must_use]
166#[cfg(any(test, feature = "cpu-parity"))]
167#[allow(clippy::too_many_arguments)]
168pub fn csr_queue_delta_enqueue_cpu(
169    active_queue: &[u32],
170    active_len: u32,
171    edge_offsets: &[u32],
172    edge_targets: &[u32],
173    edge_kind_mask: &[u32],
174    accumulator: &[u32],
175    node_count: u32,
176    next_queue_capacity: usize,
177    allow_mask: u32,
178) -> (Vec<u32>, Vec<u32>, u32) {
179    let mut accumulator = accumulator.to_vec();
180    let mut next_queue = Vec::new();
181    let next_len = try_csr_queue_delta_enqueue_cpu_into(
182        active_queue,
183        active_len,
184        edge_offsets,
185        edge_targets,
186        edge_kind_mask,
187        &mut accumulator,
188        node_count,
189        next_queue_capacity,
190        allow_mask,
191        &mut next_queue,
192    )
193    .unwrap_or_else(|err| {
194        panic!("csr_queue_delta_enqueue CPU oracle received malformed input. {err}")
195    });
196    (accumulator, next_queue, next_len)
197}
198
199/// Fallible CPU reference for queue-to-queue delta expansion into caller storage.
200///
201/// On validation failure both `accumulator` and `next_queue` are left unchanged.
202#[cfg(any(test, feature = "cpu-parity"))]
203#[allow(clippy::too_many_arguments)]
204pub fn try_csr_queue_delta_enqueue_cpu_into(
205    active_queue: &[u32],
206    active_len: u32,
207    edge_offsets: &[u32],
208    edge_targets: &[u32],
209    edge_kind_mask: &[u32],
210    accumulator: &mut Vec<u32>,
211    node_count: u32,
212    next_queue_capacity: usize,
213    allow_mask: u32,
214    next_queue: &mut Vec<u32>,
215) -> Result<u32, String> {
216    let layout = super::csr_frontier_queue::validate_csr_queue_graph(
217        node_count,
218        edge_offsets,
219        edge_targets,
220        edge_kind_mask,
221    )?;
222    if accumulator.len() != layout.words {
223        return Err(format!(
224            "Fix: csr_queue_delta_enqueue requires accumulator.len() == bitset_words(node_count), got len={} but expected {} for node_count={node_count}.",
225            accumulator.len(),
226            layout.words
227        ));
228    }
229    crate::graph::scratch::reserve_graph_items(
230        next_queue,
231        next_queue_capacity,
232        "CSR queue delta CPU oracle",
233        "next active frontier queue",
234    )?;
235
236    let mut next_tmp = Vec::with_capacity(next_queue_capacity);
237    let mut accumulator_tmp = accumulator.clone();
238    let take = (active_len as usize).min(active_queue.len());
239    let mut next_seen = 0_u32;
240
241    for &src in &active_queue[..take] {
242        if src >= node_count {
243            continue;
244        }
245        let start = edge_offsets[src as usize] as usize;
246        let end = edge_offsets[src as usize + 1] as usize;
247        for edge in start..end {
248            if edge_kind_mask[edge] & allow_mask == 0 {
249                continue;
250            }
251            let dst = edge_targets[edge];
252            let word = dst as usize / 32;
253            let bit = 1_u32 << (dst % 32);
254            let old = accumulator_tmp[word];
255            if old & bit != 0 {
256                continue;
257            }
258            accumulator_tmp[word] = old | bit;
259            if next_tmp.len() < next_queue_capacity {
260                next_tmp.push(dst);
261            }
262            next_seen = next_seen.saturating_add(1);
263        }
264    }
265
266    *accumulator = accumulator_tmp;
267    next_queue.clear();
268    next_queue.extend_from_slice(&next_tmp);
269    Ok(next_seen)
270}
271
272#[cfg(test)]
273mod tests {
274    use super::*;
275
276    #[test]
277    fn emitted_program_has_stable_delta_queue_shape() {
278        let program = csr_queue_delta_enqueue(
279            "active_queue",
280            "active_len",
281            "edge_offsets",
282            "edge_targets",
283            "edge_kind_mask",
284            "accumulator",
285            "next_queue",
286            "next_len",
287            64,
288            7,
289            8,
290            16,
291            1,
292        );
293
294        assert_eq!(
295            program.workgroup_size,
296            CSR_QUEUE_DELTA_ENQUEUE_WORKGROUP_SIZE
297        );
298        assert_eq!(program.buffers.len(), 8);
299    }
300
301    #[test]
302    fn delta_enqueue_rejects_offset_count_overflow_without_panic() {
303        let result = std::panic::catch_unwind(|| {
304            csr_queue_delta_enqueue(
305                "active_queue",
306                "active_len",
307                "edge_offsets",
308                "edge_targets",
309                "edge_kind_mask",
310                "accumulator",
311                "next_queue",
312                "next_len",
313                u32::MAX,
314                0,
315                1,
316                1,
317                1,
318            )
319        });
320
321        assert!(
322            result.is_ok(),
323            "CSR queue delta builder must reject offset-count overflow without panicking"
324        );
325        let program = result.unwrap();
326        assert!(program.stats().trap());
327        let entry = format!("{:?}", program.entry());
328        assert!(
329            entry.contains("node_count + 1 overflows u32"),
330            "Fix: trap must retain the CSR offset-count overflow diagnostic, got: {entry}"
331        );
332    }
333
334    #[test]
335    fn cpu_delta_enqueue_only_emits_first_time_discoveries() {
336        let edge_offsets = [0, 3, 4, 4, 4, 4];
337        let edge_targets = [1, 2, 3, 4];
338        let edge_kind_mask = [1, 1, 2, 1];
339        let accumulator = vec![0b00001];
340
341        let (accumulator, next_queue, next_len) = csr_queue_delta_enqueue_cpu(
342            &[0, 1],
343            2,
344            &edge_offsets,
345            &edge_targets,
346            &edge_kind_mask,
347            &accumulator,
348            5,
349            8,
350            1,
351        );
352
353        assert_eq!(accumulator, vec![0b10111]);
354        assert_eq!(next_queue, vec![1, 2, 4]);
355        assert_eq!(next_len, 3);
356    }
357
358    #[test]
359    fn cpu_delta_enqueue_reports_queue_pressure_without_clobbering_accumulator() {
360        let edge_offsets = [0, 3, 3, 3, 3];
361        let edge_targets = [1, 2, 3];
362        let edge_kind_mask = [1, 1, 1];
363        let mut accumulator = vec![0b0001];
364        let mut next_queue = Vec::new();
365
366        let next_len = try_csr_queue_delta_enqueue_cpu_into(
367            &[0],
368            1,
369            &edge_offsets,
370            &edge_targets,
371            &edge_kind_mask,
372            &mut accumulator,
373            4,
374            2,
375            1,
376            &mut next_queue,
377        )
378        .expect("Fix: canonical queue delta graph should enqueue bounded discoveries");
379
380        assert_eq!(accumulator, vec![0b1111]);
381        assert_eq!(next_queue, vec![1, 2]);
382        assert_eq!(next_len, 3);
383    }
384
385    #[test]
386    fn cpu_delta_enqueue_rejects_bad_accumulator_without_clobbering_outputs() {
387        let mut accumulator = vec![0xCAFE_BABE, 0xDEAD_BEEF];
388        let mut next_queue = vec![9, 8, 7];
389
390        let err = try_csr_queue_delta_enqueue_cpu_into(
391            &[0],
392            1,
393            &[0, 1],
394            &[0],
395            &[1],
396            &mut accumulator,
397            1,
398            4,
399            1,
400            &mut next_queue,
401        )
402        .expect_err("wrong accumulator width must fail before mutation");
403
404        assert!(err.contains("accumulator.len() == bitset_words(node_count)"));
405        assert_eq!(accumulator, vec![0xCAFE_BABE, 0xDEAD_BEEF]);
406        assert_eq!(next_queue, vec![9, 8, 7]);
407    }
408}
vyre_primitives/graph/csr_queue_delta/mod.rs

vyre_primitives/graph/csr_queue_delta/
mod.rs
