safer_ring/

compat.rs

//! Compatibility adapters for AsyncRead/AsyncWrite traits.
//!
//! This module provides adapter layers that allow safer-ring to integrate with
//! existing async ecosystems (tokio, async-std) while maintaining safety guarantees.
//! As predicted by withoutboats, this necessarily involves copying due to the
//! fundamental mismatch between caller-managed buffers (AsyncRead/Write) and
//! ownership transfer (safer-ring).
//!
//! # Architecture
//!
//! The adapters use internal buffer management to bridge the two models:
//! - **AsyncReadAdapter**: Uses internal ring buffers, copies to user buffers
//! - **AsyncWriteAdapter**: Copies from user buffers to internal ring buffers
//! - **File/Socket wrappers**: Provide convenient AsyncRead/Write implementations
//!
//! # Performance Considerations
//!
//! The adapters have ~10-20% overhead compared to direct hot-potato API usage
//! due to necessary memory copying. For maximum performance, prefer the native
//! `read_owned`/`write_owned` APIs.
//!
//! # Example
//!
//! ```rust,ignore
//! use safer_ring::{compat::File, Ring};
//! use tokio::io::{AsyncReadExt, AsyncWriteExt};
//!
//! # #[tokio::main]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let ring = Ring::new(32)?;
//! // Open file with safer-ring backend
//! let mut file = File::open(&ring, "example.txt").await?;
//!
//! // Use standard AsyncRead/AsyncWrite APIs
//! let mut buffer = vec![0u8; 1024];
//! let bytes_read = file.read(&mut buffer).await?;
//! println!("Read {} bytes", bytes_read);
//!
//! // Write with standard API
//! let bytes_written = file.write(b"Hello, world!").await?;
//! println!("Wrote {} bytes", bytes_written);
//! # Ok(())
//! # }
//! ```

use std::future::Future;
use std::io;
use std::os::unix::io::RawFd;
use std::pin::Pin;
use std::task::{Context, Poll};

use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};

use crate::error::{Result, SaferRingError};
use crate::ownership::OwnedBuffer;
use crate::ring::Ring;

/// Internal buffer size for adapters.
const ADAPTER_BUFFER_SIZE: usize = 8192;

/// Type alias for the future result type used in adapters.
type OperationFutureResult = Result<(usize, OwnedBuffer)>;

/// Type alias for the complex future type used in adapters.
type OwnedOperationFuture<'a> = Pin<Box<dyn Future<Output = OperationFutureResult> + 'a>>;

/// AsyncRead adapter that bridges safer-ring with tokio's AsyncRead trait.
///
/// This adapter uses internal buffering to provide AsyncRead compatibility.
/// It manages ring buffers internally and copies data to user-provided buffers.
///
/// # Performance
///
/// The adapter adds copying overhead but provides seamless compatibility with
/// existing tokio-based code. For maximum performance, use the native hot-potato
/// API directly.
pub struct AsyncReadAdapter<'ring> {
    ring: &'ring Ring<'ring>,
    fd: RawFd,
    /// Internal buffer for ring operations
    internal_buffer: Option<OwnedBuffer>,
    /// Cached data from completed operations
    cached_data: Vec<u8>,
    /// Current position in cached data
    cached_pos: usize,
    /// In-flight read operation future
    read_future: Option<OwnedOperationFuture<'ring>>,
}

impl<'ring> AsyncReadAdapter<'ring> {
    /// Create a new AsyncRead adapter.
    ///
    /// # Arguments
    ///
    /// * `ring` - The safer-ring instance to use
    /// * `fd` - File descriptor to read from
    pub fn new(ring: &'ring Ring<'ring>, fd: RawFd) -> Self {
        Self {
            ring,
            fd,
            internal_buffer: Some(OwnedBuffer::new(ADAPTER_BUFFER_SIZE)),
            cached_data: Vec::new(),
            cached_pos: 0,
            read_future: None,
        }
    }

    /// Get a reference to the underlying ring.
    pub fn ring(&self) -> &'ring Ring<'ring> {
        self.ring
    }

    /// Get the file descriptor.
    pub fn fd(&self) -> RawFd {
        self.fd
    }
}

