refactor(driver): handle panicking

Author: George-Miao

compio-driver/src/asyncify.rs

@@ -76,7 +76,7 @@ fn worker(
         counter.fetch_add(1, Ordering::AcqRel);
         let _guard = CounterGuard(counter);
         while let Ok(f) = receiver.recv_timeout(timeout) {
-            f.run();
+            f.run()
         }
     }
 }

compio-driver/src/key.rs

@@ -60,14 +60,17 @@ impl<C: crate::Carry + ?Sized> RawOp<C> {
     ///
     /// [`Poll::Pending`]: std::task::Poll::Pending
     pub fn operate_blocking(&mut self) -> io::Result<usize> {
-        use std::task::Poll;
+        use std::{panic::AssertUnwindSafe, task::Poll};
+
+        use crate::panic::catch_unwind_io;
 
         let optr = self.extra_mut().optr();
-        let res = unsafe { self.carrier.operate(optr.cast()) };
-        match res {
-            Poll::Pending => unreachable!("this operation is not overlapped"),
-            Poll::Ready(res) => res,
-        }
+        catch_unwind_io(AssertUnwindSafe(|| unsafe {
+            match self.carrier.operate(optr.cast()) {
+                Poll::Pending => unreachable!("this operation is not overlapped"),
+                Poll::Ready(res) => res,
+            }
+        }))
     }
 }
 

compio-driver/src/lib.rs

@@ -26,6 +26,7 @@ use compio_log::instrument;
 
 mod control;
 mod macros;
+mod panic;
 
 mod key;
 pub use key::Key;
@@ -55,6 +56,7 @@ use crate::{
     buffer_pool::{BufferAlloc, BufferPoolRoot},
     key::ErasedKey,
     op::OpCodeFlag,
+    panic::resume_unwind_io,
 };
 
 mod sys_slice;
