Merge pull request #248 from Berrysoft:dev/async-fd

feat(fs): AsyncFd

Author: George-Miao

compio-driver/src/lib.rs

@@ -138,6 +138,12 @@ macro_rules! impl_raw_fd {
                 }
             }
         }
+        #[cfg(windows)]
+        impl std::os::windows::io::AsRawHandle for $t {
+            fn as_raw_handle(&self) -> std::os::windows::io::RawHandle {
+                self.$inner.as_raw_handle()
+            }
+        }
     };
     ($t:ty, $it:ty, $inner:ident,socket) => {
         $crate::impl_raw_fd!($t, $it, $inner);
@@ -149,6 +155,12 @@ macro_rules! impl_raw_fd {
                 }
             }
         }
+        #[cfg(windows)]
+        impl std::os::windows::io::AsRawSocket for $t {
+            fn as_raw_socket(&self) -> std::os::windows::io::RawSocket {
+                self.$inner.as_raw_socket()
+            }
+        }
     };
 }
 

compio-fs/src/async_fd.rs

@@ -0,0 +1,192 @@
+use std::io;
+#[cfg(unix)]
+use std::os::fd::{FromRawFd, RawFd};
+#[cfg(windows)]
+use std::os::windows::io::{
+    AsRawHandle, AsRawSocket, FromRawHandle, FromRawSocket, RawHandle, RawSocket,
+};
+
+use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut};
+use compio_driver::{
+    op::{BufResultExt, Recv, Send},
+    AsRawFd, SharedFd, ToSharedFd,
+};
+use compio_io::{AsyncRead, AsyncWrite};
+use compio_runtime::{Attacher, Runtime};
+#[cfg(unix)]
+use {
+    compio_buf::{IoVectoredBuf, IoVectoredBufMut},
+    compio_driver::op::{RecvVectored, SendVectored},
+};
+
+/// A wrapper for IO source, providing implementations for [`AsyncRead`] and
+/// [`AsyncWrite`].
+#[derive(Debug)]
+pub struct AsyncFd<T: AsRawFd> {
+    inner: Attacher<T>,
+}
+
+impl<T: AsRawFd> AsyncFd<T> {
+    /// Create [`AsyncFd`] and attach the source to the current runtime.
+    pub fn new(source: T) -> io::Result<Self> {
+        Ok(Self {
+            inner: Attacher::new(source)?,
+        })
+    }
+
+    /// Create [`AsyncFd`] without attaching the source.
+    ///
+    /// # Safety
+    ///
+    /// The user should handle the attachment correctly.
+    pub unsafe fn new_unchecked(source: T) -> Self {
+        Self {
+            inner: Attacher::new_unchecked(source),
+        }
+    }
+}
+
+impl<T: AsRawFd + 'static> AsyncRead for AsyncFd<T> {
+    #[inline]
+    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
+        (&*self).read(buf).await
+    }
+
+    #[cfg(unix)]
+    #[inline]
+    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
+        (&*self).read_vectored(buf).await
+    }
+}
+
+impl<T: AsRawFd + 'static> AsyncRead for &AsyncFd<T> {
+    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
+        let fd = self.inner.to_shared_fd();
+        let op = Recv::new(fd, buf);
+        Runtime::current()
+            .submit(op)
+            .await
+            .into_inner()
+            .map_advanced()
+    }
+
+    #[cfg(unix)]
+    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
+        let fd = self.inner.to_shared_fd();
+        let op = RecvVectored::new(fd, buf);
+        Runtime::current()
+            .submit(op)
+            .await
+            .into_inner()
+            .map_advanced()
+    }
+}
+
+impl<T: AsRawFd + 'static> AsyncWrite for AsyncFd<T> {
+    #[inline]
+    async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
+        (&*self).write(buf).await
+    }
+
+    #[cfg(unix)]
+    #[inline]
+    async fn write_vectored<V: IoVectoredBuf>(&mut self, buf: V) -> BufResult<usize, V> {
+        (&*self).write_vectored(buf).await
+    }
+
+    #[inline]
+    async fn flush(&mut self) -> io::Result<()> {
+        (&*self).flush().await
+    }
+
+    #[inline]
+    async fn shutdown(&mut self) -> io::Result<()> {
+        (&*self).shutdown().await
+    }
+}
+
+impl<T: AsRawFd + 'static> AsyncWrite for &AsyncFd<T> {
+    async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
+        let fd = self.inner.to_shared_fd();
+        let op = Send::new(fd, buf);
+        Runtime::current().submit(op).await.into_inner()
+    }
+
+    #[cfg(unix)]
+    async fn write_vectored<V: IoVectoredBuf>(&mut self, buf: V) -> BufResult<usize, V> {
+        let fd = self.inner.to_shared_fd();
+        let op = SendVectored::new(fd, buf);
+        Runtime::current().submit(op).await.into_inner()
+    }
+
+    async fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+
+    async fn shutdown(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl<T: AsRawFd> IntoInner for AsyncFd<T> {
+    type Inner = SharedFd<T>;
+
+    fn into_inner(self) -> Self::Inner {
+        self.inner.into_inner()
+    }
+}
+
+impl<T: AsRawFd> AsRawFd for AsyncFd<T> {
+    fn as_raw_fd(&self) -> compio_driver::RawFd {
+        self.inner.as_raw_fd()
+    }
+}
+
+#[cfg(windows)]
+impl<T: AsRawFd + AsRawHandle> AsRawHandle for AsyncFd<T> {
+    fn as_raw_handle(&self) -> RawHandle {
+        self.inner.as_raw_handle()
+    }
+}
+
+#[cfg(windows)]
+impl<T: AsRawFd + AsRawSocket> AsRawSocket for AsyncFd<T> {
+    fn as_raw_socket(&self) -> RawSocket {
+        self.inner.as_raw_socket()
+    }
+}
+
+impl<T: AsRawFd> ToSharedFd<T> for AsyncFd<T> {
+    fn to_shared_fd(&self) -> SharedFd<T> {
+        self.inner.to_shared_fd()
+    }
+}
+
+impl<T: AsRawFd> Clone for AsyncFd<T> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+#[cfg(unix)]
+impl<T: AsRawFd + FromRawFd> FromRawFd for AsyncFd<T> {
+    unsafe fn from_raw_fd(fd: RawFd) -> Self {
+        Self::new_unchecked(FromRawFd::from_raw_fd(fd))
+    }
+}
+
+#[cfg(windows)]
+impl<T: AsRawFd + FromRawHandle> FromRawHandle for AsyncFd<T> {
+    unsafe fn from_raw_handle(handle: RawHandle) -> Self {
+        Self::new_unchecked(FromRawHandle::from_raw_handle(handle))
+    }
+}
+
+#[cfg(windows)]
+impl<T: AsRawFd + FromRawSocket> FromRawSocket for AsyncFd<T> {
+    unsafe fn from_raw_socket(sock: RawSocket) -> Self {
+        Self::new_unchecked(FromRawSocket::from_raw_socket(sock))
+    }
+}

compio-fs/src/lib.rs

@@ -23,6 +23,9 @@ pub use stdio::*;
 mod utils;
 pub use utils::*;
 
+mod async_fd;
+pub use async_fd::*;
+
 #[cfg(windows)]
 pub mod named_pipe;
 

compio-fs/src/named_pipe.rs

@@ -26,7 +26,7 @@ use windows_sys::Win32::{
     },
 };
 