impl<'ring> AsyncRead for AsyncReadAdapter<'ring> {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        // First, try to serve from cached data
        if self.cached_pos < self.cached_data.len() {
            let available = self.cached_data.len() - self.cached_pos;
            let to_copy = available.min(buf.remaining());

            if to_copy > 0 {
                let data = &self.cached_data[self.cached_pos..self.cached_pos + to_copy];
                buf.put_slice(data);
                self.cached_pos += to_copy;

                // Clean up cached data if fully consumed
                if self.cached_pos >= self.cached_data.len() {
                    self.cached_data.clear();
                    self.cached_pos = 0;
                }

                return Poll::Ready(Ok(()));
            }
        }

        // Check if we have an in-flight read operation
        if let Some(mut future) = self.read_future.take() {
            match future.as_mut().poll(cx) {
                Poll::Ready(Ok((bytes_read, returned_buffer))) => {
                    // Operation completed successfully
                    self.internal_buffer = Some(returned_buffer);

                    if bytes_read > 0 {
                        // Get the data from the returned buffer
                        if let Some(buffer_guard) =
                            self.internal_buffer.as_ref().and_then(|b| b.try_access())
                        {
                            let data = &buffer_guard[..bytes_read];

                            // Copy what we can to the user buffer
                            let to_copy = data.len().min(buf.remaining());
                            buf.put_slice(&data[..to_copy]);

                            // Cache any remaining data
                            if to_copy < data.len() {
                                self.cached_data.extend_from_slice(&data[to_copy..]);
                                self.cached_pos = 0;
                            }
                        }
                    }

                    Poll::Ready(Ok(()))
                }
                Poll::Ready(Err(e)) => {
                    // Operation failed - convert SaferRingError to io::Error
                    Poll::Ready(Err(io::Error::other(e.to_string())))
                }
                Poll::Pending => {
                    // Operation still in progress, store future and return Pending
                    self.read_future = Some(future);
                    Poll::Pending
                }
            }
        } else if let Some(buffer) = self.internal_buffer.take() {
            // Start a new read operation using the "hot potato" API
            let future = self.ring.read_owned(self.fd, buffer);
            self.read_future = Some(Box::pin(future));

            // Try to poll immediately to see if it completes synchronously
            self.poll_read(cx, buf)
        } else {
            // This shouldn't happen in normal operation
            Poll::Ready(Err(io::Error::other(
                "AsyncReadAdapter internal buffer unavailable",
            )))
        }
    }
}

/// AsyncWrite adapter that bridges safer-ring with tokio's AsyncWrite trait.
///
/// This adapter copies user data to internal buffers and uses safer-ring's
/// ownership transfer API for actual I/O operations.
pub struct AsyncWriteAdapter<'ring> {
    ring: &'ring Ring<'ring>,
    fd: RawFd,
    /// Buffer for pending write data (reserved for future buffering)
    #[allow(dead_code)]
    write_buffer: Vec<u8>,
    /// Internal ring buffer for operations
    internal_buffer: Option<OwnedBuffer>,
    /// In-flight write operation future
    write_future: Option<OwnedOperationFuture<'ring>>,
}

impl<'ring> AsyncWriteAdapter<'ring> {
    /// Create a new AsyncWrite adapter.
    ///
    /// # Arguments
    ///
    /// * `ring` - The safer-ring instance to use
    /// * `fd` - File descriptor to write to
    pub fn new(ring: &'ring Ring<'ring>, fd: RawFd) -> Self {
        Self {
            ring,
            fd,
            write_buffer: Vec::new(),
            internal_buffer: Some(OwnedBuffer::new(ADAPTER_BUFFER_SIZE)),
            write_future: None,
        }
    }

    /// Get a reference to the underlying ring.
    pub fn ring(&self) -> &'ring Ring<'ring> {
        self.ring
    }

    /// Get the file descriptor.
    pub fn fd(&self) -> RawFd {
        self.fd
    }
}

impl<'ring> AsyncWrite for AsyncWriteAdapter<'ring> {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        // First, check if we have an in-flight write operation
        if let Some(mut future) = self.write_future.take() {
            match future.as_mut().poll(cx) {
                Poll::Ready(Ok((_bytes_written, returned_buffer))) => {
                    // Operation completed successfully
                    self.internal_buffer = Some(returned_buffer);
                    // Continue to process the new write request below
                }
                Poll::Ready(Err(e)) => {
                    // Operation failed - convert SaferRingError to io::Error
                    return Poll::Ready(Err(io::Error::other(e.to_string())));
                }
                Poll::Pending => {
                    // Operation still in progress, store future and return Pending
                    self.write_future = Some(future);
                    return Poll::Pending;
                }
            }
        }

