feat(tui): adopt an asynchronous Tokio runtime for the TUI frontend

Refactor TUI event handling and rendering to improve responsiveness and use a unified managed client worker.

crates/irtt-app/src/cmd/tui/run.rs

@@ -1,133 +1,168 @@
-use std::{
-    collections::HashSet,
-    io,
-    sync::atomic::{AtomicBool, Ordering},
-    thread,
-    time::{Duration, Instant},
-};
+use std::{collections::HashSet, future::poll_fn, io, time::Duration};
 
-use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal};
+use super::ui::{TuiConfig, TuiState, TuiStatus, TuiTerminal};
 use crate::{
     cmd::tui::args::TuiArgs,
     shared::client::{
-        is_shutdown_requested,
         session::{
-            drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
-            request_managed_stop_once, ManagedDrainState,
+            drain_final_events, peer_close_run_error, request_managed_stop_for_peer_close,
+            request_managed_stop_once,
         },
+        worker::ManagedWorker,
     },
 };
 use crossterm::{
-    event::{self, Event, KeyCode, KeyEventKind, KeyModifiers},
+    event::{Event, EventStream, KeyCode, KeyEventKind, KeyModifiers},
     terminal,
 };
-use irtt_client::managed::{
-    BlockingManagedClient, ManagedEndReason, ManagedEvent, ManagedEventSubscription,
-    ManagedEventTryRecvError, TargetInstance,
+use futures_core::Stream;
+use irtt_client::{
+    managed::{ManagedClient, ManagedEndReason, ManagedEvent, TargetInstance},
+    ClientEvent,
 };
 use ratatui::layout::Rect;
+use tokio::{
+    sync::{broadcast::error::RecvError, watch},
+    time::{sleep, Instant},
+};
 
 const RENDER_INTERVAL: Duration = Duration::from_millis(250);
-const TUI_WAIT_SLICE: Duration = Duration::from_millis(20);
-const INPUT_EVENT_WORK_BUDGET: usize = 128;
 
-pub fn run_tui(
+pub async fn run_tui(
     args: TuiArgs,
-    shutdown_requested: &AtomicBool,
+    mut shutdown: watch::Receiver<bool>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let setup = args
         .prepare()
         .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
     let continuous = args.is_continuous();
+    if *shutdown.borrow() {
+        return Ok(());
+    }
     let mut terminal = TuiTerminal::enter()?;
     let mut state = TuiState::with_target_labels(
         TuiConfig::from_args(&args, &setup),
         setup.targets.iter().map(|target| target.label.clone()),
     );
-    let mut next_render = Instant::now();
-    if is_shutdown_requested(shutdown_requested) {
-        return Ok(());
-    }
     state.set_status(TuiStatus::Opening);
-    render_if_due(&mut terminal, &state, &mut next_render, true)?;
-    let (owner, mut events) = match BlockingManagedClient::start_with_subscription(
-        setup.managed_config(),
-        setup.managed_targets(),
-    ) {
+    terminal.draw(&state)?;
+    let (task, handle) = match ManagedClient::task(setup.managed_config(), setup.managed_targets())
+    {
         Ok(value) => value,
         Err(error) => {
             state.set_error(error.to_string());
-            render_if_due(&mut terminal, &state, &mut next_render, true)?;
+            terminal.draw(&state)?;
             return Err(Box::new(error));
         }
     };
-    let handle = owner.handle();
+    // Both subscriptions exist before the measurement task can be polled.
+    let mut events = handle.subscribe()?;
+    let mut status = handle.subscribe_status();
+    let worker = match ManagedWorker::start(task, handle.clone()) {
+        Ok(worker) => worker,
+        Err(error) => {
+            state.set_run_error(error.to_string());
+            terminal.draw(&state)?;
+            return Err(error.into());
+        }
+    };
+    let completion = worker.join();
+    tokio::pin!(completion);
+    // Input and rendering stay on the frontend executor. EventStream's drop
+    // wakes its reader; no terminal task can outlive the terminal guard.
+    let mut input = EventStream::new();
+    let render = sleep(RENDER_INTERVAL);
+    tokio::pin!(render);
     let mut interrupted = false;
     let mut stop_requested = false;
     let mut terminal_targets = HashSet::new();
-    let mut dropped_events = 0;
-    let mut subscription_closed = false;
-    loop {
-        if is_shutdown_requested(shutdown_requested) {
-            interrupted = true;
-            if request_managed_stop_once(&mut stop_requested) {
-                drop(handle.stop());
+    let mut dropped_events = 0_u64;
+    let mut events_closed = false;
+    let mut status_closed = false;
+    let outcome = loop {
+        let mut force_render = false;
+        // Default select fairness lets input, status and rendering progress
+        // even when measurement events are continuously ready.
+        tokio::select! {
+            outcome = &mut completion => break outcome,
+            _ = shutdown.changed(), if !interrupted => {
+                interrupted = true;
+                state.set_status(TuiStatus::Interrupted);
+                force_render = true;
             }
-        }
-        if handle_input(&mut state, shutdown_requested)? {
-            render_if_due(&mut terminal, &state, &mut next_render, true)?;
-        }
-        if state.quit_requested {
-            interrupted = true;
-            if request_managed_stop_once(&mut stop_requested) {
-                drop(handle.stop());
+            changed = status.changed(), if !status_closed => {
+                status_closed = changed.is_err();
+                let snapshot = status.borrow_and_update().clone();
+                if request_managed_stop_for_peer_close(
+                    continuous, interrupted, snapshot.peer_closed_target_outcomes,
+                    &mut stop_requested,
+                ) {
+                    drop(handle.stop());
+                }
+                force_render = true;
             }
+            event = events.recv(), if !events_closed => match event {
+                Ok(event) => force_render = process_tui_event(
+                    event, &mut state, &mut terminal_targets,
+                ),
+                Err(RecvError::Lagged(count)) => {
+                    dropped_events = dropped_events.saturating_add(count);
+                    state.mark_dropped_managed_events(dropped_events);
+                    force_render = true;
+                }
+                Err(RecvError::Closed) => events_closed = true,
+            },
+            event = poll_fn(|cx| std::pin::Pin::new(&mut input).poll_next(cx)) => {
+                let event = event.ok_or_else(|| io::Error::new(
+                    io::ErrorKind::UnexpectedEof, "terminal input closed",
+                ))??;
+                match event {
+                    Event::Resize(_, _) => force_render = true,
+                    Event::Key(key) if key.kind != KeyEventKind::Release => {
+                        match key.code {
+                            KeyCode::Char('q') | KeyCode::Char('c')
+                                if key.code == KeyCode::Char('q')
+                                    || key.modifiers.contains(KeyModifiers::CONTROL) => {
+                                interrupted = true;
+                                state.set_status(TuiStatus::Interrupted);
+                                force_render = true;
+                            }
+                            _ => {
+                                let (width, height) = terminal::size()?;
+                                force_render = state.handle_key(
+                                    key.code, Rect::new(0, 0, width, height),
+                                );
+                            }
+                        }
+                    }
+                    _ => {}
+                }
+            }
+            _ = &mut render, if !state.paused => force_render = true,
         }
-        if request_managed_stop_for_peer_close(
-            continuous,
-            interrupted,
-            handle.status().peer_closed_target_outcomes,
-            &mut stop_requested,
-        ) {
+        if interrupted && request_managed_stop_once(&mut stop_requested) {
             drop(handle.stop());
         }
-        let previous_dropped = dropped_events;
-        let drain_state = drain_tui_events(
-            &mut events,
-            &mut state,
-            &mut terminal_targets,
-            &mut dropped_events,
-        );
-        if dropped_events != previous_dropped {
-            // Data loss changes the meaning of displayed statistics, even while
-            // ordinary measurement refreshes are paused.
-            render_if_due(&mut terminal, &state, &mut next_render, true)?;
+        if force_render {
+            terminal.draw(&state)?;
+            // Schedule from the completed draw, never catch up missed frames.
+            render.as_mut().reset(Instant::now() + RENDER_INTERVAL);
         }
-        match drain_state {
-            ManagedDrainState::Empty => {
-                thread::sleep(managed_tui_wait_duration(&next_render, state.paused));
-            }
-            ManagedDrainState::Closed => subscription_closed = true,
-            ManagedDrainState::BudgetExhausted => {}
-        }
-        if handle.status().final_outcome.is_some() || subscription_closed {
-            break;
-        }
-        render_if_due(&mut terminal, &state, &mut next_render, false)?;
-    }
-    if interrupted {
-        state.set_status(TuiStatus::Interrupted);
-        render_if_due(&mut terminal, &state, &mut next_render, true)?;
-    }
+    };
     state.set_status(TuiStatus::Closing);
-    render_if_due(&mut terminal, &state, &mut next_render, true)?;
-    let outcome = owner.join()?;
-    drain_final_tui_events(
-        &mut events,
-        &mut state,
-        &mut terminal_targets,
-        &mut dropped_events,
-    );
+    terminal.draw(&state)?;
+    let outcome = match outcome {
+        Ok(outcome) => outcome,
+        Err(error) => {
+            state.set_run_error(error.to_string());
+            terminal.draw(&state)?;
+            return Err(error);
+        }
+    };
+    drain_final_events(&mut events, &mut dropped_events, |event| {
+        process_tui_event(event, &mut state, &mut terminal_targets);
+        Ok::<(), std::convert::Infallible>(())
+    })?;
     state.mark_dropped_managed_events(dropped_events);
     if outcome.discarded_target_outcomes != 0 {
         state.set_run_error(format!(
@@ -140,7 +175,7 @@ pub fn run_tui(
             state.process_target_outcome(target);
         }
     }
-    interrupted |= is_shutdown_requested(shutdown_requested);
+    interrupted |= *shutdown.borrow();
     let error = match &outcome.end_reason {
         ManagedEndReason::DriverFailed(failure) => {
             Some(format!("managed driver failed: {failure}"))
@@ -160,11 +195,11 @@ pub fn run_tui(
     };
     if let Some(error) = error {
         state.set_run_error(error.clone());
-        render_if_due(&mut terminal, &state, &mut next_render, true)?;
+        terminal.draw(&state)?;
         return Err(error.into());
     }
     state.set_status(TuiStatus::Complete);
-    render_if_due(&mut terminal, &state, &mut next_render, true)?;
+    terminal.draw(&state)?;
     Ok(())
 }
 
@@ -172,117 +207,36 @@ fn process_tui_event(
     event: ManagedEvent,
     state: &mut TuiState,
     terminal_targets: &mut HashSet<TargetInstance>,
-) {
+) -> bool {
     match event {
-        ManagedEvent::Client { target, event } => state.process_target_event(&target, &event),
+        ManagedEvent::Client { target, event } => {
+            let force_render = matches!(
+                event,
+                ClientEvent::SessionStarted(_)
+                    | ClientEvent::NoTestCompleted(_)
+                    | ClientEvent::SessionClosed { .. }
+                    | ClientEvent::Warning { .. }
+            );
+            state.process_target_event(&target, &event);
+            force_render
+        }
         ManagedEvent::TargetFinished { outcome } => {
             terminal_targets.insert(outcome.target.clone());
             state.process_target_outcome(&outcome);
+            true
         }
-        _ => {}
+        _ => false,
     }
 }
-fn drain_tui_events(
-    events: &mut ManagedEventSubscription,
-    state: &mut TuiState,
-    terminal_targets: &mut HashSet<TargetInstance>,
-    dropped_events: &mut u64,
-) -> ManagedDrainState {
-    let previous_dropped = *dropped_events;
-    let drained = drain_managed_events(events, dropped_events, |event| {
-        process_tui_event(event, state, terminal_targets);
-        Ok::<(), std::convert::Infallible>(())
-    })
-    .expect("processing TUI managed events is infallible");
-    if *dropped_events != previous_dropped {
-        state.mark_dropped_managed_events(*dropped_events);
-    }
-    drained
-}
-
-fn drain_final_tui_events(
-    events: &mut ManagedEventSubscription,
-    state: &mut TuiState,
-    terminal_targets: &mut HashSet<TargetInstance>,
-    dropped_events: &mut u64,
-) {
-    loop {
-        match events.try_recv() {
-            Ok(event) => process_tui_event(event, state, terminal_targets),
-            Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
-            Err(ManagedEventTryRecvError::Lagged(count)) => {
-                *dropped_events = dropped_events.saturating_add(count);
-            }
-        }
-    }
-}
-fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Result<bool> {
-    let mut force_render = false;
-    for _ in 0..INPUT_EVENT_WORK_BUDGET {
-        if !event::poll(Duration::ZERO)? {
-            break;
-        }
-        let key = match event::read()? {
-            Event::Key(key) => key,
-            Event::Resize(_, _) => {
-                force_render = true;
-                continue;
-            }
-            _ => continue,
-        };
-        if key.kind == KeyEventKind::Release {
-            continue;
-        }
-        match key.code {
-            KeyCode::Char('q') | KeyCode::Char('c')
-                if key.code == KeyCode::Char('q')
-                    || key.modifiers.contains(KeyModifiers::CONTROL) =>
-            {
-                state.quit_requested = true;
-                shutdown_requested.store(true, Ordering::Relaxed);
-                force_render = true;
-                break;
-            }
-            _ => {
-                let (width, height) = terminal::size()?;
-                force_render |= state.handle_key(key.code, Rect::new(0, 0, width, height));
-            }
-        }
-    }
-    Ok(force_render)
-}
-fn render_if_due(
-    terminal: &mut TuiTerminal,
-    state: &TuiState,
-    next_render: &mut Instant,
-    force: bool,
-) -> io::Result<()> {
-    let now = Instant::now();
-    if should_render(now, *next_render, state.paused, force) {
-        terminal.draw(state)?;
-        *next_render = now + RENDER_INTERVAL;
-    }
-    Ok(())
-}
-fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration {
-    let render_wait = if paused {
-        TUI_WAIT_SLICE
-    } else {
-        next_render.saturating_duration_since(Instant::now())
-    };
-    render_wait.min(TUI_WAIT_SLICE)
-}
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use ratatui::{backend::TestBackend, Terminal};
 
-    #[test]
-    fn subscription_lag_keeps_incomplete_statistics_visible_while_paused() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn subscription_lag_keeps_incomplete_statistics_visible_while_paused() {
         for paused in [false, true] {
-            // The production subscription is a broadcast receiver. Overflow the
-            // real channel rather than synthesizing a lag result or UI state.
             let (sender, mut events) = tokio::sync::broadcast::channel(1);
             sender.send(ManagedEvent::Started).unwrap();
             sender.send(ManagedEvent::Started).unwrap();
@@ -290,14 +244,12 @@ mod tests {
             if paused {
                 state.toggle_pause();
             }
-            let mut dropped = 0;
-            let mut terminal_targets = HashSet::new();
-            drain_tui_events(&mut events, &mut state, &mut terminal_targets, &mut dropped);
-            assert_eq!(dropped, 1);
+            let Err(RecvError::Lagged(dropped)) = events.recv().await else {
+                panic!("subscription must report real broadcast overflow");
+            };
+            state.mark_dropped_managed_events(dropped);
             let mut terminal = Terminal::new(TestBackend::new(200, 40)).unwrap();
             for _ in 0..2 {
-                // Empty drains must not clear the warning on later redraws.
-                drain_tui_events(&mut events, &mut state, &mut terminal_targets, &mut dropped);
                 terminal
                     .draw(|frame| super::super::ui::draw_dashboard(frame, &state))
                     .unwrap();