feat: initial implementation of std::process

Author: dignifiedquire

Cargo.toml

@@ -46,6 +46,11 @@ std = [
   "once_cell",
   "pin-utils",
   "slab",
+  "crossbeam-queue",
+  "async-signals",
+  "libc",
+  "lazy_static",
+  "async-macros",
 ]
 alloc = [
   "futures-core/alloc",
@@ -72,6 +77,11 @@ once_cell = { version = "1.2.0", optional = true }
 pin-project-lite = { version = "0.1.2", optional = true }
 pin-utils = { version = "0.1.0-alpha.4", optional = true }
 slab = { version = "0.4.2", optional = true }
+crossbeam-queue = { version = "0.2.1", optional = true }
+async-signals = { version = "0.1.0", optional = true }
+lazy_static = { version = "1.4.0", optional = true }
+async-macros = { version = "2.0.0", optional = true }
+libc = { version = "0.2.67", optional = true }
 
 [dev-dependencies]
 femme = "1.3.0"

src/lib.rs

@@ -270,6 +270,7 @@ cfg_default! {
     pub mod fs;
     pub mod path;
     pub mod net;
+    pub mod process;
 }
 
 cfg_unstable! {

src/net/driver/mod.rs

@@ -29,7 +29,7 @@ struct Readers {
     /// (cf. `Watcher::poll_read_ready`)
     ready: bool,
     /// The `Waker`s blocked on reading.
-    wakers: Vec<Waker>
+    wakers: Vec<Waker>,
 }
 
 /// The set of `Waker`s interested in write readiness.
@@ -39,7 +39,7 @@ struct Writers {
     /// (cf. `Watcher::poll_write_ready`)
     ready: bool,
     /// The `Waker`s blocked on writing.
-    wakers: Vec<Waker>
+    wakers: Vec<Waker>,
 }
 
 /// The state of a networking driver.
@@ -88,8 +88,14 @@ impl Reactor {
         // Allocate an entry and insert it into the slab.
         let entry = Arc::new(Entry {
             token,
-            readers: Mutex::new(Readers { ready: false, wakers: Vec::new() }),
-            writers: Mutex::new(Writers { ready: false, wakers: Vec::new() }),
+            readers: Mutex::new(Readers {
+                ready: false,
+                wakers: Vec::new(),
+            }),
+            writers: Mutex::new(Writers {
+                ready: false,
+                wakers: Vec::new(),
+            }),
         });
         vacant.insert(entry.clone());
 
@@ -249,6 +255,43 @@ impl<T: Evented> Watcher<T> {
         Poll::Pending
     }
 
+    /// Polls the inner I/O source for a non-blocking read operation.
+    ///
+    /// If the operation returns an error of the `io::ErrorKind::WouldBlock` kind, the current task
+    /// will be registered for wakeup when the I/O source becomes readable.
+    pub fn poll_read_with_mut<'a, F, R>(
+        &'a mut self,
+        cx: &mut Context<'_>,
+        mut f: F,
+    ) -> Poll<io::Result<R>>
+    where
+        F: FnMut(&mut T) -> io::Result<R>,
+    {
+        // If the operation isn't blocked, return its result.
+        match f(self.source.as_mut().unwrap()) {
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            res => return Poll::Ready(res),
+        }
+
+        // Lock the waker list.
+        let mut readers = self.entry.readers.lock().unwrap();
+
+        // Try running the operation again.
+        match f(self.source.as_mut().unwrap()) {
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            res => return Poll::Ready(res),
+        }
+
+        // Register the task if it isn't registered already.
+        if readers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {
+            readers.wakers.push(cx.waker().clone());
+        }
+
+        readers.ready = false;
+
+        Poll::Pending
+    }
+
     /// Polls the inner I/O source for a non-blocking write operation.
     ///
     /// If the operation returns an error of the `io::ErrorKind::WouldBlock` kind, the current task
@@ -286,6 +329,43 @@ impl<T: Evented> Watcher<T> {
         Poll::Pending
     }
 
+    /// Polls the inner I/O source for a non-blocking write operation.
+    ///
+    /// If the operation returns an error of the `io::ErrorKind::WouldBlock` kind, the current task
+    /// will be registered for wakeup when the I/O source becomes writable.
+    pub fn poll_write_with_mut<'a, F, R>(
+        &'a mut self,
+        cx: &mut Context<'_>,
+        mut f: F,
+    ) -> Poll<io::Result<R>>
+    where
+        F: FnMut(&mut T) -> io::Result<R>,
+    {
+        // If the operation isn't blocked, return its result.
+        match f(self.source.as_mut().unwrap()) {
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            res => return Poll::Ready(res),
+        }
+
+        // Lock the waker list.
+        let mut writers = self.entry.writers.lock().unwrap();
+
+        // Try running the operation again.
+        match f(self.source.as_mut().unwrap()) {
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            res => return Poll::Ready(res),
+        }
+
+        // Register the task if it isn't registered already.
+        if writers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {
+            writers.wakers.push(cx.waker().clone());
+        }
+
+        writers.ready = false;
+
+        Poll::Pending
+    }
+
     /// Polls the inner I/O source until a non-blocking read can be performed.
     ///
     /// If non-blocking reads are currently not possible, the `Waker`
@@ -296,7 +376,7 @@ impl<T: Evented> Watcher<T> {
         // Lock the waker list.
         let mut readers = self.entry.readers.lock().unwrap();
         if readers.ready {
-            return Poll::Ready(())
+            return Poll::Ready(());
         }
         // Register the task if it isn't registered already.
         if readers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {
@@ -314,7 +394,7 @@ impl<T: Evented> Watcher<T> {
         // Lock the waker list.
         let mut writers = self.entry.writers.lock().unwrap();
         if writers.ready {
-            return Poll::Ready(())
+            return Poll::Ready(());
         }
         // Register the task if it isn't registered already.
         if writers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {

src/process/kill.rs

@@ -0,0 +1,73 @@
+use crate::io;
+use crate::prelude::*;
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A drop guard which ensures the child process is killed on drop to maintain
+/// the contract of dropping a Future leads to "cancellation".
+#[derive(Debug)]
+pub(crate) struct ChildDropGuard<T: Kill> {
+    inner: T,
+    kill_on_drop: bool,
+}
+
+impl<T: Kill> ChildDropGuard<T> {
+    pub(crate) fn new(inner: T) -> Self {
+        Self {
+            inner,
+            kill_on_drop: true,
+        }
+    }
+
+    pub(crate) fn forget(&mut self) {
+        self.kill_on_drop = false;
+    }
+}
+
+impl<T: Kill> Kill for ChildDropGuard<T> {
+    fn kill(&mut self) -> io::Result<()> {
+        let ret = self.inner.kill();
+
+        if ret.is_ok() {
+            self.kill_on_drop = false;
+        }
+
+        ret
+    }
+}
+
+impl<T: Kill> Drop for ChildDropGuard<T> {
+    fn drop(&mut self) {
+        if self.kill_on_drop {
+            drop(self.kill());
+        }
+    }
+}
+
+impl<T: Future + Kill + Unpin> Future for ChildDropGuard<T> {
+    type Output = T::Output;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let ret = Pin::new(&mut self.inner).poll(cx);
+
+        if let Poll::Ready(_) = ret {
+            // Avoid the overhead of trying to kill a reaped process
+            self.kill_on_drop = false;
+        }
+
+        ret
+    }
+}
+
+/// An interface for killing a running process.
+pub(crate) trait Kill {
+    /// Forcefully kill the process.
+    fn kill(&mut self) -> io::Result<()>;
+}
+
+impl<'a, T: 'a + Kill> Kill for &'a mut T {
+    fn kill(&mut self) -> io::Result<()> {
+        (**self).kill()
+    }
+}