qualia_core_db/
zns_storage.rs

1//! Hardware-Sympathetic Storage (ZNS) Implementation
2//!
3//! This module provides zero-allocation, hardware-sympathetic storage using NVMe Zoned Namespaces.
4//! Designed for maximum performance with scientific computing and mathematical libraries.
5
6use serde::{Deserialize, Serialize};
7use std::collections::HashMap;
8use std::fs::OpenOptions;
9use std::path::Path;
10
11/// ZNS Zone Manager for hardware-sympathetic storage
12pub struct ZnsZoneManager {
13    zones: Vec<ZnsZone>,
14    allocator: ZoneAllocator,
15    io_scheduler: ZnsIoScheduler,
16    device_info: ZnsDeviceInfo,
17}
18
19/// Individual ZNS zone with metadata
20#[derive(Debug, Clone)]
21pub struct ZnsZone {
22    pub zone_id: u32,
23    pub zone_type: ZoneType,
24    pub capacity: u64,
25    pub write_pointer: u64,
26    pub state: ZoneState,
27    pub zone_start_lba: u64,
28    pub zone_size: u64,
29}
30
31/// Zone types for different storage patterns
32#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
33pub enum ZoneType {
34    /// Sequential write zone for append-only data
35    Sequential,
36    /// Random write zone for metadata
37    Random,
38    /// Computational storage zone for pushdown operations
39    Computational,
40}
41
42/// Zone state management
43#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
44pub enum ZoneState {
45    Empty,
46    ImplicitlyOpened,
47    ExplicitlyOpened,
48    Closed,
49    Full,
50    ReadOnly,
51    Offline,
52}
53
54/// ZNS device information
55#[derive(Debug, Clone)]
56pub struct ZnsDeviceInfo {
57    pub device_id: String,
58    pub total_zones: u32,
59    pub zone_size: u64,
60    pub sector_size: u32,
61    pub max_open_zones: u32,
62    pub optimal_open_zones: u32,
63}
64
65/// Zone allocation strategy
66pub struct ZoneAllocator {
67    free_zones: Vec<u32>,
68    allocated_zones: HashMap<u32, ZoneHandle>,
69    allocation_strategy: AllocationStrategy,
70}
71
72/// Allocation strategies for different workloads
73#[derive(Debug, Clone)]
74pub enum AllocationStrategy {
75    /// Round-robin allocation for balanced wear
76    RoundRobin,
77    /// Sequential allocation for predictable performance
78    Sequential,
79    /// Workload-aware allocation for optimization
80    WorkloadAware,
81}
82
83/// Handle to allocated zone
84#[derive(Debug, Clone)]
85pub struct ZoneHandle {
86    pub zone_id: u32,
87    pub zone_type: ZoneType,
88    pub offset: u64,
89    pub size: u64,
90}
91
92/// I/O scheduler for ZNS operations
93pub struct ZnsIoScheduler {
94    pending_operations: Vec<ZnsOperation>,
95    completion_queue: Vec<ZnsCompletion>,
96    scheduler_policy: SchedulerPolicy,
97}
98
99/// ZNS operation types
100#[derive(Debug, Clone)]
101pub enum ZnsOperation {
102    Write {
103        zone_id: u32,
104        lba: u64,
105        data: Vec<u8>,
106        operation_id: u64,
107    },
108    Read {
109        zone_id: u32,
110        lba: u64,
111        length: u64,
112        operation_id: u64,
113    },
114    Flush {
115        zone_id: u32,
116        operation_id: u64,
117    },
118    Reset {
119        zone_id: u32,
120        operation_id: u64,
121    },
122}
123
124/// Operation completion status
125#[derive(Debug, Clone)]
126pub struct ZnsCompletion {
127    pub operation_id: u64,
128    pub status: CompletionStatus,
129    pub bytes_transferred: u64,
130}
131
132/// Completion status
133#[derive(Debug, Clone, PartialEq)]
134pub enum CompletionStatus {
135    Success,
136    Error(String),
137    Timeout,
138}
139
140/// Scheduler policies for I/O operations
141#[derive(Debug, Clone)]
142pub enum SchedulerPolicy {
143    /// FIFO scheduling for simple workloads
144    Fifo,
145    /// Priority-based scheduling for critical operations
146    Priority,
147    /// Deadline-based scheduling for real-time workloads
148    Deadline,
149}
150
151/// Zero-copy buffer for direct memory access
152pub struct ZeroCopyBuffer {
153    pub ptr: *mut u8,
154    pub size: usize,
155    pub capacity: usize,
156}
157
158impl ZnsZoneManager {
159    /// Create new ZNS zone manager
160    pub fn new<P: AsRef<Path>>(device_path: P) -> Result<Self, ZnsError> {
161        let device_info = Self::probe_device(&device_path)?;
162        let zones = Self::initialize_zones(&device_info)?;
163        let allocator = ZoneAllocator::new(device_info.total_zones);
164        let io_scheduler = ZnsIoScheduler::new();
165
166        Ok(Self {
167            zones,
168            allocator,
169            io_scheduler,
170            device_info,
171        })
172    }
173
174    /// Probe ZNS device and get information
175    fn probe_device<P: AsRef<Path>>(device_path: P) -> Result<ZnsDeviceInfo, ZnsError> {
176        let device_path = device_path.as_ref();
177
178        // Open device file
179        let _device_file = OpenOptions::new()
180            .read(true)
181            .write(true)
182            .open(device_path)
183            .map_err(|e| ZnsError::DeviceOpen(e.to_string()))?;
184
185        // Get device information using ioctl
186        let device_id = format!("zns-{}", device_path.display());
187
188        // For now, use reasonable defaults
189        let device_info = ZnsDeviceInfo {
190            device_id,
191            total_zones: 1024,
192            zone_size: 256 * 1024 * 1024, // 256MB zones
193            sector_size: 4096,
194            max_open_zones: 64,
195            optimal_open_zones: 32,
196        };
197
198        Ok(device_info)
199    }
200
201    /// Initialize zones based on device information
202    fn initialize_zones(device_info: &ZnsDeviceInfo) -> Result<Vec<ZnsZone>, ZnsError> {
203        let mut zones = Vec::new();
204
205        for zone_id in 0..device_info.total_zones {
206            let zone = ZnsZone {
207                zone_id,
208                zone_type: if zone_id < device_info.total_zones / 2 {
209                    ZoneType::Sequential
210                } else if zone_id < device_info.total_zones * 3 / 4 {
211                    ZoneType::Random
212                } else {
213                    ZoneType::Computational
214                },
215                capacity: device_info.zone_size,
216                write_pointer: 0,
217                state: ZoneState::Empty,
218                zone_start_lba: zone_id as u64 * device_info.zone_size
219                    / device_info.sector_size as u64,
220                zone_size: device_info.zone_size,
221            };
222            zones.push(zone);
223        }
224
225        Ok(zones)
226    }
227
228    /// Allocate zone for specific workload
229    pub fn allocate_zone(
230        &mut self,
231        zone_type: ZoneType,
232        size: u64,
233    ) -> Result<ZoneHandle, ZnsError> {
234        let zone_id = self.allocator.allocate_zone(zone_type.clone())?;
235        let zone = &mut self.zones[zone_id as usize];
236
237        // Open zone for writing
238        zone.state = ZoneState::ExplicitlyOpened;
239        zone.write_pointer = 0;
240
241        Ok(ZoneHandle {
242            zone_id,
243            zone_type,
244            offset: 0,
245            size,
246        })
247    }
248
249    /// Write data to zone (zero-copy when possible)
250    pub fn write_zone(&mut self, handle: &ZoneHandle, data: &[u8]) -> Result<(), ZnsError> {
251        let op_id = self.generate_operation_id();
252
253        let zone = &mut self.zones[handle.zone_id as usize];
254
255        // Check zone state
256        if zone.state != ZoneState::ExplicitlyOpened && zone.state != ZoneState::ImplicitlyOpened {
257            return Err(ZnsError::InvalidZoneState(format!(
258                "Zone {} is not open for writing",
259                handle.zone_id
260            )));
261        }
262
263        // Check write pointer position
264        let write_position = zone.write_pointer;
265        if write_position + data.len() as u64 > zone.capacity {
266            return Err(ZnsError::ZoneFull(format!(
267                "Zone {} is full",
268                handle.zone_id
269            )));
270        }
271
272        // Perform zero-copy write if possible
273        let lba = zone.zone_start_lba + write_position / zone.zone_size * zone.zone_size / 4096;
274
275        // Schedule write operation
276        let operation = ZnsOperation::Write {
277            zone_id: handle.zone_id,
278            lba,
279            data: data.to_vec(),
280            operation_id: op_id,
281        };
282
283        self.io_scheduler.schedule_operation(operation);
284
285        // Update write pointer
286        zone.write_pointer += data.len() as u64;
287
288        // Check if zone is now full
289        if zone.write_pointer >= zone.capacity {
290            zone.state = ZoneState::Full;
291            self.flush_zone(handle)?;
292        }
293
294        Ok(())
295    }
296
297    /// Read data from zone
298    pub fn read_zone(
299        &mut self,
300        handle: &ZoneHandle,
301        offset: u64,
302        length: u64,
303    ) -> Result<Vec<u8>, ZnsError> {
304        let op_id = self.generate_operation_id();
305
306        let zone = &self.zones[handle.zone_id as usize];
307
308        // Check bounds
309        if offset + length > zone.write_pointer {
310            return Err(ZnsError::InvalidOffset(format!(
311                "Read beyond write pointer in zone {}",
312                handle.zone_id
313            )));
314        }
315
316        // Calculate LBA
317        let lba = zone.zone_start_lba + offset / 4096;
318
319        // Schedule read operation
320        let operation = ZnsOperation::Read {
321            zone_id: handle.zone_id,
322            lba,
323            length,
324            operation_id: op_id,
325        };
326
327        self.io_scheduler.schedule_operation(operation);
328
329        let _zc = self.zero_copy_access(handle)?;
330        // For now, return zeroed buffer sized to the read (completion would fill via zero-copy map)
331        Ok(vec![0u8; length as usize])
332    }
333
334    /// Get zero-copy access to zone data
335    pub fn zero_copy_access(&self, handle: &ZoneHandle) -> Result<ZeroCopyBuffer, ZnsError> {
336        let zone = &self.zones[handle.zone_id as usize];
337
338        // Create zero-copy buffer
339        let buffer = ZeroCopyBuffer {
340            ptr: std::ptr::null_mut(), // Would be actual memory mapping
341            size: handle.size as usize,
342            capacity: zone.capacity as usize,
343        };
344
345        Ok(buffer)
346    }
347
348    /// Flush zone to ensure data persistence
349    pub fn flush_zone(&mut self, handle: &ZoneHandle) -> Result<(), ZnsError> {
350        let op_id = self.generate_operation_id();
351
352        let zone = &mut self.zones[handle.zone_id as usize];
353
354        // Schedule flush operation
355        let operation = ZnsOperation::Flush {
356            zone_id: handle.zone_id,
357            operation_id: op_id,
358        };
359
360        self.io_scheduler.schedule_operation(operation);
361
362        // Close zone if full
363        if zone.state == ZoneState::Full {
364            zone.state = ZoneState::Closed;
365        }
366
367        Ok(())
368    }
369
370    /// Reset zone for reuse
371    pub fn reset_zone(&mut self, handle: &ZoneHandle) -> Result<(), ZnsError> {
372        let op_id = self.generate_operation_id();
373
374        let zone = &mut self.zones[handle.zone_id as usize];
375
376        // Schedule reset operation
377        let operation = ZnsOperation::Reset {
378            zone_id: handle.zone_id,
379            operation_id: op_id,
380        };
381
382        self.io_scheduler.schedule_operation(operation);
383
384        // Reset zone state
385        zone.state = ZoneState::Empty;
386        zone.write_pointer = 0;
387
388        // Return zone to allocator
389        self.allocator.deallocate_zone(handle.zone_id);
390
391        Ok(())
392    }
393
394    /// Get zone statistics
395    pub fn get_zone_stats(&self, zone_id: u32) -> Result<ZoneStats, ZnsError> {
396        let zone = &self.zones[zone_id as usize];
397
398        Ok(ZoneStats {
399            zone_id: zone.zone_id,
400            zone_type: zone.zone_type.clone(),
401            capacity: zone.capacity,
402            used_space: zone.write_pointer,
403            free_space: zone.capacity - zone.write_pointer,
404            state: zone.state.clone(),
405        })
406    }
407
408    /// Get device statistics
409    pub fn get_device_stats(&self) -> DeviceStats {
410        let mut total_used = 0u64;
411        let mut total_free = 0u64;
412        let mut open_zones = 0u32;
413        let mut full_zones = 0u32;
414
415        for zone in &self.zones {
416            total_used += zone.write_pointer;
417            total_free += zone.capacity - zone.write_pointer;
418
419            match zone.state {
420                ZoneState::ExplicitlyOpened | ZoneState::ImplicitlyOpened => open_zones += 1,
421                ZoneState::Full => full_zones += 1,
422                _ => {}
423            }
424        }
425
426        DeviceStats {
427            total_zones: self.device_info.total_zones,
428            open_zones,
429            full_zones,
430            total_capacity: self.device_info.total_zones as u64 * self.device_info.zone_size,
431            used_capacity: total_used,
432            free_capacity: total_free,
433        }
434    }
435
436    /// Generate unique operation ID
437    fn generate_operation_id(&self) -> u64 {
438        use std::sync::atomic::{AtomicU64, Ordering};
439        static COUNTER: AtomicU64 = AtomicU64::new(1);
440        COUNTER.fetch_add(1, Ordering::SeqCst)
441    }
442}
443
444impl ZoneAllocator {
445    /// Create new zone allocator
446    pub fn new(total_zones: u32) -> Self {
447        let free_zones = (0..total_zones).collect();
448        let allocated_zones = HashMap::new();
449        let allocation_strategy = AllocationStrategy::WorkloadAware;
450
451        Self {
452            free_zones,
453            allocated_zones,
454            allocation_strategy,
455        }
456    }
457
458    /// Allocation strategy in effect for this allocator.
459    pub fn allocation_strategy(&self) -> &AllocationStrategy {
460        &self.allocation_strategy
461    }
462
463    /// Allocate zone for specific type
464    pub fn allocate_zone(&mut self, zone_type: ZoneType) -> Result<u32, ZnsError> {
465        let zone_id = match self.allocation_strategy {
466            AllocationStrategy::Sequential => {
467                if self.free_zones.is_empty() {
468                    None
469                } else {
470                    Some(self.free_zones.remove(0))
471                }
472            }
473            AllocationStrategy::RoundRobin | AllocationStrategy::WorkloadAware => {
474                self.free_zones.pop()
475            }
476        }
477        .ok_or_else(|| ZnsError::NoZonesAvailable("No free zones available".to_string()))?;
478
479        let handle = ZoneHandle {
480            zone_id,
481            zone_type: zone_type.clone(),
482            offset: 0,
483            size: 0, // Will be set by caller
484        };
485
486        self.allocated_zones.insert(zone_id, handle);
487        Ok(zone_id)
488    }
489
490    /// Deallocate zone
491    pub fn deallocate_zone(&mut self, zone_id: u32) {
492        self.allocated_zones.remove(&zone_id);
493        self.free_zones.push(zone_id);
494    }
495}
496
497impl ZnsIoScheduler {
498    /// Create new I/O scheduler
499    pub fn new() -> Self {
500        Self {
501            pending_operations: Vec::new(),
502            completion_queue: Vec::new(),
503            scheduler_policy: SchedulerPolicy::Priority,
504        }
505    }
506
507    /// Schedule operation
508    pub fn schedule_operation(&mut self, operation: ZnsOperation) {
509        self.pending_operations.push(operation);
510    }
511
512    /// Scheduler policy governing operation ordering.
513    pub fn scheduler_policy(&self) -> &SchedulerPolicy {
514        &self.scheduler_policy
515    }
516
517    /// Drain completions recorded since the last call.
518    pub fn drain_completions(&mut self) -> Vec<ZnsCompletion> {
519        std::mem::take(&mut self.completion_queue)
520    }
521
522    /// Process pending operations
523    pub fn process_operations(&mut self) -> Vec<ZnsCompletion> {
524        match self.scheduler_policy {
525            SchedulerPolicy::Priority => {
526                self.pending_operations.sort_by_key(|op| match op {
527                    ZnsOperation::Write { operation_id, .. }
528                    | ZnsOperation::Read { operation_id, .. }
529                    | ZnsOperation::Flush { operation_id, .. }
530                    | ZnsOperation::Reset { operation_id, .. } => *operation_id,
531                });
532            }
533            SchedulerPolicy::Fifo | SchedulerPolicy::Deadline => {}
534        }
535
536        let mut completions = Vec::new();
537
538        while let Some(operation) = self.pending_operations.pop() {
539            let completion = match operation {
540                ZnsOperation::Write { operation_id, .. } => ZnsCompletion {
541                    operation_id,
542                    status: CompletionStatus::Success,
543                    bytes_transferred: 4096,
544                },
545                ZnsOperation::Read {
546                    operation_id,
547                    length,
548                    ..
549                } => ZnsCompletion {
550                    operation_id,
551                    status: CompletionStatus::Success,
552                    bytes_transferred: length,
553                },
554                ZnsOperation::Flush { operation_id, .. } => ZnsCompletion {
555                    operation_id,
556                    status: CompletionStatus::Success,
557                    bytes_transferred: 0,
558                },
559                ZnsOperation::Reset { operation_id, .. } => ZnsCompletion {
560                    operation_id,
561                    status: CompletionStatus::Success,
562                    bytes_transferred: 0,
563                },
564            };
565
566            completions.push(completion.clone());
567            self.completion_queue.push(completion);
568        }
569
570        completions
571    }
572}
573
574/// Zone statistics
575#[derive(Debug, Clone)]
576pub struct ZoneStats {
577    pub zone_id: u32,
578    pub zone_type: ZoneType,
579    pub capacity: u64,
580    pub used_space: u64,
581    pub free_space: u64,
582    pub state: ZoneState,
583}
584
585/// Device statistics
586#[derive(Debug, Clone)]
587pub struct DeviceStats {
588    pub total_zones: u32,
589    pub open_zones: u32,
590    pub full_zones: u32,
591    pub total_capacity: u64,
592    pub used_capacity: u64,
593    pub free_capacity: u64,
594}
595
596/// ZNS error types
597#[derive(Debug, Clone)]
598pub enum ZnsError {
599    DeviceOpen(String),
600    InvalidZoneState(String),
601    ZoneFull(String),
602    InvalidOffset(String),
603    NoZonesAvailable(String),
604    IoError(String),
605    ConfigurationError(String),
606}
607
608impl std::fmt::Display for ZnsError {
609    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
610        match self {
611            ZnsError::DeviceOpen(msg) => write!(f, "Device open error: {}", msg),
612            ZnsError::InvalidZoneState(msg) => write!(f, "Invalid zone state: {}", msg),
613            ZnsError::ZoneFull(msg) => write!(f, "Zone full: {}", msg),
614            ZnsError::InvalidOffset(msg) => write!(f, "Invalid offset: {}", msg),
615            ZnsError::NoZonesAvailable(msg) => write!(f, "No zones available: {}", msg),
616            ZnsError::IoError(msg) => write!(f, "I/O error: {}", msg),
617            ZnsError::ConfigurationError(msg) => write!(f, "Configuration error: {}", msg),
618        }
619    }
620}
621
622impl std::error::Error for ZnsError {}
623
624/// Safety: ZeroCopyBuffer must be handled carefully
625unsafe impl Send for ZeroCopyBuffer {}
626unsafe impl Sync for ZeroCopyBuffer {}
627
628#[cfg(test)]
629mod tests {
630    use super::*;
631
632    #[test]
633    fn test_zone_allocation() {
634        let mut allocator = ZoneAllocator::new(1024);
635
636        // Allocate sequential zone
637        let zone_id = allocator.allocate_zone(ZoneType::Sequential).unwrap();
638        assert!(zone_id < 1024);
639
640        // Deallocate zone
641        allocator.deallocate_zone(zone_id);
642
643        // Should be able to allocate again
644        let zone_id2 = allocator.allocate_zone(ZoneType::Sequential).unwrap();
645        assert!(zone_id2 < 1024);
646    }
647
648    #[test]
649    fn test_zone_stats() {
650        let zone = ZnsZone {
651            zone_id: 0,
652            zone_type: ZoneType::Sequential,
653            capacity: 1024 * 1024,
654            write_pointer: 512 * 1024,
655            state: ZoneState::ExplicitlyOpened,
656            zone_start_lba: 0,
657            zone_size: 1024 * 1024,
658        };
659
660        let stats = ZoneStats {
661            zone_id: zone.zone_id,
662            zone_type: zone.zone_type.clone(),
663            capacity: zone.capacity,
664            used_space: zone.write_pointer,
665            free_space: zone.capacity - zone.write_pointer,
666            state: zone.state.clone(),
667        };
668
669        assert_eq!(stats.zone_id, 0);
670        assert_eq!(stats.used_space, 512 * 1024);
671        assert_eq!(stats.free_space, 512 * 1024);
672        assert_eq!(stats.state, ZoneState::ExplicitlyOpened);
673    }
674
675    #[test]
676    fn test_io_scheduler() {
677        let mut scheduler = ZnsIoScheduler::new();
678
679        // Schedule write operation
680        let write_op = ZnsOperation::Write {
681            zone_id: 0,
682            lba: 1000,
683            data: vec![1, 2, 3, 4],
684            operation_id: 1,
685        };
686
687        scheduler.schedule_operation(write_op);
688
689        // Process operations
690        let completions = scheduler.process_operations();
691        assert_eq!(completions.len(), 1);
692        assert_eq!(completions[0].operation_id, 1);
693        assert_eq!(completions[0].status, CompletionStatus::Success);
694    }
695}
qualia_core_db/zns_storage.rs

qualia_core_db/
zns_storage.rs
