qualia_core_db/
csd_storage.rs

1//! NVMe Computational Storage (CSD) Implementation
2//!
3//! This module provides computational storage pushdown using NVMe CSD (Computational Storage Device).
4//! Designed for hardware-accelerated mathematical computations and query processing.
5
6use crate::q_hash;
7use std::collections::HashMap;
8use std::fs::OpenOptions;
9#[cfg(unix)]
10use std::os::unix::io::AsRawFd;
11
12use serde::{Deserialize, Serialize};
13use std::path::Path;
14
15/// CSD Manager for computational storage operations
16pub struct CsdManager {
17    devices: HashMap<String, CsdDevice>,
18    functions: HashMap<String, CsdFunction>,
19    scheduler: CsdScheduler,
20    performance_monitor: CsdPerformanceMonitor,
21}
22
23/// Compact, zero-heap view of a discovered CSD device.
24#[derive(Debug, Clone, Copy, PartialEq)]
25pub struct CsdDeviceHandle {
26    pub device_id_hash: u64,
27    pub max_concurrent_operations: u32,
28    pub max_data_size: u64,
29    pub memory_size: u64,
30    pub compute_units: u32,
31}
32
33/// Compact, zero-heap view of a registered computational function.
34#[derive(Debug, Clone, Copy, PartialEq)]
35pub struct CsdFunctionHandle {
36    pub function_id_hash: u64,
37    pub operation_tag: u8,
38    pub parameter_count: u16,
39    pub bytecode_len: u32,
40}
41
42/// CSD device information
43#[derive(Debug, Clone)]
44pub struct CsdDevice {
45    pub device_id: String,
46    pub device_path: String,
47    pub capabilities: CsdCapabilities,
48    pub supported_functions: Vec<String>,
49    pub device_stats: CsdDeviceStats,
50}
51
52/// CSD device capabilities
53#[derive(Debug, Clone, Serialize, Deserialize)]
54pub struct CsdCapabilities {
55    pub max_concurrent_operations: u32,
56    pub max_data_size: u64,
57    pub supported_operations: Vec<CsdOperationType>,
58    pub memory_size: u64,
59    pub compute_units: u32,
60    pub clock_speed: f64,
61}
62
63/// CSD operations supported by device
64#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
65pub enum CsdOperationType {
66    MatrixMultiply,
67    VectorDotProduct,
68    Convolution,
69    Filter,
70    Aggregate,
71    Sort,
72    Search,
73    Custom(String),
74}
75
76/// CSD function for computational operations
77#[derive(Debug, Clone)]
78pub struct CsdFunction {
79    pub function_id: String,
80    pub operation: CsdOperationType,
81    pub parameters: Vec<FunctionParameter>,
82    pub bytecode: Vec<u8>,
83    pub performance_profile: PerformanceProfile,
84}
85
86/// Function parameter definition
87#[derive(Debug, Clone, Serialize, Deserialize)]
88pub struct FunctionParameter {
89    pub name: String,
90    pub param_type: ParameterType,
91    pub size: u64,
92    pub is_input: bool,
93    pub is_output: bool,
94}
95
96/// Parameter types
97#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
98pub enum ParameterType {
99    Matrix,
100    Vector,
101    Scalar,
102    Tensor,
103    Buffer,
104}
105
106/// Performance profile for functions
107#[derive(Debug, Clone)]
108pub struct PerformanceProfile {
109    pub expected_execution_time: f64,
110    pub memory_usage: u64,
111    pub compute_intensity: f64,
112    pub data_intensity: f64,
113}
114
115/// CSD device statistics
116#[derive(Debug, Clone)]
117pub struct CsdDeviceStats {
118    pub operations_completed: u64,
119    pub total_execution_time: u64,
120    pub average_execution_time: f64,
121    pub data_processed: u64,
122    pub error_count: u64,
123    pub utilization: f64,
124}
125
126/// CSD scheduler for operation management
127pub struct CsdScheduler {
128    pending_operations: Vec<CsdOperationRequest>,
129    running_operations: HashMap<u64, CsdRunningOperation>,
130    completion_queue: Vec<CsdCompletion>,
131    scheduling_policy: SchedulingPolicy,
132}
133
134/// CSD operation request to be executed
135#[derive(Debug, Clone, Serialize, Deserialize)]
136pub struct CsdOperationRequest {
137    pub operation_id: u64,
138    pub function_id: String,
139    pub device_id: String,
140    pub inputs: Vec<OperationInput>,
141    pub outputs: Vec<OperationOutput>,
142    pub priority: OperationPriority,
143    pub deadline: Option<u64>,
144}
145
146/// Operation input
147#[derive(Debug, Clone, Serialize, Deserialize)]
148pub struct OperationInput {
149    pub name: String,
150    pub data: Vec<u8>,
151    pub location: DataLocation,
152}
153
154/// Operation output
155#[derive(Debug, Clone, Serialize, Deserialize)]
156pub struct OperationOutput {
157    pub name: String,
158    pub size: u64,
159    pub location: DataLocation,
160}
161
162/// Data location for operations
163#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
164pub enum DataLocation {
165    HostMemory,
166    DeviceMemory,
167    PersistentStorage,
168}
169
170/// Operation priority
171#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
172pub enum OperationPriority {
173    Low,
174    Normal,
175    High,
176    Critical,
177}
178
179/// Running operation tracking
180#[derive(Debug, Clone)]
181pub struct CsdRunningOperation {
182    pub operation_id: u64,
183    pub device_id: String,
184    pub start_time: u64,
185    pub progress: f64,
186}
187
188/// Operation completion result
189#[derive(Debug, Clone)]
190pub struct CsdCompletion {
191    pub operation_id: u64,
192    pub status: CompletionStatus,
193    pub execution_time: u64,
194    pub outputs: Vec<OperationOutput>,
195    pub error_message: Option<String>,
196}
197
198/// Completion status
199#[derive(Debug, Clone, PartialEq)]
200pub enum CompletionStatus {
201    Success,
202    Error,
203    Timeout,
204    Cancelled,
205}
206
207/// Scheduling policies
208#[derive(Debug, Clone)]
209pub enum SchedulingPolicy {
210    /// First-In-First-Out
211    Fifo,
212    /// Priority-based scheduling
213    Priority,
214    /// Shortest-Job-First
215    ShortestJobFirst,
216    /// Deadline-Driven Scheduling
217    Deadline,
218    /// Load-Balanced Scheduling
219    LoadBalanced,
220}
221
222/// CSD performance monitor
223pub struct CsdPerformanceMonitor {
224    device_metrics: HashMap<String, CsdDeviceMetrics>,
225    function_metrics: HashMap<String, CsdFunctionMetrics>,
226    global_metrics: CsdGlobalMetrics,
227}
228
229/// Device performance metrics
230#[derive(Debug, Clone)]
231pub struct CsdDeviceMetrics {
232    pub device_id: String,
233    pub utilization: f64,
234    pub throughput: f64,
235    pub latency: f64,
236    pub error_rate: f64,
237    pub power_consumption: f64,
238}
239
240/// Function performance metrics
241#[derive(Debug, Clone)]
242pub struct CsdFunctionMetrics {
243    pub function_id: String,
244    pub execution_count: u64,
245    pub total_execution_time: u64,
246    pub average_execution_time: f64,
247    pub success_rate: f64,
248    pub data_throughput: f64,
249}
250
251/// Global performance metrics
252#[derive(Debug, Clone)]
253pub struct CsdGlobalMetrics {
254    pub total_operations: u64,
255    pub total_execution_time: u64,
256    pub average_execution_time: f64,
257    pub total_data_processed: u64,
258    pub overall_throughput: f64,
259    pub system_utilization: f64,
260}
261
262/// Mathematical computation builder
263pub struct MathComputationBuilder {
264    operations: Vec<CsdOperationRequest>,
265    data_dependencies: HashMap<String, Vec<String>>,
266    execution_plan: ExecutionPlan,
267}
268
269/// Execution plan for computations
270#[derive(Debug, Clone)]
271pub struct ExecutionPlan {
272    pub stages: Vec<ExecutionStage>,
273    pub parallel_groups: Vec<Vec<String>>,
274    pub estimated_time: f64,
275    pub resource_requirements: ResourceRequirements,
276}
277
278/// Execution stage
279#[derive(Debug, Clone)]
280pub struct ExecutionStage {
281    pub stage_id: u32,
282    pub operations: Vec<String>,
283    pub dependencies: Vec<String>,
284    pub estimated_time: f64,
285}
286
287/// Resource requirements
288#[derive(Debug, Clone)]
289pub struct ResourceRequirements {
290    pub memory_usage: u64,
291    pub compute_units: u32,
292    pub bandwidth: f64,
293}
294
295impl CsdManager {
296    /// Create new CSD manager
297    pub fn new() -> Self {
298        Self {
299            devices: HashMap::new(),
300            functions: HashMap::new(),
301            scheduler: CsdScheduler::new(),
302            performance_monitor: CsdPerformanceMonitor::new(),
303        }
304    }
305
306    pub fn register_device(&mut self, device: CsdDevice) -> Result<(), CsdError> {
307        self.devices.insert(device.device_id.clone(), device);
308        Ok(())
309    }
310
311    /// Discover CSD devices
312    pub fn discover_devices(&mut self) -> Result<Vec<String>, CsdError> {
313        let mut discovered_devices = Vec::new();
314
315        // Scan for CSD devices
316        for i in 0..16 {
317            let device_path = format!("/dev/nvme{}", i);
318            if Path::new(&device_path).exists() {
319                if let Ok(device) = self.probe_device(&device_path) {
320                    discovered_devices.push(device.device_id.clone());
321                    self.devices.insert(device.device_id.clone(), device);
322                }
323            }
324        }
325
326        Ok(discovered_devices)
327    }
328
329    /// Discover devices into a caller-owned fixed buffer.
330    pub fn discover_devices_into(
331        &mut self,
332        out: &mut [CsdDeviceHandle],
333    ) -> Result<usize, CsdError> {
334        let mut written = 0;
335
336        for i in 0..16 {
337            let device_path = format!("/dev/nvme{}", i);
338            if Path::new(&device_path).exists() {
339                if let Ok(device) = self.probe_device(&device_path) {
340                    if written >= out.len() {
341                        return Err(CsdError::BufferTooSmall(
342                            "device discovery output buffer exhausted".to_string(),
343                        ));
344                    }
345                    out[written] = Self::device_handle(&device);
346                    written += 1;
347                    self.devices.insert(device.device_id.clone(), device);
348                }
349            }
350        }
351
352        Ok(written)
353    }
354
355    /// Probe CSD device
356    fn probe_device(&self, device_path: &str) -> Result<CsdDevice, CsdError> {
357        let _device_file = OpenOptions::new()
358            .read(true)
359            .write(true)
360            .open(device_path)
361            .map_err(|e| CsdError::DeviceOpen(e.to_string()))?;
362
363        // Get device information using ioctl
364        let device_id = format!("csd-{}", device_path);
365
366        // For now, use reasonable defaults
367        let capabilities = CsdCapabilities {
368            max_concurrent_operations: 16,
369            max_data_size: 1024 * 1024 * 1024, // 1GB
370            supported_operations: vec![
371                CsdOperationType::MatrixMultiply,
372                CsdOperationType::VectorDotProduct,
373                CsdOperationType::Convolution,
374                CsdOperationType::Filter,
375                CsdOperationType::Aggregate,
376            ],
377            memory_size: 8 * 1024 * 1024 * 1024, // 8GB
378            compute_units: 64,
379            clock_speed: 1.5, // 1.5 GHz
380        };
381
382        let device = CsdDevice {
383            device_id: device_id.clone(),
384            device_path: device_path.to_string(),
385            capabilities,
386            supported_functions: vec![], // Will be populated later
387            device_stats: CsdDeviceStats {
388                operations_completed: 0,
389                total_execution_time: 0,
390                average_execution_time: 0.0,
391                data_processed: 0,
392                error_count: 0,
393                utilization: 0.0,
394            },
395        };
396
397        Ok(device)
398    }
399
400    /// Register computational function
401    pub fn register_function(&mut self, function: CsdFunction) -> Result<(), CsdError> {
402        // Validate function
403        self.validate_function(&function)?;
404
405        // Store function
406        self.functions
407            .insert(function.function_id.clone(), function);
408
409        Ok(())
410    }
411
412    /// Execute computational operation
413    pub fn execute_operation(&mut self, operation: CsdOperationRequest) -> Result<u64, CsdError> {
414        // Validate operation
415        self.validate_operation(&operation)?;
416
417        // Schedule operation
418        self.scheduler.schedule_operation(operation.clone())?;
419
420        Ok(operation.operation_id)
421    }
422
423    /// Execute matrix multiplication
424    pub fn matrix_multiply(
425        &mut self,
426        device_id: &str,
427        a: &[f32],
428        b: &[f32],
429        dimensions: (usize, usize, usize),
430    ) -> Result<Vec<f32>, CsdError> {
431        let mut result = vec![0.0f32; dimensions.0 * dimensions.2];
432        let written = self.matrix_multiply_into(device_id, a, b, dimensions, &mut result)?;
433        result.truncate(written);
434        Ok(result)
435    }
436
437    /// Execute matrix multiplication into a caller-owned output buffer.
438    pub fn matrix_multiply_into(
439        &mut self,
440        device_id: &str,
441        a: &[f32],
442        b: &[f32],
443        dimensions: (usize, usize, usize),
444        out: &mut [f32],
445    ) -> Result<usize, CsdError> {
446        self.ensure_device_exists(device_id)?;
447        let (rows_a, shared, cols_b) = dimensions;
448        if a.len() != rows_a * shared {
449            return Err(CsdError::InvalidOperation(
450                "matrix A dimensions do not match input length".to_string(),
451            ));
452        }
453        if b.len() != shared * cols_b {
454            return Err(CsdError::InvalidOperation(
455                "matrix B dimensions do not match input length".to_string(),
456            ));
457        }
458        let required = rows_a * cols_b;
459        if out.len() < required {
460            return Err(CsdError::BufferTooSmall(
461                "matrix multiply output buffer too small".to_string(),
462            ));
463        }
464
465        let operation_id = self.generate_operation_id();
466        let operation = CsdOperationRequest {
467            operation_id,
468            function_id: "matrix_multiply".to_string(),
469            device_id: device_id.to_string(),
470            inputs: vec![
471                OperationInput {
472                    name: "matrix_a".to_string(),
473                    data: self.f32_slice_to_bytes(a),
474                    location: DataLocation::HostMemory,
475                },
476                OperationInput {
477                    name: "matrix_b".to_string(),
478                    data: self.f32_slice_to_bytes(b),
479                    location: DataLocation::HostMemory,
480                },
481                OperationInput {
482                    name: "dimensions".to_string(),
483                    data: self.serialize_dimensions(dimensions),
484                    location: DataLocation::HostMemory,
485                },
486            ],
487            outputs: vec![OperationOutput {
488                name: "result".to_string(),
489                size: (required * 4) as u64,
490                location: DataLocation::HostMemory,
491            }],
492            priority: OperationPriority::Normal,
493            deadline: None,
494        };
495        if self.execute_operation(operation).is_ok() {
496            if let Ok(completion) = self.wait_for_completion(operation_id) {
497                if let Some(output) = completion.outputs.first() {
498                    let floats = self.bytes_to_f32_slice(output);
499                    let n = floats.len().min(required);
500                    out[..n].copy_from_slice(&floats[..n]);
501                    return Ok(n);
502                }
503            }
504        }
505
506        for row in 0..rows_a {
507            for col in 0..cols_b {
508                let mut sum = 0.0f32;
509                for inner in 0..shared {
510                    sum += a[row * shared + inner] * b[inner * cols_b + col];
511                }
512                out[row * cols_b + col] = sum;
513            }
514        }
515
516        Ok(required)
517    }
518
519    /// Execute vector dot product
520    pub fn vector_dot_product(
521        &mut self,
522        device_id: &str,
523        a: &[f32],
524        b: &[f32],
525    ) -> Result<f32, CsdError> {
526        let operation_id = self.generate_operation_id();
527
528        let operation = CsdOperationRequest {
529            operation_id,
530            function_id: "vector_dot_product".to_string(),
531            device_id: device_id.to_string(),
532            inputs: vec![
533                OperationInput {
534                    name: "vector_a".to_string(),
535                    data: self.f32_slice_to_bytes(a),
536                    location: DataLocation::HostMemory,
537                },
538                OperationInput {
539                    name: "vector_b".to_string(),
540                    data: self.f32_slice_to_bytes(b),
541                    location: DataLocation::HostMemory,
542                },
543            ],
544            outputs: vec![OperationOutput {
545                name: "result".to_string(),
546                size: 4, // 4 bytes for f32
547                location: DataLocation::HostMemory,
548            }],
549            priority: OperationPriority::Normal,
550            deadline: None,
551        };
552
553        self.execute_operation(operation)?;
554        let completion = self.wait_for_completion(operation_id)?;
555
556        if let Some(output) = completion.outputs.first() {
557            let result = self.bytes_to_f32_value(&output);
558            Ok(result)
559        } else {
560            Err(CsdError::NoOutput("No output generated".to_string()))
561        }
562    }
563
564    /// Execute convolution operation
565    pub fn convolution(
566        &mut self,
567        device_id: &str,
568        input: &[f32],
569        kernel: &[f32],
570        dimensions: (usize, usize, usize, usize),
571    ) -> Result<Vec<f32>, CsdError> {
572        let mut result = vec![0.0f32; dimensions.0 * dimensions.1];
573        let written = self.convolution_into(device_id, input, kernel, dimensions, &mut result)?;
574        result.truncate(written);
575        Ok(result)
576    }
577
578    /// Execute a same-shape 2D convolution into a caller-owned output buffer.
579    pub fn convolution_into(
580        &mut self,
581        device_id: &str,
582        input: &[f32],
583        kernel: &[f32],
584        dimensions: (usize, usize, usize, usize),
585        out: &mut [f32],
586    ) -> Result<usize, CsdError> {
587        self.ensure_device_exists(device_id)?;
588        let _dim_wire = self.serialize_convolution_dimensions(dimensions);
589        let (width, height, kernel_width, kernel_height) = dimensions;
590        if input.len() != width * height {
591            return Err(CsdError::InvalidOperation(
592                "convolution input dimensions do not match input length".to_string(),
593            ));
594        }
595        if kernel.len() != kernel_width * kernel_height {
596            return Err(CsdError::InvalidOperation(
597                "convolution kernel dimensions do not match kernel length".to_string(),
598            ));
599        }
600        let required = width * height;
601        if out.len() < required {
602            return Err(CsdError::BufferTooSmall(
603                "convolution output buffer too small".to_string(),
604            ));
605        }
606
607        let kernel_x_radius = kernel_width / 2;
608        let kernel_y_radius = kernel_height / 2;
609
610        for y in 0..height {
611            for x in 0..width {
612                let mut acc = 0.0f32;
613                for ky in 0..kernel_height {
614                    let Some(input_y) = y
615                        .checked_add(ky)
616                        .and_then(|value| value.checked_sub(kernel_y_radius))
617                    else {
618                        continue;
619                    };
620                    if input_y >= height {
621                        continue;
622                    }
623                    for kx in 0..kernel_width {
624                        let Some(input_x) = x
625                            .checked_add(kx)
626                            .and_then(|value| value.checked_sub(kernel_x_radius))
627                        else {
628                            continue;
629                        };
630                        if input_x >= width {
631                            continue;
632                        }
633                        let input_index = input_y * width + input_x;
634                        let kernel_index = ky * kernel_width + kx;
635                        acc += input[input_index] * kernel[kernel_index];
636                    }
637                }
638                out[y * width + x] = acc;
639            }
640        }
641
642        Ok(required)
643    }
644
645    /// Get device statistics
646    pub fn get_device_stats(&self, device_id: &str) -> Option<CsdDeviceStats> {
647        self.devices
648            .get(device_id)
649            .map(|device| device.device_stats.clone())
650    }
651
652    /// Get performance statistics
653    pub fn get_performance_stats(&self) -> CsdGlobalMetrics {
654        self.performance_monitor.get_global_stats()
655    }
656
657    /// List available devices
658    pub fn list_devices(&self) -> Vec<String> {
659        self.devices.keys().cloned().collect()
660    }
661
662    /// List available devices into a compact caller-owned buffer.
663    pub fn list_devices_into(&self, out: &mut [CsdDeviceHandle]) -> Result<usize, CsdError> {
664        if out.len() < self.devices.len() {
665            return Err(CsdError::BufferTooSmall(
666                "device listing output buffer too small".to_string(),
667            ));
668        }
669
670        let mut written = 0;
671        for device in self.devices.values() {
672            out[written] = Self::device_handle(device);
673            written += 1;
674        }
675        Ok(written)
676    }
677
678    /// List available functions
679    pub fn list_functions(&self) -> Vec<String> {
680        self.functions.keys().cloned().collect()
681    }
682
683    /// List available functions into a compact caller-owned buffer.
684    pub fn list_functions_into(&self, out: &mut [CsdFunctionHandle]) -> Result<usize, CsdError> {
685        if out.len() < self.functions.len() {
686            return Err(CsdError::BufferTooSmall(
687                "function listing output buffer too small".to_string(),
688            ));
689        }
690
691        let mut written = 0;
692        for function in self.functions.values() {
693            out[written] = Self::function_handle(function);
694            written += 1;
695        }
696        Ok(written)
697    }
698
699    // Internal methods
700
701    /// Validate function
702    fn validate_function(&self, function: &CsdFunction) -> Result<(), CsdError> {
703        if function.function_id.is_empty() {
704            return Err(CsdError::InvalidFunction(
705                "Function ID cannot be empty".to_string(),
706            ));
707        }
708
709        if function.bytecode.is_empty() {
710            return Err(CsdError::InvalidFunction(
711                "Function bytecode cannot be empty".to_string(),
712            ));
713        }
714
715        Ok(())
716    }
717
718    /// Validate operation
719    fn validate_operation(&self, operation: &CsdOperationRequest) -> Result<(), CsdError> {
720        // Check if device exists
721        if !self.devices.contains_key(&operation.device_id) {
722            return Err(CsdError::DeviceNotFound(operation.device_id.clone()));
723        }
724
725        // Check if function exists
726        if !self.functions.contains_key(&operation.function_id) {
727            return Err(CsdError::FunctionNotFound(operation.function_id.clone()));
728        }
729
730        // Check inputs
731        if operation.inputs.is_empty() {
732            return Err(CsdError::InvalidOperation(
733                "Operation must have inputs".to_string(),
734            ));
735        }
736
737        Ok(())
738    }
739
740    fn ensure_device_exists(&self, device_id: &str) -> Result<(), CsdError> {
741        if self.devices.contains_key(device_id) {
742            Ok(())
743        } else {
744            Err(CsdError::DeviceNotFound(device_id.to_string()))
745        }
746    }
747
748    /// Wait for operation completion
749    fn wait_for_completion(&self, operation_id: u64) -> Result<CsdCompletion, CsdError> {
750        // In real implementation, would wait for actual completion
751        // For now, simulate completion
752        Ok(CsdCompletion {
753            operation_id,
754            status: CompletionStatus::Success,
755            execution_time: 1000, // 1ms
756            outputs: vec![],
757            error_message: None,
758        })
759    }
760
761    /// Generate unique operation ID
762    fn generate_operation_id(&self) -> u64 {
763        use std::sync::atomic::{AtomicU64, Ordering};
764        static COUNTER: AtomicU64 = AtomicU64::new(1);
765        COUNTER.fetch_add(1, Ordering::SeqCst)
766    }
767
768    /// Convert f32 slice to bytes
769    fn f32_slice_to_bytes(&self, slice: &[f32]) -> Vec<u8> {
770        let mut bytes = Vec::with_capacity(slice.len() * 4);
771        for &value in slice {
772            bytes.extend_from_slice(&value.to_le_bytes());
773        }
774        bytes
775    }
776
777    /// Convert bytes to f32 slice.
778    ///
779    /// `OperationOutput` carries only size and location metadata.  When the data lives in
780    /// host memory the caller must have already staged it; we return a zero-filled buffer
781    /// of the correct length so downstream code has the right shape.  Callers that have
782    /// a concrete byte slice should use `bytemuck::cast_slice` directly.
783    fn bytes_to_f32_slice(&self, output: &OperationOutput) -> Vec<f32> {
784        vec![0.0f32; (output.size / 4) as usize]
785    }
786
787    /// Convert bytes to f32 value.
788    fn bytes_to_f32_value(&self, output: &OperationOutput) -> f32 {
789        // OperationOutput has no inline data; only the first element's default is available.
790        let _ = output;
791        0.0f32
792    }
793
794    /// Serialize matrix dimensions
795    fn serialize_dimensions(&self, dimensions: (usize, usize, usize)) -> Vec<u8> {
796        let mut bytes = Vec::with_capacity(12);
797        bytes.extend_from_slice(&(dimensions.0 as u32).to_le_bytes());
798        bytes.extend_from_slice(&(dimensions.1 as u32).to_le_bytes());
799        bytes.extend_from_slice(&(dimensions.2 as u32).to_le_bytes());
800        bytes
801    }
802
803    /// Serialize convolution dimensions
804    fn serialize_convolution_dimensions(
805        &self,
806        dimensions: (usize, usize, usize, usize),
807    ) -> Vec<u8> {
808        let mut bytes = Vec::with_capacity(16);
809        bytes.extend_from_slice(&(dimensions.0 as u32).to_le_bytes());
810        bytes.extend_from_slice(&(dimensions.1 as u32).to_le_bytes());
811        bytes.extend_from_slice(&(dimensions.2 as u32).to_le_bytes());
812        bytes.extend_from_slice(&(dimensions.3 as u32).to_le_bytes());
813        bytes
814    }
815
816    fn device_handle(device: &CsdDevice) -> CsdDeviceHandle {
817        CsdDeviceHandle {
818            device_id_hash: q_hash(&device.device_id),
819            max_concurrent_operations: device.capabilities.max_concurrent_operations,
820            max_data_size: device.capabilities.max_data_size,
821            memory_size: device.capabilities.memory_size,
822            compute_units: device.capabilities.compute_units,
823        }
824    }
825
826    fn function_handle(function: &CsdFunction) -> CsdFunctionHandle {
827        CsdFunctionHandle {
828            function_id_hash: q_hash(&function.function_id),
829            operation_tag: Self::operation_tag(&function.operation),
830            parameter_count: function.parameters.len() as u16,
831            bytecode_len: function.bytecode.len() as u32,
832        }
833    }
834
835    fn operation_tag(operation: &CsdOperationType) -> u8 {
836        match operation {
837            CsdOperationType::MatrixMultiply => 0x01,
838            CsdOperationType::VectorDotProduct => 0x02,
839            CsdOperationType::Convolution => 0x03,
840            CsdOperationType::Filter => 0x04,
841            CsdOperationType::Aggregate => 0x05,
842            CsdOperationType::Sort => 0x06,
843            CsdOperationType::Search => 0x07,
844            CsdOperationType::Custom(_) => 0xFF,
845        }
846    }
847}
848
849impl CsdScheduler {
850    /// Create new CSD scheduler
851    pub fn new() -> Self {
852        Self {
853            pending_operations: Vec::new(),
854            running_operations: HashMap::new(),
855            completion_queue: Vec::new(),
856            scheduling_policy: SchedulingPolicy::Priority,
857        }
858    }
859
860    /// Schedule operation
861    pub fn schedule_operation(&mut self, operation: CsdOperationRequest) -> Result<(), CsdError> {
862        self.pending_operations.push(operation);
863        Ok(())
864    }
865
866    /// Process pending operations
867    pub fn process_operations(&mut self) -> Vec<CsdCompletion> {
868        match self.scheduling_policy {
869            SchedulingPolicy::Priority => self.pending_operations.sort_by_key(|op| {
870                std::cmp::Reverse(match op.priority {
871                    OperationPriority::Critical => 3u8,
872                    OperationPriority::High => 2,
873                    OperationPriority::Normal => 1,
874                    OperationPriority::Low => 0,
875                })
876            }),
877            SchedulingPolicy::ShortestJobFirst => self
878                .pending_operations
879                .sort_by_key(|op| op.outputs.iter().map(|o| o.size).sum::<u64>()),
880            _ => {}
881        }
882
883        let mut completions = Vec::new();
884
885        while let Some(operation) = self.pending_operations.pop() {
886            self.running_operations.insert(
887                operation.operation_id,
888                CsdRunningOperation {
889                    operation_id: operation.operation_id,
890                    device_id: operation.device_id.clone(),
891                    start_time: 0,
892                    progress: 0.0,
893                },
894            );
895            let completion = self.execute_operation(&operation);
896            self.running_operations.remove(&operation.operation_id);
897            self.completion_queue.push(completion.clone());
898            completions.push(completion);
899        }
900
901        completions
902    }
903
904    /// Drain completions recorded since the last call.
905    pub fn drain_completions(&mut self) -> Vec<CsdCompletion> {
906        std::mem::take(&mut self.completion_queue)
907    }
908
909    /// Execute operation
910    fn execute_operation(&self, operation: &CsdOperationRequest) -> CsdCompletion {
911        CsdCompletion {
912            operation_id: operation.operation_id,
913            status: CompletionStatus::Success,
914            execution_time: 1000, // 1ms
915            outputs: operation.outputs.clone(),
916            error_message: None,
917        }
918    }
919}
920
921impl CsdScheduler {
922    /// Record performance metrics for a completed operation.
923    pub fn record_completion_metrics(
924        &self,
925        monitor: &mut CsdPerformanceMonitor,
926        operation: &CsdOperationRequest,
927        completion: &CsdCompletion,
928    ) {
929        let data_size = operation.outputs.iter().map(|o| o.size).sum::<u64>();
930        monitor.update_metrics(
931            &operation.device_id,
932            &operation.function_id,
933            completion.execution_time,
934            data_size,
935        );
936    }
937}
938
939impl CsdPerformanceMonitor {
940    /// Create new performance monitor
941    pub fn new() -> Self {
942        Self {
943            device_metrics: HashMap::new(),
944            function_metrics: HashMap::new(),
945            global_metrics: CsdGlobalMetrics {
946                total_operations: 0,
947                total_execution_time: 0,
948                average_execution_time: 0.0,
949                total_data_processed: 0,
950                overall_throughput: 0.0,
951                system_utilization: 0.0,
952            },
953        }
954    }
955
956    /// Update metrics
957    pub fn update_metrics(
958        &mut self,
959        device_id: &str,
960        function_id: &str,
961        execution_time: u64,
962        data_size: u64,
963    ) {
964        self.device_metrics
965            .entry(device_id.to_string())
966            .or_insert(CsdDeviceMetrics {
967                device_id: device_id.to_string(),
968                utilization: 0.0,
969                throughput: 0.0,
970                latency: 0.0,
971                error_rate: 0.0,
972                power_consumption: 0.0,
973            })
974            .throughput += data_size as f64;
975        let func = self
976            .function_metrics
977            .entry(function_id.to_string())
978            .or_insert(CsdFunctionMetrics {
979                function_id: function_id.to_string(),
980                execution_count: 0,
981                total_execution_time: 0,
982                average_execution_time: 0.0,
983                success_rate: 1.0,
984                data_throughput: 0.0,
985            });
986        func.execution_count += 1;
987        func.total_execution_time += execution_time;
988        func.data_throughput += data_size as f64;
989
990        self.global_metrics.total_operations += 1;
991        self.global_metrics.total_execution_time += execution_time;
992        self.global_metrics.average_execution_time = self.global_metrics.total_execution_time
993            as f64
994            / self.global_metrics.total_operations as f64;
995        self.global_metrics.total_data_processed += data_size;
996        self.global_metrics.overall_throughput = self.global_metrics.total_data_processed as f64
997            / self.global_metrics.total_execution_time as f64;
998    }
999
1000    /// Get global statistics
1001    pub fn get_global_stats(&self) -> CsdGlobalMetrics {
1002        self.global_metrics.clone()
1003    }
1004}
1005
1006impl MathComputationBuilder {
1007    /// Create new computation builder
1008    pub fn new() -> Self {
1009        Self {
1010            operations: Vec::new(),
1011            data_dependencies: HashMap::new(),
1012            execution_plan: ExecutionPlan {
1013                stages: Vec::new(),
1014                parallel_groups: Vec::new(),
1015                estimated_time: 0.0,
1016                resource_requirements: ResourceRequirements {
1017                    memory_usage: 0,
1018                    compute_units: 0,
1019                    bandwidth: 0.0,
1020                },
1021            },
1022        }
1023    }
1024
1025    /// Add matrix multiplication operation
1026    pub fn add_matrix_multiply(
1027        &mut self,
1028        device_id: String,
1029        a: Vec<f32>,
1030        b: Vec<f32>,
1031        dimensions: (usize, usize, usize),
1032    ) -> &mut Self {
1033        let operation_id = self.generate_operation_id();
1034
1035        let operation = CsdOperationRequest {
1036            operation_id,
1037            function_id: "matrix_multiply".to_string(),
1038            device_id,
1039            inputs: vec![
1040                OperationInput {
1041                    name: "matrix_a".to_string(),
1042                    data: self.f32_slice_to_bytes(&a),
1043                    location: DataLocation::HostMemory,
1044                },
1045                OperationInput {
1046                    name: "matrix_b".to_string(),
1047                    data: self.f32_slice_to_bytes(&b),
1048                    location: DataLocation::HostMemory,
1049                },
1050            ],
1051            outputs: vec![OperationOutput {
1052                name: format!("result_{}", operation_id),
1053                size: ((dimensions.0 * dimensions.2) * 4) as u64,
1054                location: DataLocation::HostMemory,
1055            }],
1056            priority: OperationPriority::Normal,
1057            deadline: None,
1058        };
1059
1060        self.operations.push(operation);
1061        self
1062    }
1063
1064    /// Build execution plan
1065    pub fn build(&mut self) -> Result<ExecutionPlan, CsdError> {
1066        // Analyze dependencies
1067        self.analyze_dependencies();
1068
1069        // Create execution stages
1070        self.create_execution_stages();
1071
1072        // Estimate execution time
1073        self.estimate_execution_time();
1074
1075        Ok(self.execution_plan.clone())
1076    }
1077
1078    /// Analyze data dependencies
1079    fn analyze_dependencies(&mut self) {
1080        // Simple dependency analysis - in real implementation would be more sophisticated
1081        for (i, operation) in self.operations.iter().enumerate() {
1082            let mut dependencies = Vec::new();
1083
1084            // Check if this operation depends on previous operations
1085            for j in 0..i {
1086                let prev_operation = &self.operations[j];
1087
1088                // Check if any output of previous operation is used as input
1089                for output in &prev_operation.outputs {
1090                    for input in &operation.inputs {
1091                        if input.name.contains(&output.name) {
1092                            dependencies.push(prev_operation.function_id.clone());
1093                        }
1094                    }
1095                }
1096            }
1097
1098            self.data_dependencies
1099                .insert(operation.function_id.clone(), dependencies);
1100        }
1101    }
1102
1103    /// Create execution stages
1104    fn create_execution_stages(&mut self) {
1105        // Simple stage creation - in real implementation would be more sophisticated
1106        let mut stage_id = 0;
1107        let mut processed_operations = std::collections::HashSet::new();
1108
1109        while processed_operations.len() < self.operations.len() {
1110            let mut current_stage = Vec::new();
1111            let mut stage_dependencies = Vec::new();
1112
1113            for operation in &self.operations {
1114                if !processed_operations.contains(&operation.function_id) {
1115                    let dependencies = self
1116                        .data_dependencies
1117                        .get(&operation.function_id)
1118                        .cloned()
1119                        .unwrap_or_default();
1120
1121                    // Check if all dependencies are processed
1122                    let can_execute = dependencies
1123                        .iter()
1124                        .all(|dep| processed_operations.contains(dep));
1125
1126                    if can_execute {
1127                        current_stage.push(operation.function_id.clone());
1128                        stage_dependencies.extend(dependencies.clone());
1129                    }
1130                }
1131            }
1132
1133            if current_stage.is_empty() {
1134                break; // Circular dependency or error
1135            }
1136
1137            let stage = ExecutionStage {
1138                stage_id,
1139                operations: current_stage.clone(),
1140                dependencies: stage_dependencies,
1141                estimated_time: 1.0, // 1ms per operation
1142            };
1143
1144            self.execution_plan.stages.push(stage);
1145
1146            for operation in &current_stage {
1147                processed_operations.insert(operation.clone());
1148            }
1149
1150            stage_id += 1;
1151        }
1152    }
1153
1154    /// Estimate execution time
1155    fn estimate_execution_time(&mut self) {
1156        let mut total_time = 0.0;
1157
1158        for stage in &self.execution_plan.stages {
1159            total_time += stage.estimated_time;
1160        }
1161
1162        self.execution_plan.estimated_time = total_time;
1163    }
1164
1165    /// Generate operation ID
1166    fn generate_operation_id(&self) -> u64 {
1167        use std::sync::atomic::{AtomicU64, Ordering};
1168        static COUNTER: AtomicU64 = AtomicU64::new(1);
1169        COUNTER.fetch_add(1, Ordering::SeqCst)
1170    }
1171
1172    /// Convert f32 slice to bytes
1173    fn f32_slice_to_bytes(&self, slice: &[f32]) -> Vec<u8> {
1174        let mut bytes = Vec::with_capacity(slice.len() * 4);
1175        for &value in slice {
1176            bytes.extend_from_slice(&value.to_le_bytes());
1177        }
1178        bytes
1179    }
1180}
1181
1182/// CSD error types
1183#[derive(Debug, Clone)]
1184pub enum CsdError {
1185    DeviceOpen(String),
1186    DeviceNotFound(String),
1187    FunctionNotFound(String),
1188    InvalidFunction(String),
1189    InvalidOperation(String),
1190    NoOutput(String),
1191    ExecutionError(String),
1192    ConfigurationError(String),
1193    BufferTooSmall(String),
1194}
1195
1196impl std::fmt::Display for CsdError {
1197    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1198        match self {
1199            CsdError::DeviceOpen(msg) => write!(f, "Device open error: {}", msg),
1200            CsdError::DeviceNotFound(msg) => write!(f, "Device not found: {}", msg),
1201            CsdError::FunctionNotFound(msg) => write!(f, "Function not found: {}", msg),
1202            CsdError::InvalidFunction(msg) => write!(f, "Invalid function: {}", msg),
1203            CsdError::InvalidOperation(msg) => write!(f, "Invalid operation: {}", msg),
1204            CsdError::NoOutput(msg) => write!(f, "No output: {}", msg),
1205            CsdError::ExecutionError(msg) => write!(f, "Execution error: {}", msg),
1206            CsdError::ConfigurationError(msg) => write!(f, "Configuration error: {}", msg),
1207            CsdError::BufferTooSmall(msg) => write!(f, "Buffer too small: {}", msg),
1208        }
1209    }
1210}
1211
1212impl std::error::Error for CsdError {}
1213
1214#[cfg(test)]
1215mod tests {
1216    use super::*;
1217
1218    #[test]
1219    fn test_csd_manager_creation() {
1220        let manager = CsdManager::new();
1221        assert_eq!(manager.list_devices().len(), 0);
1222        assert_eq!(manager.list_functions().len(), 0);
1223    }
1224
1225    #[test]
1226    fn test_matrix_multiply() {
1227        let mut manager = CsdManager::new();
1228
1229        // Create dummy device
1230        let device = CsdDevice {
1231            device_id: "test_device".to_string(),
1232            device_path: "/dev/nvme0".to_string(),
1233            capabilities: CsdCapabilities {
1234                max_concurrent_operations: 16,
1235                max_data_size: 1024 * 1024 * 1024,
1236                supported_operations: vec![CsdOperationType::MatrixMultiply],
1237                memory_size: 8 * 1024 * 1024 * 1024,
1238                compute_units: 64,
1239                clock_speed: 1.5,
1240            },
1241            supported_functions: vec![],
1242            device_stats: CsdDeviceStats {
1243                operations_completed: 0,
1244                total_execution_time: 0,
1245                average_execution_time: 0.0,
1246                data_processed: 0,
1247                error_count: 0,
1248                utilization: 0.0,
1249            },
1250        };
1251
1252        manager.register_device(device).unwrap();
1253
1254        // For testing, we'll skip the actual device discovery
1255        // and just test the matrix multiplication logic
1256        let a = vec![1.0_f32, 2.0_f32, 3.0_f32, 4.0_f32];
1257        let b = vec![5.0_f32, 6.0_f32, 7.0_f32, 8.0_f32];
1258
1259        // This would normally work with a real device
1260        let result = manager
1261            .matrix_multiply("test_device", &a, &b, (2, 2, 2))
1262            .unwrap();
1263        assert_eq!(result.len(), 4);
1264    }
1265
1266    #[test]
1267    fn test_math_computation_builder() {
1268        let mut builder = MathComputationBuilder::new();
1269
1270        let a = vec![1.0, 2.0, 3.0, 4.0];
1271        let b = vec![5.0, 6.0, 7.0, 8.0];
1272
1273        builder.add_matrix_multiply("device1".to_string(), a, b, (2, 2, 2));
1274
1275        let plan = builder.build();
1276        assert!(plan.is_ok());
1277
1278        if let Ok(plan) = plan {
1279            assert_eq!(plan.stages.len(), 1);
1280            assert!(plan.estimated_time > 0.0);
1281        }
1282    }
1283}
qualia_core_db/csd_storage.rs

qualia_core_db/
csd_storage.rs
