move to tokio

13 files changed, 338 insertions, 360 deletions

diff --git a/src/main.rs b/src/main.rs
index a7d3f4b..e9c420d 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -36,17 +36,18 @@ struct Opt {
     status_fd: Option<std::os::unix::io::RawFd>,
 }
 
+#[tokio::main]
 async fn async_main(opt: Opt) -> anyhow::Result<i32> {
     if let Some(command) = opt.command {
-        let shell_write = opt.status_fd.and_then(|fd| {
+        let mut shell_write = opt.status_fd.and_then(|fd| {
             nix::sys::stat::fstat(fd).ok().map(|_| {
                 // Safety: we don't create File instances for or read/write
                 // data on this fd anywhere else
-                unsafe { async_std::fs::File::from_raw_fd(fd) }
+                unsafe { tokio::fs::File::from_raw_fd(fd) }
             })
         });
 
-        return runner::run(&command, shell_write.as_ref()).await;
+        return runner::run(&command, &mut shell_write).await;
     }
 
     shell::main().await
@@ -54,7 +55,7 @@ async fn async_main(opt: Opt) -> anyhow::Result<i32> {
 
 #[paw::main]
 fn main(opt: Opt) {
-    match async_std::task::block_on(async_main(opt)) {
+    match async_main(opt) {
         Ok(code) => {
             std::process::exit(code);
         }
diff --git a/src/mutex.rs b/src/mutex.rs
index df28ffc..e49df9c 100644
--- a/src/mutex.rs
+++ b/src/mutex.rs
@@ -1,10 +1,10 @@
-pub type Mutex<T> = async_std::sync::Arc<async_std::sync::Mutex<T>>;
-pub type Guard<T> = async_std::sync::MutexGuardArc<T>;
+pub type Mutex<T> = std::sync::Arc<tokio::sync::Mutex<T>>;
+pub type Guard<T> = tokio::sync::OwnedMutexGuard<T>;
 
-pub fn new<T>(t: T) -> async_std::sync::Arc<async_std::sync::Mutex<T>> {
-    async_std::sync::Arc::new(async_std::sync::Mutex::new(t))
+pub fn new<T>(t: T) -> Mutex<T> {
+    std::sync::Arc::new(tokio::sync::Mutex::new(t))
 }
 
 pub fn clone<T>(m: &Mutex<T>) -> Mutex<T> {
-    async_std::sync::Arc::clone(m)
+    std::sync::Arc::clone(m)
 }
diff --git a/src/parse/ast.rs b/src/parse/ast.rs
index e2d5840..46aa63a 100644
--- a/src/parse/ast.rs
+++ b/src/parse/ast.rs
@@ -97,7 +97,7 @@ impl Pipeline {
                 .into_iter()
                 .map(|exe| exe.eval(env))
                 .collect::<futures_util::stream::FuturesOrdered<_>>()
-                .collect::<Result<_, _>>()
+                .try_collect()
                 .await?,
         })
     }
@@ -137,7 +137,7 @@ impl Exe {
                     arg.eval(env).await.map(IntoIterator::into_iter)
                 })
                 .collect::<futures_util::stream::FuturesOrdered<_>>()
-                .collect::<Result<Vec<_>, _>>()
+                .try_collect::<Vec<_>>()
                 .await?
                 .into_iter()
                 .flatten()
@@ -147,7 +147,7 @@ impl Exe {
                 .into_iter()
                 .map(|arg| arg.eval(env))
                 .collect::<futures_util::stream::FuturesOrdered<_>>()
-                .collect::<Result<_, _>>()
+                .try_collect()
                 .await?,
         })
     }
@@ -330,12 +330,12 @@ impl WordPart {
         match self {
             Self::Alternation(_) => unreachable!(),
             Self::Substitution(commands) => {
-                let mut cmd = async_std::process::Command::new(
+                let mut cmd = tokio::process::Command::new(
                     std::env::current_exe().unwrap(),
                 );
                 cmd.args(&["-c", &commands]);
-                cmd.stdin(async_std::process::Stdio::inherit());
-                cmd.stderr(async_std::process::Stdio::inherit());
+                cmd.stdin(std::process::Stdio::inherit());
+                cmd.stderr(std::process::Stdio::inherit());
                 let mut out =
                     String::from_utf8(cmd.output().await.unwrap().stdout)
                         .unwrap();
@@ -408,15 +408,20 @@ impl Redirect {
         let mut iter = pair.into_inner();
 
         let prefix = iter.next().unwrap().as_str();
-        let (from, dir) = if let Some(from) = prefix.strip_suffix(">>") {
-            (from, super::Direction::Append)
-        } else if let Some(from) = prefix.strip_suffix('>') {
-            (from, super::Direction::Out)
-        } else if let Some(from) = prefix.strip_suffix('<') {
-            (from, super::Direction::In)
-        } else {
-            unreachable!()
-        };
+        let (from, dir) = prefix.strip_suffix(">>").map_or_else(
+            || {
+                prefix.strip_suffix('>').map_or_else(
+                    || {
+                        (
+                            prefix.strip_suffix('<').unwrap(),
+                            super::Direction::In,
+                        )
+                    },
+                    |from| (from, super::Direction::Out),
+                )
+            },
+            |from| (from, super::Direction::Append),
+        );
         let from = if from.is_empty() {
             match dir {
                 super::Direction::In => 0,
diff --git a/src/prelude.rs b/src/prelude.rs
index c647591..9c14a4b 100644
--- a/src/prelude.rs
+++ b/src/prelude.rs
@@ -1,10 +1,10 @@
 pub use crate::env::Env;
 
-pub use async_std::io::{ReadExt as _, WriteExt as _};
-pub use async_std::stream::StreamExt as _;
-pub use futures_lite::future::FutureExt as _;
+pub use futures_util::future::FutureExt as _;
+pub use futures_util::stream::StreamExt as _;
+pub use futures_util::stream::TryStreamExt as _;
+pub use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
 
-pub use async_std::os::unix::process::CommandExt as _;
 pub use std::os::unix::ffi::{OsStrExt as _, OsStringExt as _};
 pub use std::os::unix::io::{AsRawFd as _, FromRawFd as _, IntoRawFd as _};
 pub use std::os::unix::process::ExitStatusExt as _;
diff --git a/src/runner/builtins/command.rs b/src/runner/builtins/command.rs
index c0d3a84..e0e1853 100644
--- a/src/runner/builtins/command.rs
+++ b/src/runner/builtins/command.rs
@@ -97,7 +97,7 @@ impl Cfg {
 pub struct Io {
     fds: std::collections::HashMap<
         std::os::unix::io::RawFd,
-        std::sync::Arc<File>,
+        std::sync::Arc<tokio::sync::Mutex<File>>,
     >,
 }
 
@@ -108,7 +108,7 @@ impl Io {
         }
     }
 
-    fn stdin(&self) -> Option<std::sync::Arc<File>> {
+    fn stdin(&self) -> Option<std::sync::Arc<tokio::sync::Mutex<File>>> {
         self.fds.get(&0).map(std::sync::Arc::clone)
     }
 
@@ -120,11 +120,13 @@ impl Io {
             0,
             // Safety: we just acquired stdin via into_raw_fd, which acquires
             // ownership of the fd, so we are now the sole owner
-            std::sync::Arc::new(unsafe { File::input(stdin.into_raw_fd()) }),
+            std::sync::Arc::new(tokio::sync::Mutex::new(unsafe {
+                File::input(stdin.into_raw_fd())
+            })),
         );
     }
 
-    fn stdout(&self) -> Option<std::sync::Arc<File>> {
+    fn stdout(&self) -> Option<std::sync::Arc<tokio::sync::Mutex<File>>> {
         self.fds.get(&1).map(std::sync::Arc::clone)
     }
 
@@ -136,13 +138,13 @@ impl Io {
             1,
             // Safety: we just acquired stdout via into_raw_fd, which acquires
             // ownership of the fd, so we are now the sole owner
-            std::sync::Arc::new(unsafe {
+            std::sync::Arc::new(tokio::sync::Mutex::new(unsafe {
                 File::output(stdout.into_raw_fd())
-            }),
+            })),
         );
     }
 
-    fn stderr(&self) -> Option<std::sync::Arc<File>> {
+    fn stderr(&self) -> Option<std::sync::Arc<tokio::sync::Mutex<File>>> {
         self.fds.get(&2).map(std::sync::Arc::clone)
     }
 
@@ -154,9 +156,9 @@ impl Io {
             2,
             // Safety: we just acquired stderr via into_raw_fd, which acquires
             // ownership of the fd, so we are now the sole owner
-            std::sync::Arc::new(unsafe {
+            std::sync::Arc::new(tokio::sync::Mutex::new(unsafe {
                 File::output(stderr.into_raw_fd())
-            }),
+            })),
         );
     }
 
@@ -172,13 +174,17 @@ impl Io {
                         crate::parse::Direction::In => {
                             // Safety: we just opened fd, and nothing else has
                             // or can use it
-                            std::sync::Arc::new(unsafe { File::input(fd) })
+                            std::sync::Arc::new(tokio::sync::Mutex::new(
+                                unsafe { File::input(fd) },
+                            ))
                         }
                         crate::parse::Direction::Out
                         | crate::parse::Direction::Append => {
                             // Safety: we just opened fd, and nothing else has
                             // or can use it
-                            std::sync::Arc::new(unsafe { File::output(fd) })
+                            std::sync::Arc::new(tokio::sync::Mutex::new(
+                                unsafe { File::output(fd) },
+                            ))
                         }
                     }
                 }
@@ -190,7 +196,7 @@ impl Io {
     pub async fn read_line_stdin(&self) -> anyhow::Result<(String, bool)> {
         let mut buf = vec![];
         if let Some(fh) = self.stdin() {
-            if let File::In(fh) = &*fh {
+            if let File::In(fh) = &mut *fh.clone().lock_owned().await {
                 // we have to read only a single character at a time here
                 // because stdin needs to be shared across all commands in the
                 // command list, some of which may be builtins and others of
@@ -199,9 +205,7 @@ impl Io {
                 // no longer be available to the next command, since we have
                 // them buffered in memory rather than them being on the stdin
                 // pipe.
-                let mut bytes = fh.bytes();
-                while let Some(byte) = bytes.next().await {
-                    let byte = byte?;
+                while let Ok(byte) = fh.read_u8().await {
                     buf.push(byte);
                     if byte == b'\n' {
                         break;
@@ -219,8 +223,8 @@ impl Io {
 
     pub async fn write_stdout(&self, buf: &[u8]) -> anyhow::Result<()> {
         if let Some(fh) = self.stdout() {
-            if let File::Out(fh) = &*fh {
-                Ok((&*fh).write_all(buf).await.map(|_| ())?)
+            if let File::Out(fh) = &mut *fh.clone().lock_owned().await {
+                Ok(fh.write_all(buf).await.map(|_| ())?)
             } else {
                 Ok(())
             }
@@ -231,8 +235,8 @@ impl Io {
 
     pub async fn write_stderr(&self, buf: &[u8]) -> anyhow::Result<()> {
         if let Some(fh) = self.stderr() {
-            if let File::Out(fh) = &*fh {
-                Ok((&*fh).write_all(buf).await.map(|_| ())?)
+            if let File::Out(fh) = &mut *fh.clone().lock_owned().await {
+                Ok(fh.write_all(buf).await.map(|_| ())?)
             } else {
                 Ok(())
             }
@@ -244,7 +248,7 @@ impl Io {
     pub fn setup_command(mut self, cmd: &mut crate::runner::Command) {
         if let Some(stdin) = self.fds.remove(&0) {
             if let Ok(stdin) = std::sync::Arc::try_unwrap(stdin) {
-                let stdin = stdin.into_raw_fd();
+                let stdin = stdin.into_inner().into_raw_fd();
                 if stdin != 0 {
                     // Safety: we just acquired stdin via into_raw_fd, which
                     // acquires ownership of the fd, so we are now the sole
@@ -256,7 +260,7 @@ impl Io {
         }
         if let Some(stdout) = self.fds.remove(&1) {
             if let Ok(stdout) = std::sync::Arc::try_unwrap(stdout) {
-                let stdout = stdout.into_raw_fd();
+                let stdout = stdout.into_inner().into_raw_fd();
                 if stdout != 1 {
                     // Safety: we just acquired stdout via into_raw_fd, which
                     // acquires ownership of the fd, so we are now the sole
@@ -268,7 +272,7 @@ impl Io {
         }
         if let Some(stderr) = self.fds.remove(&2) {
             if let Ok(stderr) = std::sync::Arc::try_unwrap(stderr) {
-                let stderr = stderr.into_raw_fd();
+                let stderr = stderr.into_inner().into_raw_fd();
                 if stderr != 2 {
                     // Safety: we just acquired stderr via into_raw_fd, which
                     // acquires ownership of the fd, so we are now the sole
@@ -291,23 +295,24 @@ impl Drop for Io {
 
 #[derive(Debug)]
 pub enum File {
-    In(async_std::fs::File),
-    Out(async_std::fs::File),
+    In(tokio::fs::File),
+    Out(tokio::fs::File),
 }
 
 impl File {
     // Safety: fd must not be owned by any other File object
     pub unsafe fn input(fd: std::os::unix::io::RawFd) -> Self {
-        Self::In(async_std::fs::File::from_raw_fd(fd))
+        Self::In(tokio::fs::File::from_raw_fd(fd))
     }
 
     // Safety: fd must not be owned by any other File object
     pub unsafe fn output(fd: std::os::unix::io::RawFd) -> Self {
-        Self::Out(async_std::fs::File::from_raw_fd(fd))
+        Self::Out(tokio::fs::File::from_raw_fd(fd))
     }
 
-    fn maybe_drop(file: std::sync::Arc<Self>) {
+    fn maybe_drop(file: std::sync::Arc<tokio::sync::Mutex<Self>>) {
         if let Ok(file) = std::sync::Arc::try_unwrap(file) {
+            let file = file.into_inner();
             if file.as_raw_fd() <= 2 {
                 let _ = file.into_raw_fd();
             }
@@ -326,7 +331,10 @@ impl std::os::unix::io::AsRawFd for File {
 impl std::os::unix::io::IntoRawFd for File {
     fn into_raw_fd(self) -> std::os::unix::io::RawFd {
         match self {
-            Self::In(fh) | Self::Out(fh) => fh.into_raw_fd(),
+            Self::In(fh) | Self::Out(fh) => {
+                // XXX
+                fh.try_into_std().unwrap().into_raw_fd()
+            }
         }
     }
 }
@@ -373,7 +381,7 @@ impl<'a> Child<'a> {
     ) -> std::pin::Pin<
         Box<
             dyn std::future::Future<
-                    Output = anyhow::Result<async_std::process::ExitStatus>,
+                    Output = anyhow::Result<std::process::ExitStatus>,
                 > + Send
                 + Sync
                 + 'a,
diff --git a/src/runner/builtins/mod.rs b/src/runner/builtins/mod.rs
index 5205856..87b5ae7 100644
--- a/src/runner/builtins/mod.rs
+++ b/src/runner/builtins/mod.rs
@@ -88,7 +88,7 @@ fn cd(
                 dir.display()
             );
         }
-        async_std::process::ExitStatus::from_raw(0)
+        std::process::ExitStatus::from_raw(0)
     }
 
     Ok(command::Child::new_fut(async move {
@@ -119,7 +119,7 @@ fn set(
         };
 
         std::env::set_var(k, v);
-        async_std::process::ExitStatus::from_raw(0)
+        std::process::ExitStatus::from_raw(0)
     }
 
     Ok(command::Child::new_fut(async move {
@@ -145,7 +145,7 @@ fn unset(
         };
 
         std::env::remove_var(k);
-        async_std::process::ExitStatus::from_raw(0)
+        std::process::ExitStatus::from_raw(0)
     }
 
     Ok(command::Child::new_fut(async move {
@@ -174,7 +174,7 @@ fn echo(
                         .write_stderr(format!("echo: {}", e).as_bytes())
                         .await
                         .unwrap();
-                    return async_std::process::ExitStatus::from_raw(1 << 8);
+                    return std::process::ExitStatus::from_raw(1 << 8);
                 }
             };
         }
@@ -188,7 +188,7 @@ fn echo(
             }
         }
 
-        async_std::process::ExitStatus::from_raw(0)
+        std::process::ExitStatus::from_raw(0)
     }
 
     Ok(command::Child::new_fut(async move {
@@ -221,11 +221,7 @@ fn read(
         };
 
         std::env::set_var(var, val);
-        async_std::process::ExitStatus::from_raw(if done {
-            1 << 8
-        } else {
-            0
-        })
+        std::process::ExitStatus::from_raw(if done { 1 << 8 } else { 0 })
     }
 
     Ok(command::Child::new_fut(async move {
diff --git a/src/runner/command.rs b/src/runner/command.rs
index 5d4c11e..c7224e6 100644
--- a/src/runner/command.rs
+++ b/src/runner/command.rs
@@ -27,7 +27,7 @@ impl Command {
     pub fn new_binary(exe: crate::parse::Exe) -> Self {
         let exe_path = exe.exe().to_path_buf();
         let redirects = exe.redirects().to_vec();
-        let mut cmd = async_std::process::Command::new(exe.exe());
+        let mut cmd = tokio::process::Command::new(exe.exe());
         cmd.args(exe.args());
         Self {
             inner: Inner::Binary(cmd),
@@ -146,19 +146,19 @@ impl Command {
 }
 
 pub enum Inner {
-    Binary(async_std::process::Command),
+    Binary(tokio::process::Command),
     Builtin(super::builtins::Command),
 }
 
 pub enum Child<'a> {
-    Binary(async_std::process::Child),
+    Binary(tokio::process::Child),
     Builtin(super::builtins::Child<'a>),
 }
 
 impl<'a> Child<'a> {
     pub fn id(&self) -> Option<u32> {
         match self {
-            Self::Binary(child) => Some(child.id()),
+            Self::Binary(child) => child.id(),
             Self::Builtin(child) => child.id(),
         }
     }
@@ -176,7 +176,7 @@ impl<'a> Child<'a> {
     > {
         Box::pin(async move {
             match self {
-                Self::Binary(child) => Ok(child.status_no_drop().await?),
+                Self::Binary(mut child) => Ok(child.wait().await?),
                 Self::Builtin(child) => Ok(child.status().await?),
             }
         })
diff --git a/src/runner/mod.rs b/src/runner/mod.rs
index 1a5003f..d06b332 100644
--- a/src/runner/mod.rs
+++ b/src/runner/mod.rs
@@ -70,7 +70,7 @@ enum Frame {
 
 pub async fn run(
     commands: &str,
-    shell_write: Option<&async_std::fs::File>,
+    shell_write: &mut Option<tokio::fs::File>,
 ) -> anyhow::Result<i32> {
     let mut env = Env::new_from_env()?;
     run_commands(commands, &mut env, shell_write).await?;
@@ -86,7 +86,7 @@ pub async fn run(
 async fn run_commands(
     commands: &str,
     env: &mut Env,
-    shell_write: Option<&async_std::fs::File>,
+    shell_write: &mut Option<tokio::fs::File>,
 ) -> anyhow::Result<()> {
     let commands = crate::parse::ast::Commands::parse(commands)?;
     let commands = commands.commands();
@@ -152,7 +152,7 @@ async fn run_commands(
                                         .map(IntoIterator::into_iter)
                                 })
                                 .collect::<futures_util::stream::FuturesOrdered<_>>()
-                                .collect::<Result<Vec<_>, _>>().await?
+                                .try_collect::<Vec<_>>().await?
                                 .into_iter()
                                 .flatten()
                                 .collect()
@@ -231,7 +231,7 @@ async fn run_commands(
 async fn run_pipeline(
     pipeline: crate::parse::ast::Pipeline,
     env: &mut Env,
-    shell_write: Option<&async_std::fs::File>,
+    shell_write: &mut Option<tokio::fs::File>,
 ) -> anyhow::Result<()> {
     write_event(shell_write, Event::RunPipeline(env.idx(), pipeline.span()))
         .await?;
@@ -240,9 +240,9 @@ async fn run_pipeline(
     // level would not be safe, because in the case of a command line like
     // "echo foo; ls", we would pass the stdout fd to the ls process while it
     // is still open here, and may still have data buffered.
-    let stdin = unsafe { async_std::fs::File::from_raw_fd(0) };
-    let stdout = unsafe { async_std::fs::File::from_raw_fd(1) };
-    let stderr = unsafe { async_std::fs::File::from_raw_fd(2) };
+    let stdin = unsafe { std::fs::File::from_raw_fd(0) };
+    let stdout = unsafe { std::fs::File::from_raw_fd(1) };
+    let stderr = unsafe { std::fs::File::from_raw_fd(2) };
     let mut io = builtins::Io::new();
     io.set_stdin(stdin);
     io.set_stdout(stdout);
@@ -265,10 +265,10 @@ async fn run_pipeline(
 }
 
 async fn write_event(
-    fh: Option<&async_std::fs::File>,
+    fh: &mut Option<tokio::fs::File>,
     event: Event,
 ) -> anyhow::Result<()> {
-    if let Some(mut fh) = fh {
+    if let Some(fh) = fh {
         fh.write_all(&bincode::serialize(&event)?).await?;
         fh.flush().await?;
     }
@@ -322,11 +322,11 @@ async fn wait_children(
     pg: Option<nix::unistd::Pid>,
     env: &Env,
     io: &builtins::Io,
-    shell_write: Option<&async_std::fs::File>,
+    shell_write: &mut Option<tokio::fs::File>,
 ) -> std::process::ExitStatus {
     enum Res {
         Child(nix::Result<nix::sys::wait::WaitStatus>),
-        Builtin(Option<(anyhow::Result<std::process::ExitStatus>, bool)>),
+        Builtin((anyhow::Result<std::process::ExitStatus>, bool)),
     }
 
     macro_rules! bail {
@@ -353,7 +353,8 @@ async fn wait_children(
             (sys::id_to_pid(child.id().unwrap()), (child, i == count - 1))
         })
         .collect();
-    let mut builtins: futures_util::stream::FuturesUnordered<_> =
+    let mut builtin_count = builtins.len();
+    let builtins: futures_util::stream::FuturesUnordered<_> =
         builtins
             .into_iter()
             .map(|(i, child)| async move {
@@ -361,47 +362,40 @@ async fn wait_children(
             })
             .collect();
 
-    let (wait_w, wait_r) = async_std::channel::unbounded();
-    let new_wait = move || {
-        if let Some(pg) = pg {
-            let wait_w = wait_w.clone();
-            async_std::task::spawn(async move {
-                let res = blocking::unblock(move || {
-                    nix::sys::wait::waitpid(
-                        sys::neg_pid(pg),
-                        Some(nix::sys::wait::WaitPidFlag::WUNTRACED),
-                    )
-                })
-                .await;
-                if wait_w.is_closed() {
-                    // we shouldn't be able to drop real process terminations
+    let (wait_w, wait_r) = tokio::sync::mpsc::unbounded_channel();
+    if let Some(pg) = pg {
+        tokio::task::spawn_blocking(move || loop {
+            let res = nix::sys::wait::waitpid(
+                sys::neg_pid(pg),
+                Some(nix::sys::wait::WaitPidFlag::WUNTRACED),
+            );
+            match wait_w.send(res) {
+                Ok(_) => {}
+                Err(tokio::sync::mpsc::error::SendError(res)) => {
+                    // we should never drop wait_r while there are still valid
+                    // things to read
                     assert!(res.is_err());
-                } else {
-                    wait_w.send(res).await.unwrap();
+                    break;
                 }
-            });
-        }
-    };
-
-    new_wait();
-    loop {
-        if children.is_empty() && builtins.is_empty() {
-            break;
-        }
+            }
+        });
+    }
 
-        let child = async { Res::Child(wait_r.recv().await.unwrap()) };
-        let builtin = async {
-            Res::Builtin(if builtins.is_empty() {
-                std::future::pending().await
-            } else {
-                builtins.next().await
-            })
-        };
-        match child.race(builtin).await {
+    let mut stream: futures_util::stream::SelectAll<_> = [
+        tokio_stream::wrappers::UnboundedReceiverStream::new(wait_r)
+            .map(Res::Child)
+            .boxed(),
+        builtins.map(Res::Builtin).boxed(),
+    ]
+    .into_iter()
+    .collect();
+    while let Some(res) = stream.next().await {
+        match res {
             Res::Child(Ok(status)) => {
                 match status {
-                    // we can't call child.status() here to unify these branches
-                    // because our waitpid call already collected the status
+                    // we can't call child.status() here to unify these
+                    // branches because our waitpid call already collected the
+                    // status
                     nix::sys::wait::WaitStatus::Exited(pid, code) => {
                         let (_, last) = children.remove(&pid).unwrap();
                         if last {
@@ -449,12 +443,11 @@ async fn wait_children(
                     }
                     _ => {}
                 }
-                new_wait();
             }
             Res::Child(Err(e)) => {
                 bail!(e);
             }
-            Res::Builtin(Some((Ok(status), last))) => {
+            Res::Builtin((Ok(status), last)) => {
                 // this conversion is safe because the Signal enum is
                 // repr(i32)
                 #[allow(clippy::as_conversions)]
@@ -470,11 +463,15 @@ async fn wait_children(
                 if last {
                     final_status = Some(status);
                 }
+                builtin_count -= 1;
             }
-            Res::Builtin(Some((Err(e), _))) => {
+            Res::Builtin((Err(e), _)) => {
                 bail!(e);
             }
-            Res::Builtin(None) => {}
+        }
+
+        if children.is_empty() && builtin_count == 0 {
+            break;
         }
     }
 
diff --git a/src/shell/event.rs b/src/shell/event.rs
index 025f3c4..ad14705 100644
--- a/src/shell/event.rs
+++ b/src/shell/event.rs
@@ -11,22 +11,23 @@ pub enum Event {
 }
 
 pub struct Reader {
-    pending: async_std::sync::Mutex<Pending>,
-    cvar: async_std::sync::Condvar,
+    pending: tokio::sync::Mutex<Pending>,
+    cvar: tokio::sync::Notify,
 }
 
 impl Reader {
     pub fn new(
-        input: async_std::channel::Receiver<Event>,
-    ) -> async_std::sync::Arc<Self> {
-        let this = async_std::sync::Arc::new(Self {
-            pending: async_std::sync::Mutex::new(Pending::new()),
-            cvar: async_std::sync::Condvar::new(),
-        });
+        mut input: tokio::sync::mpsc::UnboundedReceiver<Event>,
+    ) -> std::sync::Arc<Self> {
+        let this = Self {
+            pending: tokio::sync::Mutex::new(Pending::new()),
+            cvar: tokio::sync::Notify::new(),
+        };
+        let this = std::sync::Arc::new(this);
         {
-            let this = async_std::sync::Arc::clone(&this);
-            async_std::task::spawn(async move {
-                while let Ok(event) = input.recv().await {
+            let this = this.clone();
+            tokio::task::spawn(async move {
+                while let Some(event) = input.recv().await {
                     this.new_event(Some(event)).await;
                 }
                 this.new_event(None).await;
@@ -36,13 +37,14 @@ impl Reader {
     }
 
     pub async fn recv(&self) -> Option<Event> {
-        let mut pending = self
-            .cvar
-            .wait_until(self.pending.lock().await, |pending| {
-                pending.has_event()
-            })
-            .await;
-        pending.get_event()
+        loop {
+            let mut pending = self.pending.lock().await;
+            if pending.has_event() {
+                return pending.get_event();
+            }
+            drop(pending);
+            self.cvar.notified().await;
+        }
     }
 
     async fn new_event(&self, event: Option<Event>) {
diff --git a/src/shell/history/entry.rs b/src/shell/history/entry.rs
index a45d99d..97e8a7b 100644
--- a/src/shell/history/entry.rs
+++ b/src/shell/history/entry.rs
@@ -16,8 +16,8 @@ pub struct Entry {
     visual_bell: bool,
     real_bell_pending: bool,
     fullscreen: Option<bool>,
-    input: async_std::channel::Sender<Vec<u8>>,
-    resize: async_std::channel::Sender<(u16, u16)>,
+    input: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    resize: tokio::sync::mpsc::UnboundedSender<(u16, u16)>,
     start_time: time::OffsetDateTime,
     start_instant: std::time::Instant,
 }
@@ -27,8 +27,8 @@ impl Entry {
         cmdline: String,
         env: Env,
         size: (u16, u16),
-        input: async_std::channel::Sender<Vec<u8>>,
-        resize: async_std::channel::Sender<(u16, u16)>,
+        input: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+        resize: tokio::sync::mpsc::UnboundedSender<(u16, u16)>,
     ) -> Self {
         let span = (0, cmdline.len());
         Self {
@@ -229,13 +229,13 @@ impl Entry {
 
     pub async fn send_input(&self, bytes: Vec<u8>) {
         if self.running() {
-            self.input.send(bytes).await.unwrap();
+            self.input.send(bytes).unwrap();
         }
     }
 
     pub async fn resize(&mut self, size: (u16, u16)) {
         if self.running() {
-            self.resize.send(size).await.unwrap();
+            self.resize.send(size).unwrap();
             self.vt.set_size(size.0, size.1);
         }
     }
@@ -341,11 +341,11 @@ impl Entry {
     pub async fn finish(
         &mut self,
         env: Env,
-        event_w: async_std::channel::Sender<Event>,
+        event_w: tokio::sync::mpsc::UnboundedSender<Event>,
     ) {
         self.state = State::Exited(ExitInfo::new(env.latest_status()));
         self.env = env;
-        event_w.send(Event::PtyClose).await.unwrap();
+        event_w.send(Event::PtyClose).unwrap();
     }
 
     fn exit_info(&self) -> Option<&ExitInfo> {
@@ -369,12 +369,12 @@ impl Entry {
 }
 
 struct ExitInfo {
-    status: async_std::process::ExitStatus,
+    status: std::process::ExitStatus,
     instant: std::time::Instant,
 }
 
 impl ExitInfo {
-    fn new(status: async_std::process::ExitStatus) -> Self {
+    fn new(status: std::process::ExitStatus) -> Self {
         Self {
             status,
             instant: std::time::Instant::now(),
diff --git a/src/shell/history/mod.rs b/src/shell/history/mod.rs
index 1bc4e62..2eeab0b 100644
--- a/src/shell/history/mod.rs
+++ b/src/shell/history/mod.rs
@@ -67,7 +67,7 @@ impl History {
         out: &mut impl textmode::Textmode,
         idx: usize,
     ) {
-        let mut entry = self.entries[idx].lock_arc().await;
+        let mut entry = self.entries[idx].clone().lock_owned().await;
         entry.render_fullscreen(out);
     }
 
@@ -78,7 +78,7 @@ impl History {
     pub async fn resize(&mut self, size: (u16, u16)) {
         self.size = size;
         for entry in &self.entries {
-            entry.lock_arc().await.resize(size).await;
+            entry.clone().lock_owned().await.resize(size).await;
         }
     }
 
@@ -86,10 +86,10 @@ impl History {
         &mut self,
         cmdline: &str,
         env: &Env,
-        event_w: async_std::channel::Sender<Event>,
+        event_w: tokio::sync::mpsc::UnboundedSender<Event>,
     ) -> anyhow::Result<usize> {
-        let (input_w, input_r) = async_std::channel::unbounded();
-        let (resize_w, resize_r) = async_std::channel::unbounded();
+        let (input_w, input_r) = tokio::sync::mpsc::unbounded_channel();
+        let (resize_w, resize_r) = tokio::sync::mpsc::unbounded_channel();
 
         let entry = crate::mutex::new(Entry::new(
             cmdline.to_string(),
@@ -112,7 +112,7 @@ impl History {
     }
 
     pub async fn entry(&self, idx: usize) -> crate::mutex::Guard<Entry> {
-        self.entries[idx].lock_arc().await
+        self.entries[idx].clone().lock_owned().await
     }
 
     pub fn entry_count(&self) -> usize {
@@ -173,7 +173,7 @@ impl History {
         for (idx, entry) in
             self.entries.iter().enumerate().rev().skip(self.scroll_pos)
         {
-            let entry = entry.lock_arc().await;
+            let entry = entry.clone().lock_owned().await;
             let focused = focus.map_or(false, |focus| idx == focus);
             used_lines +=
                 entry.lines(self.entry_count(), focused && !scrolling);
@@ -221,13 +221,13 @@ fn run_commands(
     cmdline: String,
     entry: crate::mutex::Mutex<Entry>,
     mut env: Env,
-    input_r: async_std::channel::Receiver<Vec<u8>>,
-    resize_r: async_std::channel::Receiver<(u16, u16)>,
-    event_w: async_std::channel::Sender<Event>,
+    input_r: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
+    resize_r: tokio::sync::mpsc::UnboundedReceiver<(u16, u16)>,
+    event_w: tokio::sync::mpsc::UnboundedSender<Event>,
 ) {
-    async_std::task::spawn(async move {
+    tokio::task::spawn(async move {
         let pty = match pty::Pty::new(
-            entry.lock_arc().await.size(),
+            entry.clone().lock_owned().await.size(),
             &entry,
             input_r,
             resize_r,
@@ -235,14 +235,12 @@ fn run_commands(
         ) {
             Ok(pty) => pty,
             Err(e) => {
-                let mut entry = entry.lock_arc().await;
+                let mut entry = entry.clone().lock_owned().await;
                 entry.process(
                     format!("nbsh: failed to allocate pty: {}\r\n", e)
                         .as_bytes(),
                 );
-                env.set_status(async_std::process::ExitStatus::from_raw(
-                    1 << 8,
-                ));
+                env.set_status(std::process::ExitStatus::from_raw(1 << 8));
                 entry.finish(env, event_w).await;
                 return;
             }
@@ -254,7 +252,7 @@ fn run_commands(
             {
                 Ok(status) => status,
                 Err(e) => {
-                    let mut entry = entry.lock_arc().await;
+                    let mut entry = entry.clone().lock_owned().await;
                     entry.process(
                         format!(
                             "nbsh: failed to spawn {}: {}\r\n",
@@ -262,7 +260,7 @@ fn run_commands(
                         )
                         .as_bytes(),
                     );
-                    env.set_status(async_std::process::ExitStatus::from_raw(
+                    env.set_status(std::process::ExitStatus::from_raw(
                         1 << 8,
                     ));
                     entry.finish(env, event_w).await;
@@ -271,7 +269,7 @@ fn run_commands(
             };
         env.set_status(status);
 
-        entry.lock_arc().await.finish(env, event_w).await;
+        entry.clone().lock_owned().await.finish(env, event_w).await;
         pty.close().await;
     });
 }
@@ -280,12 +278,19 @@ async fn spawn_commands(
     cmdline: &str,
     pty: &pty::Pty,
     env: &mut Env,
-    event_w: async_std::channel::Sender<Event>,
-) -> anyhow::Result<async_std::process::ExitStatus> {
+    event_w: tokio::sync::mpsc::UnboundedSender<Event>,
+) -> anyhow::Result<std::process::ExitStatus> {
+    enum Res {
+        Read(crate::runner::Event),
+        Exit(std::io::Result<std::process::ExitStatus>),
+    }
+
     let mut cmd = pty_process::Command::new(std::env::current_exe()?);
     cmd.args(&["-c", cmdline, "--status-fd", "3"]);
     env.apply(&mut cmd);
     let (from_r, from_w) = nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC)?;
+    // Safety: from_r was just opened above and is not used anywhere else
+    let fh = unsafe { std::fs::File::from_raw_fd(from_r) };
     // Safety: dup2 is an async-signal-safe function
     unsafe {
         cmd.pre_exec(move || {
@@ -293,90 +298,63 @@ async fn spawn_commands(
             Ok(())
         });
     }
-    let child = pty.spawn(cmd)?;
+    let mut child = pty.spawn(cmd)?;
     nix::unistd::close(from_w)?;
 
-    let (read_w, read_r) = async_std::channel::unbounded();
-    let new_read = move || {
-        let read_w = read_w.clone();
-        async_std::task::spawn(async move {
-            let event = blocking::unblock(move || {
-                // Safety: from_r was just opened above and is only
-                // referenced in this closure, which takes ownership of it
-                // at the start and returns ownership of it at the end
-                let fh = unsafe { std::fs::File::from_raw_fd(from_r) };
-                let event = bincode::deserialize_from(&fh);
-                let _ = fh.into_raw_fd();
-                event
-            })
-            .await;
-            if read_w.is_closed() {
-                // we should never drop read_r while there are still valid
-                // things to read
-                assert!(event.is_err());
-            } else {
-                read_w.send(event).await.unwrap();
+    let (read_w, read_r) = tokio::sync::mpsc::unbounded_channel();
+    tokio::task::spawn_blocking(move || loop {
+        let event = bincode::deserialize_from(&fh);
+        match event {
+            Ok(event) => {
+                read_w.send(event).unwrap();
+            }
+            Err(e) => {
+                match &*e {
+                    bincode::ErrorKind::Io(io_e) => {
+                        assert!(
+                            io_e.kind() == std::io::ErrorKind::UnexpectedEof
+                        );
+                    }
+                    e => {
+                        panic!("{}", e);
+                    }
+                }
+                break;
             }
-        });
-    };
-
-    new_read();
-    let mut read_done = false;
-    let mut exit_done = None;
-    loop {
-        enum Res {
-            Read(bincode::Result<crate::runner::Event>),
-            Exit(std::io::Result<std::process::ExitStatus>),
         }
+    });
 
-        let read_r = read_r.clone();
-        let read = async move { Res::Read(read_r.recv().await.unwrap()) };
-        let exit = async {
-            Res::Exit(if exit_done.is_none() {
-                child.status_no_drop().await
-            } else {
-                std::future::pending().await
-            })
-        };
-        match read.or(exit).await {
-            Res::Read(Ok(event)) => match event {
+    let mut stream: futures_util::stream::SelectAll<_> = [
+        tokio_stream::wrappers::UnboundedReceiverStream::new(read_r)
+            .map(Res::Read)
+            .boxed(),
+        futures_util::stream::once(child.wait())
+            .map(Res::Exit)
+            .boxed(),
+    ]
+    .into_iter()
+    .collect();
+    let mut exit_status = None;
+    while let Some(res) = stream.next().await {
+        match res {
+            Res::Read(event) => match event {
                 crate::runner::Event::RunPipeline(idx, span) => {
-                    event_w
-                        .send(Event::ChildRunPipeline(idx, span))
-                        .await
-                        .unwrap();
-                    new_read();
+                    event_w.send(Event::ChildRunPipeline(idx, span)).unwrap();
                 }
                 crate::runner::Event::Suspend(idx) => {
-                    event_w.send(Event::ChildSuspend(idx)).await.unwrap();
-                    new_read();
+                    event_w.send(Event::ChildSuspend(idx)).unwrap();
                 }
                 crate::runner::Event::Exit(new_env) => {
                     *env = new_env;
-                    read_done = true;
                 }
             },
-            Res::Read(Err(e)) => {
-                if let bincode::ErrorKind::Io(io_e) = &*e {
-                    if io_e.kind() == std::io::ErrorKind::UnexpectedEof {
-                        read_done = true;
-                    } else {
-                        anyhow::bail!(e);
-                    }
-                } else {
-                    anyhow::bail!(e);
-                }
-            }
             Res::Exit(Ok(status)) => {
-                exit_done = Some(status);
+                exit_status = Some(status);
             }
             Res::Exit(Err(e)) => {
                 anyhow::bail!(e);
             }
         }
-        if let (true, Some(status)) = (read_done, exit_done) {
-            nix::unistd::close(from_r)?;
-            return Ok(status);
-        }
     }
+    Ok(exit_status.unwrap())
 }
diff --git a/src/shell/history/pty.rs b/src/shell/history/pty.rs
index 5a51e73..acfe500 100644
--- a/src/shell/history/pty.rs
+++ b/src/shell/history/pty.rs
@@ -1,26 +1,26 @@
 use crate::shell::prelude::*;
 
 pub struct Pty {
-    pty: async_std::sync::Arc<pty_process::Pty>,
-    close_w: async_std::channel::Sender<()>,
+    pts: pty_process::Pts,
+    close_w: tokio::sync::mpsc::UnboundedSender<()>,
 }
 
 impl Pty {
     pub fn new(
         size: (u16, u16),
         entry: &crate::mutex::Mutex<super::Entry>,
-        input_r: async_std::channel::Receiver<Vec<u8>>,
-        resize_r: async_std::channel::Receiver<(u16, u16)>,
-        event_w: async_std::channel::Sender<Event>,
+        input_r: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
+        resize_r: tokio::sync::mpsc::UnboundedReceiver<(u16, u16)>,
+        event_w: tokio::sync::mpsc::UnboundedSender<Event>,
     ) -> anyhow::Result<Self> {
-        let (close_w, close_r) = async_std::channel::unbounded();
+        let (close_w, close_r) = tokio::sync::mpsc::unbounded_channel();
 
         let pty = pty_process::Pty::new()?;
         pty.resize(pty_process::Size::new(size.0, size.1))?;
-        let pty = async_std::sync::Arc::new(pty);
+        let pts = pty.pts()?;
 
-        async_std::task::spawn(pty_task(
-            async_std::sync::Arc::clone(&pty),
+        tokio::task::spawn(pty_task(
+            pty,
             crate::mutex::clone(entry),
             input_r,
             resize_r,
@@ -28,80 +28,74 @@ impl Pty {
             event_w,
         ));
 
-        Ok(Self { pty, close_w })
+        Ok(Self { pts, close_w })
     }
 
     pub fn spawn(
         &self,
         mut cmd: pty_process::Command,
-    ) -> anyhow::Result<async_std::process::Child> {
-        Ok(cmd.spawn(&self.pty)?)
+    ) -> anyhow::Result<tokio::process::Child> {
+        Ok(cmd.spawn(&self.pts)?)
     }
 
     pub async fn close(&self) {
-        self.close_w.send(()).await.unwrap();
+        self.close_w.send(()).unwrap();
     }
 }
 
 async fn pty_task(
-    pty: async_std::sync::Arc<pty_process::Pty>,
+    pty: pty_process::Pty,
     entry: crate::mutex::Mutex<super::Entry>,
-    input_r: async_std::channel::Receiver<Vec<u8>>,
-    resize_r: async_std::channel::Receiver<(u16, u16)>,
-    close_r: async_std::channel::Receiver<()>,
-    event_w: async_std::channel::Sender<Event>,
+    input_r: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
+    resize_r: tokio::sync::mpsc::UnboundedReceiver<(u16, u16)>,
+    close_r: tokio::sync::mpsc::UnboundedReceiver<()>,
+    event_w: tokio::sync::mpsc::UnboundedSender<Event>,
 ) {
-    loop {
-        enum Res {
-            Read(Result<usize, std::io::Error>),
-            Write(Result<Vec<u8>, async_std::channel::RecvError>),
-            Resize(Result<(u16, u16), async_std::channel::RecvError>),
-            Close(Result<(), async_std::channel::RecvError>),
-        }
-        let mut buf = [0_u8; 4096];
-        let read = async { Res::Read((&*pty).read(&mut buf).await) };
-        let write = async { Res::Write(input_r.recv().await) };
-        let resize = async { Res::Resize(resize_r.recv().await) };
-        let close = async { Res::Close(close_r.recv().await) };
-        match read.race(write).race(resize).or(close).await {
+    enum Res {
+        Read(Result<bytes::Bytes, std::io::Error>),
+        Write(Vec<u8>),
+        Resize((u16, u16)),
+        Close(()),
+    }
+
+    let (pty_r, mut pty_w) = pty.into_split();
+    let mut stream: futures_util::stream::SelectAll<_> = [
+        tokio_util::io::ReaderStream::new(pty_r)
+            .map(Res::Read)
+            .boxed(),
+        tokio_stream::wrappers::UnboundedReceiverStream::new(input_r)
+            .map(Res::Write)
+            .boxed(),
+        tokio_stream::wrappers::UnboundedReceiverStream::new(resize_r)
+            .map(Res::Resize)
+            .boxed(),
+        tokio_stream::wrappers::UnboundedReceiverStream::new(close_r)
+            .map(Res::Close)
+            .boxed(),
+    ]
+    .into_iter()
+    .collect();
+    while let Some(res) = stream.next().await {
+        match res {
             Res::Read(res) => match res {
                 Ok(bytes) => {
-                    entry.lock_arc().await.process(&buf[..bytes]);
-                    event_w.send(Event::PtyOutput).await.unwrap();
+                    entry.clone().lock_owned().await.process(&bytes);
+                    event_w.send(Event::PtyOutput).unwrap();
                 }
                 Err(e) => {
-                    if e.raw_os_error() == Some(libc::EIO) {
-                        continue;
-                    }
                     panic!("pty read failed: {:?}", e);
                 }
             },
-            Res::Write(res) => match res {
-                Ok(bytes) => {
-                    (&*pty).write(&bytes).await.unwrap();
-                }
-                Err(e) => {
-                    panic!("failed to read from input channel: {}", e);
-                }
-            },
-            Res::Resize(res) => match res {
-                Ok(size) => {
-                    pty.resize(pty_process::Size::new(size.0, size.1))
-                        .unwrap();
-                }
-                Err(e) => {
-                    panic!("failed to read from resize channel: {}", e);
-                }
-            },
-            Res::Close(res) => match res {
-                Ok(()) => {
-                    event_w.send(Event::PtyClose).await.unwrap();
-                    return;
-                }
-                Err(e) => {
-                    panic!("failed to read from close channel: {}", e);
-                }
-            },
+            Res::Write(bytes) => {
+                pty_w.write(&bytes).await.unwrap();
+            }
+            Res::Resize(size) => pty_w
+                .resize(pty_process::Size::new(size.0, size.1))
+                .unwrap(),
+            Res::Close(()) => {
+                event_w.send(Event::PtyClose).unwrap();
+                return;
+            }
         }
     }
 }
diff --git a/src/shell/mod.rs b/src/shell/mod.rs
index 9c4002b..82d2021 100644
--- a/src/shell/mod.rs
+++ b/src/shell/mod.rs
@@ -10,7 +10,7 @@ mod prelude;
 mod readline;
 
 pub async fn main() -> anyhow::Result<i32> {
-    let mut input = textmode::Input::new().await?;
+    let mut input = textmode::blocking::Input::new()?;
     let mut output = textmode::Output::new().await?;
 
     // avoid the guards getting stuck in a task that doesn't run to
@@ -18,23 +18,23 @@ pub async fn main() -> anyhow::Result<i32> {
     let _input_guard = input.take_raw_guard();
     let _output_guard = output.take_screen_guard();
 
-    let (event_w, event_r) = async_std::channel::unbounded();
+    let (event_w, event_r) = tokio::sync::mpsc::unbounded_channel();
 
     {
-        // nix::sys::signal::Signal is repr(i32)
-        #[allow(clippy::as_conversions)]
-        let signals = signal_hook_async_std::Signals::new(&[
-            nix::sys::signal::Signal::SIGWINCH as i32,
-        ])?;
+        let mut signals = tokio::signal::unix::signal(
+            tokio::signal::unix::SignalKind::window_change(),
+        )?;
         let event_w = event_w.clone();
-        async_std::task::spawn(async move {
-            // nix::sys::signal::Signal is repr(i32)
-            #[allow(clippy::as_conversions)]
-            let mut signals = async_std::stream::once(
-                nix::sys::signal::Signal::SIGWINCH as i32,
-            )
-            .chain(signals);
-            while signals.next().await.is_some() {
+        tokio::task::spawn(async move {
+            event_w
+                .send(Event::Resize(terminal_size::terminal_size().map_or(
+                    (24, 80),
+                    |(terminal_size::Width(w), terminal_size::Height(h))| {
+                        (h, w)
+                    },
+                )))
+                .unwrap();
+            while signals.recv().await.is_some() {
                 event_w
                     .send(Event::Resize(
                         terminal_size::terminal_size().map_or(
@@ -45,7 +45,6 @@ pub async fn main() -> anyhow::Result<i32> {
                             )| { (h, w) },
                         ),
                     ))
-                    .await
                     .unwrap();
             }
         });
@@ -53,9 +52,9 @@ pub async fn main() -> anyhow::Result<i32> {
 
     {
         let event_w = event_w.clone();
-        async_std::task::spawn(async move {
-            while let Some(key) = input.read_key().await.unwrap() {
-                event_w.send(Event::Key(key)).await.unwrap();
+        std::thread::spawn(move || {
+            while let Some(key) = input.read_key().unwrap() {
+                event_w.send(Event::Key(key)).unwrap();
             }
         });
     }
@@ -63,33 +62,35 @@ pub async fn main() -> anyhow::Result<i32> {
     // redraw the clock every second
     {
         let event_w = event_w.clone();
-        async_std::task::spawn(async move {
-            let first_sleep = 1_000_000_000_u64.saturating_sub(
-                time::OffsetDateTime::now_utc().nanosecond().into(),
-            );
-            async_std::task::sleep(std::time::Duration::from_nanos(
-                first_sleep,
-            ))
-            .await;
-            let mut interval = async_std::stream::interval(
+        tokio::task::spawn(async move {
+            let now_clock = time::OffsetDateTime::now_utc();
+            let now_instant = tokio::time::Instant::now();
+            let mut interval = tokio::time::interval_at(
+                now_instant
+                    + std::time::Duration::from_nanos(
+                        1_000_000_000_u64
+                            .saturating_sub(now_clock.nanosecond().into()),
+                    ),
                 std::time::Duration::from_secs(1),
             );
-            event_w.send(Event::ClockTimer).await.unwrap();
-            while interval.next().await.is_some() {
-                event_w.send(Event::ClockTimer).await.unwrap();
+            loop {
+                interval.tick().await;
+                event_w.send(Event::ClockTimer).unwrap();
             }
         });
     }
 
-    let (git_w, git_r): (async_std::channel::Sender<std::path::PathBuf>, _) =
-        async_std::channel::unbounded();
+    let (git_w, mut git_r): (
+        tokio::sync::mpsc::UnboundedSender<std::path::PathBuf>,
+        _,
+    ) = tokio::sync::mpsc::unbounded_channel();
     {
         let event_w = event_w.clone();
         // clippy can't tell that we assign to this later
         #[allow(clippy::no_effect_underscore_binding)]
         let mut _active_watcher = None;
-        async_std::task::spawn(async move {
-            while let Ok(mut dir) = git_r.recv().await {
+        tokio::task::spawn(async move {
+            while let Some(mut dir) = git_r.recv().await {
                 while let Ok(newer_dir) = git_r.try_recv() {
                     dir = newer_dir;
                 }
@@ -97,7 +98,8 @@ pub async fn main() -> anyhow::Result<i32> {
                 if repo.is_some() {
                     let (sync_watch_w, sync_watch_r) =
                         std::sync::mpsc::channel();
-                    let (watch_w, watch_r) = async_std::channel::unbounded();
+                    let (watch_w, mut watch_r) =
+                        tokio::sync::mpsc::unbounded_channel();
                     let mut watcher = notify::RecommendedWatcher::new(
                         sync_watch_w,
                         std::time::Duration::from_millis(100),
@@ -106,31 +108,25 @@ pub async fn main() -> anyhow::Result<i32> {
                     watcher
                         .watch(&dir, notify::RecursiveMode::Recursive)
                         .unwrap();
-                    async_std::task::spawn(blocking::unblock(move || {
+                    tokio::task::spawn_blocking(move || {
                         while let Ok(event) = sync_watch_r.recv() {
                             let watch_w = watch_w.clone();
-                            let send_failed =
-                                async_std::task::block_on(async move {
-                                    watch_w.send(event).await.is_err()
-                                });
+                            let send_failed = watch_w.send(event).is_err();
                             if send_failed {
                                 break;
                             }
                         }
-                    }));
+                    });
                     let event_w = event_w.clone();
-                    async_std::task::spawn(async move {
-                        while watch_r.recv().await.is_ok() {
+                    tokio::task::spawn(async move {
+                        while watch_r.recv().await.is_some() {
                             let repo = git2::Repository::discover(&dir).ok();
-                            let info = blocking::unblock(|| {
+                            let info = tokio::task::spawn_blocking(|| {
                                 repo.map(|repo| git::Info::new(&repo))
                             })
-                            .await;
-                            if event_w
-                                .send(Event::GitInfo(info))
-                                .await
-                                .is_err()
-                            {
+                            .await
+                            .unwrap();
+                            if event_w.send(Event::GitInfo(info)).is_err() {
                                 break;
                             }
                         }
@@ -139,24 +135,25 @@ pub async fn main() -> anyhow::Result<i32> {
                 } else {
                     _active_watcher = None;
                 }
-                let info = blocking::unblock(|| {
+                let info = tokio::task::spawn_blocking(|| {
                     repo.map(|repo| git::Info::new(&repo))
                 })
-                .await;
-                event_w.send(Event::GitInfo(info)).await.unwrap();
+                .await
+                .unwrap();
+                event_w.send(Event::GitInfo(info)).unwrap();
             }
         });
     }
 
     let mut shell = Shell::new(crate::info::get_offset())?;
     let mut prev_dir = shell.env.pwd().to_path_buf();
-    git_w.send(prev_dir.clone()).await.unwrap();
+    git_w.send(prev_dir.clone()).unwrap();
     let event_reader = event::Reader::new(event_r);
     while let Some(event) = event_reader.recv().await {
         let dir = shell.env().pwd();
         if dir != prev_dir {
             prev_dir = dir.to_path_buf();
-            git_w.send(dir.to_path_buf()).await.unwrap();
+            git_w.send(dir.to_path_buf()).unwrap();
         }
         match shell.handle_event(event, &event_w).await {
             Some(Action::Refresh) => {
@@ -322,7 +319,7 @@ impl Shell {
     pub async fn handle_event(
         &mut self,
         event: Event,
-        event_w: &async_std::channel::Sender<Event>,
+        event_w: &tokio::sync::mpsc::UnboundedSender<Event>,
     ) -> Option<Action> {
         match event {
             Event::Key(key) => {
@@ -405,7 +402,7 @@ impl Shell {
     async fn handle_key_escape(
         &mut self,
         key: textmode::Key,
-        event_w: async_std::channel::Sender<Event>,
+        event_w: tokio::sync::mpsc::UnboundedSender<Event>,
     ) -> Option<Action> {
         match key {
             textmode::Key::Ctrl(b'd') => {
@@ -514,7 +511,7 @@ impl Shell {
     async fn handle_key_readline(
         &mut self,
         key: textmode::Key,
-        event_w: async_std::channel::Sender<Event>,
+        event_w: tokio::sync::mpsc::UnboundedSender<Event>,
     ) -> Option<Action> {
         match key {
             textmode::Key::Char(c) => {