1use std::collections::{HashMap, VecDeque};
16use std::fs::{File, OpenOptions};
17use std::io::Write;
18use std::path::PathBuf;
19#[cfg(feature = "simd")]
20use std::sync::atomic::{AtomicBool, Ordering};
21use std::sync::{Arc, RwLock};
22
23#[cfg(feature = "gpu")]
24use torsh_core::sync::RwLockExt;
25use torsh_core::{
26 dtype::TensorElement,
27 error::{Result, TorshError},
28};
29
30use crate::memory_pool::global_acquire_uninit;
31
32#[cfg(feature = "simd")]
34use scirs2_core::simd_aligned::AlignedVec;
35
36#[cfg(unix)]
37use std::os::unix::fs::FileExt;
38#[cfg(windows)]
39use std::os::windows::fs::FileExt;
40
41const MEMORY_MAPPING_THRESHOLD: usize = 1024 * 1024 * 1024;
43
44#[cfg(feature = "simd")]
47const ALIGNED_STORAGE_THRESHOLD: usize = 1024;
48
49#[cfg(feature = "simd")]
52const SIMD_OPTIMIZED_THRESHOLD: usize = 10240;
53
54#[cfg(feature = "simd")]
86pub struct SimdStorage<T> {
87 original: AlignedVec<T>,
90 cow: RwLock<Option<AlignedVec<T>>>,
92 mutated: AtomicBool,
94 shared: AtomicBool,
96}
97
98#[cfg(feature = "simd")]
99impl<T> SimdStorage<T> {
100 pub fn new(data: AlignedVec<T>) -> Self {
102 Self {
103 original: data,
104 cow: RwLock::new(None),
105 mutated: AtomicBool::new(false),
106 shared: AtomicBool::new(false),
107 }
108 }
109
110 pub fn len(&self) -> usize {
114 self.original.len()
115 }
116
117 pub fn is_empty(&self) -> bool {
119 self.original.is_empty()
120 }
121
122 pub fn capacity(&self) -> usize {
124 self.original.capacity()
125 }
126
127 pub fn is_mutated(&self) -> bool {
130 self.mutated.load(Ordering::Acquire)
131 }
132
133 pub fn try_as_slice(&self) -> Option<&[T]> {
140 if self.is_mutated() {
141 None
142 } else {
143 Some(self.original.as_slice())
144 }
145 }
146
147 pub fn mark_shared(&self) {
149 self.shared.store(true, Ordering::SeqCst);
150 }
151
152 pub fn is_shared(&self) -> bool {
154 self.shared.load(Ordering::SeqCst)
155 }
156}
157
158#[cfg(feature = "simd")]
159impl<T: Copy> SimdStorage<T> {
160 pub fn with_slice<R>(&self, f: impl FnOnce(&[T]) -> R) -> R {
162 if !self.is_mutated() {
163 return f(self.original.as_slice());
164 }
165 let guard = self.cow.read().unwrap_or_else(|e| e.into_inner());
169 match guard.as_ref() {
170 Some(buffer) => f(buffer.as_slice()),
171 None => f(self.original.as_slice()),
174 }
175 }
176
177 pub fn with_slice_mut<R>(&self, f: impl FnOnce(&mut [T]) -> R) -> Result<R> {
180 let mut guard = self.cow.write().unwrap_or_else(|e| e.into_inner());
181 if guard.is_none() {
182 let source = self.original.as_slice();
183 let mut buffer = AlignedVec::with_capacity(source.len()).map_err(|e| {
184 TorshError::InvalidArgument(format!("Failed to create SIMD COW buffer: {e}"))
185 })?;
186 if !source.is_empty() {
187 unsafe {
191 std::ptr::copy_nonoverlapping(
192 source.as_ptr(),
193 buffer.as_mut_ptr(),
194 source.len(),
195 );
196 buffer.set_len(source.len());
197 }
198 }
199 *guard = Some(buffer);
200 self.mutated.store(true, Ordering::Release);
203 }
204
205 match guard.as_mut() {
206 Some(buffer) => Ok(f(buffer.as_mut_slice())),
207 None => Err(TorshError::SynchronizationError(
208 "SIMD copy-on-write buffer disappeared".to_string(),
209 )),
210 }
211 }
212
213 pub fn to_vec(&self) -> Vec<T> {
215 self.with_slice(|slice| slice.to_vec())
216 }
217}
218
219#[cfg(feature = "simd")]
220impl<T> std::fmt::Debug for SimdStorage<T> {
221 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
222 f.debug_struct("SimdStorage")
223 .field("len", &self.original.len())
224 .field("mutated", &self.mutated.load(Ordering::Relaxed))
225 .field("shared", &self.shared.load(Ordering::Relaxed))
226 .finish()
227 }
228}
229
230#[cfg(feature = "gpu")]
257pub struct DeviceBuffer {
258 ptr: u64,
260 bytes: usize,
262 dtype: torsh_core::dtype::DType,
264 backend: Arc<dyn oxicuda_backend::ComputeBackend>,
266}
267
268#[cfg(feature = "gpu")]
269impl DeviceBuffer {
270 pub(crate) fn adopt(
275 ptr: u64,
276 bytes: usize,
277 dtype: torsh_core::dtype::DType,
278 backend: Arc<dyn oxicuda_backend::ComputeBackend>,
279 ) -> Self {
280 Self {
281 ptr,
282 bytes,
283 dtype,
284 backend,
285 }
286 }
287
288 pub fn ptr(&self) -> u64 {
290 self.ptr
291 }
292
293 pub fn bytes(&self) -> usize {
295 self.bytes
296 }
297
298 pub fn dtype(&self) -> torsh_core::dtype::DType {
300 self.dtype
301 }
302
303 pub fn backend(&self) -> &Arc<dyn oxicuda_backend::ComputeBackend> {
305 &self.backend
306 }
307}
308
309#[cfg(feature = "gpu")]
310impl Drop for DeviceBuffer {
311 fn drop(&mut self) {
312 let _ = self.backend.free(self.ptr);
315 }
316}
317
318#[cfg(feature = "gpu")]
319impl std::fmt::Debug for DeviceBuffer {
320 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
321 f.debug_struct("DeviceBuffer")
322 .field("ptr", &format_args!("{:#x}", self.ptr))
323 .field("bytes", &self.bytes)
324 .field("dtype", &self.dtype)
325 .field("backend", &self.backend.name())
326 .finish()
327 }
328}
329
330pub enum TensorStorage<T: TensorElement> {
332 InMemory(Arc<RwLock<Vec<T>>>),
334 MemoryMapped(Arc<RwLock<MemoryMappedStorage<T>>>),
336 #[cfg(feature = "simd")]
338 Aligned(Arc<RwLock<AlignedVec<T>>>),
339 #[cfg(feature = "simd")]
346 SimdOptimized(Arc<SimdStorage<T>>),
347 #[cfg(feature = "gpu")]
352 Device {
353 buffer: Arc<DeviceBuffer>,
355 host_cache: Arc<RwLock<Option<Vec<T>>>>,
361 },
362}
363
364impl<T: TensorElement> std::fmt::Debug for TensorStorage<T> {
365 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
366 match self {
367 Self::InMemory(data) => f.debug_tuple("InMemory").field(data).finish(),
368 Self::MemoryMapped(storage) => f.debug_tuple("MemoryMapped").field(storage).finish(),
369 #[cfg(feature = "simd")]
370 Self::Aligned(_) => f.debug_tuple("Aligned").field(&"<AlignedVec>").finish(),
371 #[cfg(feature = "simd")]
372 Self::SimdOptimized(storage) => f.debug_tuple("SimdOptimized").field(storage).finish(),
373 #[cfg(feature = "gpu")]
374 Self::Device { buffer, .. } => f.debug_tuple("Device").field(buffer).finish(),
375 }
376 }
377}
378
379#[derive(Debug)]
381pub struct MemoryMappedStorage<T: TensorElement> {
382 file: File,
384 file_path: PathBuf,
386 num_elements: usize,
388 cache: HashMap<usize, T>,
390 max_cache_size: usize,
392 access_pattern: VecDeque<usize>,
394 is_temporary: bool,
396}
397
398impl<T: TensorElement + Copy> TensorStorage<T> {
399 pub fn in_memory(data: Vec<T>) -> Self {
401 Self::InMemory(Arc::new(RwLock::new(data)))
402 }
403
404 pub fn memory_mapped(data: Vec<T>, file_path: Option<PathBuf>) -> Result<Self> {
406 let storage = MemoryMappedStorage::new(data, file_path)?;
407 Ok(Self::MemoryMapped(Arc::new(RwLock::new(storage))))
408 }
409
410 pub fn memory_mapped_filled(
417 num_elements: usize,
418 value: T,
419 file_path: Option<PathBuf>,
420 ) -> Result<Self> {
421 let storage = MemoryMappedStorage::new_filled(num_elements, value, file_path)?;
422 Ok(Self::MemoryMapped(Arc::new(RwLock::new(storage))))
423 }
424
425 #[cfg(feature = "simd")]
427 pub fn aligned(data: Vec<T>) -> Result<Self> {
428 Ok(Self::Aligned(Arc::new(RwLock::new(Self::to_aligned_vec(
429 &data,
430 )?))))
431 }
432
433 #[cfg(feature = "simd")]
439 pub(crate) fn aligned_from_slice(data: &[T]) -> Result<Self> {
440 Ok(Self::Aligned(Arc::new(RwLock::new(Self::to_aligned_vec(
441 data,
442 )?))))
443 }
444
445 #[cfg(feature = "simd")]
451 fn to_aligned_vec(data: &[T]) -> Result<AlignedVec<T>> {
452 let mut aligned_vec = AlignedVec::with_capacity(data.len()).map_err(|e| {
453 TorshError::InvalidArgument(format!("Failed to create aligned storage: {e}"))
454 })?;
455
456 if !data.is_empty() {
457 unsafe {
462 std::ptr::copy_nonoverlapping(data.as_ptr(), aligned_vec.as_mut_ptr(), data.len());
463 aligned_vec.set_len(data.len());
464 }
465 }
466
467 Ok(aligned_vec)
468 }
469
470 pub fn fast_result(data: Vec<T>) -> Self {
480 Self::InMemory(Arc::new(RwLock::new(data)))
481 }
482
483 #[cfg(feature = "simd")]
495 pub fn simd_optimized(data: Vec<T>) -> Result<Self> {
496 let aligned_vec = Self::to_aligned_vec(&data)?;
497 let simd_storage = SimdStorage::new(aligned_vec);
498 Ok(Self::SimdOptimized(Arc::new(simd_storage)))
499 }
500
501 pub fn create_optimal(data: Vec<T>) -> Result<Self> {
509 let size_bytes = data.len() * std::mem::size_of::<T>();
510
511 if size_bytes >= MEMORY_MAPPING_THRESHOLD {
512 Self::memory_mapped(data, None)
514 } else {
515 #[cfg(feature = "simd")]
516 {
517 if size_bytes >= SIMD_OPTIMIZED_THRESHOLD {
518 return Self::simd_optimized(data);
521 } else if size_bytes >= ALIGNED_STORAGE_THRESHOLD {
522 return Self::aligned(data);
524 }
525 }
526 Ok(Self::in_memory(data))
528 }
529 }
530
531 #[cfg(feature = "gpu")]
535 pub(crate) fn device(buffer: Arc<DeviceBuffer>) -> Self {
536 Self::Device {
537 buffer,
538 host_cache: Arc::new(RwLock::new(None)),
539 }
540 }
541
542 pub fn is_device(&self) -> bool {
547 #[cfg(feature = "gpu")]
548 {
549 matches!(self, Self::Device { .. })
550 }
551 #[cfg(not(feature = "gpu"))]
552 {
553 false
554 }
555 }
556
557 #[cfg(feature = "gpu")]
559 pub(crate) fn device_buffer(&self) -> Option<&Arc<DeviceBuffer>> {
560 match self {
561 Self::Device { buffer, .. } => Some(buffer),
562 _ => None,
563 }
564 }
565
566 #[cfg(feature = "gpu")]
573 fn with_host_cache<R, F>(
574 buffer: &Arc<DeviceBuffer>,
575 host_cache: &RwLock<Option<Vec<T>>>,
576 f: F,
577 ) -> Result<R>
578 where
579 F: FnOnce(&[T]) -> Result<R>,
580 T: Copy,
581 {
582 {
585 let guard = host_cache.read_or_recover();
586 if let Some(cached) = guard.as_ref() {
587 return f(cached);
588 }
589 }
590
591 let downloaded = Self::download(buffer)?;
595 {
596 let mut guard = host_cache.write_or_recover();
597 if guard.is_none() {
598 *guard = Some(downloaded);
599 }
600 }
601
602 let guard = host_cache.read_or_recover();
603 match guard.as_ref() {
604 Some(cached) => f(cached),
605 None => Err(TorshError::SynchronizationError(
606 "device host cache disappeared".to_string(),
607 )),
608 }
609 }
610
611 #[cfg(feature = "gpu")]
613 fn download(buffer: &Arc<DeviceBuffer>) -> Result<Vec<T>>
614 where
615 T: Copy,
616 {
617 let element_size = std::mem::size_of::<T>();
618 if element_size == 0 || buffer.bytes() % element_size != 0 {
619 return Err(TorshError::InvalidOperation(format!(
620 "device buffer of {} bytes does not hold whole {}-byte elements",
621 buffer.bytes(),
622 element_size
623 )));
624 }
625 if buffer.dtype() != T::dtype() {
628 return Err(TorshError::InvalidOperation(format!(
629 "device buffer holds {} but the tensor element type is {}",
630 buffer.dtype(),
631 T::dtype()
632 )));
633 }
634
635 let count = buffer.bytes() / element_size;
636 let mut raw = vec![0u8; buffer.bytes()];
637 buffer
638 .backend()
639 .copy_dtoh(&mut raw, buffer.ptr())
640 .map_err(|e| TorshError::InvalidOperation(format!("device download failed: {e}")))?;
641
642 let mut out: Vec<T> = Vec::with_capacity(count);
643 if count > 0 {
644 unsafe {
651 std::ptr::copy_nonoverlapping(
652 raw.as_ptr(),
653 out.as_mut_ptr().cast::<u8>(),
654 buffer.bytes(),
655 );
656 out.set_len(count);
657 }
658 }
659 Ok(out)
660 }
661
662 pub fn len(&self) -> usize {
664 match self {
665 Self::InMemory(data) => {
666 data.read().map(|guard| guard.len()).unwrap_or(0) }
668 Self::MemoryMapped(storage) => {
669 storage.read().map(|guard| guard.num_elements).unwrap_or(0) }
671 #[cfg(feature = "simd")]
672 Self::Aligned(data) => {
673 data.read().map(|guard| guard.len()).unwrap_or(0) }
675 #[cfg(feature = "simd")]
676 Self::SimdOptimized(storage) => storage.len(), #[cfg(feature = "gpu")]
678 Self::Device { buffer, .. } => {
679 let element_size = std::mem::size_of::<T>();
681 if element_size == 0 {
682 0
683 } else {
684 buffer.bytes() / element_size
685 }
686 }
687 }
688 }
689
690 pub fn is_empty(&self) -> bool {
692 self.len() == 0
693 }
694
695 pub fn get(&self, index: usize) -> Result<T>
697 where
698 T: Copy,
699 {
700 match self {
701 Self::InMemory(data) => {
702 let data_guard = data.read().map_err(|_| {
703 TorshError::SynchronizationError("Lock poisoned during read".to_string())
704 })?;
705 data_guard
706 .get(index)
707 .copied()
708 .ok_or_else(|| TorshError::IndexOutOfBounds {
709 index,
710 size: data_guard.len(),
711 })
712 }
713 Self::MemoryMapped(storage) => storage
714 .write()
715 .map_err(|_| {
716 TorshError::SynchronizationError("Lock poisoned during write".to_string())
717 })?
718 .get(index),
719 #[cfg(feature = "simd")]
720 Self::Aligned(data) => {
721 let data_guard = data.read().map_err(|_| {
722 TorshError::SynchronizationError("Lock poisoned during read".to_string())
723 })?;
724 if index >= data_guard.len() {
725 Err(TorshError::IndexOutOfBounds {
726 index,
727 size: data_guard.len(),
728 })
729 } else {
730 Ok(data_guard.as_slice()[index])
731 }
732 }
733 #[cfg(feature = "simd")]
734 Self::SimdOptimized(storage) => {
735 storage.with_slice(|slice| {
737 slice
738 .get(index)
739 .copied()
740 .ok_or_else(|| TorshError::IndexOutOfBounds {
741 index,
742 size: slice.len(),
743 })
744 })
745 }
746 #[cfg(feature = "gpu")]
747 Self::Device { buffer, host_cache } => {
748 Self::with_host_cache(buffer, host_cache, |slice| {
749 slice
750 .get(index)
751 .copied()
752 .ok_or_else(|| TorshError::IndexOutOfBounds {
753 index,
754 size: slice.len(),
755 })
756 })
757 }
758 }
759 }
760
761 pub fn set(&self, index: usize, value: T) -> Result<()>
763 where
764 T: Copy,
765 {
766 match self {
767 Self::InMemory(data) => {
768 let mut data_guard = data.write().map_err(|_| {
769 TorshError::SynchronizationError("Lock poisoned during write".to_string())
770 })?;
771 if index >= data_guard.len() {
772 return Err(TorshError::IndexOutOfBounds {
773 index,
774 size: data_guard.len(),
775 });
776 }
777 data_guard[index] = value;
778 Ok(())
779 }
780 Self::MemoryMapped(storage) => storage
781 .write()
782 .map_err(|_| {
783 TorshError::SynchronizationError("Lock poisoned during write".to_string())
784 })?
785 .set(index, value),
786 #[cfg(feature = "simd")]
787 Self::Aligned(data) => {
788 let mut data_guard = data.write().map_err(|_| {
789 TorshError::SynchronizationError("Lock poisoned during write".to_string())
790 })?;
791 if index >= data_guard.len() {
792 return Err(TorshError::IndexOutOfBounds {
793 index,
794 size: data_guard.len(),
795 });
796 }
797 (*data_guard).set(index, value);
799 Ok(())
800 }
801 #[cfg(feature = "simd")]
802 Self::SimdOptimized(storage) => {
803 storage.with_slice_mut(|slice| {
806 let size = slice.len();
807 match slice.get_mut(index) {
808 Some(slot) => {
809 *slot = value;
810 Ok(())
811 }
812 None => Err(TorshError::IndexOutOfBounds { index, size }),
813 }
814 })?
815 }
816 #[cfg(feature = "gpu")]
817 Self::Device { .. } => Err(Self::device_is_immutable()),
818 }
819 }
820
821 #[cfg(feature = "gpu")]
826 fn device_is_immutable() -> TorshError {
827 TorshError::InvalidOperation(
828 "device-resident storage is immutable; call make_unique() or to_device(DeviceType::Cpu) first"
829 .to_string(),
830 )
831 }
832
833 pub fn get_slice(&self, start: usize, len: usize) -> Result<Vec<T>>
835 where
836 T: Copy,
837 {
838 match self {
839 Self::InMemory(data) => {
840 let data_guard = data.read().map_err(|_| {
841 TorshError::SynchronizationError("Lock poisoned during read".to_string())
842 })?;
843 if start + len > data_guard.len() {
844 return Err(TorshError::IndexOutOfBounds {
845 index: start + len - 1,
846 size: data_guard.len(),
847 });
848 }
849 Ok(data_guard[start..start + len].to_vec())
850 }
851 Self::MemoryMapped(storage) => storage
852 .write()
853 .map_err(|_| {
854 TorshError::SynchronizationError("Lock poisoned during write".to_string())
855 })?
856 .get_slice(start, len),
857 #[cfg(feature = "simd")]
858 Self::Aligned(data) => {
859 let data_guard = data.read().map_err(|_| {
860 TorshError::SynchronizationError("Lock poisoned during read".to_string())
861 })?;
862 if start + len > data_guard.len() {
863 return Err(TorshError::IndexOutOfBounds {
864 index: start + len - 1,
865 size: data_guard.len(),
866 });
867 }
868 let slice = data_guard.as_slice();
869 Ok(slice[start..start + len].to_vec())
870 }
871 #[cfg(feature = "simd")]
872 Self::SimdOptimized(storage) => storage.with_slice(|slice| {
873 if start + len > slice.len() {
874 return Err(TorshError::IndexOutOfBounds {
875 index: start + len - 1,
876 size: slice.len(),
877 });
878 }
879 Ok(slice[start..start + len].to_vec())
880 }),
881 #[cfg(feature = "gpu")]
882 Self::Device { buffer, host_cache } => {
883 Self::with_host_cache(buffer, host_cache, |slice| {
884 if start + len > slice.len() {
885 return Err(TorshError::IndexOutOfBounds {
886 index: start + len - 1,
887 size: slice.len(),
888 });
889 }
890 Ok(slice[start..start + len].to_vec())
891 })
892 }
893 }
894 }
895
896 pub fn set_slice(&self, start: usize, values: &[T]) -> Result<()>
898 where
899 T: Copy,
900 {
901 match self {
902 Self::InMemory(data) => {
903 let mut data_guard = data.write().map_err(|_| {
904 TorshError::SynchronizationError("Lock poisoned during write".to_string())
905 })?;
906 if start + values.len() > data_guard.len() {
907 return Err(TorshError::IndexOutOfBounds {
908 index: start + values.len() - 1,
909 size: data_guard.len(),
910 });
911 }
912 data_guard[start..start + values.len()].copy_from_slice(values);
913 Ok(())
914 }
915 Self::MemoryMapped(storage) => storage
916 .write()
917 .map_err(|_| {
918 TorshError::SynchronizationError("Lock poisoned during write".to_string())
919 })?
920 .set_slice(start, values),
921 #[cfg(feature = "simd")]
922 Self::Aligned(data) => {
923 let mut data_guard = data.write().map_err(|_| {
924 TorshError::SynchronizationError("Lock poisoned during write".to_string())
925 })?;
926 if start + values.len() > data_guard.len() {
927 return Err(TorshError::IndexOutOfBounds {
928 index: start + values.len() - 1,
929 size: data_guard.len(),
930 });
931 }
932 let slice = data_guard.as_mut_slice();
934 slice[start..start + values.len()].copy_from_slice(values);
935 Ok(())
936 }
937 #[cfg(feature = "simd")]
938 Self::SimdOptimized(storage) => storage.with_slice_mut(|slice| {
939 let size = slice.len();
940 if start + values.len() > size {
941 return Err(TorshError::IndexOutOfBounds {
942 index: start + values.len() - 1,
943 size,
944 });
945 }
946 slice[start..start + values.len()].copy_from_slice(values);
947 Ok(())
948 })?,
949 #[cfg(feature = "gpu")]
950 Self::Device { .. } => Err(Self::device_is_immutable()),
951 }
952 }
953
954 pub fn to_vec(&self) -> Result<Vec<T>>
956 where
957 T: Copy,
958 {
959 match self {
960 Self::InMemory(data) => Ok(data
961 .read()
962 .map_err(|_| {
963 TorshError::SynchronizationError("Lock poisoned during read".to_string())
964 })?
965 .clone()),
966 Self::MemoryMapped(storage) => storage
967 .write()
968 .map_err(|_| {
969 TorshError::SynchronizationError("Lock poisoned during write".to_string())
970 })?
971 .to_vec(),
972 #[cfg(feature = "simd")]
973 Self::Aligned(data) => {
974 let data_guard = data.read().map_err(|_| {
975 TorshError::SynchronizationError("Lock poisoned during read".to_string())
976 })?;
977 Ok(data_guard.as_slice().to_vec())
978 }
979 #[cfg(feature = "simd")]
980 Self::SimdOptimized(storage) => Ok(storage.to_vec()),
981 #[cfg(feature = "gpu")]
982 Self::Device { buffer, host_cache } => {
983 Self::with_host_cache(buffer, host_cache, |slice| Ok(slice.to_vec()))
984 }
985 }
986 }
987
988 pub fn storage_type(&self) -> &'static str {
990 match self {
991 Self::InMemory(_) => "in_memory",
992 Self::MemoryMapped(_) => "memory_mapped",
993 #[cfg(feature = "simd")]
994 Self::Aligned(_) => "aligned_simd",
995 #[cfg(feature = "simd")]
996 Self::SimdOptimized(_) => "simd_optimized",
997 #[cfg(feature = "gpu")]
998 Self::Device { .. } => "device",
999 }
1000 }
1001
1002 pub fn memory_usage(&self) -> usize {
1004 match self {
1005 Self::InMemory(data) => {
1006 data.read()
1007 .map(|guard| guard.len() * std::mem::size_of::<T>())
1008 .unwrap_or(0) }
1010 Self::MemoryMapped(storage) => {
1011 storage
1012 .read()
1013 .map(|storage_guard| {
1014 storage_guard.cache.len() * std::mem::size_of::<T>()
1016 + std::mem::size_of::<MemoryMappedStorage<T>>()
1017 })
1018 .unwrap_or(std::mem::size_of::<MemoryMappedStorage<T>>()) }
1020 #[cfg(feature = "simd")]
1021 Self::Aligned(data) => {
1022 data.read()
1023 .map(|data_guard| {
1024 data_guard.capacity() * std::mem::size_of::<T>()
1026 })
1027 .unwrap_or(0) }
1029 #[cfg(feature = "simd")]
1030 Self::SimdOptimized(storage) => {
1031 let buffers = if storage.is_mutated() { 2 } else { 1 };
1034 storage.capacity() * std::mem::size_of::<T>() * buffers
1035 }
1036 #[cfg(feature = "gpu")]
1037 Self::Device { buffer, host_cache } => {
1038 let cached = host_cache
1040 .read_or_recover()
1041 .as_ref()
1042 .map_or(0, |cache| cache.len() * std::mem::size_of::<T>());
1043 buffer.bytes() + cached
1044 }
1045 }
1046 }
1047
1048 pub fn with_slice<R, F>(&self, f: F) -> Result<R>
1071 where
1072 F: FnOnce(&[T]) -> Result<R>,
1073 T: Copy,
1074 {
1075 match self {
1076 Self::InMemory(data) => {
1077 let data_guard = data.read().map_err(|_| {
1078 TorshError::SynchronizationError("Lock poisoned during read".to_string())
1079 })?;
1080 f(data_guard.as_slice())
1081 }
1082 Self::MemoryMapped(storage) => {
1083 let vec = storage
1085 .write()
1086 .map_err(|_| {
1087 TorshError::SynchronizationError("Lock poisoned during write".to_string())
1088 })?
1089 .to_vec()?;
1090 f(&vec)
1091 }
1092 #[cfg(feature = "simd")]
1093 Self::Aligned(data) => {
1094 let data_guard = data.read().map_err(|_| {
1095 TorshError::SynchronizationError("Lock poisoned during read".to_string())
1096 })?;
1097 f(data_guard.as_slice())
1098 }
1099 #[cfg(feature = "simd")]
1100 Self::SimdOptimized(storage) => {
1101 storage.with_slice(f)
1103 }
1104 #[cfg(feature = "gpu")]
1105 Self::Device { buffer, host_cache } => Self::with_host_cache(buffer, host_cache, f),
1106 }
1107 }
1108
1109 #[cfg(feature = "simd")]
1121 pub fn try_as_slice_direct(&self) -> Option<&[T]> {
1122 match self {
1123 Self::SimdOptimized(storage) => storage.try_as_slice(),
1124 _ => None,
1125 }
1126 }
1127
1128 pub fn with_slice_mut<R, F>(&self, f: F) -> Result<R>
1152 where
1153 F: FnOnce(&mut [T]) -> Result<R>,
1154 T: Copy,
1155 {
1156 match self {
1157 Self::InMemory(data) => {
1158 let mut data_guard = data.write().map_err(|_| {
1159 TorshError::SynchronizationError("Lock poisoned during write".to_string())
1160 })?;
1161 f(data_guard.as_mut_slice())
1162 }
1163 Self::MemoryMapped(_) => {
1164 Err(TorshError::InvalidArgument(
1166 "Memory-mapped storage does not support mutable slice access".to_string(),
1167 ))
1168 }
1169 #[cfg(feature = "simd")]
1170 Self::Aligned(data) => {
1171 let mut data_guard = data.write().map_err(|_| {
1172 TorshError::SynchronizationError("Lock poisoned during write".to_string())
1173 })?;
1174 f(data_guard.as_mut_slice())
1175 }
1176 #[cfg(feature = "simd")]
1177 Self::SimdOptimized(storage) => {
1178 storage.with_slice_mut(f)?
1180 }
1181 #[cfg(feature = "gpu")]
1182 Self::Device { .. } => Err(Self::device_is_immutable()),
1183 }
1184 }
1185}
1186
1187fn unique_backing_path() -> PathBuf {
1193 static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
1194 let seq = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1195 let nanos = std::time::SystemTime::now()
1196 .duration_since(std::time::UNIX_EPOCH)
1197 .unwrap_or_default()
1198 .as_nanos();
1199 std::env::temp_dir().join(format!(
1200 "torsh_tensor_{pid}_{nanos}_{seq}.mmap",
1201 pid = std::process::id()
1202 ))
1203}
1204
1205impl<T: TensorElement> MemoryMappedStorage<T> {
1206 fn open_backing_file(file_path: Option<PathBuf>) -> Result<(File, PathBuf, bool)> {
1208 let (file_path, is_temporary) = match file_path {
1209 Some(path) => (path, false),
1210 None => (unique_backing_path(), true),
1213 };
1214
1215 let file = OpenOptions::new()
1216 .create(true)
1217 .read(true)
1218 .write(true)
1219 .truncate(true)
1220 .open(&file_path)
1221 .map_err(|e| {
1222 TorshError::IoError(format!("Failed to create memory-mapped file: {e}"))
1223 })?;
1224
1225 Ok((file, file_path, is_temporary))
1226 }
1227
1228 pub fn new(data: Vec<T>, file_path: Option<PathBuf>) -> Result<Self> {
1230 let (mut file, file_path, is_temporary) = Self::open_backing_file(file_path)?;
1231
1232 let data_bytes = unsafe {
1234 std::slice::from_raw_parts(
1235 data.as_ptr() as *const u8,
1236 std::mem::size_of_val(data.as_slice()),
1237 )
1238 };
1239 file.write_all(data_bytes).map_err(|e| {
1240 TorshError::IoError(format!("Failed to write to memory-mapped file: {e}"))
1241 })?;
1242 file.flush()
1243 .map_err(|e| TorshError::IoError(format!("Failed to flush memory-mapped file: {e}")))?;
1244
1245 Ok(Self {
1246 file,
1247 file_path,
1248 num_elements: data.len(),
1249 cache: HashMap::new(),
1250 max_cache_size: 10000, access_pattern: VecDeque::new(),
1252 is_temporary,
1253 })
1254 }
1255
1256 pub fn new_filled(num_elements: usize, value: T, file_path: Option<PathBuf>) -> Result<Self>
1263 where
1264 T: Copy,
1265 {
1266 let (mut file, file_path, is_temporary) = Self::open_backing_file(file_path)?;
1267
1268 let element_size = std::mem::size_of::<T>();
1269 if element_size > 0 && num_elements > 0 {
1270 const TARGET_CHUNK_BYTES: usize = 1024 * 1024;
1272 let chunk_elements = (TARGET_CHUNK_BYTES / element_size).clamp(1, num_elements);
1273 let chunk = vec![value; chunk_elements];
1274 let chunk_bytes = unsafe {
1275 std::slice::from_raw_parts(
1276 chunk.as_ptr() as *const u8,
1277 std::mem::size_of_val(chunk.as_slice()),
1278 )
1279 };
1280
1281 let mut written = 0usize;
1282 while written < num_elements {
1283 let remaining = num_elements - written;
1284 let this_chunk = remaining.min(chunk_elements);
1285 file.write_all(&chunk_bytes[..this_chunk * element_size])
1286 .map_err(|e| {
1287 TorshError::IoError(format!("Failed to write to memory-mapped file: {e}"))
1288 })?;
1289 written += this_chunk;
1290 }
1291 }
1292
1293 file.flush()
1294 .map_err(|e| TorshError::IoError(format!("Failed to flush memory-mapped file: {e}")))?;
1295
1296 Ok(Self {
1297 file,
1298 file_path,
1299 num_elements,
1300 cache: HashMap::new(),
1301 max_cache_size: 10000,
1302 access_pattern: VecDeque::new(),
1303 is_temporary,
1304 })
1305 }
1306
1307 pub fn file_path(&self) -> &std::path::Path {
1309 &self.file_path
1310 }
1311
1312 pub fn get(&mut self, index: usize) -> Result<T>
1314 where
1315 T: Copy,
1316 {
1317 if index >= self.num_elements {
1318 return Err(TorshError::IndexOutOfBounds {
1319 index,
1320 size: self.num_elements,
1321 });
1322 }
1323
1324 if let Some(&value) = self.cache.get(&index) {
1326 self.update_access_pattern(index);
1327 return Ok(value);
1328 }
1329
1330 let value = self.read_element_from_file(index)?;
1332
1333 if self.cache.len() < self.max_cache_size {
1335 self.cache.insert(index, value);
1336 } else {
1337 self.evict_lru();
1339 self.cache.insert(index, value);
1340 }
1341
1342 self.update_access_pattern(index);
1343 Ok(value)
1344 }
1345
1346 pub fn set(&mut self, index: usize, value: T) -> Result<()>
1348 where
1349 T: Copy,
1350 {
1351 if index >= self.num_elements {
1352 return Err(TorshError::IndexOutOfBounds {
1353 index,
1354 size: self.num_elements,
1355 });
1356 }
1357
1358 self.cache.insert(index, value);
1360
1361 self.write_element_to_file(index, value)?;
1363 self.update_access_pattern(index);
1364 Ok(())
1365 }
1366
1367 pub fn get_slice(&mut self, start: usize, len: usize) -> Result<Vec<T>>
1374 where
1375 T: Copy,
1376 {
1377 if start + len > self.num_elements {
1378 return Err(TorshError::IndexOutOfBounds {
1379 index: start + len - 1,
1380 size: self.num_elements,
1381 });
1382 }
1383
1384 if len == 0 {
1385 return Ok(Vec::new());
1386 }
1387
1388 let element_size = std::mem::size_of::<T>();
1389 let mut buf = global_acquire_uninit::<T>(len);
1390
1391 if element_size == 0 {
1392 let uninit = buf.as_uninit_slice_mut();
1394 for slot in uninit.iter_mut().take(len) {
1395 slot.write(unsafe { std::mem::zeroed() });
1398 }
1399 return Ok(buf.into_vec(len));
1400 }
1401
1402 {
1403 let byte_len = len * element_size;
1404 let ptr = buf.as_uninit_slice_mut().as_mut_ptr() as *mut u8;
1405 let byte_buf = unsafe {
1412 std::ptr::write_bytes(ptr, 0, byte_len);
1413 std::slice::from_raw_parts_mut(ptr, byte_len)
1414 };
1415 self.read_bytes_at(byte_buf, (start * element_size) as u64)?;
1416 }
1417
1418 Ok(buf.into_vec(len))
1419 }
1420
1421 fn read_bytes_at(&mut self, buffer: &mut [u8], offset: u64) -> Result<()> {
1423 #[cfg(unix)]
1424 {
1425 self.file.read_exact_at(buffer, offset).map_err(|e| {
1426 TorshError::IoError(format!("Failed to read from memory-mapped file: {e}"))
1427 })?;
1428 }
1429
1430 #[cfg(windows)]
1431 {
1432 let mut read_total = 0usize;
1433 while read_total < buffer.len() {
1434 let n = self
1435 .file
1436 .seek_read(&mut buffer[read_total..], offset + read_total as u64)
1437 .map_err(|e| {
1438 TorshError::IoError(format!("Failed to read from memory-mapped file: {e}"))
1439 })?;
1440 if n == 0 {
1441 return Err(TorshError::IoError(
1442 "Unexpected end of memory-mapped file".to_string(),
1443 ));
1444 }
1445 read_total += n;
1446 }
1447 }
1448
1449 #[cfg(not(any(unix, windows)))]
1450 {
1451 use std::io::{Read, Seek, SeekFrom};
1452 self.file.seek(SeekFrom::Start(offset)).map_err(|e| {
1453 TorshError::IoError(format!("Failed to seek in memory-mapped file: {e}"))
1454 })?;
1455 self.file.read_exact(buffer).map_err(|e| {
1456 TorshError::IoError(format!("Failed to read from memory-mapped file: {e}"))
1457 })?;
1458 }
1459
1460 Ok(())
1461 }
1462
1463 pub fn set_slice(&mut self, start: usize, values: &[T]) -> Result<()>
1465 where
1466 T: Copy,
1467 {
1468 if start + values.len() > self.num_elements {
1469 return Err(TorshError::IndexOutOfBounds {
1470 index: start + values.len() - 1,
1471 size: self.num_elements,
1472 });
1473 }
1474
1475 for (i, &value) in values.iter().enumerate() {
1476 self.set(start + i, value)?;
1477 }
1478 Ok(())
1479 }
1480
1481 pub fn to_vec(&mut self) -> Result<Vec<T>>
1483 where
1484 T: Copy,
1485 {
1486 self.get_slice(0, self.num_elements)
1487 }
1488
1489 fn read_element_from_file(&mut self, index: usize) -> Result<T>
1491 where
1492 T: Copy,
1493 {
1494 let offset = index * std::mem::size_of::<T>();
1495 let mut buffer = vec![0u8; std::mem::size_of::<T>()];
1496
1497 #[cfg(unix)]
1498 {
1499 self.file
1500 .read_exact_at(&mut buffer, offset as u64)
1501 .map_err(|e| {
1502 TorshError::IoError(format!("Failed to read from memory-mapped file: {e}"))
1503 })?;
1504 }
1505
1506 #[cfg(windows)]
1507 {
1508 self.file
1509 .seek_read(&mut buffer, offset as u64)
1510 .map_err(|e| {
1511 TorshError::IoError(format!("Failed to read from memory-mapped file: {e}"))
1512 })?;
1513 }
1514
1515 #[cfg(not(any(unix, windows)))]
1516 {
1517 self.file
1518 .seek(SeekFrom::Start(offset as u64))
1519 .map_err(|e| {
1520 TorshError::IoError(format!("Failed to seek in memory-mapped file: {e}"))
1521 })?;
1522 self.file.read_exact(&mut buffer).map_err(|e| {
1523 TorshError::IoError(format!("Failed to read from memory-mapped file: {e}"))
1524 })?;
1525 }
1526
1527 let value = unsafe { std::ptr::read_unaligned(buffer.as_ptr() as *const T) };
1530 Ok(value)
1531 }
1532
1533 fn write_element_to_file(&mut self, index: usize, value: T) -> Result<()>
1535 where
1536 T: Copy,
1537 {
1538 let offset = index * std::mem::size_of::<T>();
1539 let buffer = unsafe {
1540 std::slice::from_raw_parts(&value as *const T as *const u8, std::mem::size_of::<T>())
1541 };
1542
1543 #[cfg(unix)]
1544 {
1545 self.file.write_all_at(buffer, offset as u64).map_err(|e| {
1546 TorshError::IoError(format!("Failed to write to memory-mapped file: {e}"))
1547 })?;
1548 }
1549
1550 #[cfg(windows)]
1551 {
1552 self.file.seek_write(buffer, offset as u64).map_err(|e| {
1553 TorshError::IoError(format!("Failed to write to memory-mapped file: {e}"))
1554 })?;
1555 }
1556
1557 #[cfg(not(any(unix, windows)))]
1558 {
1559 self.file
1560 .seek(SeekFrom::Start(offset as u64))
1561 .map_err(|e| {
1562 TorshError::IoError(format!("Failed to seek in memory-mapped file: {e}"))
1563 })?;
1564 self.file.write_all(buffer).map_err(|e| {
1565 TorshError::IoError(format!("Failed to write to memory-mapped file: {e}"))
1566 })?;
1567 }
1568
1569 Ok(())
1570 }
1571
1572 fn update_access_pattern(&mut self, index: usize) {
1574 self.access_pattern.push_back(index);
1575 if self.access_pattern.len() > self.max_cache_size {
1576 self.access_pattern.pop_front();
1577 }
1578 }
1579
1580 fn evict_lru(&mut self) {
1582 if let Some(lru_index) = self.access_pattern.front().copied() {
1583 self.cache.remove(&lru_index);
1584 }
1585 }
1586}
1587
1588impl<T: TensorElement> Drop for MemoryMappedStorage<T> {
1589 fn drop(&mut self) {
1590 if self.is_temporary {
1591 let _ = std::fs::remove_file(&self.file_path);
1593 }
1594 }
1595}
1596
1597impl<T: TensorElement> Clone for TensorStorage<T> {
1598 fn clone(&self) -> Self {
1599 match self {
1600 Self::InMemory(data) => Self::InMemory(Arc::clone(data)),
1601 Self::MemoryMapped(storage) => Self::MemoryMapped(Arc::clone(storage)),
1602 #[cfg(feature = "simd")]
1603 Self::Aligned(data) => Self::Aligned(Arc::clone(data)),
1604 #[cfg(feature = "simd")]
1605 Self::SimdOptimized(storage) => {
1606 storage.mark_shared();
1608 Self::SimdOptimized(Arc::clone(storage))
1609 }
1610 #[cfg(feature = "gpu")]
1611 Self::Device { buffer, host_cache } => Self::Device {
1612 buffer: Arc::clone(buffer),
1615 host_cache: Arc::clone(host_cache),
1616 },
1617 }
1618 }
1619}
1620
1621#[cfg(test)]
1622mod tests {
1623 use super::*;
1624
1625 #[test]
1626 fn test_in_memory_storage() {
1627 let data = vec![1.0f32, 2.0, 3.0, 4.0];
1628 let storage = TensorStorage::in_memory(data.clone());
1629
1630 assert_eq!(storage.len(), 4);
1631 assert!(!storage.is_empty());
1632 assert_eq!(storage.storage_type(), "in_memory");
1633
1634 assert_eq!(storage.get(0).expect("get(0) failed"), 1.0);
1635 assert_eq!(storage.get(3).expect("get(3) failed"), 4.0);
1636
1637 let slice = storage.get_slice(1, 2).expect("get_slice failed");
1638 assert_eq!(slice, vec![2.0, 3.0]);
1639 }
1640
1641 #[test]
1642 fn test_optimal_storage_selection() {
1643 let small_data = vec![1.0f32; 200];
1645 let small_storage =
1646 TensorStorage::create_optimal(small_data).expect("create_optimal failed");
1647
1648 #[cfg(feature = "simd")]
1649 {
1650 assert_eq!(small_storage.storage_type(), "in_memory");
1652 }
1653 #[cfg(not(feature = "simd"))]
1654 {
1655 assert_eq!(small_storage.storage_type(), "in_memory");
1657 }
1658 }
1659
1660 #[test]
1661 fn test_memory_usage_calculation() {
1662 let data = vec![1.0f32; 1000];
1663 let storage = TensorStorage::in_memory(data);
1664 let expected_size = 1000 * std::mem::size_of::<f32>();
1665 assert_eq!(storage.memory_usage(), expected_size);
1666 }
1667
1668 #[test]
1669 #[cfg(feature = "simd")]
1670 fn test_aligned_storage() {
1671 let data = vec![1.0f32, 2.0, 3.0, 4.0];
1672 let storage =
1673 TensorStorage::aligned(data.clone()).expect("aligned storage creation failed");
1674
1675 assert_eq!(storage.len(), 4);
1676 assert!(!storage.is_empty());
1677 assert_eq!(storage.storage_type(), "aligned_simd");
1678
1679 assert_eq!(storage.get(0).expect("get(0) failed"), 1.0);
1681 assert_eq!(storage.get(3).expect("get(3) failed"), 4.0);
1682
1683 let slice = storage.get_slice(1, 2).expect("get_slice failed");
1685 assert_eq!(slice, vec![2.0, 3.0]);
1686
1687 let vec = storage.to_vec().expect("to_vec failed");
1689 assert_eq!(vec, data);
1690 }
1691
1692 #[test]
1693 #[cfg(feature = "simd")]
1694 fn test_optimal_storage_selection_with_aligned() {
1695 let medium_data = vec![1.0f32; 2000]; let medium_storage = TensorStorage::create_optimal(medium_data)
1698 .expect("create_optimal for medium data failed");
1699 assert_eq!(medium_storage.storage_type(), "aligned_simd");
1700
1701 let small_data = vec![1.0f32; 100]; let small_storage = TensorStorage::create_optimal(small_data)
1704 .expect("create_optimal for small data failed");
1705 assert_eq!(small_storage.storage_type(), "in_memory");
1706 }
1707}