move to tokio

13 files changed, 212 insertions, 216 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index f845022..f6533e8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Changelog
 
+## [Unreleased]
+
+### Changed
+
+* migrated to tokio
+
 ## [0.3.0] - 2021-12-15
 
 ### Added
diff --git a/Cargo.toml b/Cargo.toml
index 085302c..7d82025 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -14,21 +14,24 @@ include = ["src/**/*", "LICENSE", "README.md", "CHANGELOG.md"]
 
 [dependencies]
 itoa = "1.0.1"
-nix = "0.23.0"
+nix = "0.23.1"
 terminal_size = "0.1.17"
-vt100 = "0.15.0"
+vt100 = "0.15.1"
 
-blocking = { version = "1.1.0", optional = true }
-futures-lite = { version = "1.12.0", optional = true }
+tokio = { version = "1.17.0", features = ["io-std", "io-util", "rt", "rt-multi-thread"], optional = true }
 
 [features]
 default = []
-async = ["blocking", "futures-lite"]
+async = ["tokio"]
 
 [dev-dependencies]
-assert_cmd = "2.0.2"
-assert_fs = "1.0.6"
+assert_cmd = "2.0.4"
+assert_fs = "1.0.7"
 escargot = "0.5.7"
-libc = "0.2.112"
-pty-process = { version = "0.2.0", features = ["backend-smol"] }
-smol = "1.2.5"
+libc = "0.2.119"
+pty-process = { version = "0.2.0", features = ["async"] }
+tokio = { version = "1.17.0", features = ["full"] }
+
+[patch.crates-io]
+nix = { path = "../src/nix" }
+pty-process = { path = "../pty-process" }
diff --git a/examples/basic.rs b/examples/basic.rs
index dce12cc..db56d20 100644
--- a/examples/basic.rs
+++ b/examples/basic.rs
@@ -1,10 +1,10 @@
 use textmode::Textmode as _;
 
 #[cfg(feature = "async")]
