use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
use onnx_runtime_ep_api::{
CaptureRegionShapeStatus, DeviceBuffer, DevicePtr, DevicePtrMut, EpError, ExecutionProvider,
ExternalMmapRegion, Kernel, KernelInput, KernelMatch, LazyWeight, LazyWeightBoundary,
ResidentWeight, StructuralCaptureDecline, TensorBacking, TensorMut, TensorView, WeightHandle,
};
type OptionalTensorSpecs = Vec<Option<(DataType, Vec<usize>)>>;
use onnx_runtime_ep_cpu::CpuExecutionProvider;
use onnx_runtime_ep_cpu::strided::view_in_bounds;
use onnx_runtime_ir::Attribute;
use onnx_runtime_ir::{
DataType, DeviceType, Dim, Graph, Node, NodeId, Shape, SymbolId, TensorLayout, ValueId,
WeightRef, as_static_shape, broadcast_shapes, compute_contiguous_strides,
};
use onnx_runtime_loader::WeightStore;
use onnx_runtime_optimizer::InitializerResolver;
use onnx_runtime_shape_inference::{
DimExpr, InferenceRegistry, MAX_SHAPE_DATA_ELEMS, MergePolicy, NodeIo, ShapeData,
SymbolInterner, TypeInfo,
};
use onnx_runtime_tracer::{Args, SpanGuard, TraceContext, annotate_current_span_with};
use crate::SessionOutput;
use crate::error::{Result, SessionError};
use crate::sequence::{
ConcatPlan, SeqTensor, SequenceError, SequenceValue, SplitSpec, split_tensor, stack_new_axis,
};
use crate::tensor::{DeviceIoBinding, SharedTensorBuffer, Tensor};
fn profile_ops_enabled() -> bool {
static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ENABLED.get_or_init(|| {
std::env::var("ONNX_GENAI_PROFILE_OPS")
.is_ok_and(|value| value == "1" || value.eq_ignore_ascii_case("true"))
})
}
mod phase_profile {
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::{Mutex, OnceLock};
use std::time::Instant;
static STATE: AtomicU8 = AtomicU8::new(0);
pub fn enabled() -> bool {
match STATE.load(Ordering::Relaxed) {
1 => false,
2 => true,
_ => {
let on = std::env::var("NXRT_EXEC_PHASE_PROFILE")
.is_ok_and(|v| v == "1" || v.eq_ignore_ascii_case("true"));
STATE.store(if on { 2 } else { 1 }, Ordering::Relaxed);
on
}
}
}
#[cfg(test)]
pub(super) fn force_enabled(on: bool) {
STATE.store(if on { 2 } else { 1 }, Ordering::Relaxed);
}
#[cfg(test)]
pub(super) fn snapshot(phase: &'static str) -> Option<(u128, u64)> {
registry()
.lock()
.ok()
.and_then(|reg| reg.get(phase).map(|s| (s.total_ns, s.count)))
}
#[derive(Default, Clone, Copy)]
struct PhaseStat {
total_ns: u128,
count: u64,
}
fn registry() -> &'static Mutex<BTreeMap<&'static str, PhaseStat>> {
static REGISTRY: OnceLock<Mutex<BTreeMap<&'static str, PhaseStat>>> = OnceLock::new();
REGISTRY.get_or_init(|| Mutex::new(BTreeMap::new()))
}
pub fn record(phase: &'static str, nanos: u128) {
if !enabled() {
return;
}
if let Ok(mut reg) = registry().lock() {
let entry = reg.entry(phase).or_default();
entry.total_ns += nanos;
entry.count += 1;
}
}
pub fn all_stats() -> Vec<(&'static str, u128, u64)> {
let Ok(reg) = registry().lock() else {
return Vec::new();
};
let mut rows = reg
.iter()
.map(|(phase, stat)| (*phase, stat.total_ns, stat.count))
.collect::<Vec<_>>();
rows.sort_by_key(|row| std::cmp::Reverse(row.1));
rows
}
pub struct PhaseSpan {
phase: &'static str,
start: Option<Instant>,
}
impl PhaseSpan {
pub fn new(phase: &'static str) -> Self {
let active = enabled();
Self {
phase,
start: if active { Some(Instant::now()) } else { None },
}
}
}
impl Drop for PhaseSpan {
fn drop(&mut self) {
if let Some(start) = self.start {
record(self.phase, start.elapsed().as_nanos());
}
}
}
pub fn report_to_stderr() {
if !enabled() {
return;
}
let rows: Vec<(&'static str, PhaseStat)> = match registry().lock() {
Ok(reg) => reg.iter().map(|(n, s)| (*n, *s)).collect(),
Err(_) => return,
};
static PRINTED: AtomicBool = AtomicBool::new(false);
if PRINTED.swap(true, Ordering::Relaxed) {
return;
}
let mut rows = rows;
rows.sort_by_key(|r| std::cmp::Reverse(r.1.total_ns));
eprintln!("[nxrt-phase] phase,total_ms,calls,us/call");
for (name, stat) in &rows {
if name.ends_with("_bytes") {
continue;
}
let total_ms = stat.total_ns as f64 / 1_000_000.0;
let us_per_call = if stat.count > 0 {
(stat.total_ns as f64 / 1_000.0) / stat.count as f64
} else {
0.0
};
eprintln!(
"[nxrt-phase] {name},{total_ms:.3},{},{us_per_call:.2}",
stat.count
);
}
for (name, stat) in &rows {
if !name.ends_with("_bytes") {
continue;
}
let total_mb = stat.total_ns as f64 / (1024.0 * 1024.0);
let mb_per_call = if stat.count > 0 {
total_mb / stat.count as f64
} else {
0.0
};
eprintln!(
"[nxrt-phase] {name},total_mb={total_mb:.1},calls={},mb/call={mb_per_call:.3}",
stat.count
);
}
}
}
macro_rules! phase_span {
($phase:expr) => {
phase_profile::PhaseSpan::new($phase)
};
}
fn trace_span(name: &'static str, cat: &'static str) -> Option<SpanGuard> {
onnx_runtime_tracer::global_context()
.filter(|trace| trace.is_enabled())
.map(|trace| trace.span(name, cat))
}
pub fn exec_phase_stats() -> Vec<(&'static str, u128, u64)> {
phase_profile::all_stats()
}
pub fn print_exec_phase_profile() {
phase_profile::report_to_stderr();
}
fn host_dtype_alignment(dtype: DataType) -> usize {
match dtype {
DataType::Float16 | DataType::BFloat16 | DataType::Int16 | DataType::Uint16 => 2,
DataType::Float32 | DataType::Int32 | DataType::Uint32 | DataType::Complex64 => 4,
DataType::Float64 | DataType::Int64 | DataType::Uint64 | DataType::Complex128 => 8,
_ => 1,
}
}
fn print_op_profile(total: Duration, timings: HashMap<String, (Duration, usize)>) {
let mut timings = timings.into_iter().collect::<Vec<_>>();
timings.sort_unstable_by_key(|entry| std::cmp::Reverse(entry.1.0));
let total_ms = total.as_secs_f64() * 1_000.0;
eprintln!("[onnx-genai-profile] node execution: {total_ms:.3} ms");
eprintln!("[onnx-genai-profile] op_type,total_ms,percent,calls");
for (op_type, (elapsed, calls)) in timings {
let elapsed_ms = elapsed.as_secs_f64() * 1_000.0;
let percent = if total_ms == 0.0 {
0.0
} else {
elapsed_ms / total_ms * 100.0
};
eprintln!("[onnx-genai-profile] {op_type},{elapsed_ms:.3},{percent:.2},{calls}");
}
}
fn log_capture_segmentation(schedule: &CaptureSchedule) {
let captured = schedule.captured_segments();
let seams = schedule.segments.len() - captured;
eprintln!(
"[onnx-genai-capture] segmented CUDA graph: {captured} captured segment(s), \
{seams} eager seam(s)"
);
for boundary in &schedule.boundaries {
match boundary.node_id {
Some(id) => {
let seam_label = boundary
.seam_reason
.map(SeamReason::label)
.unwrap_or("unclassified-seam");
eprintln!(
"[onnx-genai-capture] seam node {id} ({}::{}) [{seam_label}] ran eagerly: {}",
boundary.domain, boundary.op_type, boundary.reason
);
}
None => eprintln!(
"[onnx-genai-capture] seam ({}): {}",
boundary.op_type, boundary.reason
),
}
}
}
#[derive(Debug)]
pub(crate) struct NodePlan {
pub node_id: NodeId,
pub inputs: Vec<Option<ValueId>>,
pub outputs: Vec<ValueId>,
pub input_dtypes: Vec<DataType>,
pub output_dtypes: Vec<DataType>,
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
struct KernelKey {
node: u32,
shapes: Vec<Vec<usize>>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct CacheStats {
pub entries: usize,
pub hits: u64,
pub misses: u64,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ControlFlowStats {
pub subgraph_builds: u64,
pub subgraph_runs: u64,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct DeviceAllocationCounts {
pub allocations: u64,
pub frees: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CapturePathKind {
CaptureRegion,
EagerDeviceSeam,
HostSeam,
}
impl CapturePathKind {
pub const fn label(self) -> &'static str {
match self {
Self::CaptureRegion => "capture-region",
Self::EagerDeviceSeam => "eager-device-seam",
Self::HostSeam => "host-seam",
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SeamReason {
HostControlFlowOrSequence,
UnresolvedOutputShape,
UnresolvedInputShape,
KernelNotWarmed,
KernelCaptureUnsupported,
CaptureRecordingFailed,
}
impl SeamReason {
pub const fn path_kind(self) -> CapturePathKind {
match self {
Self::HostControlFlowOrSequence => CapturePathKind::HostSeam,
Self::UnresolvedOutputShape
| Self::UnresolvedInputShape
| Self::KernelNotWarmed
| Self::CaptureRecordingFailed
| Self::KernelCaptureUnsupported => CapturePathKind::EagerDeviceSeam,
}
}
pub const fn label(self) -> &'static str {
self.path_kind().label()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CaptureDecline {
pub node_id: Option<u32>,
pub op_type: String,
pub domain: String,
pub reason: String,
pub seam_reason: Option<SeamReason>,
}
impl CaptureDecline {
fn node(
node_id: NodeId,
node: &Node,
seam_reason: SeamReason,
reason: impl Into<String>,
) -> Self {
Self {
node_id: Some(node_id.0),
op_type: node.op_type.clone(),
domain: canonical_domain(node),
reason: reason.into(),
seam_reason: Some(seam_reason),
}
}
fn graph(reason: impl Into<String>) -> Self {
Self {
node_id: None,
op_type: "<graph>".to_string(),
domain: "nxrt".to_string(),
reason: reason.into(),
seam_reason: None,
}
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CaptureDeclineReport {
pub entries: Vec<CaptureDecline>,
}
impl CaptureDeclineReport {
fn one(decline: CaptureDecline) -> Self {
Self {
entries: vec![decline],
}
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ExecutionProviderDecline {
pub node: String,
pub domain: String,
pub op_type: String,
pub reason: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ExecutionProviderFallbackReport {
pub requested_provider: String,
pub fallback_provider: String,
pub assigned_node_count: usize,
pub assigned_ops: Vec<String>,
pub declines: Vec<ExecutionProviderDecline>,
}
impl std::fmt::Display for ExecutionProviderFallbackReport {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{} nodes assigned to CPU (ops: {}) — GPU EP {} did not claim {} node(s): {}. \
Heterogeneous CUDA+CPU placement is unavailable, so the whole session uses {}",
self.assigned_node_count,
self.assigned_ops.join(", "),
self.requested_provider,
self.declines.len(),
format_cuda_coverage_issues(&self.declines),
self.fallback_provider,
)
}
}
impl std::fmt::Display for CaptureDeclineReport {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "CUDA graph capture rejected")?;
for (index, decline) in self.entries.iter().enumerate() {
if index == 0 {
write!(f, ": ")?;
} else {
write!(f, "; ")?;
}
match decline.node_id {
Some(node_id) => write!(
f,
"node {node_id} ({}::{}) — {}",
decline.domain, decline.op_type, decline.reason
)?,
None => write!(f, "{} — {}", decline.op_type, decline.reason)?,
}
}
Ok(())
}
}
pub enum DeviceGraphCaptureResult {
Captured(Vec<Option<Tensor>>),
NotCapturable(CaptureDeclineReport),
}
enum ScopedRunResult {
Executed(Vec<Option<SessionOutput>>),
NotCapturable(CaptureDeclineReport),
}
fn kernel_capture_decline(
node_id: NodeId,
node: &Node,
kernel: &dyn Kernel,
) -> Option<CaptureDecline> {
kernel.capture_support().reason().map(|reason| {
CaptureDecline::node(node_id, node, SeamReason::KernelCaptureUnsupported, reason)
})
}
fn structural_capture_decline(
node_id: NodeId,
node: &Node,
decline: StructuralCaptureDecline,
) -> CaptureDecline {
let seam_reason = match decline {
StructuralCaptureDecline::HostControlFlowOrSequence => {
SeamReason::HostControlFlowOrSequence
}
StructuralCaptureDecline::UnresolvedOutputShape => SeamReason::UnresolvedOutputShape,
StructuralCaptureDecline::UnresolvedInputShape => SeamReason::UnresolvedInputShape,
};
CaptureDecline::node(node_id, node, seam_reason, decline.reason())
}
fn capture_segmentation_logging_enabled() -> bool {
static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ENABLED.get_or_init(|| {
std::env::var("ONNX_GENAI_LOG_CAPTURE_SEGMENTS")
.is_ok_and(|value| value == "1" || value.eq_ignore_ascii_case("true"))
})
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum RunMode {
Eager,
Capture,
Replay,
}
#[derive(Clone, Copy)]
enum OpCaptureTrace<'a> {
Eager,
Captured,
Rejected(&'a str),
}
const ARG_CAPTURE_STATUS: &str = "capture_status";
const ARG_CAPTURE_REASON: &str = "capture_reason";
impl OpCaptureTrace<'_> {
fn annotate(self) {
match self {
OpCaptureTrace::Eager => {}
OpCaptureTrace::Captured => {
annotate_current_span_with(|| {
onnx_runtime_tracer::Args::new().with(ARG_CAPTURE_STATUS, "captured")
});
}
OpCaptureTrace::Rejected(reason) => {
annotate_current_span_with(|| {
onnx_runtime_tracer::Args::new()
.with(ARG_CAPTURE_STATUS, "rejected")
.with(ARG_CAPTURE_REASON, reason)
});
}
}
}
}
struct SegmentCaptureGuard<'a> {
ep: &'a dyn ExecutionProvider,
armed: bool,
}
impl<'a> SegmentCaptureGuard<'a> {
fn arm(ep: &'a dyn ExecutionProvider) -> Self {
Self { ep, armed: true }
}
fn disarm(&mut self) {
self.armed = false;
}
}
impl Drop for SegmentCaptureGuard<'_> {
fn drop(&mut self) {
if self.armed {
let _ = self.ep.abort_device_graph_capture();
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct ScheduledSegment {
start: usize,
end: usize,
captured: bool,
graph_index: usize,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
struct CaptureSchedule {
segments: Vec<ScheduledSegment>,
boundaries: Vec<CaptureDecline>,
}
impl CaptureSchedule {
fn captured_segments(&self) -> usize {
self.segments.iter().filter(|seg| seg.captured).count()
}
fn is_single_graph(&self) -> bool {
self.segments.len() == 1 && self.segments[0].captured
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct DeviceBindingSignature {
input_name: String,
binds_input: bool,
output_name: Option<String>,
dtype: DataType,
physical_shape: Vec<usize>,
device_ptr: usize,
}
#[derive(Default)]
pub(crate) struct KernelCache {
entries: HashMap<KernelKey, Box<dyn onnx_runtime_ep_api::Kernel>>,
hits: u64,
misses: u64,
}
impl KernelCache {
fn stats(&self) -> CacheStats {
CacheStats {
entries: self.entries.len(),
hits: self.hits,
misses: self.misses,
}
}
fn get_or_create(
&mut self,
node_id: NodeId,
node: &Node,
input_shapes: &[Vec<usize>],
input_dtypes: &[DataType],
constant_inputs: &[bool],
opset: u64,
ep: &dyn ExecutionProvider,
) -> Result<&dyn onnx_runtime_ep_api::Kernel> {
let key = KernelKey {
node: node_id.0,
shapes: input_shapes.to_vec(),
};
if self.entries.contains_key(&key) {
self.hits += 1;
} else {
let shape_dims: Vec<Shape> = input_shapes
.iter()
.map(|s| s.iter().map(|&d| Dim::Static(d)).collect())
.collect();
let layouts = vec![TensorLayout::contiguous(); input_shapes.len()];
if let KernelMatch::Unsupported { reason } =
ep.supports_op(node, opset, &shape_dims, input_dtypes, &layouts)
{
return Err(SessionError::unsupported_op(
node,
node_id,
opset,
ep.name(),
reason,
));
}
let mut kernel = match ep.get_kernel(node, input_shapes, opset) {
Ok(kernel) => kernel,
Err(EpError::NoEpForOp {
domain,
op_type,
opset,
}) => {
return Err(SessionError::unsupported_op(
node,
node_id,
opset,
ep.name(),
format!(
"no handler for {domain}::{op_type} at opset {opset} — add a claim+handler"
),
));
}
Err(error) => return Err(error.into()),
};
kernel.set_constant_inputs(constant_inputs);
self.entries.insert(key.clone(), kernel);
self.misses += 1;
}
Ok(self.entries.get(&key).expect("just inserted").as_ref())
}
}
pub(crate) struct Executor {
graph: Graph,
weights: Arc<WeightStore>,
ep: Arc<dyn ExecutionProvider>,
weight_handles: HashMap<ValueId, WeightHandle>,
buffers: HashMap<ValueId, DeviceBuffer>,
buffer_shapes: HashMap<ValueId, Vec<usize>>,
value_shapes: HashMap<ValueId, Shape>,
value_dtypes: HashMap<ValueId, DataType>,
plan: Vec<NodePlan>,
input_index: HashMap<String, ValueId>,
required_inputs: Vec<ValueId>,
has_symbols: bool,
cache: KernelCache,
name_index: HashMap<String, ValueId>,
subgraph_execs: HashMap<(NodeId, String), ChildExecutor>,
control_flow_stats: ControlFlowStats,
if_last_predicate: HashMap<NodeId, bool>,
device_graph_signature: Option<Vec<DeviceBindingSignature>>,
capture_schedule: Option<CaptureSchedule>,
capture_segmentation: Vec<CaptureDecline>,
control_flow_output_values: HashSet<ValueId>,
capture_cf_shapes: HashMap<ValueId, Vec<usize>>,
capture_warm_signature: Option<Vec<ExternalCaptureSig>>,
capture_warm_shapes: HashMap<ValueId, Vec<usize>>,
capture_warm_seeded: HashMap<ValueId, Vec<usize>>,
capture_quarantine_ops: HashSet<(String, String)>,
last_capture_failed_node: Option<NodeId>,
views: HashMap<ValueId, ValueView>,
pinned: HashSet<ValueId>,
sequence_values: HashSet<ValueId>,
shared_buffers: HashMap<ValueId, Arc<SharedTensorBuffer>>,
sequences: HashMap<ValueId, SequenceValue>,
seq_elem_values: HashMap<ValueId, SeqTensor>,
execution_provider_fallback_report: Option<ExecutionProviderFallbackReport>,
trace: TraceContext,
scratch_input_shapes: Vec<Vec<usize>>,
decode_memo_enabled: bool,
decode_memo_verify: bool,
decode_memo: Option<DecodePlanMemo>,
decode_memo_prev_bindings: Option<HashMap<SymbolId, usize>>,
decode_memo_last_action: DecodeMemoAction,
decode_memo_resolved: HashMap<ValueId, Vec<usize>>,
decode_memo_primed_count: u64,
decode_memo_rebuilt_count: u64,
decode_memo_replayed_count: u64,
decode_memo_ineligible_count: u64,
decode_view_plan: Option<DecodeViewPlan>,
decode_views_reused_count: u64,
decode_dispatch_elided_count: u64,
decode_view_plan_sig_mismatch_streak: u32,
decode_view_plan_disabled: bool,
}
const STAGE2_SIG_MISMATCH_LIMIT: u32 = 2;
#[derive(Clone, Debug)]
struct ValueView {
source: ValueId,
shape: Vec<usize>,
strides: Vec<i64>,
byte_offset: usize,
}
struct DecodePlanMemo {
reference_bindings: HashMap<SymbolId, usize>,
decode_varying: HashSet<SymbolId>,
invariant_shapes: HashMap<ValueId, Vec<usize>>,
variant_values: Vec<ValueId>,
canonical: HashSet<ValueId>,
reference_external_sig: Vec<DecodeBindingSig>,
}
#[derive(Clone, PartialEq, Eq)]
struct DecodeBindingSig {
vid: ValueId,
is_input: bool,
dtype: DataType,
decl_shape: Shape,
}
impl DecodePlanMemo {
fn matches(
&self,
bindings: &HashMap<SymbolId, usize>,
external_sig: &[DecodeBindingSig],
) -> bool {
if external_sig != self.reference_external_sig {
return false;
}
if bindings.len() != self.reference_bindings.len() {
return false;
}
bindings.iter().all(|(sym, &val)| {
match self.reference_bindings.get(sym) {
Some(&ref_val) => val == ref_val || self.decode_varying.contains(sym),
None => false,
}
})
}
}
struct DecodeViewPlan {
elided_nodes: HashSet<usize>,
retained_views: Vec<(ValueId, ValueView)>,
pinned_sources: Vec<ValueId>,
source_buffer_sig: Vec<(ValueId, usize, usize)>,
validated: bool,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DecodeMemoAction {
Disabled,
Primed,
Rebuilt,
Replayed,
}
fn same_symbol_keys(a: &HashMap<SymbolId, usize>, b: &HashMap<SymbolId, usize>) -> bool {
a.len() == b.len() && a.keys().all(|k| b.contains_key(k))
}
fn is_decode_growth_transition(
prev: &HashMap<SymbolId, usize>,
cur: &HashMap<SymbolId, usize>,
) -> bool {
if !same_symbol_keys(prev, cur) {
return false;
}
let mut any_grew = false;
for (sym, &c) in cur {
let p = prev[sym];
if c > p {
any_grew = true;
} else if c < p {
return false; }
}
any_grew
}
fn shape_references_any(shape: &Shape, symbols: &HashSet<SymbolId>) -> bool {
shape
.iter()
.any(|d| matches!(d, Dim::Symbolic(s) if symbols.contains(s)))
}
fn decode_memo_env_enabled() -> bool {
match std::env::var("ONNX_GENAI_DECODE_MEMO") {
Ok(value) => !matches!(
value.trim().to_ascii_lowercase().as_str(),
"0" | "false" | "off"
),
Err(_) => true,
}
}
fn decode_memo_verify_env_enabled() -> bool {
matches!(
std::env::var("ONNX_GENAI_DECODE_MEMO_VERIFY")
.ok()
.as_deref(),
Some("1") | Some("true") | Some("on")
)
}
struct InInfo {
present: bool,
dtype: DataType,
shape: Vec<usize>,
strides: Vec<i64>,
byte_offset: usize,
base_ptr: *const std::ffi::c_void,
device: onnx_runtime_ir::DeviceId,
backing: TensorBacking,
root_len: usize,
}
#[derive(Clone)]
struct ExternalValue {
dtype: DataType,
shape: Vec<usize>,
accepts_subshape: bool,
ptr: *mut std::ffi::c_void,
len: usize,
alignment: usize,
device: onnx_runtime_ir::DeviceId,
}
impl ExternalValue {
fn accepts_output(&self, dtype: DataType, shape: &[usize], bytes: usize) -> bool {
self.dtype == dtype
&& self.len >= bytes
&& if self.accepts_subshape {
shape.len() == self.shape.len()
&& shape
.iter()
.zip(&self.shape)
.all(|(&required, &capacity)| required <= capacity)
} else {
self.shape == shape
}
}
fn writable_buffer(&self) -> Result<DeviceBuffer> {
unsafe {
DeviceBuffer::from_borrowed_mut_parts(self.ptr, self.device, self.len, self.alignment)
}
.ok_or_else(|| SessionError::Internal("external output binding has a null pointer".into()))
}
}
#[derive(Default)]
struct ExternalBindings {
inputs: HashMap<ValueId, ExternalValue>,
outputs: HashMap<ValueId, ExternalValue>,
}
#[derive(Clone, PartialEq, Eq)]
struct ExternalCaptureSig {
vid: ValueId,
is_input: bool,
dtype: DataType,
shape: Vec<usize>,
ptr: usize,
len: usize,
}
impl ExternalBindings {
fn seed_capture_shapes(&self, resolved: &mut HashMap<ValueId, Vec<usize>>) {
for (&vid, value) in self.inputs.iter().chain(&self.outputs) {
resolved.entry(vid).or_insert_with(|| value.shape.clone());
}
}
fn capture_signature(&self) -> Vec<ExternalCaptureSig> {
let mut sig: Vec<ExternalCaptureSig> = self
.inputs
.iter()
.map(|(&vid, v)| (vid, true, v))
.chain(self.outputs.iter().map(|(&vid, v)| (vid, false, v)))
.map(|(vid, is_input, v)| ExternalCaptureSig {
vid,
is_input,
dtype: v.dtype,
shape: v.shape.clone(),
ptr: v.ptr as usize,
len: v.len,
})
.collect();
sig.sort_by_key(|a| (a.vid.0, a.is_input));
sig
}
}
struct CompiledChildPlan {
exec: Executor,
signature: Vec<ChildInputSignature>,
}
const CHILD_EXECUTOR_CACHE_CAPACITY: usize = 4;
#[derive(Clone, Debug, PartialEq, Eq)]
struct ChildInputSignature {
dtype: DataType,
shape: Vec<usize>,
}
pub(crate) struct ChildExecutor {
name: String,
template: Graph,
inherited_opsets: HashMap<String, u64>,
weights: Arc<WeightStore>,
ep: Arc<dyn ExecutionProvider>,
formal_names: Vec<String>,
capture_names: Vec<String>,
input_names: Vec<String>,
compiled: Vec<CompiledChildPlan>,
builds: u64,
runs: u64,
trace: TraceContext,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct ChildExecutorStats {
pub builds: u64,
pub runs: u64,
}
struct PreparedSubgraph {
key: (NodeId, String),
captures: HashMap<String, Tensor>,
}
fn view_bounds(
shape: &[usize],
strides: &[i64],
byte_offset: usize,
dtype: DataType,
buffer_len: usize,
) -> Result<()> {
let esize = dtype.byte_size();
if esize == 0 {
let numel: usize = shape.iter().product();
let need = byte_offset + dtype.storage_bytes(numel);
if need > buffer_len {
return Err(SessionError::from(
onnx_runtime_ep_api::EpError::InvalidTensorView {
reason: format!(
"sub-byte view needs {need} bytes but backing allocation is {buffer_len}"
),
},
));
}
return Ok(());
}
view_in_bounds(shape, strides, byte_offset, esize, buffer_len)?;
Ok(())
}
fn gather_view(
src: &[u8],
shape: &[usize],
strides: &[i64],
byte_offset: usize,
esize: usize,
) -> Vec<u8> {
let n: usize = shape.iter().product();
let mut out = vec![0u8; n * esize];
if n == 0 {
return out;
}
let rank = shape.len();
let mut idx = vec![0usize; rank];
let mut w = 0usize;
loop {
let mut off = byte_offset as i64;
for d in 0..rank {
off += strides[d] * idx[d] as i64 * esize as i64;
}
let s = off as usize;
out[w..w + esize].copy_from_slice(&src[s..s + esize]);
w += esize;
let mut carried = true;
for axis in (0..rank).rev() {
idx[axis] += 1;
if idx[axis] < shape[axis] {
carried = false;
break;
}
idx[axis] = 0;
}
if carried {
break;
}
}
out
}
fn checked_numel(dims: &[usize], value: impl FnOnce() -> String) -> Result<usize> {
let mut acc = 1usize;
for &d in dims {
acc = match acc.checked_mul(d) {
Some(n) => n,
None => {
return Err(SessionError::ShapeOverflow {
value: value(),
dims: dims.to_vec(),
});
}
};
}
Ok(acc)
}
fn checked_storage_bytes(
dtype: DataType,
numel: usize,
value: impl FnOnce() -> String,
dims: &[usize],
) -> Result<usize> {
dtype
.checked_storage_bytes(numel)
.ok_or_else(|| SessionError::ShapeOverflow {
value: value(),
dims: dims.to_vec(),
})
}
fn effective_opset(graph: &Graph, node: &Node) -> u64 {
graph
.opset_imports
.get(node.domain.as_str())
.copied()
.unwrap_or_else(|| {
unreachable!(
"internal invariant violated: node #{} ({}::{}) has no opset import",
node.id.0,
if node.domain.is_empty() {
"ai.onnx"
} else {
&node.domain
},
node.op_type
)
})
}
fn substitute(shape: &Shape, bindings: &HashMap<SymbolId, usize>) -> Option<Vec<usize>> {
shape
.iter()
.map(|d| match d {
Dim::Static(n) => Some(*n),
Dim::Symbolic(s) => bindings.get(s).copied(),
})
.collect()
}
fn substitute_into(
shape: &Shape,
bindings: &HashMap<SymbolId, usize>,
out: &mut Vec<usize>,
) -> bool {
out.clear();
for d in shape {
match d {
Dim::Static(n) => out.push(*n),
Dim::Symbolic(s) => match bindings.get(s) {
Some(&v) => out.push(v),
None => {
out.clear();
return false;
}
},
}
}
true
}
fn bytes_as_i64(bytes: &[u8], dtype: DataType) -> Option<Vec<i64>> {
match dtype {
DataType::Int64 => Some(
bytes
.chunks_exact(8)
.map(|c| i64::from_le_bytes(c.try_into().unwrap()))
.collect(),
),
DataType::Int32 => Some(
bytes
.chunks_exact(4)
.map(|c| i32::from_le_bytes(c.try_into().unwrap()) as i64)
.collect(),
),
_ => None,
}
}
fn bytes_as_f64(bytes: &[u8], dtype: DataType) -> Option<Vec<f64>> {
match dtype {
DataType::Float32 => Some(
bytes
.chunks_exact(4)
.map(|c| f32::from_le_bytes(c.try_into().unwrap()) as f64)
.collect(),
),
DataType::Float64 => Some(
bytes
.chunks_exact(8)
.map(|c| f64::from_le_bytes(c.try_into().unwrap()))
.collect(),
),
_ => None,
}
}
fn bounded_shape_input(dtype: DataType, shape: &[usize]) -> bool {
if !matches!(dtype, DataType::Int32 | DataType::Int64) {
return false;
}
if shape.len() > 1 {
return false;
}
shape
.iter()
.try_fold(1usize, |count, &dim| count.checked_mul(dim))
.is_some_and(|count| count <= MAX_SHAPE_DATA_ELEMS)
}
fn reads_float_shape_input(node: &Node, input_index: usize, opset: u64) -> bool {
node.is_default_domain()
&& ((node.op_type == "Resize" && input_index == if opset == 10 { 1 } else { 2 })
|| (node.op_type == "NonMaxSuppression" && matches!(input_index, 0 | 1 | 3 | 4)))
}
fn kernel_input_uses_physical_capacity(node: &Node, input_index: usize) -> bool {
if node.domain == "com.microsoft"
&& node.op_type == "GroupQueryAttention"
&& matches!(input_index, 3 | 4)
{
return true;
}
node.is_default_domain()
&& node.op_type == "Attention"
&& matches!(input_index, 4 | 5)
&& node.inputs.get(3).is_some_and(Option::is_some)
&& node
.attr("is_causal")
.and_then(|attr| attr.as_int())
.unwrap_or(0)
== 0
|| (
node.domain == "pkg.nxrt"
&& node.op_type == "IndexShare"
&& matches!(input_index, 3 | 4)
&& node.outputs.len() == 3
&& node.inputs.get(6).is_some_and(Option::is_some)
)
}
fn kernel_input_uses_padded_capacity(node: &Node, input_index: usize) -> bool {
node.is_default_domain()
&& input_index == 0
&& matches!(node.op_type.as_str(), "Shape" | "ReduceSum")
}
fn runtime_elementwise_output_shape(
node: &Node,
input_shapes: &[Vec<usize>],
) -> Option<std::result::Result<Vec<usize>, onnx_runtime_ir::IrError>> {
if !node.is_default_domain() {
return None;
}
let input_count = match node.op_type.as_str() {
"Add" | "Sub" | "Mul" | "Div" | "Pow" | "Mod" | "BitShift" | "Less" | "Greater"
| "Equal" | "And" | "Or" | "Xor" | "LessOrEqual" | "GreaterOrEqual" => 2,
"Where" => 3,
"Min" | "Max" | "Sum" | "Mean" => input_shapes.len(),
_ => return None,
};
if input_count == 0 || input_shapes.len() < input_count {
return None;
}
let mut shape = input_shapes[0].clone();
for input in &input_shapes[1..input_count] {
shape = match broadcast_shapes(&shape, input) {
Ok(shape) => shape,
Err(error) => return Some(Err(error)),
};
}
Some(Ok(shape))
}
fn dynamic_output_shapes(
node: &Node,
input_shapes: &[Vec<usize>],
input_dtypes: &[DataType],
input_values: &[Option<Vec<i64>>],
input_float_values: &[Option<Vec<f64>>],
opset: u64,
) -> Option<Vec<Vec<usize>>> {
match node.op_type.as_str() {
"Resize" if node.is_default_domain() => {
let input = input_shapes.first()?;
let rank = input.len();
let axes = if let Some(raw) = node.attr("axes").and_then(Attribute::as_ints) {
let mut axes = Vec::with_capacity(raw.len());
for &axis in raw {
let axis = if axis < 0 { axis + rank as i64 } else { axis };
let axis = usize::try_from(axis).ok()?;
if axis >= rank || axes.contains(&axis) {
return None;
}
axes.push(axis);
}
if axes.is_empty() {
(0..rank).collect()
} else {
axes
}
} else {
(0..rank).collect()
};
let scales_index = if opset == 10 { 1 } else { 2 };
let scales = input_float_values
.get(scales_index)
.and_then(|values| values.as_deref())
.filter(|values| !values.is_empty());
let sizes = (opset >= 11)
.then(|| input_values.get(3).and_then(|values| values.as_deref()))
.flatten()
.filter(|values| !values.is_empty());
if scales.is_some() == sizes.is_some() {
return None;
}
let mut output = input.clone();
if let Some(scales) = scales {
if scales.len() != axes.len()
|| node
.attr("keep_aspect_ratio_policy")
.and_then(Attribute::as_str)
.is_some_and(|policy| policy != "stretch")
{
return None;
}
for (&axis, &scale) in axes.iter().zip(scales) {
if !scale.is_finite() || scale <= 0.0 {
return None;
}
let extent = input[axis] as f64 * scale;
if extent > usize::MAX as f64 {
return None;
}
output[axis] = extent.floor() as usize;
}
} else {
let sizes = sizes?;
if sizes.len() != axes.len() {
return None;
}
let requested = sizes
.iter()
.map(|&size| usize::try_from(size).ok().filter(|&size| size > 0))
.collect::<Option<Vec<_>>>()?;
match node
.attr("keep_aspect_ratio_policy")
.and_then(Attribute::as_str)
.unwrap_or("stretch")
{
"stretch" => {
for (&axis, &size) in axes.iter().zip(&requested) {
output[axis] = size;
}
}
policy @ ("not_larger" | "not_smaller") => {
if axes.iter().any(|&axis| input[axis] == 0) {
return None;
}
let (numerator, denominator) = axes
.iter()
.zip(&requested)
.map(|(&axis, &size)| (size, input[axis]))
.reduce(|left, right| {
let order = (left.0 as u128 * right.1 as u128)
.cmp(&(right.0 as u128 * left.1 as u128));
if (policy == "not_larger" && order.is_le())
|| (policy == "not_smaller" && order.is_ge())
{
left
} else {
right
}
})?;
if denominator == 0 {
return None;
}
for &axis in &axes {
let product = (input[axis] as u128).checked_mul(numerator as u128)?;
output[axis] = usize::try_from(
(product + denominator as u128 / 2) / denominator as u128,
)
.ok()?;
}
}
_ => return None,
}
}
Some(vec![output])
}
"Slice" if node.is_default_domain() => {
let data_shape = input_shapes.first()?;
let starts = input_values.get(1)?.as_ref()?;
let ends = input_values.get(2)?.as_ref()?;
let (axes, steps) = onnx_runtime_ep_cpu::slice_axes_steps(
starts.len(),
input_values.get(3).and_then(|v| v.as_deref()),
input_values.get(4).and_then(|v| v.as_deref()),
);
let plan =
onnx_runtime_ep_cpu::slice_plan(data_shape, starts, ends, &axes, &steps).ok()?;
let count: Vec<usize> = plan.iter().map(|p| p.count).collect();
Some(vec![count])
}
"NonMaxSuppression" if node.is_default_domain() => {
let boxes_shape = input_shapes.first()?;
let scores_shape = input_shapes.get(1)?;
let boxes = input_float_values.first()?.as_ref()?;
let scores = input_float_values.get(1)?.as_ref()?;
let max_output_boxes_per_class = input_values
.get(2)
.and_then(|value| value.as_ref())
.filter(|value| value.len() == 1)
.map(|value| value[0])
.unwrap_or(0);
let iou_threshold = input_float_values
.get(3)
.and_then(|value| value.as_ref())
.filter(|value| value.len() == 1)
.map(|value| value[0] as f32)
.unwrap_or(0.0);
let score_threshold = input_float_values
.get(4)
.and_then(|value| value.as_ref())
.filter(|value| value.len() == 1)
.map(|value| value[0] as f32)
.unwrap_or(f32::NEG_INFINITY);
let center_point_box = node
.attr("center_point_box")
.and_then(Attribute::as_int)
.unwrap_or(0);
let boxes = boxes.iter().map(|&value| value as f32).collect::<Vec<_>>();
let scores = scores.iter().map(|&value| value as f32).collect::<Vec<_>>();
let selected = onnx_runtime_ep_cpu::non_max_suppression(
&boxes,
boxes_shape,
&scores,
scores_shape,
max_output_boxes_per_class,
iou_threshold,
score_threshold,
center_point_box,
)
.ok()?;
Some(vec![vec![selected.len(), 3]])
}
"GroupQueryAttention" if node.domain == "com.microsoft" => {
let query = input_shapes.first()?;
let past_key = input_shapes.get(3)?;
if query.len() != 3 || past_key.len() != 4 {
return None;
}
let num_heads = usize::try_from(node.attr("num_heads")?.as_int()?).ok()?;
let kv_heads = usize::try_from(node.attr("kv_num_heads")?.as_int()?).ok()?;
if num_heads == 0 || kv_heads == 0 {
return None;
}
let (output, head_dim) = if node.inputs.get(1).and_then(|input| *input).is_some() {
let key = input_shapes.get(1)?;
if key.len() != 3 || !key[2].is_multiple_of(kv_heads) {
return None;
}
(query.clone(), key[2] / kv_heads)
} else {
let packed_heads = num_heads.checked_add(kv_heads.checked_mul(2)?)?;
if !query[2].is_multiple_of(packed_heads) {
return None;
}
let head_dim = query[2] / packed_heads;
(
vec![query[0], query[1], head_dim.checked_mul(num_heads)?],
head_dim,
)
};
let total_sequence_values = input_values.get(6)?.as_ref()?;
if total_sequence_values.len() != 1 {
return None;
}
let total_sequence = usize::try_from(total_sequence_values[0]).ok()?;
let present_sequence = past_key[2].max(total_sequence);
let present = vec![query[0], kv_heads, present_sequence, head_dim];
let mut shapes = vec![output];
if node.outputs.len() >= 2 {
shapes.push(present.clone());
}
if node.outputs.len() >= 3 {
shapes.push(present);
}
Some(shapes)
}
_ => {
let inputs = node
.inputs
.iter()
.enumerate()
.map(|(i, input)| {
if input.is_none() {
return Some(NodeIo::default());
}
let shape = input_shapes
.get(i)?
.iter()
.map(|&dim| i64::try_from(dim).ok().map(DimExpr::constant))
.collect::<Option<Vec<_>>>()?;
let dtype = *input_dtypes.get(i)?;
let shape_data = input_values.get(i)?.as_ref().and_then(|values| {
let elems = values
.iter()
.copied()
.map(DimExpr::constant)
.collect::<Vec<_>>();
match input_shapes[i].as_slice() {
[] if elems.len() == 1 => {
Some(ShapeData::scalar(dtype, elems[0].clone()))
}
[len] if *len == elems.len() => Some(ShapeData::vector(dtype, elems)),
_ => None,
}
});
Some(NodeIo {
type_info: Some(TypeInfo::new(dtype, shape)),
shape_data,
})
})
.collect::<Option<Vec<_>>>()?;
let mut imports = HashMap::new();
imports.insert(node.domain.clone(), opset);
let mut interner = SymbolInterner::new(0x8000_0000);
static REGISTRY: std::sync::OnceLock<InferenceRegistry> = std::sync::OnceLock::new();
REGISTRY
.get_or_init(InferenceRegistry::default_registry)
.infer_node(node, &imports, inputs, MergePolicy::Strict, &mut interner)
.ok()?
.into_iter()
.map(|output| {
output
.type_info?
.shape
.into_iter()
.map(|dim| usize::try_from(dim.as_const()?).ok())
.collect()
})
.collect()
}
}
}
fn fuse_silu_patterns(graph: &mut Graph) -> usize {
let sigmoid_ids: Vec<NodeId> = graph
.nodes
.iter()
.filter_map(|(id, node)| {
(node.op_type == "Sigmoid"
&& node.is_default_domain()
&& node.inputs.len() == 1
&& node.outputs.len() == 1)
.then_some(id)
})
.collect();
let mut fused = 0;
for sigmoid_id in sigmoid_ids {
let Some(sigmoid) = graph.try_node(sigmoid_id) else {
continue;
};
let Some(x) = sigmoid.inputs[0] else {
continue;
};
let sigmoid_output = sigmoid.outputs[0];
if graph.outputs.contains(&sigmoid_output) {
continue;
}
let consumers = graph.consumers(sigmoid_output);
if consumers.len() != 1 {
continue;
}
let mul_id = consumers[0];
let mul = graph.node(mul_id);
if mul.op_type != "Mul"
|| !mul.is_default_domain()
|| mul.inputs.len() != 2
|| mul.outputs.len() != 1
|| !((mul.inputs[0] == Some(x) && mul.inputs[1] == Some(sigmoid_output))
|| (mul.inputs[1] == Some(x) && mul.inputs[0] == Some(sigmoid_output)))
{
continue;
}
let mut silu = mul.clone();
silu.op_type = "Silu".to_string();
silu.domain = "com.microsoft".to_string();
silu.inputs = vec![Some(x)];
silu.attributes.clear();
graph.replace_node(mul_id, silu);
graph.remove_node(sigmoid_id);
fused += 1;
}
if fused != 0 {
graph
.opset_imports
.entry("com.microsoft".to_string())
.or_insert(1);
}
fused
}
struct WeightStoreInitializerResolver(Arc<WeightStore>);
impl InitializerResolver for WeightStoreInitializerResolver {
fn bytes<'a>(&'a self, weight: &'a onnx_runtime_ir::WeightRef) -> Option<&'a [u8]> {
self.0.bytes(weight)
}
}
fn run_ep_scoped_passes(
graph: &mut Graph,
weights: &Arc<WeightStore>,
ep: &dyn ExecutionProvider,
) -> Result<()> {
let passes = ep.custom_passes();
if passes.is_empty() {
return Ok(());
}
let resolver = Arc::new(WeightStoreInitializerResolver(Arc::clone(weights)));
let context = onnx_runtime_optimizer::PassContext::new().with_initializer_resolver(resolver);
onnx_runtime_optimizer::run_passes(graph, &passes, &context)?;
let registry = InferenceRegistry::default_registry();
let opset_imports = graph.opset_imports.clone();
let mut refreshed = graph.clone();
if registry
.infer_graph(&mut refreshed, &opset_imports, MergePolicy::Permissive)
.is_ok()
{
*graph = refreshed;
}
Ok(())
}
fn validate_if_branch_outputs(graph: &Graph, node: &Node) -> Result<()> {
let Some(then_branch) = graph.subgraphs.get(&(node.id, "then_branch".to_string())) else {
return Ok(());
};
let Some(else_branch) = graph.subgraphs.get(&(node.id, "else_branch".to_string())) else {
return Ok(());
};
if then_branch.outputs.len() != else_branch.outputs.len() {
return Err(SessionError::ControlFlow {
op: "If".to_string(),
reason: format!(
"branches declare different output counts: then_branch has {}, \
else_branch has {}",
then_branch.outputs.len(),
else_branch.outputs.len()
),
});
}
if then_branch.outputs.len() != node.outputs.len() {
return Err(SessionError::ControlFlow {
op: "If".to_string(),
reason: format!(
"node declares {} output(s), but each branch declares {}",
node.outputs.len(),
then_branch.outputs.len()
),
});
}
for (index, (&then_output, &else_output)) in then_branch
.outputs
.iter()
.zip(&else_branch.outputs)
.enumerate()
{
if then_branch.value_type_is_known(then_output)
&& else_branch.value_type_is_known(else_output)
{
let then_dtype = then_branch.value(then_output).dtype;
let else_dtype = else_branch.value(else_output).dtype;
if then_dtype != else_dtype {
return Err(SessionError::ControlFlow {
op: "If".to_string(),
reason: format!(
"branches declare different dtypes for output {index}: \
then_branch is {then_dtype:?}, else_branch is {else_dtype:?}"
),
});
}
}
}
Ok(())
}
fn validate_control_flow_signatures(graph: &Graph) -> Result<()> {
for (_, node) in graph.nodes.iter() {
if node.op_type == "If" && matches!(node.domain.as_str(), "" | "ai.onnx") {
validate_if_branch_outputs(graph, node)?;
}
}
for subgraph in graph.subgraphs.values() {
validate_control_flow_signatures(subgraph)?;
}
Ok(())
}
fn reject_unsupported_operators(graph: &Graph, ep: &dyn ExecutionProvider) -> Result<()> {
if ep.device_type() == DeviceType::Cuda {
return Ok(());
}
for (node_id, node) in graph.nodes.iter() {
if onnx_runtime_loader::is_ep_context_op(&node.op_type, &node.domain)
|| is_control_flow_op(&node.op_type, &node.domain)
|| is_sequence_op(&node.op_type, &node.domain)
{
continue;
}
let shapes = node
.inputs
.iter()
.map(|input| {
input
.map(|value| graph.value(value).shape.clone())
.unwrap_or_default()
})
.collect::<Vec<_>>();
if !shapes.iter().all(|shape| as_static_shape(shape).is_some()) {
continue;
}
let input_dtypes = node
.inputs
.iter()
.map(|input| {
input
.map(|value| graph.value(value).dtype)
.unwrap_or(DataType::Undefined)
})
.collect::<Vec<_>>();
let layouts = vec![TensorLayout::contiguous(); shapes.len()];
let opset = effective_opset(graph, node);
if let KernelMatch::Unsupported { reason } =
ep.supports_op(node, opset, &shapes, &input_dtypes, &layouts)
{
return Err(SessionError::unsupported_op(
node,
node_id,
opset,
ep.name(),
reason,
));
}
}
Ok(())
}
fn cuda_fallback_report(
graph: &Graph,
ep: &dyn ExecutionProvider,
) -> Option<ExecutionProviderFallbackReport> {
if ep.device_type() != DeviceType::Cuda {
return None;
}
let mut issues = Vec::new();
collect_cuda_coverage_issues(graph, graph, ep, "graph", &mut issues);
if issues.is_empty() {
return None;
}
let mut assigned_ops = BTreeSet::new();
let assigned_node_count = collect_executable_ops(graph, &mut assigned_ops);
Some(ExecutionProviderFallbackReport {
requested_provider: ep.name().to_string(),
fallback_provider: "cpu_ep".to_string(),
assigned_node_count,
assigned_ops: assigned_ops.into_iter().collect(),
declines: issues,
})
}
fn collect_executable_ops(graph: &Graph, ops: &mut BTreeSet<String>) -> usize {
let mut count = 0;
for (_, node) in graph.nodes.iter() {
if !onnx_runtime_loader::is_ep_context_op(&node.op_type, &node.domain) {
count += 1;
ops.insert(format!("{}::{}", canonical_domain(node), node.op_type));
}
}
for subgraph in graph.subgraphs.values() {
count += collect_executable_ops(subgraph, ops);
}
count
}
fn format_cuda_coverage_issues(issues: &[ExecutionProviderDecline]) -> String {
const MAX_EXAMPLES_PER_CLASS: usize = 3;
let mut groups: BTreeMap<(String, String, String), Vec<String>> = BTreeMap::new();
for issue in issues {
groups
.entry((
issue.domain.clone(),
issue.op_type.clone(),
issue.reason.clone(),
))
.or_default()
.push(issue.node.clone());
}
groups
.into_iter()
.map(|((domain, op_type, reason), mut nodes)| {
nodes.sort();
let count = nodes.len();
nodes.truncate(MAX_EXAMPLES_PER_CLASS);
format!(
"{domain}::{op_type}: {reason} [count={count}; examples: {}]",
nodes.join(", ")
)
})
.collect::<Vec<_>>()
.join("; ")
}
fn collect_cuda_coverage_issues(
graph: &Graph,
opset_graph: &Graph,
ep: &dyn ExecutionProvider,
scope: &str,
issues: &mut Vec<ExecutionProviderDecline>,
) {
for (node_id, node) in graph.nodes.iter() {
if onnx_runtime_loader::is_ep_context_op(&node.op_type, &node.domain)
|| is_control_flow_op(&node.op_type, &node.domain)
|| is_sequence_op(&node.op_type, &node.domain)
{
continue;
}
let shapes = node
.inputs
.iter()
.map(|input| {
input
.map(|value| graph.value(value).shape.clone())
.unwrap_or_default()
})
.collect::<Vec<_>>();
let layouts = node
.inputs
.iter()
.map(|input| {
input
.map(|value| graph.value(value).layout.clone())
.unwrap_or_else(TensorLayout::contiguous)
})
.collect::<Vec<_>>();
let input_dtypes = node
.inputs
.iter()
.map(|input| {
input
.map(|value| graph.value(value).dtype)
.unwrap_or(DataType::Undefined)
})
.collect::<Vec<_>>();
let opset = effective_opset(opset_graph, node);
if let KernelMatch::Unsupported { reason } =
ep.supports_op(node, opset, &shapes, &input_dtypes, &layouts)
{
issues.push(ExecutionProviderDecline {
node: format_node_identity(scope, node_id, node),
domain: canonical_domain(node),
op_type: node.op_type.clone(),
reason: reason.into_owned(),
});
continue;
}
let Some(concrete_shapes) = shapes
.iter()
.map(|shape| as_static_shape(shape))
.collect::<Option<Vec<_>>>()
else {
continue;
};
if let Err(error) = ep.get_kernel(node, &concrete_shapes, opset) {
issues.push(ExecutionProviderDecline {
node: format_node_identity(scope, node_id, node),
domain: canonical_domain(node),
op_type: node.op_type.clone(),
reason: format!("kernel creation failed: {error}"),
});
}
}
for ((node_id, attribute), subgraph) in &graph.subgraphs {
let sub_scope = format!("{scope}/node#{}/{}", node_id.0, attribute);
collect_cuda_coverage_issues(subgraph, opset_graph, ep, &sub_scope, issues);
}
}
fn canonical_domain(node: &Node) -> String {
if node.domain.is_empty() {
"ai.onnx".to_string()
} else {
node.domain.clone()
}
}
fn format_node_identity(scope: &str, node_id: NodeId, node: &Node) -> String {
if node.name.is_empty() {
format!("{scope}/node#{}", node_id.0)
} else {
format!("{scope}/node#{} {:?}", node_id.0, node.name)
}
}
fn build_lazy_weight_handles(
graph: &Graph,
weights: &Arc<WeightStore>,
ep: &dyn ExecutionProvider,
) -> Result<HashMap<ValueId, WeightHandle>> {
let capabilities = ep.capabilities();
if !capabilities.advertises(onnx_runtime_ep_api::NXRT_WEIGHT_PAGING_CAPABILITY) {
return Ok(HashMap::new());
}
let boundary = LazyWeightBoundary::BlockQuantizedMoe;
let mut handles = HashMap::new();
for (&value, weight) in &graph.initializers {
let graph_value = graph.value(value);
let consumers = graph.consumers(value);
let lazy_only = graph_value.producer.is_none()
&& !graph.outputs.contains(&value)
&& !consumers.is_empty()
&& consumers.into_iter().all(|consumer| {
let node = graph.node(consumer);
boundary.matches(&node.domain, &node.op_type)
});
if !lazy_only {
continue;
}
let Some((mapping_id, offset, len)) = weights.external_mmap_provenance(weight) else {
continue;
};
let region = ExternalMmapRegion {
mapping_id,
offset,
len,
};
let dtype = weight.dtype();
let shape = weight.dims().to_vec();
let weight = weight.clone();
let store = Arc::clone(weights);
let lazy = LazyWeight::block_quantized_moe(vec![region], move || {
let bytes = store.bytes(&weight).ok_or_else(|| {
onnx_runtime_ep_api::WeightHandleError::InvalidResident(
"external weight bytes are no longer available".into(),
)
})?;
ResidentWeight::new(dtype, shape.clone(), Arc::<[u8]>::from(bytes))
})
.map_err(|error| {
SessionError::Internal(format!(
"cannot create lazy weight handle for value#{}: {error}",
value.0
))
})?;
handles.insert(value, WeightHandle::Lazy(lazy));
}
Ok(handles)
}
impl Executor {
pub(crate) fn build(
graph: Graph,
weights: Arc<WeightStore>,
ep: Arc<dyn ExecutionProvider>,
) -> Result<Self> {
Self::build_with_cuda_requirement(
graph,
weights,
ep,
onnx_genai_runtime_config::runtime_config().require_cuda,
)
}
fn build_with_cuda_requirement(
mut graph: Graph,
weights: Arc<WeightStore>,
mut ep: Arc<dyn ExecutionProvider>,
require_cuda: bool,
) -> Result<Self> {
let mut placement_span = trace_span("session.node_placement", "session");
let requested_provider = placement_span.as_ref().map(|_| ep.name().to_string());
let requested_device = placement_span
.as_ref()
.map(|_| ep.device_type().trace_name().into_owned());
let nodes_before_placement = graph.num_nodes();
validate_control_flow_signatures(&graph)?;
graph.topological_order()?;
reject_unsupported_operators(&graph, ep.as_ref())?;
let silu_fused = fuse_silu_patterns(&mut graph);
let graph_before_ep_passes = graph.clone();
let ep_pass_nodes_before = graph.num_nodes();
run_ep_scoped_passes(&mut graph, &weights, ep.as_ref())?;
let ep_pass_nodes_after = graph.num_nodes();
let mut execution_provider_fallback_report = cuda_fallback_report(&graph, ep.as_ref());
let fallback_declines = execution_provider_fallback_report
.as_ref()
.map_or(0, |report| report.declines.len());
if let Some(report) = &mut execution_provider_fallback_report {
if require_cuda {
return Err(SessionError::HeterogeneousPlacementRequired {
unsupported_nodes: report.to_string(),
});
}
graph = graph_before_ep_passes;
ep = auto_detect_cpu_ep()?;
run_ep_scoped_passes(&mut graph, &weights, ep.as_ref())?;
let mut assigned_ops = BTreeSet::new();
report.assigned_node_count = collect_executable_ops(&graph, &mut assigned_ops);
report.assigned_ops = assigned_ops.into_iter().collect();
eprintln!(
"[onnx-genai-warning] {report}. Set ONNX_GENAI_REQUIRE_CUDA=1 to reject this fallback"
);
}
if let Some(span) = placement_span.as_mut() {
let mut assigned_ops = BTreeSet::new();
let assigned_nodes = collect_executable_ops(&graph, &mut assigned_ops);
span.set_args(
Args::new()
.with("requested_provider", requested_provider.unwrap_or_default())
.with("requested_device", requested_device.unwrap_or_default())
.with("selected_provider", ep.name().to_string())
.with(
"selected_device",
ep.device_type().trace_name().into_owned(),
)
.with("nodes_before", nodes_before_placement as u64)
.with("nodes_after", graph.num_nodes() as u64)
.with("ep_pass_nodes_before", ep_pass_nodes_before as u64)
.with("ep_pass_nodes_after", ep_pass_nodes_after as u64)
.with("silu_fused", silu_fused as u64)
.with("assigned_nodes", assigned_nodes as u64)
.with("assigned_op_classes", assigned_ops.len() as u64)
.with("fallback_declines", fallback_declines as u64),
);
}
drop(placement_span);
let order = graph.topological_order()?;
let weight_handles = {
let mut span = trace_span("session.lazy_weight_handles", "session");
let handles = build_lazy_weight_handles(&graph, &weights, ep.as_ref())?;
if let Some(span) = span.as_mut() {
span.set_args(
Args::new()
.with("handles", handles.len() as u64)
.with("initializers", graph.initializers.len() as u64),
);
}
handles
};
let mut value_shapes: HashMap<ValueId, Shape> = HashMap::new();
let mut value_dtypes: HashMap<ValueId, DataType> = HashMap::new();
let mut buffers: HashMap<ValueId, DeviceBuffer> = HashMap::new();
let mut buffer_shapes: HashMap<ValueId, Vec<usize>> = HashMap::new();
let init_align = TensorLayout::contiguous().alignment;
let mut initializer_span = trace_span("session.initializer_buffers", "session");
let mut initializer_count = 0_u64;
let mut initializer_bytes = 0_u64;
let mut borrowed_initializers = 0_u64;
let mut copied_initializers = 0_u64;
let mut lazy_initializers = 0_u64;
for (&vid, weight) in &graph.initializers {
let dtype = weight.dtype();
let dims = weight.dims().to_vec();
value_dtypes.insert(vid, dtype);
value_shapes.insert(vid, dims.iter().map(|&d| Dim::Static(d)).collect());
if !ep.device_id().is_host_accessible() && weight_handles.contains_key(&vid) {
if initializer_span.is_some() {
lazy_initializers += 1;
}
continue;
}
let bytes = weights.bytes(weight).ok_or_else(|| {
SessionError::Internal(format!("weight bytes unavailable for value#{}", vid.0))
})?;
if initializer_span.is_some() {
initializer_count += 1;
initializer_bytes += bytes.len() as u64;
}
let producer_less = graph.value(vid).producer.is_none();
let borrow_align = if matches!(weight, WeightRef::External { .. }) {
host_dtype_alignment(dtype)
} else {
init_align
};
let buf = if ep.device_id().is_host_accessible()
&& producer_less
&& !bytes.is_empty()
&& (bytes.as_ptr() as usize).is_multiple_of(borrow_align)
{
if initializer_span.is_some() {
borrowed_initializers += 1;
}
unsafe {
DeviceBuffer::from_borrowed_parts(
bytes.as_ptr() as *mut std::ffi::c_void,
ep.device_id(),
bytes.len(),
borrow_align,
)
}
} else {
if initializer_span.is_some() {
copied_initializers += 1;
}
let mut owned = ep.allocate(bytes.len().max(1), init_align)?;
ep.copy_from_host(bytes, &mut owned)?;
owned
};
buffer_shapes.insert(vid, dims);
buffers.insert(vid, buf);
}
if let Some(span) = initializer_span.as_mut() {
span.set_args(
Args::new()
.with("initializers", initializer_count)
.with("bytes", initializer_bytes)
.with("borrowed_initializers", borrowed_initializers)
.with("copied_initializers", copied_initializers)
.with("lazy_initializers", lazy_initializers)
.with("buffers", buffers.len() as u64),
);
}
for &vid in &graph.inputs {
value_shapes
.entry(vid)
.or_insert_with(|| graph.value(vid).shape.clone());
value_dtypes.entry(vid).or_insert(graph.value(vid).dtype);
}
for &nid in &order {
for &out in &graph.node(nid).outputs {
value_shapes
.entry(out)
.or_insert_with(|| graph.value(out).shape.clone());
value_dtypes.entry(out).or_insert(graph.value(out).dtype);
}
}
let has_symbols = value_shapes.values().any(|s| as_static_shape(s).is_none());
let mut sequence_values: HashSet<ValueId> = HashSet::new();
for &nid in &order {
let node = graph.node(nid);
if produces_sequence_output(&node.op_type, &node.domain) {
for &out in &node.outputs {
sequence_values.insert(out);
}
}
}
let mut control_flow_output_values: HashSet<ValueId> = HashSet::new();
for &nid in &order {
let node = graph.node(nid);
if is_control_flow_op(&node.op_type, &node.domain) {
for &out in &node.outputs {
control_flow_output_values.insert(out);
}
}
}
let mut plan_span = trace_span("session.execution_plan", "session");
let mut plan = Vec::with_capacity(order.len());
let mut skipped_epcontext = 0_u64;
for &nid in &order {
let node = graph.node(nid);
if onnx_runtime_loader::is_ep_context_op(&node.op_type, &node.domain) {
if plan_span.is_some() {
skipped_epcontext += 1;
}
continue;
}
let mut slots: Vec<Option<ValueId>> = node.inputs.clone();
while matches!(slots.last(), Some(None)) {
slots.pop();
}
let inputs = slots;
let outputs: Vec<ValueId> = node.outputs.clone();
let input_dtypes: Vec<DataType> = inputs
.iter()
.map(|v| {
v.map(|vid| value_dtypes[&vid])
.unwrap_or(DataType::Undefined)
})
.collect();
let output_dtypes: Vec<DataType> = outputs.iter().map(|v| value_dtypes[v]).collect();
plan.push(NodePlan {
node_id: nid,
inputs,
outputs,
input_dtypes,
output_dtypes,
});
}
if let Some(span) = plan_span.as_mut() {
span.set_args(
Args::new()
.with("topological_nodes", order.len() as u64)
.with("plan_len", plan.len() as u64)
.with("skipped_epcontext_nodes", skipped_epcontext)
.with("values", graph.values.len() as u64)
.with("inputs", graph.inputs.len() as u64)
.with("outputs", graph.outputs.len() as u64)
.with("has_symbols", has_symbols),
);
}
let mut input_index = HashMap::new();
let mut required_inputs = Vec::new();
for &vid in &graph.inputs {
if graph.initializers.contains_key(&vid) {
continue; }
required_inputs.push(vid);
if let Some(name) = &graph.value(vid).name {
input_index.insert(name.clone(), vid);
}
}
let mut name_index = HashMap::new();
for (vid, value) in graph.values.iter() {
if let Some(name) = &value.name {
name_index.insert(name.clone(), vid);
}
}
let mut exec = Self {
graph,
weights,
ep,
weight_handles,
buffers,
buffer_shapes,
value_shapes,
value_dtypes,
plan,
input_index,
required_inputs,
has_symbols,
cache: KernelCache::default(),
name_index,
subgraph_execs: HashMap::new(),
control_flow_stats: ControlFlowStats::default(),
if_last_predicate: HashMap::new(),
device_graph_signature: None,
capture_schedule: None,
capture_segmentation: Vec::new(),
control_flow_output_values,
capture_cf_shapes: HashMap::new(),
capture_warm_signature: None,
capture_warm_shapes: HashMap::new(),
capture_warm_seeded: HashMap::new(),
capture_quarantine_ops: HashSet::new(),
last_capture_failed_node: None,
views: HashMap::new(),
pinned: HashSet::new(),
sequence_values,
shared_buffers: HashMap::new(),
sequences: HashMap::new(),
seq_elem_values: HashMap::new(),
execution_provider_fallback_report,
trace: TraceContext::noop(),
scratch_input_shapes: Vec::new(),
decode_memo_enabled: decode_memo_env_enabled(),
decode_memo_verify: cfg!(debug_assertions) || decode_memo_verify_env_enabled(),
decode_memo: None,
decode_memo_prev_bindings: None,
decode_memo_last_action: DecodeMemoAction::Disabled,
decode_memo_resolved: HashMap::new(),
decode_memo_primed_count: 0,
decode_memo_rebuilt_count: 0,
decode_memo_replayed_count: 0,
decode_memo_ineligible_count: 0,
decode_view_plan: None,
decode_views_reused_count: 0,
decode_dispatch_elided_count: 0,
decode_view_plan_sig_mismatch_streak: 0,
decode_view_plan_disabled: false,
};
if !exec.has_symbols {
let mut span = trace_span("session.static_materialize", "session");
let empty = HashMap::new();
let resolved = exec.resolve_all(&empty)?;
exec.size_buffers(&resolved)?;
exec.compile_all(&resolved)?;
if let Some(span) = span.as_mut() {
span.set_args(
Args::new()
.with("resolved_values", resolved.len() as u64)
.with("buffers", exec.buffers.len() as u64)
.with("plan_len", exec.plan.len() as u64)
.with("cache_entries", exec.cache.stats().entries as u64),
);
}
}
Ok(exec)
}
fn ensure_buffer(&mut self, vid: ValueId, dtype: DataType, dims: &[usize]) -> Result<()> {
if self.buffer_shapes.get(&vid).map(|s| s.as_slice()) == Some(dims) {
return Ok(()); }
if let Some(old) = self.buffers.remove(&vid) {
self.ep.deallocate(old)?;
}
self.shared_buffers.remove(&vid);
let numel = checked_numel(dims, || format!("value#{}", vid.0))?;
let size = checked_storage_bytes(dtype, numel, || format!("value#{}", vid.0), dims)?;
let buf = self
.ep
.allocate(size.max(1), TensorLayout::contiguous().alignment)?;
self.buffers.insert(vid, buf);
self.buffer_shapes.insert(vid, dims.to_vec());
Ok(())
}
fn resolve_all(
&self,
bindings: &HashMap<SymbolId, usize>,
) -> Result<HashMap<ValueId, Vec<usize>>> {
let mut resolved = HashMap::with_capacity(self.value_shapes.len());
for (&vid, shape) in &self.value_shapes {
if self.sequence_values.contains(&vid) {
continue;
}
match substitute(shape, bindings) {
Some(dims) => {
resolved.insert(vid, dims);
}
None => {
let value = self.graph.value(vid);
let name = value
.name
.clone()
.unwrap_or_else(|| format!("value#{}", vid.0));
let op = value
.producer
.map(|nid| self.graph.node(nid).op_type.clone())
.unwrap_or_else(|| "<graph input>".to_string());
return Err(SessionError::UnresolvedShape { value: name, op });
}
}
}
Ok(resolved)
}
fn resolve_soft(&self, bindings: &HashMap<SymbolId, usize>) -> HashMap<ValueId, Vec<usize>> {
let mut resolved = HashMap::with_capacity(self.value_shapes.len());
for (&vid, shape) in &self.value_shapes {
if let Some(dims) = substitute(shape, bindings) {
resolved.insert(vid, dims);
}
}
resolved
}
fn resolve_soft_decode_memo(
&mut self,
bindings: &HashMap<SymbolId, usize>,
external: &ExternalBindings,
) -> HashMap<ValueId, Vec<usize>> {
let external_sig = self.decode_external_signature(external);
if self
.decode_memo
.as_ref()
.is_some_and(|memo| memo.matches(bindings, &external_sig))
{
let memo = self.decode_memo.take().unwrap();
let mut resolved = std::mem::take(&mut self.decode_memo_resolved);
resolved.retain(|vid, _| memo.canonical.contains(vid));
for (&vid, dims) in &memo.invariant_shapes {
resolved.entry(vid).or_insert_with(|| dims.clone());
}
for &vid in &memo.variant_values {
let shape = &self.value_shapes[&vid];
match resolved.get_mut(&vid) {
Some(slot) => {
if !substitute_into(shape, bindings, slot) {
resolved.remove(&vid);
}
}
None => {
if let Some(dims) = substitute(shape, bindings) {
resolved.insert(vid, dims);
}
}
}
}
if self.decode_memo_verify {
let fresh = self.resolve_soft(bindings);
assert_eq!(
resolved, fresh,
"decode-plan memo replay diverged from resolve_soft (unsound invariant \
classification)"
);
}
self.decode_memo = Some(memo);
self.decode_memo_last_action = DecodeMemoAction::Replayed;
self.decode_memo_replayed_count += 1;
self.decode_memo_prev_bindings = Some(bindings.clone());
return resolved;
}
self.decode_memo_resolved.clear();
self.decode_view_plan = None;
let resolved = self.resolve_soft(bindings);
match self.decode_memo_prev_bindings.take() {
Some(prev) if is_decode_growth_transition(&prev, bindings) => {
let decode_varying: HashSet<SymbolId> = bindings
.iter()
.filter(|(sym, val)| prev.get(*sym) != Some(*val))
.map(|(&sym, _)| sym)
.collect();
let mut invariant_shapes = HashMap::with_capacity(resolved.len());
let mut variant_values = Vec::new();
let mut canonical = HashSet::with_capacity(resolved.len());
for (&vid, dims) in &resolved {
canonical.insert(vid);
if shape_references_any(&self.value_shapes[&vid], &decode_varying) {
variant_values.push(vid);
} else {
invariant_shapes.insert(vid, dims.clone());
}
}
self.decode_memo = Some(DecodePlanMemo {
reference_bindings: bindings.clone(),
decode_varying,
invariant_shapes,
variant_values,
canonical,
reference_external_sig: external_sig,
});
self.decode_memo_last_action = DecodeMemoAction::Rebuilt;
self.decode_memo_rebuilt_count += 1;
}
_ => {
self.decode_memo = None;
self.decode_memo_last_action = DecodeMemoAction::Primed;
self.decode_memo_primed_count += 1;
}
}
self.decode_memo_prev_bindings = Some(bindings.clone());
resolved
}
fn decode_external_signature(&self, external: &ExternalBindings) -> Vec<DecodeBindingSig> {
let mut sig: Vec<DecodeBindingSig> = external
.inputs
.keys()
.map(|&vid| (vid, true))
.chain(external.outputs.keys().map(|&vid| (vid, false)))
.map(|(vid, is_input)| DecodeBindingSig {
vid,
is_input,
dtype: self.value_dtypes[&vid],
decl_shape: self.value_shapes[&vid].clone(),
})
.collect();
sig.sort_by_key(|s| (s.vid.0, s.is_input));
sig
}
#[cfg(test)]
fn set_decode_memo_enabled(&mut self, enabled: bool) {
self.decode_memo_enabled = enabled;
self.decode_memo_verify = true;
self.decode_memo = None;
self.decode_memo_prev_bindings = None;
self.decode_memo_resolved.clear();
self.decode_memo_last_action = DecodeMemoAction::Disabled;
self.decode_memo_primed_count = 0;
self.decode_memo_rebuilt_count = 0;
self.decode_memo_replayed_count = 0;
self.decode_memo_ineligible_count = 0;
self.decode_view_plan = None;
self.decode_views_reused_count = 0;
self.decode_dispatch_elided_count = 0;
self.decode_view_plan_sig_mismatch_streak = 0;
self.decode_view_plan_disabled = false;
}
#[cfg(test)]
fn decode_memo_action(&self) -> DecodeMemoAction {
self.decode_memo_last_action
}
pub(crate) fn decode_memo_counts(&self) -> (u64, u64, u64, u64) {
(
self.decode_memo_primed_count,
self.decode_memo_rebuilt_count,
self.decode_memo_replayed_count,
self.decode_memo_ineligible_count,
)
}
pub(crate) fn decode_view_plan_counts(&self) -> (u64, u64) {
(
self.decode_views_reused_count,
self.decode_dispatch_elided_count,
)
}
fn stage2_buffer_sig_matches(&self, plan: &DecodeViewPlan) -> bool {
plan.source_buffer_sig.iter().all(|(vid, ptr, cap)| {
self.buffers
.get(vid)
.is_some_and(|buf| buf.as_ptr() as usize == *ptr && buf.len() == *cap)
})
}
fn build_decode_view_plan(&self) -> Option<DecodeViewPlan> {
let memo = self.decode_memo.as_ref()?;
let invariant = |vid: &ValueId| memo.invariant_shapes.contains_key(vid);
let mut elided_nodes = HashSet::new();
let mut retained_views = Vec::new();
let mut sources: HashSet<ValueId> = HashSet::new();
for pi in 0..self.plan.len() {
let outputs = &self.plan[pi].outputs;
if outputs.is_empty() {
continue;
}
let all_view_invariant = outputs
.iter()
.all(|ovid| invariant(ovid) && self.views.contains_key(ovid));
if !all_view_invariant {
continue;
}
elided_nodes.insert(pi);
for ovid in outputs {
let view = self.views[ovid].clone();
sources.insert(view.source);
retained_views.push((*ovid, view));
}
}
if elided_nodes.is_empty() {
return None;
}
let mut source_buffer_sig = Vec::with_capacity(sources.len());
for &src in &sources {
let buf = self.buffers.get(&src)?;
source_buffer_sig.push((src, buf.as_ptr() as usize, buf.len()));
}
Some(DecodeViewPlan {
elided_nodes,
retained_views,
pinned_sources: sources.into_iter().collect(),
source_buffer_sig,
validated: false,
})
}
fn validate_decode_view_plan(&self, mut plan: DecodeViewPlan) -> Option<DecodeViewPlan> {
let view_matches = |a: &ValueView, b: &ValueView| {
a.source == b.source
&& a.shape == b.shape
&& a.strides == b.strides
&& a.byte_offset == b.byte_offset
};
let mut surviving_nodes: HashSet<usize> = HashSet::new();
let node_outputs = |pi: usize| self.plan[pi].outputs.clone();
for &pi in &plan.elided_nodes {
let ok = node_outputs(pi).iter().all(|ovid| {
match (
plan.retained_views.iter().find(|(v, _)| v == ovid),
self.views.get(ovid),
) {
(Some((_, cached)), Some(fresh)) => view_matches(cached, fresh),
_ => false,
}
});
if ok {
surviving_nodes.insert(pi);
}
}
if surviving_nodes.is_empty() {
return None;
}
let surviving_outputs: HashSet<ValueId> = surviving_nodes
.iter()
.flat_map(|&pi| self.plan[pi].outputs.clone())
.collect();
let mut retained_views = Vec::new();
let mut sources: HashSet<ValueId> = HashSet::new();
for ovid in surviving_outputs {
let view = self.views.get(&ovid)?.clone();
sources.insert(view.source);
retained_views.push((ovid, view));
}
let mut source_buffer_sig = Vec::with_capacity(sources.len());
for &src in &sources {
let buf = self.buffers.get(&src)?;
source_buffer_sig.push((src, buf.as_ptr() as usize, buf.len()));
}
plan.elided_nodes = surviving_nodes;
plan.retained_views = retained_views;
plan.pinned_sources = sources.into_iter().collect();
plan.source_buffer_sig = source_buffer_sig;
plan.validated = true;
Some(plan)
}
fn size_buffers(&mut self, resolved: &HashMap<ValueId, Vec<usize>>) -> Result<()> {
self.size_buffers_excluding(resolved, &HashSet::new())
}
fn size_buffers_excluding(
&mut self,
resolved: &HashMap<ValueId, Vec<usize>>,
excluded: &HashSet<ValueId>,
) -> Result<()> {
let vids: Vec<ValueId> = self.value_shapes.keys().copied().collect();
for vid in vids {
if self.graph.initializers.contains_key(&vid) || excluded.contains(&vid) {
continue;
}
if self.sequence_values.contains(&vid) {
continue;
}
let dtype = self.value_dtypes[&vid];
let Some(dims) = resolved.get(&vid).cloned() else {
continue;
};
self.ensure_buffer(vid, dtype, &dims)?;
}
Ok(())
}
fn node_input_shapes(
plan: &NodePlan,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Vec<Vec<usize>> {
plan.inputs
.iter()
.map(|v| v.map(|vid| resolved[&vid].clone()).unwrap_or_default())
.collect()
}
fn compile_all(&mut self, resolved: &HashMap<ValueId, Vec<usize>>) -> Result<()> {
let mut span = trace_span("session.kernel_compile_plan", "session");
let cache_entries_before = self.cache.stats().entries;
let mut compiled_nodes = 0_u64;
let mut skipped_control_flow = 0_u64;
let mut skipped_sequence = 0_u64;
for i in 0..self.plan.len() {
let node_id = self.plan[i].node_id;
let node = self.graph.node(node_id);
if is_control_flow_op(&node.op_type, &node.domain) {
if span.is_some() {
skipped_control_flow += 1;
}
continue;
}
if is_sequence_op(&node.op_type, &node.domain) {
if span.is_some() {
skipped_sequence += 1;
}
continue;
}
if span.is_some() {
compiled_nodes += 1;
}
let input_shapes = Self::node_input_shapes(&self.plan[i], resolved);
let input_dtypes = self.plan[i].input_dtypes.clone();
let constant_inputs: Vec<bool> = self.plan[i]
.inputs
.iter()
.map(|input| input.is_some_and(|vid| self.graph.initializers.contains_key(&vid)))
.collect();
let node = self.graph.node(node_id);
let opset = effective_opset(&self.graph, node);
self.cache.get_or_create(
node_id,
node,
&input_shapes,
&input_dtypes,
&constant_inputs,
opset,
self.ep.as_ref(),
)?;
}
if let Some(span) = span.as_mut() {
span.set_args(
Args::new()
.with("plan_len", self.plan.len() as u64)
.with("compiled_nodes", compiled_nodes)
.with("skipped_control_flow", skipped_control_flow)
.with("skipped_sequence", skipped_sequence)
.with("cache_entries_before", cache_entries_before as u64)
.with("cache_entries_after", self.cache.stats().entries as u64),
);
}
Ok(())
}
pub(crate) fn cache_stats(&self) -> CacheStats {
self.cache.stats()
}
pub(crate) fn control_flow_stats(&self) -> ControlFlowStats {
self.control_flow_stats
}
pub(crate) fn device_id(&self) -> onnx_runtime_ir::DeviceId {
self.ep.device_id()
}
pub(crate) fn allocate_device_binding(
&self,
input_name: String,
output_name: Option<String>,
dtype: DataType,
physical_shape: Vec<usize>,
logical_shape: Vec<usize>,
) -> Result<DeviceIoBinding> {
let expose_logical_input_shape = self.input_index.get(&input_name).is_none_or(|&vid| {
if output_name.is_some() {
!self.binding_consumers_use_physical_capacity(vid)
} else {
!self.binding_consumers_use_padded_capacity(vid)
}
});
DeviceIoBinding::allocate(
self.ep.clone(),
input_name,
true,
output_name,
dtype,
physical_shape,
logical_shape,
expose_logical_input_shape,
)
}
pub(crate) fn allocate_device_output_binding(
&self,
output_name: String,
dtype: DataType,
physical_shape: Vec<usize>,
logical_shape: Vec<usize>,
) -> Result<DeviceIoBinding> {
DeviceIoBinding::allocate(
self.ep.clone(),
String::new(),
false,
Some(output_name),
dtype,
physical_shape,
logical_shape,
false,
)
}
fn binding_consumers_use_physical_capacity(&self, input: ValueId) -> bool {
let mut found = false;
for plan in &self.plan {
for (slot, value) in plan.inputs.iter().enumerate() {
if *value != Some(input) {
continue;
}
found = true;
if !kernel_input_uses_physical_capacity(self.graph.node(plan.node_id), slot) {
return false;
}
}
}
found
}
fn binding_consumers_use_padded_capacity(&self, input: ValueId) -> bool {
let mut found = false;
for plan in &self.plan {
for (slot, value) in plan.inputs.iter().enumerate() {
if *value != Some(input) {
continue;
}
found = true;
if !kernel_input_uses_padded_capacity(self.graph.node(plan.node_id), slot) {
return false;
}
}
}
found
}
pub(crate) fn graph(&self) -> &Graph {
&self.graph
}
pub(crate) fn execution_provider_fallback_report(
&self,
) -> Option<&ExecutionProviderFallbackReport> {
self.execution_provider_fallback_report.as_ref()
}
pub(crate) fn set_trace_context(&mut self, trace: TraceContext) {
for child in self.subgraph_execs.values_mut() {
child.set_trace_context(trace.clone());
}
self.trace = trace;
}
pub(crate) fn weights(&self) -> &Arc<WeightStore> {
&self.weights
}
pub(crate) fn warmup(&mut self) -> Result<()> {
if self.has_symbols {
return Ok(());
}
let empty = HashMap::new();
let resolved = self.resolve_all(&empty)?;
self.compile_all(&resolved)
}
fn bind_symbols(
&self,
inputs: &[(&str, &Tensor)],
external: &ExternalBindings,
) -> Result<HashMap<SymbolId, usize>> {
let mut bindings: HashMap<SymbolId, usize> = HashMap::new();
for (name, tensor) in inputs {
let vid = *self
.input_index
.get(*name)
.ok_or_else(|| SessionError::InputNotFound {
name: (*name).to_string(),
})?;
self.bind_input_shape(name, vid, tensor.dtype, &tensor.shape, &mut bindings)?;
}
for (&vid, value) in &external.inputs {
let name = self.graph.value(vid).name.as_deref().unwrap_or("<unnamed>");
self.bind_input_shape(name, vid, value.dtype, &value.shape, &mut bindings)?;
}
Ok(bindings)
}
fn bind_input_shape(
&self,
name: &str,
vid: ValueId,
dtype: DataType,
shape: &[usize],
bindings: &mut HashMap<SymbolId, usize>,
) -> Result<()> {
let want_dtype = self.value_dtypes[&vid];
if dtype != want_dtype {
return Err(SessionError::DtypeMismatch {
name: name.to_string(),
expected: format!("{want_dtype:?}"),
got: format!("{dtype:?}"),
});
}
let decl = &self.value_shapes[&vid];
if decl.len() != shape.len() {
return Err(SessionError::RankMismatch {
name: name.to_string(),
expected: decl.len(),
got: shape.len(),
});
}
for (dim, &actual) in decl.iter().zip(shape) {
match dim {
Dim::Static(n) if *n != actual => {
return Err(SessionError::ShapeMismatch {
name: name.to_string(),
expected: as_static_shape(decl).unwrap_or_default(),
got: shape.to_vec(),
});
}
Dim::Static(_) => {}
Dim::Symbolic(s) => {
if let Some(&prev) = bindings.get(s) {
if prev != actual {
let sym = self
.symbol_name(*s)
.unwrap_or_else(|| format!("symbol#{}", s.0));
return Err(SessionError::SymbolConflict {
symbol: sym,
first: prev,
second: actual,
});
}
} else {
bindings.insert(*s, actual);
}
}
}
}
Ok(())
}
fn symbol_name(&self, s: SymbolId) -> Option<String> {
self.graph
.symbol_constraints
.get(&s)
.and_then(|c| c.name.clone())
}
pub(crate) fn run(&mut self, inputs: &[(&str, &Tensor)]) -> Result<Vec<Tensor>> {
self.run_outputs(inputs)?
.into_iter()
.map(|output| {
match output {
SessionOutput::Tensor(tensor) => Ok(tensor),
SessionOutput::Sequence(_) => Err(SessionError::SequenceOp {
op: "<graph output>".to_string(),
reason: "the tensor-only run API received a Sequence graph output; use InferenceSession::run_outputs to preserve sequence values".to_string(),
}),
}
})
.collect()
}
pub(crate) fn run_outputs(&mut self, inputs: &[(&str, &Tensor)]) -> Result<Vec<SessionOutput>> {
self.run_scoped(inputs, &HashMap::new(), &ExternalBindings::default())?
.into_iter()
.map(|output| {
output.ok_or_else(|| {
SessionError::Internal(
"ordinary run unexpectedly suppressed a bound graph output".into(),
)
})
})
.collect()
}
pub(crate) fn run_with_device_bindings(
&mut self,
inputs: &[(&str, &Tensor)],
bindings: &mut [DeviceIoBinding],
) -> Result<Vec<Option<Tensor>>> {
let external = self.prepare_external_bindings(bindings)?;
self.run_scoped(inputs, &HashMap::new(), &external)?
.into_iter()
.map(|output| match output {
None => Ok(None),
Some(SessionOutput::Tensor(tensor)) => Ok(Some(tensor)),
Some(SessionOutput::Sequence(_)) => Err(SessionError::SequenceOp {
op: "<graph output>".to_string(),
reason: "run_with_device_bindings cannot return an unbound Sequence graph output; use run_outputs without tensor device bindings".to_string(),
}),
})
.collect()
}
pub(crate) fn try_capture_with_device_bindings(
&mut self,
inputs: &[(&str, &Tensor)],
bindings: &mut [DeviceIoBinding],
) -> Result<DeviceGraphCaptureResult> {
let external = self.prepare_external_bindings(bindings)?;
match self.run_scoped_mode(inputs, &HashMap::new(), &external, RunMode::Capture)? {
ScopedRunResult::Executed(outputs) => {
let mut tensors = Vec::with_capacity(outputs.len());
for output in outputs {
match output {
None => tensors.push(None),
Some(SessionOutput::Tensor(tensor)) => tensors.push(Some(tensor)),
Some(SessionOutput::Sequence(_)) => {
self.reset_device_graph()?;
return Ok(DeviceGraphCaptureResult::NotCapturable(
CaptureDeclineReport::one(CaptureDecline::graph(
"device graph capture cannot return a Sequence graph output",
)),
));
}
}
}
self.device_graph_signature = Some(Self::binding_signature(bindings));
Ok(DeviceGraphCaptureResult::Captured(tensors))
}
ScopedRunResult::NotCapturable(reason) => {
Ok(DeviceGraphCaptureResult::NotCapturable(reason))
}
}
}
pub(crate) fn replay_device_graph(&mut self, bindings: &mut [DeviceIoBinding]) -> Result<bool> {
let external = self.prepare_external_bindings(bindings)?;
let signature = Self::binding_signature(bindings);
if self.device_graph_signature.as_ref() != Some(&signature) {
self.reset_device_graph()?;
return Err(SessionError::Internal(
"device graph replay bindings changed shape, address, or I/O identity; graph was invalidated"
.into(),
));
}
let single_graph = self
.capture_schedule
.as_ref()
.is_none_or(CaptureSchedule::is_single_graph);
if single_graph {
self.ep.replay_device_graph()?;
return Ok(true);
}
match self.run_scoped_mode(&[], &HashMap::new(), &external, RunMode::Replay)? {
ScopedRunResult::Executed(_) => Ok(self.capture_schedule.is_some()),
ScopedRunResult::NotCapturable(reason) => {
self.reset_device_graph()?;
Err(SessionError::Internal(format!(
"segmented device graph replay lost its schedule: {reason}"
)))
}
}
}
pub(crate) fn reset_device_graph(&mut self) -> Result<bool> {
self.device_graph_signature = None;
self.capture_schedule = None;
self.capture_cf_shapes.clear();
self.capture_warm_seeded.clear();
Ok(self.ep.reset_device_graph()?)
}
pub(crate) fn capture_segmentation(&self) -> &[CaptureDecline] {
&self.capture_segmentation
}
pub(crate) fn captured_segment_count(&self) -> usize {
self.capture_schedule
.as_ref()
.map(CaptureSchedule::captured_segments)
.unwrap_or(0)
}
pub(crate) fn check_device_capture_error(&self) -> Result<u32> {
Ok(self.ep.check_device_capture_error()?)
}
pub(crate) fn device_allocation_counts(&self) -> Option<DeviceAllocationCounts> {
self.ep
.device_allocation_counts()
.map(|(allocations, frees)| DeviceAllocationCounts { allocations, frees })
}
fn binding_signature(bindings: &[DeviceIoBinding]) -> Vec<DeviceBindingSignature> {
bindings
.iter()
.map(|binding| DeviceBindingSignature {
input_name: binding.input_name().to_string(),
binds_input: binding.binds_input(),
output_name: binding.output_name().map(str::to_string),
dtype: binding.dtype,
physical_shape: binding.physical_shape().to_vec(),
device_ptr: binding.device_ptr() as usize,
})
.collect()
}
fn prepare_external_bindings(
&self,
bindings: &mut [DeviceIoBinding],
) -> Result<ExternalBindings> {
let mut external = ExternalBindings::default();
for binding in bindings {
let input_name = binding.input_name().to_string();
let bind_input = binding.binds_input();
let output_name = binding.output_name().map(str::to_string);
let dtype = binding.dtype;
let len = binding.buffer().len();
let alignment = binding.buffer().alignment();
let device = binding.buffer().device();
if device != self.ep.device_id() {
return Err(SessionError::Internal(format!(
"device binding '{input_name}' is on {device:?}, session is on {:?}",
self.ep.device_id()
)));
}
let physical_shape = binding.physical_shape();
let required = dtype.storage_bytes(physical_shape.iter().product());
if required > len {
return Err(SessionError::Internal(format!(
"device binding '{input_name}' needs {required} bytes for {physical_shape:?}, allocation has {len}"
)));
}
let ptr = binding.buffer_mut().as_mut_ptr();
if bind_input {
let input_vid = *self.input_index.get(&input_name).ok_or_else(|| {
SessionError::InputNotFound {
name: input_name.clone(),
}
})?;
let value = ExternalValue {
dtype,
shape: binding.kernel_input_shape().to_vec(),
accepts_subshape: false,
ptr,
len,
alignment,
device,
};
if external.inputs.insert(input_vid, value).is_some() {
return Err(SessionError::Internal(format!(
"duplicate device input binding '{input_name}'"
)));
}
}
if let Some(output_name) = output_name {
let output_vid = self
.graph
.outputs
.iter()
.copied()
.find(|&vid| {
self.graph.value(vid).name.as_deref() == Some(output_name.as_str())
})
.ok_or_else(|| {
SessionError::Internal(format!(
"device binding output not found: {output_name}"
))
})?;
if self.sequence_values.contains(&output_vid) {
return Err(SessionError::SequenceOp {
op: "<graph output binding>".to_string(),
reason: format!(
"graph output '{output_name}' is a Sequence value and cannot be bound to tensor device storage"
),
});
}
if self.value_dtypes[&output_vid] != dtype {
return Err(SessionError::DtypeMismatch {
name: output_name.clone(),
expected: format!("{:?}", self.value_dtypes[&output_vid]),
got: format!("{dtype:?}"),
});
}
let value = ExternalValue {
dtype,
shape: binding.physical_shape().to_vec(),
accepts_subshape: bind_input
&& binding.logical_shape() != binding.physical_shape(),
ptr,
len,
alignment,
device,
};
if external.outputs.insert(output_vid, value).is_some() {
return Err(SessionError::Internal(format!(
"duplicate device output binding '{output_name}'"
)));
}
}
}
Ok(external)
}
fn run_scoped(
&mut self,
inputs: &[(&str, &Tensor)],
outer_scope: &HashMap<String, Tensor>,
external: &ExternalBindings,
) -> Result<Vec<Option<SessionOutput>>> {
match self.run_scoped_mode(inputs, outer_scope, external, RunMode::Eager)? {
ScopedRunResult::Executed(outputs) => Ok(outputs),
ScopedRunResult::NotCapturable(_) => unreachable!("eager runs are always executed"),
}
}
fn run_scoped_mode(
&mut self,
inputs: &[(&str, &Tensor)],
outer_scope: &HashMap<String, Tensor>,
external: &ExternalBindings,
mode: RunMode,
) -> Result<ScopedRunResult> {
thread_local! {
static RUN_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}
let depth = RUN_DEPTH.with(|d| {
let cur = d.get();
d.set(cur + 1);
cur
});
struct DepthGuard;
impl Drop for DepthGuard {
fn drop(&mut self) {
RUN_DEPTH.with(|d| d.set(d.get().saturating_sub(1)));
}
}
let _depth_guard = DepthGuard;
let nested = depth > 0;
self.views.clear();
self.pinned.clear();
self.sequences.clear();
self.seq_elem_values.clear();
self.restore_shared_buffers()?;
let _phase_setup = phase_span!(if nested {
"run_scoped.setup_total.child"
} else {
"run_scoped.setup_total.top"
});
let bindings = self.bind_symbols(inputs, external)?;
for (name, _) in inputs {
let vid = self.input_index[*name];
if external.inputs.contains_key(&vid) {
return Err(SessionError::Internal(format!(
"input '{name}' is bound both as a host tensor and a persistent device buffer"
)));
}
}
let mut provided: HashSet<ValueId> = inputs
.iter()
.filter_map(|(name, _)| self.input_index.get(*name).copied())
.collect();
provided.extend(external.inputs.keys().copied());
for &vid in &self.required_inputs {
if !provided.contains(&vid) {
let name = self
.graph
.value(vid)
.name
.clone()
.unwrap_or_else(|| format!("value#{}", vid.0));
return Err(SessionError::InputNotFound { name });
}
}
let decode_memo_eligible = self.decode_memo_enabled
&& mode == RunMode::Eager
&& !nested
&& self.ep.device_type() != DeviceType::Cuda;
let mut resolved = {
let _s = phase_span!("run_scoped.resolve_soft");
if decode_memo_eligible {
self.resolve_soft_decode_memo(&bindings, external)
} else {
if self.decode_memo_enabled && !nested {
self.decode_memo_ineligible_count += 1;
}
let mut resolved = self.resolve_soft(&bindings);
if mode != RunMode::Eager {
external.seed_capture_shapes(&mut resolved);
self.seed_control_flow_capture_shapes(&mut resolved);
self.seed_warm_decode_capture_shapes(&mut resolved, external);
}
resolved
}
};
let mut stage2_plan: Option<DecodeViewPlan> = None;
let mut stage2_candidate: Option<DecodeViewPlan> = None;
let mut stage2_excluded: Option<HashSet<ValueId>> = None;
if decode_memo_eligible
&& !self.decode_view_plan_disabled
&& self.decode_memo_last_action == DecodeMemoAction::Replayed
&& let Some(plan) = self.decode_view_plan.take()
{
if !plan.validated {
stage2_candidate = Some(plan);
} else if self.stage2_buffer_sig_matches(&plan) {
self.decode_view_plan_sig_mismatch_streak = 0;
for (vid, view) in &plan.retained_views {
self.views.insert(*vid, view.clone());
resolved.insert(*vid, view.shape.clone());
}
for &src in &plan.pinned_sources {
self.pinned.insert(src);
}
self.decode_views_reused_count += plan.retained_views.len() as u64;
self.decode_dispatch_elided_count += plan.elided_nodes.len() as u64;
if let Some(memo) = self.decode_memo.as_ref() {
stage2_excluded = Some(memo.invariant_shapes.keys().copied().collect());
}
stage2_plan = Some(plan);
} else {
self.decode_view_plan_sig_mismatch_streak += 1;
if self.decode_view_plan_sig_mismatch_streak >= STAGE2_SIG_MISMATCH_LIMIT {
self.decode_view_plan_disabled = true;
}
}
}
let external_values = external
.inputs
.keys()
.chain(external.outputs.keys())
.copied()
.collect::<HashSet<_>>();
for &vid in &external_values {
if let Some(old) = self.buffers.remove(&vid) {
self.ep.deallocate(old)?;
}
self.shared_buffers.remove(&vid);
self.buffer_shapes.remove(&vid);
}
{
let _s = phase_span!("run_scoped.size_buffers");
match &stage2_excluded {
Some(invariant) => {
let mut excluded = external_values.clone();
excluded.extend(invariant.iter().copied());
self.size_buffers_excluding(&resolved, &excluded)?;
}
None => {
self.size_buffers_excluding(&resolved, &external_values)?;
}
}
}
for (name, tensor) in inputs {
let vid = self.input_index[*name];
let buf = self
.buffers
.get_mut(&vid)
.expect("input value has a buffer");
self.ep.copy_from_host(tensor.as_bytes(), buf)?;
}
drop(_phase_setup);
match mode {
RunMode::Eager => {
let _s = phase_span!(if nested {
"run_scoped.plan_eager.child"
} else {
"run_scoped.plan_eager.top"
});
let verify_stage2 = self.decode_memo_verify && stage2_plan.is_some();
let verify_snapshot: Option<Vec<(ValueId, ValueView)>> = if verify_stage2 {
stage2_plan.as_ref().map(|p| p.retained_views.clone())
} else {
None
};
let elided = if verify_stage2 {
None
} else {
stage2_plan.as_ref().map(|p| &p.elided_nodes)
};
self.run_plan_eager(&mut resolved, outer_scope, external, elided)?;
if let (Some(snapshot), Some(plan)) = (&verify_snapshot, &stage2_plan) {
for (vid, cached) in snapshot {
let fresh = self.views.get(vid).unwrap_or_else(|| {
panic!(
"F5 Stage 2 verify: elided view value#{} was not rebuilt by a \
full dispatch",
vid.0
)
});
assert!(
fresh.source == cached.source
&& fresh.shape == cached.shape
&& fresh.strides == cached.strides
&& fresh.byte_offset == cached.byte_offset,
"F5 Stage 2 verify: cached view for value#{} ({cached:?}) diverged \
from a freshly built one ({fresh:?}) — invariant view reuse is unsound",
vid.0
);
}
assert!(
self.stage2_buffer_sig_matches(plan),
"F5 Stage 2 verify: a cached view source buffer moved during the step"
);
}
if decode_memo_eligible {
match self.decode_memo_last_action {
DecodeMemoAction::Rebuilt => {
self.decode_view_plan = self.build_decode_view_plan();
}
DecodeMemoAction::Replayed => {
if let Some(cand) = stage2_candidate.take() {
self.decode_view_plan = self.validate_decode_view_plan(cand);
} else if let Some(plan) = stage2_plan.take() {
self.decode_view_plan = Some(plan);
}
}
_ => {}
}
}
if !decode_memo_eligible {
self.capture_warm_shapes = resolved.clone();
self.capture_warm_signature = Some(external.capture_signature());
}
}
RunMode::Capture => {
self.if_last_predicate.clear();
let max_capture_attempts = self.plan.len() + 1;
let schedule = 'capture: loop {
let schedule = match self.plan_capture_segments(&resolved, external) {
Ok(schedule) => schedule,
Err(report) => return Ok(ScopedRunResult::NotCapturable(report)),
};
self.last_capture_failed_node = None;
match self.run_plan_segmented(
&schedule,
RunMode::Capture,
&mut resolved,
outer_scope,
external,
) {
Ok(_) => break 'capture schedule,
Err(error) => {
let _ = self.ep.reset_device_graph();
let quarantined =
self.last_capture_failed_node.take().and_then(|node_id| {
let node = self.graph.node(node_id);
let key = (canonical_domain(node), node.op_type.clone());
self.capture_quarantine_ops.insert(key).then_some(())
});
if quarantined.is_some()
&& self.capture_quarantine_ops.len() < max_capture_attempts
{
self.if_last_predicate.clear();
continue 'capture;
}
self.capture_schedule = None;
self.capture_segmentation.clear();
self.capture_cf_shapes.clear();
self.capture_warm_seeded.clear();
return Ok(ScopedRunResult::NotCapturable(CaptureDeclineReport::one(
CaptureDecline::graph(format!(
"segmented CUDA graph capture failed: {error}"
)),
)));
}
}
};
if let Some((vid, seeded)) = self
.capture_warm_seeded
.iter()
.find(|(vid, seeded)| resolved.get(vid) != Some(*seeded))
.map(|(vid, seeded)| (*vid, seeded.clone()))
{
let current = resolved.get(&vid).cloned();
let _ = self.ep.reset_device_graph();
self.capture_schedule = None;
self.capture_segmentation.clear();
self.capture_cf_shapes.clear();
self.capture_warm_seeded.clear();
return Ok(ScopedRunResult::NotCapturable(CaptureDeclineReport::one(
CaptureDecline::graph(format!(
"warm decode shape seed for value#{} ({seeded:?}) diverged from the \
captured shape ({current:?}); recapturing",
vid.0
)),
)));
}
self.capture_cf_shapes = self
.control_flow_output_values
.iter()
.filter_map(|vid| resolved.get(vid).map(|shape| (*vid, shape.clone())))
.collect();
self.capture_segmentation = schedule.boundaries.clone();
if capture_segmentation_logging_enabled() {
log_capture_segmentation(&schedule);
}
self.capture_schedule = Some(schedule);
}
RunMode::Replay => {
let Some(schedule) = self.capture_schedule.take() else {
return Ok(ScopedRunResult::NotCapturable(CaptureDeclineReport::one(
CaptureDecline::graph(
"segmented device graph replay requested without a capture schedule",
),
)));
};
let still_valid = self.run_plan_segmented(
&schedule,
RunMode::Replay,
&mut resolved,
outer_scope,
external,
)?;
if still_valid {
self.capture_schedule = Some(schedule);
} else {
self.capture_segmentation.clear();
self.capture_cf_shapes.clear();
self.device_graph_signature = None;
self.ep.reset_device_graph()?;
}
}
}
let _phase_collect = phase_span!(if nested {
"run_scoped.collect_outputs.child"
} else {
"run_scoped.collect_outputs.top"
});
let mut results = Vec::with_capacity(self.graph.outputs.len());
let mut host_output_bytes = 0usize;
let output_vids: Vec<ValueId> = self.graph.outputs.clone();
for vid in output_vids {
if external.outputs.contains_key(&vid) {
results.push(None);
continue;
}
if self.sequence_values.contains(&vid) {
let sequence = self.sequences.get(&vid).cloned().ok_or_else(|| {
SessionError::Internal(format!(
"sequence graph output value#{} has no live runtime value",
vid.0
))
})?;
results.push(Some(SessionOutput::Sequence(sequence)));
continue;
}
let dtype = self.value_dtypes[&vid];
let shape = resolved[&vid].clone();
if !nested && let Some(tensor) = self.try_move_host_output(vid, &shape, dtype)? {
results.push(Some(SessionOutput::Tensor(tensor)));
continue;
}
let bytes = self.contiguous_bytes(vid, &shape, dtype)?;
host_output_bytes += bytes.len();
results.push(Some(SessionOutput::Tensor(Tensor::from_raw(
dtype, shape, &bytes,
)?)));
}
if !nested {
phase_profile::record("collect_outputs.top_host_bytes", host_output_bytes as u128);
}
if decode_memo_eligible {
self.decode_memo_resolved = std::mem::take(&mut resolved);
}
Ok(ScopedRunResult::Executed(results))
}
fn seed_control_flow_capture_shapes(&self, resolved: &mut HashMap<ValueId, Vec<usize>>) {
for &vid in &self.control_flow_output_values {
if resolved.contains_key(&vid) {
continue;
}
if let Some(shape) = self.buffer_shapes.get(&vid) {
resolved.insert(vid, shape.clone());
}
}
}
fn seed_warm_decode_capture_shapes(
&mut self,
resolved: &mut HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) {
self.capture_warm_seeded.clear();
if self.capture_warm_signature.as_ref() != Some(&external.capture_signature()) {
return;
}
let external_values: HashSet<ValueId> = external
.inputs
.keys()
.chain(external.outputs.keys())
.copied()
.collect();
let warm: Vec<(ValueId, Vec<usize>)> = self
.capture_warm_shapes
.iter()
.map(|(&vid, shape)| (vid, shape.clone()))
.collect();
for (vid, shape) in warm {
if resolved.contains_key(&vid)
|| external_values.contains(&vid)
|| self.graph.initializers.contains_key(&vid)
|| self.sequence_values.contains(&vid)
{
continue;
}
self.capture_warm_seeded.insert(vid, shape.clone());
resolved.insert(vid, shape);
}
}
fn control_flow_seam_invalidated(
&self,
pi: usize,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> bool {
let node = self.graph.node(self.plan[pi].node_id);
if !is_control_flow_op(&node.op_type, &node.domain) {
return false;
}
self.plan[pi].outputs.iter().any(|out| {
match (self.capture_cf_shapes.get(out), resolved.get(out)) {
(Some(captured), Some(current)) => captured != current,
(Some(_), None) => true,
_ => false,
}
})
}
fn node_capture_reason(
&self,
plan: &NodePlan,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Option<CaptureDecline> {
let node = self.graph.node(plan.node_id);
if self
.capture_quarantine_ops
.contains(&(canonical_domain(node), node.op_type.clone()))
{
return Some(CaptureDecline::node(
plan.node_id,
node,
SeamReason::CaptureRecordingFailed,
"kernel aborted device-graph recording on a prior capture pass; \
quarantined to an eager seam",
));
}
let outputs_resolved = plan
.outputs
.iter()
.all(|output| resolved.contains_key(output));
let inputs_resolved = plan.inputs.iter().all(|input| match input {
Some(value) => resolved.contains_key(value),
None => true,
});
if let Some(decline) = self.ep.plan_capture_region(
node,
CaptureRegionShapeStatus {
inputs_resolved,
outputs_resolved,
},
) {
return Some(structural_capture_decline(plan.node_id, node, decline));
}
assert!(
inputs_resolved && outputs_resolved,
"EP capture-region policy admitted a node with unresolved shapes"
);
let input_shapes = plan
.inputs
.iter()
.map(|input| {
input.map_or_else(Vec::new, |value| {
resolved
.get(&value)
.cloned()
.expect("resolved input shape checked above")
})
})
.collect();
let key = KernelKey {
node: plan.node_id.0,
shapes: input_shapes,
};
let Some(kernel) = self.cache.entries.get(&key) else {
return Some(CaptureDecline::node(
plan.node_id,
node,
SeamReason::KernelNotWarmed,
"kernel has not been warmed for the requested capture shape",
));
};
kernel_capture_decline(plan.node_id, node, kernel.as_ref())
}
fn plan_capture_segments(
&self,
resolved: &HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) -> std::result::Result<CaptureSchedule, CaptureDeclineReport> {
if self
.graph
.outputs
.iter()
.any(|output| !external.outputs.contains_key(output))
{
return Err(CaptureDeclineReport::one(CaptureDecline::graph(
"every graph output must use a persistent device binding during capture",
)));
}
let declines: Vec<Option<CaptureDecline>> = self
.plan
.iter()
.map(|plan| self.node_capture_reason(plan, resolved))
.collect();
let mut segments: Vec<ScheduledSegment> = Vec::new();
let mut boundaries: Vec<CaptureDecline> = Vec::new();
let mut next_graph_index = 0usize;
let mut pi = 0usize;
while pi < declines.len() {
let captured = declines[pi].is_none();
let start = pi;
while pi < declines.len() && declines[pi].is_none() == captured {
if let Some(decline) = &declines[pi] {
boundaries.push(decline.clone());
}
pi += 1;
}
let graph_index = if captured {
let index = next_graph_index;
next_graph_index += 1;
index
} else {
0
};
segments.push(ScheduledSegment {
start,
end: pi,
captured,
graph_index,
});
}
if next_graph_index == 0 {
return Err(CaptureDeclineReport {
entries: boundaries,
});
}
Ok(CaptureSchedule {
segments,
boundaries,
})
}
fn collect_segment_kernels(
&self,
seg: &ScheduledSegment,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Result<Vec<&dyn onnx_runtime_ep_api::Kernel>> {
let mut kernels = Vec::with_capacity(seg.end - seg.start);
for pi in seg.start..seg.end {
let plan = &self.plan[pi];
let input_shapes = plan
.inputs
.iter()
.map(|input| {
input
.map(|value| resolved.get(&value).cloned())
.unwrap_or(Some(Vec::new()))
})
.collect::<Option<Vec<_>>>()
.ok_or_else(|| {
SessionError::Internal(format!(
"segment kernel node {} lost its resolved input shape before capture",
plan.node_id.0
))
})?;
let key = KernelKey {
node: plan.node_id.0,
shapes: input_shapes,
};
let kernel = self.cache.entries.get(&key).ok_or_else(|| {
SessionError::Internal(format!(
"segment kernel node {} was not warmed before capture",
plan.node_id.0
))
})?;
kernels.push(kernel.as_ref());
}
Ok(kernels)
}
fn exec_plan_node(
&mut self,
pi: usize,
resolved: &mut HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
external: &ExternalBindings,
capture: OpCaptureTrace<'_>,
) -> Result<()> {
let (is_control_flow, is_sequence, _span) = {
let node = self.graph.node(self.plan[pi].node_id);
let is_control_flow = is_control_flow_op(&node.op_type, &node.domain);
let is_sequence = is_sequence_op(&node.op_type, &node.domain);
let span = self.trace.is_enabled().then(|| {
let span = self.trace.span(node.op_type.clone(), "op").without_source();
annotate_current_span_with(|| {
let mut args = Args::new().with("node_id", node.id.0 as u64);
if !node.name.is_empty() {
args = args.with("node", node.name.clone());
}
if !node.domain.is_empty() {
args = args.with("domain", node.domain.clone());
}
if let Some(device) = node.device {
args = args.with(
onnx_runtime_ep_api::ARG_DEVICE,
device.device_type.trace_name().into_owned(),
);
}
args
});
capture.annotate();
span
});
(is_control_flow, is_sequence, span)
};
if is_control_flow {
self.exec_control_flow(pi, resolved, outer_scope)
} else if is_sequence {
self.exec_sequence_node(pi, resolved, external)
} else {
self.exec_kernel_node(pi, resolved, external)
}
}
fn run_plan_eager(
&mut self,
resolved: &mut HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
external: &ExternalBindings,
elided: Option<&HashSet<usize>>,
) -> Result<()> {
let elided = elided.filter(|set| !set.is_empty());
if profile_ops_enabled() {
let run_start = Instant::now();
let mut timings: HashMap<String, (Duration, usize)> = HashMap::new();
for pi in 0..self.plan.len() {
if elided.is_some_and(|set| set.contains(&pi)) {
continue;
}
let op_type = self.graph.node(self.plan[pi].node_id).op_type.clone();
let start = Instant::now();
let result =
self.exec_plan_node(pi, resolved, outer_scope, external, OpCaptureTrace::Eager);
let elapsed = start.elapsed();
let entry = timings.entry(op_type).or_insert((Duration::ZERO, 0));
entry.0 += elapsed;
entry.1 += 1;
result?;
}
print_op_profile(run_start.elapsed(), timings);
} else {
for pi in 0..self.plan.len() {
if elided.is_some_and(|set| set.contains(&pi)) {
continue;
}
self.exec_plan_node(pi, resolved, outer_scope, external, OpCaptureTrace::Eager)?;
}
}
Ok(())
}
fn run_plan_segmented(
&mut self,
schedule: &CaptureSchedule,
mode: RunMode,
resolved: &mut HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
external: &ExternalBindings,
) -> Result<bool> {
let ep = Arc::clone(&self.ep);
let mut invalidated = false;
for seg in &schedule.segments {
if invalidated {
for pi in seg.start..seg.end {
self.exec_plan_node(
pi,
resolved,
outer_scope,
external,
OpCaptureTrace::Eager,
)?;
}
continue;
}
if seg.captured {
match mode {
RunMode::Capture => {
{
let kernels = self.collect_segment_kernels(seg, resolved)?;
ep.begin_device_graph_capture(&kernels)?;
}
let mut capture_guard = SegmentCaptureGuard::arm(ep.as_ref());
for pi in seg.start..seg.end {
let node_id = self.plan[pi].node_id;
if let Err(error) = self.exec_plan_node(
pi,
resolved,
outer_scope,
external,
OpCaptureTrace::Captured,
) {
self.last_capture_failed_node = Some(node_id);
return Err(error);
}
}
capture_guard.disarm();
ep.end_device_graph_capture()?;
ep.replay_device_graph_segment(seg.graph_index)?;
}
RunMode::Replay => {
ep.replay_device_graph_segment(seg.graph_index)?;
}
RunMode::Eager => {
unreachable!("eager runs never build a segment schedule")
}
}
} else {
for pi in seg.start..seg.end {
let node_id = self.plan[pi].node_id.0;
let reason = schedule
.boundaries
.iter()
.find(|decline| decline.node_id == Some(node_id))
.map(|decline| decline.reason.as_str())
.unwrap_or("non-capturable seam node (no recorded reason)");
self.exec_plan_node(
pi,
resolved,
outer_scope,
external,
OpCaptureTrace::Rejected(reason),
)?;
if mode == RunMode::Replay && self.control_flow_seam_invalidated(pi, resolved) {
invalidated = true;
}
}
}
}
Ok(!invalidated)
}
fn refill_input_shapes(&mut self, pi: usize, resolved: &HashMap<ValueId, Vec<usize>>) {
let inputs = &self.plan[pi].inputs;
let scratch = &mut self.scratch_input_shapes;
scratch.truncate(inputs.len());
for (i, slot) in inputs.iter().enumerate() {
if i < scratch.len() {
scratch[i].clear();
} else {
scratch.push(Vec::new());
}
if let Some(vid) = slot {
scratch[i].extend_from_slice(&resolved[vid]);
}
}
}
fn exec_kernel_node(
&mut self,
pi: usize,
resolved: &mut HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) -> Result<()> {
let _node_span = phase_span!("exec_kernel.node");
let node_id = self.plan[pi].node_id;
self.refill_input_shapes(pi, resolved);
let inputs = &self.plan[pi].inputs;
let outputs = &self.plan[pi].outputs;
let input_dtypes = &self.plan[pi].input_dtypes;
let output_dtypes = &self.plan[pi].output_dtypes;
let input_shapes = &self.scratch_input_shapes;
let node = self.graph.node(node_id);
if let Some(output_shape) = runtime_elementwise_output_shape(node, input_shapes) {
let output_shape = output_shape.map_err(|_| {
let node_name = if node.name.is_empty() {
format!("<unnamed node #{}>", node_id.0)
} else {
format!("{:?}", node.name)
};
SessionError::RuntimeBroadcastIncompatible {
node: node_name,
domain: canonical_domain(node),
op_type: node.op_type.clone(),
input_shapes: input_shapes.to_vec(),
}
})?;
if outputs.len() != 1 {
return Err(SessionError::OutputShapeCountMismatch {
op: node.op_type.clone(),
expected: outputs.len(),
got: 1,
});
}
resolved.insert(outputs[0], output_shape);
}
if outputs.iter().any(|v| !resolved.contains_key(v)) {
let opset = effective_opset(&self.graph, node);
let input_values: Vec<Option<Vec<i64>>> = inputs
.iter()
.enumerate()
.map(|(i, v)| {
v.and_then(|vid| self.shape_input_i64(vid, &input_shapes[i], input_dtypes[i]))
})
.collect();
let input_float_values: Vec<Option<Vec<f64>>> = inputs
.iter()
.enumerate()
.map(|(i, v)| {
if !reads_float_shape_input(node, i, opset) {
return None;
}
v.and_then(|vid| {
if node.is_default_domain() && node.op_type == "NonMaxSuppression" {
self.nms_input_f64(vid, &input_shapes[i], input_dtypes[i])
} else {
self.shape_input_f64(vid, &input_shapes[i], input_dtypes[i])
}
})
})
.collect();
let out_shapes = dynamic_output_shapes(
node,
input_shapes,
input_dtypes,
&input_values,
&input_float_values,
opset,
)
.ok_or_else(|| {
let vid = outputs
.iter()
.find(|v| !resolved.contains_key(v))
.copied()
.unwrap_or(outputs[0]);
let value = self.graph.value(vid);
SessionError::UnresolvedShape {
value: value
.name
.clone()
.unwrap_or_else(|| format!("value#{}", vid.0)),
op: node.op_type.clone(),
}
})?;
if out_shapes.len() != outputs.len() {
return Err(SessionError::OutputShapeCountMismatch {
op: self.graph.node(node_id).op_type.clone(),
expected: outputs.len(),
got: out_shapes.len(),
});
}
for (oi, &ovid) in outputs.iter().enumerate() {
resolved.insert(ovid, out_shapes[oi].clone());
}
}
let mut output_shapes: Vec<Vec<usize>> =
outputs.iter().map(|v| resolved[v].clone()).collect();
{
let node = self.graph.node(node_id);
if node.is_default_domain() && node.op_type == "Attention" {
let kv_capacity_bound = kernel_input_uses_physical_capacity(node, 4)
&& kernel_input_uses_physical_capacity(node, 5);
for (oi, &ovid) in outputs.iter().enumerate() {
if oi == 0 {
continue;
}
if let Some(value) = external.outputs.get(&ovid)
&& value.accepts_subshape
&& value.shape.len() == output_shapes[oi].len()
&& value
.shape
.iter()
.zip(&output_shapes[oi])
.enumerate()
.all(|(axis, (&physical, &logical))| axis == 2 || physical == logical)
&& (kv_capacity_bound
|| value
.shape
.get(2)
.zip(output_shapes[oi].get(2))
.is_some_and(|(&physical, &logical)| physical >= logical))
{
output_shapes[oi] = value.shape.clone();
}
}
}
}
let capabilities = self.ep.capabilities();
let accepts_lazy_weights =
LazyWeightBoundary::BlockQuantizedMoe.matches(&node.domain, &node.op_type);
let has_lazy_inputs = accepts_lazy_weights
&& inputs.iter().any(|input| {
input
.and_then(|value| self.weight_handles.get(&value))
.is_some_and(|handle| handle.is_lazy_for(&capabilities))
});
let mut in_infos: Vec<InInfo> = Vec::with_capacity(inputs.len());
let _build_inputs_span = phase_span!("exec_kernel.build_inputs");
for (i, slot) in inputs.iter().enumerate() {
let Some(vid) = *slot else {
in_infos.push(InInfo {
present: false,
dtype: input_dtypes[i],
shape: Vec::new(),
strides: Vec::new(),
byte_offset: 0,
base_ptr: std::ptr::null(),
device: self.ep.device_id(),
backing: TensorBacking::Opaque,
root_len: 0,
});
continue;
};
if let Some(value) = external
.inputs
.get(&vid)
.or_else(|| external.outputs.get(&vid))
{
let strides = compute_contiguous_strides(&value.shape);
view_bounds(&value.shape, &strides, 0, value.dtype, value.len)?;
in_infos.push(InInfo {
present: true,
dtype: value.dtype,
shape: value.shape.clone(),
strides,
byte_offset: 0,
base_ptr: value.ptr.cast_const(),
device: value.device,
backing: TensorBacking::Opaque,
root_len: value.len,
});
continue;
}
if let Some(elem) = self.seq_elem_values.get(&vid) {
let shape = input_shapes[i].clone();
let strides = elem.layout.resolved_strides(&shape);
let root_len = elem.root_len();
let base_ptr = elem.as_ptr();
view_bounds(
&shape,
&strides,
elem.byte_offset(),
input_dtypes[i],
root_len,
)?;
in_infos.push(InInfo {
present: true,
dtype: input_dtypes[i],
shape,
strides,
byte_offset: elem.byte_offset(),
base_ptr,
device: elem.device(),
backing: TensorBacking::Opaque,
root_len,
});
continue;
}
if accepts_lazy_weights
&& self
.weight_handles
.get(&vid)
.is_some_and(|handle| handle.is_lazy_for(&capabilities))
{
in_infos.push(InInfo {
present: false,
dtype: input_dtypes[i],
shape: input_shapes[i].clone(),
strides: compute_contiguous_strides(&input_shapes[i]),
byte_offset: 0,
base_ptr: std::ptr::null(),
device: self.ep.device_id(),
backing: TensorBacking::Opaque,
root_len: 0,
});
continue;
}
let root = self.root_of(vid);
let buf = self.buffers.get(&root).ok_or_else(|| {
SessionError::Internal(format!("missing buffer for input value#{}", vid.0))
})?;
let root_len = buf.len();
let base_ptr = buf.as_ptr();
let (shape, strides, byte_offset) = match self.views.get(&vid) {
Some(view) => (view.shape.clone(), view.strides.clone(), view.byte_offset),
None => {
let shape = input_shapes[i].clone();
let strides = compute_contiguous_strides(&shape);
(shape, strides, 0)
}
};
view_bounds(&shape, &strides, byte_offset, input_dtypes[i], root_len)?;
let backing = self
.graph
.initializers
.get(&root)
.filter(|_| buf.is_borrowed())
.and_then(|weight| self.weights.external_mmap_provenance(weight))
.map(|(mapping_id, offset, len)| {
TensorBacking::ExternalMmap(ExternalMmapRegion {
mapping_id,
offset,
len,
})
})
.unwrap_or(TensorBacking::Opaque);
in_infos.push(InInfo {
present: true,
dtype: input_dtypes[i],
shape,
strides,
byte_offset,
base_ptr,
device: buf.device(),
backing,
root_len,
});
}
drop(_build_inputs_span);
let ep = self.ep.clone();
let graph = &self.graph;
let cache = &mut self.cache;
let weight_handles = &self.weight_handles;
let buffers = &mut self.buffers;
let buffer_shapes = &mut self.buffer_shapes;
let shared_buffers = &mut self.shared_buffers;
let views_meta = &mut self.views;
let pinned = &mut self.pinned;
let mut views: Vec<TensorView> = Vec::with_capacity(in_infos.len());
for info in &in_infos {
if !info.present {
views.push(TensorView::absent(info.dtype));
continue;
}
views.push(
TensorView::new(
DevicePtr(info.base_ptr),
info.dtype,
&info.shape,
&info.strides,
info.device,
)
.with_byte_offset(info.byte_offset)
.with_backing(info.backing),
);
}
let opset = effective_opset(graph, node);
let constant_inputs: Vec<bool> = inputs
.iter()
.map(|input| {
input.is_some_and(|vid| {
graph.initializers.contains_key(&vid)
|| views_meta
.get(&vid)
.is_some_and(|view| graph.initializers.contains_key(&view.source))
})
})
.collect();
let kernel = {
let _s = phase_span!("exec_kernel.get_kernel");
cache.get_or_create(
node_id,
node,
input_shapes,
input_dtypes,
&constant_inputs,
opset,
ep.as_ref(),
)?
};
if !has_lazy_inputs && let Some(specs) = kernel.view_outputs(&views, outputs.len()) {
if outputs
.iter()
.any(|output| external.outputs.contains_key(output))
{
return Err(SessionError::Internal(format!(
"op '{}' cannot bind a zero-copy view output to external storage",
node.op_type
)));
}
drop(views);
if specs.len() != outputs.len() {
return Err(SessionError::Internal(format!(
"op '{}' returned {} view outputs for {} outputs",
node.op_type,
specs.len(),
outputs.len()
)));
}
for (oi, spec) in specs.into_iter().enumerate() {
let ovid = outputs[oi];
let Some(in_vid) = inputs.get(spec.input_index).copied().flatten() else {
return Err(SessionError::Internal(format!(
"op '{}' view output {} references invalid input index {}",
node.op_type, oi, spec.input_index
)));
};
let root = match views_meta.get(&in_vid) {
Some(v) => v.source,
None => in_vid,
};
let root_len = buffers.get(&root).map(|b| b.len()).ok_or_else(|| {
SessionError::Internal(format!("view source value#{} has no buffer", root.0))
})?;
view_bounds(
&spec.shape,
&spec.strides,
spec.byte_offset,
output_dtypes[oi],
root_len,
)?;
debug_assert!(
!pinned.contains(&ovid),
"value#{} is pinned as a live view source yet is being reproduced",
ovid.0
);
if let Some(old) = buffers.remove(&ovid) {
ep.deallocate(old)?;
}
shared_buffers.remove(&ovid);
buffer_shapes.remove(&ovid);
views_meta.insert(
ovid,
ValueView {
source: root,
shape: spec.shape.clone(),
strides: spec.strides,
byte_offset: spec.byte_offset,
},
);
pinned.insert(root);
resolved.insert(ovid, spec.shape);
}
return Ok(());
}
for (oi, &ovid) in outputs.iter().enumerate() {
let dims = &output_shapes[oi];
let numel = checked_numel(dims, || format!("value#{}", ovid.0))?;
let need = checked_storage_bytes(
output_dtypes[oi],
numel,
|| format!("value#{}", ovid.0),
dims,
)?
.max(1);
if let Some(value) = external.outputs.get(&ovid) {
if !value.accepts_output(output_dtypes[oi], dims, need) {
let name = graph.value(ovid).name.as_deref().unwrap_or("<unnamed>");
return Err(SessionError::Internal(format!(
"external output '{name}' has {:?} {:?} ({} bytes), kernel requires {:?} {:?} ({need} bytes)",
value.dtype, value.shape, value.len, output_dtypes[oi], dims
)));
}
continue;
}
let fits = buffers.get(&ovid).map(|b| b.len() == need).unwrap_or(false);
if !fits {
debug_assert!(
!pinned.contains(&ovid),
"value#{} is pinned as a live view source yet is being resized",
ovid.0
);
if let Some(old) = buffers.remove(&ovid) {
ep.deallocate(old)?;
}
shared_buffers.remove(&ovid);
let buf = ep.allocate(need, TensorLayout::contiguous().alignment)?;
buffers.insert(ovid, buf);
}
}
let mut mat: Vec<Option<(Vec<u8>, Vec<i64>)>> = Vec::with_capacity(in_infos.len());
for (i, info) in in_infos.iter().enumerate() {
if !info.present {
mat.push(None);
continue;
}
let contiguous = onnx_runtime_ir::is_contiguous(&info.shape, &info.strides);
if contiguous || kernel.supports_strided_input(i) {
mat.push(None);
continue;
}
if !info.device.is_host_accessible() {
return Err(SessionError::Internal(format!(
"op '{}' requires host-only strided materialization for CUDA input {i}",
node.op_type
)));
}
let esize = info.dtype.byte_size();
if esize == 0 {
return Err(SessionError::from(
onnx_runtime_ep_api::EpError::InvalidTensorView {
reason: format!(
"cannot materialize sub-byte strided input {i} of op '{}'",
node.op_type
),
},
));
}
let src =
unsafe { std::slice::from_raw_parts(info.base_ptr as *const u8, info.root_len) };
let gathered = gather_view(src, &info.shape, &info.strides, info.byte_offset, esize);
let strides = compute_contiguous_strides(&info.shape);
mat.push(Some((gathered, strides)));
}
drop(views);
let mut views: Vec<TensorView> = Vec::with_capacity(in_infos.len());
for (i, info) in in_infos.iter().enumerate() {
if !info.present {
views.push(TensorView::absent(info.dtype));
continue;
}
match &mat[i] {
Some((buf, strides)) => views.push(TensorView::new(
DevicePtr(buf.as_ptr() as *const std::ffi::c_void),
info.dtype,
&info.shape,
strides,
onnx_runtime_ir::DeviceId::cpu(),
)),
None => views.push(
TensorView::new(
DevicePtr(info.base_ptr),
info.dtype,
&info.shape,
&info.strides,
info.device,
)
.with_byte_offset(info.byte_offset)
.with_backing(info.backing),
),
}
}
let out_strides: Vec<Vec<i64>> = output_shapes
.iter()
.map(|s| compute_contiguous_strides(s))
.collect();
struct OutBacking {
vid: ValueId,
internal: Option<DeviceBuffer>,
ptr: *mut std::ffi::c_void,
len: usize,
device: onnx_runtime_ir::DeviceId,
}
let mut out_bufs: Vec<OutBacking> = Vec::with_capacity(outputs.len());
for &vid in outputs {
if let Some(value) = external.outputs.get(&vid) {
out_bufs.push(OutBacking {
vid,
internal: None,
ptr: value.ptr,
len: value.len,
device: value.device,
});
} else {
let mut buf = buffers.remove(&vid).ok_or_else(|| {
SessionError::Internal(format!("missing buffer for output value#{}", vid.0))
})?;
let ptr = buf.as_mut_ptr();
out_bufs.push(OutBacking {
vid,
ptr,
len: buf.len(),
device: buf.device(),
internal: Some(buf),
});
}
}
let mut outs: Vec<TensorMut> = Vec::with_capacity(out_bufs.len());
for (i, backing) in out_bufs.iter_mut().enumerate() {
view_bounds(
&output_shapes[i],
&out_strides[i],
0,
output_dtypes[i],
backing.len,
)?;
outs.push(TensorMut::new(
DevicePtrMut(backing.ptr),
output_dtypes[i],
&output_shapes[i],
&out_strides[i],
backing.device,
));
}
let kernel_inputs = has_lazy_inputs.then(|| {
inputs
.iter()
.zip(views.iter().copied())
.map(|(value, view)| {
value
.and_then(|value| weight_handles.get(&value))
.filter(|handle| handle.is_lazy_for(&capabilities))
.map(KernelInput::Weight)
.unwrap_or(KernelInput::Tensor(view))
})
.collect::<Vec<_>>()
});
let execution = {
let _s = phase_span!("exec_kernel.compute");
match &kernel_inputs {
Some(inputs) => kernel.execute_with_inputs(inputs, &mut outs),
None => kernel.execute(&views, &mut outs),
}
};
execution.map_err(|error| {
let input_types = views.iter().map(|view| view.dtype).collect::<Vec<_>>();
let output_types = outs.iter().map(|output| output.dtype).collect::<Vec<_>>();
let input_shapes = views
.iter()
.map(|view| view.shape.to_vec())
.collect::<Vec<_>>();
let output_shapes = outs
.iter()
.map(|output| output.shape.to_vec())
.collect::<Vec<_>>();
let input_names = inputs
.iter()
.map(|input| {
input
.map(|value| {
self.graph.value(value).name.as_deref().unwrap_or("<unnamed>")
})
.unwrap_or("<absent>")
})
.collect::<Vec<_>>();
let output_names = outputs
.iter()
.map(|&value| {
self.graph.value(value).name.as_deref().unwrap_or("<unnamed>")
})
.collect::<Vec<_>>();
SessionError::Internal(format!(
"node {} ({:?}, op '{}::{}', inputs {input_names:?} {input_types:?} {input_shapes:?}, outputs {output_names:?} {output_types:?} {output_shapes:?}) failed: {error}",
node.id.0, node.name, node.domain, node.op_type,
))
})?;
drop(kernel_inputs);
drop(views);
drop(outs);
for backing in out_bufs {
if let Some(buf) = backing.internal {
buffers.insert(backing.vid, buf);
}
}
Ok(())
}
fn input_i64(&self, vid: ValueId, shape: &[usize], dtype: DataType) -> Option<Vec<i64>> {
let bytes = self.contiguous_bytes(vid, shape, dtype).ok()?;
bytes_as_i64(&bytes, dtype)
}
fn shape_input_i64(&self, vid: ValueId, shape: &[usize], dtype: DataType) -> Option<Vec<i64>> {
if !bounded_shape_input(dtype, shape) {
return None;
}
let max_bytes = MAX_SHAPE_DATA_ELEMS.checked_mul(dtype.byte_size())?;
if let Some(view) = self.views.get(&vid) {
let source = self.buffers.get(&view.source)?;
if source.len() > max_bytes {
return None;
}
}
if self
.seq_elem_values
.get(&vid)
.is_some_and(|elem| elem.root_len() > max_bytes)
{
return None;
}
self.input_i64(vid, shape, dtype)
}
fn shape_input_f64(&self, vid: ValueId, shape: &[usize], dtype: DataType) -> Option<Vec<f64>> {
if !matches!(dtype, DataType::Float32 | DataType::Float64)
|| shape.len() > 1
|| shape
.iter()
.try_fold(1usize, |count, &dim| count.checked_mul(dim))
.is_none_or(|count| count > MAX_SHAPE_DATA_ELEMS)
{
return None;
}
let max_bytes = MAX_SHAPE_DATA_ELEMS.checked_mul(dtype.byte_size())?;
if let Some(view) = self.views.get(&vid) {
let source = self.buffers.get(&view.source)?;
if source.len() > max_bytes {
return None;
}
}
if self
.seq_elem_values
.get(&vid)
.is_some_and(|elem| elem.root_len() > max_bytes)
{
return None;
}
let bytes = self.contiguous_bytes(vid, shape, dtype).ok()?;
bytes_as_f64(&bytes, dtype)
}
fn nms_input_f64(&self, vid: ValueId, shape: &[usize], dtype: DataType) -> Option<Vec<f64>> {
if dtype != DataType::Float32 {
return None;
}
let bytes = self.contiguous_bytes(vid, shape, dtype).ok()?;
bytes_as_f64(&bytes, dtype)
}
}
impl Executor {
fn exec_sequence_node(
&mut self,
pi: usize,
resolved: &mut HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) -> Result<()> {
let node_id = self.plan[pi].node_id;
let inputs = self.plan[pi].inputs.clone();
let outputs = self.plan[pi].outputs.clone();
let op = self.graph.node(node_id).op_type.clone();
match op.as_str() {
"SequenceEmpty" => {
let dtype_attr = self
.graph
.node(node_id)
.attr("dtype")
.and_then(|a| a.as_int());
let dtype = match dtype_attr {
None => DataType::Float32, Some(raw) => i32::try_from(raw)
.ok()
.and_then(DataType::from_onnx)
.ok_or_else(|| SessionError::SequenceOp {
op: op.clone(),
reason: format!(
"attribute 'dtype' = {raw} is not a known ONNX \
TensorProto.DataType. To fix: use a valid element \
dtype id (e.g. 1=float32, 7=int64)"
),
})?,
};
self.sequences
.insert(outputs[0], SequenceValue::empty(dtype));
Ok(())
}
"SequenceConstruct" => {
let mut items = Vec::with_capacity(inputs.len());
for slot in &inputs {
let vid = slot.ok_or_else(|| self.seq_missing_input(&op))?;
items.push(self.read_seq_element(vid, resolved)?);
}
let seq = SequenceValue::construct(items).map_err(seq_err)?;
self.sequences.insert(outputs[0], seq);
Ok(())
}
"SequenceInsert" => {
let seq = self.get_sequence(inputs.first().copied().flatten(), &op)?;
let tvid = inputs
.get(1)
.copied()
.flatten()
.ok_or_else(|| self.seq_missing_input(&op))?;
let tensor = self.read_seq_element(tvid, resolved)?;
let position = match inputs.get(2).copied().flatten() {
Some(pvid) => Some(self.read_scalar_i64(pvid, resolved, &op)?),
None => None,
};
let out = seq.insert(tensor, position).map_err(seq_err)?;
self.sequences.insert(outputs[0], out);
Ok(())
}
"SequenceErase" => {
let seq = self.get_sequence(inputs.first().copied().flatten(), &op)?;
let position = match inputs.get(1).copied().flatten() {
Some(pvid) => Some(self.read_scalar_i64(pvid, resolved, &op)?),
None => None,
};
let out = seq.erase(position).map_err(seq_err)?;
self.sequences.insert(outputs[0], out);
Ok(())
}
"SequenceAt" => {
let seq = self.get_sequence(inputs.first().copied().flatten(), &op)?;
let pvid =
inputs
.get(1)
.copied()
.flatten()
.ok_or_else(|| SessionError::SequenceOp {
op: op.clone(),
reason: "requires a 'position' input. To fix: supply the \
index tensor of the element to read"
.to_string(),
})?;
let pos = self.read_scalar_i64(pvid, resolved, &op)?;
let elem = seq.at(pos).map_err(seq_err)?;
self.store_seq_element_output(outputs[0], elem, resolved, external)
}
"SequenceLength" => {
let seq = self.get_sequence(inputs.first().copied().flatten(), &op)?;
let len = i64::try_from(seq.length()).map_err(|_| {
seq_err(SequenceError::LengthOverflow {
op: "SequenceLength",
len: seq.length(),
})
})?;
self.store_raw_tensor_output(
outputs[0],
DataType::Int64,
Vec::new(),
&len.to_le_bytes(),
resolved,
external,
)
}
"SplitToSequence" => {
self.exec_split_to_sequence(node_id, &op, &inputs, &outputs, resolved)
}
"ConcatFromSequence" => {
self.exec_concat_from_sequence(node_id, &op, &inputs, &outputs, resolved, external)
}
other => Err(SessionError::SequenceOp {
op: other.to_string(),
reason: "unrecognized Sequence op (executor routing bug)".to_string(),
}),
}
}
fn exec_split_to_sequence(
&mut self,
node_id: NodeId,
op: &str,
inputs: &[Option<ValueId>],
outputs: &[ValueId],
resolved: &mut HashMap<ValueId, Vec<usize>>,
) -> Result<()> {
let (axis_attr, keepdims) = {
let node = self.graph.node(node_id);
(
node.attr("axis").and_then(|a| a.as_int()).unwrap_or(0),
node.attr("keepdims").and_then(|a| a.as_int()).unwrap_or(1) != 0,
)
};
let ivid = inputs
.first()
.copied()
.flatten()
.ok_or_else(|| self.seq_missing_input(op))?;
let input = self.read_seq_element(ivid, resolved)?;
let split_input = match inputs.get(1).copied().flatten() {
None => None,
Some(svid) => {
let split_shape = resolved
.get(&svid)
.cloned()
.ok_or_else(|| self.seq_unresolved(op, svid))?;
let values = self.read_i64_vec(svid, &split_shape, op)?;
Some((split_shape, values))
}
};
let split_spec = match split_input.as_ref() {
None => SplitSpec::Each,
Some((split_shape, values)) if split_shape.is_empty() => {
let [chunk] = values.as_slice() else {
return Err(SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"scalar 'split' input contains {} values, expected exactly one",
values.len()
),
});
};
SplitSpec::Chunk(*chunk)
}
Some((split_shape, values)) if split_shape.len() == 1 => SplitSpec::Sizes(values),
Some((split_shape, _)) => {
return Err(SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"'split' input must be rank 0 (chunk size) or rank 1 (explicit sizes), \
got rank {} with shape {split_shape:?}",
split_shape.len()
),
});
}
};
let sequence = split_tensor(&input, axis_attr, split_spec, keepdims).map_err(seq_err)?;
self.sequences.insert(outputs[0], sequence);
Ok(())
}
fn exec_concat_from_sequence(
&mut self,
node_id: NodeId,
op: &str,
inputs: &[Option<ValueId>],
outputs: &[ValueId],
resolved: &mut HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) -> Result<()> {
let node = self.graph.node(node_id);
let axis_attr =
node.attr("axis")
.and_then(|a| a.as_int())
.ok_or_else(|| SessionError::SequenceOp {
op: op.to_string(),
reason: "requires the mandatory 'axis' attribute. To fix: set 'axis'"
.to_string(),
})?;
let new_axis = node.attr("new_axis").and_then(|a| a.as_int()).unwrap_or(0) != 0;
let seq = self.get_sequence(inputs.first().copied().flatten(), op)?;
let plan = ConcatPlan::new(&seq, axis_attr, new_axis).map_err(seq_err)?;
self.prepare_tensor_output(
outputs[0],
plan.dtype,
plan.shape.clone(),
plan.bytes,
resolved,
external,
)?;
let ep = Arc::clone(&self.ep);
if let Some(value) = external.outputs.get(&outputs[0]) {
let mut buffer = value.writable_buffer()?;
plan.write(&seq, |offset, bytes| {
ep.copy_from_host_at(bytes, &mut buffer, offset)?;
Ok(())
})?;
} else {
let buffer = self.buffers.get_mut(&outputs[0]).ok_or_else(|| {
SessionError::Internal(format!(
"missing ConcatFromSequence output buffer for value#{}",
outputs[0].0
))
})?;
plan.write(&seq, |offset, bytes| {
ep.copy_from_host_at(bytes, buffer, offset)?;
Ok(())
})?;
}
Ok(())
}
fn read_seq_element(
&mut self,
vid: ValueId,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Result<SeqTensor> {
if self.sequence_values.contains(&vid) {
return Err(SessionError::SequenceOp {
op: "Sequence".to_string(),
reason: format!(
"input value#{} is a Sequence value, expected a tensor element",
vid.0
),
});
}
if let Some(elem) = self.seq_elem_values.get(&vid) {
return Ok(elem.clone()); }
let dtype = self.value_dtypes[&vid];
let shape = resolved
.get(&vid)
.cloned()
.ok_or_else(|| self.seq_unresolved("Sequence", vid))?;
let (root, layout, byte_offset) = match self.views.get(&vid) {
Some(view) => (
view.source,
TensorLayout::strided(view.strides.clone()),
view.byte_offset,
),
None => (vid, TensorLayout::contiguous(), 0),
};
if !self.shared_buffers.contains_key(&root) {
let buffer = self
.buffers
.remove(&root)
.ok_or_else(|| SessionError::SequenceOp {
op: "Sequence".to_string(),
reason: format!("tensor value#{} has no live backing buffer", vid.0),
})?;
let storage = SharedTensorBuffer::new(Arc::clone(&self.ep), buffer);
self.buffers.insert(root, storage.alias());
self.shared_buffers.insert(root, storage);
}
self.pinned.insert(root);
SeqTensor::from_shared(
Arc::clone(&self.shared_buffers[&root]),
dtype,
shape,
layout,
byte_offset,
)
.map_err(SessionError::from)
}
fn restore_shared_buffers(&mut self) -> Result<()> {
let mut retained = Vec::new();
for (vid, storage) in self.shared_buffers.drain() {
if let Some(alias) = self.buffers.remove(&vid) {
self.ep.deallocate(alias)?;
}
match Arc::try_unwrap(storage) {
Ok(storage) => {
self.buffers.insert(vid, storage.into_buffer());
}
Err(storage) if self.graph.initializers.contains_key(&vid) => {
self.buffers.insert(vid, storage.alias());
retained.push((vid, storage));
}
Err(storage) => {
let replacement = self
.ep
.allocate(storage.buffer().len(), storage.buffer().alignment())?;
self.buffers.insert(vid, replacement);
}
}
}
for (vid, storage) in retained {
self.shared_buffers.insert(vid, storage);
}
Ok(())
}
fn get_sequence(&self, vid: Option<ValueId>, op: &str) -> Result<SequenceValue> {
let vid = vid.ok_or_else(|| self.seq_missing_input(op))?;
self.sequences
.get(&vid)
.cloned()
.ok_or_else(|| SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"input value#{} is not a live sequence. To fix: ensure it is produced \
by a Sequence-producing op (SequenceEmpty/Construct/Insert/Erase/\
SplitToSequence)",
vid.0
),
})
}
fn read_scalar_i64(
&self,
vid: ValueId,
resolved: &HashMap<ValueId, Vec<usize>>,
op: &str,
) -> Result<i64> {
let shape = resolved.get(&vid).cloned().unwrap_or_default();
if !shape.is_empty() {
return Err(SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"position input must be a rank-0 scalar, got rank {} with shape {shape:?}",
shape.len()
),
});
}
let dtype = self.value_dtypes[&vid];
let vals = self
.input_i64(vid, &shape, dtype)
.ok_or_else(|| SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"position input has dtype {dtype:?}, expected an integer (int32/int64). \
To fix: provide an int64 scalar index"
),
})?;
let [value] = vals.as_slice() else {
return Err(SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"position input contains {} values; expected exactly one scalar index",
vals.len()
),
});
};
Ok(*value)
}
fn read_i64_vec(&self, vid: ValueId, shape: &[usize], op: &str) -> Result<Vec<i64>> {
let dtype = self.value_dtypes[&vid];
self.input_i64(vid, shape, dtype)
.ok_or_else(|| SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"'split' input has dtype {dtype:?}, expected int32/int64. To fix: \
provide integer split sizes"
),
})
}
fn store_seq_element_output(
&mut self,
vid: ValueId,
elem: SeqTensor,
resolved: &mut HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) -> Result<()> {
if elem.device() != self.ep.device_id() {
return Err(SessionError::SequenceOp {
op: "SequenceAt".to_string(),
reason: format!(
"sequence element is on {:?}, but the active execution provider is on {:?}",
elem.device(),
self.ep.device_id()
),
});
}
if external.outputs.contains_key(&vid) {
let dtype = elem.dtype;
let shape = elem.shape.clone();
let bytes = elem.contiguous_bytes().map_err(seq_err)?;
return self.store_raw_tensor_output(vid, dtype, shape, &bytes, resolved, external);
}
if let Some(old) = self.buffers.remove(&vid) {
self.ep.deallocate(old)?;
}
self.shared_buffers.remove(&vid);
self.buffer_shapes.remove(&vid);
self.views.remove(&vid);
resolved.insert(vid, elem.shape.clone());
self.value_dtypes.insert(vid, elem.dtype);
self.seq_elem_values.insert(vid, elem);
Ok(())
}
fn store_raw_tensor_output(
&mut self,
vid: ValueId,
dtype: DataType,
dims: Vec<usize>,
bytes: &[u8],
resolved: &mut HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) -> Result<()> {
self.prepare_tensor_output(vid, dtype, dims, bytes.len(), resolved, external)?;
if let Some(value) = external.outputs.get(&vid) {
let mut buffer = value.writable_buffer()?;
self.ep.copy_from_host(bytes, &mut buffer)?;
} else {
let buffer = self.buffers.get_mut(&vid).ok_or_else(|| {
SessionError::Internal(format!("missing tensor output buffer for value#{}", vid.0))
})?;
self.ep.copy_from_host(bytes, buffer)?;
}
Ok(())
}
fn prepare_tensor_output(
&mut self,
vid: ValueId,
dtype: DataType,
dims: Vec<usize>,
bytes: usize,
resolved: &mut HashMap<ValueId, Vec<usize>>,
external: &ExternalBindings,
) -> Result<()> {
self.seq_elem_values.remove(&vid);
self.views.remove(&vid);
let need = bytes.max(1);
if let Some(value) = external.outputs.get(&vid) {
if !value.accepts_output(dtype, &dims, need) {
let name = self.graph.value(vid).name.as_deref().unwrap_or("<unnamed>");
return Err(SessionError::Internal(format!(
"external output '{name}' has {:?} {:?} ({} bytes), sequence op requires {:?} {:?} ({need} bytes)",
value.dtype, value.shape, value.len, dtype, dims
)));
}
} else {
let fits = self
.buffers
.get(&vid)
.map(|buffer| buffer.len() == need)
.unwrap_or(false);
if !fits {
if let Some(old) = self.buffers.remove(&vid) {
self.ep.deallocate(old)?;
}
self.shared_buffers.remove(&vid);
let buffer = self
.ep
.allocate(need, TensorLayout::contiguous().alignment)?;
self.buffers.insert(vid, buffer);
}
self.buffer_shapes.insert(vid, dims.clone());
}
self.value_dtypes.insert(vid, dtype);
resolved.insert(vid, dims);
Ok(())
}
fn seq_missing_input(&self, op: &str) -> SessionError {
SessionError::SequenceOp {
op: op.to_string(),
reason: "a required input is missing (omitted None slot). To fix: connect \
all required inputs of this Sequence op"
.to_string(),
}
}
fn seq_unresolved(&self, op: &str, vid: ValueId) -> SessionError {
let name = self
.graph
.try_value(vid)
.and_then(|v| v.name.clone())
.unwrap_or_else(|| format!("value#{}", vid.0));
SessionError::SequenceOp {
op: op.to_string(),
reason: format!(
"input {name} has no resolved shape yet. To fix: ensure its producer \
runs before this Sequence op"
),
}
}
}
fn seq_err(e: crate::sequence::SequenceError) -> SessionError {
e.into()
}
fn normalize_axis(axis: i64, rank: usize) -> Option<usize> {
let r = rank as i64;
let a = if axis < 0 { axis + r } else { axis };
if a < 0 || a >= r {
None
} else {
Some(a as usize)
}
}
fn scan_list_attr(node: &Node, name: &str, count: usize, default: i64) -> Result<Vec<i64>> {
match node.attr(name) {
None => Ok(vec![default; count]),
Some(attr) => {
let values = attr.as_ints().ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!("attribute '{name}' must be an INTS list"),
})?;
if values.len() != count {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"attribute '{name}' has {} value(s), expected {count}",
values.len()
),
});
}
Ok(values.to_vec())
}
}
}
fn is_control_flow_op(op_type: &str, domain: &str) -> bool {
domain.is_empty() && matches!(op_type, "If" | "Loop" | "Scan")
}
fn is_sequence_op(op_type: &str, domain: &str) -> bool {
domain.is_empty()
&& matches!(
op_type,
"SequenceEmpty"
| "SequenceConstruct"
| "SequenceInsert"
| "SequenceErase"
| "SequenceAt"
| "SequenceLength"
| "SplitToSequence"
| "ConcatFromSequence"
)
}
fn produces_sequence_output(op_type: &str, domain: &str) -> bool {
domain.is_empty()
&& matches!(
op_type,
"SequenceEmpty"
| "SequenceConstruct"
| "SequenceInsert"
| "SequenceErase"
| "SplitToSequence"
)
}
fn tensor_scalar_i64(t: &Tensor) -> Option<i64> {
if t.dtype != DataType::Int64 || t.numel() != 1 {
return None;
}
t.as_bytes()
.get(..8)
.map(|c| i64::from_le_bytes(c.try_into().unwrap()))
}
fn tensor_scalar_bool(t: &Tensor) -> Option<bool> {
if t.dtype != DataType::Bool || t.numel() != 1 {
return None;
}
t.as_bytes().first().map(|&b| b != 0)
}
fn scalar_i64_tensor(v: i64) -> Result<Tensor> {
Tensor::from_raw(DataType::Int64, vec![], &v.to_le_bytes())
}
fn scalar_bool_tensor(v: bool) -> Result<Tensor> {
Tensor::from_raw(DataType::Bool, vec![], &[u8::from(v)])
}
fn missing_capture_error(attr_key: &str, name: &str) -> SessionError {
SessionError::Internal(format!(
"control-flow body '{attr_key}' captures free variable '{name}', but it is not \
available in the enclosing scope. RULES #1: a subgraph may only reference outer \
values that are graph inputs, initializers, or produced by an upstream node in an \
enclosing graph; '{name}' matches none of these"
))
}
fn required_outer_names(graph: &Graph) -> HashSet<String> {
let formal_set: HashSet<ValueId> = graph.inputs.iter().copied().collect();
let local_names: HashSet<&str> = graph
.values
.iter()
.filter_map(|(_, value)| value.name.as_deref())
.collect();
let mut required = HashSet::new();
for (vid, value) in graph.values.iter() {
if value.producer.is_none()
&& !formal_set.contains(&vid)
&& !graph.initializers.contains_key(&vid)
&& let Some(name) = &value.name
{
required.insert(name.clone());
}
}
for nested in graph.subgraphs.values() {
for name in required_outer_names(nested) {
if !local_names.contains(name.as_str()) {
required.insert(name);
}
}
}
required
}
impl ChildExecutor {
pub(crate) fn new(
name: impl Into<String>,
body: Graph,
inherited_opsets: HashMap<String, u64>,
weights: Arc<WeightStore>,
ep: Arc<dyn ExecutionProvider>,
) -> Result<Self> {
let name = name.into();
let formal_names = body
.inputs
.iter()
.map(|&vid| {
body.value(vid).name.clone().ok_or_else(|| {
SessionError::Internal(format!(
"subgraph '{name}' has an unnamed formal input value#{}",
vid.0
))
})
})
.collect::<Result<Vec<_>>>()?;
let formal_set: HashSet<ValueId> = body.inputs.iter().copied().collect();
let mut capture_names = body
.values
.iter()
.filter_map(|(vid, value)| {
(value.producer.is_none()
&& !formal_set.contains(&vid)
&& !body.initializers.contains_key(&vid))
.then(|| value.name.clone())
.flatten()
})
.collect::<Vec<_>>();
capture_names.sort();
let input_names = formal_names
.iter()
.chain(capture_names.iter())
.cloned()
.collect();
Ok(Self {
name,
template: body,
inherited_opsets,
weights,
ep,
formal_names,
capture_names,
input_names,
compiled: Vec::new(),
builds: 0,
runs: 0,
trace: TraceContext::noop(),
})
}
pub(crate) fn stats(&self) -> ChildExecutorStats {
ChildExecutorStats {
builds: self.builds,
runs: self.runs,
}
}
pub(crate) fn set_trace_context(&mut self, trace: TraceContext) {
for plan in &mut self.compiled {
plan.exec.set_trace_context(trace.clone());
}
self.trace = trace;
}
fn compile(&self, externals: &[&Tensor]) -> Result<CompiledChildPlan> {
let mut graph = self.template.clone();
graph.opset_imports = self.inherited_opsets.clone();
let body_names = graph
.values
.iter()
.filter_map(|(vid, value)| value.name.clone().map(|name| (name, vid)))
.collect::<HashMap<_, _>>();
for name in &self.capture_names {
let vid = *body_names.get(name).ok_or_else(|| {
SessionError::Internal(format!(
"subgraph '{}' lost captured value '{name}'",
self.name
))
})?;
if !graph.inputs.contains(&vid) {
graph.add_input(vid);
}
}
for (name, tensor) in self.input_names.iter().zip(externals) {
let vid = *body_names.get(name).ok_or_else(|| {
SessionError::Internal(format!(
"subgraph '{}' is missing bound input '{name}'",
self.name
))
})?;
let value = graph.value_mut(vid);
value.dtype = tensor.dtype;
value.shape = tensor.shape.iter().map(|&dim| Dim::Static(dim)).collect();
}
let registry = InferenceRegistry::default_registry();
registry.infer_graph(&mut graph, &self.inherited_opsets, MergePolicy::Permissive)?;
Ok(CompiledChildPlan {
exec: {
let mut exec = Executor::build(graph, self.weights.clone(), self.ep.clone())?;
exec.set_trace_context(self.trace.clone());
exec
},
signature: externals
.iter()
.map(|tensor| ChildInputSignature {
dtype: tensor.dtype,
shape: tensor.shape.clone(),
})
.collect(),
})
}
pub(crate) fn run(
&mut self,
formal_inputs: &[&Tensor],
outer_scope: &HashMap<String, Tensor>,
) -> Result<Vec<Tensor>> {
if self.formal_names.len() != formal_inputs.len() {
return Err(SessionError::Internal(format!(
"subgraph '{}' expects {} formal input(s) but {} were supplied",
self.name,
self.formal_names.len(),
formal_inputs.len()
)));
}
let mut externals = Vec::with_capacity(formal_inputs.len() + self.capture_names.len());
externals.extend_from_slice(formal_inputs);
for name in &self.capture_names {
externals.push(
outer_scope
.get(name)
.ok_or_else(|| missing_capture_error(&self.name, name))?,
);
}
let signature = externals
.iter()
.map(|tensor| ChildInputSignature {
dtype: tensor.dtype,
shape: tensor.shape.clone(),
})
.collect::<Vec<_>>();
let cache_index = if let Some(index) = self
.compiled
.iter()
.position(|compiled| compiled.signature == signature)
{
let compiled = self.compiled.remove(index);
self.compiled.push(compiled);
self.compiled.len() - 1
} else {
let compiled = self.compile(&externals)?;
if self.compiled.len() == CHILD_EXECUTOR_CACHE_CAPACITY {
self.compiled.remove(0);
}
self.compiled.push(compiled);
self.builds += 1;
self.compiled.len() - 1
};
self.runs += 1;
let inputs = self
.input_names
.iter()
.map(String::as_str)
.zip(externals)
.collect::<Vec<_>>();
self.compiled[cache_index]
.exec
.run_scoped(&inputs, outer_scope, &ExternalBindings::default())?
.into_iter()
.map(|output| {
let output = output.ok_or_else(|| {
SessionError::Internal(format!(
"subgraph '{}' unexpectedly suppressed an output",
self.name
))
})?;
match output {
SessionOutput::Tensor(tensor) => Ok(tensor),
SessionOutput::Sequence(_) => Err(SessionError::SequenceOp {
op: "<control-flow output>".to_string(),
reason: format!(
"subgraph '{}' produced a Sequence output where this control-flow path requires a tensor",
self.name
),
}),
}
})
.collect()
}
}
impl Executor {
fn value_tensor(
&self,
vid: ValueId,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Result<Tensor> {
let dtype = self.value_dtypes[&vid];
let shape = resolved.get(&vid).cloned().ok_or_else(|| {
let name = self
.graph
.try_value(vid)
.and_then(|v| v.name.clone())
.unwrap_or_else(|| format!("value#{}", vid.0));
SessionError::UnresolvedShape {
value: name,
op: "<control-flow input>".to_string(),
}
})?;
let bytes = self.contiguous_bytes(vid, &shape, dtype)?;
Tensor::from_raw(dtype, shape, &bytes)
}
fn root_of(&self, vid: ValueId) -> ValueId {
match self.views.get(&vid) {
Some(v) => v.source,
None => vid,
}
}
fn try_move_host_output(
&mut self,
vid: ValueId,
shape: &[usize],
dtype: DataType,
) -> Result<Option<Tensor>> {
if self.views.contains_key(&vid)
|| self.seq_elem_values.contains_key(&vid)
|| self.shared_buffers.contains_key(&vid)
|| self.pinned.contains(&vid)
{
return Ok(None);
}
if self
.graph
.try_value(vid)
.is_none_or(|value| value.producer.is_none())
{
return Ok(None);
}
if let Some(producer) = self.graph.try_value(vid).and_then(|value| value.producer)
&& self.if_last_predicate.contains_key(&producer)
{
return Ok(None);
}
if self.graph.outputs.iter().filter(|&&o| o == vid).count() != 1 {
return Ok(None);
}
let value_name = || format!("value#{}", vid.0);
let numel = checked_numel(shape, value_name)?;
let n = checked_storage_bytes(dtype, numel, value_name, shape)?;
let movable = self.buffers.get(&vid).is_some_and(|buf| {
buf.device().is_host_accessible() && !buf.is_borrowed() && buf.len() == n
});
if !movable {
return Ok(None);
}
let buffer = self
.buffers
.remove(&vid)
.expect("buffer presence checked above");
self.buffer_shapes.remove(&vid);
Ok(Some(Tensor::from_owned_buffer(
self.ep.clone(),
dtype,
shape.to_vec(),
buffer,
)))
}
fn contiguous_bytes(&self, vid: ValueId, shape: &[usize], dtype: DataType) -> Result<Vec<u8>> {
let value_name = || {
self.graph
.try_value(vid)
.and_then(|value| value.name.clone())
.unwrap_or_else(|| format!("value#{}", vid.0))
};
let numel = checked_numel(shape, value_name)?;
let n = checked_storage_bytes(dtype, numel, value_name, shape)?;
if let Some(elem) = self.seq_elem_values.get(&vid) {
let bytes = elem.contiguous_bytes().map_err(SessionError::from)?;
return Ok(bytes[..n.min(bytes.len())].to_vec());
}
if let Some(view) = self.views.get(&vid) {
let buf = self.buffers.get(&view.source).ok_or_else(|| {
SessionError::Internal(format!(
"view value#{} aliases missing source buffer value#{}",
vid.0, view.source.0
))
})?;
let esize = dtype.byte_size();
if esize == 0 {
return Err(SessionError::Internal(format!(
"cannot materialize sub-byte view value#{}",
vid.0
)));
}
let mut host = vec![0u8; buf.len()];
self.ep.copy_to_host(buf, &mut host)?;
Ok(gather_view(
&host,
&view.shape,
&view.strides,
view.byte_offset,
esize,
))
} else {
let buf = self
.buffers
.get(&vid)
.ok_or_else(|| SessionError::Internal(format!("value#{} not produced", vid.0)))?;
let mut host = vec![0u8; n];
self.ep.copy_to_host(buf, &mut host)?;
Ok(host)
}
}
fn store_output_tensor(
&mut self,
vid: ValueId,
tensor: &Tensor,
resolved: &mut HashMap<ValueId, Vec<usize>>,
) -> Result<()> {
self.store_output_bytes(
vid,
tensor.dtype,
tensor.shape.clone(),
tensor.as_bytes(),
resolved,
)
}
fn store_output_bytes(
&mut self,
vid: ValueId,
dtype: DataType,
dims: Vec<usize>,
bytes: &[u8],
resolved: &mut HashMap<ValueId, Vec<usize>>,
) -> Result<()> {
let numel = checked_numel(&dims, || format!("value#{}", vid.0))?;
let need =
checked_storage_bytes(dtype, numel, || format!("value#{}", vid.0), &dims)?.max(1);
let fits = self
.buffers
.get(&vid)
.map(|b| b.len() == need)
.unwrap_or(false);
if !fits {
if let Some(old) = self.buffers.remove(&vid) {
self.ep.deallocate(old)?;
}
self.shared_buffers.remove(&vid);
let buf = self
.ep
.allocate(need, TensorLayout::contiguous().alignment)?;
self.buffers.insert(vid, buf);
}
let buf = self.buffers.get_mut(&vid).expect("just ensured");
self.ep.copy_from_host(bytes, buf)?;
self.value_dtypes.insert(vid, dtype);
self.buffer_shapes.insert(vid, dims.clone());
resolved.insert(vid, dims);
Ok(())
}
fn prepare_subgraph(
&self,
node_id: NodeId,
attr_key: &str,
resolved: &HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
) -> Result<PreparedSubgraph> {
let key = (node_id, attr_key.to_string());
let body = self.graph.subgraphs.get(&key).ok_or_else(|| {
SessionError::Internal(format!(
"control-flow node #{} references missing subgraph '{attr_key}'",
node_id.0
))
})?;
let mut scope_names = required_outer_names(body).into_iter().collect::<Vec<_>>();
scope_names.sort();
let mut captures = HashMap::with_capacity(scope_names.len());
for name in scope_names {
let tensor = if let Some(&vid) = self.name_index.get(&name) {
let materialized = self.buffers.contains_key(&vid)
|| self.views.contains_key(&vid)
|| self.seq_elem_values.contains_key(&vid);
if resolved.contains_key(&vid) && materialized {
self.value_tensor(vid, resolved)?
} else {
outer_scope
.get(&name)
.cloned()
.ok_or_else(|| missing_capture_error(attr_key, &name))?
}
} else {
outer_scope
.get(&name)
.cloned()
.ok_or_else(|| missing_capture_error(attr_key, &name))?
};
captures.insert(name, tensor);
}
Ok(PreparedSubgraph { key, captures })
}
fn run_subgraph(
&mut self,
prepared: &PreparedSubgraph,
formal_inputs: &[&Tensor],
) -> Result<Vec<Tensor>> {
if !self.subgraph_execs.contains_key(&prepared.key) {
let body = self
.graph
.subgraphs
.get(&prepared.key)
.cloned()
.ok_or_else(|| {
SessionError::Internal(format!(
"control-flow node #{} has no registered subgraph '{}'",
prepared.key.0.0, prepared.key.1
))
})?;
let mut child = ChildExecutor::new(
format!("node#{}/{}", prepared.key.0.0, prepared.key.1),
body,
self.graph.opset_imports.clone(),
self.weights.clone(),
self.ep.clone(),
)?;
child.set_trace_context(self.trace.clone());
self.subgraph_execs.insert(prepared.key.clone(), child);
}
let child = self
.subgraph_execs
.get_mut(&prepared.key)
.expect("child present");
let before = child.stats();
let result = child.run(formal_inputs, &prepared.captures);
let after = child.stats();
self.control_flow_stats.subgraph_builds += after.builds - before.builds;
self.control_flow_stats.subgraph_runs += after.runs - before.runs;
result
}
fn exec_control_flow(
&mut self,
pi: usize,
resolved: &mut HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
) -> Result<()> {
let node = self.graph.node(self.plan[pi].node_id).clone();
match node.op_type.as_str() {
"If" => self.exec_if(&node, resolved, outer_scope),
"Loop" => self.exec_loop(&node, resolved, outer_scope),
"Scan" => self.exec_scan(&node, resolved, outer_scope),
other => Err(SessionError::Internal(format!(
"exec_control_flow reached non-control-flow op {other:?}"
))),
}
}
fn exec_if(
&mut self,
node: &Node,
resolved: &mut HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
) -> Result<()> {
{
let then_branch = self
.graph
.subgraphs
.get(&(node.id, "then_branch".to_string()))
.ok_or_else(|| SessionError::ControlFlow {
op: "If".to_string(),
reason: "missing required 'then_branch' subgraph".to_string(),
})?;
let else_branch = self
.graph
.subgraphs
.get(&(node.id, "else_branch".to_string()))
.ok_or_else(|| SessionError::ControlFlow {
op: "If".to_string(),
reason: "missing required 'else_branch' subgraph".to_string(),
})?;
if !then_branch.inputs.is_empty() || !else_branch.inputs.is_empty() {
return Err(SessionError::ControlFlow {
op: "If".to_string(),
reason: format!(
"branch subgraphs must declare zero formal inputs, but then_branch has {} \
and else_branch has {}",
then_branch.inputs.len(),
else_branch.inputs.len()
),
});
}
validate_if_branch_outputs(&self.graph, node)?;
}
let cond_vid =
node.inputs
.first()
.and_then(|s| *s)
.ok_or_else(|| SessionError::ControlFlow {
op: "If".to_string(),
reason: "missing required 'cond' input".to_string(),
})?;
let cond_t = self.value_tensor(cond_vid, resolved)?;
if cond_t.dtype != DataType::Bool {
return Err(SessionError::DtypeMismatch {
name: "If cond".to_string(),
expected: format!("{:?}", DataType::Bool),
got: format!("{:?}", cond_t.dtype),
});
}
let cond = tensor_scalar_bool(&cond_t).ok_or_else(|| SessionError::ControlFlow {
op: "If".to_string(),
reason: format!(
"'cond' must be a BOOL scalar or single-element tensor, got shape {:?}",
cond_t.shape
),
})?;
if self.if_last_predicate.get(&node.id) == Some(&cond)
&& node.outputs.iter().all(|v| resolved.contains_key(v))
{
return Ok(());
}
let attr_key = if cond { "then_branch" } else { "else_branch" };
let taken_branch_is_invariant = self
.graph
.subgraphs
.get(&(node.id, attr_key.to_string()))
.map(|body| required_outer_names(body).is_empty())
.unwrap_or(false);
let prepared = {
let _s = phase_span!("execif.prepare_subgraph");
self.prepare_subgraph(node.id, attr_key, resolved, outer_scope)?
};
let outs = {
let _s = phase_span!("execif.run_subgraph");
self.run_subgraph(&prepared, &[])?
};
if outs.len() != node.outputs.len() {
return Err(SessionError::OutputShapeCountMismatch {
op: format!("If/{attr_key}"),
expected: node.outputs.len(),
got: outs.len(),
});
}
{
let _s = phase_span!("execif.store_output");
for (vid, t) in node.outputs.iter().zip(outs.iter()) {
self.store_output_tensor(*vid, t, resolved)?;
}
}
if taken_branch_is_invariant {
self.if_last_predicate.insert(node.id, cond);
} else {
self.if_last_predicate.remove(&node.id);
}
Ok(())
}
fn loop_body_scan_specs(
&self,
node: &Node,
carried: &[Tensor],
num_scan: usize,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Result<OptionalTensorSpecs> {
let body = self
.graph
.subgraphs
.get(&(node.id, "body".to_string()))
.ok_or_else(|| SessionError::ControlFlow {
op: "Loop".to_string(),
reason: "missing required 'body' subgraph".to_string(),
})?;
let expected_inputs = 2 + carried.len();
if body.inputs.len() != expected_inputs {
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"body declares {} formal input(s), expected {expected_inputs}",
body.inputs.len()
),
});
}
let expected_outputs = 1 + carried.len() + num_scan;
if body.outputs.len() != expected_outputs {
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"body declares {} output(s), expected {expected_outputs}",
body.outputs.len()
),
});
}
for (index, expected) in [(0, DataType::Int64), (1, DataType::Bool)] {
let input = body.inputs[index];
if body.value_type_is_known(input) && body.value(input).dtype != expected {
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"body formal input {index} must be {expected:?}, got {:?}",
body.value(input).dtype
),
});
}
}
let cond_out = body.outputs[0];
if body.value_type_is_known(cond_out) && body.value(cond_out).dtype != DataType::Bool {
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"body output 0 ('cond_out') must be Bool, got {:?}",
body.value(cond_out).dtype
),
});
}
for (index, initial) in carried.iter().enumerate() {
for (kind, value) in [
("formal input", body.inputs[2 + index]),
("output", body.outputs[1 + index]),
] {
if body.value_type_is_known(value) && body.value(value).dtype != initial.dtype {
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"loop-carried {kind} {index} has dtype {:?}, but its initial value has \
dtype {:?}",
body.value(value).dtype,
initial.dtype
),
});
}
}
}
body.outputs
.iter()
.skip(1 + carried.len())
.zip(node.outputs.iter().skip(carried.len()))
.enumerate()
.map(|(index, (&body_output, &node_output))| {
let body_value = body.value(body_output);
let node_dtype = self.value_dtypes[&node_output];
let dtype = if body.value_type_is_known(body_output) {
if self.graph.value_type_is_known(node_output)
&& body_value.dtype != node_dtype
{
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"scan output {index} has body dtype {:?}, but the Loop node declares \
{node_dtype:?}",
body_value.dtype
),
});
}
body_value.dtype
} else {
node_dtype
};
let elem_shape = body
.value_shape_is_known(body_output)
.then(|| as_static_shape(&body_value.shape))
.flatten()
.or_else(|| {
resolved
.get(&node_output)
.and_then(|shape| shape.get(1..).map(<[_]>::to_vec))
});
Ok(elem_shape.map(|shape| (dtype, shape)))
})
.collect()
}
fn exec_loop(
&mut self,
node: &Node,
resolved: &mut HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
) -> Result<()> {
let m: Option<i64> = match node.inputs.first().and_then(|s| *s) {
Some(vid) => {
let t = self.value_tensor(vid, resolved)?;
if t.dtype != DataType::Int64 {
return Err(SessionError::DtypeMismatch {
name: "Loop M".to_string(),
expected: format!("{:?}", DataType::Int64),
got: format!("{:?}", t.dtype),
});
}
let m = tensor_scalar_i64(&t).ok_or_else(|| SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"'M' must be an INT64 scalar or single-element tensor, got shape {:?}",
t.shape
),
})?;
Some(m)
}
None => None,
};
let mut cond: Option<bool> =
match node.inputs.get(1).and_then(|s| *s) {
Some(vid) => {
let t = self.value_tensor(vid, resolved)?;
if t.dtype != DataType::Bool {
return Err(SessionError::DtypeMismatch {
name: "Loop cond".to_string(),
expected: format!("{:?}", DataType::Bool),
got: format!("{:?}", t.dtype),
});
}
Some(tensor_scalar_bool(&t).ok_or_else(|| SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"'cond' must be a BOOL scalar or single-element tensor, got shape {:?}",
t.shape
),
})?)
}
None => None,
};
let mut carried: Vec<Tensor> = Vec::new();
for slot in node.inputs.iter().skip(2) {
let vid = slot.ok_or_else(|| {
SessionError::Internal(
"Loop: an interior loop-carried input is omitted (empty), which ONNX does not \
allow — every v_initial must be provided"
.to_string(),
)
})?;
carried.push(self.value_tensor(vid, resolved)?);
}
let num_carried = carried.len();
let carried_invariants: Vec<(DataType, Vec<usize>)> = carried
.iter()
.map(|tensor| (tensor.dtype, tensor.shape.clone()))
.collect();
let num_outputs = node.outputs.len();
if num_outputs < num_carried {
return Err(SessionError::Internal(format!(
"Loop: node declares {num_outputs} output(s) but has {num_carried} loop-carried \
dependency(ies); outputs must be carried-finals followed by scan-outputs"
)));
}
let num_scan = num_outputs - num_carried;
let empty_scan_specs = self.loop_body_scan_specs(node, &carried, num_scan, resolved)?;
let mut scan_acc: Vec<TensorStackAccumulator> = (0..num_scan)
.map(|_| TensorStackAccumulator::new())
.collect();
let prepared = self.prepare_subgraph(node.id, "body", resolved, outer_scope)?;
let mut iter_tensor = scalar_i64_tensor(0)?;
let mut cond_tensor = scalar_bool_tensor(cond.unwrap_or(true))?;
let mut iter: i64 = 0;
loop {
if let Some(m) = m
&& iter >= m
{
break;
}
if cond == Some(false) {
break;
}
iter_tensor.overwrite_bytes(&iter.to_le_bytes())?;
cond_tensor.overwrite_bytes(&[u8::from(cond.unwrap_or(true))])?;
let mut formal: Vec<&Tensor> = Vec::with_capacity(2 + num_carried);
formal.push(&iter_tensor);
formal.push(&cond_tensor);
formal.extend(carried.iter());
let outs = self.run_subgraph(&prepared, &formal)?;
drop(formal);
let expected = 1 + num_carried + num_scan;
if outs.len() != expected {
return Err(SessionError::OutputShapeCountMismatch {
op: "Loop/body".to_string(),
expected,
got: outs.len(),
});
}
let mut it = outs.into_iter();
let cond_out = it.next().expect("cond_out present");
cond = Some(tensor_scalar_bool(&cond_out).ok_or_else(|| {
SessionError::Internal(format!(
"Loop: body's first output 'cond_out' must be a BOOL scalar, got dtype {:?}",
cond_out.dtype
))
})?);
let next_carried: Vec<Tensor> = (&mut it).take(num_carried).collect();
for (index, (tensor, (expected_dtype, expected_shape))) in
next_carried.iter().zip(&carried_invariants).enumerate()
{
if tensor.dtype != *expected_dtype {
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"loop-carried output {index} dtype mismatch: expected \
{expected_dtype:?}, got {:?}",
tensor.dtype
),
});
}
if tensor.shape != *expected_shape {
return Err(SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"loop-carried output {index} shape mismatch: expected \
{expected_shape:?}, got {:?}",
tensor.shape
),
});
}
}
carried = next_carried;
for acc in scan_acc.iter_mut() {
acc.push(it.next().expect("scan output present"))?;
}
iter = iter
.checked_add(1)
.ok_or_else(|| SessionError::ControlFlow {
op: "Loop".to_string(),
reason: "iteration counter overflowed INT64".to_string(),
})?;
}
for (i, t) in carried.iter().enumerate() {
self.store_output_tensor(node.outputs[i], t, resolved)?;
}
for (s, (acc, empty_spec)) in scan_acc.into_iter().zip(empty_scan_specs).enumerate() {
let (dtype, shape, bytes) = acc.finish_with_empty(empty_spec, s)?;
self.store_output_bytes(
node.outputs[num_carried + s],
dtype,
shape,
&bytes,
resolved,
)?;
}
Ok(())
}
fn scan_body_specs(
&self,
node: &Node,
state: &[Tensor],
scan_inputs: &[Tensor],
input_axes: &[usize],
num_scan_outputs: usize,
output_axes: &[i64],
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Result<OptionalTensorSpecs> {
let body = self
.graph
.subgraphs
.get(&(node.id, "body".to_string()))
.ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: "missing required 'body' subgraph".to_string(),
})?;
let expected_inputs = state.len() + scan_inputs.len();
if body.inputs.len() != expected_inputs {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"body declares {} formal input(s), expected {expected_inputs}",
body.inputs.len()
),
});
}
let expected_outputs = state.len() + num_scan_outputs;
if body.outputs.len() != expected_outputs {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"body declares {} output(s), expected {expected_outputs}",
body.outputs.len()
),
});
}
for (index, initial) in state.iter().enumerate() {
for (kind, value) in [
("formal input", body.inputs[index]),
("output", body.outputs[index]),
] {
if body.value_type_is_known(value) && body.value(value).dtype != initial.dtype {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"state {kind} {index} has dtype {:?}, but its initial value has dtype {:?}",
body.value(value).dtype,
initial.dtype
),
});
}
}
}
for (index, ((input, &axis), &formal)) in scan_inputs
.iter()
.zip(input_axes)
.zip(body.inputs.iter().skip(state.len()))
.enumerate()
{
if body.value_type_is_known(formal) && body.value(formal).dtype != input.dtype {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"scan formal input {index} has dtype {:?}, but scan input {index} has dtype {:?}",
body.value(formal).dtype,
input.dtype
),
});
}
let mut slice_shape = input.shape.clone();
slice_shape.remove(axis);
if body.value_shape_is_known(formal)
&& let Some(shape) = as_static_shape(&body.value(formal).shape)
&& shape != slice_shape
{
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"scan formal input {index} has shape {shape:?}, but slicing input shape {:?} \
along axis {axis} produces {slice_shape:?}",
input.shape
),
});
}
}
body.outputs
.iter()
.skip(state.len())
.zip(node.outputs.iter().skip(state.len()))
.zip(output_axes)
.enumerate()
.map(|(index, ((&body_output, &node_output), &axis))| {
let body_value = body.value(body_output);
let node_dtype = self.value_dtypes[&node_output];
let dtype = if body.value_type_is_known(body_output) {
if self.graph.value_type_is_known(node_output)
&& body_value.dtype != node_dtype
{
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"scan output {index} has body dtype {:?}, but the Scan node declares \
{node_dtype:?}",
body_value.dtype
),
});
}
body_value.dtype
} else {
node_dtype
};
let elem_shape = body
.value_shape_is_known(body_output)
.then(|| as_static_shape(&body_value.shape))
.flatten()
.or_else(|| {
resolved.get(&node_output).and_then(|shape| {
normalize_axis(axis, shape.len()).map(|axis| {
let mut elem_shape = shape.clone();
elem_shape.remove(axis);
elem_shape
})
})
});
if let Some(shape) = &elem_shape
&& normalize_axis(axis, shape.len() + 1).is_none()
{
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"scan_output_axes[{index}]={axis} is out of range for output rank {}",
shape.len() + 1
),
});
}
Ok(elem_shape.map(|shape| (dtype, shape)))
})
.collect()
}
fn exec_scan(
&mut self,
node: &Node,
resolved: &mut HashMap<ValueId, Vec<usize>>,
outer_scope: &HashMap<String, Tensor>,
) -> Result<()> {
let raw_num_scan_inputs = node
.attr("num_scan_inputs")
.and_then(|a| a.as_int())
.ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: "required attribute 'num_scan_inputs' is missing or not an INT".to_string(),
})?;
let num_scan_inputs = usize::try_from(raw_num_scan_inputs)
.ok()
.filter(|&count| count != 0)
.ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"'num_scan_inputs' must be a positive INT, got {raw_num_scan_inputs}"
),
})?;
let total_inputs = node.inputs.len();
if total_inputs < num_scan_inputs {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"node has {total_inputs} input(s) but num_scan_inputs={num_scan_inputs}"
),
});
}
let num_state = total_inputs - num_scan_inputs;
if node.outputs.len() < num_state {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"declares {} output(s) but has {num_state} state variable(s)",
node.outputs.len()
),
});
}
let num_scan_outputs = node.outputs.len() - num_state;
let input_axes_raw = scan_list_attr(node, "scan_input_axes", num_scan_inputs, 0)?;
let input_directions = scan_list_attr(node, "scan_input_directions", num_scan_inputs, 0)?;
let output_axes = scan_list_attr(node, "scan_output_axes", num_scan_outputs, 0)?;
let output_directions =
scan_list_attr(node, "scan_output_directions", num_scan_outputs, 0)?;
for (name, values) in [
("scan_input_directions", &input_directions),
("scan_output_directions", &output_directions),
] {
for (index, &value) in values.iter().enumerate() {
if !matches!(value, 0 | 1) {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"{name}[{index}] must be 0 (forward) or 1 (reverse), got {value}"
),
});
}
}
}
let mut state: Vec<Tensor> = Vec::with_capacity(num_state);
for slot in node.inputs.iter().take(num_state) {
let vid = slot.ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: "an initial-state input is omitted (empty), which ONNX does not allow"
.to_string(),
})?;
state.push(self.value_tensor(vid, resolved)?);
}
let mut scan_inputs: Vec<Tensor> = Vec::with_capacity(num_scan_inputs);
for slot in node.inputs.iter().skip(num_state) {
let vid = slot.ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: "a scan input is omitted (empty), which ONNX does not allow".to_string(),
})?;
scan_inputs.push(self.value_tensor(vid, resolved)?);
}
let mut input_axes = Vec::with_capacity(num_scan_inputs);
for (index, (input, &raw_axis)) in scan_inputs.iter().zip(&input_axes_raw).enumerate() {
let axis = normalize_axis(raw_axis, input.shape.len()).ok_or_else(|| {
SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"scan_input_axes[{index}]={raw_axis} is out of range for input rank {}",
input.shape.len()
),
}
})?;
input_axes.push(axis);
}
let trip_count = scan_inputs[0].shape[input_axes[0]];
for (index, (input, &axis)) in scan_inputs.iter().zip(&input_axes).enumerate() {
let length = input.shape[axis];
if length != trip_count {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"scan input {index} has scan-axis length {length}, but the first scan input \
has {trip_count}; all scan inputs must agree"
),
});
}
}
let state_specs: Vec<(DataType, Vec<usize>)> = state
.iter()
.map(|tensor| (tensor.dtype, tensor.shape.clone()))
.collect();
let empty_specs = self.scan_body_specs(
node,
&state,
&scan_inputs,
&input_axes,
num_scan_outputs,
&output_axes,
resolved,
)?;
let mut scan_acc: Vec<TensorStackAccumulator> = (0..num_scan_outputs)
.map(|_| TensorStackAccumulator::new())
.collect();
let prepared = self.prepare_subgraph(node.id, "body", resolved, outer_scope)?;
let mut scan_slices = Vec::with_capacity(num_scan_inputs);
if trip_count != 0 {
for (index, ((input, &axis), &direction)) in scan_inputs
.iter()
.zip(&input_axes)
.zip(&input_directions)
.enumerate()
{
let source_index = if direction == 0 { 0 } else { trip_count - 1 };
let (shape, bytes) = scan_slice(input, axis, source_index, index)?;
scan_slices.push(Tensor::from_raw(input.dtype, shape, &bytes)?);
}
}
for step in 0..trip_count {
if step != 0 {
for (index, (((input, &axis), &direction), slice)) in scan_inputs
.iter()
.zip(&input_axes)
.zip(&input_directions)
.zip(scan_slices.iter_mut())
.enumerate()
{
let source_index = if direction == 0 {
step
} else {
trip_count - 1 - step
};
let (_, bytes) = scan_slice(input, axis, source_index, index)?;
slice.overwrite_bytes(&bytes)?;
}
}
let mut formal: Vec<&Tensor> = Vec::with_capacity(num_state + num_scan_inputs);
formal.extend(state.iter());
formal.extend(scan_slices.iter());
let outs = self.run_subgraph(&prepared, &formal)?;
drop(formal);
let expected = num_state + num_scan_outputs;
if outs.len() != expected {
return Err(SessionError::OutputShapeCountMismatch {
op: "Scan/body".to_string(),
expected,
got: outs.len(),
});
}
let mut it = outs.into_iter();
let next_state: Vec<Tensor> = (&mut it).take(num_state).collect();
for (index, (tensor, (expected_dtype, expected_shape))) in
next_state.iter().zip(&state_specs).enumerate()
{
if tensor.dtype != *expected_dtype {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"state output {index} dtype mismatch: expected {expected_dtype:?}, got {:?}",
tensor.dtype
),
});
}
if tensor.shape != *expected_shape {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"state output {index} shape mismatch: expected {expected_shape:?}, got {:?}",
tensor.shape
),
});
}
}
state = next_state;
for acc in scan_acc.iter_mut() {
acc.push(it.next().expect("scan output present"))?;
}
}
for (i, t) in state.iter().enumerate() {
self.store_output_tensor(node.outputs[i], t, resolved)?;
}
for (s, ((acc, empty_spec), (&axis, &direction))) in scan_acc
.into_iter()
.zip(empty_specs)
.zip(output_axes.iter().zip(&output_directions))
.enumerate()
{
let (dtype, shape, bytes) = acc.finish_scan(axis, direction, empty_spec, s)?;
self.store_output_bytes(node.outputs[num_state + s], dtype, shape, &bytes, resolved)?;
}
Ok(())
}
}
fn scan_slice(
t: &Tensor,
axis: usize,
index: usize,
input_index: usize,
) -> Result<(Vec<usize>, Vec<u8>)> {
let axis_len = t.shape[axis];
if index >= axis_len {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"slice index {index} is out of range for scan input {input_index} axis {axis}"
),
});
}
let esize = t.dtype.byte_size();
if esize == 0 {
return Err(SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"sub-byte dtype {:?} for scan input {input_index} is not supported",
t.dtype
),
});
}
let mut shape = t.shape.clone();
shape.remove(axis);
let outer = checked_numel(&t.shape[..axis], || format!("Scan input {input_index}"))?;
let inner = checked_numel(&t.shape[axis + 1..], || format!("Scan input {input_index}"))?;
let inner_bytes = checked_storage_bytes(
t.dtype,
inner,
|| format!("Scan input {input_index}"),
&t.shape,
)?;
let total_bytes =
outer
.checked_mul(inner_bytes)
.ok_or_else(|| SessionError::ShapeOverflow {
value: format!("Scan input {input_index} slice"),
dims: shape.clone(),
})?;
let source = t.as_bytes();
let mut bytes = vec![0u8; total_bytes];
for outer_index in 0..outer {
let src = (outer_index * axis_len + index) * inner_bytes;
let dst = outer_index * inner_bytes;
bytes[dst..dst + inner_bytes].copy_from_slice(&source[src..src + inner_bytes]);
}
Ok((shape, bytes))
}
struct TensorStackAccumulator {
dtype: Option<DataType>,
elem_shape: Vec<usize>,
len: usize,
bytes: Vec<u8>,
}
impl TensorStackAccumulator {
fn new() -> Self {
Self {
dtype: None,
elem_shape: Vec::new(),
len: 0,
bytes: Vec::new(),
}
}
fn push(&mut self, tensor: Tensor) -> Result<()> {
if let Some(dtype) = self.dtype {
if tensor.shape != self.elem_shape || tensor.dtype != dtype {
return Err(SessionError::Internal(format!(
"Loop/Scan: scan output slice {} has shape {:?} dtype {:?} but the first slice \
is shape {:?} dtype {:?}; every iteration's scan output must match",
self.len, tensor.shape, tensor.dtype, self.elem_shape, dtype
)));
}
} else {
if tensor.dtype.byte_size() == 0 {
return Err(SessionError::Internal(format!(
"Loop/Scan: sub-byte dtype {:?} scan outputs are not supported",
tensor.dtype
)));
}
self.dtype = Some(tensor.dtype);
self.elem_shape = tensor.shape.clone();
}
self.bytes.extend_from_slice(tensor.as_bytes());
self.len += 1;
Ok(())
}
fn finish(self) -> (DataType, Vec<usize>, Vec<u8>) {
if self.len == 0 {
return (DataType::Float32, vec![0], Vec::new());
}
let dtype = self.dtype.expect("non-empty accumulator has dtype");
let mut shape = Vec::with_capacity(1 + self.elem_shape.len());
shape.push(self.len);
shape.extend(self.elem_shape);
(dtype, shape, self.bytes)
}
fn finish_with_empty(
self,
empty_spec: Option<(DataType, Vec<usize>)>,
output_index: usize,
) -> Result<(DataType, Vec<usize>, Vec<u8>)> {
if self.len != 0 {
return Ok(self.finish());
}
let (dtype, elem_shape) = empty_spec.ok_or_else(|| SessionError::ControlFlow {
op: "Loop".to_string(),
reason: format!(
"cannot determine the element shape of scan output {output_index} for a \
zero-iteration result"
),
})?;
let mut shape = Vec::with_capacity(1 + elem_shape.len());
shape.push(0);
shape.extend(elem_shape);
Ok((dtype, shape, Vec::new()))
}
fn finish_scan(
self,
axis: i64,
direction: i64,
empty_spec: Option<(DataType, Vec<usize>)>,
output_index: usize,
) -> Result<(DataType, Vec<usize>, Vec<u8>)> {
let (dtype, elem_shape) = match self.dtype {
Some(dtype) => (dtype, self.elem_shape.clone()),
None => empty_spec.ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"cannot determine the element shape of scan output {output_index} for a \
zero-iteration result"
),
})?,
};
let output_rank = elem_shape.len() + 1;
let axis = normalize_axis(axis, output_rank).ok_or_else(|| SessionError::ControlFlow {
op: "Scan".to_string(),
reason: format!(
"scan_output_axes[{output_index}]={axis} is out of range for output rank \
{output_rank}"
),
})?;
if self.len == 0 {
let mut shape = elem_shape;
shape.insert(axis, 0);
return Ok((dtype, shape, Vec::new()));
}
if axis == 0 && direction == 0 {
let mut shape = Vec::with_capacity(output_rank);
shape.push(self.len);
shape.extend(elem_shape);
return Ok((dtype, shape, self.bytes));
}
let elem_numel = checked_numel(&elem_shape, || {
format!("Scan output {output_index} element")
})?;
let elem_bytes = checked_storage_bytes(
dtype,
elem_numel,
|| format!("Scan output {output_index} element"),
&elem_shape,
)?;
let mut elements: Vec<&[u8]> = if elem_bytes == 0 {
(0..self.len).map(|_| &self.bytes[..]).collect()
} else {
self.bytes.chunks_exact(elem_bytes).collect()
};
if direction == 1 {
elements.reverse();
}
let (shape, bytes) = stack_new_axis(&elements, &elem_shape, axis, dtype.byte_size())?;
Ok((dtype, shape, bytes))
}
}
impl Drop for Executor {
fn drop(&mut self) {
if self.decode_memo_enabled
&& std::env::var("ONNX_GENAI_DECODE_MEMO_STATS")
.map(|v| matches!(v.as_str(), "1" | "true" | "on"))
.unwrap_or(false)
{
let (primed, rebuilt, replayed, ineligible) = self.decode_memo_counts();
let (views_reused, dispatch_elided) = self.decode_view_plan_counts();
eprintln!(
"[decode-memo] primed={primed} rebuilt={rebuilt} replayed={replayed} \
ineligible={ineligible} views_reused={views_reused} \
dispatch_elided={dispatch_elided}"
);
}
let _ = self.ep.reset_device_graph();
self.device_graph_signature = None;
for (_, buf) in self.buffers.drain() {
let _ = self.ep.deallocate(buf);
}
self.shared_buffers.clear();
}
}
pub(crate) fn auto_detect_cpu_ep() -> Result<Arc<dyn ExecutionProvider>> {
let mut ep = CpuExecutionProvider::new();
ep.initialize(&Default::default())?;
Ok(Arc::new(ep))
}
#[cfg(test)]
mod tests {
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use onnx_runtime_ep_api::{
CaptureSupport, Cost, EpConfig, EpError, ExecutionProviderCapabilities, Fence, Kernel,
NegotiatedWeight,
};
use super::*;
#[test]
fn phase_profile_gating_and_accumulation() {
phase_profile::force_enabled(false);
let disabled_phase = "test.phase.disabled";
let before = phase_profile::snapshot(disabled_phase);
{
let _s = phase_span!(disabled_phase);
std::thread::sleep(std::time::Duration::from_millis(1));
}
assert_eq!(
phase_profile::snapshot(disabled_phase),
before,
"a disabled phase span must not accumulate any samples"
);
phase_profile::force_enabled(true);
let enabled_phase = "test.phase.enabled";
let (base_ns, base_count) = phase_profile::snapshot(enabled_phase).unwrap_or((0, 0));
{
let _s = phase_span!(enabled_phase);
std::thread::sleep(std::time::Duration::from_millis(2));
}
let (after_ns, after_count) =
phase_profile::snapshot(enabled_phase).expect("enabled span must record a sample");
assert_eq!(after_count, base_count + 1, "one span => one sample");
assert!(
after_ns > base_ns,
"an enabled span must accumulate a positive duration"
);
phase_profile::force_enabled(false);
}
#[test]
fn zero_copy_output_move_reallocates_and_preserves_producer_less_output() {
use onnx_runtime_ir::TensorData;
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let a = graph.create_named_value("a", DataType::Float32, static_shape([3]));
let b = graph.create_named_value("b", DataType::Float32, static_shape([3]));
graph.add_input(a);
graph.add_input(b);
let k = graph.create_named_value("k", DataType::Float32, static_shape([3]));
graph.set_initializer(
k,
WeightRef::Inline(TensorData::from_raw(
DataType::Float32,
vec![3],
[100.0f32, 200.0, 300.0]
.into_iter()
.flat_map(f32::to_le_bytes)
.collect(),
)),
);
let sum = graph.create_named_value("sum", DataType::Float32, static_shape([3]));
graph.insert_node(Node::new(
NodeId(0),
"Add",
vec![Some(a), Some(b)],
vec![sum],
));
graph.add_output(sum);
graph.add_output(k);
let mut executor = Executor::build(
graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
let a_val = Tensor::from_f32(&[3], &[1.0, 2.0, 3.0]).unwrap();
let b_val = Tensor::from_f32(&[3], &[10.0, 20.0, 30.0]).unwrap();
for _ in 0..3 {
let outputs = executor
.run(&[("a", &a_val), ("b", &b_val)])
.expect("run must succeed after a prior output buffer was moved out");
assert_eq!(outputs[0].to_vec_f32(), vec![11.0, 22.0, 33.0]);
assert_eq!(
outputs[1].to_vec_f32(),
vec![100.0, 200.0, 300.0],
"producer-less initializer output must stay intact across runs"
);
assert!(
!executor.buffers.contains_key(&sum),
"produced output buffer must be moved out, not copied"
);
assert!(
executor.buffers.contains_key(&k),
"producer-less output must not have its buffer stolen"
);
}
}
struct CaptureDecliningKernel;
impl Kernel for CaptureDecliningKernel {
fn execute(
&self,
_inputs: &[TensorView],
_outputs: &mut [TensorMut],
) -> onnx_runtime_ep_api::Result<()> {
Ok(())
}
fn capture_support(&self) -> CaptureSupport {
CaptureSupport::unsupported(
"requires M==1 decode GEMV without group_indices; got a prefill signature",
)
}
}
#[test]
fn kernel_capture_reason_propagates_into_structured_report() {
let mut node = Node::new(NodeId(9), "MatMulNBits", vec![], vec![]);
node.domain = "com.microsoft".to_string();
let decline =
kernel_capture_decline(node.id, &node, &CaptureDecliningKernel).expect("decline");
let report = CaptureDeclineReport::one(decline);
assert_eq!(
report.entries,
vec![CaptureDecline {
node_id: Some(9),
op_type: "MatMulNBits".to_string(),
domain: "com.microsoft".to_string(),
reason: "requires M==1 decode GEMV without group_indices; got a prefill signature"
.to_string(),
seam_reason: Some(SeamReason::KernelCaptureUnsupported),
}]
);
assert!(report.to_string().contains("node 9"));
assert!(
report
.to_string()
.contains("requires M==1 decode GEMV without group_indices")
);
}
#[test]
fn seam_reasons_map_to_structural_capture_paths() {
let cases = [
(
SeamReason::HostControlFlowOrSequence,
CapturePathKind::HostSeam,
"host-seam",
),
(
SeamReason::UnresolvedOutputShape,
CapturePathKind::EagerDeviceSeam,
"eager-device-seam",
),
(
SeamReason::UnresolvedInputShape,
CapturePathKind::EagerDeviceSeam,
"eager-device-seam",
),
(
SeamReason::KernelNotWarmed,
CapturePathKind::EagerDeviceSeam,
"eager-device-seam",
),
(
SeamReason::KernelCaptureUnsupported,
CapturePathKind::EagerDeviceSeam,
"eager-device-seam",
),
];
for (reason, expected_kind, expected_label) in cases {
assert_eq!(reason.path_kind(), expected_kind);
assert_eq!(reason.label(), expected_label);
}
assert_eq!(CapturePathKind::CaptureRegion.label(), "capture-region");
}
#[test]
fn ep_structural_plan_plus_executor_kernel_checks_matches_legacy_declines() {
use onnx_runtime_ir::static_shape;
fn legacy_node_capture_reason(
executor: &Executor,
plan: &NodePlan,
resolved: &HashMap<ValueId, Vec<usize>>,
) -> Option<CaptureDecline> {
let node = executor.graph.node(plan.node_id);
if is_control_flow_op(&node.op_type, &node.domain)
|| is_sequence_op(&node.op_type, &node.domain)
{
return Some(CaptureDecline::node(
plan.node_id,
node,
SeamReason::HostControlFlowOrSequence,
"control-flow and sequence nodes are not device-graph capturable",
));
}
if plan
.outputs
.iter()
.any(|output| !resolved.contains_key(output))
{
return Some(CaptureDecline::node(
plan.node_id,
node,
SeamReason::UnresolvedOutputShape,
"data-dependent output shape was unresolved before capture",
));
}
let Some(input_shapes) = plan
.inputs
.iter()
.map(|input| {
input
.map(|value| resolved.get(&value).cloned())
.unwrap_or(Some(Vec::new()))
})
.collect::<Option<Vec<_>>>()
else {
return Some(CaptureDecline::node(
plan.node_id,
node,
SeamReason::UnresolvedInputShape,
"data-dependent input shape was unresolved before capture",
));
};
let key = KernelKey {
node: plan.node_id.0,
shapes: input_shapes,
};
let Some(kernel) = executor.cache.entries.get(&key) else {
return Some(CaptureDecline::node(
plan.node_id,
node,
SeamReason::KernelNotWarmed,
"kernel has not been warmed for the requested capture shape",
));
};
kernel_capture_decline(plan.node_id, node, kernel.as_ref())
}
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
for index in 0..6 {
let input = graph.create_named_value(
format!("input_{index}"),
DataType::Float32,
static_shape([1]),
);
let output = graph.create_named_value(
format!("output_{index}"),
DataType::Float32,
static_shape([1]),
);
graph.add_input(input);
graph.add_output(output);
graph.insert_node(Node::new(
NodeId(0),
"Identity",
vec![Some(input)],
vec![output],
));
}
let mut executor = Executor::build(
graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().expect("CPU EP"),
)
.expect("representative static graph");
let mut resolved = executor
.value_shapes
.iter()
.filter_map(|(&value, shape)| as_static_shape(shape).map(|shape| (value, shape)))
.collect::<HashMap<_, _>>();
let keys = executor
.plan
.iter()
.map(|plan| KernelKey {
node: plan.node_id.0,
shapes: plan
.inputs
.iter()
.map(|input| {
input
.map(|value| resolved[&value].clone())
.unwrap_or_default()
})
.collect(),
})
.collect::<Vec<_>>();
executor.graph.node_mut(executor.plan[0].node_id).op_type = "If".to_string();
resolved.remove(&executor.plan[0].outputs[0]);
resolved.remove(&executor.plan[0].inputs[0].expect("present input"));
resolved.remove(&executor.plan[1].outputs[0]);
resolved.remove(&executor.plan[1].inputs[0].expect("present input"));
resolved.remove(&executor.plan[2].inputs[0].expect("present input"));
executor.cache.entries.remove(&keys[3]);
executor
.cache
.entries
.insert(keys[4].clone(), Box::new(CaptureDecliningKernel));
let legacy = executor
.plan
.iter()
.map(|plan| legacy_node_capture_reason(&executor, plan, &resolved))
.collect::<Vec<_>>();
let refactored = executor
.plan
.iter()
.map(|plan| executor.node_capture_reason(plan, &resolved))
.collect::<Vec<_>>();
assert_eq!(refactored, legacy);
assert_eq!(
refactored
.iter()
.map(|decline| decline.as_ref().and_then(|decline| decline.seam_reason))
.collect::<Vec<_>>(),
vec![
Some(SeamReason::HostControlFlowOrSequence),
Some(SeamReason::UnresolvedOutputShape),
Some(SeamReason::UnresolvedInputShape),
Some(SeamReason::KernelNotWarmed),
Some(SeamReason::KernelCaptureUnsupported),
None,
]
);
}
#[test]
fn capture_shapes_seed_unresolved_external_values_without_overwriting_resolved_shapes() {
let external_value = |shape| ExternalValue {
dtype: DataType::Float32,
shape,
accepts_subshape: false,
ptr: std::ptr::null_mut(),
len: 0,
alignment: 1,
device: onnx_runtime_ir::DeviceId::cpu(),
};
let mut external = ExternalBindings::default();
external
.inputs
.insert(ValueId(0), external_value(vec![1, 2]));
external
.outputs
.insert(ValueId(1), external_value(vec![1, 4, 128, 64]));
external
.outputs
.insert(ValueId(2), external_value(vec![1, 4, 128, 64]));
let mut resolved = HashMap::from([(ValueId(0), vec![1, 1])]);
external.seed_capture_shapes(&mut resolved);
assert_eq!(resolved[&ValueId(0)], vec![1, 1]);
assert_eq!(resolved[&ValueId(1)], vec![1, 4, 128, 64]);
assert_eq!(resolved[&ValueId(2)], vec![1, 4, 128, 64]);
}
#[test]
fn only_gqa_cache_inputs_use_physical_capacity_as_kernel_geometry() {
let mut gqa = Node::new(NodeId(0), "GroupQueryAttention", vec![], vec![]);
gqa.domain = "com.microsoft".to_string();
let attention = Node::new(NodeId(1), "Attention", vec![], vec![]);
assert!(kernel_input_uses_physical_capacity(&gqa, 3));
assert!(kernel_input_uses_physical_capacity(&gqa, 4));
assert!(!kernel_input_uses_physical_capacity(&gqa, 0));
assert!(!kernel_input_uses_physical_capacity(&attention, 4));
}
#[test]
fn only_capacity_aware_inputs_keep_physical_capacity() {
let shape = Node::new(NodeId(0), "Shape", vec![], vec![]);
let reduce_sum = Node::new(NodeId(1), "ReduceSum", vec![], vec![]);
let cumsum = Node::new(NodeId(2), "CumSum", vec![], vec![]);
let unsqueeze = Node::new(NodeId(3), "Unsqueeze", vec![], vec![]);
assert!(kernel_input_uses_padded_capacity(&shape, 0));
assert!(kernel_input_uses_padded_capacity(&reduce_sum, 0));
assert!(!kernel_input_uses_padded_capacity(&cumsum, 0));
assert!(!kernel_input_uses_padded_capacity(&unsqueeze, 0));
assert!(!kernel_input_uses_padded_capacity(&shape, 1));
let indexer_add = Node::new(NodeId(4), "Add", vec![], vec![]);
let indexer_cast = Node::new(NodeId(5), "Cast", vec![], vec![]);
assert!(!kernel_input_uses_padded_capacity(&indexer_add, 0));
assert!(!kernel_input_uses_padded_capacity(&indexer_cast, 0));
}
struct WeightDeliveryKernel {
deliveries: Arc<std::sync::Mutex<Vec<&'static str>>>,
}
impl WeightDeliveryKernel {
fn copy_bytes(bytes: &[u8], output: &mut TensorMut<'_>) -> onnx_runtime_ep_api::Result<()> {
if bytes.len() != output.byte_size() {
return Err(EpError::KernelFailed(
"test output byte count mismatch".into(),
));
}
unsafe {
std::ptr::copy_nonoverlapping(
bytes.as_ptr(),
output.data.0.cast::<u8>(),
bytes.len(),
);
}
Ok(())
}
}
impl Kernel for WeightDeliveryKernel {
fn execute(
&self,
inputs: &[TensorView],
outputs: &mut [TensorMut],
) -> onnx_runtime_ep_api::Result<()> {
self.deliveries.lock().unwrap().push("resident");
let bytes = unsafe {
std::slice::from_raw_parts(inputs[0].data_ptr::<u8>(), inputs[0].byte_size())
};
Self::copy_bytes(bytes, &mut outputs[0])
}
fn execute_with_inputs(
&self,
inputs: &[KernelInput<'_>],
outputs: &mut [TensorMut],
) -> onnx_runtime_ep_api::Result<()> {
match &inputs[0] {
KernelInput::Tensor(view) => self.execute(std::slice::from_ref(view), outputs),
KernelInput::Weight(handle) => {
self.deliveries.lock().unwrap().push("lazy");
let NegotiatedWeight::Lazy(lazy) =
handle.negotiate(&ExecutionProviderCapabilities::nxrt_weight_paging())?
else {
return Err(EpError::KernelFailed(
"nxrt test EP expected a lazy WeightHandle".into(),
));
};
let resident = lazy.materialize()?;
Self::copy_bytes(resident.bytes(), &mut outputs[0])
}
}
}
}
struct WeightDeliveryEp {
cpu: CpuExecutionProvider,
lazy: bool,
optional_input_contract: bool,
deliveries: Arc<std::sync::Mutex<Vec<&'static str>>>,
device: onnx_runtime_ir::DeviceId,
allocations: Arc<AtomicUsize>,
host_uploads: Arc<AtomicUsize>,
}
impl WeightDeliveryEp {
fn new(lazy: bool, deliveries: Arc<std::sync::Mutex<Vec<&'static str>>>) -> Self {
Self::with_device(
lazy,
deliveries,
onnx_runtime_ir::DeviceId::cpu(),
Arc::new(AtomicUsize::new(0)),
Arc::new(AtomicUsize::new(0)),
)
}
fn non_host(
lazy: bool,
deliveries: Arc<std::sync::Mutex<Vec<&'static str>>>,
allocations: Arc<AtomicUsize>,
host_uploads: Arc<AtomicUsize>,
) -> Self {
Self::with_device(
lazy,
deliveries,
onnx_runtime_ir::DeviceId::new(onnx_runtime_ir::DeviceType::Custom(7), 0),
allocations,
host_uploads,
)
}
fn with_device(
lazy: bool,
deliveries: Arc<std::sync::Mutex<Vec<&'static str>>>,
device: onnx_runtime_ir::DeviceId,
allocations: Arc<AtomicUsize>,
host_uploads: Arc<AtomicUsize>,
) -> Self {
let mut cpu = CpuExecutionProvider::new();
cpu.initialize(&EpConfig::default()).unwrap();
Self {
cpu,
lazy,
optional_input_contract: false,
deliveries,
device,
allocations,
host_uploads,
}
}
fn copy_bytes(
&self,
src: *const u8,
dst: *mut u8,
size: usize,
) -> onnx_runtime_ep_api::Result<()> {
if size != 0 {
unsafe { std::ptr::copy_nonoverlapping(src, dst, size) };
}
Ok(())
}
}
impl ExecutionProvider for WeightDeliveryEp {
fn name(&self) -> &str {
if self.lazy {
"nxrt_test_ep"
} else {
"stock_test_ep"
}
}
fn device_type(&self) -> onnx_runtime_ir::DeviceType {
self.device.device_type
}
fn device_id(&self) -> onnx_runtime_ir::DeviceId {
self.device
}
fn capabilities(&self) -> ExecutionProviderCapabilities {
if self.lazy {
ExecutionProviderCapabilities::nxrt_weight_paging()
} else {
ExecutionProviderCapabilities::stock()
}
}
fn initialize(&mut self, _config: &EpConfig) -> onnx_runtime_ep_api::Result<()> {
Ok(())
}
fn shutdown(&mut self) -> onnx_runtime_ep_api::Result<()> {
Ok(())
}
fn supports_op(
&self,
op: &Node,
opset: u64,
_shapes: &[Shape],
input_dtypes: &[DataType],
_layouts: &[TensorLayout],
) -> KernelMatch {
if self.optional_input_contract && op.op_type == "OptionalContract" {
if input_dtypes == [DataType::Float32, DataType::Undefined, DataType::Bool] {
return KernelMatch::Supported {
cost: Cost::ZERO,
required_input_layouts: None,
output_layouts: vec![TensorLayout::contiguous()],
};
}
return KernelMatch::unsupported(format!(
"OptionalContract requires [Float32, Undefined, Bool] input dtypes, got {input_dtypes:?}"
));
}
if LazyWeightBoundary::BlockQuantizedMoe.matches(&op.domain, &op.op_type)
|| (op.is_default_domain() && op.op_type == "Identity")
{
KernelMatch::Supported {
cost: Cost::ZERO,
required_input_layouts: None,
output_layouts: vec![TensorLayout::contiguous()],
}
} else {
KernelMatch::unsupported(format!(
"no handler for {}::{} at opset {opset} — test EP intentionally declines this op",
canonical_domain(op),
op.op_type
))
}
}
fn get_kernel(
&self,
_op: &Node,
_shapes: &[Vec<usize>],
_opset: u64,
) -> onnx_runtime_ep_api::Result<Box<dyn Kernel>> {
Ok(Box::new(WeightDeliveryKernel {
deliveries: Arc::clone(&self.deliveries),
}))
}
fn allocate(
&self,
size: usize,
alignment: usize,
) -> onnx_runtime_ep_api::Result<DeviceBuffer> {
self.allocations.fetch_add(1, Ordering::Relaxed);
if self.device.is_host_accessible() {
return self.cpu.allocate(size, alignment);
}
let layout = std::alloc::Layout::from_size_align(size.max(1), alignment)
.map_err(|_| EpError::AlignmentError)?;
let ptr = unsafe { std::alloc::alloc(layout) };
if ptr.is_null() {
return Err(EpError::OutOfMemory {
requested: size,
available: 0,
});
}
Ok(unsafe { DeviceBuffer::from_raw_parts(ptr.cast(), self.device, size, alignment) })
}
fn deallocate(&self, buffer: DeviceBuffer) -> onnx_runtime_ep_api::Result<()> {
if self.device.is_host_accessible() {
return self.cpu.deallocate(buffer);
}
let size = buffer.len();
let alignment = buffer.alignment();
let ptr = buffer.into_raw().cast::<u8>();
let layout = std::alloc::Layout::from_size_align(size.max(1), alignment)
.expect("test EP allocated this layout");
unsafe { std::alloc::dealloc(ptr, layout) };
Ok(())
}
fn copy(
&self,
src: &DeviceBuffer,
dst: &mut DeviceBuffer,
size: usize,
) -> onnx_runtime_ep_api::Result<()> {
if size > src.len() || size > dst.len() {
return Err(EpError::KernelFailed("test EP copy out of bounds".into()));
}
self.copy_bytes(src.as_ptr().cast(), dst.as_mut_ptr().cast(), size)
}
fn copy_async(
&self,
src: &DeviceBuffer,
dst: &mut DeviceBuffer,
size: usize,
) -> onnx_runtime_ep_api::Result<Fence> {
self.copy(src, dst, size)?;
Ok(Fence::default())
}
fn sync(&self) -> onnx_runtime_ep_api::Result<()> {
Ok(())
}
fn copy_from_host(
&self,
src: &[u8],
dst: &mut DeviceBuffer,
) -> onnx_runtime_ep_api::Result<()> {
if src.len() > dst.len() {
return Err(EpError::KernelFailed(
"test EP host upload out of bounds".into(),
));
}
self.host_uploads.fetch_add(1, Ordering::Relaxed);
self.copy_bytes(src.as_ptr(), dst.as_mut_ptr().cast(), src.len())
}
fn copy_to_host(
&self,
src: &DeviceBuffer,
dst: &mut [u8],
) -> onnx_runtime_ep_api::Result<()> {
if dst.len() > src.len() {
return Err(EpError::KernelFailed(
"test EP host download out of bounds".into(),
));
}
self.copy_bytes(src.as_ptr().cast(), dst.as_mut_ptr(), dst.len())
}
}
fn weight_delivery_fixture() -> (Graph, Arc<WeightStore>, std::path::PathBuf) {
static NEXT_FILE: AtomicU64 = AtomicU64::new(0);
let root = std::env::var_os("CARGO_TARGET_DIR")
.map(std::path::PathBuf::from)
.unwrap_or_else(|| std::env::current_dir().unwrap().join("target"))
.join("weight-handle-tests");
std::fs::create_dir_all(&root).unwrap();
let id = NEXT_FILE.fetch_add(1, Ordering::Relaxed);
let path = root.join(format!(
"block-quantized-moe-{}-{id}.bin",
std::process::id()
));
std::fs::write(&path, [1u8, 2, 3, 4]).unwrap();
let mut graph = Graph::new();
graph.opset_imports.insert("pkg.nxrt".into(), 1);
let weight = graph.create_named_value("weight", DataType::Uint8, static_shape([4]));
graph.set_initializer(
weight,
WeightRef::External {
path: path.clone(),
offset: 0,
length: 4,
dtype: DataType::Uint8,
dims: vec![4],
},
);
let output = graph.create_named_value("output", DataType::Uint8, static_shape([4]));
let mut node = Node::new(
NodeId(0),
"BlockQuantizedMoE",
vec![Some(weight)],
vec![output],
);
node.domain = "pkg.nxrt".into();
graph.insert_node(node);
graph.add_output(output);
let mut store = WeightStore::new();
store.map_external(&path).unwrap();
(graph, Arc::new(store), path)
}
#[test]
fn claim_time_optional_input_dtype_is_undefined_not_silently_float32() {
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 1);
let data = graph.create_named_value("data", DataType::Float32, static_shape([1]));
let training_mode =
graph.create_named_value("training_mode", DataType::Bool, static_shape([]));
let output = graph.create_named_value("output", DataType::Float32, static_shape([1]));
graph.add_input(data);
graph.add_input(training_mode);
graph.add_output(output);
graph.insert_node(Node::new(
NodeId(0),
"OptionalContract",
vec![Some(data), None, Some(training_mode)],
vec![output],
));
let deliveries = Arc::new(std::sync::Mutex::new(Vec::new()));
let mut ep = WeightDeliveryEp::new(false, deliveries);
ep.optional_input_contract = true;
let executor = Executor::build(graph, Arc::new(WeightStore::new()), Arc::new(ep));
assert!(
executor.is_ok(),
"an omitted optional input must reach supports_op as DataType::Undefined"
);
}
#[test]
fn executor_opens_per_op_span_only_when_tracing_enabled() {
use onnx_runtime_tracer::TraceContext;
{
let (graph, weights, path) = weight_delivery_fixture();
let deliveries = Arc::new(std::sync::Mutex::new(Vec::new()));
let ep: Arc<dyn ExecutionProvider> =
Arc::new(WeightDeliveryEp::new(false, Arc::clone(&deliveries)));
let mut executor = Executor::build(graph, weights, ep).unwrap();
let (trace, events) = TraceContext::in_memory();
trace.set_enabled(false);
executor.set_trace_context(trace);
let _ = executor.run(&[]).unwrap();
drop(executor);
std::fs::remove_file(path).unwrap();
assert!(
events.events().is_empty(),
"a disabled trace context must not open op spans"
);
}
{
let (graph, weights, path) = weight_delivery_fixture();
let deliveries = Arc::new(std::sync::Mutex::new(Vec::new()));
let ep: Arc<dyn ExecutionProvider> =
Arc::new(WeightDeliveryEp::new(false, Arc::clone(&deliveries)));
let mut executor = Executor::build(graph, weights, ep).unwrap();
let (trace, events) = TraceContext::in_memory();
executor.set_trace_context(trace);
let _ = executor.run(&[]).unwrap();
drop(executor);
std::fs::remove_file(path).unwrap();
let spans = events.events();
assert_eq!(spans.len(), 1, "one op span per executed node");
assert_eq!(spans[0].name, "BlockQuantizedMoE");
assert_eq!(spans[0].cat, "op");
}
}
#[test]
fn op_capture_trace_annotates_span_with_status_and_reason() {
use onnx_runtime_tracer::TraceContext;
{
let (trace, events) = TraceContext::in_memory();
{
let _span = trace.span("MatMulNBits", "op");
OpCaptureTrace::Rejected(
"kernel declares CaptureSupport::Unsupported: per-call workspace alloc",
)
.annotate();
}
let recorded = events.events();
assert_eq!(recorded.len(), 1);
let args = recorded[0].args.as_ref().unwrap();
assert_eq!(args[ARG_CAPTURE_STATUS], "rejected");
assert!(
args[ARG_CAPTURE_REASON]
.as_str()
.unwrap()
.contains("CaptureSupport::Unsupported")
);
}
{
let (trace, events) = TraceContext::in_memory();
{
let _span = trace.span("MatMulNBits", "op");
OpCaptureTrace::Captured.annotate();
}
let recorded = events.events();
let args = recorded[0].args.as_ref().unwrap();
assert_eq!(args[ARG_CAPTURE_STATUS], "captured");
}
{
let (trace, events) = TraceContext::in_memory();
{
let _span = trace.span("MatMulNBits", "op");
OpCaptureTrace::Eager.annotate();
}
let recorded = events.events();
assert!(
recorded[0]
.args
.as_ref()
.map(|a| a.get(ARG_CAPTURE_STATUS).is_none())
.unwrap_or(true),
"eager ops carry no capture status"
);
}
}
#[test]
fn executor_selects_lazy_or_resident_weight_delivery_from_ep_capability() {
for (lazy, expected) in [(true, "lazy"), (false, "resident")] {
let (graph, weights, path) = weight_delivery_fixture();
let deliveries = Arc::new(std::sync::Mutex::new(Vec::new()));
let ep: Arc<dyn ExecutionProvider> =
Arc::new(WeightDeliveryEp::new(lazy, Arc::clone(&deliveries)));
let mut executor = Executor::build(graph, weights, ep).unwrap();
let outputs = executor.run(&[]).unwrap();
assert_eq!(outputs[0].as_bytes(), &[1, 2, 3, 4]);
assert_eq!(&*deliveries.lock().unwrap(), &[expected]);
drop(executor);
std::fs::remove_file(path).unwrap();
}
}
#[test]
fn non_host_lazy_only_initializer_skips_eager_device_residency() {
for (lazy, expected_allocations, expected_uploads, expected_delivery) in
[(true, 1, 0, "lazy"), (false, 2, 1, "resident")]
{
let (graph, weights, path) = weight_delivery_fixture();
let deliveries = Arc::new(std::sync::Mutex::new(Vec::new()));
let allocations = Arc::new(AtomicUsize::new(0));
let host_uploads = Arc::new(AtomicUsize::new(0));
let ep: Arc<dyn ExecutionProvider> = Arc::new(WeightDeliveryEp::non_host(
lazy,
Arc::clone(&deliveries),
Arc::clone(&allocations),
Arc::clone(&host_uploads),
));
let mut executor = Executor::build(graph, weights, ep).unwrap();
assert_eq!(
allocations.load(Ordering::Relaxed),
expected_allocations,
"lazy nxrt builds only the output; stock EPs also allocate the initializer"
);
assert_eq!(
host_uploads.load(Ordering::Relaxed),
expected_uploads,
"lazy nxrt must not upload the initializer during build"
);
let outputs = executor.run(&[]).unwrap();
assert_eq!(outputs[0].as_bytes(), &[1, 2, 3, 4]);
assert_eq!(&*deliveries.lock().unwrap(), &[expected_delivery]);
assert_eq!(
host_uploads.load(Ordering::Relaxed),
expected_uploads,
"dispatch must not introduce a second EP upload"
);
drop(executor);
std::fs::remove_file(path).unwrap();
}
}
#[test]
fn initializer_shared_with_resident_consumer_uses_one_device_copy() {
let (mut graph, weights, path) = weight_delivery_fixture();
graph.opset_imports.insert(String::new(), 17);
let weight = graph
.values
.iter()
.find_map(|(vid, value)| (value.name.as_deref() == Some("weight")).then_some(vid))
.unwrap();
let resident_output =
graph.create_named_value("resident_output", DataType::Uint8, static_shape([4]));
graph.insert_node(Node::new(
NodeId(1),
"Identity",
vec![Some(weight)],
vec![resident_output],
));
graph.add_output(resident_output);
let deliveries = Arc::new(std::sync::Mutex::new(Vec::new()));
let allocations = Arc::new(AtomicUsize::new(0));
let host_uploads = Arc::new(AtomicUsize::new(0));
let ep: Arc<dyn ExecutionProvider> = Arc::new(WeightDeliveryEp::non_host(
true,
Arc::clone(&deliveries),
Arc::clone(&allocations),
Arc::clone(&host_uploads),
));
let mut executor = Executor::build(graph, weights, ep).unwrap();
assert!(
!executor.weight_handles.contains_key(&weight),
"a resident consumer makes the single eager device copy authoritative"
);
assert_eq!(allocations.load(Ordering::Relaxed), 3);
assert_eq!(host_uploads.load(Ordering::Relaxed), 1);
let outputs = executor.run(&[]).unwrap();
assert_eq!(outputs[0].as_bytes(), &[1, 2, 3, 4]);
assert_eq!(outputs[1].as_bytes(), &[1, 2, 3, 4]);
assert_eq!(&*deliveries.lock().unwrap(), &["resident", "resident"]);
assert_eq!(
host_uploads.load(Ordering::Relaxed),
1,
"both consumers must share the one resident initializer"
);
drop(executor);
std::fs::remove_file(path).unwrap();
}
#[test]
fn coverage_collector_surfaces_ep_decline_reason() {
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let input = graph.create_named_value("x", DataType::Float32, vec![Dim::Static(1)]);
let output = graph.create_named_value("y", DataType::Float32, vec![Dim::Static(1)]);
graph.insert_node(Node::new(
NodeId(0),
"NotRegistered",
vec![Some(input)],
vec![output],
));
let ep = CpuExecutionProvider::new();
let mut issues = Vec::new();
collect_cuda_coverage_issues(&graph, &graph, &ep, "graph", &mut issues);
assert_eq!(issues.len(), 1);
assert_eq!(issues[0].op_type, "NotRegistered");
assert_eq!(issues[0].domain, "ai.onnx");
assert!(
issues[0]
.reason
.contains("no handler for ai.onnx::NotRegistered at opset 17"),
"{}",
issues[0].reason
);
assert!(
!issues[0].reason.contains("unsupported by"),
"{}",
issues[0].reason
);
}
#[test]
fn cuda_coverage_report_groups_all_distinct_failure_classes_deterministically() {
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let input = graph.create_named_value("x", DataType::Float32, vec![Dim::Static(1)]);
let op_types = [
"RepeatedMissing",
"Missing08",
"RepeatedMissing",
"Missing07",
"Missing06",
"RepeatedMissing",
"Missing05",
"Missing04",
"Missing03",
"Missing02",
"Missing01",
"Missing00",
"RepeatedMissing",
];
for (index, op_type) in op_types.into_iter().enumerate() {
let output = graph.create_named_value(
format!("output_{index}"),
DataType::Float32,
vec![Dim::Static(1)],
);
graph.insert_node(Node::new(
NodeId(index as u32),
op_type,
vec![Some(input)],
vec![output],
));
}
let ep = WeightDeliveryEp::with_device(
false,
Arc::new(std::sync::Mutex::new(Vec::new())),
onnx_runtime_ir::DeviceId::cuda(0),
Arc::new(AtomicUsize::new(0)),
Arc::new(AtomicUsize::new(0)),
);
let report = || {
cuda_fallback_report(&graph, &ep)
.expect("CUDA declines must produce a fallback report")
.to_string()
};
let first = report();
let second = report();
assert_eq!(first, second);
assert!(first.contains("13 nodes assigned to CPU"));
assert!(first.contains("GPU EP stock_test_ep did not claim 13 node(s)"));
assert!(first.contains("the whole session uses cpu_ep"));
assert_eq!(first.matches("ai.onnx::RepeatedMissing:").count(), 1);
assert!(first.contains("ai.onnx::RepeatedMissing: no handler"));
assert!(first.contains("[count=4; examples: graph/node#0, graph/node#12, graph/node#2]"));
assert!(!first.contains("graph/node#5"));
for op_type in [
"Missing00",
"Missing01",
"Missing02",
"Missing03",
"Missing04",
"Missing05",
"Missing06",
"Missing07",
"Missing08",
] {
assert_eq!(
first.matches(&format!("ai.onnx::{op_type}:")).count(),
1,
"{first}"
);
assert!(
first.contains(&format!("ai.onnx::{op_type}: no handler")),
"{first}"
);
}
assert!(!first.contains("more unsupported node"));
}
#[test]
fn cuda_decline_warns_and_falls_back_to_cpu_unless_strict() {
let graph = || {
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let input = graph.create_named_value("input", DataType::Float32, vec![Dim::Static(1)]);
let output =
graph.create_named_value("output", DataType::Float32, vec![Dim::Static(1)]);
graph.add_input(input);
graph.add_output(output);
graph.insert_node(Node::new(
NodeId(0),
"Relu",
vec![Some(input)],
vec![output],
));
graph
};
let cuda_ep = || {
Arc::new(WeightDeliveryEp::with_device(
false,
Arc::new(std::sync::Mutex::new(Vec::new())),
onnx_runtime_ir::DeviceId::cuda(0),
Arc::new(AtomicUsize::new(0)),
Arc::new(AtomicUsize::new(0)),
)) as Arc<dyn ExecutionProvider>
};
let exec = Executor::build_with_cuda_requirement(
graph(),
Arc::new(WeightStore::new()),
cuda_ep(),
false,
)
.expect("default CUDA decline must use the CPU fallback");
assert_eq!(exec.device_id().device_type, DeviceType::Cpu);
let report = exec
.execution_provider_fallback_report()
.expect("fallback must remain observable");
assert_eq!(report.assigned_node_count, 1);
assert_eq!(report.assigned_ops, ["ai.onnx::Relu"]);
assert_eq!(report.declines.len(), 1);
assert_eq!(report.declines[0].op_type, "Relu");
assert!(report.declines[0].reason.contains("intentionally declines"));
let strict = Executor::build_with_cuda_requirement(
graph(),
Arc::new(WeightStore::new()),
cuda_ep(),
true,
)
.err()
.expect("strict CUDA must reject CPU fallback");
assert!(strict.to_string().contains("ONNX_GENAI_REQUIRE_CUDA=1"));
}
#[test]
fn sequence_executor_preserves_element_arc_identity() {
use onnx_runtime_ir::{TensorData, WeightRef, static_shape};
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let input = graph.create_named_value("input", DataType::Float32, static_shape([2]));
graph.set_initializer(
input,
WeightRef::Inline(TensorData::from_raw(
DataType::Float32,
vec![2],
[7.0f32, 8.0]
.into_iter()
.flat_map(f32::to_le_bytes)
.collect(),
)),
);
let zero = graph.create_named_value("zero", DataType::Int64, static_shape([]));
graph.set_initializer(
zero,
WeightRef::Inline(TensorData::from_raw(
DataType::Int64,
vec![],
0i64.to_le_bytes().to_vec(),
)),
);
let one = graph.create_named_value("one", DataType::Int64, static_shape([]));
graph.set_initializer(
one,
WeightRef::Inline(TensorData::from_raw(
DataType::Int64,
vec![],
1i64.to_le_bytes().to_vec(),
)),
);
let first_sequence = graph.create_value(DataType::Float32, static_shape([]));
graph.insert_node(Node::new(
NodeId(0),
"SequenceConstruct",
vec![Some(input)],
vec![first_sequence],
));
let first_at = graph.create_value(DataType::Float32, static_shape([2]));
graph.insert_node(Node::new(
NodeId(0),
"SequenceAt",
vec![Some(first_sequence), Some(zero)],
vec![first_at],
));
let inserted_sequence = graph.create_value(DataType::Float32, static_shape([]));
graph.insert_node(Node::new(
NodeId(0),
"SequenceInsert",
vec![Some(first_sequence), Some(first_at)],
vec![inserted_sequence],
));
let second_at = graph.create_value(DataType::Float32, static_shape([2]));
graph.insert_node(Node::new(
NodeId(0),
"SequenceAt",
vec![Some(inserted_sequence), Some(one)],
vec![second_at],
));
graph.add_output(second_at);
let mut executor = Executor::build(
graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
let output = executor.run(&[]).unwrap();
assert_eq!(output[0].to_vec_f32(), vec![7.0, 8.0]);
let original = &executor.sequences[&first_sequence].elements()[0];
let first_at_arc = &executor.seq_elem_values[&first_at];
let inserted = &executor.sequences[&inserted_sequence].elements()[1];
let second_at_arc = &executor.seq_elem_values[&second_at];
assert!(original.shares_storage_with(first_at_arc));
assert!(original.shares_storage_with(inserted));
assert!(original.shares_storage_with(second_at_arc));
assert_eq!(original.as_ptr(), executor.buffers[&input].as_ptr());
}
#[test]
fn fuses_only_single_consumer_silu_pattern() {
let mut graph = Graph::new();
let shape = vec![Dim::Static(2)];
let x = graph.create_named_value("x", DataType::Float32, shape.clone());
let sigmoid_out = graph.create_named_value("sigmoid", DataType::Float32, shape.clone());
let silu_out = graph.create_named_value("silu", DataType::Float32, shape);
graph.add_input(x);
graph.add_output(silu_out);
graph.insert_node(Node::new(
NodeId(0),
"Sigmoid",
vec![Some(x)],
vec![sigmoid_out],
));
graph.insert_node(Node::new(
NodeId(0),
"Mul",
vec![Some(sigmoid_out), Some(x)],
vec![silu_out],
));
assert_eq!(fuse_silu_patterns(&mut graph), 1);
assert_eq!(graph.num_nodes(), 1);
let fused = graph.nodes.values().next().unwrap();
assert_eq!(fused.op_type, "Silu");
assert_eq!(fused.domain, "com.microsoft");
assert_eq!(fused.inputs, vec![Some(x)]);
assert_eq!(fused.outputs, vec![silu_out]);
assert_eq!(graph.opset_imports["com.microsoft"], 1);
}
#[test]
fn does_not_fuse_silu_when_sigmoid_has_second_consumer() {
let mut graph = Graph::new();
let shape = vec![Dim::Static(2)];
let x = graph.create_named_value("x", DataType::Float32, shape.clone());
let sigmoid_out = graph.create_named_value("sigmoid", DataType::Float32, shape.clone());
let mul_out = graph.create_named_value("mul", DataType::Float32, shape.clone());
let identity_out = graph.create_named_value("identity", DataType::Float32, shape);
graph.add_input(x);
graph.add_output(mul_out);
graph.add_output(identity_out);
graph.insert_node(Node::new(
NodeId(0),
"Sigmoid",
vec![Some(x)],
vec![sigmoid_out],
));
graph.insert_node(Node::new(
NodeId(0),
"Mul",
vec![Some(x), Some(sigmoid_out)],
vec![mul_out],
));
graph.insert_node(Node::new(
NodeId(0),
"Identity",
vec![Some(sigmoid_out)],
vec![identity_out],
));
assert_eq!(fuse_silu_patterns(&mut graph), 0);
assert_eq!(graph.num_nodes(), 3);
assert_eq!(
graph
.nodes
.values()
.filter(|node| node.op_type == "Sigmoid")
.count(),
1
);
assert_eq!(
graph
.nodes
.values()
.filter(|node| node.op_type == "Mul")
.count(),
1
);
assert!(graph.validate().is_ok());
}
#[test]
fn view_bounds_rejects_out_of_bounds_view() {
let shape = [2usize, 3];
let strides = compute_contiguous_strides(&shape);
let err = view_bounds(&shape, &strides, 0, DataType::Float32, 16);
assert!(err.is_err(), "gate must reject an oversized view");
assert!(view_bounds(&shape, &strides, 0, DataType::Float32, 24).is_ok());
}
#[test]
fn view_bounds_rejects_offset_overrun() {
let shape = [4usize];
let strides = compute_contiguous_strides(&shape);
assert!(view_bounds(&shape, &strides, 8, DataType::Float32, 16).is_err());
assert!(view_bounds(&shape, &strides, 0, DataType::Float32, 16).is_ok());
}
#[test]
fn substitute_resolves_bound_symbols_only() {
let mut bindings = HashMap::new();
bindings.insert(SymbolId(0), 7usize);
let shape = vec![Dim::Symbolic(SymbolId(0)), Dim::Static(4)];
assert_eq!(substitute(&shape, &bindings), Some(vec![7, 4]));
let unbound = vec![Dim::Symbolic(SymbolId(1)), Dim::Static(4)];
assert_eq!(substitute(&unbound, &bindings), None);
}
#[test]
fn checked_numel_detects_overflow() {
assert_eq!(checked_numel(&[2, 3, 4], || "v".into()).unwrap(), 24);
assert_eq!(checked_numel(&[], || "v".into()).unwrap(), 1);
let huge = [usize::MAX, 2];
let err = checked_numel(&huge, || "value#9".into());
assert!(matches!(err, Err(SessionError::ShapeOverflow { .. })));
}
#[test]
fn checked_storage_bytes_detects_byte_overflow() {
let numel = usize::MAX / 4;
let err = checked_storage_bytes(DataType::Float64, numel, || "value#9".into(), &[numel]);
assert!(matches!(err, Err(SessionError::ShapeOverflow { .. })));
assert_eq!(
checked_storage_bytes(DataType::Float32, 4, || "v".into(), &[4]).unwrap(),
16
);
}
#[test]
fn dynamic_output_shapes_slice_is_single_output() {
let node = Node::new(NodeId(0), "Slice", vec![], vec![]);
let input_shapes = vec![vec![4usize, 2]];
let input_values = vec![
None, Some(vec![1]), Some(vec![3]), Some(vec![0]), Some(vec![1]), ];
let input_dtypes = vec![
DataType::Float32,
DataType::Int64,
DataType::Int64,
DataType::Int64,
DataType::Int64,
];
let out =
dynamic_output_shapes(&node, &input_shapes, &input_dtypes, &input_values, &[], 17)
.unwrap();
assert_eq!(out.len(), 1, "Slice must resolve exactly one output shape");
assert_eq!(out[0], vec![2, 2]);
let mut custom_slice = Node::new(NodeId(1), "Slice", vec![], vec![ValueId(0)]);
custom_slice.domain = "example.custom".into();
assert!(
dynamic_output_shapes(
&custom_slice,
&input_shapes,
&input_dtypes,
&input_values,
&[],
17
)
.is_none(),
"ONNX Slice semantics must not be applied to an unrelated custom-domain op"
);
let other = Node::new(
NodeId(2),
"NxrtNeverRegisteredSentinelOp",
vec![],
vec![ValueId(0)],
);
assert!(
dynamic_output_shapes(&other, &input_shapes, &input_dtypes, &input_values, &[], 17)
.is_none()
);
}
#[test]
fn dynamic_output_shapes_unsqueeze_supports_input_and_attribute_axes() {
use onnx_runtime_ir::Attribute;
let input_axes = Node::new(
NodeId(0),
"Unsqueeze",
vec![Some(ValueId(0)), Some(ValueId(1))],
vec![ValueId(2)],
);
assert_eq!(
dynamic_output_shapes(
&input_axes,
&[vec![2, 3], vec![2]],
&[DataType::Float32, DataType::Int64],
&[None, Some(vec![0, -1])],
&[],
17,
),
Some(vec![vec![1, 2, 3, 1]])
);
let mut attribute_axes = Node::new(
NodeId(1),
"Unsqueeze",
vec![Some(ValueId(0))],
vec![ValueId(1)],
);
attribute_axes
.attributes
.insert("axes".into(), Attribute::Ints(vec![1, -1]));
assert_eq!(
dynamic_output_shapes(
&attribute_axes,
&[vec![2, 3]],
&[DataType::Float32],
&[None],
&[],
11,
),
Some(vec![vec![2, 1, 3, 1]])
);
}
#[test]
fn dynamic_output_shapes_resize_reads_runtime_scales() {
let node = Node::new(
NodeId(0),
"Resize",
vec![Some(ValueId(0)), Some(ValueId(1)), Some(ValueId(2))],
vec![ValueId(3)],
);
assert_eq!(
dynamic_output_shapes(
&node,
&[vec![1, 128, 13, 13], vec![8], vec![4]],
&[DataType::Float32, DataType::Float32, DataType::Float32],
&[None, None, None],
&[None, None, Some(vec![1.0, 1.0, 2.0, 2.0])],
11,
),
Some(vec![vec![1, 128, 26, 26]])
);
}
#[test]
fn dynamic_output_shapes_non_max_suppression_counts_selected_boxes() {
let node = Node::new(
NodeId(0),
"NonMaxSuppression",
(0..5).map(|index| Some(ValueId(index))).collect(),
vec![ValueId(5)],
);
let shapes = vec![vec![1, 3, 4], vec![1, 1, 3], vec![], vec![], vec![]];
let dtypes = vec![
DataType::Float32,
DataType::Float32,
DataType::Int64,
DataType::Float32,
DataType::Float32,
];
let ints = vec![None, None, Some(vec![2]), None, None];
let floats = vec![
Some(vec![0., 0., 1., 1., 0., 0., 0.9, 0.9, 2., 2., 3., 3.]),
Some(vec![0.9, 0.8, 0.7]),
None,
Some(vec![0.5]),
Some(vec![0.0]),
];
assert_eq!(
dynamic_output_shapes(&node, &shapes, &dtypes, &ints, &floats, 11),
Some(vec![vec![2, 3]])
);
}
#[test]
fn dynamic_output_shapes_gqa_supports_packed_qkv() {
use onnx_runtime_ir::{Attribute, ValueId};
let mut node = Node::new(
NodeId(0),
"GroupQueryAttention",
vec![
Some(ValueId(0)),
None,
None,
Some(ValueId(3)),
Some(ValueId(4)),
Some(ValueId(5)),
Some(ValueId(6)),
],
vec![ValueId(7), ValueId(8), ValueId(9)],
);
node.domain = "com.microsoft".into();
node.attributes
.insert("num_heads".into(), Attribute::Int(14));
node.attributes
.insert("kv_num_heads".into(), Attribute::Int(2));
let input_shapes = vec![
vec![1, 1, 1152],
vec![],
vec![],
vec![1, 2, 16, 64],
vec![1, 2, 16, 64],
vec![1],
vec![],
];
let input_values = vec![None, None, None, None, None, None, Some(vec![17])];
assert_eq!(
dynamic_output_shapes(
&node,
&input_shapes,
&[
DataType::Float32,
DataType::Undefined,
DataType::Undefined,
DataType::Float32,
DataType::Float32,
DataType::Int32,
DataType::Int32,
],
&input_values,
&[],
1,
),
Some(vec![
vec![1, 1, 896],
vec![1, 2, 17, 64],
vec![1, 2, 17, 64],
])
);
}
#[test]
fn effective_opset_reads_graph_import() {
let mut graph = Graph::default();
graph.opset_imports.insert(String::new(), 12);
let node = Node::new(NodeId(0), "Softmax", vec![], vec![]);
assert_eq!(effective_opset(&graph, &node), 12);
graph.opset_imports.insert(String::new(), 0);
assert_eq!(effective_opset(&graph, &node), 0);
}
#[test]
#[should_panic(expected = "internal invariant violated")]
fn effective_opset_requires_validated_import() {
effective_opset(
&Graph::default(),
&Node::new(NodeId(0), "Softmax", vec![], vec![]),
);
}
#[test]
fn child_executor_binds_formals_captures_and_inline_initializers_in_output_order() {
use onnx_runtime_ir::{TensorData, WeightRef, static_shape};
let mut body = Graph::new();
let formal = body.create_named_value("formal", DataType::Float32, static_shape([2]));
body.add_input(formal);
let captured = body.create_named_value("captured", DataType::Float32, static_shape([2]));
let one = body.create_named_value("one", DataType::Float32, static_shape([2]));
body.set_initializer(
one,
WeightRef::Inline(TensorData::from_raw(
DataType::Float32,
vec![2],
[1.0f32, 1.0]
.into_iter()
.flat_map(f32::to_le_bytes)
.collect(),
)),
);
let sum = body.create_named_value("sum", DataType::Float32, static_shape([2]));
body.insert_node(Node::new(
NodeId(0),
"Add",
vec![Some(formal), Some(captured)],
vec![sum],
));
let adjusted = body.create_named_value("adjusted", DataType::Float32, static_shape([2]));
body.insert_node(Node::new(
NodeId(0),
"Add",
vec![Some(sum), Some(one)],
vec![adjusted],
));
body.add_output(adjusted);
body.add_output(sum);
let mut opsets = HashMap::new();
opsets.insert(String::new(), 17);
let mut child = ChildExecutor::new(
"direct-test",
body,
opsets,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
let mut outer_scope = HashMap::new();
outer_scope.insert(
"captured".to_string(),
Tensor::from_f32(&[2], &[10.0, 20.0]).unwrap(),
);
let first = Tensor::from_f32(&[2], &[2.0, 3.0]).unwrap();
let outputs = child.run(&[&first], &outer_scope).unwrap();
assert_eq!(outputs.len(), 2);
assert_eq!(outputs[0].to_vec_f32(), vec![13.0, 24.0]);
assert_eq!(outputs[1].to_vec_f32(), vec![12.0, 23.0]);
assert_eq!(child.stats(), ChildExecutorStats { builds: 1, runs: 1 });
let second = Tensor::from_f32(&[2], &[-1.0, 4.0]).unwrap();
let outputs = child.run(&[&second], &outer_scope).unwrap();
assert_eq!(outputs[0].to_vec_f32(), vec![10.0, 25.0]);
assert_eq!(outputs[1].to_vec_f32(), vec![9.0, 24.0]);
assert_eq!(
child.stats(),
ChildExecutorStats { builds: 1, runs: 2 },
"matching input signatures must reuse the compiled child plan"
);
}
fn unary_child(name: &str) -> ChildExecutor {
let mut body = Graph::new();
let input = body.create_named_value("input", DataType::Float32, Vec::new());
body.add_input(input);
let output = body.create_named_value("output", DataType::Float32, Vec::new());
body.insert_node(Node::new(
NodeId(0),
"Relu",
vec![Some(input)],
vec![output],
));
body.add_output(output);
let mut opsets = HashMap::new();
opsets.insert(String::new(), 17);
ChildExecutor::new(
name,
body,
opsets,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap()
}
#[test]
fn child_executor_reuses_a_signature_after_an_intervening_signature() {
let mut child = unary_child("a-b-a");
let outer_scope = HashMap::new();
let a = Tensor::from_f32(&[1], &[-1.0]).unwrap();
let b = Tensor::from_f32(&[2], &[-2.0, 3.0]).unwrap();
assert_eq!(
child.run(&[&a], &outer_scope).unwrap()[0].to_vec_f32(),
vec![0.0]
);
assert_eq!(
child.run(&[&b], &outer_scope).unwrap()[0].to_vec_f32(),
vec![0.0, 3.0]
);
assert_eq!(
child.run(&[&a], &outer_scope).unwrap()[0].to_vec_f32(),
vec![0.0]
);
assert_eq!(child.stats(), ChildExecutorStats { builds: 2, runs: 3 });
}
#[test]
fn child_executor_lru_evicts_oldest_signature_only() {
let mut child = unary_child("lru-eviction");
let outer_scope = HashMap::new();
let inputs = (1..=CHILD_EXECUTOR_CACHE_CAPACITY + 1)
.map(|len| Tensor::from_f32(&[len], &vec![len as f32; len]).unwrap())
.collect::<Vec<_>>();
for input in &inputs {
child.run(&[input], &outer_scope).unwrap();
}
assert_eq!(
child.stats(),
ChildExecutorStats {
builds: (CHILD_EXECUTOR_CACHE_CAPACITY + 1) as u64,
runs: (CHILD_EXECUTOR_CACHE_CAPACITY + 1) as u64,
}
);
child.run(&[&inputs[0]], &outer_scope).unwrap();
child.run(&[inputs.last().unwrap()], &outer_scope).unwrap();
assert_eq!(
child.stats(),
ChildExecutorStats {
builds: (CHILD_EXECUTOR_CACHE_CAPACITY + 2) as u64,
runs: (CHILD_EXECUTOR_CACHE_CAPACITY + 3) as u64,
},
"the evicted oldest signature must rebuild while a recent entry remains cached"
);
}
fn captured_add_child(name: &str) -> ChildExecutor {
let mut body = Graph::new();
let input = body.create_named_value("input", DataType::Float32, Vec::new());
body.add_input(input);
let captured = body.create_named_value("captured", DataType::Float32, Vec::new());
let output = body.create_named_value("output", DataType::Float32, Vec::new());
body.insert_node(Node::new(
NodeId(0),
"Add",
vec![Some(input), Some(captured)],
vec![output],
));
body.add_output(output);
let mut opsets = HashMap::new();
opsets.insert(String::new(), 17);
ChildExecutor::new(
name,
body,
opsets,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap()
}
#[test]
fn child_executor_cached_plan_rebinds_captures_without_stale_state() {
let mut child = captured_add_child("capture-shadowing");
let a_input = Tensor::from_f32(&[1], &[1.0]).unwrap();
let b_input = Tensor::from_f32(&[2], &[2.0, 3.0]).unwrap();
let mut scope = HashMap::new();
scope.insert(
"captured".to_string(),
Tensor::from_f32(&[1], &[10.0]).unwrap(),
);
assert_eq!(
child.run(&[&a_input], &scope).unwrap()[0].to_vec_f32(),
vec![11.0]
);
scope.insert(
"captured".to_string(),
Tensor::from_f32(&[2], &[20.0, 30.0]).unwrap(),
);
assert_eq!(
child.run(&[&b_input], &scope).unwrap()[0].to_vec_f32(),
vec![22.0, 33.0]
);
scope.insert(
"captured".to_string(),
Tensor::from_f32(&[1], &[40.0]).unwrap(),
);
let cached = child.run(&[&a_input], &scope).unwrap()[0].to_vec_f32();
let mut fresh = captured_add_child("capture-shadowing-fresh");
let freshly_compiled = fresh.run(&[&a_input], &scope).unwrap()[0].to_vec_f32();
assert_eq!(cached, vec![41.0]);
assert_eq!(cached, freshly_compiled);
assert_eq!(child.stats(), ChildExecutorStats { builds: 2, runs: 3 });
}
use onnx_runtime_ir::{WeightRef, static_shape};
use std::path::PathBuf;
fn weightstream_tmp_dir() -> PathBuf {
let dir = PathBuf::from(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../target/weightstream_test"
));
std::fs::create_dir_all(&dir).expect("create weight-streaming test dir");
dir
}
fn f32_le(data: &[f32]) -> Vec<u8> {
data.iter().flat_map(|v| v.to_le_bytes()).collect()
}
#[test]
fn aligned_external_initializer_is_borrowed_zero_copy() {
let align = TensorLayout::contiguous().alignment;
let path = weightstream_tmp_dir().join("aligned_init.bin");
let w_data = [1.0f32, 2.0, 3.0, 4.0];
std::fs::write(&path, f32_le(&w_data)).unwrap();
let mut store = WeightStore::new();
store.map_external(&path).unwrap();
let mut g = Graph::new();
g.opset_imports.insert(String::new(), 17);
let w = g.create_named_value("W", DataType::Float32, static_shape([4]));
g.set_initializer(
w,
WeightRef::External {
path: path.clone(),
offset: 0, length: 16,
dtype: DataType::Float32,
dims: vec![4],
},
);
let y = g.create_value(DataType::Float32, static_shape([4]));
g.insert_node(Node::new(NodeId(0), "Relu", vec![Some(w)], vec![y]));
g.add_output(y);
let ep = auto_detect_cpu_ep().unwrap();
let exec = Executor::build(g, Arc::new(store), ep).unwrap();
let weight = &exec.graph.initializers[&w];
let src = exec.weights().bytes(weight).unwrap();
assert!(
(src.as_ptr() as usize).is_multiple_of(align),
"mmap window must be aligned for this test to exercise the zero-copy path"
);
let buf = &exec.buffers[&w];
assert!(
buf.is_borrowed(),
"aligned initializer must be borrowed, not copied"
);
assert_eq!(
buf.as_ptr() as *const u8,
src.as_ptr(),
"zero-copy: the buffer must alias the mmap bytes (no copy)"
);
let _ = std::fs::remove_file(&path);
}
#[test]
fn device_unaligned_external_initializer_is_borrowed_at_dtype_alignment() {
let align = TensorLayout::contiguous().alignment;
let path = weightstream_tmp_dir().join("unaligned_init.bin");
let offset = 8usize;
let w_data = [5.0f32, 6.0, 7.0, 8.0];
let mut file = vec![0u8; offset];
file.extend_from_slice(&f32_le(&w_data));
std::fs::write(&path, &file).unwrap();
let mut store = WeightStore::new();
store.map_external(&path).unwrap();
let mut g = Graph::new();
g.opset_imports.insert(String::new(), 17);
let w = g.create_named_value("W", DataType::Float32, static_shape([4]));
g.set_initializer(
w,
WeightRef::External {
path: path.clone(),
offset,
length: 16,
dtype: DataType::Float32,
dims: vec![4],
},
);
let x = g.create_named_value("X", DataType::Float32, static_shape([4]));
g.add_input(x);
let y = g.create_value(DataType::Float32, static_shape([4]));
g.insert_node(Node::new(NodeId(0), "Add", vec![Some(x), Some(w)], vec![y]));
g.add_output(y);
let ep = auto_detect_cpu_ep().unwrap();
let mut exec = Executor::build(g, Arc::new(store), ep).unwrap();
let weight = &exec.graph.initializers[&w];
let src = exec.weights().bytes(weight).unwrap();
assert!(
!(src.as_ptr() as usize).is_multiple_of(align),
"window must be unaligned for this test to exercise the fallback"
);
let buf = &exec.buffers[&w];
assert!(
buf.is_borrowed(),
"dtype-aligned mmap initializer must remain borrowed"
);
assert_eq!(
buf.as_ptr() as *const u8,
src.as_ptr(),
"zero-copy buffer must alias the mmap window"
);
assert_eq!(buf.alignment(), std::mem::align_of::<f32>());
let x_tensor = Tensor::from_f32(&[4], &[10.0, 20.0, 30.0, 40.0]).unwrap();
let out = exec.run(&[("X", &x_tensor)]).unwrap();
assert_eq!(out.len(), 1);
let got = out[0].to_vec_f32();
let want = [15.0f32, 26.0, 37.0, 48.0];
assert_eq!(got.len(), want.len());
for (g, w) in got.iter().zip(want.iter()) {
assert!((g - w).abs() < 1e-5, "got {g}, want {w}");
}
let _ = std::fs::remove_file(&path);
}
#[test]
fn unaligned_external_qmoe_keeps_route_first_enabled_and_matches_legacy() {
use std::ffi::OsString;
use std::sync::{Mutex, OnceLock};
static ENV_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
let _env_guard = ENV_LOCK
.get_or_init(|| Mutex::new(()))
.lock()
.expect("weight-offload env lock");
struct RestoreEnv(Option<OsString>);
impl Drop for RestoreEnv {
fn drop(&mut self) {
if let Some(value) = self.0.take() {
unsafe { std::env::set_var(onnx_runtime_ep_cpu::WEIGHT_OFFLOAD_ENV, value) };
} else {
unsafe { std::env::remove_var(onnx_runtime_ep_cpu::WEIGHT_OFFLOAD_ENV) };
}
}
}
let _restore = RestoreEnv(std::env::var_os(onnx_runtime_ep_cpu::WEIGHT_OFFLOAD_ENV));
let fixture = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../onnx-runtime-ep-cpu/tests/fixtures/qmoe_weight_offload/model.onnx");
let input_values: Vec<f32> = (0..64).map(|index| index as f32 * 0.03125 - 1.0).collect();
let router_values = vec![
9.0, 0.0, 0.0, 0.0, 0.0, 9.0, 0.0, 0.0, 0.0, 0.0, 9.0, 0.0, 0.0, 0.0, 0.0, 9.0,
];
let input = Tensor::from_f32(&[4, 16], &input_values).unwrap();
let router = Tensor::from_f32(&[4, 4], &router_values).unwrap();
unsafe { std::env::set_var(onnx_runtime_ep_cpu::WEIGHT_OFFLOAD_ENV, "0") };
let (legacy_graph, legacy_weights) =
onnx_runtime_loader::load_model_with_weights(&fixture).unwrap();
let mut legacy =
Executor::build(legacy_graph, legacy_weights, auto_detect_cpu_ep().unwrap()).unwrap();
let legacy_output = legacy.run(&[("X", &input), ("router", &router)]).unwrap();
unsafe { std::env::set_var(onnx_runtime_ep_cpu::WEIGHT_OFFLOAD_ENV, "1") };
let before = onnx_runtime_ep_cpu::weight_offload_stats();
let (offload_graph, offload_weights) =
onnx_runtime_loader::load_model_with_weights(&fixture).unwrap();
let mut offload = Executor::build(
offload_graph,
offload_weights,
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
for (&value, weight) in &offload.graph.initializers {
let WeightRef::External { .. } = weight else {
continue;
};
let source = offload.weights.bytes(weight).unwrap();
assert!(
!(source.as_ptr() as usize).is_multiple_of(TensorLayout::contiguous().alignment)
);
let buffer = &offload.buffers[&value];
assert!(buffer.is_borrowed());
assert_eq!(buffer.as_ptr() as *const u8, source.as_ptr());
}
let offload_output = offload.run(&[("X", &input), ("router", &router)]).unwrap();
let after = onnx_runtime_ep_cpu::weight_offload_stats();
assert_eq!(
offload_output[0].to_vec_f32(),
legacy_output[0].to_vec_f32()
);
assert!(
after.layer_executions
>= before
.layer_executions
.checked_add(1)
.expect("layer execution counter overflow")
);
assert!(after.bytes_read_from_mmap > before.bytes_read_from_mmap);
}
#[test]
fn producer_backed_initializer_is_not_borrowed() {
let align = TensorLayout::contiguous().alignment;
let path = weightstream_tmp_dir().join("producer_backed_init.bin");
let w_data = [1.0f32, 2.0, 3.0, 4.0];
std::fs::write(&path, f32_le(&w_data)).unwrap();
let mut store = WeightStore::new();
store.map_external(&path).unwrap();
let mut g = Graph::new();
g.opset_imports.insert(String::new(), 17);
let x = g.create_named_value("X", DataType::Float32, static_shape([4]));
g.add_input(x);
let w = g.create_named_value("W", DataType::Float32, static_shape([4]));
g.set_initializer(
w,
WeightRef::External {
path: path.clone(),
offset: 0, length: 16,
dtype: DataType::Float32,
dims: vec![4],
},
);
g.insert_node(Node::new(NodeId(0), "Identity", vec![Some(x)], vec![w]));
let y = g.create_value(DataType::Float32, static_shape([4]));
g.insert_node(Node::new(NodeId(1), "Add", vec![Some(x), Some(w)], vec![y]));
g.add_output(y);
assert!(
g.value(w).producer.is_some(),
"test setup: initializer value must have a producer",
);
let ep = auto_detect_cpu_ep().unwrap();
let exec = Executor::build(g, Arc::new(store), ep).unwrap();
let weight = &exec.graph.initializers[&w];
let src = exec.weights().bytes(weight).unwrap();
assert!(
(src.as_ptr() as usize).is_multiple_of(align),
"mmap window must be aligned so only the producer guard prevents borrowing",
);
let buf = &exec.buffers[&w];
assert!(
!buf.is_borrowed(),
"producer-backed initializer must fall back to an owned writable copy",
);
assert_ne!(
buf.as_ptr() as *const u8,
src.as_ptr(),
"producer-backed initializer must not alias read-only mmap bytes",
);
let _ = std::fs::remove_file(&path);
}
#[test]
fn warm_decode_seeding_admits_previously_unresolved_capture_safe_node() {
use onnx_runtime_ir::{Attribute, static_shape};
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 13);
let start = graph.create_named_value("start", DataType::Int64, static_shape([]));
let limit = graph.create_named_value("limit", DataType::Int64, static_shape([]));
let delta = graph.create_named_value("delta", DataType::Int64, static_shape([]));
graph.add_input(start);
graph.add_input(limit);
graph.add_input(delta);
let len_sym = graph.intern_symbol("range_len");
let r = graph.create_named_value("r", DataType::Int64, vec![len_sym.into()]);
graph.insert_node(Node::new(
NodeId(0),
"Range",
vec![Some(start), Some(limit), Some(delta)],
vec![r],
));
let y = graph.create_named_value("y", DataType::Float32, vec![len_sym.into()]);
let mut cast = Node::new(NodeId(0), "Cast", vec![Some(r)], vec![y]);
cast.attributes
.insert("to".into(), Attribute::Int(DataType::Float32 as i64));
graph.insert_node(cast);
graph.add_output(y);
let mut exec = Executor::build(
graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
exec.set_decode_memo_enabled(false);
let zero = Tensor::from_raw(DataType::Int64, vec![], &0i64.to_le_bytes()).unwrap();
let four = Tensor::from_raw(DataType::Int64, vec![], &4i64.to_le_bytes()).unwrap();
let one = Tensor::from_raw(DataType::Int64, vec![], &1i64.to_le_bytes()).unwrap();
let inputs = [("start", &zero), ("limit", &four), ("delta", &one)];
let cast_pi = exec
.plan
.iter()
.position(|p| exec.graph.node(p.node_id).op_type == "Cast")
.expect("plan contains the Cast node");
let bindings = exec
.bind_symbols(&inputs, &ExternalBindings::default())
.unwrap();
let pre = exec.resolve_soft(&bindings);
assert!(
!pre.contains_key(&r) && !pre.contains_key(&y),
"Range's runtime-length output (and its Cast) must be data-dependent (unresolved)"
);
let pre_seam = exec
.node_capture_reason(&exec.plan[cast_pi], &pre)
.and_then(|decline| decline.seam_reason);
assert!(
matches!(
pre_seam,
Some(SeamReason::UnresolvedInputShape) | Some(SeamReason::UnresolvedOutputShape)
),
"without seeding the Cast must be an unresolved-shape seam; got {pre_seam:?}"
);
let out = exec.run(&inputs).unwrap();
assert_eq!(out[0].to_vec_f32(), vec![0.0, 1.0, 2.0, 3.0]);
let bindings2 = exec
.bind_symbols(&inputs, &ExternalBindings::default())
.unwrap();
let mut post = exec.resolve_soft(&bindings2);
assert!(
!post.contains_key(&r),
"resolve_soft alone still omits the data-dependent value"
);
exec.seed_warm_decode_capture_shapes(&mut post, &ExternalBindings::default());
assert_eq!(
post.get(&r),
Some(&vec![4usize]),
"warm seeding must restore Range's exact eager-resolved output shape"
);
assert_eq!(post.get(&y), Some(&vec![4usize]));
let post_seam = exec
.node_capture_reason(&exec.plan[cast_pi], &post)
.and_then(|decline| decline.seam_reason);
assert!(
!matches!(
post_seam,
Some(SeamReason::UnresolvedInputShape) | Some(SeamReason::UnresolvedOutputShape)
),
"warm-seeded decode shapes must clear the unresolved-shape seam; got {post_seam:?}"
);
let mut mismatched = exec.resolve_soft(&bindings2);
let mut other = ExternalBindings::default();
other.inputs.insert(
start,
ExternalValue {
dtype: DataType::Int64,
shape: vec![],
accepts_subshape: false,
ptr: 0x1000 as *mut std::ffi::c_void,
len: 8,
alignment: 8,
device: onnx_runtime_ir::DeviceId::cpu(),
},
);
exec.seed_warm_decode_capture_shapes(&mut mismatched, &other);
assert!(
!mismatched.contains_key(&r),
"a changed persistent-binding signature must withhold the warm seed"
);
}
#[test]
fn quarantined_op_type_is_forced_to_a_capture_recording_failed_seam() {
use onnx_runtime_ir::{Attribute, static_shape};
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 13);
let x = graph.create_named_value("x", DataType::Int64, static_shape([4]));
graph.add_input(x);
let y = graph.create_named_value("y", DataType::Float32, static_shape([4]));
let mut cast = Node::new(NodeId(0), "Cast", vec![Some(x)], vec![y]);
cast.attributes
.insert("to".into(), Attribute::Int(DataType::Float32 as i64));
graph.insert_node(cast);
graph.add_output(y);
let mut exec = Executor::build(
graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
let cast_pi = exec
.plan
.iter()
.position(|p| exec.graph.node(p.node_id).op_type == "Cast")
.expect("plan contains the Cast node");
let xt = Tensor::from_raw(
DataType::Int64,
vec![4],
&[0i64, 1, 2, 3]
.iter()
.flat_map(|v| v.to_le_bytes())
.collect::<Vec<u8>>(),
)
.unwrap();
let bindings = exec
.bind_symbols(&[("x", &xt)], &ExternalBindings::default())
.unwrap();
let resolved = exec.resolve_soft(&bindings);
let pre_seam = exec
.node_capture_reason(&exec.plan[cast_pi], &resolved)
.and_then(|decline| decline.seam_reason);
assert!(
!matches!(pre_seam, Some(SeamReason::CaptureRecordingFailed)),
"a non-quarantined statically-shaped node must not be a recording-failed seam; \
got {pre_seam:?}"
);
exec.capture_quarantine_ops
.insert(("ai.onnx".to_string(), "Cast".to_string()));
let post = exec.node_capture_reason(&exec.plan[cast_pi], &resolved);
assert_eq!(
post.and_then(|decline| decline.seam_reason),
Some(SeamReason::CaptureRecordingFailed),
"a quarantined op-type must be forced to a CaptureRecordingFailed eager seam"
);
}
#[cfg(test)]
struct DecodeMemoIds {
batch: SymbolId,
seq: SymbolId,
x2: ValueId,
ymul: ValueId,
}
#[cfg(test)]
fn decode_memo_test_graph() -> (Graph, DecodeMemoIds) {
use onnx_runtime_ir::TensorData;
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let batch = graph.intern_symbol("batch");
let seq = graph.intern_symbol("seq");
let x = graph.create_named_value("x", DataType::Float32, vec![batch.into(), seq.into()]);
graph.add_input(x);
let x2 = graph.create_named_value("x2", DataType::Float32, vec![batch.into(), seq.into()]);
graph.insert_node(Node::new(
NodeId(0),
"Add",
vec![Some(x), Some(x)],
vec![x2],
));
graph.add_output(x2);
let y = graph.create_named_value("y", DataType::Float32, static_shape([4]));
graph.add_input(y);
let w = graph.create_named_value("w", DataType::Float32, static_shape([4]));
graph.set_initializer(
w,
WeightRef::Inline(TensorData::from_raw(
DataType::Float32,
vec![4],
[1.0f32, 2.0, 3.0, 4.0]
.into_iter()
.flat_map(f32::to_le_bytes)
.collect(),
)),
);
let ymul = graph.create_named_value("ymul", DataType::Float32, static_shape([4]));
graph.insert_node(Node::new(
NodeId(0),
"Mul",
vec![Some(y), Some(w)],
vec![ymul],
));
graph.add_output(ymul);
(
graph,
DecodeMemoIds {
batch,
seq,
x2,
ymul,
},
)
}
#[cfg(test)]
fn decode_memo_run(exec: &mut Executor, batch: usize, seq: usize) -> Vec<Vec<f32>> {
let x = Tensor::from_f32(
&[batch, seq],
&(0..batch * seq).map(|i| i as f32 + 1.0).collect::<Vec<_>>(),
)
.unwrap();
let y = Tensor::from_f32(&[4], &[10.0, 20.0, 30.0, 40.0]).unwrap();
exec.run(&[("x", &x), ("y", &y)])
.unwrap()
.into_iter()
.map(|t| t.to_vec_f32())
.collect()
}
#[test]
fn decode_memo_env_default_on_unless_explicitly_disabled() {
use std::ffi::OsString;
use std::sync::{Mutex, OnceLock};
static ENV_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
let _env_guard = ENV_LOCK
.get_or_init(|| Mutex::new(()))
.lock()
.expect("decode-memo env lock");
struct RestoreEnv(Option<OsString>);
impl Drop for RestoreEnv {
fn drop(&mut self) {
match self.0.take() {
Some(value) => unsafe { std::env::set_var("ONNX_GENAI_DECODE_MEMO", value) },
None => unsafe { std::env::remove_var("ONNX_GENAI_DECODE_MEMO") },
}
}
}
let _restore = RestoreEnv(std::env::var_os("ONNX_GENAI_DECODE_MEMO"));
unsafe { std::env::remove_var("ONNX_GENAI_DECODE_MEMO") };
assert!(decode_memo_env_enabled(), "unset must default ON");
for off in ["0", "false", "off", "FALSE", "Off", " 0 ", "\tOFF\n"] {
unsafe { std::env::set_var("ONNX_GENAI_DECODE_MEMO", off) };
assert!(!decode_memo_env_enabled(), "{off:?} must disable the memo");
}
for on in [
"1", "true", "on", "ON", "True", " on ", "", " ", "yes", "2", "banana",
] {
unsafe { std::env::set_var("ONNX_GENAI_DECODE_MEMO", on) };
assert!(decode_memo_env_enabled(), "{on:?} must keep the memo ON");
}
}
#[test]
fn decode_plan_memo_rebuilds_and_replays() {
let (graph, ids) = decode_memo_test_graph();
let mut exec = Executor::build(
graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
exec.set_decode_memo_enabled(true);
decode_memo_run(&mut exec, 1, 4);
assert_eq!(exec.decode_memo_action(), DecodeMemoAction::Primed);
assert!(exec.decode_memo.is_none());
decode_memo_run(&mut exec, 1, 5);
assert_eq!(exec.decode_memo_action(), DecodeMemoAction::Rebuilt);
let memo = exec.decode_memo.as_ref().expect("memo built");
assert!(memo.decode_varying.contains(&ids.seq));
assert!(!memo.decode_varying.contains(&ids.batch));
assert!(memo.invariant_shapes.contains_key(&ids.ymul));
assert!(memo.variant_values.contains(&ids.x2));
let out6 = decode_memo_run(&mut exec, 1, 6);
assert_eq!(exec.decode_memo_action(), DecodeMemoAction::Replayed);
let out7 = decode_memo_run(&mut exec, 1, 7);
assert_eq!(exec.decode_memo_action(), DecodeMemoAction::Replayed);
assert_eq!(out6[0].len(), 6);
assert_eq!(out7[0].len(), 7);
decode_memo_run(&mut exec, 2, 7);
assert_eq!(exec.decode_memo_action(), DecodeMemoAction::Rebuilt);
let memo = exec.decode_memo.as_ref().expect("memo rebuilt");
assert_eq!(memo.reference_bindings.get(&ids.batch), Some(&2));
}
#[test]
fn decode_plan_memo_is_token_exact_over_128_steps() {
const STEPS: usize = 130;
let (off_graph, _) = decode_memo_test_graph();
let mut off = Executor::build(
off_graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
off.set_decode_memo_enabled(false);
assert!(!off.decode_memo_enabled);
let (on_graph, _) = decode_memo_test_graph();
let mut on = Executor::build(
on_graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
on.set_decode_memo_enabled(true);
let mut replays = 0usize;
for step in 0..STEPS {
let seq = 3 + step; let ref_out = decode_memo_run(&mut off, 1, seq);
let memo_out = decode_memo_run(&mut on, 1, seq);
assert_eq!(
ref_out, memo_out,
"decode-plan memo diverged from the reference at step {step} (seq={seq})"
);
if on.decode_memo_action() == DecodeMemoAction::Replayed {
replays += 1;
}
}
assert!(
replays >= STEPS - 2,
"expected the memo to replay in steady state; only {replays}/{STEPS} replays"
);
}
#[test]
fn decode_plan_memo_fires_on_persistent_kv_bindings() {
use onnx_runtime_ir::{TensorData, static_shape};
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let l = graph.intern_symbol("L");
let kv = graph.create_named_value("kv", DataType::Float32, vec![l.into()]);
graph.add_input(kv);
let kvout = graph.create_named_value("kvout", DataType::Float32, vec![l.into()]);
graph.insert_node(Node::new(NodeId(0), "Relu", vec![Some(kv)], vec![kvout]));
graph.add_output(kvout);
let y = graph.create_named_value("y", DataType::Float32, static_shape([4]));
graph.add_input(y);
let w = graph.create_named_value("w", DataType::Float32, static_shape([4]));
graph.set_initializer(
w,
WeightRef::Inline(TensorData::from_raw(
DataType::Float32,
vec![4],
[1.0f32, 2.0, 3.0, 4.0]
.into_iter()
.flat_map(f32::to_le_bytes)
.collect(),
)),
);
let ymul = graph.create_named_value("ymul", DataType::Float32, static_shape([4]));
graph.insert_node(Node::new(
NodeId(0),
"Mul",
vec![Some(y), Some(w)],
vec![ymul],
));
graph.add_output(ymul);
let mut exec = Executor::build(
graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
exec.set_decode_memo_enabled(true);
const CAP: usize = 128;
let mut kv_binding = exec
.allocate_device_binding(
"kv".into(),
Some("kvout".into()),
DataType::Float32,
vec![CAP],
vec![1],
)
.unwrap();
let ptr0 = kv_binding.device_ptr();
let y_tensor = Tensor::from_f32(&[4], &[10.0, 20.0, 30.0, 40.0]).unwrap();
let mut replays = 0usize;
for step in 0..8usize {
let len = 4 + step; kv_binding.set_logical_shape(vec![len]).unwrap();
let bytes: Vec<u8> = (0..len).flat_map(|i| (i as f32).to_le_bytes()).collect();
kv_binding.write_bytes(0, &bytes).unwrap();
exec.run_with_device_bindings(
&[("y", &y_tensor)],
std::slice::from_mut(&mut kv_binding),
)
.unwrap();
if exec.decode_memo_action() == DecodeMemoAction::Replayed {
replays += 1;
}
}
assert_eq!(kv_binding.device_ptr(), ptr0);
let (primed, rebuilt, replayed, ineligible) = exec.decode_memo_counts();
assert_eq!(
ineligible, 0,
"persistent-KV decode must be memo-eligible, not excluded (the F5 regression)"
);
assert!(primed >= 1, "the first decode step must prime the memo");
assert!(
replayed >= 1,
"steady persistent-KV decode must replay the memo \
(primed={primed} rebuilt={rebuilt} replayed={replayed})"
);
assert_eq!(replays as u64, replayed);
}
#[cfg(test)]
fn stage2_view_graph() -> (Graph, ValueId) {
use onnx_runtime_ir::{TensorData, static_shape};
let mut graph = Graph::new();
graph.opset_imports.insert(String::new(), 17);
let l = graph.intern_symbol("L");
let kv = graph.create_named_value("kv", DataType::Float32, vec![l.into()]);
graph.add_input(kv);
let kvout = graph.create_named_value("kvout", DataType::Float32, vec![l.into()]);
graph.insert_node(Node::new(NodeId(0), "Relu", vec![Some(kv)], vec![kvout]));
graph.add_output(kvout);
let y = graph.create_named_value("y", DataType::Float32, static_shape([4]));
graph.add_input(y);
let w = graph.create_named_value("w", DataType::Float32, static_shape([4]));
graph.set_initializer(
w,
WeightRef::Inline(TensorData::from_raw(
DataType::Float32,
vec![4],
[1.0f32, 2.0, 3.0, 4.0]
.into_iter()
.flat_map(f32::to_le_bytes)
.collect(),
)),
);
let ymul = graph.create_named_value("ymul", DataType::Float32, static_shape([4]));
graph.insert_node(Node::new(
NodeId(0),
"Mul",
vec![Some(y), Some(w)],
vec![ymul],
));
let yshape = graph.create_named_value("yshape", DataType::Int64, static_shape([2]));
graph.set_initializer(
yshape,
WeightRef::Inline(TensorData::from_raw(
DataType::Int64,
vec![2],
[2i64, 2].into_iter().flat_map(i64::to_le_bytes).collect(),
)),
);
let yview = graph.create_named_value("yview", DataType::Float32, static_shape([2, 2]));
graph.insert_node(Node::new(
NodeId(0),
"Reshape",
vec![Some(ymul), Some(yshape)],
vec![yview],
));
graph.add_output(yview);
(graph, ymul)
}
#[cfg(test)]
fn stage2_run(
exec: &mut Executor,
kv_binding: &mut DeviceIoBinding,
len: usize,
y_bias: f32,
) -> Vec<f32> {
kv_binding.set_logical_shape(vec![len]).unwrap();
let bytes: Vec<u8> = (0..len).flat_map(|i| (i as f32).to_le_bytes()).collect();
kv_binding.write_bytes(0, &bytes).unwrap();
let y = Tensor::from_f32(
&[4],
&[y_bias + 1.0, y_bias + 2.0, y_bias + 3.0, y_bias + 4.0],
)
.unwrap();
let outs = exec
.run_with_device_bindings(&[("y", &y)], std::slice::from_mut(kv_binding))
.unwrap();
outs.into_iter()
.flatten()
.next()
.expect("yview output")
.to_vec_f32()
}
#[cfg(test)]
fn stage2_kv_binding(exec: &Executor) -> DeviceIoBinding {
exec.allocate_device_binding(
"kv".into(),
Some("kvout".into()),
DataType::Float32,
vec![256],
vec![1],
)
.unwrap()
}
#[test]
fn decode_view_plan_fires_and_is_token_exact_over_128_steps() {
const STEPS: usize = 130;
let (off_graph, _) = stage2_view_graph();
let mut off = Executor::build(
off_graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
off.set_decode_memo_enabled(false);
assert!(!off.decode_memo_enabled, "reference must run memo-OFF");
let mut off_kv = stage2_kv_binding(&off);
let (on_graph, _) = stage2_view_graph();
let mut on = Executor::build(
on_graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
on.set_decode_memo_enabled(true);
let mut on_kv = stage2_kv_binding(&on);
for step in 0..STEPS {
let len = 4 + step; let bias = step as f32; let ref_out = stage2_run(&mut off, &mut off_kv, len, bias);
let memo_out = stage2_run(&mut on, &mut on_kv, len, bias);
assert_eq!(
ref_out, memo_out,
"Stage 2 view reuse diverged from the reference at step {step} (L={len})"
);
}
let (views_reused, dispatch_elided) = on.decode_view_plan_counts();
assert!(
views_reused > 0 && dispatch_elided > 0,
"Stage 2 must fire on steady decode (views_reused={views_reused}, \
dispatch_elided={dispatch_elided})"
);
assert!(
views_reused as usize >= STEPS - 4,
"expected steady Stage 2 reuse; only {views_reused}/{STEPS} views reused"
);
assert!(
on.decode_view_plan.is_some(),
"the cached view plan must survive steady-state replay"
);
}
#[test]
fn decode_view_plan_rebuilds_on_source_buffer_move() {
use onnx_runtime_ir::TensorLayout;
let (off_graph, _) = stage2_view_graph();
let mut off = Executor::build(
off_graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
let mut off_kv = stage2_kv_binding(&off);
let (on_graph, ymul) = stage2_view_graph();
let mut on = Executor::build(
on_graph,
Arc::new(WeightStore::new()),
auto_detect_cpu_ep().unwrap(),
)
.unwrap();
on.set_decode_memo_enabled(true);
let mut on_kv = stage2_kv_binding(&on);
for step in 0..6usize {
let len = 4 + step;
let bias = step as f32;
let r = stage2_run(&mut off, &mut off_kv, len, bias);
let m = stage2_run(&mut on, &mut on_kv, len, bias);
assert_eq!(r, m, "warmup diverged at step {step}");
}
assert!(
on.decode_view_plan.is_some(),
"view plan must be built before the realloc test"
);
let (reused_before, _) = on.decode_view_plan_counts();
let old = on.buffers.remove(&ymul).expect("ymul buffer");
let cap = old.len();
let fresh = on
.ep
.allocate(cap, TensorLayout::contiguous().alignment)
.unwrap();
let moved_ptr = fresh.as_ptr() as usize;
assert_ne!(
moved_ptr,
old.as_ptr() as usize,
"the replacement buffer must not reuse the original address, or the \
signature check below is not being exercised"
);
on.ep.deallocate(old).unwrap();
on.buffers.insert(ymul, fresh);
assert!(
!on.stage2_buffer_sig_matches(on.decode_view_plan.as_ref().unwrap()),
"the forced realloc must break the buffer-identity signature"
);
let len = 4 + 6;
let bias = 6.0f32;
let ref_out = stage2_run(&mut off, &mut off_kv, len, bias);
let memo_out = stage2_run(&mut on, &mut on_kv, len, bias);
assert_eq!(
ref_out, memo_out,
"a moved source buffer must force a rebuild, never serve a stale view"
);
let (reused_after, _) = on.decode_view_plan_counts();
assert_eq!(
reused_after, reused_before,
"the mismatched step must NOT reuse cached views (would be stale)"
);
let healed = on.buffers.get(&ymul).expect("ymul rebound").as_ptr() as usize;
assert!(healed == moved_ptr || healed != 0, "ymul must be backed");
}
}