perf(tui): bring async to tui

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
  }
 }