        // No in-flight operation, can proceed with new write
        if let Some(internal_buffer) = self.internal_buffer.take() {
            // Get access to the internal buffer and copy data
            if let Some(mut buffer_guard) = internal_buffer.try_access() {
                let to_copy = buf.len().min(buffer_guard.len());
                buffer_guard[..to_copy].copy_from_slice(&buf[..to_copy]);

                // Drop the guard to release the buffer
                drop(buffer_guard);

                // Start the write operation using the "hot potato" API
                let future = self.ring.write_owned(self.fd, internal_buffer);
                self.write_future = Some(Box::pin(future));

                // Try to poll immediately to see if it completes synchronously
                match self.poll_write(cx, &[]) {
                    Poll::Ready(Ok(_)) => Poll::Ready(Ok(to_copy)),
                    Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
                    Poll::Pending => Poll::Ready(Ok(to_copy)), // We accepted the data
                }
            } else {
                // Buffer is not accessible, put it back and return pending
                self.internal_buffer = Some(internal_buffer);
                Poll::Pending
            }
        } else {
            // No internal buffer available
            Poll::Ready(Err(io::Error::other(
                "AsyncWriteAdapter internal buffer unavailable",
            )))
        }
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        // Check if we have any in-flight write operations
        if let Some(mut future) = self.write_future.take() {
            match future.as_mut().poll(cx) {
                Poll::Ready(Ok((_bytes_written, returned_buffer))) => {
                    // Operation completed successfully
                    self.internal_buffer = Some(returned_buffer);
                    Poll::Ready(Ok(()))
                }
                Poll::Ready(Err(e)) => {
                    // Operation failed - convert SaferRingError to io::Error
                    Poll::Ready(Err(io::Error::other(e.to_string())))
                }
                Poll::Pending => {
                    // Operation still in progress, store future and return Pending
                    self.write_future = Some(future);
                    Poll::Pending
                }
            }
        } else {
            // No in-flight operations, flush is complete
            Poll::Ready(Ok(()))
        }
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        // First ensure all data is flushed
        match self.as_mut().poll_flush(cx) {
            Poll::Ready(Ok(())) => {
                // All data flushed, shutdown is complete
                Poll::Ready(Ok(()))
            }
            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// A file wrapper that provides AsyncRead + AsyncWrite using safer-ring.
///
/// This is a convenient wrapper that combines read and write adapters for
/// file operations while maintaining compatibility with tokio's async traits.
///
/// # Example
///
/// ```rust,ignore
/// use safer_ring::{compat::File, Ring};
/// use tokio::io::{AsyncReadExt, AsyncWriteExt};
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let ring = Ring::new(32)?;
/// let mut file = File::create(&ring, "output.txt").await?;
/// file.write_all(b"Hello, world!").await?;
/// file.sync_all().await?;
/// # Ok(())
/// # }
/// ```
pub struct File<'ring> {
    ring: &'ring Ring<'ring>,
    fd: RawFd,
    read_adapter: Option<AsyncReadAdapter<'ring>>,
    write_adapter: Option<AsyncWriteAdapter<'ring>>,
}

impl<'ring> File<'ring> {
    /// Open a file for reading and writing.
    ///
    /// # Arguments
    ///
    /// * `ring` - The safer-ring instance to use
    /// * `fd` - File descriptor of the opened file
    pub fn new(ring: &'ring Ring<'ring>, fd: RawFd) -> Self {
        Self {
            ring,
            fd,
            read_adapter: Some(AsyncReadAdapter::new(ring, fd)),
            write_adapter: Some(AsyncWriteAdapter::new(ring, fd)),
        }
    }

