Merge pull request #224 from Berrysoft/dev/process

Author: George-Miao

Cargo.toml

@@ -12,6 +12,7 @@ members = [
     "compio-io",
     "compio-tls",
     "compio-log",
+    "compio-process",
 ]
 resolver = "2"
 
@@ -34,6 +35,7 @@ compio-signal = { path = "./compio-signal", version = "0.2.0" }
 compio-dispatcher = { path = "./compio-dispatcher", version = "0.2.0" }
 compio-log = { path = "./compio-log", version = "0.1.0" }
 compio-tls = { path = "./compio-tls", version = "0.2.0", default-features = false }
+compio-process = { path = "./compio-process", version = "0.1.0" }
 
 flume = "0.11.0"
 cfg-if = "1.0.0"

compio-driver/src/iocp/cp/mod.rs

@@ -22,9 +22,9 @@ use compio_buf::arrayvec::ArrayVec;
 use compio_log::*;
 use windows_sys::Win32::{
     Foundation::{
-        RtlNtStatusToDosError, ERROR_BAD_COMMAND, ERROR_HANDLE_EOF, ERROR_IO_INCOMPLETE,
-        ERROR_NO_DATA, FACILITY_NTWIN32, INVALID_HANDLE_VALUE, NTSTATUS, STATUS_PENDING,
-        STATUS_SUCCESS,
+        RtlNtStatusToDosError, ERROR_BAD_COMMAND, ERROR_BROKEN_PIPE, ERROR_HANDLE_EOF,
+        ERROR_IO_INCOMPLETE, ERROR_NO_DATA, ERROR_PIPE_CONNECTED, ERROR_PIPE_NOT_CONNECTED,
+        FACILITY_NTWIN32, INVALID_HANDLE_VALUE, NTSTATUS, STATUS_PENDING, STATUS_SUCCESS,
     },
     Storage::FileSystem::SetFileCompletionNotificationModes,
     System::{
@@ -173,7 +173,12 @@ impl CompletionPort {
             } else {
                 let error = unsafe { RtlNtStatusToDosError(overlapped.base.Internal as _) };
                 match error {
-                    ERROR_IO_INCOMPLETE | ERROR_HANDLE_EOF | ERROR_NO_DATA => Ok(0),
+                    ERROR_IO_INCOMPLETE
+                    | ERROR_HANDLE_EOF
+                    | ERROR_BROKEN_PIPE
+                    | ERROR_PIPE_CONNECTED
+                    | ERROR_PIPE_NOT_CONNECTED
+                    | ERROR_NO_DATA => Ok(0),
                     _ => Err(io::Error::from_raw_os_error(error as _)),
                 }
             };

compio-driver/src/iocp/mod.rs

@@ -97,6 +97,24 @@ impl AsRawFd for OwnedSocket {
     }
 }
 
+impl AsRawFd for std::process::ChildStdin {
+    fn as_raw_fd(&self) -> RawFd {
+        self.as_raw_handle() as _
+    }
+}
+
+impl AsRawFd for std::process::ChildStdout {
+    fn as_raw_fd(&self) -> RawFd {
+        self.as_raw_handle() as _
+    }
+}
+
+impl AsRawFd for std::process::ChildStderr {
+    fn as_raw_fd(&self) -> RawFd {
+        self.as_raw_handle() as _
+    }
+}
+
 impl From<OwnedHandle> for OwnedFd {
     fn from(value: OwnedHandle) -> Self {
         Self::File(value)
@@ -109,6 +127,24 @@ impl From<std::fs::File> for OwnedFd {
     }
 }
 
+impl From<std::process::ChildStdin> for OwnedFd {
+    fn from(value: std::process::ChildStdin) -> Self {
+        Self::File(OwnedHandle::from(value))
+    }
+}
+
+impl From<std::process::ChildStdout> for OwnedFd {
+    fn from(value: std::process::ChildStdout) -> Self {
+        Self::File(OwnedHandle::from(value))
+    }
+}
+
+impl From<std::process::ChildStderr> for OwnedFd {
+    fn from(value: std::process::ChildStderr) -> Self {
+        Self::File(OwnedHandle::from(value))
+    }
+}
+
 impl From<OwnedSocket> for OwnedFd {
     fn from(value: OwnedSocket) -> Self {
         Self::Socket(value)

compio-driver/src/iocp/op.rs

@@ -19,8 +19,9 @@ use windows_sys::{
     core::GUID,
     Win32::{
         Foundation::{
-            CloseHandle, GetLastError, ERROR_HANDLE_EOF, ERROR_IO_INCOMPLETE, ERROR_IO_PENDING,
-            ERROR_NOT_FOUND, ERROR_NO_DATA, ERROR_PIPE_CONNECTED,
+            CloseHandle, GetLastError, ERROR_BROKEN_PIPE, ERROR_HANDLE_EOF, ERROR_IO_INCOMPLETE,
+            ERROR_IO_PENDING, ERROR_NOT_FOUND, ERROR_NO_DATA, ERROR_PIPE_CONNECTED,
+            ERROR_PIPE_NOT_CONNECTED,
         },
         Networking::WinSock::{
             closesocket, setsockopt, shutdown, socklen_t, WSAIoctl, WSARecv, WSARecvFrom, WSASend,
@@ -45,9 +46,12 @@ fn winapi_result(transferred: u32) -> Poll<io::Result<usize>> {
     assert_ne!(error, 0);
     match error {
         ERROR_IO_PENDING => Poll::Pending,
-        ERROR_IO_INCOMPLETE | ERROR_HANDLE_EOF | ERROR_PIPE_CONNECTED | ERROR_NO_DATA => {
-            Poll::Ready(Ok(transferred as _))
-        }
+        ERROR_IO_INCOMPLETE
+        | ERROR_HANDLE_EOF
+        | ERROR_BROKEN_PIPE
+        | ERROR_PIPE_CONNECTED
+        | ERROR_PIPE_NOT_CONNECTED
+        | ERROR_NO_DATA => Poll::Ready(Ok(transferred as _)),
         _ => Poll::Ready(Err(io::Error::from_raw_os_error(error as _))),
     }
 }

compio-driver/src/iour/op.rs

@@ -555,3 +555,15 @@ impl<T: IoVectoredBuf, S> IntoInner for SendToVectored<T, S> {
         self.buffer
     }
 }
+
+impl<S: AsRawFd> OpCode for PollOnce<S> {
+    fn create_entry(self: Pin<&mut Self>) -> OpEntry {
+        let flags = match self.interest {
+            Interest::Readable => libc::POLLIN,
+            Interest::Writable => libc::POLLOUT,
+        };
+        opcode::PollAdd::new(Fd(self.fd.as_raw_fd()), flags as _)
+            .build()
+            .into()
+    }
+}

compio-driver/src/op.rs

@@ -16,8 +16,8 @@ pub use crate::sys::op::{
 };
 #[cfg(unix)]
 pub use crate::sys::op::{
-    CreateDir, CreateSocket, FileStat, HardLink, OpenFile, PathStat, ReadVectoredAt, Rename,
-    Symlink, Unlink, WriteVectoredAt,
+    CreateDir, CreateSocket, FileStat, HardLink, Interest, OpenFile, PathStat, PollOnce,
+    ReadVectoredAt, Rename, Symlink, Unlink, WriteVectoredAt,
 };
 use crate::{
     sys::{sockaddr_storage, socklen_t},

compio-driver/src/poll/mod.rs

@@ -17,7 +17,7 @@ use crossbeam_queue::SegQueue;
 pub(crate) use libc::{sockaddr_storage, socklen_t};
 use polling::{Event, Events, PollMode, Poller};
 
-use crate::{syscall, AsyncifyPool, Entry, Key, OutEntries, ProactorBuilder};
+use crate::{op::Interest, syscall, AsyncifyPool, Entry, Key, OutEntries, ProactorBuilder};
 
 pub(crate) mod op;
 
@@ -84,15 +84,6 @@ pub struct WaitArg {
     pub interest: Interest,
 }
 
-/// The interest of the operation
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum Interest {
-    /// Represents a read operation.
-    Readable,
-    /// Represents a write operation.
-    Writable,
-}
-
 #[derive(Debug, Default)]
 struct FdQueue {
     read_queue: VecDeque<usize>,

compio-driver/src/poll/op.rs

@@ -748,3 +748,18 @@ impl<T: IoVectoredBuf, S> IntoInner for SendToVectored<T, S> {
         self.buffer
     }
 }
+
+impl<S: AsRawFd> OpCode for PollOnce<S> {
+    fn pre_submit(self: Pin<&mut Self>) -> io::Result<Decision> {
+        Ok(Decision::wait_for(self.fd.as_raw_fd(), self.interest))
+    }
+
+    fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
+        match self.interest {
+            Interest::Readable => debug_assert!(event.readable),
+            Interest::Writable => debug_assert!(event.writable),
+        }
+
+        Poll::Ready(Ok(0))
+    }
+}

compio-driver/src/unix/op.rs

@@ -369,3 +369,25 @@ impl<T: IoVectoredBuf, S> IntoInner for SendVectored<T, S> {
         self.buffer
     }
 }
+
+/// The interest to poll a file descriptor.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Interest {
+    /// Represents a read operation.
+    Readable,
+    /// Represents a write operation.
+    Writable,
+}
+
+/// Poll a file descriptor for specified [`Interest`].
+pub struct PollOnce<S> {
+    pub(crate) fd: SharedFd<S>,
+    pub(crate) interest: Interest,
+}
+
+impl<S> PollOnce<S> {
+    /// Create [`PollOnce`].
+    pub fn new(fd: SharedFd<S>, interest: Interest) -> Self {
+        Self { fd, interest }
+    }
+}

compio-fs/src/named_pipe.rs

@@ -168,8 +168,8 @@ impl NamedPipeServer {
     ///
     /// // Write fails with an OS-specific error after client has been
     /// // disconnected.
-    /// let e = client.write("ping").await.0.unwrap_err();
-    /// assert_eq!(e.raw_os_error(), Some(ERROR_PIPE_NOT_CONNECTED as i32));
+    /// let e = client.write("ping").await.0.unwrap();
+    /// assert_eq!(e, 0);
     /// # })
     /// ```
     pub fn disconnect(&self) -> io::Result<()> {