-use crate::{File, OpenOptions};
+use crate::{AsyncFd, File, OpenOptions};
 
 /// A [Windows named pipe] server.
 ///
@@ -88,7 +88,7 @@ use crate::{File, OpenOptions};
 /// [Windows named pipe]: https://docs.microsoft.com/en-us/windows/win32/ipc/named-pipes
 #[derive(Debug, Clone)]
 pub struct NamedPipeServer {
-    handle: File,
+    handle: AsyncFd<std::fs::File>,
 }
 
 impl NamedPipeServer {
@@ -189,8 +189,7 @@ impl AsyncRead for NamedPipeServer {
 impl AsyncRead for &NamedPipeServer {
     #[inline]
     async fn read<B: IoBufMut>(&mut self, buffer: B) -> BufResult<usize, B> {
-        // The position is ignored.
-        self.handle.read_at(buffer, 0).await
+        (&self.handle).read(buffer).await
     }
 }
 
@@ -214,8 +213,7 @@ impl AsyncWrite for NamedPipeServer {
 impl AsyncWrite for &NamedPipeServer {
     #[inline]
     async fn write<T: IoBuf>(&mut self, buffer: T) -> BufResult<usize, T> {
-        // The position is ignored.
-        (&self.handle).write_at(buffer, 0).await
+        (&self.handle).write(buffer).await
     }
 
     #[inline]
@@ -974,7 +972,7 @@ impl ServerOptions {
         )?;
 
         Ok(NamedPipeServer {
-            handle: File::from_std(unsafe { std::fs::File::from_raw_handle(h as _) })?,
+            handle: AsyncFd::new(unsafe { std::fs::File::from_raw_handle(h as _) })?,
         })
     }
 }

compio-fs/src/stdio/unix.rs

@@ -1,35 +1,43 @@
-use std::{io, mem::ManuallyDrop, os::fd::FromRawFd};
+use std::io;
 
 use compio_buf::{BufResult, IoBuf, IoBufMut, IoVectoredBuf, IoVectoredBufMut};
 use compio_driver::{AsRawFd, RawFd};
 use compio_io::{AsyncRead, AsyncWrite};
 
 #[cfg(doc)]
 use super::{stderr, stdin, stdout};
-use crate::pipe::{Receiver, Sender};
+use crate::AsyncFd;
 
 /// A handle to the standard input stream of a process.
 ///
 /// See [`stdin`].
 #[derive(Debug, Clone)]
-pub struct Stdin(ManuallyDrop<Receiver>);
+pub struct Stdin(AsyncFd<RawFd>);
 
 impl Stdin {
     pub(crate) fn new() -> Self {
-        // SAFETY: we don't drop it
-        Self(ManuallyDrop::new(unsafe {
-            Receiver::from_raw_fd(libc::STDIN_FILENO)
-        }))
+        // SAFETY: no need to attach on unix
+        Self(unsafe { AsyncFd::new_unchecked(libc::STDIN_FILENO) })
     }
 }
 
 impl AsyncRead for Stdin {
     async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
-        self.0.read(buf).await
+        (&*self).read(buf).await
     }
 
     async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
-        self.0.read_vectored(buf).await
+        (&*self).read_vectored(buf).await
+    }
+}
+
+impl AsyncRead for &Stdin {
+    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
+        (&self.0).read(buf).await
+    }
+
+    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
+        (&self.0).read_vectored(buf).await
     }
 }
 
