use std::sync::{Arc, RwLock, TryLockError};
use dlpk::sys::{DLDataType, DLDevice, DLPackVersion};
use dlpk::{DLPackPointerCast, DLPackTensor, GetDLPackDataType};
use ndarray::ArrayD;
use crate::error::ParseError;
pub trait Array: std::any::Any + Send + Sync {
fn as_any(&self) -> &dyn std::any::Any;
fn as_any_mut(&mut self) -> &mut dyn std::any::Any;
fn shape(&self) -> Vec<usize>;
fn dtype(&self) -> DLDataType;
fn device(&self) -> DLDevice;
fn as_dlpack(
&self,
device: DLDevice,
stream: Option<i64>,
max_version: DLPackVersion,
) -> Result<DLPackTensor, ParseError>;
fn copy(&self) -> Box<dyn Array>;
}
impl<T> Array for Arc<RwLock<ArrayD<T>>>
where
T: 'static + Send + Sync + Clone + Default + GetDLPackDataType + DLPackPointerCast,
{
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn std::any::Any {
self
}
fn shape(&self) -> Vec<usize> {
match self.try_read() {
Ok(lock) => lock.shape().to_vec(),
Err(TryLockError::Poisoned(_)) => panic!("readcon-core array lock is poisoned"),
Err(TryLockError::WouldBlock) => panic!("readcon-core array is already locked"),
}
}
fn dtype(&self) -> DLDataType {
T::get_dlpack_data_type()
}
fn device(&self) -> DLDevice {
DLDevice::cpu()
}
fn as_dlpack(
&self,
device: DLDevice,
_stream: Option<i64>,
_max_version: DLPackVersion,
) -> Result<DLPackTensor, ParseError> {
if device != DLDevice::cpu() {
return Err(ParseError::ValidationError(format!(
"Arc<RwLock<ArrayD>> is CPU-only; requested device {device:?} unsupported"
)));
}
let lock = match self.try_read() {
Ok(lock) => lock,
Err(TryLockError::Poisoned(_)) => {
return Err(ParseError::ValidationError(
"readcon-core array lock is poisoned".into(),
));
}
Err(TryLockError::WouldBlock) => {
return Err(ParseError::ValidationError(
"readcon-core array is already locked".into(),
));
}
};
let owned: ArrayD<T> = lock.to_owned();
DLPackTensor::try_from(owned).map_err(|e| {
ParseError::ValidationError(format!("dlpk ArrayD conversion failed: {e}"))
})
}
fn copy(&self) -> Box<dyn Array> {
Box::new(Arc::clone(self))
}
}
pub fn array_from_shape<T>(shape: &[usize]) -> Box<dyn Array>
where
T: 'static + Send + Sync + Clone + Default + GetDLPackDataType + DLPackPointerCast,
{
let arr: ArrayD<T> = ArrayD::default(ndarray::IxDyn(shape));
Box::new(Arc::new(RwLock::new(arr)))
}
pub fn allocate_array_on_device(
shape: &[usize],
device: DLDevice,
) -> Result<Box<dyn Array>, ParseError> {
if device == DLDevice::cpu() {
return Ok(array_from_shape::<f64>(shape));
}
#[cfg(feature = "cuda")]
{
use dlpk::sys::DLDeviceType;
if device.device_type == DLDeviceType::kDLCUDA {
return crate::cuda_array::allocate_cuda_f64(shape, device.device_id);
}
}
Err(ParseError::ValidationError(format!(
"no device allocator in this build for {device:?}; use caller-supplied device buffers via from_dlpack / array_from_host_f64_on_device, or build with `--features cuda` for CUDA devices"
)))
}
pub struct DeviceTaggedF64Array {
shape: Vec<usize>,
device: DLDevice,
data: Arc<Vec<f64>>,
}
impl DeviceTaggedF64Array {
pub fn new(shape: &[usize], data: Vec<f64>, device: DLDevice) -> Result<Self, ParseError> {
let n: usize = shape.iter().product();
if data.len() != n {
return Err(ParseError::ValidationError(format!(
"device-tagged array: expected {n} f64 values for shape {shape:?}, got {}",
data.len()
)));
}
Ok(Self {
shape: shape.to_vec(),
device,
data: Arc::new(data),
})
}
}
pub fn array_from_host_f64_on_device(
shape: &[usize],
data: Vec<f64>,
device: DLDevice,
) -> Result<Box<dyn Array>, ParseError> {
Ok(Box::new(DeviceTaggedF64Array::new(shape, data, device)?))
}
pub fn from_dlpack_f64(tensor: &DLPackTensor) -> Result<Box<dyn Array>, ParseError> {
let device = tensor.device();
let shape: Vec<usize> = tensor.shape().iter().map(|&d| d as usize).collect();
let n: usize = shape.iter().product();
let dtype = tensor.dtype();
if dtype.code != dlpk::sys::DLDataTypeCode::kDLFloat || dtype.bits != 64 {
return Err(ParseError::ValidationError(format!(
"from_dlpack_f64: expected f64, got dtype code={:?} bits={}",
dtype.code, dtype.bits
)));
}
let ptr = tensor
.data_ptr::<f64>()
.map_err(|e| ParseError::ValidationError(format!("from_dlpack_f64 data_ptr: {e}")))?;
let mut data = vec![0.0f64; n];
if n > 0 {
unsafe {
std::ptr::copy_nonoverlapping(ptr, data.as_mut_ptr(), n);
}
}
array_from_host_f64_on_device(&shape, data, device)
}
struct DeviceTaggedManager {
data: Arc<Vec<f64>>,
shape: Vec<i64>,
}
unsafe extern "C" fn device_tagged_deleter(managed: *mut dlpk::sys::DLManagedTensorVersioned) {
if managed.is_null() {
return;
}
unsafe {
let ctx = (*managed).manager_ctx;
if !ctx.is_null() {
let _ = Box::from_raw(ctx as *mut DeviceTaggedManager);
(*managed).manager_ctx = std::ptr::null_mut();
}
}
}
impl Array for DeviceTaggedF64Array {
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn std::any::Any {
self
}
fn shape(&self) -> Vec<usize> {
self.shape.clone()
}
fn dtype(&self) -> DLDataType {
f64::get_dlpack_data_type()
}
fn device(&self) -> DLDevice {
self.device
}
fn as_dlpack(
&self,
device: DLDevice,
_stream: Option<i64>,
_max_version: DLPackVersion,
) -> Result<DLPackTensor, ParseError> {
if device != self.device {
return Err(ParseError::ValidationError(format!(
"device mismatch: array is on {:?}, requested {:?}",
self.device, device
)));
}
let manager = Box::new(DeviceTaggedManager {
data: Arc::clone(&self.data),
shape: self.shape.iter().map(|&d| d as i64).collect(),
});
let data_ptr = manager.data.as_ptr() as *mut std::ffi::c_void;
let shape_ptr = manager.shape.as_ptr() as *mut i64;
let mut managed = dlpk::sys::DLManagedTensorVersioned {
version: dlpk::sys::DLPackVersion {
major: dlpk::sys::DLPACK_MAJOR_VERSION,
minor: dlpk::sys::DLPACK_MINOR_VERSION,
},
manager_ctx: std::ptr::null_mut(),
deleter: Some(device_tagged_deleter),
dl_tensor: dlpk::sys::DLTensor {
data: data_ptr,
device: self.device,
ndim: self.shape.len() as i32,
dtype: f64::get_dlpack_data_type(),
shape: shape_ptr,
strides: std::ptr::null_mut(),
byte_offset: 0,
},
flags: 0,
};
managed.manager_ctx = Box::into_raw(manager) as *mut std::ffi::c_void;
Ok(unsafe { DLPackTensor::from_raw(managed) })
}
fn copy(&self) -> Box<dyn Array> {
Box::new(Self {
shape: self.shape.clone(),
device: self.device,
data: Arc::clone(&self.data),
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn array_from_shape_reports_shape_and_dtype() {
let a: Box<dyn Array> = array_from_shape::<f64>(&[5, 3]);
assert_eq!(a.shape(), vec![5, 3]);
let dt = a.dtype();
assert_eq!(dt.code, dlpk::sys::DLDataTypeCode::kDLFloat);
assert_eq!(dt.bits, 64);
assert_eq!(dt.lanes, 1);
assert_eq!(a.device(), DLDevice::cpu());
}
#[test]
fn array_copy_shares_storage_via_arc() {
let a = array_from_shape::<f64>(&[2, 3]);
let b = a.copy();
assert_eq!(a.shape(), b.shape());
}
#[test]
fn array_dlpack_export_round_trip() {
let a = array_from_shape::<f64>(&[4, 3]);
let tensor = a
.as_dlpack(DLDevice::cpu(), None, DLPackVersion::current())
.expect("DLPack export should succeed for CPU array");
assert_eq!(tensor.shape(), &[4, 3]);
}
#[cfg(not(feature = "cuda"))]
#[test]
fn allocate_non_cpu_fails_clearly() {
match allocate_array_on_device(&[2, 3], DLDevice::cuda(0)) {
Ok(_) => panic!("non-CPU allocate must fail without --features cuda"),
Err(err) => {
let msg = format!("{err:?}");
assert!(
msg.contains("no device allocator") || msg.contains("allocator"),
"{msg}"
);
}
}
let cpu = allocate_array_on_device(&[2, 3], DLDevice::cpu()).unwrap();
assert_eq!(cpu.device(), DLDevice::cpu());
}
#[cfg(feature = "cuda")]
#[test]
fn allocate_cuda_succeeds_with_feature() {
let a = allocate_array_on_device(&[2, 3], DLDevice::cuda(0))
.expect("CUDA allocate must succeed with --features cuda and a driver");
assert_eq!(a.device(), DLDevice::cuda(0));
assert_eq!(
a.device().device_type,
dlpk::sys::DLDeviceType::kDLCUDA
);
let t = a
.as_dlpack(DLDevice::cuda(0), None, DLPackVersion::current())
.expect("matching as_dlpack");
assert_eq!(t.device().device_type, dlpk::sys::DLDeviceType::kDLCUDA);
let cpu = allocate_array_on_device(&[2, 3], DLDevice::cpu()).unwrap();
assert_eq!(cpu.device(), DLDevice::cpu());
}
#[test]
fn cuda_tagged_preserves_device_and_matching_as_dlpack() {
let data: Vec<f64> = (0..6).map(|i| i as f64).collect();
let a = array_from_host_f64_on_device(&[2, 3], data.clone(), DLDevice::cuda(0)).unwrap();
assert_eq!(a.device(), DLDevice::cuda(0));
assert_eq!(a.shape(), vec![2, 3]);
let mismatch = a
.as_dlpack(DLDevice::cpu(), None, DLPackVersion::current())
.unwrap_err();
assert!(
format!("{mismatch:?}").contains("device mismatch"),
"{mismatch:?}"
);
let tensor = a
.as_dlpack(DLDevice::cuda(0), None, DLPackVersion::current())
.expect("matching CUDA device export");
assert_eq!(tensor.device(), DLDevice::cuda(0));
assert_eq!(tensor.shape(), &[2, 3]);
let back = from_dlpack_f64(&tensor).unwrap();
assert_eq!(back.device(), DLDevice::cuda(0));
assert_eq!(back.shape(), vec![2, 3]);
let again = back
.as_dlpack(DLDevice::cuda(0), None, DLPackVersion::current())
.unwrap();
assert_eq!(again.device(), DLDevice::cuda(0));
}
#[test]
fn cpu_tagged_path_unchanged() {
let a = array_from_host_f64_on_device(&[1, 3], vec![1.0, 2.0, 3.0], DLDevice::cpu()).unwrap();
assert_eq!(a.device(), DLDevice::cpu());
let t = a
.as_dlpack(DLDevice::cpu(), None, DLPackVersion::current())
.unwrap();
assert_eq!(t.device(), DLDevice::cpu());
let back = from_dlpack_f64(&t).unwrap();
assert_eq!(back.device(), DLDevice::cpu());
}
}