feat(fs, unix): use AsyncFd in stdio

Author: Berrysoft

compio-fs/src/async_fd.rs

@@ -13,6 +13,11 @@ use compio_driver::{
 };
 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`].
@@ -46,6 +51,12 @@ impl<T: AsRawFd + 'static> AsyncRead for AsyncFd<T> {
     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> {
@@ -58,6 +69,17 @@ impl<T: AsRawFd + 'static> AsyncRead for &AsyncFd<T> {
             .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> {
@@ -66,6 +88,12 @@ impl<T: AsRawFd + 'static> AsyncWrite for AsyncFd<T> {
         (&*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
@@ -84,6 +112,13 @@ impl<T: AsRawFd + 'static> AsyncWrite for &AsyncFd<T> {
         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(())
     }

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) })
     }
 }