@@ -43,14 +51,12 @@ impl AsRawFd for Stdin {
 ///
 /// See [`stdout`].
 #[derive(Debug, Clone)]
-pub struct Stdout(ManuallyDrop<Sender>);
+pub struct Stdout(AsyncFd<RawFd>);
 
 impl Stdout {
     pub(crate) fn new() -> Self {
-        // SAFETY: we don't drop it
-        Self(ManuallyDrop::new(unsafe {
-            Sender::from_raw_fd(libc::STDOUT_FILENO)
-        }))
+        // SAFETY: no need to attach on unix
+        Self(unsafe { AsyncFd::new_unchecked(libc::STDOUT_FILENO) })
     }
 }
 
@@ -82,14 +88,12 @@ impl AsRawFd for Stdout {
 ///
 /// See [`stderr`].
 #[derive(Debug, Clone)]
-pub struct Stderr(ManuallyDrop<Sender>);
+pub struct Stderr(AsyncFd<RawFd>);
 
 impl Stderr {
     pub(crate) fn new() -> Self {
-        // SAFETY: we don't drop it
-        Self(ManuallyDrop::new(unsafe {
-            Sender::from_raw_fd(libc::STDERR_FILENO)
-        }))
+        // SAFETY: no need to attach on unix
+        Self(unsafe { AsyncFd::new_unchecked(libc::STDERR_FILENO) })
     }
 }