perf(tui): bring async to tui (#9132)

### Description

This PR makes the driving the TUI async, the actual operation on the
state is still completely sync. The primary driver behind this PR is it
allows us to make use of
[`crossterm::event::EventStream`](https://docs.rs/crossterm/latest/crossterm/event/struct.EventStream.html)
which seems to be far more performant than polling for user input.

The first commit of the PR just changes types from `std::sync::mpsc` to
`tokio::sync::mpsc` (and make use of `tokio::sync::oneshot` for our
callbacks instead of a channel of size 1).

The second commit removes our usage of `crossterm::event::poll` in favor
of a dedicated task that reads and forwards events from
`crossterm::event::EventStream`.

The final commit moves the production of ticks to it's own task to avoid
the need for timing out our reads.

### Testing Instructions

Notice large reduction in CPU usage from `turbo` when tasks are not
producing output and the TUI is just waiting for user input.

Before
<img width="331" alt="Screenshot 2024-09-10 at 2 57 45 PM"
src="https://github.com/user-attachments/assets/ad85fa7a-7b55-4459-a6c0-c0a7931e8738">

After
<img width="393" alt="Screenshot 2024-09-10 at 2 56 17 PM"
src="https://github.com/user-attachments/assets/9b6793b2-bca6-4baa-ab4d-3182e561212e">

crates/turborepo-lib/src/commands/run.rs

@@ -57,7 +57,7 @@ pub async fn run(base: CommandBase, telemetry: CommandEventBuilder) -> Result<i3
 
         // We only stop if it's the TUI, for the web UI we don't need to stop
         if let Some(UISender::Tui(sender)) = sender {
-            sender.stop();
+            sender.stop().await;
         }
 
         if let Some(handle) = handle {

crates/turborepo-lib/src/run/mod.rs

@@ -253,7 +253,8 @@ impl Run {
         }
 
         let (sender, receiver) = TuiSender::new();
-        let handle = tokio::task::spawn_blocking(move || Ok(tui::run_app(task_names, receiver)?));
+        let handle =
+            tokio::task::spawn(async move { Ok(tui::run_app(task_names, receiver).await?) });
 
         Ok(Some((sender, handle)))
     }
@@ -454,7 +455,8 @@ impl Run {
             global_env,
             ui_sender,
             is_watch,
-        );
+        )
+        .await;
 
         if self.opts.run_opts.dry_run.is_some() {
             visitor.dry_run();

crates/turborepo-lib/src/task_graph/visitor.rs

@@ -106,7 +106,7 @@ impl<'a> Visitor<'a> {
     // Once we have the full picture we will go about grouping these pieces of data
     // together
     #[allow(clippy::too_many_arguments)]
-    pub fn new(
+    pub async fn new(
         package_graph: Arc<PackageGraph>,
         run_cache: Arc<RunCache>,
         run_tracker: RunTracker,
@@ -133,8 +133,11 @@ impl<'a> Visitor<'a> {
         let sink = Self::sink(run_opts);
         let color_cache = ColorSelector::default();
         // Set up correct size for underlying pty
-        if let Some(pane_size) = ui_sender.as_ref().and_then(|sender| sender.pane_size()) {
-            manager.set_pty_size(pane_size.rows, pane_size.cols);
+
+        if let Some(app) = ui_sender.as_ref() {
+            if let Some(pane_size) = app.pane_size().await {
+                manager.set_pty_size(pane_size.rows, pane_size.cols);
+            }
         }
 
         Self {
@@ -330,7 +333,7 @@ impl<'a> Visitor<'a> {
 
         if !self.is_watch {
             if let Some(handle) = &self.ui_sender {
-                handle.stop();
+                handle.stop().await;
             }
         }
 

crates/turborepo-ui/Cargo.toml

@@ -24,8 +24,9 @@ axum-server = { workspace = true }
 base64 = "0.22"
 chrono = { workspace = true }
 console = { workspace = true }
-crossterm = "0.27.0"
+crossterm = { version = "0.27.0", features = ["event-stream"] }
 dialoguer = { workspace = true }
+futures = { workspace = true }
 indicatif = { workspace = true }
 lazy_static = { workspace = true }
 nix = { version = "0.26.2", features = ["signal"] }
@@ -34,7 +35,6 @@ serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
-
 tracing = { workspace = true }
 tui-term = { workspace = true }
 turbopath = { workspace = true }

crates/turborepo-ui/src/sender.rs

@@ -62,9 +62,9 @@ impl UISender {
             UISender::Wui(sender) => sender.task(task),
         }
     }
-    pub fn stop(&self) {
+    pub async fn stop(&self) {
         match self {
-            UISender::Tui(sender) => sender.stop(),
+            UISender::Tui(sender) => sender.stop().await,
             UISender::Wui(sender) => sender.stop(),
         }
     }
@@ -75,9 +75,9 @@ impl UISender {
         }
     }
 
-    pub fn pane_size(&self) -> Option<PaneSize> {
+    pub async fn pane_size(&self) -> Option<PaneSize> {
         match self {
-            UISender::Tui(sender) => sender.pane_size(),
+            UISender::Tui(sender) => sender.pane_size().await,
             // Not applicable to the web UI
             UISender::Wui(_) => None,
         }

crates/turborepo-ui/src/tui/app.rs

@@ -2,8 +2,7 @@ use std::{
     collections::BTreeMap,
     io::{self, Stdout, Write},
     mem,
-    sync::mpsc,
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use ratatui::{
@@ -12,9 +11,13 @@ use ratatui::{
     widgets::TableState,
     Frame, Terminal,
 };
+use tokio::{
+    sync::{mpsc, oneshot},
+    time::Instant,
+};
 use tracing::{debug, trace};
 
-const FRAMERATE: Duration = Duration::from_millis(3);
+pub const FRAMERATE: Duration = Duration::from_millis(3);
 const RESIZE_DEBOUNCE_DELAY: Duration = Duration::from_millis(10);
 
 use super::{
@@ -43,7 +46,6 @@ pub struct App<W> {
     tasks: BTreeMap<String, TerminalOutput<W>>,
     tasks_by_status: TasksByStatus,
     focus: LayoutSections,
-    tty_stdin: bool,
     scroll: TableState,
     selected_task_index: usize,
     has_user_scrolled: bool,
@@ -78,8 +80,6 @@ impl<W> App<W> {
             size,
             done: false,
             focus: LayoutSections::TaskList,
-            // Check if stdin is a tty that we should read input from
-            tty_stdin: atty::is(atty::Stream::Stdin),
             tasks: tasks_by_status
                 .task_names_in_displayed_order()
                 .map(|task_name| {
@@ -112,7 +112,6 @@ impl<W> App<W> {
         let has_selection = self.get_full_task()?.has_selection();
         Ok(InputOptions {
             focus: &self.focus,
-            tty_stdin: self.tty_stdin,
             has_selection,
         })
     }
@@ -558,16 +557,19 @@ impl<W: Write> App<W> {
 
 /// Handle the rendering of the `App` widget based on events received by
 /// `receiver`
-pub fn run_app(tasks: Vec<String>, receiver: AppReceiver) -> Result<(), Error> {
+pub async fn run_app(tasks: Vec<String>, receiver: AppReceiver) -> Result<(), Error> {
     let mut terminal = startup()?;
     let size = terminal.size()?;
 
     let mut app: App<Box<dyn io::Write + Send>> = App::new(size.height, size.width, tasks);
+    let (crossterm_tx, crossterm_rx) = mpsc::channel(1024);
+    input::start_crossterm_stream(crossterm_tx);
 
-    let (result, callback) = match run_app_inner(&mut terminal, &mut app, receiver) {
-        Ok(callback) => (Ok(()), callback),
-        Err(err) => (Err(err), None),
-    };
+    let (result, callback) =
+        match run_app_inner(&mut terminal, &mut app, receiver, crossterm_rx).await {
+            Ok(callback) => (Ok(()), callback),
+            Err(err) => (Err(err), None),
+        };
 
     cleanup(terminal, app, callback)?;
 
@@ -576,18 +578,19 @@ pub fn run_app(tasks: Vec<String>, receiver: AppReceiver) -> Result<(), Error> {
 
 // Break out inner loop so we can use `?` without worrying about cleaning up the
 // terminal.
-fn run_app_inner<B: Backend + std::io::Write>(
+async fn run_app_inner<B: Backend + std::io::Write>(
     terminal: &mut Terminal<B>,
     app: &mut App<Box<dyn io::Write + Send>>,
-    receiver: AppReceiver,
-) -> Result<Option<mpsc::SyncSender<()>>, Error> {
+    mut receiver: AppReceiver,
+    mut crossterm_rx: mpsc::Receiver<crossterm::event::Event>,
+) -> Result<Option<oneshot::Sender<()>>, Error> {
     // Render initial state to paint the screen
     terminal.draw(|f| view(app, f))?;
     let mut last_render = Instant::now();
     let mut resize_debouncer = Debouncer::new(RESIZE_DEBOUNCE_DELAY);
     let mut callback = None;
     let mut needs_rerender = true;
-    while let Some(event) = poll(app.input_options()?, &receiver, last_render + FRAMERATE) {
+    while let Some(event) = poll(app.input_options()?, &mut receiver, &mut crossterm_rx).await {
         // If we only receive ticks, then there's been no state change so no update
         // needed
         if !matches!(event, Event::Tick) {
@@ -625,13 +628,31 @@ fn run_app_inner<B: Backend + std::io::Write>(
 
 /// Blocking poll for events, will only return None if app handle has been
 /// dropped
-fn poll(input_options: InputOptions, receiver: &AppReceiver, deadline: Instant) -> Option<Event> {
-    match input(input_options) {
-        Ok(Some(event)) => Some(event),
-        Ok(None) => receiver.recv(deadline).ok(),
-        // Unable to read from stdin, shut down and attempt to clean up
-        Err(_) => Some(Event::InternalStop),
-    }
+async fn poll<'a>(
+    input_options: InputOptions<'a>,
+    receiver: &mut AppReceiver,
+    crossterm_rx: &mut mpsc::Receiver<crossterm::event::Event>,
+) -> Option<Event> {
+    let input_closed = crossterm_rx.is_closed();
+    let input_fut = async {
+        crossterm_rx
+            .recv()
+            .await
+            .and_then(|event| input_options.handle_crossterm_event(event))
+    };
+    let receiver_fut = async { receiver.recv().await };
+    let event_fut = async move {
+        if input_closed {
+            receiver_fut.await
+        } else {
+            tokio::select! {
+                e = input_fut => e,
+                e = receiver_fut => e,
+            }
+        }
+    };
+
+    event_fut.await
 }
 
 const MIN_HEIGHT: u16 = 10;
@@ -672,7 +693,7 @@ fn startup() -> io::Result<Terminal<CrosstermBackend<Stdout>>> {
 fn cleanup<B: Backend + io::Write>(
     mut terminal: Terminal<B>,
     mut app: App<Box<dyn io::Write + Send>>,
-    callback: Option<mpsc::SyncSender<()>>,
+    callback: Option<oneshot::Sender<()>>,
 ) -> io::Result<()> {
     terminal.clear()?;
     crossterm::execute!(
@@ -692,7 +713,7 @@ fn cleanup<B: Backend + io::Write>(
 fn update(
     app: &mut App<Box<dyn io::Write + Send>>,
     event: Event,
-) -> Result<Option<mpsc::SyncSender<()>>, Error> {
+) -> Result<Option<oneshot::Sender<()>>, Error> {
     match event {
         Event::StartTask { task, output_logs } => {
             app.start_task(&task, output_logs)?;

crates/turborepo-ui/src/tui/event.rs

@@ -1,5 +1,6 @@
 use async_graphql::Enum;
 use serde::Serialize;
+use tokio::sync::oneshot;
 
 pub enum Event {
     StartTask {
@@ -19,8 +20,8 @@ pub enum Event {
         status: String,
         result: CacheResult,
     },
-    PaneSizeQuery(std::sync::mpsc::SyncSender<PaneSize>),
-    Stop(std::sync::mpsc::SyncSender<()>),
+    PaneSizeQuery(oneshot::Sender<PaneSize>),
+    Stop(oneshot::Sender<()>),
     // Stop initiated by the TUI itself
     InternalStop,
     Tick,

crates/turborepo-ui/src/tui/handle.rs

@@ -1,6 +1,7 @@
-use std::{sync::mpsc, time::Instant};
+use tokio::sync::{mpsc, oneshot};
 
 use super::{
+    app::FRAMERATE,
     event::{CacheResult, OutputLogs, PaneSize},
     Error, Event, TaskResult,
 };
@@ -9,12 +10,12 @@ use crate::sender::{TaskSender, UISender};
 /// Struct for sending app events to TUI rendering
 #[derive(Debug, Clone)]
 pub struct TuiSender {
-    primary: mpsc::Sender<Event>,
+    primary: mpsc::UnboundedSender<Event>,
 }
 
 /// Struct for receiving app events
 pub struct AppReceiver {
-    primary: mpsc::Receiver<Event>,
+    primary: mpsc::UnboundedReceiver<Event>,
 }
 
 impl TuiSender {
@@ -23,7 +24,17 @@ impl TuiSender {
     /// AppSender is meant to be held by the actual task runner
     /// AppReceiver should be passed to `crate::tui::run_app`
     pub fn new() -> (Self, AppReceiver) {
-        let (primary_tx, primary_rx) = mpsc::channel();
+        let (primary_tx, primary_rx) = mpsc::unbounded_channel();
+        let tick_sender = primary_tx.clone();
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(FRAMERATE);
+            loop {
+                interval.tick().await;
+                if tick_sender.send(Event::Tick).is_err() {
+                    break;
+                }
+            }
+        });
         (
             Self {
                 primary: primary_tx,
@@ -70,13 +81,13 @@ impl TuiSender {
     }
 
     /// Stop rendering TUI and restore terminal to default configuration
-    pub fn stop(&self) {
-        let (callback_tx, callback_rx) = mpsc::sync_channel(1);
+    pub async fn stop(&self) {
+        let (callback_tx, callback_rx) = oneshot::channel();
         // Send stop event, if receiver has dropped ignore error as
         // it'll be a no-op.
         self.primary.send(Event::Stop(callback_tx)).ok();
         // Wait for callback to be sent or the channel closed.
-        callback_rx.recv().ok();
+        callback_rx.await.ok();
     }
 
     /// Update the list of tasks displayed in the TUI
@@ -103,23 +114,19 @@ impl TuiSender {
     }
 
     /// Fetches the size of the terminal pane
-    pub fn pane_size(&self) -> Option<PaneSize> {
-        let (callback_tx, callback_rx) = mpsc::sync_channel(1);
+    pub async fn pane_size(&self) -> Option<PaneSize> {
+        let (callback_tx, callback_rx) = oneshot::channel();
         // Send query, if no receiver to handle the request return None
         self.primary.send(Event::PaneSizeQuery(callback_tx)).ok()?;
         // Wait for callback to be sent
-        callback_rx.recv().ok()
+        callback_rx.await.ok()
     }
 }
 
 impl AppReceiver {
     /// Receive an event, producing a tick event if no events are rec eived by
     /// the deadline.
-    pub fn recv(&self, deadline: Instant) -> Result<Event, mpsc::RecvError> {
-        match self.primary.recv_deadline(deadline) {
-            Ok(event) => Ok(event),
-            Err(mpsc::RecvTimeoutError::Timeout) => Ok(Event::Tick),
-            Err(mpsc::RecvTimeoutError::Disconnected) => Err(mpsc::RecvError),
-        }
+    pub async fn recv(&mut self) -> Option<Event> {
+        self.primary.recv().await
     }
 }