use std::sync::{Arc, OnceLock};
use onnx_runtime_ep_api::{DeviceBuffer, ExecutionProvider};
use onnx_runtime_ep_cpu::CpuExecutionProvider;
use onnx_runtime_ir::{DataType, DeviceId, TensorLayout};
use crate::error::{Result, SessionError};
pub(crate) fn shared_cpu_ep() -> Arc<CpuExecutionProvider> {
static EP: OnceLock<Arc<CpuExecutionProvider>> = OnceLock::new();
EP.get_or_init(|| {
let mut ep = CpuExecutionProvider::new();
let _ = ep.initialize(&Default::default());
Arc::new(ep)
})
.clone()
}
pub fn cpu_allocator() -> Arc<dyn ExecutionProvider> {
shared_cpu_ep()
}
pub(crate) struct SharedTensorBuffer {
buffer: Option<DeviceBuffer>,
allocator: Arc<dyn ExecutionProvider>,
import_guard: Option<Box<dyn core::any::Any + Send + Sync>>,
}
impl SharedTensorBuffer {
pub(crate) fn new(allocator: Arc<dyn ExecutionProvider>, buffer: DeviceBuffer) -> Arc<Self> {
Arc::new(Self {
buffer: Some(buffer),
allocator,
import_guard: None,
})
}
fn with_guard(
allocator: Arc<dyn ExecutionProvider>,
buffer: DeviceBuffer,
import_guard: Option<Box<dyn core::any::Any + Send + Sync>>,
) -> Arc<Self> {
Arc::new(Self {
buffer: Some(buffer),
allocator,
import_guard,
})
}
pub(crate) fn allocate_cpu(bytes: usize) -> Result<Arc<Self>> {
let allocator: Arc<dyn ExecutionProvider> = shared_cpu_ep();
let buffer = allocator.allocate(bytes.max(1), TensorLayout::contiguous().alignment)?;
Ok(Self::new(allocator, buffer))
}
pub(crate) fn buffer(&self) -> &DeviceBuffer {
self.buffer
.as_ref()
.expect("SharedTensorBuffer buffer taken only in Drop")
}
pub(crate) fn buffer_mut(&mut self) -> &mut DeviceBuffer {
self.buffer
.as_mut()
.expect("SharedTensorBuffer buffer taken only in Drop")
}
pub(crate) fn allocator(&self) -> &Arc<dyn ExecutionProvider> {
&self.allocator
}
pub(crate) fn alias(&self) -> DeviceBuffer {
let buffer = self.buffer();
unsafe {
DeviceBuffer::from_borrowed_parts(
buffer.as_ptr() as *mut std::ffi::c_void,
buffer.device(),
buffer.len(),
buffer.alignment(),
)
}
}
pub(crate) fn into_buffer(mut self) -> DeviceBuffer {
debug_assert!(
self.import_guard.is_none(),
"executor-promoted buffers never carry a foreign import guard"
);
self.buffer
.take()
.expect("SharedTensorBuffer buffer taken only by into_buffer or Drop")
}
}
impl std::fmt::Debug for SharedTensorBuffer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SharedTensorBuffer")
.field("device", &self.buffer().device())
.field("len", &self.buffer().len())
.field("ptr", &self.buffer().as_ptr())
.finish()
}
}
impl Drop for SharedTensorBuffer {
fn drop(&mut self) {
if let Some(buffer) = self.buffer.take() {
let _ = self.allocator.deallocate(buffer);
}
let _ = self.import_guard.take();
}
}
pub(crate) fn host_bytes(buffer: &DeviceBuffer) -> &[u8] {
assert!(
buffer.device().is_host_accessible(),
"host_bytes on non-host device {:?}",
buffer.device()
);
if buffer.is_empty() {
return &[];
}
unsafe { std::slice::from_raw_parts(buffer.as_ptr() as *const u8, buffer.len()) }
}
pub struct Tensor {
pub dtype: DataType,
pub shape: Vec<usize>,
pub layout: TensorLayout,
device: DeviceId,
buffer: Option<DeviceBuffer>,
allocator: Arc<dyn ExecutionProvider>,
import_guard: Option<Box<dyn core::any::Any + Send + Sync>>,
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct DeviceBindingTransferStats {
pub host_upload_calls: u64,
pub host_upload_bytes: u64,
pub host_download_calls: u64,
pub host_download_bytes: u64,
}
pub struct DeviceIoBinding {
input_name: String,
bind_input: bool,
output_name: Option<String>,
pub dtype: DataType,
physical_shape: Vec<usize>,
logical_shape: Vec<usize>,
expose_logical_input_shape: bool,
buffer: Option<DeviceBuffer>,
allocator: Arc<dyn ExecutionProvider>,
transfer_stats: DeviceBindingTransferStats,
}
impl DeviceIoBinding {
pub(crate) fn allocate(
allocator: Arc<dyn ExecutionProvider>,
input_name: String,
bind_input: bool,
output_name: Option<String>,
dtype: DataType,
physical_shape: Vec<usize>,
logical_shape: Vec<usize>,
expose_logical_input_shape: bool,
) -> Result<Self> {
validate_logical_shape(&physical_shape, &logical_shape)?;
let numel = physical_shape.iter().try_fold(1usize, |product, &dim| {
product.checked_mul(dim).ok_or_else(|| {
SessionError::Internal(format!(
"device binding '{input_name}' physical shape overflows: {physical_shape:?}"
))
})
})?;
let bytes = dtype.storage_bytes(numel).max(1);
let allocator_for_buffer = allocator.clone();
let buffer = allocator_for_buffer.allocate(bytes, TensorLayout::contiguous().alignment)?;
Ok(Self {
input_name,
bind_input,
output_name,
dtype,
physical_shape,
logical_shape,
expose_logical_input_shape,
buffer: Some(buffer),
allocator,
transfer_stats: DeviceBindingTransferStats::default(),
})
}
pub fn input_name(&self) -> &str {
&self.input_name
}
pub(crate) fn binds_input(&self) -> bool {
self.bind_input
}
pub fn output_name(&self) -> Option<&str> {
self.output_name.as_deref()
}
pub fn physical_shape(&self) -> &[usize] {
&self.physical_shape
}
pub fn logical_shape(&self) -> &[usize] {
&self.logical_shape
}
pub(crate) fn kernel_input_shape(&self) -> &[usize] {
if self.expose_logical_input_shape {
&self.logical_shape
} else {
&self.physical_shape
}
}
pub fn has_dynamic_logical_input_shape(&self) -> bool {
self.bind_input
&& self.expose_logical_input_shape
&& self.logical_shape != self.physical_shape
}
pub fn exposes_logical_input_shape(&self) -> bool {
self.bind_input && self.expose_logical_input_shape
}
pub fn set_logical_shape(&mut self, shape: Vec<usize>) -> Result<()> {
validate_logical_shape(&self.physical_shape, &shape)?;
self.logical_shape = shape;
Ok(())
}
pub fn device_ptr(&self) -> *const std::ffi::c_void {
self.buffer().as_ptr()
}
pub fn transfer_stats(&self) -> DeviceBindingTransferStats {
self.transfer_stats
}
pub fn write_bytes(&mut self, byte_offset: usize, bytes: &[u8]) -> Result<()> {
let buffer = self
.buffer
.as_mut()
.expect("DeviceIoBinding buffer taken only in Drop");
self.allocator
.copy_from_host_at(bytes, buffer, byte_offset)?;
self.transfer_stats.host_upload_calls += 1;
self.transfer_stats.host_upload_bytes += bytes.len() as u64;
Ok(())
}
pub fn read_bytes(&mut self) -> Result<Vec<u8>> {
let mut bytes = vec![0; self.buffer().len()];
self.read_bytes_into(&mut bytes)?;
Ok(bytes)
}
pub fn read_bytes_into(&mut self, bytes: &mut [u8]) -> Result<()> {
self.allocator.copy_to_host(self.buffer(), bytes)?;
self.transfer_stats.host_download_calls += 1;
self.transfer_stats.host_download_bytes += bytes.len() as u64;
Ok(())
}
pub fn device_argmax_supported(&self) -> bool {
self.allocator.device_argmax_supported()
}
pub fn device_argmax(&self, elements: usize, result: &mut DeviceIoBinding) -> Result<()> {
if !matches!(self.dtype, DataType::Float32 | DataType::Float16)
|| result.dtype != DataType::Uint32
{
return Err(SessionError::Internal(format!(
"device argmax requires f32/f16 logits and u32 result, got {:?} and {:?}",
self.dtype, result.dtype
)));
}
if !Arc::ptr_eq(&self.allocator, &result.allocator) {
return Err(SessionError::Internal(
"device argmax bindings must belong to the same execution provider".into(),
));
}
Ok(self.allocator.device_argmax(
self.buffer(),
elements,
self.dtype,
result.buffer_mut(),
)?)
}
pub(crate) fn buffer(&self) -> &DeviceBuffer {
self.buffer
.as_ref()
.expect("DeviceIoBinding buffer taken only in Drop")
}
pub(crate) fn buffer_mut(&mut self) -> &mut DeviceBuffer {
self.buffer
.as_mut()
.expect("DeviceIoBinding buffer taken only in Drop")
}
}
fn validate_logical_shape(physical: &[usize], logical: &[usize]) -> Result<()> {
if physical.len() != logical.len()
|| physical
.iter()
.zip(logical)
.any(|(&capacity, &valid)| valid > capacity)
{
return Err(SessionError::Internal(format!(
"device binding logical shape {logical:?} exceeds physical capacity {physical:?}"
)));
}
Ok(())
}
impl std::fmt::Debug for DeviceIoBinding {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DeviceIoBinding")
.field("input_name", &self.input_name)
.field("bind_input", &self.bind_input)
.field("output_name", &self.output_name)
.field("dtype", &self.dtype)
.field("physical_shape", &self.physical_shape)
.field("logical_shape", &self.logical_shape)
.field(
"expose_logical_input_shape",
&self.expose_logical_input_shape,
)
.field("device", &self.buffer().device())
.field("device_ptr", &self.device_ptr())
.field("transfer_stats", &self.transfer_stats)
.finish()
}
}
impl Drop for DeviceIoBinding {
fn drop(&mut self) {
if let Some(buffer) = self.buffer.take() {
let _ = self.allocator.reset_device_graph();
let _ = self.allocator.deallocate(buffer);
}
}
}
impl Tensor {
pub(crate) fn allocate_cpu(dtype: DataType, shape: Vec<usize>) -> Result<Self> {
let numel = shape.iter().try_fold(1usize, |product, &dim| {
product.checked_mul(dim).ok_or_else(|| {
SessionError::Internal(format!(
"Tensor::allocate_cpu: element count overflows for shape {shape:?}"
))
})
})?;
let bytes = dtype.checked_storage_bytes(numel).ok_or_else(|| {
SessionError::Internal(format!(
"Tensor::allocate_cpu: byte count overflows for shape {shape:?} dtype {dtype:?}"
))
})?;
let allocator: Arc<dyn ExecutionProvider> = shared_cpu_ep();
let layout = TensorLayout::contiguous();
let buffer = allocator.allocate(bytes.max(1), layout.alignment)?;
Ok(Self {
dtype,
shape,
layout,
device: buffer.device(),
buffer: Some(buffer),
allocator,
import_guard: None,
})
}
pub(crate) fn copy_from_host_at(&mut self, offset: usize, bytes: &[u8]) -> Result<()> {
let buffer = self.buffer.as_mut().ok_or_else(|| {
SessionError::Internal("Tensor buffer is unavailable for writing".to_string())
})?;
self.allocator.copy_from_host_at(bytes, buffer, offset)?;
Ok(())
}
pub(crate) fn from_raw_in(
allocator: Arc<dyn ExecutionProvider>,
dtype: DataType,
shape: Vec<usize>,
bytes: &[u8],
) -> Result<Self> {
let numel: usize = shape.iter().product();
let expected = dtype.storage_bytes(numel);
if bytes.len() != expected {
return Err(SessionError::Internal(format!(
"Tensor::from_raw_in: {} bytes for shape {shape:?} dtype {dtype:?}, expected {expected}",
bytes.len()
)));
}
let layout = TensorLayout::contiguous();
let align = layout.alignment;
let mut buffer = allocator.allocate(expected.max(1), align)?;
allocator.copy_from_host(bytes, &mut buffer)?;
Ok(Self {
dtype,
shape,
layout,
device: buffer.device(),
buffer: Some(buffer),
allocator,
import_guard: None,
})
}
pub fn from_raw(dtype: DataType, shape: Vec<usize>, bytes: &[u8]) -> Result<Self> {
Self::from_raw_in(shared_cpu_ep(), dtype, shape, bytes)
}
pub(crate) fn from_owned_buffer(
allocator: Arc<dyn ExecutionProvider>,
dtype: DataType,
shape: Vec<usize>,
buffer: DeviceBuffer,
) -> Self {
debug_assert!(
!buffer.is_borrowed(),
"from_owned_buffer requires an owned buffer"
);
debug_assert_eq!(
buffer.len(),
dtype.storage_bytes(shape.iter().product::<usize>()).max(1),
"from_owned_buffer size mismatch for shape {shape:?} dtype {dtype:?}",
);
let device = buffer.device();
Self {
dtype,
shape,
layout: TensorLayout::contiguous(),
device,
buffer: Some(buffer),
allocator,
import_guard: None,
}
}
pub fn from_f32(shape: &[usize], data: &[f32]) -> Result<Self> {
let mut bytes = Vec::with_capacity(data.len() * 4);
for v in data {
bytes.extend_from_slice(&v.to_le_bytes());
}
Self::from_raw(DataType::Float32, shape.to_vec(), &bytes)
}
pub fn from_i64(shape: &[usize], data: &[i64]) -> Result<Self> {
let mut bytes = Vec::with_capacity(data.len() * 8);
for v in data {
bytes.extend_from_slice(&v.to_le_bytes());
}
Self::from_raw(DataType::Int64, shape.to_vec(), &bytes)
}
pub(crate) fn into_shared_parts(
mut self,
) -> (Arc<SharedTensorBuffer>, DataType, Vec<usize>, TensorLayout) {
let buffer = self
.buffer
.take()
.expect("Tensor buffer taken only by into_shared_parts or Drop");
let storage = SharedTensorBuffer::with_guard(
Arc::clone(&self.allocator),
buffer,
self.import_guard.take(),
);
let dtype = self.dtype;
let shape = std::mem::take(&mut self.shape);
let layout = std::mem::take(&mut self.layout);
(storage, dtype, shape, layout)
}
pub fn device(&self) -> DeviceId {
self.device
}
pub fn from_borrowed_parts_with_guard(
allocator: Arc<dyn ExecutionProvider>,
dtype: DataType,
shape: Vec<usize>,
layout: TensorLayout,
buffer: DeviceBuffer,
guard: Box<dyn core::any::Any + Send + Sync>,
) -> Self {
debug_assert!(
buffer.is_borrowed(),
"from_borrowed_parts_with_guard requires a borrowed DeviceBuffer; \
an owned buffer would be freed twice (EP deallocate + guard)"
);
Self {
dtype,
shape,
layout,
device: buffer.device(),
buffer: Some(buffer),
allocator,
import_guard: Some(guard),
}
}
pub fn numel(&self) -> usize {
self.shape.iter().product()
}
pub fn device_ptr(&self) -> *const std::ffi::c_void {
if self.numel() == 0 {
std::ptr::null()
} else {
self.buffer().as_ptr()
}
}
pub fn sync(&self) -> Result<()> {
self.allocator.sync()?;
Ok(())
}
fn buffer(&self) -> &DeviceBuffer {
self.buffer
.as_ref()
.expect("Tensor buffer taken only in Drop")
}
pub fn as_bytes(&self) -> &[u8] {
let n = self.dtype.storage_bytes(self.numel());
&host_bytes(self.buffer())[..n]
}
pub(crate) fn overwrite_bytes(&mut self, bytes: &[u8]) -> Result<()> {
let expected = self.dtype.storage_bytes(self.numel());
if bytes.len() != expected {
return Err(SessionError::Internal(format!(
"Tensor::overwrite_bytes: got {} bytes for shape {:?} dtype {:?}, expected {expected}",
bytes.len(),
self.shape,
self.dtype
)));
}
let buffer = self
.buffer
.as_mut()
.expect("Tensor buffer taken only in Drop");
self.allocator.copy_from_host(bytes, buffer)?;
Ok(())
}
pub fn try_as_slice_f32(&self) -> Option<&[f32]> {
assert_eq!(
self.dtype,
DataType::Float32,
"try_as_slice_f32 on non-f32 tensor"
);
if cfg!(target_endian = "big") {
return None;
}
let bytes = self.as_bytes();
if bytes.as_ptr().align_offset(std::mem::align_of::<f32>()) != 0 {
return None;
}
Some(unsafe { std::slice::from_raw_parts(bytes.as_ptr().cast::<f32>(), self.numel()) })
}
pub fn try_as_slice_u16(&self) -> Option<&[u16]> {
assert!(
matches!(self.dtype, DataType::Float16 | DataType::BFloat16),
"try_as_slice_u16 on non-16-bit float tensor"
);
if cfg!(target_endian = "big") {
return None;
}
let bytes = self.as_bytes();
if bytes.as_ptr().align_offset(std::mem::align_of::<u16>()) != 0 {
return None;
}
Some(unsafe { std::slice::from_raw_parts(bytes.as_ptr().cast::<u16>(), self.numel()) })
}
pub fn to_vec_f32(&self) -> Vec<f32> {
assert_eq!(
self.dtype,
DataType::Float32,
"to_vec_f32 on non-f32 tensor"
);
self.try_as_slice_f32().map_or_else(
|| {
self.as_bytes()
.chunks_exact(4)
.map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
.collect()
},
<[f32]>::to_vec,
)
}
pub fn to_vec_i64(&self) -> Vec<i64> {
assert_eq!(self.dtype, DataType::Int64, "to_vec_i64 on non-i64 tensor");
self.as_bytes()
.chunks_exact(8)
.map(|c| i64::from_le_bytes(c.try_into().unwrap()))
.collect()
}
}
impl Clone for Tensor {
fn clone(&self) -> Self {
Self::from_raw_in(
self.allocator.clone(),
self.dtype,
self.shape.clone(),
self.as_bytes(),
)
.expect("Tensor::clone: re-allocation of identical bytes")
}
}
impl std::fmt::Debug for Tensor {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Tensor")
.field("dtype", &self.dtype)
.field("shape", &self.shape)
.field("device", &self.device)
.finish()
}
}
impl Drop for Tensor {
fn drop(&mut self) {
if let Some(buffer) = self.buffer.take() {
let _ = self.allocator.deallocate(buffer);
}
let _ = self.import_guard.take();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::os::raw::c_void;
use std::sync::atomic::{AtomicUsize, Ordering};
struct CountingGuard(Arc<AtomicUsize>);
impl Drop for CountingGuard {
fn drop(&mut self) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
#[test]
fn borrowed_guard_ctor_runs_guard_exactly_once_on_drop() {
let drops = Arc::new(AtomicUsize::new(0));
let mut backing = [1.0f32, 2.0, 3.0, 4.0];
let ptr = backing.as_mut_ptr() as *mut c_void;
let buffer = unsafe {
DeviceBuffer::from_borrowed_parts(ptr, DeviceId::cpu(), backing.len() * 4, 4)
};
assert!(buffer.is_borrowed());
let guard = Box::new(CountingGuard(drops.clone()));
let tensor = Tensor::from_borrowed_parts_with_guard(
shared_cpu_ep(),
DataType::Float32,
vec![4],
TensorLayout::contiguous(),
buffer,
guard,
);
assert_eq!(tensor.as_bytes().len(), 16);
assert_eq!(tensor.try_as_slice_f32().unwrap(), &[1.0, 2.0, 3.0, 4.0]);
assert_eq!(tensor.to_vec_f32(), vec![1.0, 2.0, 3.0, 4.0]);
assert_eq!(
drops.load(Ordering::SeqCst),
0,
"guard alive while tensor is"
);
drop(tensor);
assert_eq!(
drops.load(Ordering::SeqCst),
1,
"guard runs exactly once on drop"
);
}
#[test]
fn borrows_aligned_half_storage_as_raw_bits() {
let bits = [0x3c00u16, 0x4000];
let bytes = bits
.iter()
.flat_map(|value| value.to_le_bytes())
.collect::<Vec<_>>();
let tensor = Tensor::from_raw(DataType::Float16, vec![2], &bytes).unwrap();
assert_eq!(tensor.try_as_slice_u16().unwrap(), bits);
}
#[test]
fn exposes_logical_is_static_while_dynamic_tracks_current_shape() {
let mut logical_mask = DeviceIoBinding::allocate(
shared_cpu_ep(),
"attention_mask".into(),
true,
None,
DataType::Int64,
vec![1, 4096],
vec![1, 4096],
true,
)
.unwrap();
assert!(logical_mask.exposes_logical_input_shape());
assert!(!logical_mask.has_dynamic_logical_input_shape());
logical_mask.set_logical_shape(vec![1, 5]).unwrap();
assert!(logical_mask.exposes_logical_input_shape());
assert!(logical_mask.has_dynamic_logical_input_shape());
assert_eq!(logical_mask.kernel_input_shape(), &[1, 5]);
let mut physical_mask = DeviceIoBinding::allocate(
shared_cpu_ep(),
"attention_mask".into(),
true,
None,
DataType::Int64,
vec![1, 4096],
vec![1, 5],
false,
)
.unwrap();
assert!(!physical_mask.exposes_logical_input_shape());
assert!(!physical_mask.has_dynamic_logical_input_shape());
assert_eq!(physical_mask.kernel_input_shape(), &[1, 4096]);
physical_mask.set_logical_shape(vec![1, 4096]).unwrap();
assert!(!physical_mask.exposes_logical_input_shape());
}
}