1use crate::Tensor;
9use std::collections::HashMap;
10use std::sync::{Arc, RwLock};
11use torsh_core::{device::DeviceType, dtype::TensorElement, error::Result};
12
13#[cfg(feature = "gpu")]
17pub struct GpuContext;
18
19#[cfg(feature = "gpu")]
20pub struct GpuKernel;
21
22#[cfg(feature = "gpu")]
23impl GpuContext {
24 pub fn new() -> Result<Self> {
25 Err(torsh_core::error::TorshError::InvalidArgument(
26 "GPU support temporarily unavailable".to_string(),
27 ))
28 }
29}
30
31#[cfg(feature = "gpu")]
32impl GpuKernel {
33 pub fn load(_context: &GpuContext, _name: &str) -> Result<Self> {
34 Err(torsh_core::error::TorshError::InvalidArgument(
35 "GPU support temporarily unavailable".to_string(),
36 ))
37 }
38
39 pub fn auto_tune(&mut self, _tuning_params: &[(String, f32)]) -> Result<()> {
40 Err(torsh_core::error::TorshError::InvalidArgument(
41 "GPU support temporarily unavailable".to_string(),
42 ))
43 }
44
45 pub fn enable_fusion(&mut self, _enable: bool) -> Result<()> {
46 Err(torsh_core::error::TorshError::InvalidArgument(
47 "GPU support temporarily unavailable".to_string(),
48 ))
49 }
50
51 pub fn enable_tensor_cores(&mut self, _enable: bool) -> Result<()> {
52 Err(torsh_core::error::TorshError::InvalidArgument(
53 "GPU support temporarily unavailable".to_string(),
54 ))
55 }
56
57 pub fn supports_tensor_cores(&self) -> bool {
58 false
59 }
60
61 pub fn execute<T>(&self, _input: &[T], _output: &mut [T]) -> Result<()> {
62 Err(torsh_core::error::TorshError::InvalidArgument(
63 "GPU support temporarily unavailable".to_string(),
64 ))
65 }
66}
67
68#[derive(Debug, Clone)]
70pub enum DeviceOptimization {
71 Cpu(CpuOptimization),
73 Gpu(GpuOptimization),
75 Metal(MetalOptimization),
77 WebGpu(WebGpuOptimization),
79}
80
81#[derive(Debug, Clone)]
83pub struct CpuOptimization {
84 pub use_simd: bool,
86 pub thread_count: Option<usize>,
88 pub cache_friendly: bool,
90 pub numa_aware: bool,
92}
93
94#[derive(Debug, Clone)]
96pub struct GpuOptimization {
97 pub use_pinned_memory: bool,
99 pub stream_count: u32,
101 pub mixed_precision: bool,
103 pub memory_pool_size: Option<usize>,
105
106 pub use_tensor_cores: bool,
109 pub auto_kernel_tuning: bool,
111 pub use_unified_memory: bool,
113 pub multi_gpu_strategy: MultiGpuStrategy,
115 pub backend_preference: Vec<GpuBackendType>,
117 pub memory_coalescing: bool,
119 pub kernel_fusion_level: u8,
121 pub dynamic_batching: bool,
123}
124
125#[derive(Debug, Clone)]
127pub enum MultiGpuStrategy {
128 Single,
130 DataParallel,
132 ModelParallel,
134 PipelineParallel,
136 Auto,
138}
139
140#[derive(Debug, Clone, PartialEq, Eq)]
142pub enum GpuBackendType {
143 Cuda,
145 Metal,
147 WebGpu,
149 Rocm,
151 OpenCl,
153}
154
155#[derive(Debug, Clone)]
157pub struct MetalOptimization {
158 pub use_mps: bool,
160 pub command_buffer_count: u32,
162 pub auto_memory_management: bool,
164}
165
166#[derive(Debug, Clone)]
168pub struct WebGpuOptimization {
169 pub use_compute_shaders: bool,
171 pub buffer_pool_size: Option<usize>,
173 pub pipeline_caching: bool,
175}
176
177#[derive(Debug)]
179pub struct OperationScheduler {
180 device_queues: HashMap<DeviceType, Vec<ScheduledOperation>>,
182 sync_state: HashMap<DeviceType, SyncState>,
184 operation_counter: Arc<RwLock<u64>>,
186}
187
188#[derive(Debug)]
190pub struct ScheduledOperation {
191 pub id: u64,
193 pub operation: OperationType,
195 pub priority: u8,
197 pub dependencies: Vec<DeviceType>,
199}
200
201#[derive(Debug)]
203pub enum OperationType {
204 Compute,
206 Transfer,
208 Synchronization,
210}
211
212#[derive(Debug)]
214pub struct SyncState {
215 pub last_operation: std::time::Instant,
217 pub pending_transfers: usize,
219 pub available: bool,
221}
222
223impl Default for CpuOptimization {
224 fn default() -> Self {
225 Self {
226 use_simd: true,
227 thread_count: None, cache_friendly: true,
229 numa_aware: true,
230 }
231 }
232}
233
234impl Default for GpuOptimization {
235 fn default() -> Self {
236 Self {
237 use_pinned_memory: true,
238 stream_count: 4,
239 mixed_precision: false,
240 memory_pool_size: Some(1024 * 1024 * 1024), use_tensor_cores: true, auto_kernel_tuning: true, use_unified_memory: true, multi_gpu_strategy: MultiGpuStrategy::Auto, backend_preference: vec![
248 GpuBackendType::Cuda, GpuBackendType::Metal, GpuBackendType::Rocm, GpuBackendType::WebGpu, GpuBackendType::OpenCl, ],
254 memory_coalescing: true, kernel_fusion_level: 2, dynamic_batching: true, }
258 }
259}
260
261impl Default for MetalOptimization {
262 fn default() -> Self {
263 Self {
264 use_mps: true,
265 command_buffer_count: 8,
266 auto_memory_management: true,
267 }
268 }
269}
270
271impl Default for WebGpuOptimization {
272 fn default() -> Self {
273 Self {
274 use_compute_shaders: true,
275 buffer_pool_size: Some(256 * 1024 * 1024), pipeline_caching: true,
277 }
278 }
279}
280
281impl<T: TensorElement + Copy> Tensor<T> {
282 pub fn to_device(&self, target_device: DeviceType) -> Result<Self> {
284 if self.device == target_device {
285 return Ok(self.clone());
286 }
287
288 let optimization = self.get_device_optimization(target_device);
290
291 match (self.device, target_device) {
293 (DeviceType::Cpu, DeviceType::Cuda(gpu_id)) => {
294 self.cpu_to_gpu_transfer(gpu_id as u32, optimization)
295 }
296 (DeviceType::Cuda(gpu_id), DeviceType::Cpu) => {
297 self.gpu_to_cpu_transfer(gpu_id as u32, optimization)
298 }
299 (DeviceType::Cpu, DeviceType::Metal(metal_id)) => {
300 self.cpu_to_metal_transfer(metal_id as u32, optimization)
301 }
302 (DeviceType::Metal(metal_id), DeviceType::Cpu) => {
303 self.metal_to_cpu_transfer(metal_id as u32, optimization)
304 }
305 _ => {
306 self.generic_device_transfer(target_device)
308 }
309 }
310 }
311
312 fn get_device_optimization(&self, device: DeviceType) -> DeviceOptimization {
314 match device {
315 DeviceType::Cpu => DeviceOptimization::Cpu(CpuOptimization::default()),
316 DeviceType::Cuda(_) => DeviceOptimization::Gpu(GpuOptimization::default()),
317 DeviceType::Metal(_) => DeviceOptimization::Metal(MetalOptimization::default()),
318 DeviceType::Wgpu(_) => DeviceOptimization::Gpu(GpuOptimization::default()),
319 }
320 }
321
322 fn cpu_to_gpu_transfer(&self, _gpu_id: u32, optimization: DeviceOptimization) -> Result<Self> {
324 let data = self.to_vec()?;
325
326 if let DeviceOptimization::Gpu(gpu_opt) = optimization {
328 if gpu_opt.use_pinned_memory {
329 self.transfer_with_pinned_memory(data, DeviceType::Cuda(_gpu_id as usize))
331 } else {
332 Self::from_data(
334 data,
335 self.shape().dims().to_vec(),
336 DeviceType::Cuda(_gpu_id as usize),
337 )
338 }
339 } else {
340 Self::from_data(
341 data,
342 self.shape().dims().to_vec(),
343 DeviceType::Cuda(_gpu_id as usize),
344 )
345 }
346 }
347
348 fn gpu_to_cpu_transfer(&self, _gpu_id: u32, optimization: DeviceOptimization) -> Result<Self> {
350 let data = self.to_vec()?;
351
352 if let DeviceOptimization::Cpu(cpu_opt) = optimization {
354 if cpu_opt.numa_aware {
355 self.transfer_with_numa_awareness(data, DeviceType::Cpu)
357 } else {
358 Self::from_data(data, self.shape().dims().to_vec(), DeviceType::Cpu)
360 }
361 } else {
362 Self::from_data(data, self.shape().dims().to_vec(), DeviceType::Cpu)
363 }
364 }
365
366 fn cpu_to_metal_transfer(
368 &self,
369 _metal_id: u32,
370 optimization: DeviceOptimization,
371 ) -> Result<Self> {
372 let data = self.to_vec()?;
373
374 if let DeviceOptimization::Metal(metal_opt) = optimization {
376 if metal_opt.use_mps {
377 self.transfer_with_mps(data, DeviceType::Metal(_metal_id as usize))
379 } else {
380 Self::from_data(
382 data,
383 self.shape().dims().to_vec(),
384 DeviceType::Metal(_metal_id as usize),
385 )
386 }
387 } else {
388 Self::from_data(
389 data,
390 self.shape().dims().to_vec(),
391 DeviceType::Metal(_metal_id as usize),
392 )
393 }
394 }
395
396 fn metal_to_cpu_transfer(
398 &self,
399 _metal_id: u32,
400 optimization: DeviceOptimization,
401 ) -> Result<Self> {
402 let data = self.to_vec()?;
403
404 if let DeviceOptimization::Cpu(cpu_opt) = optimization {
406 if cpu_opt.cache_friendly {
407 self.transfer_with_cache_optimization(data, DeviceType::Cpu)
409 } else {
410 Self::from_data(data, self.shape().dims().to_vec(), DeviceType::Cpu)
412 }
413 } else {
414 Self::from_data(data, self.shape().dims().to_vec(), DeviceType::Cpu)
415 }
416 }
417
418 fn generic_device_transfer(&self, target_device: DeviceType) -> Result<Self> {
420 let data = self.to_vec()?;
421 Self::from_data(data, self.shape().dims().to_vec(), target_device)
422 }
423
424 fn transfer_with_pinned_memory(&self, data: Vec<T>, target_device: DeviceType) -> Result<Self> {
426 Self::from_data(data, self.shape().dims().to_vec(), target_device)
428 }
429
430 fn transfer_with_numa_awareness(
432 &self,
433 data: Vec<T>,
434 target_device: DeviceType,
435 ) -> Result<Self> {
436 Self::from_data(data, self.shape().dims().to_vec(), target_device)
438 }
439
440 fn transfer_with_mps(&self, data: Vec<T>, target_device: DeviceType) -> Result<Self> {
442 Self::from_data(data, self.shape().dims().to_vec(), target_device)
444 }
445
446 fn transfer_with_cache_optimization(
448 &self,
449 data: Vec<T>,
450 target_device: DeviceType,
451 ) -> Result<Self> {
452 let optimized_data = self.optimize_for_cache(data)?;
454 Self::from_data(optimized_data, self.shape().dims().to_vec(), target_device)
455 }
456
457 fn optimize_for_cache(&self, data: Vec<T>) -> Result<Vec<T>> {
459 Ok(data)
461 }
462
463 pub fn synchronize_devices(&self, devices: &[DeviceType]) -> Result<()> {
465 for device in devices {
467 self.synchronize_device(*device)?;
468 }
469 Ok(())
470 }
471
472 fn synchronize_device(&self, _device: DeviceType) -> Result<()> {
474 Ok(())
476 }
477
478 pub fn can_transfer_efficiently(&self, target_device: DeviceType) -> bool {
480 match (self.device, target_device) {
481 (a, b) if a == b => true,
483 (DeviceType::Cpu, DeviceType::Cuda(_)) | (DeviceType::Cuda(_), DeviceType::Cpu) => true,
485 (DeviceType::Cpu, DeviceType::Metal(_)) | (DeviceType::Metal(_), DeviceType::Cpu) => {
487 true
488 }
489 _ => false,
491 }
492 }
493
494 pub fn get_transfer_strategy(&self, target_device: DeviceType) -> TransferStrategy {
496 match (self.device, target_device) {
497 (a, b) if a == b => TransferStrategy::NoTransfer,
498 (DeviceType::Cpu, DeviceType::Cuda(_)) => TransferStrategy::DirectTransfer,
499 (DeviceType::Cuda(_), DeviceType::Cpu) => TransferStrategy::DirectTransfer,
500 (DeviceType::Cpu, DeviceType::Metal(_)) => TransferStrategy::DirectTransfer,
501 (DeviceType::Metal(_), DeviceType::Cpu) => TransferStrategy::DirectTransfer,
502 _ => TransferStrategy::ThroughCpu,
503 }
504 }
505}
506
507#[derive(Debug, Clone, PartialEq)]
509pub enum TransferStrategy {
510 NoTransfer,
512 DirectTransfer,
514 ThroughCpu,
516}
517
518impl OperationScheduler {
519 pub fn new() -> Self {
521 Self {
522 device_queues: HashMap::new(),
523 sync_state: HashMap::new(),
524 operation_counter: Arc::new(RwLock::new(0)),
525 }
526 }
527
528 pub fn schedule_operation(
530 &mut self,
531 device: DeviceType,
532 operation: OperationType,
533 priority: u8,
534 dependencies: Vec<DeviceType>,
535 ) -> Result<u64> {
536 let mut counter = self
538 .operation_counter
539 .write()
540 .expect("lock should not be poisoned");
541 *counter += 1;
542 let op_id = *counter;
543 drop(counter);
544
545 let scheduled_op = ScheduledOperation {
547 id: op_id,
548 operation,
549 priority,
550 dependencies,
551 };
552
553 self.device_queues
555 .entry(device)
556 .or_default()
557 .push(scheduled_op);
558
559 if let Some(queue) = self.device_queues.get_mut(&device) {
561 queue.sort_by(|a, b| b.priority.cmp(&a.priority));
562 }
563
564 self.sync_state.entry(device).or_insert_with(|| SyncState {
566 last_operation: std::time::Instant::now(),
567 pending_transfers: 0,
568 available: true,
569 });
570
571 Ok(op_id)
572 }
573
574 pub fn execute_next_operation(&mut self, device: DeviceType) -> Result<Option<u64>> {
576 let op = if let Some(queue) = self.device_queues.get_mut(&device) {
578 if queue.is_empty() {
579 None
580 } else {
581 Some(queue.remove(0)) }
583 } else {
584 None
585 };
586
587 if let Some(op) = op {
588 let dependencies_satisfied = self.check_dependencies(&op.dependencies)?;
590
591 if dependencies_satisfied {
592 self.execute_operation(&op)?;
594
595 if let Some(sync_state) = self.sync_state.get_mut(&device) {
597 sync_state.last_operation = std::time::Instant::now();
598 }
599
600 Ok(Some(op.id))
601 } else {
602 if let Some(queue) = self.device_queues.get_mut(&device) {
604 queue.insert(0, op);
605 }
606 Ok(None)
607 }
608 } else {
609 Ok(None)
610 }
611 }
612
613 fn check_dependencies(&self, dependencies: &[DeviceType]) -> Result<bool> {
615 for &dep_device in dependencies {
616 if let Some(sync_state) = self.sync_state.get(&dep_device) {
617 if !sync_state.available {
618 return Ok(false);
619 }
620 }
621 }
622 Ok(true)
623 }
624
625 fn execute_operation(&self, _operation: &ScheduledOperation) -> Result<()> {
627 std::thread::sleep(std::time::Duration::from_millis(1));
629 Ok(())
630 }
631
632 pub fn get_queue_length(&self, device: DeviceType) -> usize {
634 self.device_queues
635 .get(&device)
636 .map_or(0, |queue| queue.len())
637 }
638
639 pub fn clear_device_queue(&mut self, device: DeviceType) {
641 self.device_queues.remove(&device);
642 }
643}
644
645impl Default for OperationScheduler {
646 fn default() -> Self {
647 Self::new()
648 }
649}
650
651static GLOBAL_SCHEDULER: parking_lot::Mutex<Option<OperationScheduler>> =
653 parking_lot::Mutex::new(None);
654
655pub fn get_global_scheduler() -> parking_lot::MutexGuard<'static, Option<OperationScheduler>> {
657 let mut guard = GLOBAL_SCHEDULER.lock();
658 if guard.is_none() {
659 *guard = Some(OperationScheduler::new());
660 }
661 guard
662}
663
664pub fn initialize_global_scheduler() -> Result<()> {
666 let mut guard = GLOBAL_SCHEDULER.lock();
667 *guard = Some(OperationScheduler::new());
668 Ok(())
669}
670
671#[cfg(feature = "gpu")]
673impl<T: TensorElement + Copy + Default> Tensor<T> {
674 pub fn execute_gpu_kernel(&self, kernel_name: &str, _params: Vec<T>) -> Result<Self> {
676 let gpu_opt = match self.get_device_optimization(self.device) {
677 DeviceOptimization::Gpu(opt) => opt,
678 _ => {
679 return Err(torsh_core::error::TorshError::InvalidArgument(
680 "GPU kernel execution requires GPU device".to_string(),
681 ))
682 }
683 };
684
685 let gpu_context = self.create_optimal_gpu_context(&gpu_opt)?;
687
688 let input_buffer = self.create_gpu_buffer(&gpu_context, &gpu_opt)?;
690
691 let kernel = self.select_optimal_kernel(&gpu_context, kernel_name, &gpu_opt)?;
693
694 let mut output_buffer = vec![T::default(); input_buffer.len()];
696 kernel.execute(&input_buffer, &mut output_buffer)?;
697
698 self.gpu_buffer_to_tensor(output_buffer, &gpu_context, &gpu_opt)
700 }
701
702 #[allow(dead_code)]
705 fn create_optimal_gpu_context(&self, _gpu_opt: &GpuOptimization) -> Result<GpuContext> {
706 Err(torsh_core::error::TorshError::InvalidArgument(
738 "GPU backend creation temporarily disabled".to_string(),
739 ))
740 }
741
742 #[allow(dead_code)]
745 fn create_gpu_buffer(&self, _context: &GpuContext, _gpu_opt: &GpuOptimization) -> Result<Vec<T>>
746 where
747 T: Copy,
748 {
749 let data = self.to_vec()?;
750 Ok(data)
762 }
763
764 fn select_optimal_kernel(
766 &self,
767 context: &GpuContext,
768 kernel_name: &str,
769 gpu_opt: &GpuOptimization,
770 ) -> Result<GpuKernel> {
771 let mut kernel = GpuKernel::load(context, kernel_name).map_err(|e| {
772 torsh_core::error::TorshError::InvalidArgument(format!(
773 "Failed to load kernel '{}': {}",
774 kernel_name, e
775 ))
776 })?;
777
778 if gpu_opt.auto_kernel_tuning {
779 kernel.auto_tune(&[])?;
782 }
783
784 if gpu_opt.use_tensor_cores && kernel.supports_tensor_cores() {
785 kernel.enable_tensor_cores(true)?;
787 }
788
789 if gpu_opt.kernel_fusion_level > 0 {
790 kernel.enable_fusion(gpu_opt.kernel_fusion_level > 0)?;
792 }
793
794 Ok(kernel)
795 }
796
797 #[allow(dead_code)]
800 fn gpu_buffer_to_tensor(
801 &self,
802 buffer: Vec<T>, _context: &GpuContext,
804 _gpu_opt: &GpuOptimization,
805 ) -> Result<Self>
806 where
807 T: Copy,
808 {
809 Self::from_data(buffer, self.shape().dims().to_vec(), self.device)
819 }
820
821 pub fn distribute_multi_gpu(
823 &self,
824 gpu_count: usize,
825 strategy: Option<MultiGpuStrategy>,
826 ) -> Result<Vec<Self>> {
827 if gpu_count <= 1 {
828 return Ok(vec![self.clone()]);
829 }
830
831 let strategy = strategy.unwrap_or(MultiGpuStrategy::Auto);
832 let effective_strategy = match strategy {
833 MultiGpuStrategy::Auto => self.select_optimal_multi_gpu_strategy(gpu_count),
834 s => s,
835 };
836
837 match effective_strategy {
838 MultiGpuStrategy::DataParallel => self.data_parallel_distribution(gpu_count),
839 MultiGpuStrategy::ModelParallel => self.model_parallel_distribution(gpu_count),
840 MultiGpuStrategy::PipelineParallel => self.pipeline_parallel_distribution(gpu_count),
841 _ => Ok(vec![self.clone()]), }
843 }
844
845 fn select_optimal_multi_gpu_strategy(&self, gpu_count: usize) -> MultiGpuStrategy {
847 let _total_elements = self.numel();
848 let shape = self.shape();
849 let dims = shape.dims();
850
851 if dims.len() > 0 && dims[0] >= gpu_count * 4 {
853 return MultiGpuStrategy::DataParallel;
854 }
855
856 if dims.len() > 1 && dims.iter().skip(1).product::<usize>() > 1024 * 1024 {
858 return MultiGpuStrategy::ModelParallel;
859 }
860
861 if dims.len() > 3 {
863 return MultiGpuStrategy::PipelineParallel;
864 }
865
866 MultiGpuStrategy::DataParallel
868 }
869
870 fn data_parallel_distribution(&self, gpu_count: usize) -> Result<Vec<Self>> {
872 let shape = self.shape();
873 let dims = shape.dims();
874 if dims.is_empty() {
875 return Err(torsh_core::error::TorshError::InvalidArgument(
876 "Cannot distribute scalar tensor".to_string(),
877 ));
878 }
879
880 let batch_size = dims[0];
881 let chunk_size = (batch_size + gpu_count - 1) / gpu_count; let mut distributed_tensors = Vec::with_capacity(gpu_count);
884 let data = self.to_vec()?;
885 let elements_per_batch = dims.iter().skip(1).product::<usize>();
886
887 for gpu_id in 0..gpu_count {
888 let start_batch = gpu_id * chunk_size;
889 let end_batch = ((gpu_id + 1) * chunk_size).min(batch_size);
890
891 if start_batch >= batch_size {
892 break; }
894
895 let start_idx = start_batch * elements_per_batch;
896 let end_idx = end_batch * elements_per_batch;
897 let chunk_data = data[start_idx..end_idx].to_vec();
898
899 let mut chunk_dims = dims.to_vec();
900 chunk_dims[0] = end_batch - start_batch;
901
902 let chunk_tensor = Self::from_data(chunk_data, chunk_dims, DeviceType::Cuda(gpu_id))?;
903
904 distributed_tensors.push(chunk_tensor);
905 }
906
907 Ok(distributed_tensors)
908 }
909
910 fn model_parallel_distribution(&self, gpu_count: usize) -> Result<Vec<Self>> {
912 let shape = self.shape();
913 let dims = shape.dims();
914 if dims.len() < 2 {
915 return Err(torsh_core::error::TorshError::InvalidArgument(
916 "Model parallel requires at least 2D tensor".to_string(),
917 ));
918 }
919
920 let feature_dim = dims.len() - 1;
922 let feature_size = dims[feature_dim];
923 let chunk_size = (feature_size + gpu_count - 1) / gpu_count;
924
925 let mut distributed_tensors = Vec::with_capacity(gpu_count);
926 let _data = self.to_vec()?;
927
928 for gpu_id in 0..gpu_count {
929 let start_feature = gpu_id * chunk_size;
930 let end_feature = ((gpu_id + 1) * chunk_size).min(feature_size);
931
932 if start_feature >= feature_size {
933 break;
934 }
935
936 let mut chunk_dims = dims.to_vec();
939 chunk_dims[feature_dim] = end_feature - start_feature;
940
941 let chunk_size_total: usize = chunk_dims.iter().product();
943 let chunk_data = vec![T::default(); chunk_size_total];
944
945 let chunk_tensor = Self::from_data(chunk_data, chunk_dims, DeviceType::Cuda(gpu_id))?;
946
947 distributed_tensors.push(chunk_tensor);
948 }
949
950 Ok(distributed_tensors)
951 }
952
953 fn pipeline_parallel_distribution(&self, gpu_count: usize) -> Result<Vec<Self>> {
955 let mut distributed_tensors = Vec::with_capacity(gpu_count);
958
959 for gpu_id in 0..gpu_count {
960 let pipeline_tensor = Self::from_data(
961 self.to_vec()?,
962 self.shape().dims().to_vec(),
963 DeviceType::Cuda(gpu_id),
964 )?;
965 distributed_tensors.push(pipeline_tensor);
966 }
967
968 Ok(distributed_tensors)
969 }
970
971 #[allow(dead_code)]
974 pub fn enable_mixed_precision(
975 &mut self,
976 _precision: i32, ) -> Result<()> {
978 Err(torsh_core::error::TorshError::InvalidArgument(
988 "Mixed precision temporarily disabled".to_string(),
989 ))
990 }
991}
992
993#[cfg(test)]
994mod tests {
995 use super::*;
996 use crate::Tensor;
997
998 #[test]
999 fn test_device_transfer() {
1000 let tensor = Tensor::from_data(vec![1.0f32, 2.0, 3.0, 4.0], vec![2, 2], DeviceType::Cpu)
1001 .expect("tensor creation should succeed");
1002
1003 let same_device = tensor
1005 .to_device(DeviceType::Cpu)
1006 .expect("device transfer should succeed");
1007 assert_eq!(same_device.device(), DeviceType::Cpu);
1008
1009 assert_eq!(
1011 tensor.get_transfer_strategy(DeviceType::Cpu),
1012 TransferStrategy::NoTransfer
1013 );
1014 assert_eq!(
1015 tensor.get_transfer_strategy(DeviceType::Cuda(0)),
1016 TransferStrategy::DirectTransfer
1017 );
1018 }
1019
1020 #[test]
1021 fn test_operation_scheduler() {
1022 let mut scheduler = OperationScheduler::new();
1023
1024 let op1 = scheduler
1026 .schedule_operation(DeviceType::Cpu, OperationType::Compute, 5, vec![])
1027 .expect("operation should succeed");
1028
1029 let op2 = scheduler
1030 .schedule_operation(DeviceType::Cpu, OperationType::Compute, 10, vec![])
1031 .expect("operation should succeed");
1032
1033 assert_eq!(
1035 scheduler
1036 .execute_next_operation(DeviceType::Cpu)
1037 .expect("operation execution should succeed"),
1038 Some(op2)
1039 );
1040 assert_eq!(
1041 scheduler
1042 .execute_next_operation(DeviceType::Cpu)
1043 .expect("operation execution should succeed"),
1044 Some(op1)
1045 );
1046 }
1047
1048 #[test]
1049 fn test_transfer_efficiency() {
1050 let tensor = Tensor::from_data(vec![1.0f32, 2.0, 3.0, 4.0], vec![2, 2], DeviceType::Cpu)
1051 .expect("tensor creation should succeed");
1052
1053 assert!(tensor.can_transfer_efficiently(DeviceType::Cpu));
1055
1056 assert!(tensor.can_transfer_efficiently(DeviceType::Cuda(0)));
1058
1059 assert!(tensor.can_transfer_efficiently(DeviceType::Metal(0)));
1061 }
1062
1063 #[test]
1064 fn test_device_optimization_defaults() {
1065 let cpu_opt = CpuOptimization::default();
1066 assert!(cpu_opt.use_simd);
1067 assert!(cpu_opt.cache_friendly);
1068 assert!(cpu_opt.numa_aware);
1069
1070 let gpu_opt = GpuOptimization::default();
1071 assert!(gpu_opt.use_pinned_memory);
1072 assert_eq!(gpu_opt.stream_count, 4);
1073 assert!(!gpu_opt.mixed_precision);
1074 }
1075
1076 #[test]
1077 fn test_global_scheduler() {
1078 initialize_global_scheduler().expect("scheduler initialization should succeed");
1079
1080 {
1081 let mut scheduler = get_global_scheduler();
1082 let scheduler = scheduler
1083 .as_mut()
1084 .expect("mutable reference should be available");
1085
1086 let op_id = scheduler
1087 .schedule_operation(DeviceType::Cpu, OperationType::Compute, 5, vec![])
1088 .expect("scheduler initialization should succeed");
1089
1090 assert_eq!(scheduler.get_queue_length(DeviceType::Cpu), 1);
1091 assert_eq!(
1092 scheduler
1093 .execute_next_operation(DeviceType::Cpu)
1094 .expect("operation execution should succeed"),
1095 Some(op_id)
1096 );
1097 }
1098 }
1099}