ghostflow_cuda/

stream.rs

//! CUDA streams for async execution - Real Implementation

use crate::error::CudaResult;
use crate::ffi;

/// CUDA stream for asynchronous operations
#[derive(Debug)]
pub struct CudaStream {
    handle: ffi::cudaStream_t,
    is_default: bool,
}

impl CudaStream {
    /// Create new CUDA stream
    pub fn new() -> CudaResult<Self> {
        #[cfg(feature = "cuda")]
        {
            let mut handle: ffi::cudaStream_t = std::ptr::null_mut();
            
            unsafe {
                let err = ffi::cudaStreamCreate(&mut handle);
                if err != 0 {
                    return Err(CudaError::DriverError(err));
                }
            }
            
            Ok(CudaStream {
                handle,
                is_default: false,
            })
        }
        
        #[cfg(not(feature = "cuda"))]
        {
            Ok(CudaStream {
                handle: std::ptr::null_mut(),
                is_default: false,
            })
        }
    }

    /// Get default stream (stream 0)
    pub fn default_stream() -> Self {
        CudaStream {
            handle: std::ptr::null_mut(),
            is_default: true,
        }
    }

    /// Get raw handle
    pub fn handle(&self) -> ffi::cudaStream_t {
        self.handle
    }

    /// Synchronize stream - wait for all operations to complete
    pub fn synchronize(&self) -> CudaResult<()> {
        #[cfg(feature = "cuda")]
        unsafe {
            let err = ffi::cudaStreamSynchronize(self.handle);
            if err != 0 {
                return Err(CudaError::SyncError);
            }
        }
        Ok(())
    }

    /// Check if stream is complete (non-blocking)
    pub fn is_complete(&self) -> CudaResult<bool> {
        #[cfg(feature = "cuda")]
        unsafe {
            let err = ffi::cudaStreamQuery(self.handle);
            if err == 0 {
                return Ok(true);
            } else if err == 600 { // cudaErrorNotReady
                return Ok(false);
            } else {
                return Err(CudaError::DriverError(err));
            }
        }
        
        #[cfg(not(feature = "cuda"))]
        Ok(true)
    }

    /// Wait for event on this stream
    pub fn wait_event(&self, _event: &CudaEvent) -> CudaResult<()> {
        #[cfg(feature = "cuda")]
        unsafe {
            // cudaStreamWaitEvent would be called here
            // For now, just synchronize
            self.synchronize()?;
        }
        Ok(())
    }
}

impl Default for CudaStream {
    fn default() -> Self {
        Self::default_stream()
    }
}

impl Drop for CudaStream {
    fn drop(&mut self) {
        if !self.is_default && !self.handle.is_null() {
            #[cfg(feature = "cuda")]
            unsafe {
                let _ = ffi::cudaStreamDestroy(self.handle);
            }
        }
    }
}

unsafe impl Send for CudaStream {}
unsafe impl Sync for CudaStream {}

/// CUDA event for synchronization and timing
#[derive(Debug)]
pub struct CudaEvent {
    handle: ffi::cudaEvent_t,
}

impl CudaEvent {
    /// Create new event
    pub fn new() -> CudaResult<Self> {
        #[cfg(feature = "cuda")]
        {
            let mut handle: ffi::cudaEvent_t = std::ptr::null_mut();
            
            unsafe {
                let err = ffi::cudaEventCreate(&mut handle);
                if err != 0 {
                    return Err(CudaError::DriverError(err));
                }
            }
            
            Ok(CudaEvent { handle })
        }
        
        #[cfg(not(feature = "cuda"))]
        {
            Ok(CudaEvent {
                handle: std::ptr::null_mut(),
            })
        }
    }

    /// Record event on stream
    pub fn record(&self, _stream: &CudaStream) -> CudaResult<()> {
        #[cfg(feature = "cuda")]
        unsafe {
            let err = ffi::cudaEventRecord(self.handle, _stream.handle());
            if err != 0 {
                return Err(CudaError::DriverError(err));
            }
        }
        Ok(())
    }

    /// Synchronize on event - wait until event is recorded
    pub fn synchronize(&self) -> CudaResult<()> {
        #[cfg(feature = "cuda")]
        unsafe {
            let err = ffi::cudaEventSynchronize(self.handle);
            if err != 0 {
                return Err(CudaError::SyncError);
            }
        }
        Ok(())
    }

    /// Get elapsed time between two events (in milliseconds)
    pub fn elapsed_time(_start: &CudaEvent, _end: &CudaEvent) -> CudaResult<f32> {
        #[cfg(feature = "cuda")]
        {
            let mut ms: f32 = 0.0;
            
            unsafe {
                let err = ffi::cudaEventElapsedTime(&mut ms, _start.handle, _end.handle);
                if err != 0 {
                    return Err(CudaError::DriverError(err));
                }
            }
            
            Ok(ms)
        }
        
        #[cfg(not(feature = "cuda"))]
        Ok(0.0)
    }
}

impl Default for CudaEvent {
    fn default() -> Self {
        Self::new().unwrap_or(CudaEvent {
            handle: std::ptr::null_mut(),
        })
    }
}

impl Drop for CudaEvent {
    fn drop(&mut self) {
        if !self.handle.is_null() {
            #[cfg(feature = "cuda")]
            unsafe {
                let _ = ffi::cudaEventDestroy(self.handle);
            }
        }
    }
}

unsafe impl Send for CudaEvent {}
unsafe impl Sync for CudaEvent {}

/// Timer utility using CUDA events
pub struct CudaTimer {
    start: CudaEvent,
    stop: CudaEvent,
    stream: CudaStream,
}

impl CudaTimer {
    pub fn new(stream: CudaStream) -> CudaResult<Self> {
        Ok(CudaTimer {
            start: CudaEvent::new()?,
            stop: CudaEvent::new()?,
            stream,
        })
    }

    pub fn start(&self) -> CudaResult<()> {
        self.start.record(&self.stream)
    }

    pub fn stop(&self) -> CudaResult<()> {
        self.stop.record(&self.stream)
    }

    pub fn elapsed_ms(&self) -> CudaResult<f32> {
        self.stop.synchronize()?;
        CudaEvent::elapsed_time(&self.start, &self.stop)
    }
}