qualia_core_db/platform/
host.rs

1//! Host-side I/O management for calculus modality.
2//!
3//! This module provides platform-specific zero-copy streaming implementations
4//! that bypass the OS page cache for deterministic edge compute. It manages
5//! memory-mapped I/O, io_uring (Linux), and IOCP (Windows) with strict alignment
6//! requirements for DMA transfers.
7//!
8//! ## Architecture
9//!
10//! - **ZeroCopyStreamer trait**: Platform-agnostic interface for async I/O
11//! - **DmaBuffer**: Page-aligned (4096-byte) buffers for DMA transfers
12//! - **Memory pinning**: Prevents swap to ensure DMA stability
13//! - **Double-buffering**: One buffer active for reading, one inactive for DMA
14//!
15//! ## Windows FFI Firewall
16//!
17//! Windows-specific IOCP implementation is isolated in directml_bridge.rs to avoid
18//! DirectX API version conflicts. This module uses type-erased FFI functions to
19//! communicate with the Windows-specific code.
20
21use std::fs::File;
22use std::io;
23use std::path::Path;
24
25// ─── Constants ─────────────────────────────────────────────────────────────────
26
27/// OS page size for DMA alignment (4096 bytes on most systems)
28pub const PAGE_SIZE: usize = 4096;
29
30/// Default buffer size (65536 bytes = 16 pages = 8192 f64 values)
31pub const DEFAULT_BUFFER_SIZE: usize = 65536;
32
33// ─── Errors ─────────────────────────────────────────────────────────────────────
34
35#[derive(Debug)]
36pub enum IoError {
37    MisalignedOffset { offset: u64, required: u64 },
38    MisalignedBufferSize { size: usize, required: usize },
39    FileOpenError(io::Error),
40    IoError(io::Error),
41    LockError(String),
42    InvalidState(String),
43}
44
45impl From<io::Error> for IoError {
46    fn from(err: io::Error) -> Self {
47        IoError::IoError(err)
48    }
49}
50
51// ─── DMA Buffer (Page-Aligned) ─────────────────────────────────────────────────
52
53/// DMA buffer aligned to OS page boundaries (4096 bytes).
54///
55/// Required for O_DIRECT (Linux) and FILE_FLAG_NO_BUFFERING (Windows) to ensure
56/// the NVMe controller can DMA directly into the buffer without kernel intervention.
57#[repr(C, align(4096))]
58pub struct DmaBuffer<const N: usize> {
59    data: [u8; N],
60}
61
62impl<const N: usize> DmaBuffer<N> {
63    /// Creates a new zero-initialized DMA buffer.
64    ///
65    /// # Panics
66    ///
67    /// Panics if N is not a multiple of PAGE_SIZE (4096).
68    pub fn new() -> Self {
69        assert!(
70            N % PAGE_SIZE == 0,
71            "DMA buffer size must be a multiple of PAGE_SIZE (4096), got {}",
72            N
73        );
74
75        Self { data: [0u8; N] }
76    }
77
78    /// Returns the buffer as a byte slice.
79    pub fn as_slice(&self) -> &[u8] {
80        &self.data
81    }
82
83    /// Returns the buffer as a mutable byte slice.
84    pub fn as_mut_slice(&mut self) -> &mut [u8] {
85        &mut self.data
86    }
87
88    /// Returns the buffer length in bytes.
89    pub fn len(&self) -> usize {
90        N
91    }
92
93    /// Returns true if the buffer is empty.
94    pub fn is_empty(&self) -> bool {
95        N == 0
96    }
97
98    /// Returns the number of f64 values this buffer can hold.
99    pub fn f64_capacity(&self) -> usize {
100        N / 8
101    }
102}
103
104impl<const N: usize> Default for DmaBuffer<N> {
105    fn default() -> Self {
106        Self::new()
107    }
108}
109
110// ─── ZeroCopyStreamer Trait ─────────────────────────────────────────────────────
111
112/// Platform-agnostic zero-copy streaming interface.
113///
114/// This trait abstracts over Linux io_uring and Windows IOCP, providing a
115/// consistent API for asynchronous DMA transfers. The host guarantees that
116/// one buffer is always safely mutable by the OS while the core reads from the other.
117pub trait ZeroCopyStreamer: Send {
118    /// Issues an asynchronous hardware read into the INACTIVE buffer.
119    ///
120    /// # Errors
121    ///
122    /// Returns an error if the offset is not 4096-byte aligned. Both O_DIRECT
123    /// and FILE_FLAG_NO_BUFFERING require strict sector alignment.
124    fn async_read_chunk(&mut self, offset: u64) -> Result<(), IoError>;
125
126    /// Non-blocking poll for completion.
127    ///
128    /// Returns `Some(&[u8])` only if the hardware has completed the DMA transfer
129    /// into the inactive buffer. Upon returning `Some`, internally swaps
130    /// active/inactive pointers.
131    ///
132    /// Returns `None` if the transfer is still in progress.
133    fn poll_completion(&mut self) -> Option<&[u8]>;
134
135    /// Returns the currently active buffer for SIMD chunking.
136    ///
137    /// This buffer is guaranteed to be stable and readable by the core.
138    fn get_active_buffer(&self) -> &[u8];
139
140    /// Returns the buffer size in bytes.
141    fn buffer_size(&self) -> usize;
142}
143
144// ─── Windows IOCP Implementation (FFI Firewall) ───────────────────────────────────
145
146/// Windows IOCP-based zero-copy streamer using FFI firewall.
147///
148/// This implementation delegates all Windows-specific operations to directml_bridge.rs
149/// via type-erased FFI functions to avoid DirectX API version conflicts.
150#[cfg(target_os = "windows")]
151pub struct IocpGridManager {
152    handle: *mut crate::directml_bridge::IocpHandle,
153    buffer_a: DmaBuffer<DEFAULT_BUFFER_SIZE>,
154    buffer_b: DmaBuffer<DEFAULT_BUFFER_SIZE>,
155    active_buffer: BufferId,
156    pending_read: bool,
157}
158
159#[cfg(target_os = "windows")]
160#[derive(Clone, Copy, PartialEq, Eq)]
161enum BufferId {
162    A,
163    B,
164}
165
166#[cfg(target_os = "windows")]
167unsafe impl Send for IocpGridManager {}
168
169#[cfg(target_os = "windows")]
170impl IocpGridManager {
171    /// Creates a new IOCP grid manager via FFI.
172    pub fn new(file_path: &Path) -> Result<Self, IoError> {
173        let path_str = file_path
174            .to_str()
175            .ok_or_else(|| IoError::InvalidState("Invalid UTF-8 path".to_string()))?;
176
177        unsafe {
178            let mut handle = std::ptr::null_mut();
179            let status = crate::directml_bridge::iocp_create_ffi(
180                path_str.as_ptr(),
181                path_str.len(),
182                &mut handle,
183            );
184
185            if status != crate::directml_bridge::DmlStatus::Success {
186                return Err(IoError::IoError(io::Error::new(
187                    io::ErrorKind::Other,
188                    status.message(),
189                )));
190            }
191
192            Ok(Self {
193                handle,
194                buffer_a: DmaBuffer::new(),
195                buffer_b: DmaBuffer::new(),
196                active_buffer: BufferId::A,
197                pending_read: false,
198            })
199        }
200    }
201
202    fn get_inactive_buffer_mut(&mut self) -> &mut [u8] {
203        match self.active_buffer {
204            BufferId::A => self.buffer_b.as_mut_slice(),
205            BufferId::B => self.buffer_a.as_mut_slice(),
206        }
207    }
208
209    fn get_inactive_buffer(&self) -> &[u8] {
210        match self.active_buffer {
211            BufferId::A => self.buffer_b.as_slice(),
212            BufferId::B => self.buffer_a.as_slice(),
213        }
214    }
215
216    fn swap_buffers(&mut self) {
217        self.active_buffer = match self.active_buffer {
218            BufferId::A => BufferId::B,
219            BufferId::B => BufferId::A,
220        };
221    }
222}
223
224#[cfg(target_os = "windows")]
225impl ZeroCopyStreamer for IocpGridManager {
226    fn async_read_chunk(&mut self, offset: u64) -> Result<(), IoError> {
227        // Validate 4096-byte alignment
228        if offset % PAGE_SIZE as u64 != 0 {
229            return Err(IoError::MisalignedOffset {
230                offset,
231                required: PAGE_SIZE as u64,
232            });
233        }
234
235        if self.pending_read {
236            return Err(IoError::InvalidState(
237                "Read already in progress. Call poll_completion first.".to_string(),
238            ));
239        }
240
241        // Prepare the inactive DMA buffer before the async transfer posts.
242        self.get_inactive_buffer_mut().fill(0);
243        let _dma_len = self.get_inactive_buffer().len();
244
245        unsafe {
246            let status = crate::directml_bridge::iocp_async_read_ffi(self.handle, offset);
247            if status != crate::directml_bridge::DmlStatus::Success {
248                return Err(IoError::IoError(io::Error::new(
249                    io::ErrorKind::Other,
250                    status.message(),
251                )));
252            }
253        }
254
255        self.pending_read = true;
256        Ok(())
257    }
258
259    fn poll_completion(&mut self) -> Option<&[u8]> {
260        if !self.pending_read {
261            return None;
262        }
263
264        unsafe {
265            let mut buffer_ptr = std::ptr::null();
266            let mut size = 0usize;
267
268            if crate::directml_bridge::iocp_poll_completion_ffi(
269                self.handle,
270                &mut buffer_ptr,
271                &mut size,
272            ) {
273                self.pending_read = false;
274                self.swap_buffers();
275                Some(self.get_active_buffer())
276            } else {
277                None
278            }
279        }
280    }
281
282    fn get_active_buffer(&self) -> &[u8] {
283        match self.active_buffer {
284            BufferId::A => self.buffer_a.as_slice(),
285            BufferId::B => self.buffer_b.as_slice(),
286        }
287    }
288
289    fn buffer_size(&self) -> usize {
290        DEFAULT_BUFFER_SIZE
291    }
292}
293
294#[cfg(target_os = "windows")]
295impl Drop for IocpGridManager {
296    fn drop(&mut self) {
297        unsafe {
298            crate::directml_bridge::iocp_destroy_ffi(self.handle);
299        }
300    }
301}
302
303// ─── Linux io_uring Implementation ─────────────────────────────────────────────
304
305#[cfg(target_os = "linux")]
306use libc::{c_void, mlock, O_DIRECT};
307#[cfg(target_os = "linux")]
308use std::os::unix::fs::OpenOptionsExt;
309#[cfg(target_os = "linux")]
310use std::os::unix::io::AsRawFd;
311
312/// Linux io_uring-based zero-copy streamer.
313///
314/// Uses O_DIRECT to bypass the kernel page cache and achieve deterministic
315/// DMA transfers from NVMe to RAM.
316#[cfg(target_os = "linux")]
317pub struct IoUringGridManager {
318    ring: io_uring::IoUring,
319    file: File,
320    buffer_a: DmaBuffer<DEFAULT_BUFFER_SIZE>,
321    buffer_b: DmaBuffer<DEFAULT_BUFFER_SIZE>,
322    active_buffer: BufferId,
323    pending_submission: bool,
324}
325
326#[cfg(target_os = "linux")]
327#[derive(Clone, Copy, PartialEq, Eq)]
328enum BufferId {
329    A,
330    B,
331}
332
333#[cfg(target_os = "linux")]
334impl IoUringGridManager {
335    /// Creates a new io_uring grid manager.
336    ///
337    /// Opens the file with O_DIRECT, creates an io_uring instance, and pins
338    /// both buffers in physical RAM.
339    pub fn new(file_path: &Path) -> Result<Self, IoError> {
340        let file = File::options()
341            .read(true)
342            .custom_flags(O_DIRECT)
343            .open(file_path)
344            .map_err(IoError::FileOpenError)?;
345
346        let ring = io_uring::IoUring::new(8).map_err(IoError::IoError)?;
347
348        let mut manager = Self {
349            ring,
350            file,
351            buffer_a: DmaBuffer::new(),
352            buffer_b: DmaBuffer::new(),
353            active_buffer: BufferId::A,
354            pending_submission: false,
355        };
356
357        // Pin buffers in physical RAM to prevent swap
358        manager.pin_buffers()?;
359
360        Ok(manager)
361    }
362
363    /// Pins both buffers in physical RAM using mlock.
364    fn pin_buffers(&mut self) -> Result<(), IoError> {
365        unsafe {
366            let result_a = mlock(
367                self.buffer_a.as_slice().as_ptr() as *const c_void,
368                self.buffer_a.len(),
369            );
370
371            let result_b = mlock(
372                self.buffer_b.as_slice().as_ptr() as *const c_void,
373                self.buffer_b.len(),
374            );
375
376            if result_a == 0 && result_b == 0 {
377                Ok(())
378            } else {
379                Err(IoError::LockError(
380                    "Failed to pin DMA buffers in physical RAM".to_string(),
381                ))
382            }
383        }
384    }
385
386    fn get_inactive_buffer_mut(&mut self) -> &mut [u8] {
387        match self.active_buffer {
388            BufferId::A => self.buffer_b.as_mut_slice(),
389            BufferId::B => self.buffer_a.as_mut_slice(),
390        }
391    }
392
393    fn get_inactive_buffer(&self) -> &[u8] {
394        match self.active_buffer {
395            BufferId::A => self.buffer_b.as_slice(),
396            BufferId::B => self.buffer_a.as_slice(),
397        }
398    }
399
400    fn swap_buffers(&mut self) {
401        self.active_buffer = match self.active_buffer {
402            BufferId::A => BufferId::B,
403            BufferId::B => BufferId::A,
404        };
405    }
406}
407
408#[cfg(target_os = "linux")]
409impl ZeroCopyStreamer for IoUringGridManager {
410    fn async_read_chunk(&mut self, offset: u64) -> Result<(), IoError> {
411        // Validate 4096-byte alignment
412        if offset % PAGE_SIZE as u64 != 0 {
413            return Err(IoError::MisalignedOffset {
414                offset,
415                required: PAGE_SIZE as u64,
416            });
417        }
418
419        if self.pending_submission {
420            return Err(IoError::InvalidState(
421                "Read already submitted. Call poll_completion first.".to_string(),
422            ));
423        }
424
425        let read_op = io_uring::opcode::Read::new(
426            io_uring::types::Fd(self.file.as_raw_fd()),
427            self.get_inactive_buffer_mut().as_mut_ptr(),
428            self.get_inactive_buffer().len() as u32,
429        )
430        .offset(offset)
431        .build();
432
433        unsafe {
434            self.ring.submission().push(&read_op).map_err(|e| {
435                IoError::IoError(io::Error::new(io::ErrorKind::Other, e.to_string()))
436            })?;
437        }
438
439        self.pending_submission = true;
440        Ok(())
441    }
442
443    fn poll_completion(&mut self) -> Option<&[u8]> {
444        if !self.pending_submission {
445            return None;
446        }
447
448        match self.ring.submit_and_wait(1) {
449            Ok(_) => {
450                let cqe_result = self.ring.completion().next().map(|cqe| cqe.result());
451                if let Some(res) = cqe_result {
452                    self.pending_submission = false;
453
454                    if res >= 0 {
455                        self.swap_buffers();
456                        Some(self.get_active_buffer())
457                    } else {
458                        // I/O error - reset state
459                        None
460                    }
461                } else {
462                    None
463                }
464            }
465            Err(_) => None,
466        }
467    }
468
469    fn get_active_buffer(&self) -> &[u8] {
470        match self.active_buffer {
471            BufferId::A => self.buffer_a.as_slice(),
472            BufferId::B => self.buffer_b.as_slice(),
473        }
474    }
475
476    fn buffer_size(&self) -> usize {
477        DEFAULT_BUFFER_SIZE
478    }
479}
480
481#[cfg(target_os = "linux")]
482impl Drop for IoUringGridManager {
483    fn drop(&mut self) {
484        unsafe {
485            // Unlock buffers
486            let _ = libc::munlock(
487                self.buffer_a.as_slice().as_ptr() as *const c_void,
488                self.buffer_a.len(),
489            );
490            let _ = libc::munlock(
491                self.buffer_b.as_slice().as_ptr() as *const c_void,
492                self.buffer_b.len(),
493            );
494        }
495    }
496}
497
498// ─── Fallback mmap Implementation (Non-Deterministic) ───────────────────────────
499
500/// Simple mmap-based grid manager for non-real-time use cases.
501///
502/// This implementation does not bypass the page cache and may incur page faults.
503/// Use only for desktop applications where occasional stalls are acceptable.
504pub struct MmapGridManager {
505    mmap: memmap2::Mmap,
506}
507
508impl MmapGridManager {
509    /// Creates a new mmap grid manager.
510    ///
511    /// Uses madvise to hint sequential read-ahead pattern.
512    pub fn new(file_path: &Path) -> Result<Self, IoError> {
513        let file = File::open(file_path).map_err(IoError::FileOpenError)?;
514        let mmap = unsafe { memmap2::Mmap::map(&file) }.map_err(IoError::IoError)?;
515
516        // Issue madvise for aggressive read-ahead
517        #[cfg(target_os = "linux")]
518        unsafe {
519            libc::madvise(
520                mmap.as_ptr() as *mut libc::c_void,
521                mmap.len(),
522                libc::MADV_SEQUENTIAL | libc::MADV_WILLNEED,
523            );
524        }
525
526        Ok(Self { mmap })
527    }
528
529    /// Returns the entire memory-mapped slice.
530    pub fn get_slice(&self) -> &[u8] {
531        &self.mmap
532    }
533}
534
535// ─── Tests ─────────────────────────────────────────────────────────────────────
536
537#[cfg(test)]
538mod tests {
539    use super::*;
540    use std::io::Write;
541
542    #[test]
543    fn test_dma_buffer_alignment() {
544        let buffer = DmaBuffer::<4096>::new();
545
546        // Verify pointer is 4096-byte aligned
547        assert_eq!(buffer.as_slice().as_ptr() as usize % 4096, 0);
548    }
549
550    #[test]
551    #[should_panic(expected = "DMA buffer size must be a multiple of PAGE_SIZE")]
552    fn test_dma_buffer_misaligned_size() {
553        let _buffer = DmaBuffer::<4095>::new();
554    }
555
556    #[test]
557    fn test_dma_buffer_f64_capacity() {
558        let buffer = DmaBuffer::<65536>::new();
559        assert_eq!(buffer.f64_capacity(), 8192); // 65536 / 8
560    }
561
562    #[cfg(target_os = "windows")]
563    #[test]
564    fn test_iocp_offset_validation() {
565        // Create a temporary test file
566        let temp_dir = std::env::temp_dir();
567        let file_path = temp_dir.join("test_iocp.dat");
568
569        let mut file = File::create(&file_path).unwrap();
570        // Write at least 2 pages of data
571        file.write_all(&vec![0u8; 8192]).unwrap();
572        file.sync_all().unwrap();
573
574        // IOCP is currently stubbed due to DirectX API version conflicts
575        // Test that it returns the expected error
576        let result = IocpGridManager::new(&file_path);
577        assert!(result.is_err());
578
579        // Verify the error message mentions the stub
580        match result {
581            Err(IoError::IoError(e)) => {
582                let error_msg = e.to_string();
583                assert!(
584                    error_msg.contains("IOCP implementation stubbed")
585                        || error_msg.contains("DirectStorageFailed"),
586                    "Expected stub error, got: {}",
587                    error_msg
588                );
589            }
590            _ => panic!("Expected IoError::IoError with stub message"),
591        }
592
593        // Cleanup
594        std::fs::remove_file(&file_path).unwrap();
595    }
596
597    #[cfg(target_os = "linux")]
598    #[test]
599    fn test_iouring_offset_validation() {
600        // Create a temporary test file
601        let temp_dir = std::env::temp_dir();
602        let file_path = temp_dir.join("test_iouring.dat");
603
604        let mut file = File::create(&file_path).unwrap();
605        file.write_all(&vec![0u8; 8192]).unwrap();
606        file.sync_all().unwrap();
607
608        let mut manager = IoUringGridManager::new(&file_path).unwrap();
609
610        // Test valid offset (4096-byte aligned)
611        assert!(manager.async_read_chunk(4096).is_ok());
612
613        // Test invalid offset (not aligned)
614        let result = manager.async_read_chunk(4095);
615        assert!(matches!(result, Err(IoError::MisalignedOffset { .. })));
616
617        // Cleanup
618        std::fs::remove_file(&file_path).unwrap();
619    }
620}
qualia_core_db/platform/host.rs

qualia_core_db/platform/
host.rs