@@ -212,7 +214,8 @@ impl Proactor {
         }
         self.cancel.remove(&key);
         if key.is_unique() && key.has_result() {
-            Some(key.take_result())
+            let (res, buf) = key.take_result().into_parts();
+            Some(BufResult(resume_unwind_io(res), buf))
         } else {
             self.driver.cancel(key.erase());
             None
@@ -286,12 +289,14 @@ impl Proactor {
     ///
     /// # Panics
     ///
-    /// This function will panic if the [`Key`] is not unique.
+    /// This function will panic if the [`Key`] is not unique or if the
+    /// operation is blocking and it panicked in the thread pool.
     pub fn pop<T: OpCode>(&mut self, key: Key<T>) -> PushEntry<Key<T>, BufResult<usize, T>> {
         instrument!(compio_log::Level::DEBUG, "pop", ?key);
         if key.has_result() {
             self.cancel.remove(&key);
-            PushEntry::Ready(key.take_result())
+            let (res, buf) = key.take_result().into_parts();
+            PushEntry::Ready(BufResult(resume_unwind_io(res), buf))
         } else {
             PushEntry::Pending(key)
         }
@@ -302,7 +307,8 @@ impl Proactor {
     ///
     /// # Panics
     ///
-    /// This function will panic if the [`Key`] is not unique.
+    /// This function will panic if the [`Key`] is not unique or if the
+    /// operation is blocking and it panicked in the thread pool.
     pub fn pop_with_extra<T: OpCode>(
         &mut self,
         key: Key<T>,
@@ -311,8 +317,8 @@ impl Proactor {
         if key.has_result() {
             self.cancel.remove(&key);
             let extra = key.swap_extra(self.default_extra());
-            let res = key.take_result();
-            PushEntry::Ready((res, extra))
+            let (res, buf) = key.take_result().into_parts();
+            PushEntry::Ready((BufResult(resume_unwind_io(res), buf), extra))
         } else {
             PushEntry::Pending(key)
         }

compio-driver/src/panic.rs

@@ -0,0 +1,47 @@
+use std::{
+    any::Any,
+    error::Error,
+    fmt::{Debug, Display},
+    io,
+    panic::{UnwindSafe, catch_unwind, resume_unwind},
+};
+
+pub(crate) struct Panic(Box<dyn Any + Send>);
+
+// Panic is unconditionally `Sync`
+unsafe impl Sync for Panic {}
+
+impl Display for Panic {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        Debug::fmt(self, f)
+    }
+}
+
+impl Debug for Panic {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("Panic").finish_non_exhaustive()
+    }
+}
+
+impl Error for Panic {}
+
+impl From<Panic> for io::Error {
+    fn from(value: Panic) -> Self {
+        Self::other(value)
+    }
+}
+
+pub(crate) fn catch_unwind_io<F, R>(f: F) -> io::Result<R>
+where
+    F: FnOnce() -> io::Result<R> + UnwindSafe,
+{
+    catch_unwind(f).map_err(|err| io::Error::from(Panic(err)))?
+}
+
+pub(crate) fn resume_unwind_io<T>(res: io::Result<T>) -> io::Result<T> {
+    let Err(e) = res else { return res };
+    match e.downcast::<Panic>() {
+        Ok(p) => resume_unwind(p.0),
+        Err(e) => Err(e),
+    }
+}

compio-driver/src/sys/iocp/cp/global.rs

@@ -103,6 +103,7 @@ impl Port {
         self.global_port.attach(fd)
     }
 
+    #[allow(dead_code)]
     pub fn post(&self, res: io::Result<usize>, optr: *mut Overlapped) -> io::Result<()> {
         self.port.post(res, optr)
     }

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

@@ -83,6 +83,7 @@ impl CompletionPort {
         Ok(())
     }
 
+    #[allow(dead_code)]
     pub fn post(&self, res: io::Result<usize>, optr: *mut Overlapped) -> io::Result<()> {
         if let Some(overlapped) = unsafe { optr.as_mut() } {
             match &res {

compio-driver/src/sys/iocp/cp/multi.rs

@@ -22,6 +22,7 @@ impl Port {
         self.port.attach(fd)
     }
 
+    #[allow(dead_code)]
     pub fn post(&self, res: io::Result<usize>, optr: *mut Overlapped) -> io::Result<()> {
         self.port.post(res, optr)
     }

compio-driver/src/sys/iocp/fd.rs

@@ -0,0 +1,219 @@
+use super::*;
+
+/// On windows, handle and socket are in the same size.
+/// Both of them could be attached to an IOCP.
+/// Therefore, both could be seen as fd.
+pub type RawFd = HANDLE;
+
+/// Extracts raw fds.
+pub trait AsRawFd {
+    /// Extracts the raw fd.
+    fn as_raw_fd(&self) -> RawFd;
+}
+
+/// Owned handle or socket on Windows.
+#[derive(Debug)]
+pub enum OwnedFd {
+    /// Win32 handle.
+    File(OwnedHandle),
+    /// Windows socket handle.
+    Socket(OwnedSocket),
+}
+
+impl AsRawFd for OwnedFd {
+    fn as_raw_fd(&self) -> RawFd {
+        match self {
+            Self::File(fd) => fd.as_raw_handle() as _,
+            Self::Socket(s) => s.as_raw_socket() as _,
+        }
+    }
+}
+
+impl AsRawFd for RawFd {
+    fn as_raw_fd(&self) -> RawFd {
+        *self
+    }
+}
+
+impl AsRawFd for std::fs::File {
+    fn as_raw_fd(&self) -> RawFd {
+        self.as_raw_handle() as _
+    }
+}
+
+impl AsRawFd for OwnedHandle {
+    fn as_raw_fd(&self) -> RawFd {
+        self.as_raw_handle() as _
+    }
+}
+
+impl AsRawFd for socket2::Socket {
+    fn as_raw_fd(&self) -> RawFd {
+        self.as_raw_socket() as _
+    }
+}
+
+impl AsRawFd for OwnedSocket {
+    fn as_raw_fd(&self) -> RawFd {
+        self.as_raw_socket() as _
+    }
+}
+
+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)
+    }
+}
+
+impl From<std::fs::File> for OwnedFd {
+    fn from(value: std::fs::File) -> Self {
+        Self::File(OwnedHandle::from(value))
+    }
+}
+
+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)
+    }
+}
+
+impl From<socket2::Socket> for OwnedFd {
+    fn from(value: socket2::Socket) -> Self {
+        Self::Socket(OwnedSocket::from(value))
+    }
+}
+
+/// Borrowed handle or socket on Windows.
+#[derive(Debug)]
+pub enum BorrowedFd<'a> {
+    /// Win32 handle.
+    File(BorrowedHandle<'a>),
+    /// Windows socket handle.
+    Socket(BorrowedSocket<'a>),
+}
+
+impl AsRawFd for BorrowedFd<'_> {
+    fn as_raw_fd(&self) -> RawFd {
+        match self {
+            Self::File(fd) => fd.as_raw_handle() as RawFd,
+            Self::Socket(s) => s.as_raw_socket() as RawFd,
+        }
+    }
+}
+
+impl<'a> From<BorrowedHandle<'a>> for BorrowedFd<'a> {
+    fn from(value: BorrowedHandle<'a>) -> Self {
+        Self::File(value)
+    }
+}
+
+impl<'a> From<BorrowedSocket<'a>> for BorrowedFd<'a> {
+    fn from(value: BorrowedSocket<'a>) -> Self {
+        Self::Socket(value)
+    }
+}
+
+/// Extracts fds.
+pub trait AsFd {
+    /// Extracts the borrowed fd.
+    fn as_fd(&self) -> BorrowedFd<'_>;
+}
+
+impl AsFd for OwnedFd {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        match self {
+            Self::File(fd) => fd.as_fd(),
+            Self::Socket(s) => s.as_fd(),
+        }
+    }
+}
+
+impl AsFd for std::fs::File {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.as_handle().into()
+    }
+}
+
+impl AsFd for OwnedHandle {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.as_handle().into()
+    }
+}
+
+impl AsFd for BorrowedHandle<'_> {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        (*self).into()
+    }
+}
+
+impl AsFd for socket2::Socket {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.as_socket().into()
+    }
+}
+
+impl AsFd for OwnedSocket {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.as_socket().into()
+    }
+}
+
+impl AsFd for BorrowedSocket<'_> {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        (*self).into()
+    }
+}
+
+impl AsFd for std::process::ChildStdin {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.as_handle().into()
+    }
+}
+
+impl AsFd for std::process::ChildStdout {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.as_handle().into()
+    }
+}
+
+impl AsFd for std::process::ChildStderr {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.as_handle().into()
+    }
+}