cancelable builtins

4 files changed, 38 insertions, 61 deletions

diff --git a/src/ctrlc.rs b/src/ctrlc.rs
index 4c0153a..9bf53fa 100644
--- a/src/ctrlc.rs
+++ b/src/ctrlc.rs
@@ -1,7 +1,7 @@
 use crate::Session;
 use libc::c_int;
 use nix::sys::signal::*;
-use std::sync::*;
+use std::{sync::*, time::Instant};
 
 static SESSION: Mutex<Option<Arc<Mutex<Session>>>> = Mutex::new(None);
 
@@ -9,7 +9,7 @@ fn handle() {
     let Ok(mut se) = SESSION.lock() else { return };
     let Some(se) = se.as_mut() else { return };
     let Ok(mut se) = se.lock() else { return };
-    se.ctrlc.pressed = true;
+    se.ctrlc.last_press = Instant::now();
 }
 
 extern "C" fn c_handle(_signal: c_int) {
@@ -21,9 +21,16 @@ extern "C" fn c_handle(_signal: c_int) {
     });
 }
 
-#[derive(Default)]
 pub struct CtrlC {
-    pressed: bool,
+    last_press: Instant,
+}
+
+impl Default for CtrlC {
+    fn default() -> Self {
+        Self {
+            last_press: Instant::now(),
+        }
+    }
 }
 
 struct Teardown;
@@ -52,12 +59,6 @@ pub fn setup(session: Arc<Mutex<Session>>) -> impl Drop {
     Teardown
 }
 
