| 1 | //! Terminal input ingestion and fairness controls for the TUI event loop. |
| 2 | |
| 3 | use std::cell::Cell; |
| 4 | use std::io; |
| 5 | use std::sync::{ |
| 6 | Arc, Mutex, |
| 7 | atomic::{AtomicBool, Ordering}, |
| 8 | }; |
| 9 | use std::thread::{self, JoinHandle}; |
| 10 | use std::time::{Duration, Instant}; |
| 11 | |
| 12 | use crossterm::event::{self, Event}; |
| 13 | |
| 14 | const TERMINAL_INPUT_POLL_INTERVAL: Duration = Duration::from_millis(50); |
| 15 | const TERMINAL_INPUT_HEARTBEAT_INTERVAL: Duration = Duration::from_millis(500); |
| 16 | pub(super) const TERMINAL_INPUT_STALL_TIMEOUT: Duration = Duration::from_secs(5); |
| 17 | pub(super) const TERMINAL_INPUT_RECOVERY_COOLDOWN: Duration = Duration::from_secs(10); |
| 18 | const TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT: Duration = Duration::from_millis(500); |
| 19 | const TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL: Duration = Duration::from_millis(5); |
| 20 | /// Upper bound on engine events processed before yielding to terminal input. |
| 21 | pub(super) const MAX_ENGINE_EVENTS_PER_DRAIN: usize = 16; |
| 22 | /// Wall-clock budget for one engine drain batch (#1830 / #2317 input fairness). |
| 23 | pub(super) const ENGINE_DRAIN_TIME_BUDGET: Duration = Duration::from_millis(8); |
| 24 | |
| 25 | pub(super) enum TerminalInputMessage { |
| 26 | Event(ObservedTerminalEvent), |
| 27 | Heartbeat, |
| 28 | Error(io::Error), |
| 29 | } |
| 30 | |
| 31 | /// A terminal event paired with the instant the dedicated input thread read it. |
| 32 | /// |
| 33 | /// The event loop can lag behind this thread under load. Keeping receipt time |
| 34 | /// prevents that backlog from collapsing a deliberate pause between a raw |
| 35 | /// paste and Enter into a paste-speed sequence. |
| 36 | pub(crate) struct ObservedTerminalEvent { |
| 37 | pub(crate) event: Event, |
| 38 | pub(crate) observed_at: Instant, |
| 39 | } |
| 40 | |
| 41 | impl ObservedTerminalEvent { |
| 42 | pub(crate) fn new(event: Event, observed_at: Instant) -> Self { |
| 43 | Self { event, observed_at } |
| 44 | } |
| 45 | } |
| 46 | |
| 47 | /// Process-wide handle on the one terminal input pump's pause flags. |
| 48 | /// |
| 49 | /// There is one stdin per process and exactly one [`TerminalInputPump`] |
| 50 | /// reading it, so this is a singleton by construction rather than by |
| 51 | /// convention. It exists so that *handing the terminal to a child* can be one |
| 52 | /// operation instead of a rule every call site has to remember (#6165): |
| 53 | /// suspending raw mode and the alternate screen does not stop the pump |
| 54 | /// thread, which keeps calling `event::read()` on the same tty and splits the |
| 55 | /// user's keystrokes between the child and the composer. |
| 56 | /// |
| 57 | /// Known limitation: the gate only stops the pump reading. It cannot drain |
| 58 | /// input the pump already buffered, and it cannot refuse the handoff when a |
| 59 | /// cancellation key is pending — both need the receiver and the event loop's |
| 60 | /// pending queue, so they stay with [`super::prepare_terminal_input_handoff`] |
| 61 | /// at the call sites that have them. |
| 62 | static CHILD_TERMINAL_GATE: Mutex<Option<ChildTerminalGate>> = Mutex::new(None); |
| 63 | |
| 64 | #[derive(Clone)] |
| 65 | struct ChildTerminalGate { |
| 66 | paused: Arc<AtomicBool>, |
| 67 | paused_ack: Arc<AtomicBool>, |
| 68 | } |
| 69 | |
| 70 | /// Publish this pump as the process's terminal input owner. |
| 71 | pub(super) fn publish_child_terminal_gate(paused: &Arc<AtomicBool>, paused_ack: &Arc<AtomicBool>) { |
| 72 | if let Ok(mut gate) = CHILD_TERMINAL_GATE.lock() { |
| 73 | *gate = Some(ChildTerminalGate { |
| 74 | paused: Arc::clone(paused), |
| 75 | paused_ack: Arc::clone(paused_ack), |
| 76 | }); |
| 77 | } |
| 78 | } |
| 79 | |
| 80 | /// Retract `paused`'s pump, but only if it is still the published one — a |
| 81 | /// detached wedged thread must not unpublish the replacement that took over. |
| 82 | fn retract_child_terminal_gate(paused: &Arc<AtomicBool>) { |
| 83 | if let Ok(mut gate) = CHILD_TERMINAL_GATE.lock() |
| 84 | && gate |
| 85 | .as_ref() |
| 86 | .is_some_and(|current| Arc::ptr_eq(¤t.paused, paused)) |
| 87 | { |
| 88 | *gate = None; |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | /// The terminal input pump, paused for as long as this guard is alive. |
| 93 | /// |
| 94 | /// Held by [`crate::tui::external_editor::with_suspended_tui`] across the |
| 95 | /// whole child handoff, so the pump resumes on every path out — including a |
| 96 | /// child that failed to spawn or a panic unwinding through it. |
| 97 | pub(crate) struct ChildTerminalInputPause { |
| 98 | gate: Option<ChildTerminalGate>, |
| 99 | } |
| 100 | |
| 101 | /// Stop the process's terminal input pump before a child takes the tty. |
| 102 | /// |
| 103 | /// Fails closed: if the pump does not acknowledge the pause, the caller must |
| 104 | /// not run the child, because that is exactly the keystroke-splitting state |
| 105 | /// this guards against. A process with no pump published (tests, non-TUI |
| 106 | /// callers) has nothing to pause and succeeds with an inert guard, matching |
| 107 | /// [`TerminalInputPump::pause_for_child_terminal`]'s `handle.is_none()` case. |
| 108 | pub(crate) fn pause_terminal_input_for_child() -> io::Result<ChildTerminalInputPause> { |
| 109 | let Some(gate) = CHILD_TERMINAL_GATE |
| 110 | .lock() |
| 111 | .ok() |
| 112 | .and_then(|gate| gate.clone()) |
| 113 | else { |
| 114 | return Ok(ChildTerminalInputPause { gate: None }); |
| 115 | }; |
| 116 | gate.paused.store(true, Ordering::Release); |
| 117 | let deadline = Instant::now() + TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT; |
| 118 | while !gate.paused_ack.load(Ordering::Acquire) { |
| 119 | if Instant::now() >= deadline { |
| 120 | gate.paused_ack.store(false, Ordering::Release); |
| 121 | gate.paused.store(false, Ordering::Release); |
| 122 | return Err(io::Error::new( |
| 123 | io::ErrorKind::TimedOut, |
| 124 | "terminal input pump did not pause before child terminal handoff", |
| 125 | )); |
| 126 | } |
| 127 | // Blocking-call convention (#6149): a bounded retry, capped by |
| 128 | // `TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT`, in a synchronous API whose |
| 129 | // caller is about to block this very thread on a foreground editor |
| 130 | // for as long as the user keeps it open. `tokio::time` is not |
| 131 | // reachable from here and would not change what the thread does. |
| 132 | thread::sleep(TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL); |
| 133 | } |
| 134 | Ok(ChildTerminalInputPause { gate: Some(gate) }) |
| 135 | } |
| 136 | |
| 137 | impl Drop for ChildTerminalInputPause { |
| 138 | fn drop(&mut self) { |
| 139 | if let Some(gate) = self.gate.take() { |
| 140 | gate.paused_ack.store(false, Ordering::Release); |
| 141 | gate.paused.store(false, Ordering::Release); |
| 142 | } |
| 143 | } |
| 144 | } |
| 145 | |
| 146 | pub(crate) struct TerminalInputPump { |
| 147 | pub(super) rx: std::sync::mpsc::Receiver<TerminalInputMessage>, |
| 148 | pub(super) stop: Arc<AtomicBool>, |
| 149 | pub(super) paused: Arc<AtomicBool>, |
| 150 | pub(super) paused_ack: Arc<AtomicBool>, |
| 151 | pub(super) handle: Option<JoinHandle<()>>, |
| 152 | pub(super) last_alive_at: Cell<Instant>, |
| 153 | } |
| 154 | |
| 155 | pub(super) struct TerminalInputPumpParts { |
| 156 | pub(super) rx: std::sync::mpsc::Receiver<TerminalInputMessage>, |
| 157 | pub(super) stop: Arc<AtomicBool>, |
| 158 | pub(super) paused: Arc<AtomicBool>, |
| 159 | pub(super) paused_ack: Arc<AtomicBool>, |
| 160 | pub(super) handle: JoinHandle<()>, |
| 161 | } |
| 162 | |
| 163 | impl TerminalInputPump { |
| 164 | pub(super) fn spawn() -> io::Result<Self> { |
| 165 | let parts = Self::spawn_parts()?; |
| 166 | publish_child_terminal_gate(&parts.paused, &parts.paused_ack); |
| 167 | Ok(Self { |
| 168 | rx: parts.rx, |
| 169 | stop: parts.stop, |
| 170 | paused: parts.paused, |
| 171 | paused_ack: parts.paused_ack, |
| 172 | handle: Some(parts.handle), |
| 173 | last_alive_at: Cell::new(Instant::now()), |
| 174 | }) |
| 175 | } |
| 176 | |
| 177 | fn spawn_parts() -> io::Result<TerminalInputPumpParts> { |
| 178 | let (tx, rx) = std::sync::mpsc::channel(); |
| 179 | let stop = Arc::new(AtomicBool::new(false)); |
| 180 | let paused = Arc::new(AtomicBool::new(false)); |
| 181 | let paused_ack = Arc::new(AtomicBool::new(false)); |
| 182 | let thread_stop = Arc::clone(&stop); |
| 183 | let thread_paused = Arc::clone(&paused); |
| 184 | let thread_paused_ack = Arc::clone(&paused_ack); |
| 185 | let handle = thread::Builder::new() |
| 186 | .name("codewhale-terminal-input".to_string()) |
| 187 | .spawn(move || { |
| 188 | let mut last_heartbeat = Instant::now(); |
| 189 | while !thread_stop.load(Ordering::Acquire) { |
| 190 | if thread_paused.load(Ordering::Acquire) { |
| 191 | thread_paused_ack.store(true, Ordering::Release); |
| 192 | thread::sleep(TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL); |
| 193 | continue; |
| 194 | } |
| 195 | thread_paused_ack.store(false, Ordering::Release); |
| 196 | match event::poll(TERMINAL_INPUT_POLL_INTERVAL) { |
| 197 | Ok(true) if thread_stop.load(Ordering::Acquire) => break, |
| 198 | Ok(true) => match event::read() { |
| 199 | Ok(event) => { |
| 200 | last_heartbeat = Instant::now(); |
| 201 | let observed = ObservedTerminalEvent::new(event, last_heartbeat); |
| 202 | if tx.send(TerminalInputMessage::Event(observed)).is_err() { |
| 203 | break; |
| 204 | } |
| 205 | } |
| 206 | Err(err) => { |
| 207 | let _ = tx.send(TerminalInputMessage::Error(err)); |
| 208 | break; |
| 209 | } |
| 210 | }, |
| 211 | Ok(false) => { |
| 212 | let now = Instant::now(); |
| 213 | if now.duration_since(last_heartbeat) |
| 214 | >= TERMINAL_INPUT_HEARTBEAT_INTERVAL |
| 215 | { |
| 216 | last_heartbeat = now; |
| 217 | if tx.send(TerminalInputMessage::Heartbeat).is_err() { |
| 218 | break; |
| 219 | } |
| 220 | } |
| 221 | } |
| 222 | Err(err) => { |
| 223 | let _ = tx.send(TerminalInputMessage::Error(err)); |
| 224 | break; |
| 225 | } |
| 226 | } |
| 227 | } |
| 228 | })?; |
| 229 | Ok(TerminalInputPumpParts { |
| 230 | rx, |
| 231 | stop, |
| 232 | paused, |
| 233 | paused_ack, |
| 234 | handle, |
| 235 | }) |
| 236 | } |
| 237 | |
| 238 | pub(super) fn recv_timeout( |
| 239 | &self, |
| 240 | timeout: Duration, |
| 241 | ) -> io::Result<Option<ObservedTerminalEvent>> { |
| 242 | let deadline = Instant::now() + timeout; |
| 243 | loop { |
| 244 | let remaining = deadline.saturating_duration_since(Instant::now()); |
| 245 | match self.rx.recv_timeout(remaining) { |
| 246 | Ok(TerminalInputMessage::Event(event)) => { |
| 247 | self.mark_alive(); |
| 248 | return Ok(Some(event)); |
| 249 | } |
| 250 | Ok(TerminalInputMessage::Heartbeat) => { |
| 251 | self.mark_alive(); |
| 252 | if remaining.is_zero() { |
| 253 | return Ok(None); |
| 254 | } |
| 255 | } |
| 256 | Ok(TerminalInputMessage::Error(err)) => { |
| 257 | self.mark_alive(); |
| 258 | return Err(err); |
| 259 | } |
| 260 | Err(std::sync::mpsc::RecvTimeoutError::Timeout) => return Ok(None), |
| 261 | Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { |
| 262 | return Err(io::Error::new( |
| 263 | io::ErrorKind::BrokenPipe, |
| 264 | "terminal input pump disconnected", |
| 265 | )); |
| 266 | } |
| 267 | } |
| 268 | } |
| 269 | } |
| 270 | |
| 271 | pub(super) fn try_recv(&self) -> io::Result<Option<ObservedTerminalEvent>> { |
| 272 | loop { |
| 273 | match self.rx.try_recv() { |
| 274 | Ok(TerminalInputMessage::Event(event)) => { |
| 275 | self.mark_alive(); |
| 276 | return Ok(Some(event)); |
| 277 | } |
| 278 | Ok(TerminalInputMessage::Heartbeat) => { |
| 279 | self.mark_alive(); |
| 280 | } |
| 281 | Ok(TerminalInputMessage::Error(err)) => { |
| 282 | self.mark_alive(); |
| 283 | return Err(err); |
| 284 | } |
| 285 | Err(std::sync::mpsc::TryRecvError::Empty) => return Ok(None), |
| 286 | Err(std::sync::mpsc::TryRecvError::Disconnected) => return Ok(None), |
| 287 | } |
| 288 | } |
| 289 | } |
| 290 | |
| 291 | pub(super) fn mark_alive(&self) { |
| 292 | self.last_alive_at.set(Instant::now()); |
| 293 | } |
| 294 | |
| 295 | pub(super) fn stalled_for(&self, now: Instant) -> Duration { |
| 296 | now.saturating_duration_since(self.last_alive_at.get()) |
| 297 | } |
| 298 | |
| 299 | /// Async: callers are on the event-loop task, so the ack wait uses |
| 300 | /// `tokio::time::sleep` — a `thread::sleep` here would park a Tokio |
| 301 | /// worker for up to `TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT`. |
| 302 | pub(super) async fn pause_for_child_terminal(&self) -> io::Result<()> { |
| 303 | self.paused.store(true, Ordering::Release); |
| 304 | if self.handle.is_none() { |
| 305 | self.paused_ack.store(true, Ordering::Release); |
| 306 | self.mark_alive(); |
| 307 | return Ok(()); |
| 308 | } |
| 309 | |
| 310 | let deadline = Instant::now() + TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT; |
| 311 | while !self.paused_ack.load(Ordering::Acquire) { |
| 312 | if Instant::now() >= deadline { |
| 313 | self.paused_ack.store(false, Ordering::Release); |
| 314 | self.paused.store(false, Ordering::Release); |
| 315 | return Err(io::Error::new( |
| 316 | io::ErrorKind::TimedOut, |
| 317 | "terminal input pump did not pause before child terminal handoff", |
| 318 | )); |
| 319 | } |
| 320 | tokio::time::sleep(TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL).await; |
| 321 | } |
| 322 | self.mark_alive(); |
| 323 | Ok(()) |
| 324 | } |
| 325 | |
| 326 | pub(super) fn resume_after_child_terminal(&self) { |
| 327 | self.paused_ack.store(false, Ordering::Release); |
| 328 | self.paused.store(false, Ordering::Release); |
| 329 | self.mark_alive(); |
| 330 | } |
| 331 | |
| 332 | /// Replace a wedged pump thread with a freshly spawned one. |
| 333 | /// |
| 334 | /// The old thread may be blocked forever inside crossterm's blocking |
| 335 | /// `event::read` (a stalled Windows console poll, or a Unix tty that |
| 336 | /// stopped delivering bytes), so it can never be joined. Instead it is |
| 337 | /// detached: `stop` is flagged and the `JoinHandle` dropped, so if the |
| 338 | /// thread ever wakes it exits on its own (its send fails once `rx` is |
| 339 | /// replaced, and the stop flag covers the poll loop). |
| 340 | pub(super) fn restart_detached(&mut self) -> io::Result<()> { |
| 341 | self.detach_current_thread(); |
| 342 | let parts = Self::spawn_parts()?; |
| 343 | self.install_parts(parts); |
| 344 | Ok(()) |
| 345 | } |
| 346 | |
| 347 | /// Flag the current pump thread to stop and drop its handle without |
| 348 | /// joining (the thread may be wedged in a blocking terminal read). |
| 349 | pub(super) fn detach_current_thread(&mut self) { |
| 350 | self.stop.store(true, Ordering::Release); |
| 351 | let _ = self.handle.take(); |
| 352 | retract_child_terminal_gate(&self.paused); |
| 353 | } |
| 354 | |
| 355 | /// Adopt freshly spawned pump parts and reset the liveness clock. |
| 356 | pub(super) fn install_parts(&mut self, parts: TerminalInputPumpParts) { |
| 357 | publish_child_terminal_gate(&parts.paused, &parts.paused_ack); |
| 358 | self.rx = parts.rx; |
| 359 | self.stop = parts.stop; |
| 360 | self.paused = parts.paused; |
| 361 | self.paused_ack = parts.paused_ack; |
| 362 | self.handle = Some(parts.handle); |
| 363 | self.last_alive_at.set(Instant::now()); |
| 364 | } |
| 365 | } |
| 366 | |
| 367 | impl Drop for TerminalInputPump { |
| 368 | fn drop(&mut self) { |
| 369 | // `event::read` can remain blocked forever after a tty disconnect on |
| 370 | // every supported desktop platform. Joining here would turn an input |
| 371 | // failure into an application shutdown hang. Flag the cooperative |
| 372 | // stop and detach; if the read ever wakes, the loop observes `stop` |
| 373 | // (or its send fails because `rx` was dropped) and exits on its own. |
| 374 | self.stop.store(true, Ordering::Release); |
| 375 | let _ = self.handle.take(); |
| 376 | retract_child_terminal_gate(&self.paused); |
| 377 | } |
| 378 | } |
| 379 | |
| 380 | pub(super) fn engine_drain_budget_exhausted( |
| 381 | events_drained: usize, |
| 382 | started: Instant, |
| 383 | now: Instant, |
| 384 | ) -> bool { |
| 385 | events_drained >= MAX_ENGINE_EVENTS_PER_DRAIN |
| 386 | || now.saturating_duration_since(started) >= ENGINE_DRAIN_TIME_BUDGET |
| 387 | } |
| 388 |