-async fn run(
-    tm: &mut textmode::Output,
-    input: &mut textmode::Input,
-) -> textmode::Result<()> {
+#[tokio::main]
+async fn main() -> textmode::Result<()> {
+    let mut input = textmode::Input::new().await?;
+    let mut tm = textmode::Output::new().await?;
     tm.move_to(5, 5);
     tm.write_str("foo");
     input.read_key().await?;
@@ -23,16 +23,6 @@ async fn run(
     Ok(())
 }
 
-#[cfg(feature = "async")]
-fn main() {
-    smol::block_on(async {
-        let mut input = textmode::Input::new().await.unwrap();
-        let mut tm = textmode::Output::new().await.unwrap();
-        let e = run(&mut tm, &mut input).await;
-        e.unwrap();
-    });
-}
-
 #[cfg(not(feature = "async"))]
 fn main() {
     let mut tm = textmode::blocking::Output::new().unwrap();
diff --git a/examples/input.rs b/examples/input.rs
index a3b1f1d..49a7c55 100644
--- a/examples/input.rs
+++ b/examples/input.rs
@@ -1,5 +1,6 @@
 #[cfg(feature = "async")]
-async fn async_main() {
+#[tokio::main]
+async fn main() {
     let mut input = textmode::Input::new().await.unwrap();
     for arg in std::env::args().skip(1) {
         match arg.as_str() {
@@ -27,11 +28,6 @@ async fn async_main() {
     }
 }
 
-#[cfg(feature = "async")]
-fn main() {
-    smol::block_on(async { async_main().await })
-}
-
 #[cfg(not(feature = "async"))]
 fn main() {
     let mut input = textmode::blocking::Input::new().unwrap();
diff --git a/examples/tmux.rs b/examples/tmux.rs
index 5eebee4..46a4926 100644
--- a/examples/tmux.rs
+++ b/examples/tmux.rs
@@ -2,15 +2,10 @@
 mod tmux_impl;
 
 #[cfg(feature = "async")]
-async fn async_main(ex: &smol::Executor<'_>) {
+#[tokio::main]
+async fn main() {
     let tmux = tmux_impl::Tmux::new().await;
-    tmux.run(ex).await;
-}
-
-#[cfg(feature = "async")]
-fn main() {
-    let ex = smol::Executor::new();
-    smol::block_on(async { async_main(&ex).await })
+    tmux.run().await;
 }
 
 #[cfg(not(feature = "async"))]
diff --git a/examples/tmux_impl/mod.rs b/examples/tmux_impl/mod.rs
index de026e4..313aaf2 100644
--- a/examples/tmux_impl/mod.rs
+++ b/examples/tmux_impl/mod.rs
@@ -1,12 +1,13 @@
-use pty_process::Command as _;
-use smol::io::{AsyncReadExt as _, AsyncWriteExt as _};
 use textmode::Textmode as _;
+use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
 
+#[derive(Debug)]
 enum Command {
     NewWindow,
     NextWindow,
 }
 
+#[derive(Debug)]
 enum Event {
     Input(textmode::Key),
     Output,
@@ -16,8 +17,8 @@ enum Event {
 }
 
 struct Window {
-    child: std::sync::Arc<pty_process::smol::Child>,
-    vt: std::sync::Arc<smol::lock::Mutex<vt100::Parser>>,
+    vt: std::sync::Arc<tokio::sync::Mutex<vt100::Parser>>,
+    pty_w: pty_process::OwnedWritePty,
     screen: vt100::Screen,
 }
 
@@ -33,13 +34,13 @@ struct State {
     next_window_id: usize,
     notifications: std::collections::BTreeMap<usize, Notification>,
     next_notification_id: usize,
-    wevents: smol::channel::Sender<Event>,
-    revents: smol::channel::Receiver<Event>,
+    wevents: tokio::sync::mpsc::UnboundedSender<Event>,
+    revents: tokio::sync::mpsc::UnboundedReceiver<Event>,
 }
 
 impl State {
     fn new() -> Self {
-        let (sender, receiver) = smol::channel::unbounded();
+        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
         Self {
             windows: std::collections::BTreeMap::new(),
             current_window: 0,
@@ -59,7 +60,7 @@ impl State {
         self.windows.get_mut(&self.current_window).unwrap()
     }
 
-    fn next_window(&mut self, ex: &smol::Executor<'_>) {
+    fn next_window(&mut self) {
         self.current_window = self
             .windows
             .keys()
@@ -68,13 +69,10 @@ impl State {
             .skip_while(|&id| id < self.current_window)
             .nth(1)
             .unwrap();
-        self.notify(
-            ex,
-            &format!("switched to window {}", self.current_window),
-        );
+        self.notify(&format!("switched to window {}", self.current_window));
     }
 
-    fn notify(&mut self, ex: &smol::Executor<'_>, text: &str) {
+    fn notify(&mut self, text: &str) {
         let now = std::time::Instant::now();
         let expiry = now + std::time::Duration::from_secs(5);
         let text = text.to_string();
@@ -83,43 +81,35 @@ impl State {
         self.next_notification_id += 1;
         self.notifications.insert(id, notification);
         let notify = self.wevents.clone();
-        ex.spawn(async move {
-            smol::Timer::at(expiry).await;
-            notify.send(Event::Notification).await.unwrap();
-        })
-        .detach();
+        tokio::task::spawn(async move {
+            tokio::time::sleep_until(tokio::time::Instant::from_std(expiry))
+                .await;
+            notify.send(Event::Notification).unwrap();
+        });
     }
 
-    fn spawn_input_task(
-        &self,
-        ex: &smol::Executor<'_>,
-        mut input: textmode::Input,
-    ) {
+    fn spawn_input_thread(&self, mut input: textmode::blocking::Input) {
         let notify = self.wevents.clone();
-        ex.spawn(async move {
+        std::thread::spawn(move || {
             let mut waiting_for_command = false;
             input.parse_utf8(false);
             input.parse_meta(false);
             input.parse_special_keys(false);
             loop {
                 input.parse_single(waiting_for_command);
-                match input.read_key().await {
+                match input.read_key() {
                     Ok(Some(key)) => {
                         if waiting_for_command {
                             waiting_for_command = false;
                             match key {
                                 textmode::Key::Ctrl(b'n') => {
-                                    notify
-                                        .send(Event::Input(key))
-                                        .await
-                                        .unwrap();
+                                    notify.send(Event::Input(key)).unwrap();
                                 }
                                 textmode::Key::Byte(b'c') => {
                                     notify
                                         .send(Event::Command(
                                             Command::NewWindow,
                                         ))
-                                        .await
                                         .unwrap();
                                 }
                                 textmode::Key::Byte(b'n') => {
@@ -127,7 +117,6 @@ impl State {
                                         .send(Event::Command(
                                             Command::NextWindow,
                                         ))
-                                        .await
                                         .unwrap();
                                 }
                                 _ => {} // ignore
@@ -138,10 +127,7 @@ impl State {
                                     waiting_for_command = true;
                                 }
                                 _ => {
-                                    notify
-                                        .send(Event::Input(key))
-                                        .await
-                                        .unwrap();
+                                    notify.send(Event::Input(key)).unwrap();
                                 }
                             }
                         }
@@ -155,59 +141,67 @@ impl State {
                     }
                 }
             }
-        })
-        .detach();
+        });
     }
 
     fn new_window(
         &mut self,
-        ex: &smol::Executor<'_>,
-        notify: smol::channel::Sender<Event>,
+        notify: tokio::sync::mpsc::UnboundedSender<Event>,
     ) {
-        let child = smol::process::Command::new("zsh")
-            .spawn_pty(Some(&pty_process::Size::new(24, 80)))
-            .unwrap();
-        let child = std::sync::Arc::new(child);
+        let pty = pty_process::Pty::new().unwrap();
+        let pts = pty.pts().unwrap();
+        pty.resize(pty_process::Size::new(24, 80)).unwrap();
+        let mut cmd = pty_process::Command::new("zsh");
+        let mut child = cmd.spawn(&pts).unwrap();
+        let (mut pty_r, pty_w) = pty.into_split();
         let vt = vt100::Parser::default();
         let screen = vt.screen().clone();
-        let vt = std::sync::Arc::new(smol::lock::Mutex::new(vt));
+        let vt = std::sync::Arc::new(tokio::sync::Mutex::new(vt));
         let id = self.next_window_id;
         self.next_window_id += 1;
         let window = Window {
-            child: child.clone(),
+            pty_w,
             vt: vt.clone(),
             screen,
         };
         self.windows.insert(id, window);
         self.current_window = id;
-        self.notify(ex, &format!("created window {}", id));
-        ex.spawn(async move {
+        self.notify(&format!("created window {}", id));
+        tokio::task::spawn(async move {
+            let _pts = pts;
             let mut buf = [0_u8; 4096];
             loop {
-                match child.pty().read(&mut buf).await {
-                    Ok(bytes) => {
-                        vt.lock_arc().await.process(&buf[..bytes]);
-                        notify.send(Event::Output).await.unwrap();
-                    }
-                    Err(e) => {
-                        // EIO means that the process closed the other
-                        // end of the pty
-                        if e.raw_os_error() != Some(libc::EIO) {
+                tokio::select! {
+                    bytes = pty_r.read(&mut buf) => match bytes {
+                        Ok(bytes) => {
+                            if bytes == 0 {
+                                continue;
+                            }
+                            vt.clone()
+                                .lock_owned()
+                                .await
+                                .process(&buf[..bytes]);
+                            notify.send(Event::Output).unwrap();
+                        }
+                        Err(e) => {
                             eprintln!("pty read failed: {:?}", e);
+                            break;
                         }
-                        notify.send(Event::WindowExit(id)).await.unwrap();
+                    },
+                    _ = child.wait() => {
+                        notify.send(Event::WindowExit(id)).unwrap();
                         break;
-                    }
+                    },
                 }
             }
-        })
-        .detach();
+        });
     }
 
     async fn redraw_current_window(&mut self, tm: &mut textmode::Output) {
         let window = self.current_window();
         tm.clear();
-        let new_screen = window.vt.lock_arc().await.screen().clone();
+        let new_screen =
+            window.vt.clone().lock_owned().await.screen().clone();
         tm.write(&new_screen.state_formatted());
         self.draw_notifications(tm, &new_screen);
         tm.refresh().await.unwrap();
@@ -216,7 +210,8 @@ impl State {
     async fn update_current_window(&mut self, tm: &mut textmode::Output) {
         let window = self.current_window();
         let old_screen = window.screen.clone();
-        let new_screen = window.vt.lock_arc().await.screen().clone();
+        let new_screen =
+            window.vt.clone().lock_owned().await.screen().clone();
         let diff = new_screen.state_diff(&old_screen);
         self.clear_notifications(tm, &old_screen);
         tm.write(&diff);
@@ -292,88 +287,75 @@ impl State {
 
 #[must_use]
 pub struct Tmux {
-    input: textmode::Input,
+    input: textmode::blocking::Input,
     tm: textmode::Output,
     state: State,
 }
 
 impl Tmux {
     pub async fn new() -> Self {
-        let input = textmode::Input::new().await.unwrap();
+        let input = textmode::blocking::Input::new().unwrap();
         let tm = textmode::Output::new().await.unwrap();
         let state = State::new();
         Self { input, tm, state }
     }
 
-    pub async fn run(self, ex: &smol::Executor<'_>) {
+    pub async fn run(self) {
         let Self {
-            input,
+            mut input,
             mut tm,
             mut state,
         } = self;
 
-        state.spawn_input_task(ex, input);
+        let _raw_guard = input.take_raw_guard();
+        state.spawn_input_thread(input);
 
-        ex.run(async {
-            state.new_window(ex, state.wevents.clone());
+        state.new_window(state.wevents.clone());
 
-            loop {
-                match state.revents.recv().await {
-                    Ok(Event::Output) => {
-                        state.update_current_window(&mut tm).await;
+        loop {
+            match state.revents.recv().await {
+                Some(Event::Output) => {
+                    state.update_current_window(&mut tm).await;
+                }
+                Some(Event::Input(key)) => {
+                    state
+                        .current_window_mut()
+                        .pty_w
+                        .write_all(&key.into_bytes())
+                        .await
+                        .unwrap();
+                }
+                Some(Event::WindowExit(id)) => {
+                    // do this first because next_window breaks if
+                    // current_window is greater than all existing windows
+                    if state.current_window == id {
+                        state.next_window()
                     }
-                    Ok(Event::Input(key)) => {
-                        state
-                            .current_window()
-                            .child
-                            .pty()
-                            .write_all(&key.into_bytes())
-                            .await
-                            .unwrap();
+                    state.windows.remove(&id).unwrap();
+                    if state.windows.is_empty() {
+                        break;
                     }
-                    Ok(Event::WindowExit(id)) => {
-                        // do this first because next_window breaks if
-                        // current_window is greater than all existing windows
-                        if state.current_window == id {
-                            state.next_window(ex)
-                        }
-                        let mut dropped_window =
-                            state.windows.remove(&id).unwrap();
-                        // i can get_mut because at this point the future
-                        // holding the other copy of child has already been
-                        // dropped
-                        std::sync::Arc::get_mut(&mut dropped_window.child)
-                            .unwrap()
-                            .status()
-                            .await
-                            .unwrap();
-                        if state.windows.is_empty() {
-                            break;
-                        }
-                        state.notify(ex, &format!("window {} exited", id));
+                    state.notify(&format!("window {} exited", id));
 
+                    state.redraw_current_window(&mut tm).await;
+                }
+                Some(Event::Command(c)) => match c {
+                    Command::NewWindow => {
+                        state.new_window(state.wevents.clone());
                         state.redraw_current_window(&mut tm).await;
                     }
-                    Ok(Event::Command(c)) => match c {
-                        Command::NewWindow => {
-                            state.new_window(ex, state.wevents.clone());
-                            state.redraw_current_window(&mut tm).await;
-                        }
-                        Command::NextWindow => {
-                            state.next_window(ex);
-                            state.redraw_current_window(&mut tm).await;
-                        }
-                    },
-                    Ok(Event::Notification) => {
-                        state.update_current_window(&mut tm).await;
-                    }
-                    Err(e) => {
-                        eprintln!("{}", e);
-                        break;
+                    Command::NextWindow => {
+                        state.next_window();
+                        state.redraw_current_window(&mut tm).await;
                     }
+                },
+                Some(Event::Notification) => {
+                    state.update_current_window(&mut tm).await;
+                }
+                None => {
+                    break;
                 }
             }
-        })
-        .await;
+        }
     }
 }
diff --git a/src/blocking/input.rs b/src/blocking/input.rs
index 48d9ee8..63c9ac5 100644
--- a/src/blocking/input.rs
+++ b/src/blocking/input.rs
@@ -36,7 +36,8 @@ impl RawGuard {
     /// Switch back from raw mode early.
     ///
     /// # Errors
-    /// * `Error::SetTerminalMode`: failed to return the terminal from raw mode
+    /// * `Error::SetTerminalMode`: failed to return the terminal from raw
+    /// mode
     pub fn cleanup(&mut self) -> crate::error::Result<()> {
         self.termios.take().map_or(Ok(()), |termios| {
             let stdin = std::io::stdin().as_raw_fd();
diff --git a/src/input.rs b/src/input.rs
index 90213ec..7f434f5 100644
--- a/src/input.rs
+++ b/src/input.rs
@@ -1,5 +1,5 @@
-use futures_lite::io::AsyncReadExt as _;
 use std::os::unix::io::AsRawFd as _;
+use tokio::io::AsyncReadExt as _;
 
 use crate::private::Input as _;
 
@@ -16,16 +16,20 @@ impl RawGuard {
     ///
     /// # Errors
     /// * `Error::SetTerminalMode`: failed to put the terminal into raw mode
+    // spawn_blocking is uncancellable, and the tcgetattr/tcsetattr calls
+    // can't panic, so unwrap is safe here
+    #[allow(clippy::missing_panics_doc)]
     pub async fn new() -> crate::error::Result<Self> {
         let stdin = std::io::stdin().as_raw_fd();
-        let termios = blocking::unblock(move || {
+        let termios = tokio::task::spawn_blocking(move || {
             nix::sys::termios::tcgetattr(stdin)
                 .map_err(crate::error::Error::SetTerminalMode)
         })
-        .await?;
+        .await
+        .unwrap()?;
         let mut termios_raw = termios.clone();
         nix::sys::termios::cfmakeraw(&mut termios_raw);
-        blocking::unblock(move || {
+        tokio::task::spawn_blocking(move || {
             nix::sys::termios::tcsetattr(
                 stdin,
                 nix::sys::termios::SetArg::TCSANOW,
@@ -33,7 +37,8 @@ impl RawGuard {
             )
             .map_err(crate::error::Error::SetTerminalMode)
         })
-        .await?;
+        .await
+        .unwrap()?;
         Ok(Self {
             termios: Some(termios),
         })
@@ -42,11 +47,15 @@ impl RawGuard {
     /// Switch back from raw mode early.
     ///
     /// # Errors
-    /// * `Error::SetTerminalMode`: failed to return the terminal from raw mode
+    /// * `Error::SetTerminalMode`: failed to return the terminal from raw
+    /// mode
+    // spawn_blocking is uncancellable, and the tcsetattr call can't panic, so
+    // unwrap is safe here
+    #[allow(clippy::missing_panics_doc)]
     pub async fn cleanup(&mut self) -> crate::error::Result<()> {
         if let Some(termios) = self.termios.take() {
             let stdin = std::io::stdin().as_raw_fd();
-            blocking::unblock(move || {
+            tokio::task::spawn_blocking(move || {
                 nix::sys::termios::tcsetattr(
                     stdin,
                     nix::sys::termios::SetArg::TCSANOW,
@@ -55,6 +64,7 @@ impl RawGuard {
                 .map_err(crate::error::Error::SetTerminalMode)
             })
             .await
+            .unwrap()
         } else {
             Ok(())
         }
@@ -66,11 +76,19 @@ impl Drop for RawGuard {
     /// of an async drop mechanism. If this could be a problem, you should
     /// call `cleanup` manually instead.
     fn drop(&mut self) {
-        futures_lite::future::block_on(async {
-            // https://github.com/rust-lang/rust-clippy/issues/8003
-            #[allow(clippy::let_underscore_drop)]
-            let _ = self.cleanup().await;
-        });
+        // doesn't literally call `cleanup`, because calling spawn_blocking
+        // while the tokio runtime is in the process of shutting down doesn't
+        // work (spawn_blocking tasks are cancelled if the runtime starts
+        // shutting down before the task body starts running), but should be
+        // kept in sync with the actual things that `cleanup` does.
+        if let Some(termios) = self.termios.take() {
+            let stdin = std::io::stdin().as_raw_fd();
+            let _ = nix::sys::termios::tcsetattr(
+                stdin,
+                nix::sys::termios::SetArg::TCSANOW,
+                &termios,
+            );
+        }
     }
 }
 
@@ -80,8 +98,16 @@ impl Drop for RawGuard {
 /// additionally configure the types of keypresses you are interested in
 /// through the `parse_*` methods. This configuration can be changed between
 /// any two calls to [`read_key`](Input::read_key).
+///
+/// # Note
+///
+/// This is built on [`tokio::io::Stdin`], and inherits its caveats. In
+/// particular, it will likely cause a hang until one more newline is received
+/// when the tokio runtime shuts down. Because of this, it is generally
+/// recommended to spawn a thread and use
+/// [`textmode::blocking::Input`](crate::blocking::Input) instead.
 pub struct Input {
-    stdin: blocking::Unblock<std::io::Stdin>,
+    stdin: tokio::io::Stdin,
     raw: Option<RawGuard>,
 
     buf: Vec<u8>,
@@ -161,7 +187,7 @@ impl Input {
     #[must_use]
     pub fn new_without_raw() -> Self {
         Self {
-            stdin: blocking::Unblock::new(std::io::stdin()),
+            stdin: tokio::io::stdin(),
             raw: None,
             buf: Vec::with_capacity(4096),
             pos: 0,
@@ -290,7 +316,7 @@ impl Input {
 }
 
 async fn read_stdin(
-    stdin: &mut blocking::Unblock<std::io::Stdin>,
+    stdin: &mut tokio::io::Stdin,
     buf: &mut [u8],
 ) -> crate::error::Result<usize> {
     stdin
diff --git a/src/lib.rs b/src/lib.rs
index 7e89eed..3bfb9b1 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -8,11 +8,8 @@
 //! ```no_run
 //! use textmode::Textmode;
 //! # #[cfg(feature = "async")]
+//! # #[tokio::main]
 //! # fn main() -> textmode::Result<()> {
-//! #     futures_lite::future::block_on(async { run().await })
-//! # }
-//! # #[cfg(feature = "async")]
-//! # async fn run() -> textmode::Result<()> {
 //! let mut tm = textmode::Output::new().await?;
 //! tm.clear();
 //! tm.move_to(5, 5);
diff --git a/src/output.rs b/src/output.rs
index 9b72438..378b7d4 100644
--- a/src/output.rs
+++ b/src/output.rs
@@ -1,4 +1,4 @@
-use futures_lite::io::AsyncWriteExt as _;
+use tokio::io::AsyncWriteExt as _;
 
 use crate::private::Output as _;
 
@@ -16,11 +16,7 @@ impl ScreenGuard {
     /// # Errors
     /// * `Error::WriteStdout`: failed to write initialization to stdout
     pub async fn new() -> crate::error::Result<Self> {
-        write_stdout(
-            &mut blocking::Unblock::new(std::io::stdout()),
-            crate::INIT,
-        )
-        .await?;
+        write_stdout(&mut tokio::io::stdout(), crate::INIT).await?;
         Ok(Self { cleaned_up: false })
     }
 
@@ -33,11 +29,7 @@ impl ScreenGuard {
             return Ok(());
         }
         self.cleaned_up = true;
-        write_stdout(
-            &mut blocking::Unblock::new(std::io::stdout()),
-            crate::DEINIT,
-        )
-        .await
+        write_stdout(&mut tokio::io::stdout(), crate::DEINIT).await
     }
 }
 
@@ -46,10 +38,12 @@ impl Drop for ScreenGuard {
     /// of an async drop mechanism. If this could be a problem, you should
     /// call `cleanup` manually instead.
     fn drop(&mut self) {
-        futures_lite::future::block_on(async {
-            // https://github.com/rust-lang/rust-clippy/issues/8003
-            #[allow(clippy::let_underscore_drop)]
-            let _ = self.cleanup().await;
+        tokio::task::block_in_place(move || {
+            tokio::runtime::Handle::current().block_on(async {
+                // https://github.com/rust-lang/rust-clippy/issues/8003
+                #[allow(clippy::let_underscore_drop)]
+                let _ = self.cleanup().await;
+            });
         });
     }
 }
@@ -61,7 +55,7 @@ impl Drop for ScreenGuard {
 /// then call [`refresh`](Output::refresh) when you want to update the
 /// terminal on `stdout`.
 pub struct Output {
-    stdout: blocking::Unblock<std::io::Stdout>,
+    stdout: tokio::io::Stdout,
     screen: Option<ScreenGuard>,
 
     cur: vt100::Parser,
@@ -112,8 +106,9 @@ impl Output {
         };
         let cur = vt100::Parser::new(rows, cols, 0);
         let next = vt100::Parser::new(rows, cols, 0);
+
         Self {
-            stdout: blocking::Unblock::new(std::io::stdout()),
+            stdout: tokio::io::stdout(),
             screen: None,
             cur,
             next,
@@ -158,7 +153,7 @@ impl Output {
 }
 
 async fn write_stdout(
-    stdout: &mut blocking::Unblock<std::io::Stdout>,
+    stdout: &mut tokio::io::Stdout,
     buf: &[u8],
 ) -> crate::error::Result<()> {
     stdout
diff --git a/tests/fixtures/bin/Cargo.toml b/tests/fixtures/bin/Cargo.toml
index cac6fe1..4696e6d 100644
--- a/tests/fixtures/bin/Cargo.toml
+++ b/tests/fixtures/bin/Cargo.toml
@@ -6,10 +6,9 @@ edition = "2018"
 
 [dependencies]
 textmode = { path = "../../.." }
-vt100 = "*"
 
-smol = { version = "*", optional = true }
+tokio = { version = "*", features = ["full"], optional = true }
 
 [features]
 default = []
-async = ["textmode/async", "smol"]
+async = ["textmode/async", "tokio"]
diff --git a/tests/fixtures/mod.rs b/tests/fixtures/mod.rs
index 08f09c1..5a65954 100644
--- a/tests/fixtures/mod.rs
+++ b/tests/fixtures/mod.rs
@@ -1,4 +1,3 @@
-use pty_process::Command as _;
 use std::io::{BufRead as _, Read as _};
 use std::os::unix::io::AsRawFd as _;
 
@@ -62,33 +61,34 @@ pub struct BuiltFixture {
 }
 
 impl BuiltFixture {
-    pub fn run<F: FnOnce(&mut std::fs::File)>(
+    pub fn run<F: FnOnce(&mut pty_process::blocking::Pty)>(
         &mut self,
         args: &[&str],
         f: F,
     ) {
-        let mut cmd = self.run.command();
-        let mut child = cmd
-            .args(args)
-            .spawn_pty(Some(&pty_process::Size::new(24, 80)))
-            .unwrap();
+        let mut pty = pty_process::blocking::Pty::new().unwrap();
+        let pts = pty.pts().unwrap();
+        pty.resize(pty_process::Size::new(24, 80)).unwrap();
+        let mut cmd = pty_process::blocking::Command::new(self.run.path());
+        cmd.args(args);
+        let mut child = cmd.spawn(&pts).unwrap();
 
         if self.screenguard {
-            assert!(read_ready(child.pty().as_raw_fd()));
+            assert!(read_ready(pty.as_raw_fd()));
             let mut buf = vec![0u8; 1024];
-            let bytes = child.pty().read(&mut buf).unwrap();
+            let bytes = pty.read(&mut buf).unwrap();
             buf.truncate(bytes);
             assert_eq!(&buf[..], b"\x1b7\x1b[?47h\x1b[2J\x1b[H\x1b[?25h");
         } else {
             std::thread::sleep(std::time::Duration::from_millis(100));
         }
 
-        f(child.pty_mut());
+        f(&mut pty);
 
         if self.screenguard {
-            assert!(read_ready(child.pty().as_raw_fd()));
+            assert!(read_ready(pty.as_raw_fd()));
             let mut buf = vec![0u8; 1024];
-            let bytes = child.pty().read(&mut buf).unwrap();
+            let bytes = pty.read(&mut buf).unwrap();
             buf.truncate(bytes);
             assert_eq!(&buf[..], b"\x1b[?47l\x1b8\x1b[?25h");
         }
@@ -100,7 +100,7 @@ impl BuiltFixture {
 
 #[allow(dead_code)]
 #[track_caller]
-pub fn read(f: &mut std::fs::File) -> Vec<u8> {
+pub fn read(f: &mut pty_process::blocking::Pty) -> Vec<u8> {
     assert!(read_ready(f.as_raw_fd()));
     let mut buf = vec![0u8; 1024];
     let bytes = f.read(&mut buf).unwrap();
@@ -110,7 +110,9 @@ pub fn read(f: &mut std::fs::File) -> Vec<u8> {
 
 #[allow(dead_code)]
 #[track_caller]
-pub fn read_line(f: &mut std::io::BufReader<&mut std::fs::File>) -> Vec<u8> {
+pub fn read_line(
+    f: &mut std::io::BufReader<&mut pty_process::blocking::Pty>,
+) -> Vec<u8> {
     assert!(!f.buffer().is_empty() || read_ready(f.get_ref().as_raw_fd()));
     let mut buf = vec![];
     f.read_until(b'\n', &mut buf).unwrap();
diff --git a/tests/input.rs b/tests/input.rs
index c341717..5b23b6a 100644
--- a/tests/input.rs
+++ b/tests/input.rs
@@ -312,25 +312,29 @@ fn run_input_test(
 }
 
 #[track_caller]
-fn write(f: &mut std::fs::File, key: textmode::Key) {
+fn write(f: &mut pty_process::blocking::Pty, key: textmode::Key) {
     f.write_all(&key.into_bytes()).unwrap();
 }
 
 #[track_caller]
-fn read(f: &mut std::io::BufReader<&mut std::fs::File>) -> String {
+fn read(
+    f: &mut std::io::BufReader<&mut pty_process::blocking::Pty>,
+) -> String {
     std::string::String::from_utf8(fixtures::read_line(f)).unwrap()
 }
 
 #[track_caller]
 fn assert_line(
-    f: &mut std::io::BufReader<&mut std::fs::File>,
+    f: &mut std::io::BufReader<&mut pty_process::blocking::Pty>,
     expected: &str,
 ) {
     assert_eq!(read(f), format!("{}\r\n", expected));
 }
 
 #[track_caller]
-fn assert_no_more_lines(f: &mut std::io::BufReader<&mut std::fs::File>) {
+fn assert_no_more_lines(
+    f: &mut std::io::BufReader<&mut pty_process::blocking::Pty>,
+) {
     if fixtures::read_ready(f.get_ref().as_raw_fd()) || !f.buffer().is_empty()
     {
         use std::io::Read as _;