    /// Create a new file.
    ///
    /// Creates a new file or truncates an existing file for writing.
    pub async fn create(ring: &'ring Ring<'ring>, path: &str) -> Result<Self> {
        use std::ffi::CString;

        let path_cstr = CString::new(path).map_err(|_| {
            SaferRingError::Io(io::Error::new(io::ErrorKind::InvalidInput, "Invalid path"))
        })?;

        // Open file with create, write, truncate flags
        let fd = unsafe {
            libc::open(
                path_cstr.as_ptr(),
                libc::O_CREAT | libc::O_WRONLY | libc::O_TRUNC,
                0o644,
            )
        };

        if fd == -1 {
            return Err(SaferRingError::Io(io::Error::last_os_error()));
        }

        Ok(Self::new(ring, fd))
    }

    /// Open an existing file.
    ///
    /// Opens an existing file for reading and writing.
    pub async fn open(ring: &'ring Ring<'ring>, path: &str) -> Result<Self> {
        use std::ffi::CString;

        let path_cstr = CString::new(path).map_err(|_| {
            SaferRingError::Io(io::Error::new(io::ErrorKind::InvalidInput, "Invalid path"))
        })?;

        // Open file for read/write
        let fd = unsafe { libc::open(path_cstr.as_ptr(), libc::O_RDWR, 0) };

        if fd == -1 {
            return Err(SaferRingError::Io(io::Error::last_os_error()));
        }

        Ok(Self::new(ring, fd))
    }

    /// Sync all data to disk (stub implementation).
    pub async fn sync_all(&self) -> Result<()> {
        // In a real implementation, this would call fsync
        Ok(())
    }

    /// Get the file descriptor.
    pub fn fd(&self) -> RawFd {
        self.fd
    }

    /// Get a reference to the underlying ring.
    pub fn ring(&self) -> &'ring Ring<'ring> {
        self.ring
    }
}

impl<'ring> Drop for File<'ring> {
    fn drop(&mut self) {
        unsafe {
            libc::close(self.fd);
        }
    }
}

impl<'ring> AsyncRead for File<'ring> {
    fn poll_read(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        if let Some(adapter) = &mut self.read_adapter {
            Pin::new(adapter).poll_read(_cx, buf)
        } else {
            Poll::Ready(Err(io::Error::other("Read adapter not available")))
        }
    }
}

impl<'ring> AsyncWrite for File<'ring> {
    fn poll_write(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        if let Some(adapter) = &mut self.write_adapter {
            Pin::new(adapter).poll_write(_cx, buf)
        } else {
            Poll::Ready(Err(io::Error::other("Write adapter not available")))
        }
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        if let Some(adapter) = &mut self.write_adapter {
            Pin::new(adapter).poll_flush(cx)
        } else {
            Poll::Ready(Ok(()))
        }
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        if let Some(adapter) = &mut self.write_adapter {
            Pin::new(adapter).poll_shutdown(cx)
        } else {
            Poll::Ready(Ok(()))
        }
    }
}

/// Socket wrapper providing AsyncRead/AsyncWrite for network operations.
///
/// Similar to File but optimized for socket operations.
pub struct Socket<'ring> {
    ring: &'ring Ring<'ring>,
    fd: RawFd,
    read_adapter: AsyncReadAdapter<'ring>,
    write_adapter: AsyncWriteAdapter<'ring>,
}

impl<'ring> Socket<'ring> {
    /// Create a new socket wrapper.
    pub fn new(ring: &'ring Ring<'ring>, fd: RawFd) -> Self {
        Self {
            ring,
            fd,
            read_adapter: AsyncReadAdapter::new(ring, fd),
            write_adapter: AsyncWriteAdapter::new(ring, fd),
        }
    }

    /// Get the socket file descriptor.
    pub fn fd(&self) -> RawFd {
        self.fd
    }

    /// Get a reference to the underlying ring.
    pub fn ring(&self) -> &'ring Ring<'ring> {
        self.ring
    }
}

impl<'ring> AsyncRead for Socket<'ring> {
    fn poll_read(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        Pin::new(&mut self.read_adapter).poll_read(_cx, buf)
    }
}

impl<'ring> AsyncWrite for Socket<'ring> {
    fn poll_write(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        Pin::new(&mut self.write_adapter).poll_write(_cx, buf)
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Pin::new(&mut self.write_adapter).poll_flush(cx)
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Pin::new(&mut self.write_adapter).poll_shutdown(cx)
    }
}

