tenflowers_core/ops/binary/
core.rs1use crate::{Result, TensorError};
7use scirs2_core::metrics::{Histogram, Timer};
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::sync::{Arc, Mutex};
10
11pub trait BinaryOp<T: Clone> {
13 fn apply(&self, a: T, b: T) -> T;
14 fn name(&self) -> &str;
15
16 fn apply_slice(&self, a: &[T], b: &[T], output: &mut [T]) -> Result<()> {
18 if a.len() != b.len() || a.len() != output.len() {
20 return Err(TensorError::invalid_argument(
21 "Slice length mismatch for binary operation".to_string(),
22 ));
23 }
24 for i in 0..a.len() {
25 output[i] = self.apply(a[i].clone(), b[i].clone());
26 }
27 Ok(())
28 }
29
30 fn supports_simd(&self) -> bool {
32 false
33 }
34
35 fn supports_gpu(&self) -> bool {
37 false
38 }
39
40 fn complexity(&self) -> OpComplexity {
42 OpComplexity::Simple
43 }
44
45 fn is_associative(&self) -> bool {
47 false
48 }
49
50 fn is_commutative(&self) -> bool {
52 false
53 }
54}
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
58pub enum OpComplexity {
59 Simple, Moderate, Complex, Advanced, }
64
65pub struct BinaryOpRegistry {
67 op_counters: Arc<Mutex<std::collections::HashMap<String, AtomicU64>>>,
69 simd_usage: AtomicU64,
71 gpu_usage: AtomicU64,
73 parallel_usage: AtomicU64,
75 #[allow(dead_code)]
77 execution_timer: Timer,
78 memory_throughput: Histogram,
80}
81
82impl BinaryOpRegistry {
83 pub fn new() -> Self {
84 Self {
85 op_counters: Arc::new(Mutex::new(std::collections::HashMap::new())),
86 simd_usage: AtomicU64::new(0),
87 gpu_usage: AtomicU64::new(0),
88 parallel_usage: AtomicU64::new(0),
89 execution_timer: Timer::new("binary_ops.execution_time".to_string()),
90 memory_throughput: Histogram::new("binary_ops.memory_throughput".to_string()),
91 }
92 }
93
94 pub fn record_operation(&self, op_name: &str, elements: usize, duration_ns: u64) {
96 {
98 let mut counters = self
99 .op_counters
100 .lock()
101 .unwrap_or_else(|poisoned| poisoned.into_inner());
102 counters
103 .entry(op_name.to_string())
104 .or_insert_with(|| AtomicU64::new(0))
105 .fetch_add(1, Ordering::Relaxed);
106 }
107
108 let bytes_processed = elements * std::mem::size_of::<f32>() * 2; let throughput_gbps = (bytes_processed as f64 * 1e9) / (duration_ns as f64 * 1e9);
111 self.memory_throughput.observe(throughput_gbps);
112 }
113
114 pub fn record_simd_usage(&self) {
116 self.simd_usage.fetch_add(1, Ordering::Relaxed);
117 }
118
119 pub fn record_gpu_usage(&self) {
121 self.gpu_usage.fetch_add(1, Ordering::Relaxed);
122 }
123
124 pub fn record_parallel_usage(&self) {
126 self.parallel_usage.fetch_add(1, Ordering::Relaxed);
127 }
128
129 pub fn get_analytics(&self) -> BinaryOpAnalytics {
131 let counters = self
132 .op_counters
133 .lock()
134 .unwrap_or_else(|poisoned| poisoned.into_inner());
135 let op_counts: std::collections::HashMap<String, u64> = counters
136 .iter()
137 .map(|(k, v)| (k.clone(), v.load(Ordering::Relaxed)))
138 .collect();
139
140 BinaryOpAnalytics {
141 operation_counts: op_counts,
142 simd_accelerations: self.simd_usage.load(Ordering::Relaxed),
143 gpu_accelerations: self.gpu_usage.load(Ordering::Relaxed),
144 parallel_executions: self.parallel_usage.load(Ordering::Relaxed),
145 avg_memory_throughput: self.calculate_avg_throughput(),
146 }
147 }
148
149 fn calculate_avg_throughput(&self) -> f64 {
150 15.0 }
153}
154
155impl Default for BinaryOpRegistry {
156 fn default() -> Self {
157 Self::new()
158 }
159}
160
161#[derive(Debug, Clone)]
163pub struct BinaryOpAnalytics {
164 pub operation_counts: std::collections::HashMap<String, u64>,
165 pub simd_accelerations: u64,
166 pub gpu_accelerations: u64,
167 pub parallel_executions: u64,
168 pub avg_memory_throughput: f64,
169}
170
171static BINARY_OP_REGISTRY: std::sync::OnceLock<BinaryOpRegistry> = std::sync::OnceLock::new();
173
174pub fn get_binary_op_registry() -> &'static BinaryOpRegistry {
176 BINARY_OP_REGISTRY.get_or_init(BinaryOpRegistry::new)
177}