more refactoring to lay the groundwork for reintroducing builtins

5 files changed, 302 insertions, 189 deletions

diff --git a/src/state/history/builtins.rs b/src/builtins.rs
index 329aa97..e55f6ec 100644
--- a/src/state/history/builtins.rs
+++ b/src/builtins.rs
@@ -1,9 +1,9 @@
 use async_std::io::WriteExt as _;
 use std::os::unix::process::ExitStatusExt as _;
 
-type Builtin = &'static (dyn for<'a> Fn(
+type BuiltinFunc = &'static (dyn for<'a> Fn(
     &'a crate::parse::Exe,
-    &'a super::ProcessEnv,
+    &'a crate::command::Env,
 ) -> std::pin::Pin<
     Box<
         dyn std::future::Future<Output = std::process::ExitStatus>
@@ -15,16 +15,16 @@ type Builtin = &'static (dyn for<'a> Fn(
               + Send);
 
 static BUILTINS: once_cell::sync::Lazy<
-    std::collections::HashMap<&'static str, Builtin>,
+    std::collections::HashMap<&'static str, BuiltinFunc>,
 > = once_cell::sync::Lazy::new(|| {
     // all this does is convince the type system to do the right thing, i
     // don't think there's any way to just do it directly through annotations
     // or casts or whatever
-    fn coerce_builtin<F>(f: &'static F) -> Builtin
+    fn coerce_builtin<F>(f: &'static F) -> BuiltinFunc
     where
         F: for<'a> Fn(
                 &'a crate::parse::Exe,
-                &'a super::ProcessEnv,
+                &'a crate::command::Env,
             ) -> std::pin::Pin<
                 Box<
                     dyn std::future::Future<Output = std::process::ExitStatus>
@@ -55,9 +55,43 @@ static BUILTINS: once_cell::sync::Lazy<
     builtins
 });
 
+pub struct Builtin {
+    f: BuiltinFunc,
+    stdin: Box<dyn async_std::io::Read>,
+    stdout: Box<dyn async_std::io::Write>,
+    stderr: Box<dyn async_std::io::Write>,
+}
+
+impl Builtin {
+    pub fn new(exe: &crate::parse::Exe) -> Option<Self> {
+        if let Some(f) = BUILTINS.get(exe.exe()) {
+            Some(Self {
+                f,
+                stdin: Box::new(async_std::io::stdin()),
+                stdout: Box::new(async_std::io::stdout()),
+                stderr: Box::new(async_std::io::stderr()),
+            })
+        } else {
+            None
+        }
+    }
+
+    pub fn stdin(&mut self, fh: std::fs::File) {
+        self.stdin = Box::new(async_std::fs::File::from(fh));
+    }
+
+    pub fn stdout(&mut self, fh: std::fs::File) {
+        self.stdout = Box::new(async_std::fs::File::from(fh));
+    }
+
+    pub fn stderr(&mut self, fh: std::fs::File) {
+        self.stderr = Box::new(async_std::fs::File::from(fh));
+    }
+}
+
 pub fn run(
     exe: &crate::parse::Exe,
-    env: &super::ProcessEnv,
+    env: &crate::command::Env,
 ) -> Option<
     std::pin::Pin<
         Box<
@@ -80,7 +114,7 @@ pub fn run(
 
 async fn cd(
     exe: &crate::parse::Exe,
-    env: &super::ProcessEnv,
+    env: &crate::command::Env,
 ) -> async_std::process::ExitStatus {
     let dir = exe
         .args()
@@ -135,7 +169,7 @@ async fn cd(
 
 async fn and(
     exe: &crate::parse::Exe,
-    env: &super::ProcessEnv,
+    env: &crate::command::Env,
 ) -> async_std::process::ExitStatus {
     let exe = exe.shift();
     if env.latest_status().success() {
@@ -148,7 +182,7 @@ async fn and(
 
 async fn or(
     exe: &crate::parse::Exe,
-    env: &super::ProcessEnv,
+    env: &crate::command::Env,
 ) -> async_std::process::ExitStatus {
     let exe = exe.shift();
     if env.latest_status().success() {
@@ -161,7 +195,7 @@ async fn or(
 
 async fn command(
     exe: &crate::parse::Exe,
-    env: &super::ProcessEnv,
+    env: &crate::command::Env,
 ) -> async_std::process::ExitStatus {
     let exe = exe.shift();
     // super::run_binary(&exe, env).await;
@@ -170,7 +204,7 @@ async fn command(
 
 async fn builtin(
     exe: &crate::parse::Exe,
-    env: &super::ProcessEnv,
+    env: &crate::command::Env,
 ) -> async_std::process::ExitStatus {
     let exe = exe.shift();
     run(&exe, env).unwrap().await;
diff --git a/src/command.rs b/src/command.rs
new file mode 100644
index 0000000..b1453af
--- /dev/null
+++ b/src/command.rs
@@ -0,0 +1,249 @@
+use async_std::io::ReadExt as _;
+use async_std::os::unix::process::CommandExt as _;
+use async_std::stream::StreamExt as _;
+use std::os::unix::io::FromRawFd as _;
+use std::os::unix::process::ExitStatusExt as _;
+
+const PID0: nix::unistd::Pid = nix::unistd::Pid::from_raw(0);
+
+#[derive(Clone)]
+pub struct Env {
+    latest_status: async_std::process::ExitStatus,
+}
+
+impl Env {
+    pub fn new() -> Self {
+        Self {
+            latest_status: async_std::process::ExitStatus::from_raw(0),
+        }
+    }
+
+    pub fn set_status(&mut self, status: async_std::process::ExitStatus) {
+        self.latest_status = status;
+    }
+
+    pub fn latest_status(&self) -> &async_std::process::ExitStatus {
+        &self.latest_status
+    }
+}
+
+enum Command {
+    Binary(async_std::process::Command),
+    Builtin(crate::builtins::Builtin),
+}
+
+impl Command {
+    fn new(exe: &crate::parse::Exe) -> Self {
+        crate::builtins::Builtin::new(exe).map_or_else(
+            || {
+                let mut cmd = async_std::process::Command::new(exe.exe());
+                cmd.args(exe.args());
+                Self::Binary(cmd)
+            },
+            Self::Builtin,
+        )
+    }
+
+    fn stdin(&mut self, fh: std::fs::File) {
+        match self {
+            Self::Binary(cmd) => {
+                cmd.stdin(fh);
+            }
+            Self::Builtin(cmd) => {
+                cmd.stdin(fh);
+            }
+        }
+    }
+
+    fn stdout(&mut self, fh: std::fs::File) {
+        match self {
+            Self::Binary(cmd) => {
+                cmd.stdout(fh);
+            }
+            Self::Builtin(cmd) => {
+                cmd.stdout(fh);
+            }
+        }
+    }
+
+    fn stderr(&mut self, fh: std::fs::File) {
+        match self {
+            Self::Binary(cmd) => {
+                cmd.stderr(fh);
+            }
+            Self::Builtin(cmd) => {
+                cmd.stderr(fh);
+            }
+        }
+    }
+
+    unsafe fn pre_exec<F>(&mut self, f: F)
+    where
+        F: 'static + FnMut() -> std::io::Result<()> + Send + Sync,
+    {
+        match self {
+            Self::Binary(cmd) => {
+                cmd.pre_exec(f);
+            }
+            Self::Builtin(_) => {}
+        }
+    }
+
+    fn spawn(self, env: &Env) -> anyhow::Result<Child> {
+        match self {
+            Self::Binary(mut cmd) => {
+                let child = cmd.spawn()?;
+                Ok(Child::Binary(child))
+            }
+            Self::Builtin(cmd) => Ok(Child::Builtin(cmd)),
+        }
+    }
+}
+
+enum Child {
+    Binary(async_std::process::Child),
+    Builtin(crate::builtins::Builtin),
+}
+
+impl Child {
+    fn id(&self) -> Option<u32> {
+        match self {
+            Self::Binary(child) => Some(child.id()),
+            Self::Builtin(child) => todo!(),
+        }
+    }
+
+    async fn status(self) -> anyhow::Result<std::process::ExitStatus> {
+        match self {
+            Self::Binary(child) => Ok(child.status_no_drop().await?),
+            Self::Builtin(child) => todo!(),
+        }
+    }
+}
+
+pub async fn run() -> anyhow::Result<i32> {
+    let pipeline = read_pipeline().await?;
+    let env: Env = Env::new(); // todo
+    let mut cmds: Vec<_> = pipeline.exes().iter().map(Command::new).collect();
+    for i in 0..(cmds.len() - 1) {
+        let (r, w) = pipe()?;
+        cmds[i].stdout(w);
+        cmds[i + 1].stdin(r);
+    }
+
+    let mut children = vec![];
+    let mut pg_pid = None;
+    for mut cmd in cmds.drain(..) {
+        // Safety: setpgid is an async-signal-safe function
+        unsafe {
+            cmd.pre_exec(move || {
+                setpgid_child(pg_pid)?;
+                Ok(())
+            });
+        }
+        let child = cmd.spawn(&env)?;
+        if let Some(id) = child.id() {
+            let child_pid = id_to_pid(id);
+            setpgid_parent(child_pid, pg_pid)?;
+            if pg_pid.is_none() {
+                pg_pid = Some(child_pid);
+                set_foreground_pg(child_pid)?;
+            }
+        }
+        children.push(child);
+    }
+
+    let mut final_status = None;
+
+    let count = children.len();
+    let mut children: futures_util::stream::FuturesUnordered<_> =
+        children
+            .into_iter()
+            .enumerate()
+            .map(|(i, child)| async move {
+                (child.status().await, i == count - 1)
+            })
+            .collect();
+    while let Some((status, last)) = children.next().await {
+        let status = status.unwrap_or_else(|_| {
+            async_std::process::ExitStatus::from_raw(1 << 8)
+        });
+        // this conversion is safe because the Signal enum is repr(i32)
+        #[allow(clippy::as_conversions)]
+        if status.signal() == Some(nix::sys::signal::Signal::SIGINT as i32) {
+            nix::sys::signal::raise(nix::sys::signal::Signal::SIGINT)?;
+        }
+        if last {
+            final_status = Some(status);
+        }
+    }
+
+    let final_status = final_status.unwrap();
+    if let Some(signal) = final_status.signal() {
+        nix::sys::signal::raise(signal.try_into().unwrap())?;
+    }
+    Ok(final_status.code().unwrap())
+}
+
+async fn read_pipeline() -> anyhow::Result<crate::parse::Pipeline> {
+    // Safety: this code is only called by crate::history::run_pipeline, which
+    // passes data through on fd 3, and which will not spawn this process
+    // unless the pipe was successfully opened on that fd
+    let mut fd3 = unsafe { async_std::fs::File::from_raw_fd(3) };
+    let mut pipeline = String::new();
+    fd3.read_to_string(&mut pipeline).await?;
+    let ast = crate::parse::Pipeline::parse(&pipeline)?;
+    Ok(ast)
+}
+
+fn pipe() -> anyhow::Result<(std::fs::File, std::fs::File)> {
+    let (r, w) = nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC)?;
+    // Safety: these file descriptors were just returned by pipe2 above, which
+    // means they must be valid otherwise that call would have returned an
+    // error
+    Ok((unsafe { std::fs::File::from_raw_fd(r) }, unsafe {
+        std::fs::File::from_raw_fd(w)
+    }))
+}
+
+fn set_foreground_pg(pg: nix::unistd::Pid) -> anyhow::Result<()> {
+    let pty = nix::fcntl::open(
+        "/dev/tty",
+        nix::fcntl::OFlag::empty(),
+        nix::sys::stat::Mode::empty(),
+    )?;
+    nix::unistd::tcsetpgrp(pty, pg)?;
+    nix::unistd::close(pty)?;
+    nix::sys::signal::kill(neg_pid(pg), nix::sys::signal::Signal::SIGCONT)?;
+    Ok(())
+}
+
+fn setpgid_child(pg: Option<nix::unistd::Pid>) -> std::io::Result<()> {
+    nix::unistd::setpgid(PID0, pg.unwrap_or(PID0))?;
+    Ok(())
+}
+
+fn setpgid_parent(
+    pid: nix::unistd::Pid,
+    pg: Option<nix::unistd::Pid>,
+) -> anyhow::Result<()> {
+    nix::unistd::setpgid(pid, pg.unwrap_or(PID0)).or_else(|e| {
+        // EACCES means that the child already called exec, but if it did,
+        // then it also must have already called setpgid itself, so we don't
+        // care
+        if e == nix::errno::Errno::EACCES {
+            Ok(())
+        } else {
+            Err(e)
+        }
+    })?;
+    Ok(())
+}
+
+fn id_to_pid(id: u32) -> nix::unistd::Pid {
+    nix::unistd::Pid::from_raw(id.try_into().unwrap())
+}
+
+fn neg_pid(pid: nix::unistd::Pid) -> nix::unistd::Pid {
+    nix::unistd::Pid::from_raw(-pid.as_raw())
+}
diff --git a/src/main.rs b/src/main.rs
index f3b444c..91a3526 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -12,11 +12,12 @@
 #![allow(clippy::too_many_lines)]
 #![allow(clippy::type_complexity)]
 
+mod builtins;
+mod command;
 mod env;
 mod event;
 mod format;
 mod parse;
-mod pipe;
 mod state;
 
 use async_std::stream::StreamExt as _;
@@ -43,8 +44,8 @@ fn get_offset() -> time::UtcOffset {
 }
 
 async fn async_main() -> anyhow::Result<i32> {
-    if std::env::args().nth(1).as_deref() == Some("--internal-pipe-runner") {
-        return pipe::run().await;
+    if std::env::args().nth(1).as_deref() == Some("--internal-cmd-runner") {
+        return command::run().await;
     }
 
     let mut input = textmode::Input::new().await?;
diff --git a/src/pipe.rs b/src/pipe.rs
deleted file mode 100644
index e75f010..0000000
--- a/src/pipe.rs
+++ /dev/null
@@ -1,149 +0,0 @@
-use async_std::io::ReadExt as _;
-use async_std::os::unix::process::CommandExt as _;
-use async_std::stream::StreamExt as _;
-use std::os::unix::io::FromRawFd as _;
-use std::os::unix::process::ExitStatusExt as _;
-
-const PID0: nix::unistd::Pid = nix::unistd::Pid::from_raw(0);
-
-pub async fn run() -> anyhow::Result<i32> {
-    let pipeline = read_pipeline().await?;
-    let mut cmds: Vec<_> = pipeline
-        .exes()
-        .iter()
-        .map(|exe| {
-            let mut cmd = async_std::process::Command::new(exe.exe());
-            cmd.args(exe.args());
-            cmd
-        })
-        .collect();
-    for i in 0..(cmds.len() - 1) {
-        let (r, w) = pipe()?;
-        cmds[i].stdout(w);
-        cmds[i + 1].stdin(r);
-    }
-
-    let mut children = vec![];
-
-    // Safety: setpgid is an async-signal-safe function
-    unsafe {
-        cmds[0].pre_exec(|| {
-            setpgid_child(PID0)?;
-            Ok(())
-        });
-    }
-    let leader = cmds[0].spawn()?;
-    let pg_pid = id_to_pid(leader.id());
-    setpgid_parent(pg_pid, PID0)?;
-    set_foreground_pg(pg_pid)?;
-    children.push(leader);
-
-    for cmd in &mut cmds[1..] {
-        // Safety: setpgid is an async-signal-safe function
-        unsafe {
-            cmd.pre_exec(move || {
-                setpgid_child(pg_pid)?;
-                Ok(())
-            });
-        }
-        let child = cmd.spawn()?;
-        let child_pid = id_to_pid(child.id());
-        children.push(child);
-        setpgid_parent(child_pid, pg_pid)?;
-    }
-    // ensure that we don't keep the pipes open past when the children exit
-    drop(cmds);
-
-    let mut final_status = None;
-
-    let count = children.len();
-    let mut children: futures_util::stream::FuturesUnordered<_> = children
-        .into_iter()
-        .enumerate()
-        .map(|(i, child)| async move {
-            (child.status_no_drop().await, i == count - 1)
-        })
-        .collect();
-    while let Some((status, last)) = children.next().await {
-        let status = status.unwrap_or_else(|_| {
-            async_std::process::ExitStatus::from_raw(1 << 8)
-        });
-        // this conversion is safe because the Signal enum is repr(i32)
-        #[allow(clippy::as_conversions)]
-        if status.signal() == Some(nix::sys::signal::Signal::SIGINT as i32) {
-            nix::sys::signal::raise(nix::sys::signal::Signal::SIGINT)?;
-        }
-        if last {
-            final_status = Some(status);
-        }
-    }
-
-    let final_status = final_status.unwrap();
-    if let Some(signal) = final_status.signal() {
-        nix::sys::signal::raise(signal.try_into().unwrap())?;
-    }
-    Ok(final_status.code().unwrap())
-}
-
-async fn read_pipeline() -> anyhow::Result<crate::parse::Pipeline> {
-    // Safety: this code is only called by crate::history::run_pipeline, which
-    // passes data through on fd 3, and which will not spawn this process
-    // unless the pipe was successfully opened on that fd
-    let mut fd3 = unsafe { async_std::fs::File::from_raw_fd(3) };
-    let mut pipeline = String::new();
-    fd3.read_to_string(&mut pipeline).await?;
-    let ast = crate::parse::Pipeline::parse(&pipeline)?;
-    Ok(ast)
-}
-
-fn pipe() -> anyhow::Result<(std::fs::File, std::fs::File)> {
-    let (r, w) = nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC)?;
-    // Safety: these file descriptors were just returned by pipe2 above, which
-    // means they must be valid otherwise that call would have returned an
-    // error
-    Ok((unsafe { std::fs::File::from_raw_fd(r) }, unsafe {
-        std::fs::File::from_raw_fd(w)
-    }))
-}
-
-fn set_foreground_pg(pg: nix::unistd::Pid) -> anyhow::Result<()> {
-    let pty = nix::fcntl::open(
-        "/dev/tty",
-        nix::fcntl::OFlag::empty(),
-        nix::sys::stat::Mode::empty(),
-    )?;
-    nix::unistd::tcsetpgrp(pty, pg)?;
-    nix::unistd::close(pty)?;
-    nix::sys::signal::kill(neg_pid(pg), nix::sys::signal::Signal::SIGCONT)?;
-    Ok(())
-}
-
-fn setpgid_child(pg: nix::unistd::Pid) -> std::io::Result<()> {
-    nix::unistd::setpgid(id_to_pid(0), pg)?;
-    Ok(())
-}
-
-fn setpgid_parent(
-    pid: nix::unistd::Pid,
-    pg: nix::unistd::Pid,
-) -> anyhow::Result<()> {
-    nix::unistd::setpgid(pid, pg).or_else(|e| {
-        // EACCES means that the child already called exec, but if it did,
-        // then it also must have already called setpgid itself, so we don't
-        // care
-        if e == nix::errno::Errno::EACCES {
-            Ok(())
-        } else {
-            Err(e)
-        }
-    })?;
-    Ok(())
-}
-
-fn id_to_pid(id: u32) -> nix::unistd::Pid {
-    nix::unistd::Pid::from_raw(id.try_into().unwrap())
-}
-
-fn neg_pid(pid: nix::unistd::Pid) -> nix::unistd::Pid {
-    nix::unistd::Pid::from_raw(-pid.as_raw())
-}
diff --git a/src/state/history/mod.rs b/src/state/history/mod.rs
index 8680559..3ff342c 100644
--- a/src/state/history/mod.rs
+++ b/src/state/history/mod.rs
@@ -2,7 +2,6 @@ use async_std::io::WriteExt as _;
 use std::os::unix::io::FromRawFd as _;
 use std::os::unix::process::ExitStatusExt as _;
 
-mod builtins;
 mod pty;
 
 pub struct History {
@@ -100,7 +99,7 @@ impl History {
         run_commands(
             ast.clone(),
             async_std::sync::Arc::clone(&entry),
-            ProcessEnv::new(),
+            crate::command::Env::new(),
             input_r,
             resize_r,
             event_w,
@@ -514,31 +513,10 @@ impl ExitInfo {
     }
 }
 
-#[derive(Clone)]
-pub struct ProcessEnv {
-    latest_status: async_std::process::ExitStatus,
-}
-
-impl ProcessEnv {
-    fn new() -> Self {
-        Self {
-            latest_status: async_std::process::ExitStatus::from_raw(0),
-        }
-    }
-
-    fn set_status(&mut self, status: async_std::process::ExitStatus) {
-        self.latest_status = status;
-    }
-
-    fn latest_status(&self) -> &async_std::process::ExitStatus {
-        &self.latest_status
-    }
-}
-
 fn run_commands(
     ast: crate::parse::Commands,
     entry: async_std::sync::Arc<async_std::sync::Mutex<Entry>>,
-    mut env: ProcessEnv,
+    mut env: crate::command::Env,
     input_r: async_std::channel::Receiver<Vec<u8>>,
     resize_r: async_std::channel::Receiver<(u16, u16)>,
     event_w: async_std::channel::Sender<crate::event::Event>,
@@ -582,10 +560,10 @@ fn run_commands(
 async fn run_pipeline(
     pipeline: &crate::parse::Pipeline,
     pty: &pty::Pty,
-    env: &ProcessEnv,
+    env: &crate::command::Env,
 ) -> (async_std::process::ExitStatus, bool) {
     let mut cmd = pty_process::Command::new(std::env::current_exe().unwrap());
-    cmd.arg("--internal-pipe-runner");
+    cmd.arg("--internal-cmd-runner");
     let (r, w) = nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC).unwrap();
     unsafe {
         cmd.pre_exec(move || {