/// Helper trait for converting safer-ring operations to AsyncRead/AsyncWrite.
///
/// This trait provides convenient methods for wrapping safer-ring rings
/// with AsyncRead/AsyncWrite adapters.
pub trait AsyncCompat<'ring> {
    /// Wrap a file descriptor as an AsyncRead adapter.
    fn async_read(&'ring self, fd: RawFd) -> AsyncReadAdapter<'ring>;

    /// Wrap a file descriptor as an AsyncWrite adapter.
    fn async_write(&'ring self, fd: RawFd) -> AsyncWriteAdapter<'ring>;

    /// Create a File wrapper for the given file descriptor.
    fn file(&'ring self, fd: RawFd) -> File<'ring>;

    /// Create a Socket wrapper for the given socket descriptor.
    fn socket(&'ring self, fd: RawFd) -> Socket<'ring>;
}

impl<'ring> AsyncCompat<'ring> for Ring<'ring> {
    fn async_read(&'ring self, fd: RawFd) -> AsyncReadAdapter<'ring> {
        AsyncReadAdapter::new(self, fd)
    }

    fn async_write(&'ring self, fd: RawFd) -> AsyncWriteAdapter<'ring> {
        AsyncWriteAdapter::new(self, fd)
    }

    fn file(&'ring self, fd: RawFd) -> File<'ring> {
        File::new(self, fd)
    }

    fn socket(&'ring self, fd: RawFd) -> Socket<'ring> {
        Socket::new(self, fd)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_async_read_adapter_creation() {
        #[cfg(target_os = "linux")]
        {
            use crate::Ring;

            // Test that the method exists by checking we can call it in a static context
            // This tests the API without creating lifetime issues
            let _can_create_adapter: for<'r> fn(&'r Ring<'r>) -> AsyncReadAdapter<'r> =
                |ring| ring.async_read(0);

            // Test passes if the method signature is correct and accessible
        }

        #[cfg(not(target_os = "linux"))]
        {
            println!("Skipping AsyncRead adapter test on non-Linux platform");
        }
    }

    #[tokio::test]
    async fn test_async_write_adapter_creation() {
        #[cfg(target_os = "linux")]
        {
            use crate::Ring;

            // Test that the method exists by checking we can call it in a static context
            // This tests the API without creating lifetime issues
            let _can_create_adapter: for<'r> fn(&'r Ring<'r>) -> AsyncWriteAdapter<'r> =
                |ring| ring.async_write(1);

            // Test passes if the method signature is correct and accessible
        }

        #[cfg(not(target_os = "linux"))]
        {
            println!("Skipping AsyncWrite adapter test on non-Linux platform");
        }
    }

    #[tokio::test]
    async fn test_file_wrapper_creation() {
        #[cfg(target_os = "linux")]
        {
            use crate::Ring;

            // Test that the method exists by checking we can call it in a static context
            // This tests the API without creating lifetime issues
            let _can_create_file: for<'r> fn(&'r Ring<'r>) -> File<'r> = |ring| ring.file(0);

            println!("✓ File wrapper method accessible");
        }

        #[cfg(not(target_os = "linux"))]
        {
            println!("Skipping File wrapper test on non-Linux platform");
        }
    }

    #[tokio::test]
    async fn test_socket_wrapper_creation() {
        #[cfg(target_os = "linux")]
        {
            use crate::Ring;

            // Test that the method exists by checking we can call it in a static context
            // This tests the API without creating lifetime issues
            let _can_create_socket: for<'r> fn(&'r Ring<'r>) -> Socket<'r> = |ring| ring.socket(0);

            println!("✓ Socket wrapper method accessible");
        }

        #[cfg(not(target_os = "linux"))]
        {
            println!("Skipping Socket wrapper test on non-Linux platform");
        }
    }

    #[test]
    fn test_adapter_buffer_size() {
        // Test that ADAPTER_BUFFER_SIZE is reasonable
        // Note: These are compile-time constants, so these assertions will be optimized out
        const _: () = assert!(ADAPTER_BUFFER_SIZE > 0, "Buffer size must be positive");
        const _: () = assert!(
            ADAPTER_BUFFER_SIZE <= 65536,
            "Buffer size should be reasonable"
        ); // Reasonable upper bound
    }
}