fix(backend/runner): Processes are stoppable, thanks God, thank you

Author: Brayan-724

Cargo.lock

@@ -9,6 +9,7 @@ authors = ["Apika Luca"]
 test = true
 
 [dependencies]
+async-io = "2.4.1"
 hakoniwa = "1.1.0"
 lsp-types = "0.97.0"
 nix = "0.29.0"

backend/runner/Cargo.toml

@@ -1,13 +1,19 @@
+use std::{io::Read, ops, os::fd::AsFd, ptr::read};
+
+use async_io::Async;
 use hakoniwa::{Child, ExitStatus};
-use nix::sys::wait::{self, WaitPidFlag, WaitStatus};
+use nix::{
+    sys::wait::{self, WaitPidFlag, WaitStatus},
+    unistd::Pid,
+};
 
-trait HakoniwaChildExt {
+pub trait HakoniwaChildExt {
     fn try_wait(&self) -> Option<ExitStatus>;
 }
 
 impl HakoniwaChildExt for Child {
     fn try_wait(&self) -> Option<ExitStatus> {
-        match wait::waitpid(self.id(), Some(WaitPidFlag::WNOHANG)) {
+        match wait::waitpid(Pid::from_raw(self.id() as i32), Some(WaitPidFlag::WNOHANG)) {
             Ok(WaitStatus::StillAlive) => None,
             Ok(WaitStatus::Exited(_, code)) => Some(ExitStatus {
                 code,
@@ -16,13 +22,65 @@ impl HakoniwaChildExt for Child {
                 exit_code: None,
                 rusage: None,
             }),
-            Ok(WaitStatus::Signaled(_, signal, _) | WaitStatus::Stopped(_, signal)) => Some(ExitStatus {
-                code: signal as i32,
-                reason: signal.as_str().to_owned(),
-                exit_code: None,
-                rusage: None,
-            }),
+            Ok(WaitStatus::Signaled(_, signal, _) | WaitStatus::Stopped(_, signal)) => {
+                Some(ExitStatus {
+                    code: signal as i32,
+                    reason: signal.as_str().to_owned(),
+                    exit_code: None,
+                    rusage: None,
+                })
+            }
             Ok(WaitStatus::Continued(_)) => None,
+            Ok(_) => None,
+            Err(err) => {
+                println!("[ERROR] {err}");
+                None
+            }
         }
     }
 }
+
+pub struct AsyncOsReader(Async<os_pipe::PipeReader>);
+
+impl From<os_pipe::PipeReader> for AsyncOsReader {
+    fn from(value: os_pipe::PipeReader) -> Self {
+        Self(Async::new(value).expect("Cannot create async wrapper"))
+    }
+}
+
+impl AsFd for AsyncOsReader {
+    fn as_fd(&self) -> std::os::unix::prelude::BorrowedFd<'_> {
+        self.0.as_fd()
+    }
+}
+
+impl AsyncOsReader {
+    pub async fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        _ = self.0.readable().await?;
+        unsafe { self.0.get_mut() }.read(buf)
+    }
+
+    pub async fn read_to_end(&mut self, buf: &mut Vec<u8>) -> std::io::Result<usize> {
+        let mut total_bytes = 0;
+
+        loop {
+            let read_bytes = self.read(buf.as_mut_slice()).await?;
+
+            if read_bytes == 0 {
+                break;
+            }
+
+            total_bytes += read_bytes;
+        }
+
+        Ok(total_bytes)
+    }
+}
+
+impl ops::Deref for AsyncOsReader {
+    type Target = os_pipe::PipeReader;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0.get_ref()
+    }
+}

backend/runner/src/hakoniwa_ext.rs

@@ -1,15 +1,19 @@
 pub mod error;
 pub mod hakoniwa_ext;
 
+use async_io::Async;
 use hakoniwa::{Child, Command, Container, ExitStatus, Output};
+use hakoniwa_ext::{AsyncOsReader, HakoniwaChildExt};
 use nix::libc::pid_t;
 use nix::sys::signal::{self, Signal};
 use nix::unistd::Pid;
 pub use os_pipe::{PipeReader, PipeWriter};
 use std::future::Future;
 use std::io::Read;
 use std::path::{Path, PathBuf};
+use std::time::Duration;
 use tokio::sync::oneshot;
+use tokio::time::Interval;
 use tokio::{fs, io};
 
 pub const BASE_ENV: [(&str, &str); 3] = [
@@ -92,12 +96,12 @@ impl Runner {
     }
 
     pub async fn collect_output(cmd: &mut Command) -> Result<Output, hakoniwa::Error> {
-        async fn collect(stream: Option<PipeReader>) -> Vec<u8> {
+        async fn collect(stream: Option<AsyncOsReader>) -> Vec<u8> {
             let mut buf = Vec::new();
 
             let Some(mut stream) = stream else { return buf };
 
-            _ = stream.read_to_end(&mut buf);
+            _ = stream.read_to_end(&mut buf).await;
 
             buf
         }
@@ -120,10 +124,10 @@ impl Runner {
     where
         Stdout: Default + Send + 'static,
         StdoutAsync: Future<Output = Stdout> + Send + 'static,
-        StdoutFn: FnOnce(Option<PipeReader>) -> StdoutAsync,
+        StdoutFn: FnOnce(Option<AsyncOsReader>) -> StdoutAsync,
         Stderr: Default + Send + 'static,
         StderrAsync: Future<Output = Stderr> + Send + 'static,
-        StderrFn: FnOnce(Option<PipeReader>) -> StderrAsync,
+        StderrFn: FnOnce(Option<AsyncOsReader>) -> StderrAsync,
     {
         let mut child = cmd
             .envs(BASE_ENV)
@@ -133,18 +137,24 @@ impl Runner {
             .stderr(hakoniwa::Stdio::MakePipe)
             .spawn()?;
 
-        let stdout = child.stdout.take();
+        let stdout = child.stdout.take().map(AsyncOsReader::from);
         let stdout = tokio::spawn(stdout_fn(stdout));
-        let stderr = child.stderr.take();
+        let stderr = child.stderr.take().map(AsyncOsReader::from);
         let stderr = tokio::spawn(stderr_fn(stderr));
 
         let child_pid = Pid::from_raw(child.id() as pid_t);
-        let status = if let Some(abort) = abort {
+        let status = if let Some(mut abort) = abort {
             tokio::spawn(async move {
+                let mut status_check_interval = tokio::time::interval(Duration::from_millis(100));
+
                 loop {
                     tokio::select! {
-                        _ = abort => {
-                            println!("[[[[KILL]]]]");
+                        _ = status_check_interval.tick() => {
+                            if let Some(status) = child.try_wait() {
+                                return Ok(status)
+                            }
+                        }
+                        _ = &mut abort => {
                             _ = signal::kill(child_pid, Signal::SIGKILL);
                             return Ok(ExitStatus {
                                 code: 137,
@@ -153,14 +163,6 @@ impl Runner {
                                 rusage: None,
                             });
                         },
-                        else => {
-                            return Ok(ExitStatus {
-                                code: 137,
-                                reason: "Aborted".to_owned(),
-                                exit_code: None,
-                                rusage: None,
-                            });
-                        }
                     }
                 }
             })

backend/runner/src/lib.rs

@@ -83,7 +83,7 @@ async fn execute(project: ProjectExecuter) -> Result<(), ()> {
                 let buf = &mut [0; 2048];
 
                 loop {
-                    let Ok(size) = stdout.read(buf) else {
+                    let Ok(size) = stdout.read(buf).await else {
                         log::trace!("Cannot read");
                         break;
                     };