-pub fn peek(session: &Session) -> bool {
-    session.ctrlc.pressed
-}
-
-pub fn pop(session: &mut Session) -> bool {
-    let x = session.ctrlc.pressed;
-    session.ctrlc.pressed = false;
-    x
+pub fn pressed_since(session: &Session, instant: Instant) -> bool {
+    session.ctrlc.last_press > instant
 }
diff --git a/src/main.rs b/src/main.rs
index 222a5c5..9e9d640 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -8,7 +8,7 @@ use std::io::{self, IsTerminal, Read, Write};
 use std::os::unix::ffi::OsStrExt;
 use std::os::unix::io::AsRawFd;
 use std::path::Path;
-use std::process::{Command, Stdio};
+use std::process::Command;
 use std::sync::{Arc, Mutex};
 use std::thread::sleep;
 use std::time::Duration;
@@ -27,9 +27,9 @@ pub mod parse;
 pub mod raw;
 pub mod reload;
 pub mod run;
+pub mod rw;
 pub mod serialization;
 pub mod wait;
-pub mod rw;
 
 use linebuf::LineBuf;
 use raw::*;
diff --git a/src/run/mod.rs b/src/run/mod.rs
index 6046a12..09051c5 100644
--- a/src/run/mod.rs
+++ b/src/run/mod.rs
@@ -1,10 +1,8 @@
-use nix::spawn;
-
 use crate::rw::*;
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
-use std::thread::Thread;
+use std::time::Instant;
 
 use crate::parse::{self, Ast, PostExpansion, PreExpansion};
 use crate::wait::{ChildWaiter, ThreadWaiter};
@@ -55,7 +53,7 @@ impl ExecError {
                 let mut out = String::new();
                 for e in stack.iter() {
                     if !out.is_empty() {
-                        out += "\n";
+                        out += "\r\n";
                     }
                     out += &e.error_message();
                 }
@@ -173,9 +171,10 @@ impl SpawnedCmd {
 
 impl Executor {
     fn exec_loop(&mut self, mut s: SpawnedCmd, cs: &mut [Canceler]) -> Result<(), ExecError> {
+        let begin = Instant::now();
         while !s.join_timeout(50) {
-            if let Ok(mut se) = self.se.try_lock()
-                && ctrlc::pop(&mut se)
+            if let Ok(se) = self.se.try_lock()
+                && ctrlc::pressed_since(&se, begin)
             {
                 for c in cs.iter_mut() {
                     c.cancel();
@@ -217,8 +216,8 @@ impl Executor {
 
                 let handle = wait::spawn(move || {
                     let ast = ast.expand(&mut this)?;
-                    this.execute(ast, stdin, stdout)?;
-                    Ok(())
+                    let cmd = this.execute(ast, stdin, stdout);
+                    this.exec_loop(cmd, &mut [])
                 });
 
                 SpawnedCmd::Fun(handle)
@@ -313,9 +312,9 @@ impl Executor {
     pub fn execute_fun(
         session: Arc<Mutex<Session>>,
         invoke: Vec<BString>,
-        stdin: InputReader,
-        stdout: OutputWriter,
-    ) -> SpawnedCmd {
+        stdin: Input,
+        stdout: Output,
+    ) -> Result<(), ExecError> {
         let mut this = Self {
             se: session,
             args: None,
@@ -327,7 +326,11 @@ impl Executor {
             args: invoke[1..].to_vec(),
         };
 
-        this.execute_pipeline(parse::Pipes { cmds: vec![cmd] }, stdin, stdout)
+        let (stdin, c1) = InputReader::new(stdin);
+        let (stdout, c2) = OutputWriter::new(stdout);
+
+        let cmd = this.execute_pipeline(parse::Pipes { cmds: vec![cmd] }, stdin, stdout);
+        this.exec_loop(cmd, &mut [c1, c2])
     }
 }
 
@@ -376,25 +379,14 @@ impl parse::Expander for Executor {
         }
 
         let (stdin, _) = io::pipe().unwrap();
-        let (stdin, mut c1) = InputReader::new(Input::Pipe(stdin));
+        let (stdin, c1) = InputReader::new(Input::Pipe(stdin));
         let (mut expansion, stdout) = io::pipe().unwrap();
-        let (stdout, mut c2) = OutputWriter::new(Output::Pipe(stdout));
+        let (stdout, c2) = OutputWriter::new(Output::Pipe(stdout));
         let mut this = self.clone();
 
-        let se = self.se.clone();
         let t = std::thread::spawn(move || {
-            let mut spawned = this.execute(ast, stdin, stdout);
-            while !spawned.join_timeout(50) {
-                if let Ok(mut se) = se.try_lock()
-                    && ctrlc::pop(&mut se)
-                {
-                    c1.cancel();
-                    c2.cancel();
-                    spawned.cancel();
-                    break;
-                }
-            }
-            spawned.join()
+            let cmd = this.execute(ast, stdin, stdout);
+            this.exec_loop(cmd, &mut [c1, c2])
         });
         let mut buf = Vec::new();
         expansion.read_to_end(&mut buf).unwrap();
@@ -413,21 +405,10 @@ fn exec(se: Arc<Mutex<Session>>, ast: Ast<PreExpansion>) -> Result<(), ExecError
         expand_commands: true,
     };
     let ast = ast.expand(&mut exec)?;
-    let (stdin, mut c1) = InputReader::new(Input::Stdin);
-    let (stdout, mut c2) = OutputWriter::new(Output::Stdout);
-    let mut spawned = exec.execute(ast, stdin, stdout);
-    ctrlc::pop(&mut *se.lock().unwrap());
-    while !spawned.join_timeout(50) {
-        if let Ok(mut se) = se.try_lock()
-            && ctrlc::pop(&mut se)
-        {
-            c1.cancel();
-            c2.cancel();
-            spawned.cancel();
-            break;
-        }
-    }
-    spawned.join()
+    let (stdin, c1) = InputReader::new(Input::Stdin);
+    let (stdout, c2) = OutputWriter::new(Output::Stdout);
+    let cmd = exec.execute(ast, stdin, stdout);
+    exec.exec_loop(cmd, &mut [c1, c2])
 }
 
 pub fn run(se: Arc<Mutex<Session>>, cmd: Vec<u8>) {
diff --git a/src/rw.rs b/src/rw.rs
index 31808fb..b6b3a16 100644
--- a/src/rw.rs
+++ b/src/rw.rs
@@ -115,11 +115,6 @@ fn check<'a>(
 }
 
 impl InputReader {
-    fn cancel(&mut self) -> PollStatus {
-        self.canceled = true;
-        PollStatus::Cancel
-    }
-
     fn poll(&mut self) -> PollStatus {
         let stdin = io::stdin();
         let read_fd = match &self.input {