#![allow(clippy::too_many_arguments)]
use std::sync::Arc;
use crate::fixed_point_graph::FixedPointAnalysisKind;
use vyre_foundation::ir::model::expr::Ident;
use vyre_foundation::ir::{BufferAccess, BufferDecl, DataType, Expr, Node, Program};
use vyre_primitives::graph::program_graph::ProgramGraphShape;
const CLEAR_CHANGED_OP: &str = "weir::fixed_point::clear_changed";
pub(crate) fn run_resident_sequence_window_into(
backend: &dyn vyre::VyreBackend,
program: &vyre::ir::Program,
graph: &crate::fixed_point_resident::FixedPointResidentGraph,
seed_bits: &[u32],
max_iterations: u32,
scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
resident_frontier: &mut crate::fixed_point_resident::FixedPointResidentFrontierScratch,
result: &mut Vec<u32>,
kind: FixedPointAnalysisKind,
consumer: &str,
) -> Result<(), String> {
graph.require_kind(kind, consumer)?;
crate::dispatch_decode::require_positive_iterations(consumer, max_iterations)?;
let words = crate::dispatch_decode::bitset_word_capacity(consumer, graph.node_count())?;
crate::dispatch_decode::require_bitset_words(consumer, seed_bits, words)?;
crate::dispatch_decode::require_bitset_tail_clear(consumer, seed_bits, graph.node_count())?;
scratch.prepare_frontier_words(words, seed_bits)?;
let frontier_byte_len = words.checked_mul(4).ok_or_else(|| {
format!(
"{consumer} resident frontier byte length overflow. Fix: shard the graph before GPU dispatch."
)
})?;
if matches!(
kind,
FixedPointAnalysisKind::Reaching
| FixedPointAnalysisKind::Live
| FixedPointAnalysisKind::PointsToSubset
| FixedPointAnalysisKind::Slice
) {
let (frontier, changed) =
resident_frontier.ensure_inplace_refs(backend, frontier_byte_len)?;
return run_resident_changed_flag_window_resources_into(
backend,
graph,
frontier,
changed,
max_iterations,
scratch,
result,
kind,
consumer,
);
}
let (frontier_in, frontier_out) =
resident_frontier.ensure_pair_refs(backend, frontier_byte_len)?;
run_resident_sequence_window_resources_into(
backend,
program,
graph,
frontier_in,
frontier_out,
max_iterations,
scratch,
result,
consumer,
)
}
fn run_resident_sequence_window_resources_into(
backend: &dyn vyre::VyreBackend,
program: &vyre::ir::Program,
graph: &crate::fixed_point_resident::FixedPointResidentGraph,
frontier_in: &vyre::backend::Resource,
frontier_out: &vyre::backend::Resource,
max_iterations: u32,
scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
result: &mut Vec<u32>,
consumer: &str,
) -> Result<(), String> {
let words = scratch.frontier_words.len();
let frontier_byte_len = words.checked_mul(4).ok_or_else(|| {
format!(
"{consumer} resident frontier byte length overflow. Fix: shard the graph before GPU dispatch."
)
})?;
scratch.begin_frontier_density(graph.node_count());
crate::dispatch_decode::try_pack_u32_into(&scratch.frontier_words, &mut scratch.frontier_bytes, "weir u32 byte staging")?;
backend
.upload_resident_many(&[
(frontier_in, scratch.frontier_bytes.as_slice()),
(frontier_out, scratch.frontier_bytes.as_slice()),
])
.map_err(|error| error.to_string())?;
graph.resources_with_resident_frontier_into(
frontier_in,
frontier_out,
&mut scratch.resources,
)?;
let grid_override = Some([graph.node_count().max(1), 1, 1]);
if max_iterations % 2 == 0 {
graph.resources_with_resident_frontier_into(
frontier_out,
frontier_in,
&mut scratch.alternate_resources,
)?;
let repeated_steps = [
vyre::backend::ResidentDispatchStep {
program,
resources: scratch.resources.as_slice(),
grid_override,
workgroup_override: None,
},
vyre::backend::ResidentDispatchStep {
program,
resources: scratch.alternate_resources.as_slice(),
grid_override,
workgroup_override: None,
},
];
let read_ranges = [
vyre::backend::ResidentReadRange {
resource: frontier_in,
byte_offset: 0,
byte_len: frontier_byte_len,
},
vyre::backend::ResidentReadRange {
resource: frontier_out,
byte_offset: 0,
byte_len: frontier_byte_len,
},
];
backend
.dispatch_resident_repeated_sequence_read_ranges_into(
&[],
&repeated_steps,
max_iterations / 2,
&read_ranges,
&mut [
&mut scratch.frontier_bytes,
&mut scratch.previous_frontier_bytes,
],
)
.map_err(|error| error.to_string())?;
} else {
let prefix_steps = [vyre::backend::ResidentDispatchStep {
program,
resources: scratch.resources.as_slice(),
grid_override,
workgroup_override: None,
}];
if max_iterations > 1 {
graph.resources_with_resident_frontier_into(
frontier_out,
frontier_in,
&mut scratch.alternate_resources,
)?;
} else {
scratch.alternate_resources.clear();
}
let repeated_steps = [
vyre::backend::ResidentDispatchStep {
program,
resources: scratch.alternate_resources.as_slice(),
grid_override,
workgroup_override: None,
},
vyre::backend::ResidentDispatchStep {
program,
resources: scratch.resources.as_slice(),
grid_override,
workgroup_override: None,
},
];
let read_ranges = [
vyre::backend::ResidentReadRange {
resource: frontier_out,
byte_offset: 0,
byte_len: frontier_byte_len,
},
vyre::backend::ResidentReadRange {
resource: frontier_in,
byte_offset: 0,
byte_len: frontier_byte_len,
},
];
backend
.dispatch_resident_repeated_sequence_read_ranges_into(
&prefix_steps,
&repeated_steps,
max_iterations / 2,
&read_ranges,
&mut [
&mut scratch.frontier_bytes,
&mut scratch.previous_frontier_bytes,
],
)
.map_err(|error| error.to_string())?;
}
crate::dispatch_decode::unpack_exact_u32_into(
&scratch.frontier_bytes,
words,
consumer,
&mut scratch.next,
)?;
crate::dispatch_decode::unpack_exact_u32_into(
&scratch.previous_frontier_bytes,
words,
consumer,
&mut scratch.frontier_words,
)?;
crate::dispatch_decode::require_bitset_tail_clear(consumer, &scratch.next, graph.node_count())?;
crate::dispatch_decode::require_bitset_tail_clear(
consumer,
&scratch.frontier_words,
graph.node_count(),
)?;
scratch.record_compacted_frontier_window(max_iterations);
if scratch.next != scratch.frontier_words {
return Err(format!(
"{consumer} did not converge within {max_iterations} sequence-window resident iterations"
));
}
crate::fixed_point_scratch::copy_words_into(&scratch.next, result)?;
Ok(())
}
fn run_resident_changed_flag_window_resources_into(
backend: &dyn vyre::VyreBackend,
graph: &crate::fixed_point_resident::FixedPointResidentGraph,
frontier: &vyre::backend::Resource,
changed: &vyre::backend::Resource,
max_iterations: u32,
scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
result: &mut Vec<u32>,
kind: FixedPointAnalysisKind,
consumer: &str,
) -> Result<(), String> {
let words = scratch.frontier_words.len();
let frontier_byte_len = words.checked_mul(4).ok_or_else(|| {
format!(
"{consumer} resident frontier byte length overflow. Fix: shard the graph before GPU dispatch."
)
})?;
scratch.begin_frontier_density(graph.node_count());
crate::dispatch_decode::try_pack_u32_into(&scratch.frontier_words, &mut scratch.frontier_bytes, "weir u32 byte staging")?;
crate::dispatch_decode::try_pack_u32_into(&[0], &mut scratch.changed_bytes, "weir u32 byte staging")?;
backend
.upload_resident_many(&[
(frontier, scratch.frontier_bytes.as_slice()),
(changed, scratch.changed_bytes.as_slice()),
])
.map_err(|error| error.to_string())?;
graph.resources_with_resident_frontier_changed_into(
frontier,
changed,
&mut scratch.changed_resources,
)?;
let clear_changed_program = ensure_clear_changed_program(&mut scratch.clear_changed_program);
let forward_changed_program = ensure_forward_changed_program(
&mut scratch.forward_changed_program,
kind,
graph.node_count(),
graph.edge_count(),
);
let repeated_steps = [
vyre::backend::ResidentDispatchStep {
program: clear_changed_program,
resources: std::slice::from_ref(changed),
grid_override: Some([1, 1, 1]),
workgroup_override: None,
},
vyre::backend::ResidentDispatchStep {
program: forward_changed_program,
resources: scratch.changed_resources.as_slice(),
grid_override: Some([graph.node_count().max(1), 1, 1]),
workgroup_override: None,
},
];
let read_ranges = [
vyre::backend::ResidentReadRange {
resource: changed,
byte_offset: 0,
byte_len: std::mem::size_of::<u32>(),
},
vyre::backend::ResidentReadRange {
resource: frontier,
byte_offset: 0,
byte_len: frontier_byte_len,
},
];
backend
.dispatch_resident_repeated_sequence_read_ranges_into(
&[],
&repeated_steps,
max_iterations,
&read_ranges,
&mut [&mut scratch.changed_bytes, &mut scratch.frontier_bytes],
)
.map_err(|error| error.to_string())?;
let changed_word =
crate::dispatch_decode::unpack_exact_u32_scalar(&scratch.changed_bytes, consumer)?;
crate::dispatch_decode::unpack_exact_u32_into(
&scratch.frontier_bytes,
words,
consumer,
&mut scratch.next,
)?;
crate::dispatch_decode::require_bitset_tail_clear(consumer, &scratch.next, graph.node_count())?;
scratch.record_compacted_frontier_window(max_iterations);
if changed_word != 0 {
return Err(format!(
"{consumer} did not converge within {max_iterations} resident changed-flag iterations"
));
}
crate::fixed_point_scratch::copy_words_into(&scratch.next, result)?;
Ok(())
}
fn ensure_clear_changed_program(cache: &mut Option<Program>) -> &Program {
cache.get_or_insert_with(|| {
Program::wrapped(
vec![
BufferDecl::storage("changed", 0, BufferAccess::ReadWrite, DataType::U32)
.with_count(1),
],
[1, 1, 1],
vec![Node::Region {
generator: Ident::from(CLEAR_CHANGED_OP),
source_region: None,
body: Arc::new(vec![Node::store("changed", Expr::u32(0), Expr::u32(0))]),
}],
)
})
}
fn ensure_forward_changed_program(
cache: &mut Option<crate::fixed_point_scratch::FixedPointForwardChangedProgramCache>,
kind: FixedPointAnalysisKind,
node_count: u32,
edge_count: u32,
) -> &Program {
let cached = cache
.get_or_insert_with(|| build_forward_changed_program_cache(kind, node_count, edge_count));
if cached.kind != kind || cached.node_count != node_count || cached.edge_count != edge_count {
*cached = build_forward_changed_program_cache(kind, node_count, edge_count);
}
&cached.program
}
fn build_forward_changed_program_cache(
kind: FixedPointAnalysisKind,
node_count: u32,
edge_count: u32,
) -> crate::fixed_point_scratch::FixedPointForwardChangedProgramCache {
crate::fixed_point_scratch::FixedPointForwardChangedProgramCache {
kind,
node_count,
edge_count,
program: match kind {
FixedPointAnalysisKind::Slice => {
vyre_primitives::graph::csr_backward_or_changed::csr_backward_or_changed_parallel(
ProgramGraphShape::new(node_count, edge_count),
"frontier",
"changed",
u32::MAX,
)
}
_ => vyre_primitives::graph::csr_forward_or_changed::csr_forward_or_changed_parallel(
ProgramGraphShape::new(node_count, edge_count),
"frontier",
"changed",
u32::MAX,
),
},
}
}
#[cfg(test)]
mod tests {
#[test]
fn resident_sequence_production_path_has_no_panic_or_clone_staging() {
let source = include_str!("fixed_point_resident_sequence.rs");
let production = source
.split("#[cfg(test)]")
.next()
.expect("production sequence source must precede tests");
for forbidden in [
concat!("panic", "!("),
concat!(".", "expect("),
concat!("unimplemented", "!("),
concat!("todo", "!("),
"changed.clone()",
] {
assert!(
!production.contains(forbidden),
"resident sequence-window production path must not contain {forbidden}"
);
}
assert!(
production.contains("std::slice::from_ref(changed)"),
"clear-changed sequence step must borrow the resident changed handle without clone allocation"
);
}
}