From 5559aa906ed6abb6723320b604a5fc8a7a60bcee Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 01:54:30 -0700 Subject: [PATCH 01/10] io: bounded shared PTY buffer with one wake in flight The reader posted every 4 KiB read as its own UserEvent::PtyBytes(Vec) through winit's unbounded proxy channel. Under a flood it ran ahead of the parser by the whole output (peak RSS 400-650 MB) and the main thread handled ~16k events per 64 MB, each with a fresh Vec. Now the reader appends to a buffer shared with the main thread's PtyReader and posts a payload-less PtyReady only when none is in flight. Past BUFFER_CAP (256 KiB) it stops reading, so the kernel PTY buffer fills and the child blocks. The main thread swaps the whole buffer out (no copy, no steady-state allocation) and hands it to the parser via peek/consume. The close is a flag in the same state, so it is reported only once every byte read before it has been consumed; Event::PtyBytes/PtyClosed become Event::PtyReady. A batch that ends mid synchronized update now always asks for a wake at the BSU watchdog deadline (previously only when the BSU began in that batch; otherwise it requested a redraw render_term would skip). --- crates/toastty-io/src/lib.rs | 573 +++++++++++++++++++++++--- crates/toastty-io/tests/common/mod.rs | 54 +++ crates/toastty-io/tests/pty_reader.rs | 52 ++- crates/toastty-io/tests/pty_writer.rs | 37 +- crates/toastty-window/src/event.rs | 12 +- crates/toastty-window/src/lib.rs | 21 +- crates/toastty/src/main.rs | 134 +++--- crates/toastty/tests/end_to_end.rs | 99 +++-- 8 files changed, 758 insertions(+), 224 deletions(-) create mode 100644 crates/toastty-io/tests/common/mod.rs diff --git a/crates/toastty-io/src/lib.rs b/crates/toastty-io/src/lib.rs index f3e9d0bd..ddbafb2c 100644 --- a/crates/toastty-io/src/lib.rs +++ b/crates/toastty-io/src/lib.rs @@ -1,12 +1,29 @@ //! mio + winit event-loop bridge. //! -//! Background mio thread polls the PTY fd and posts a [`UserEvent`] to -//! winit's event loop via [`EventLoopProxy::send_event`]. The render thread -//! receives those events as a winit `UserEvent`, advances the parser, and -//! requests a redraw. +//! A background mio thread reads the PTY master into a bounded buffer it +//! shares with the main thread, and wakes winit's event loop with +//! [`UserEvent::PtyReady`] via [`EventLoopProxy::send_event`]. The main +//! thread drains that buffer through its [`PtyReader`] at its own pace, +//! feeds the parser, and redraws. //! //! See [`docs/decisions/pty-event-loop.md`](../../docs/decisions/pty-event-loop.md). //! +//! ## Flow control +//! +//! Two properties matter under a flood (`cat` of a big file), where the +//! child writes far faster than the main thread can parse: +//! +//! - **At most one wake in flight.** The reader posts `PtyReady` only when +//! the main thread has taken everything since the last one. winit drains +//! every queued user event before it delivers `RedrawRequested`, so the +//! old one-event-per-4-KiB-read design starved redraws for the whole +//! flood (no frame until `cat` finished). +//! - **Bounded buffering.** Once [`BUFFER_CAP`] bytes are waiting, the +//! reader stops reading until the main thread takes them. The kernel +//! PTY buffer then fills and the child blocks in `write`, instead of the +//! reader queueing hundreds of MB the main thread will take seconds to +//! parse (and that a Ctrl-C can't cancel). +//! //! ## Dep graph //! //! This crate depends only on `mio`, `winit`, `libc`, and `tracing`. In @@ -21,6 +38,7 @@ use std::io::{self, ErrorKind, Read}; use std::os::fd::{AsRawFd, BorrowedFd, FromRawFd, OwnedFd}; use std::sync::mpsc::{self, Receiver, Sender}; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; use std::thread::{self, JoinHandle}; use mio::unix::SourceFd; @@ -32,24 +50,33 @@ use winit::event_loop::EventLoopProxy; /// /// The mio thread emits these via `EventLoopProxy::send_event(T::from(...))`, /// so the main thread can match on them inside its winit `user_event` callback. -#[derive(Debug)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum UserEvent { - /// Bytes read from the PTY master. Sized so each event carries a - /// reasonable chunk (4 KiB max per wake); does not allocate per byte. - PtyBytes(Vec), - /// PTY master closed (child exited / EIO). Main thread should - /// reap the child and exit. - PtyClosed, + /// The reader has buffered bytes, or seen the PTY close, since the + /// main thread last took its buffer. Drain it with + /// [`PtyReader::peek`] / [`PtyReader::consume`]; the close is + /// [`PtyReader::is_finished`]. Carries no payload: the bytes stay in + /// the shared buffer, and the next wake is posted only after the main + /// thread has taken them, so the event queue never holds more than one. + PtyReady, } /// Token for the PTY fd inside the mio poll set. const PTY_TOKEN: Token = Token(0); -/// Per-wake read buffer size. The mio thread reads in chunks of this -/// size, packages each chunk into a `PtyBytes` event, and posts it to the -/// proxy. 4 KiB matches the kernel pipe-buffer default and keeps each -/// allocation small. -const READ_BUF_SIZE: usize = 4096; +/// How many bytes the reader buffers ahead of the main thread before it +/// stops reading (it can overshoot by one [`READ_BUF_SIZE`] batch). Big +/// enough that the reader never starves the parser — it refills this in +/// well under a millisecond, the parser needs several to get through it — +/// and small enough that a Ctrl-C during a flood takes effect after a few +/// ms of already-read output, not after hundreds of MB. +pub const BUFFER_CAP: usize = 256 * 1024; + +/// Size of the reader thread's private read buffer. A Linux PTY read +/// returns at most ~4 KiB, so the reader keeps reading into this until it +/// is full or the fd runs dry, and only then takes the shared lock: one +/// lock + memcpy per 64 KiB instead of per read. +const READ_BUF_SIZE: usize = 64 * 1024; /// Returned when the event-loop destination is gone (proxy dropped, /// channel closed). Causes the reader thread to shut down. @@ -78,30 +105,178 @@ where } } -/// Spawn a background thread that polls `fd` with mio and posts -/// [`UserEvent::PtyBytes`] to `proxy` as bytes arrive. Returns a -/// `JoinHandle` so the binary can join on shutdown. +/// State shared by the reader thread and the main thread's [`PtyReader`]. +#[derive(Debug, Default)] +struct Shared { + state: Mutex, + /// Signalled when the main thread takes a full buffer, or detaches. + space: Condvar, +} + +#[derive(Debug, Default)] +struct State { + /// Bytes read but not yet taken by the main thread. + buf: Vec, + /// A `PtyReady` was posted and the main thread hasn't taken the + /// buffer since. While set, the reader appends without posting. + wake_pending: bool, + /// The reader hit EOF / an error and has exited; `buf` holds the + /// last bytes it will ever deliver. + closed: bool, + /// The [`PtyReader`] was dropped. A reader blocked on a full buffer + /// exits instead of waiting forever. + detached: bool, +} + +impl Shared { + fn lock(&self) -> MutexGuard<'_, State> { + // A panic while holding the lock leaves only a Vec and three + // flags behind, all still consistent; keep going. + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Reader side: block while [`BUFFER_CAP`] bytes are waiting. + /// Returns `false` if the main thread detached. + fn wait_for_space(&self) -> bool { + let mut st = self.lock(); + while st.buf.len() >= BUFFER_CAP && !st.detached { + st = self + .space + .wait(st) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + !st.detached + } + + /// Reader side: append `bytes`. Returns whether the caller must post + /// a `PtyReady` (none is in flight). + fn push(&self, bytes: &[u8]) -> bool { + let mut st = self.lock(); + st.buf.extend_from_slice(bytes); + !std::mem::replace(&mut st.wake_pending, true) + } + + /// Reader side: the PTY closed. Returns whether the caller must post + /// a `PtyReady` to report it. + fn close(&self) -> bool { + let mut st = self.lock(); + st.closed = true; + !std::mem::replace(&mut st.wake_pending, true) + } + + /// Main side: swap the shared buffer with `into` (which must be + /// empty) and re-arm the wake. Returns whether the PTY has closed. + fn take(&self, into: &mut Vec) -> bool { + debug_assert!(into.is_empty()); + let mut st = self.lock(); + std::mem::swap(&mut st.buf, into); + st.wake_pending = false; + let closed = st.closed; + drop(st); + // The reader only waits while the buffer is full, so only then is + // there anyone to wake (a futex wake is a syscall even without + // waiters). + if into.len() >= BUFFER_CAP { + self.space.notify_one(); + } + closed + } +} + +/// The main thread's end of the PTY reader: the bytes read so far, in +/// order. Returned by [`spawn_pty_reader`]. +/// +/// Bytes are taken from the reader thread a whole buffer at a time (a +/// swap, so no copy and, once both buffers have grown, no allocation); +/// [`peek`](Self::peek) / [`consume`](Self::consume) then hand them out +/// in pieces so the caller can parse with a time budget and stop at any +/// byte. +#[derive(Debug)] +pub struct PtyReader { + shared: Arc, + /// The batch being consumed; `buf[pos..]` is still unparsed. + buf: Vec, + pos: usize, + /// The reader thread reported close at the last take, so `buf` holds + /// the final bytes. + closed: bool, + /// Held, not joined: the thread may be parked in `poll` on an fd that + /// stays open until the process exits. + handle: Option>, +} + +impl PtyReader { + /// Up to `max` unconsumed bytes, oldest first, without consuming them. + /// Takes the reader thread's buffer when the current batch is used + /// up. Empty when nothing is buffered. + pub fn peek(&mut self, max: usize) -> &[u8] { + if self.pos == self.buf.len() && !self.closed { + self.buf.clear(); + self.pos = 0; + self.closed = self.shared.take(&mut self.buf); + } + let end = self.buf.len().min(self.pos.saturating_add(max)); + &self.buf[self.pos..end] + } + + /// Mark the first `n` bytes of the last [`peek`](Self::peek) as + /// consumed. + pub fn consume(&mut self, n: usize) { + debug_assert!(n <= self.buf.len() - self.pos); + self.pos = (self.pos + n).min(self.buf.len()); + } + + /// Whether unconsumed bytes are already in hand (not counting any + /// the reader thread buffered since the last take). + pub fn has_unconsumed(&self) -> bool { + self.pos < self.buf.len() + } + + /// The PTY has closed and every byte read before the close has been + /// consumed. Valid after a [`peek`](Self::peek) that returned empty. + pub fn is_finished(&self) -> bool { + self.closed && self.pos == self.buf.len() + } + + /// The reader thread's handle, for tests that wait for it to exit. + #[doc(hidden)] + pub fn take_join_handle(&mut self) -> Option> { + self.handle.take() + } +} + +impl Drop for PtyReader { + fn drop(&mut self) { + // Release a reader blocked on a full buffer so its thread exits. + self.shared.lock().detached = true; + self.shared.space.notify_all(); + } +} + +/// Spawn a background thread that polls `fd` with mio, buffers what it +/// reads, and posts [`UserEvent::PtyReady`] to `proxy` when there is +/// something new to take. Returns the main thread's [`PtyReader`]. /// /// The thread: /// - dups `fd` so it owns its own copy (caller keeps the original for /// `write` / `resize` / `try_wait`); /// - sets the duped fd non-blocking; /// - registers it with mio for `READABLE`; -/// - on wake, drains it into a 4 KiB buffer (one allocation per wake) -/// and posts `UserEvent::PtyBytes(bytes)`; -/// - exits cleanly when `fd` returns EOF / EIO (posts -/// [`UserEvent::PtyClosed`]) or when `proxy.send_event` fails (the -/// event loop has gone away). +/// - on wake, drains it into the shared buffer, stopping while +/// [`BUFFER_CAP`] bytes are waiting (see the module docs); +/// - exits cleanly when `fd` returns EOF / EIO (marks the reader closed +/// and wakes the loop), when `proxy.send_event` fails (the event loop +/// has gone away) or when the [`PtyReader`] is dropped while it waits +/// for space. /// /// # Errors /// /// Returns `io::Error` only if the initial setup fails (dup, fcntl, mio /// `Poll::new`, `register`). Once the thread is running, errors are /// logged via `tracing::warn` and the thread exits. -pub fn spawn_pty_reader( - fd: BorrowedFd<'_>, - proxy: EventLoopProxy, -) -> io::Result> +pub fn spawn_pty_reader(fd: BorrowedFd<'_>, proxy: EventLoopProxy) -> io::Result where T: From + Send + 'static, { @@ -114,7 +289,7 @@ where pub fn spawn_pty_reader_with_sink( fd: BorrowedFd<'_>, sink: S, -) -> io::Result> { +) -> io::Result { // Dup the fd up-front so the thread owns its own `OwnedFd`. We rely // on libc::dup; rustix's dup is similar but we already need libc for // fcntl below, so keep it consistent. @@ -138,11 +313,19 @@ pub fn spawn_pty_reader_with_sink( poll.registry() .register(&mut SourceFd(&source_fd), PTY_TOKEN, Interest::READABLE)?; + let shared = Arc::new(Shared::default()); + let thread_shared = Arc::clone(&shared); let handle = thread::Builder::new() .name("toastty-io::pty-reader".into()) - .spawn(move || run_loop(owned, poll, &sink))?; + .spawn(move || run_loop(owned, poll, &thread_shared, &sink))?; - Ok(handle) + Ok(PtyReader { + shared, + buf: Vec::new(), + pos: 0, + closed: false, + handle: Some(handle), + }) } /// Wrapper around `fcntl` to set `O_NONBLOCK`. @@ -161,22 +344,29 @@ fn set_nonblocking(fd: &OwnedFd) -> io::Result<()> { Ok(()) } -fn run_loop(owned: OwnedFd, mut poll: Poll, sink: &S) { +fn run_loop(owned: OwnedFd, mut poll: Poll, shared: &Shared, sink: &S) { // Wrap the OwnedFd in std::fs::File so we get `Read`. The fd is // already non-blocking; reads will return `WouldBlock` between // pipe-fills. let mut file = std::fs::File::from(owned); let mut events = Events::with_capacity(8); + let mut buf = vec![0u8; READ_BUF_SIZE].into_boxed_slice(); debug!("toastty-io::pty-reader started"); + let close = || { + if shared.close() { + let _ = sink.send(UserEvent::PtyReady); + } + }; + loop { if let Err(e) = poll.poll(&mut events, None) { if e.kind() == ErrorKind::Interrupted { continue; } warn!("pty-reader poll error: {e}; exiting"); - let _ = sink.send(UserEvent::PtyClosed); + close(); return; } @@ -184,23 +374,23 @@ fn run_loop(owned: OwnedFd, mut poll: Poll, sink: &S) { if ev.token() != PTY_TOKEN { continue; } - // Read in a loop until WouldBlock — mio is level-triggered by - // default, but draining each wake keeps p99 latency low and - // gives the parser the largest contiguous chunks possible. - match drain_into_proxy(&mut file, sink) { + // mio is edge-triggered: read until WouldBlock before polling + // again, or the rest of the data would never raise another + // event. + match drain(&mut file, &mut buf, shared, sink) { DrainResult::Continue => {} DrainResult::Closed => { - debug!("pty-reader EOF; sending PtyClosed"); - let _ = sink.send(UserEvent::PtyClosed); + debug!("pty-reader EOF; closing"); + close(); return; } - DrainResult::ProxyDropped => { - debug!("pty-reader proxy dropped; exiting"); + DrainResult::Detached => { + debug!("pty-reader consumer gone; exiting"); return; } DrainResult::Error(e) => { - warn!("pty-reader read error: {e}; sending PtyClosed and exiting"); - let _ = sink.send(UserEvent::PtyClosed); + warn!("pty-reader read error: {e}; closing and exiting"); + close(); return; } } @@ -212,37 +402,64 @@ fn run_loop(owned: OwnedFd, mut poll: Poll, sink: &S) { enum DrainResult { /// Read everything currently available; loop back to poll. Continue, - /// EOF (0 bytes returned). + /// EOF (0 bytes returned, or EIO). Closed, - /// `proxy.send_event` failed — the event loop has been dropped. - ProxyDropped, + /// The event loop or the [`PtyReader`] has gone away. + Detached, /// I/O error other than `WouldBlock` / `Interrupted`. Error(io::Error), } -fn drain_into_proxy(file: &mut std::fs::File, sink: &S) -> DrainResult { +/// Read `file` until it would block, handing the bytes to the main thread +/// a [`READ_BUF_SIZE`] batch at a time. +fn drain( + file: &mut std::fs::File, + buf: &mut [u8], + shared: &Shared, + sink: &S, +) -> DrainResult { + let mut filled = 0; loop { - let mut buf = vec![0u8; READ_BUF_SIZE]; - match file.read(&mut buf) { - Ok(0) => return DrainResult::Closed, + if filled == buf.len() && !flush(&buf[..filled], &mut filled, shared, sink) { + return DrainResult::Detached; + } + // Back-pressure: don't start another batch while the main thread + // is a full buffer behind. + if filled == 0 && !shared.wait_for_space() { + return DrainResult::Detached; + } + let outcome = match file.read(&mut buf[filled..]) { + Ok(0) => DrainResult::Closed, Ok(n) => { - buf.truncate(n); - trace!("pty-reader read {n} bytes"); - if sink.send(UserEvent::PtyBytes(buf)).is_err() { - return DrainResult::ProxyDropped; - } - } - Err(ref e) if e.kind() == ErrorKind::WouldBlock => { - return DrainResult::Continue; + filled += n; + continue; } - Err(ref e) if e.kind() == ErrorKind::Interrupted => {} + Err(ref e) if e.kind() == ErrorKind::WouldBlock => DrainResult::Continue, + Err(ref e) if e.kind() == ErrorKind::Interrupted => continue, // EIO on Linux when the PTY slave is closed; treat as EOF. - Err(e) if is_eio(&e) => return DrainResult::Closed, - Err(e) => return DrainResult::Error(e), + Err(e) if is_eio(&e) => DrainResult::Closed, + Err(e) => DrainResult::Error(e), + }; + // Whatever ends the drain, deliver what was read first. + if !flush(&buf[..filled], &mut filled, shared, sink) { + return DrainResult::Detached; } + return outcome; } } +/// Hand `bytes` (the reader's first `*filled` bytes) to the main thread +/// and post a wake if none is in flight. Returns `false` if the event +/// loop is gone. +fn flush(bytes: &[u8], filled: &mut usize, shared: &Shared, sink: &S) -> bool { + *filled = 0; + if bytes.is_empty() { + return true; + } + trace!("pty-reader buffered {} bytes", bytes.len()); + !shared.push(bytes) || sink.send(UserEvent::PtyReady).is_ok() +} + fn is_eio(e: &io::Error) -> bool { e.raw_os_error() == Some(libc::EIO) } @@ -446,11 +663,233 @@ mod tests { #[test] fn user_event_debug_renders() { - let ev = UserEvent::PtyBytes(b"hello".to_vec()); - let s = format!("{ev:?}"); - assert!(s.contains("PtyBytes")); - let ev = UserEvent::PtyClosed; - let s = format!("{ev:?}"); - assert!(s.contains("PtyClosed")); + let s = format!("{:?}", UserEvent::PtyReady); + assert!(s.contains("PtyReady")); + } + + // ---- shared buffer / back-pressure ---------------------------------- + + use std::io::Write as _; + use std::os::fd::AsFd; + use std::os::unix::net::UnixStream; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::{Duration, Instant}; + + /// Counts wakes and forwards them so a test can block on the next one. + struct CountingSink { + wakes: Arc, + tx: Mutex>, + } + + impl EventSink for CountingSink { + fn send(&self, _ev: UserEvent) -> Result<(), SinkClosed> { + self.wakes.fetch_add(1, Ordering::SeqCst); + let tx = self.tx.lock().map_err(|_| SinkClosed)?; + tx.send(()).map_err(|_| SinkClosed) + } + } + + struct Harness { + reader: PtyReader, + /// The child's end; dropping it closes the reader's side. + child: Option, + wakes: Arc, + rx: mpsc::Receiver<()>, + _ours: UnixStream, + } + + fn harness() -> Harness { + let (ours, child) = UnixStream::pair().expect("socketpair"); + child.set_nonblocking(true).expect("nonblocking"); + let wakes = Arc::new(AtomicUsize::new(0)); + let (tx, rx) = mpsc::channel(); + let sink = CountingSink { + wakes: Arc::clone(&wakes), + tx: Mutex::new(tx), + }; + let reader = spawn_pty_reader_with_sink(ours.as_fd(), sink).expect("spawn"); + Harness { + reader, + child: Some(child), + wakes, + rx, + _ours: ours, + } + } + + /// Write as much of `data` as the socket takes without blocking. + fn write_some(child: &mut UnixStream, data: &[u8]) -> usize { + let mut off = 0; + while off < data.len() { + match child.write(&data[off..]) { + Ok(n) => off += n, + Err(e) if e.kind() == ErrorKind::WouldBlock => break, + Err(e) => panic!("write: {e}"), + } + } + off + } + + /// Wait until `f` holds, failing the test after 5 s. + fn wait_until(mut f: impl FnMut() -> bool) { + let deadline = Instant::now() + Duration::from_secs(5); + while !f() { + assert!(Instant::now() < deadline, "timed out"); + thread::sleep(Duration::from_millis(2)); + } + } + + /// Everything currently buffered, consumed `step` bytes at a time. + fn drain_all(reader: &mut PtyReader, step: usize) -> Vec { + let mut out = Vec::new(); + loop { + let chunk = reader.peek(step); + if chunk.is_empty() { + return out; + } + assert!(chunk.len() <= step); + out.extend_from_slice(chunk); + let n = chunk.len(); + reader.consume(n); + } + } + + fn pattern(len: usize) -> Vec { + (0..len).map(|i| (i % 251) as u8).collect() + } + + #[test] + fn one_wake_per_take_however_many_reads() { + let mut h = harness(); + let child = h.child.as_mut().unwrap(); + // Many separate writes while the main thread takes nothing: the + // reader buffers them all behind a single wake. + for i in 0..50u8 { + child.write_all(&[i; 100]).unwrap(); + thread::sleep(Duration::from_micros(200)); + } + h.rx.recv_timeout(Duration::from_secs(5)) + .expect("first wake"); + wait_until(|| h.reader.shared.lock().buf.len() == 5000); + assert_eq!(h.wakes.load(Ordering::SeqCst), 1); + + let got = drain_all(&mut h.reader, 4096); + assert_eq!(got.len(), 5000); + assert!( + got.chunks(100) + .enumerate() + .all(|(i, c)| c.iter().all(|&b| usize::from(b) == i)) + ); + + // Taking the buffer re-arms the wake: the next write posts again. + h.child.as_mut().unwrap().write_all(b"more").unwrap(); + h.rx.recv_timeout(Duration::from_secs(5)) + .expect("second wake"); + assert_eq!(h.wakes.load(Ordering::SeqCst), 2); + assert_eq!(drain_all(&mut h.reader, 4096), b"more"); + } + + #[test] + fn reader_stops_at_the_cap_and_the_writer_blocks() { + let mut h = harness(); + let data = pattern(BUFFER_CAP * 4); + let child = h.child.as_mut().unwrap(); + let mut written = 0; + // Push until the socket stops taking bytes for a while: the reader + // has stopped reading, so the kernel buffer stays full. + let mut stalled_since = Instant::now(); + while stalled_since.elapsed() < Duration::from_millis(200) { + let n = write_some(child, &data[written..]); + if n > 0 { + written += n; + stalled_since = Instant::now(); + } + assert!( + written < data.len(), + "the reader never applied back-pressure" + ); + thread::sleep(Duration::from_millis(1)); + } + let buffered = h.reader.shared.lock().buf.len(); + assert!(buffered >= BUFFER_CAP, "stopped early: {buffered}"); + assert!( + buffered < BUFFER_CAP + READ_BUF_SIZE, + "overshot: {buffered}" + ); + assert_eq!(h.wakes.load(Ordering::SeqCst), 1); + + // Consuming frees space; the reader resumes and every byte arrives + // in order. + let mut got = Vec::new(); + while got.len() < data.len() { + written += write_some(h.child.as_mut().unwrap(), &data[written..]); + got.extend(drain_all(&mut h.reader, 10_000)); + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(got, data); + } + + #[test] + fn close_is_reported_only_after_every_byte_is_consumed() { + let mut h = harness(); + let data = pattern(100_000); + let mut child = h.child.take().unwrap(); + child.set_nonblocking(false).unwrap(); + let writer = thread::spawn(move || { + child.write_all(&data).unwrap(); + // Dropping `child` closes the socket: EOF on the reader's end. + }); + let mut got = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let chunk = h.reader.peek(7000); + if chunk.is_empty() { + if h.reader.is_finished() { + break; + } + assert!(Instant::now() < deadline, "no close"); + let _ = h.rx.recv_timeout(Duration::from_millis(50)); + continue; + } + got.extend_from_slice(chunk); + let n = chunk.len(); + h.reader.consume(n); + } + writer.join().unwrap(); + // Finished only once the last byte was handed out. + assert_eq!(got, pattern(100_000)); + } + + #[test] + fn partial_consume_keeps_the_rest_for_later() { + let mut h = harness(); + h.child.as_mut().unwrap().write_all(b"hello world").unwrap(); + h.rx.recv_timeout(Duration::from_secs(5)).expect("wake"); + wait_until(|| h.reader.shared.lock().buf.len() == 11); + assert_eq!(h.reader.peek(5), b"hello"); + h.reader.consume(5); + assert!(h.reader.has_unconsumed()); + assert_eq!(h.reader.peek(100), b" world"); + h.reader.consume(6); + assert!(!h.reader.has_unconsumed()); + assert!(h.reader.peek(100).is_empty()); + assert!(!h.reader.is_finished()); + } + + #[test] + fn dropping_the_reader_releases_a_thread_blocked_on_a_full_buffer() { + let mut h = harness(); + let data = pattern(BUFFER_CAP * 2); + let mut written = 0; + wait_until(|| { + written += write_some(h.child.as_mut().unwrap(), &data[written..]); + h.reader.shared.lock().buf.len() >= BUFFER_CAP + }); + let handle = h.reader.take_join_handle().unwrap(); + // The thread is parked in `wait_for_space`; the socket is still + // open, so only the detach can end it. + drop(h.reader); + wait_until(|| handle.is_finished()); + handle.join().unwrap(); } } diff --git a/crates/toastty-io/tests/common/mod.rs b/crates/toastty-io/tests/common/mod.rs new file mode 100644 index 00000000..bb31afbe --- /dev/null +++ b/crates/toastty-io/tests/common/mod.rs @@ -0,0 +1,54 @@ +//! Shared test glue: a channel-backed [`EventSink`] and a helper that +//! turns the reader's wake + shared-buffer API back into "the next batch +//! of bytes, or the close", which is what these tests want to assert on. + +use std::sync::Mutex; +use std::sync::mpsc; +use std::time::Duration; + +use toastty_io::{EventSink, PtyReader, SinkClosed, UserEvent}; + +/// Test sink: forwards `UserEvent` into an mpsc channel. +pub(crate) struct ChannelSink(pub(crate) Mutex>); + +impl EventSink for ChannelSink { + fn send(&self, ev: UserEvent) -> Result<(), SinkClosed> { + let g = self.0.lock().map_err(|_| SinkClosed)?; + g.send(ev).map_err(|_| SinkClosed) + } +} + +/// One step of what the reader delivered. +pub(crate) enum Batch { + Bytes(Vec), + Closed, +} + +/// Everything buffered right now, or the close once every byte is out. +/// Waits up to `timeout` for a wake when there is neither; `None` on +/// timeout. +pub(crate) fn recv_batch( + reader: &mut PtyReader, + rx: &mpsc::Receiver, + timeout: Duration, +) -> Option { + loop { + let mut out = Vec::new(); + loop { + let chunk = reader.peek(usize::MAX); + if chunk.is_empty() { + break; + } + out.extend_from_slice(chunk); + let n = chunk.len(); + reader.consume(n); + } + if !out.is_empty() { + return Some(Batch::Bytes(out)); + } + if reader.is_finished() { + return Some(Batch::Closed); + } + rx.recv_timeout(timeout).ok()?; + } +} diff --git a/crates/toastty-io/tests/pty_reader.rs b/crates/toastty-io/tests/pty_reader.rs index 7510c5c5..637f5deb 100644 --- a/crates/toastty-io/tests/pty_reader.rs +++ b/crates/toastty-io/tests/pty_reader.rs @@ -6,27 +6,23 @@ //! winit-side wiring is one trait impl and is exercised end-to-end by //! the binary's smoke run. +mod common; + use std::sync::Mutex; use std::sync::mpsc; use std::time::{Duration, Instant}; -use toastty_io::{EventSink, SinkClosed, UserEvent, spawn_pty_reader_with_sink}; +use common::{Batch, ChannelSink, recv_batch}; +use toastty_io::{PtyReader, UserEvent, spawn_pty_reader_with_sink}; use toastty_pty::{Pty, PtySpec}; -/// Test sink: forwards `UserEvent` into an mpsc channel. -struct ChannelSink(Mutex>); - -impl EventSink for ChannelSink { - fn send(&self, ev: UserEvent) -> Result<(), SinkClosed> { - let g = self.0.lock().map_err(|_| SinkClosed)?; - g.send(ev).map_err(|_| SinkClosed) - } -} - -/// Collect events from the channel for at most `max` time, returning all -/// `PtyBytes` payloads concatenated and a flag indicating whether -/// `PtyClosed` was seen. -fn collect(rx: &mpsc::Receiver, max: Duration) -> (Vec, bool) { +/// Collect output for at most `max` time, returning all bytes +/// concatenated and a flag indicating whether the close was seen. +fn collect( + reader: &mut PtyReader, + rx: &mpsc::Receiver, + max: Duration, +) -> (Vec, bool) { let deadline = Instant::now() + max; let mut bytes = Vec::new(); let mut closed = false; @@ -35,13 +31,13 @@ fn collect(rx: &mpsc::Receiver, max: Duration) -> (Vec, bool) { if now >= deadline { break; } - match rx.recv_timeout(deadline - now) { - Ok(UserEvent::PtyBytes(mut b)) => bytes.append(&mut b), - Ok(UserEvent::PtyClosed) => { + match recv_batch(reader, rx, deadline - now) { + Some(Batch::Bytes(mut b)) => bytes.append(&mut b), + Some(Batch::Closed) => { closed = true; break; } - Err(_) => break, + None => break, } } (bytes, closed) @@ -54,9 +50,9 @@ fn reads_echo_output_and_observes_close() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _join = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); - let (bytes, closed) = collect(&rx, Duration::from_secs(3)); + let (bytes, closed) = collect(&mut reader, &rx, Duration::from_secs(3)); let s = String::from_utf8_lossy(&bytes); assert!(s.contains("hello, toastty-io"), "got: {s:?}"); assert!(closed, "expected PtyClosed once child exits"); @@ -71,7 +67,7 @@ fn reads_multiple_chunks_from_cat() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _join = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); // Write a few separate lines; cat echoes them back. for line in &[b"alpha\n".as_slice(), b"beta\n", b"gamma\n"] { @@ -83,10 +79,10 @@ fn reads_multiple_chunks_from_cat() { let start = Instant::now(); let mut bytes = Vec::new(); while start.elapsed() < Duration::from_secs(2) { - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(UserEvent::PtyBytes(mut b)) => bytes.append(&mut b), - Ok(UserEvent::PtyClosed) | Err(mpsc::RecvTimeoutError::Disconnected) => break, - Err(mpsc::RecvTimeoutError::Timeout) => { + match recv_batch(&mut reader, &rx, Duration::from_millis(200)) { + Some(Batch::Bytes(mut b)) => bytes.append(&mut b), + Some(Batch::Closed) => break, + None => { if !bytes.is_empty() { // Got something + nothing new for 200 ms — done. break; @@ -110,9 +106,9 @@ fn reports_close_when_child_exits() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _join = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); - let (_bytes, closed) = collect(&rx, Duration::from_secs(3)); + let (_bytes, closed) = collect(&mut reader, &rx, Duration::from_secs(3)); assert!(closed, "expected PtyClosed when child exits"); drop(pty); } diff --git a/crates/toastty-io/tests/pty_writer.rs b/crates/toastty-io/tests/pty_writer.rs index 04ec309f..0f043da9 100644 --- a/crates/toastty-io/tests/pty_writer.rs +++ b/crates/toastty-io/tests/pty_writer.rs @@ -8,25 +8,17 @@ //! the foreground app in paste-collect mode. `PtyWriter` must deliver every //! byte in order regardless of ring size. +mod common; + use std::os::fd::{AsRawFd, BorrowedFd}; use std::sync::Mutex; use std::sync::mpsc; use std::time::{Duration, Instant}; -use toastty_io::{ - EventSink, SinkClosed, UserEvent, spawn_pty_reader_with_sink, spawn_pty_writer, -}; +use common::{Batch, ChannelSink, recv_batch}; +use toastty_io::{PtyReader, UserEvent, spawn_pty_reader_with_sink, spawn_pty_writer}; use toastty_pty::{Pty, PtySpec}; -struct ChannelSink(Mutex>); - -impl EventSink for ChannelSink { - fn send(&self, ev: UserEvent) -> Result<(), SinkClosed> { - let g = self.0.lock().map_err(|_| SinkClosed)?; - g.send(ev).map_err(|_| SinkClosed) - } -} - /// Put the tty behind `fd` into raw mode so bytes round-trip 1:1 through /// `cat` (no echo, no CR/LF translation) and the ~1 KiB input ring is the /// only bottleneck. @@ -47,7 +39,12 @@ fn make_raw(fd: BorrowedFd<'_>) { /// Collect echoed bytes from the reader channel until `want` bytes have /// arrived or `max` elapses. -fn collect_until(rx: &mpsc::Receiver, want: usize, max: Duration) -> Vec { +fn collect_until( + reader: &mut PtyReader, + rx: &mpsc::Receiver, + want: usize, + max: Duration, +) -> Vec { let deadline = Instant::now() + max; let mut bytes = Vec::new(); while bytes.len() < want { @@ -55,9 +52,9 @@ fn collect_until(rx: &mpsc::Receiver, want: usize, max: Duration) -> if now >= deadline { break; } - match rx.recv_timeout(deadline - now) { - Ok(UserEvent::PtyBytes(mut b)) => bytes.append(&mut b), - Ok(UserEvent::PtyClosed) | Err(_) => break, + match recv_batch(reader, rx, deadline - now) { + Some(Batch::Bytes(mut b)) => bytes.append(&mut b), + Some(Batch::Closed) | None => break, } } bytes @@ -84,7 +81,7 @@ fn large_paste_delivers_every_byte_including_terminator() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); let writer = spawn_pty_writer(pty.master_fd()).expect("spawn writer"); // 8 KiB: many times the ~1 KiB ring, so the writer parks on POLLOUT @@ -92,7 +89,7 @@ fn large_paste_delivers_every_byte_including_terminator() { let paste = bracketed_paste(8192); writer.send(&paste); - let got = collect_until(&rx, paste.len(), Duration::from_secs(5)); + let got = collect_until(&mut reader, &rx, paste.len(), Duration::from_secs(5)); assert_eq!( got.len(), @@ -119,7 +116,7 @@ fn writes_are_delivered_in_order() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("spawn reader"); let writer = spawn_pty_writer(pty.master_fd()).expect("spawn writer"); // Interleave a big (ring-spanning) send with small ones; order must be @@ -136,7 +133,7 @@ fn writes_are_delivered_in_order() { expected.extend_from_slice(chunk); } - let got = collect_until(&rx, expected.len(), Duration::from_secs(5)); + let got = collect_until(&mut reader, &rx, expected.len(), Duration::from_secs(5)); assert_eq!(got, expected, "bytes arrived out of order or incomplete"); drop(writer); diff --git a/crates/toastty-window/src/event.rs b/crates/toastty-window/src/event.rs index a59491da..6c092c0b 100644 --- a/crates/toastty-window/src/event.rs +++ b/crates/toastty-window/src/event.rs @@ -207,13 +207,11 @@ pub enum Event { /// A user-event posted via `WindowHandle::wake`. Used by the mio PTY /// thread to nudge the event loop. User, - /// PTY bytes arrived on the master fd (delivered via - /// `toastty_io::spawn_pty_reader`). The binary feeds these to the - /// parser and requests a redraw. - PtyBytes(Vec), - /// PTY master closed (child exited / EIO). The binary should reap - /// and exit. - PtyClosed, + /// The PTY reader (`toastty_io::spawn_pty_reader`) has new bytes + /// buffered, or saw the master close. The binary drains its + /// `toastty_io::PtyReader`, feeds the parser and requests a redraw; + /// once the reader reports finished it reaps the child and exits. + PtyReady, /// IME committed text — should be inserted into the editor (written to the PTY), /// the same as typed text. Arrives via the platform text-input protocol /// (e.g. fcitx5 Hangul/CJK composition), NOT through `Event::Key`. diff --git a/crates/toastty-window/src/lib.rs b/crates/toastty-window/src/lib.rs index a641cbe8..bdeac08d 100644 --- a/crates/toastty-window/src/lib.rs +++ b/crates/toastty-window/src/lib.rs @@ -90,26 +90,22 @@ pub enum WindowError { /// Custom user-event payload — sent via [`WindowHandle::wake`] from /// background threads (e.g. the mio PTY thread). /// -/// `PtyBytes` and `PtyClosed` correspond 1:1 to -/// [`toastty_io::UserEvent`]; the `From` impl below makes the -/// `spawn_pty_reader>` call site work -/// without any per-call conversion. +/// `PtyReady` corresponds 1:1 to [`toastty_io::UserEvent`]; the `From` +/// impl below makes the `spawn_pty_reader>` +/// call site work without any per-call conversion. #[derive(Debug)] pub enum UserEvent { /// Wake the event loop. Surfaces as [`Event::User`]. Wake, - /// PTY bytes ready, from `toastty-io::spawn_pty_reader`. Surfaces as - /// [`Event::PtyBytes`]. - PtyBytes(Vec), - /// PTY closed (child exited / EIO). Surfaces as [`Event::PtyClosed`]. - PtyClosed, + /// The PTY reader has bytes (or the close) for the main thread, from + /// `toastty-io::spawn_pty_reader`. Surfaces as [`Event::PtyReady`]. + PtyReady, } impl From for UserEvent { fn from(ev: toastty_io::UserEvent) -> Self { match ev { - toastty_io::UserEvent::PtyBytes(b) => UserEvent::PtyBytes(b), - toastty_io::UserEvent::PtyClosed => UserEvent::PtyClosed, + toastty_io::UserEvent::PtyReady => UserEvent::PtyReady, } } } @@ -559,8 +555,7 @@ impl ApplicationHandler for Runner<'_, A> { fn user_event(&mut self, event_loop: &ActiveEventLoop, ev: UserEvent) { let mapped = match ev { UserEvent::Wake => Event::User, - UserEvent::PtyBytes(b) => Event::PtyBytes(b), - UserEvent::PtyClosed => Event::PtyClosed, + UserEvent::PtyReady => Event::PtyReady, }; self.dispatch(mapped, event_loop); } diff --git a/crates/toastty/src/main.rs b/crates/toastty/src/main.rs index 774c4ceb..4ff66905 100644 --- a/crates/toastty/src/main.rs +++ b/crates/toastty/src/main.rs @@ -6,9 +6,9 @@ //! //! The render thread (winit main thread on macOS) owns the parser, //! `Term`, and `Renderer`. A background mio thread reads bytes from -//! the PTY master and posts them via `EventLoopProxy::send_event` as -//! `Event::PtyBytes(...)`; the binary feeds those to the parser and -//! requests a redraw. +//! the PTY master into a bounded buffer and wakes the loop with +//! `Event::PtyReady`; the binary drains that buffer into the parser +//! and requests a redraw. #![deny(unsafe_op_in_unsafe_fn)] @@ -279,9 +279,10 @@ struct Toastty { parser: Parser, term: Term, pty: Option, - /// `JoinHandle` for the mio reader; held so we don't drop it. The - /// thread exits on its own when the PTY closes or the proxy goes away. - reader: Option>, + /// The main thread's end of the mio reader: the PTY output read so + /// far, drained on `Event::PtyReady`. The thread exits on its own when + /// the PTY closes or the proxy goes away. + reader: Option, /// Background writer that drains queued bytes to the PTY master in /// order. Routing writes through here (instead of a direct /// `Pty::write`) guarantees the whole payload lands even when the @@ -307,7 +308,7 @@ struct Toastty { /// other reload keeps it. See [`toastty::font_zoom`]. font_zoom: FontZoom, /// Last title we pushed to `winit::Window::set_title`. Tracked so - /// we don't churn the compositor on every `PtyBytes` batch — only + /// we don't churn the compositor on every `PtyReady` batch — only /// change the title when it actually differs. last_title: Option, /// Most recent mouse pixel position; tracked so motion events that @@ -945,13 +946,73 @@ impl Toastty { } } - /// Feed a batch of PTY bytes through the parser. Returns whether a - /// fresh BSU (mode 2026 begin-synchronized-update) just went high - /// during this batch — the caller uses that signal to schedule the - /// watchdog redraw via `ControlSignal::RedrawIn(BSU_TIMEOUT)`. - fn handle_pty_bytes(&mut self, bytes: &[u8]) -> bool { + /// Drain the PTY reader into the parser (`Event::PtyReady`). Bytes + /// the reader buffers past its cap wait in the kernel's buffer, which + /// blocks the child: memory stays flat however fast it writes. + fn handle_pty_ready(&mut self) -> ControlSignal { + let Some(mut reader) = self.reader.take() else { + return ControlSignal::Continue; + }; + let mut parsed = false; + loop { + let avail = reader.peek(usize::MAX); + if avail.is_empty() { + break; + } + let now = Instant::now(); + let n = avail.len(); + self.handle_pty_bytes(avail); + self.stats.pty_batch(now, n, now.elapsed()); + reader.consume(n); + parsed = true; + } + let finished = reader.is_finished(); + self.reader = Some(reader); + if finished { + // Every byte read before the close is parsed by now. + return self.handle_pty_closed(); + } + if !parsed { + // A wake for bytes an earlier pass already took. + return ControlSignal::Continue; + } + // New output may have rewritten or moved the hovered link; + // re-resolve it (clears when it's gone). + if self.term.link_hover().is_some() { + self.refresh_link_hover(); + } + if self.term.pause_rendering() { + // Mid synchronized update: `render_term` would just skip, so + // instead of redrawing now, wake at the BSU watchdog deadline. + // `handle_pty_bytes` fires the watchdog when more bytes + // arrive after it; this covers BSU-then-silence. + let left = self.term.sync_output_started_at().map_or(BSU_TIMEOUT, |t| { + (t + BSU_TIMEOUT).saturating_duration_since(Instant::now()) + }); + return ControlSignal::RedrawIn(left); + } + // RedrawIn(ZERO) — not Continue — to force the event loop to + // wake immediately. On macOS, plain `request_redraw()` doesn't + // wake from Wait reliably; RedrawIn sets ControlFlow::Poll, + // which does. + ControlSignal::RedrawIn(Duration::ZERO) + } + + /// The PTY closed and its last bytes are parsed: reap and exit. + fn handle_pty_closed(&mut self) -> ControlSignal { + debug!("pty closed; exiting"); + self.stats.write(); + toastty::perf_stats::finish_profiler(); + if let Some(pty) = self.pty.as_mut() { + let _ = pty.wait(); + } + ControlSignal::Exit + } + + /// Feed one slice of PTY bytes through the parser, then service what + /// it queued (title, clipboard, replies, BSU watchdog). + fn handle_pty_bytes(&mut self, bytes: &[u8]) { self.pty_log.log(Direction::FromApp, bytes); - let was_paused = self.term.pause_rendering(); // Refresh the foreground-process CWD before parsing this // batch — if an RGP `r;path=...` arrives in these bytes, // `Term::register_asset_by_path` needs to know the current @@ -1006,11 +1067,6 @@ impl Toastty { w.request_redraw(); } } - - // A fresh BSU started in this batch — caller will schedule a - // wake-up at BSU_TIMEOUT so the watchdog above eventually - // fires even if the app sends no more bytes. - !was_paused && self.term.pause_rendering() } } @@ -1164,7 +1220,7 @@ impl Toastty { // Spawn the mio reader thread. It posts user-events via the // winit `EventLoopProxy`, which arrive in `App::event` as - // `Event::PtyBytes` / `Event::PtyClosed`. + // `Event::PtyReady`. let proxy = handle.event_loop_proxy(); let reader = match toastty_io::spawn_pty_reader(pty.master_fd(), proxy) { Ok(j) => j, @@ -1370,7 +1426,7 @@ impl Toastty { // guaranteed: the loop's RedrawIn handler issues the // `request_redraw` + `ControlFlow::Poll` wake, which a bare // `request_redraw()` from `Wait` doesn't do reliably on macOS - // (same reasoning as the `Event::PtyBytes` path). + // (same reasoning as the `Event::PtyReady` path). ControlSignal::RedrawIn(Duration::ZERO) } else { self.close_now() @@ -2937,43 +2993,7 @@ impl App for Toastty { signal } } - Event::PtyBytes(bytes) => { - let at = Instant::now(); - let fresh_bsu = self.handle_pty_bytes(&bytes); - self.stats.pty_batch(at, bytes.len(), at.elapsed()); - // New output may have rewritten or moved the hovered - // link; re-resolve it (clears when it's gone). - if self.term.link_hover().is_some() { - self.refresh_link_hover(); - } - // If a BSU just went high we still want to wake the - // event loop, but at the watchdog deadline rather than - // immediately — `render_term` would just skip while the - // pause is active. The watchdog inside - // `handle_pty_bytes` already fires if the previous BSU - // had already expired; this branch covers the - // "BSU-then-silence" case where no further bytes - // arrive to drive the watchdog. - // - // RedrawIn(ZERO) — not Continue — to force the event - // loop to wake immediately. On macOS, plain - // `request_redraw()` doesn't wake from Wait reliably; - // RedrawIn sets ControlFlow::WaitUntil(now) which does. - if fresh_bsu { - ControlSignal::RedrawIn(BSU_TIMEOUT) - } else { - ControlSignal::RedrawIn(Duration::ZERO) - } - } - Event::PtyClosed => { - debug!("pty closed; exiting"); - self.stats.write(); - toastty::perf_stats::finish_profiler(); - if let Some(pty) = self.pty.as_mut() { - let _ = pty.wait(); - } - ControlSignal::Exit - } + Event::PtyReady => self.handle_pty_ready(), Event::User => { // The config watcher wakes the loop via Wake on file // changes. Drain its pending flag (atomic swap) and diff --git a/crates/toastty/tests/end_to_end.rs b/crates/toastty/tests/end_to_end.rs index d23e56a9..2ac7cb25 100644 --- a/crates/toastty/tests/end_to_end.rs +++ b/crates/toastty/tests/end_to_end.rs @@ -1,14 +1,14 @@ //! End-to-end (sans-renderer) integration: PTY → mio reader → parser → Term. //! //! Builds the same data pipeline `main.rs` does without the wgpu/winit -//! halves. Exercises the `Vec` payload path between +//! halves. Exercises the shared-buffer path between //! `toastty-io::spawn_pty_reader_with_sink` and the binary's parser. use std::sync::Mutex; use std::sync::mpsc; use std::time::{Duration, Instant}; -use toastty_io::{EventSink, SinkClosed, UserEvent, spawn_pty_reader_with_sink}; +use toastty_io::{EventSink, PtyReader, SinkClosed, UserEvent, spawn_pty_reader_with_sink}; use toastty_parser::Parser; use toastty_pty::{Pty, PtySpec, WinSize}; use toastty_term::Term; @@ -27,6 +27,41 @@ impl EventSink for ChannelSink { } } +/// One step of what the reader delivered. +enum Batch { + Bytes(Vec), + Closed, +} + +/// Everything buffered right now, or the close once every byte is out. +/// Waits up to `timeout` for a wake when there is neither; `None` on +/// timeout. +fn recv_batch( + reader: &mut PtyReader, + rx: &mpsc::Receiver, + timeout: Duration, +) -> Option { + loop { + let mut out = Vec::new(); + loop { + let chunk = reader.peek(usize::MAX); + if chunk.is_empty() { + break; + } + out.extend_from_slice(chunk); + let n = chunk.len(); + reader.consume(n); + } + if !out.is_empty() { + return Some(Batch::Bytes(out)); + } + if reader.is_finished() { + return Some(Batch::Closed); + } + rx.recv_timeout(timeout).ok()?; + } +} + #[test] fn pty_echo_drives_term_through_parser() { // Spawn `echo hello` under a PTY at a known grid size. @@ -42,7 +77,7 @@ fn pty_echo_drives_term_through_parser() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); let mut parser = Parser::new(); let mut term = Term::new(24, 80, 0); @@ -50,10 +85,10 @@ fn pty_echo_drives_term_through_parser() { let deadline = Instant::now() + Duration::from_secs(3); let mut closed = false; while !closed && Instant::now() < deadline { - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(UserEvent::PtyBytes(b)) => parser.advance(&mut term, &b), - Ok(UserEvent::PtyClosed) => closed = true, - Err(_) => {} + match recv_batch(&mut reader, &rx, Duration::from_millis(200)) { + Some(Batch::Bytes(b)) => parser.advance(&mut term, &b), + Some(Batch::Closed) => closed = true, + None => {} } } assert!(closed, "expected PtyClosed event"); @@ -91,7 +126,7 @@ fn keyboard_encoder_round_trips_through_cat() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); // Encode "hi\r" the way the binary would. let bytes_h = encode_key( @@ -130,8 +165,8 @@ fn keyboard_encoder_round_trips_through_cat() { let deadline = Instant::now() + Duration::from_secs(2); let mut total = Vec::new(); while Instant::now() < deadline { - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(UserEvent::PtyBytes(b)) => { + match recv_batch(&mut reader, &rx, Duration::from_millis(200)) { + Some(Batch::Bytes(b)) => { total.extend_from_slice(&b); parser.advance(&mut term, &b); let s = String::from_utf8_lossy(&total); @@ -139,8 +174,8 @@ fn keyboard_encoder_round_trips_through_cat() { break; } } - Ok(UserEvent::PtyClosed) => break, - Err(_) => {} + Some(Batch::Closed) => break, + None => {} } } let s = String::from_utf8_lossy(&total); @@ -162,7 +197,7 @@ fn resize_propagates_through_pty_and_term() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); // Resize using the same helper main.rs uses. let (cols, rows) = grid_dims_from_pixels(1280, 800, 10.0, 20.0); @@ -184,13 +219,13 @@ fn resize_propagates_through_pty_and_term() { let mut closed = false; let mut all = Vec::new(); while !closed && Instant::now() < deadline { - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(UserEvent::PtyBytes(b)) => { + match recv_batch(&mut reader, &rx, Duration::from_millis(200)) { + Some(Batch::Bytes(b)) => { all.extend_from_slice(&b); parser.advance(&mut term, &b); } - Ok(UserEvent::PtyClosed) => closed = true, - Err(_) => {} + Some(Batch::Closed) => closed = true, + None => {} } } let s = String::from_utf8_lossy(&all); @@ -218,18 +253,18 @@ fn pty_drives_bracketed_paste_mode_through_term() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); let mut parser = Parser::new(); let mut term = Term::new(24, 80, 0); let deadline = Instant::now() + Duration::from_secs(3); while Instant::now() < deadline { - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(UserEvent::PtyBytes(b)) => { + match recv_batch(&mut reader, &rx, Duration::from_millis(200)) { + Some(Batch::Bytes(b)) => { parser.advance(&mut term, &b); } - Ok(UserEvent::PtyClosed) => break, - Err(_) => {} + Some(Batch::Closed) => break, + None => {} } if term.bracketed_paste() { break; @@ -257,18 +292,18 @@ fn pty_drives_focus_reporting_mode_through_term() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); let mut parser = Parser::new(); let mut term = Term::new(24, 80, 0); let deadline = Instant::now() + Duration::from_secs(3); while Instant::now() < deadline { - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(UserEvent::PtyBytes(b)) => { + match recv_batch(&mut reader, &rx, Duration::from_millis(200)) { + Some(Batch::Bytes(b)) => { parser.advance(&mut term, &b); } - Ok(UserEvent::PtyClosed) => break, - Err(_) => {} + Some(Batch::Closed) => break, + None => {} } if term.report_focus() { break; @@ -302,7 +337,7 @@ fn shift_letter_under_kitty_disambiguate_writes_plain_uppercase_to_pty() { let (tx, rx) = mpsc::channel(); let sink = ChannelSink(Mutex::new(tx)); - let _reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); + let mut reader = spawn_pty_reader_with_sink(pty.master_fd(), sink).expect("reader"); // Flag bit 1 == disambiguate-only, the same value zellij pushes. let kitty_flags = 1u8; @@ -347,8 +382,8 @@ fn shift_letter_under_kitty_disambiguate_writes_plain_uppercase_to_pty() { let deadline = Instant::now() + Duration::from_secs(2); let mut total = Vec::new(); while Instant::now() < deadline { - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(UserEvent::PtyBytes(b)) => { + match recv_batch(&mut reader, &rx, Duration::from_millis(200)) { + Some(Batch::Bytes(b)) => { total.extend_from_slice(&b); parser.advance(&mut term, &b); let s = String::from_utf8_lossy(&total); @@ -356,8 +391,8 @@ fn shift_letter_under_kitty_disambiguate_writes_plain_uppercase_to_pty() { break; } } - Ok(UserEvent::PtyClosed) => break, - Err(_) => {} + Some(Batch::Closed) => break, + None => {} } } let s = String::from_utf8_lossy(&total); From 3444b42a576c0263d751bfc094c076631ae57951 Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 01:55:10 -0700 Subject: [PATCH 02/10] ingest: parse floods with a frame budget and yield to the redraw Parsing everything the reader had buffered before each frame drew nothing until a flood ended: winit hands out every queued user event before RedrawRequested, and the handler never ran out of bytes. handle_pty_ready now parses in 16 KiB slices. Output that drains before a frame interval passes is drawn at once as before. When output keeps coming it stops once a frame is due (16.7 ms after the last frame started, and after parsing at least as long as that frame took) and answers the new ControlSignal::Yield. The window runner redraws, drops reader wakes until about_to_wait (which runs after the iteration's redraws on X11, Wayland and macOS) and there delivers PtyReady again: one parse slice plus one frame per loop iteration. A check at the top of the handler covers output arriving about as fast as it parses. Mode 2026: never yields while rendering is paused; once a frame is due mid update, parses only up to the next ESU so the frame lands between updates even when ESU and the next BSU are back to back. The pacing policy lives in toastty::ingest with unit tests; the design is recorded in docs/decisions/pty-event-loop.md. --- Cargo.lock | 1 + crates/toastty-window/src/event.rs | 12 ++ crates/toastty-window/src/lib.rs | 46 ++++++- crates/toastty/Cargo.toml | 1 + crates/toastty/src/ingest.rs | 191 +++++++++++++++++++++++++++++ crates/toastty/src/lib.rs | 1 + crates/toastty/src/main.rs | 51 ++++++-- docs/decisions/pty-event-loop.md | 58 +++++++++ 8 files changed, 350 insertions(+), 11 deletions(-) create mode 100644 crates/toastty/src/ingest.rs diff --git a/Cargo.lock b/Cargo.lock index f2cd3044..ed523d86 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4997,6 +4997,7 @@ dependencies = [ "base64 0.22.1", "icy_sixel", "image", + "memchr", "notify", "pollster", "pprof", diff --git a/crates/toastty-window/src/event.rs b/crates/toastty-window/src/event.rs index 6c092c0b..15eea3ee 100644 --- a/crates/toastty-window/src/event.rs +++ b/crates/toastty-window/src/event.rs @@ -211,6 +211,9 @@ pub enum Event { /// buffered, or saw the master close. The binary drains its /// `toastty_io::PtyReader`, feeds the parser and requests a redraw; /// once the reader reports finished it reaps the child and exits. + /// + /// Also delivered after a [`ControlSignal::Yield`] frame, to resume + /// parsing where the binary left off. PtyReady, /// IME committed text — should be inserted into the editor (written to the PTY), /// the same as typed text. Arrives via the platform text-input protocol @@ -238,4 +241,13 @@ pub enum ControlSignal { Exit, /// Schedule a redraw at most `Duration` from now. RedrawIn(Duration), + /// Redraw now, like `RedrawIn(Duration::ZERO)`, and deliver + /// [`Event::PtyReady`] again once that frame is out. Returned from + /// `PtyReady` when the app stopped parsing with bytes left so a frame + /// could be drawn (flood output). Reader wakes that arrive before the + /// frame are dropped: the resumed `PtyReady` drains everything + /// buffered anyway, and delivering them would parse again before the + /// frame (winit hands out every queued user event before it gets to + /// `RedrawRequested`). + Yield, } diff --git a/crates/toastty-window/src/lib.rs b/crates/toastty-window/src/lib.rs index bdeac08d..73b37c4e 100644 --- a/crates/toastty-window/src/lib.rs +++ b/crates/toastty-window/src/lib.rs @@ -262,6 +262,7 @@ pub fn run(options: WindowOptions, mut app: A) -> Result<(), WindowError app: &mut app, last_status: ControlSignal::Continue, redraw_at: None, + resume_pty: false, }; event_loop.set_control_flow(ControlFlow::Wait); @@ -316,6 +317,8 @@ fn schedule(sig: ControlSignal, is_redraw: bool, redraw_at: Option, now redraw_at, } } + // Yield redraws now exactly like RedrawIn(0); the resumed + // PtyReady is the Runner's business (see `Runner::resume_pty`). // Duration::ZERO means "wake immediately" — under // ControlFlow::WaitUntil(now), winit's EventLoopWaker re-arms // the CFRunLoopTimer every iteration because Instant::now() @@ -324,6 +327,11 @@ fn schedule(sig: ControlSignal, is_redraw: bool, redraw_at: Option, now // CFRunLoopTimerSetNextFireDate. ControlFlow::Poll dedups // against a fixed past instant inside EventLoopWaker::start, so // the arm happens once and subsequent iterations short-circuit. + ControlSignal::Yield => Plan { + flow: Flow::Poll, + redraw_now: true, + redraw_at, + }, ControlSignal::RedrawIn(d) if d.is_zero() => Plan { flow: Flow::Poll, redraw_now: true, @@ -356,6 +364,11 @@ struct Runner<'a, A: App> { /// When the next scheduled redraw is due (from a /// `ControlSignal::RedrawIn`), if any. redraw_at: Option, + /// The app answered `ControlSignal::Yield`: it stopped parsing PTY + /// output so a frame could be drawn. Until `about_to_wait` (which + /// every platform runs after the iteration's `RedrawRequested`) + /// delivers the resuming `Event::PtyReady`, reader wakes are dropped. + resume_pty: bool, } impl Runner<'_, A> { @@ -363,6 +376,9 @@ impl Runner<'_, A> { let is_redraw = matches!(ev, Event::Redraw); let sig = self.app.event(ev); self.last_status = sig; + if sig == ControlSignal::Yield { + self.resume_pty = true; + } self.apply_control(sig, is_redraw, event_loop); } @@ -555,6 +571,8 @@ impl ApplicationHandler for Runner<'_, A> { fn user_event(&mut self, event_loop: &ActiveEventLoop, ev: UserEvent) { let mapped = match ev { UserEvent::Wake => Event::User, + // Covered by the resume in `about_to_wait`, after the frame. + UserEvent::PtyReady if self.resume_pty => return, UserEvent::PtyReady => Event::PtyReady, }; self.dispatch(mapped, event_loop); @@ -572,8 +590,18 @@ impl ApplicationHandler for Runner<'_, A> { } } - fn about_to_wait(&mut self, _event_loop: &ActiveEventLoop) { - // No-op. We rely on `ControlFlow` set in `apply_control`. + fn about_to_wait(&mut self, event_loop: &ActiveEventLoop) { + // Resume PTY parsing after a `Yield`. winit runs this after the + // iteration's user events and redraws on X11, Wayland and macOS + // alike, so the frame the app yielded for is out by now. Calling + // the app directly (rather than posting a wake) keeps it to one + // parse slice per iteration without a channel round trip; the + // `Poll` flow `Yield` set means the loop doesn't sleep before the + // next frame. Otherwise we rely on the `ControlFlow` set in + // `apply_control`. + if std::mem::take(&mut self.resume_pty) { + self.dispatch(Event::PtyReady, event_loop); + } } } @@ -1044,6 +1072,20 @@ mod tests { ); } + #[test] + fn yield_redraws_at_once_and_keeps_the_deadline() { + let now = Instant::now(); + let pending = Some(now + ms(300)); + assert_eq!( + schedule(ControlSignal::Yield, false, pending, now), + schedule(ControlSignal::RedrawIn(ms(0)), false, pending, now), + ); + let plan = schedule(ControlSignal::Yield, false, pending, now); + assert_eq!(plan.flow, Flow::Poll); + assert!(plan.redraw_now); + assert_eq!(plan.redraw_at, pending); + } + #[test] fn control_signal_variants_are_distinct() { assert_ne!(ControlSignal::Continue, ControlSignal::Exit); diff --git a/crates/toastty/Cargo.toml b/crates/toastty/Cargo.toml index 564573b8..3caa1dd0 100644 --- a/crates/toastty/Cargo.toml +++ b/crates/toastty/Cargo.toml @@ -17,6 +17,7 @@ path = "src/main.rs" anyhow.workspace = true base64 = "0.22" icy_sixel = "0.5" +memchr.workspace = true image = { version = "0.25", default-features = false, features = ["png", "jpeg", "gif", "webp", "bmp"] } # `wayland-data-control` makes arboard use the native Wayland clipboard # (wl_data_control) when WAYLAND_DISPLAY is set, falling back to X11 diff --git a/crates/toastty/src/ingest.rs b/crates/toastty/src/ingest.rs new file mode 100644 index 00000000..4b830cfb --- /dev/null +++ b/crates/toastty/src/ingest.rs @@ -0,0 +1,191 @@ +//! PTY ingest pacing: how much buffered output to parse before letting a +//! frame through. +//! +//! The parser, `Term` and renderer all live on the main thread, so under a +//! flood (`cat` of a big file) every millisecond spent parsing is one not +//! spent drawing. Parsing everything the reader has buffered before each +//! frame drew nothing at all until the flood ended. Instead the +//! `Event::PtyReady` handler parses in slices and, once a frame is due, +//! stops and answers `ControlSignal::Yield`: the loop draws, then resumes +//! parsing. Small writes (typing, prompts) are parsed in one go and drawn +//! at once, exactly as before; only output that outlasts a frame interval +//! is paced. +//! +//! See `docs/decisions/pty-event-loop.md` (2026-10-22 section). + +use std::sync::LazyLock; +use std::time::{Duration, Instant}; + +use memchr::memmem; + +/// Most bytes parsed between two "is a frame due?" checks. At the +/// slowest parse rates seen (~13 MB/s, `seq`-style short lines) this is +/// ~1.3 ms, fine enough to keep frames on time; big enough that the +/// per-slice work in `handle_pty_bytes` stays noise. +pub const SLICE: usize = 16 * 1024; + +/// Target frame period while output floods in (~60 Hz). The display rate +/// isn't known here (winit only reports it per monitor, and it can change +/// as the window moves); 60 Hz is the common floor and keeps the render +/// share of a flood small. +pub const FRAME_INTERVAL: Duration = Duration::from_micros(16_667); + +/// Least time to parse before yielding again, so a flood always makes +/// progress. The longer of this and the last frame's render time is +/// used, which keeps parsing at >= half the main thread when frames are +/// slow (e.g. a huge window). +const MIN_PARSE: Duration = Duration::from_millis(1); + +/// How long a requested frame may hold up parsing. Redraws arrive within +/// the same loop iteration on every platform winit supports, so this only +/// guards against one that never comes: parsing then carries on as if no +/// frame were pending. +const MAX_FRAME_WAIT: Duration = Duration::from_millis(100); + +/// Mode 2026 end-synchronized-update, `CSI ? 2026 l`. +const ESU: &[u8] = b"\x1b[?2026l"; + +static ESU_FINDER: LazyLock> = LazyLock::new(|| memmem::Finder::new(ESU)); + +/// Pacing state for the PTY ingest. One per window, owned by the app. +#[derive(Debug, Default)] +pub struct Ingest { + /// Start of the last frame that was actually drawn. + last_frame: Option, + /// How long that frame took to render. + last_render: Duration, + /// When the ingest asked for a redraw that hasn't run yet. + frame_requested: Option, +} + +impl Ingest { + /// A frame was drawn between `start` and `end`. + pub fn frame_drawn(&mut self, start: Instant, end: Instant) { + self.last_frame = Some(start); + self.last_render = end.saturating_duration_since(start); + } + + /// The redraw handler ran (drawn or skipped). + pub fn redraw_handled(&mut self) { + self.frame_requested = None; + } + + /// The ingest answered with a redraw request at `now`. + pub fn frame_requested(&mut self, now: Instant) { + self.frame_requested.get_or_insert(now); + } + + fn frame_due(&self, now: Instant) -> bool { + self.last_frame + .is_none_or(|t| now.saturating_duration_since(t) >= FRAME_INTERVAL) + } + + /// At the top of `PtyReady`: a redraw the ingest asked for is still + /// queued and a frame is due, so yield without parsing. Without this, + /// output arriving about as fast as it parses would deliver a fresh + /// wake each time the handler drained the buffer, and winit hands out + /// every queued user event before `RedrawRequested`. + #[must_use] + pub fn should_wait_for_frame(&self, now: Instant) -> bool { + self.frame_requested + .is_some_and(|at| now.saturating_duration_since(at) < MAX_FRAME_WAIT) + && self.frame_due(now) + } + + /// Between slices: stop parsing (begun at `parse_start`) to draw a + /// frame. The caller also holds off while mode 2026 pauses rendering, + /// since that frame would be skipped. + #[must_use] + pub fn should_yield(&self, now: Instant, parse_start: Instant) -> bool { + let min_parse = self.last_render.clamp(MIN_PARSE, FRAME_INTERVAL); + self.frame_due(now) && now.saturating_duration_since(parse_start) >= min_parse + } +} + +/// How much of `avail` to parse next once a frame is due but mode 2026 +/// is holding it back: up to and including the first ESU, if there is +/// one. Ending the slice right after the ESU lets the frame go out before +/// the next synchronized update begins (apps often send ESU and the next +/// BSU back to back, so a slice boundary anywhere else would almost +/// always land inside an update). +#[must_use] +pub fn slice_to_esu(avail: &[u8]) -> usize { + ESU_FINDER + .find(avail) + .map_or(avail.len(), |i| i + ESU.len()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ms(n: u64) -> Duration { + Duration::from_millis(n) + } + + #[test] + fn the_first_frame_is_due_at_once() { + let ing = Ingest::default(); + let now = Instant::now(); + // Still parses for MIN_PARSE first. + assert!(!ing.should_yield(now, now)); + assert!(ing.should_yield(now + ms(1), now)); + } + + #[test] + fn yields_a_frame_interval_after_the_last_frame() { + let mut ing = Ingest::default(); + let t0 = Instant::now(); + ing.frame_drawn(t0, t0 + ms(2)); + let parse_start = t0 + ms(2); + assert!(!ing.should_yield(t0 + ms(10), parse_start)); + assert!(!ing.should_yield(t0 + ms(16), parse_start)); + assert!(ing.should_yield(t0 + ms(17), parse_start)); + } + + #[test] + fn slow_frames_get_at_least_as_much_parse_time() { + let mut ing = Ingest::default(); + let t0 = Instant::now(); + // A 30 ms frame: one is due as soon as it ends, but parse for as + // long as it took first. + ing.frame_drawn(t0, t0 + ms(30)); + let parse_start = t0 + ms(30); + assert!(!ing.should_yield(t0 + ms(40), parse_start)); + // Capped at one frame interval. + assert!(ing.should_yield(parse_start + FRAME_INTERVAL, parse_start)); + } + + #[test] + fn waits_for_a_requested_frame_only_while_one_is_due_and_pending() { + let mut ing = Ingest::default(); + let t0 = Instant::now(); + ing.frame_drawn(t0, t0 + ms(1)); + assert!(!ing.should_wait_for_frame(t0 + ms(20)), "nothing requested"); + ing.frame_requested(t0 + ms(5)); + assert!(!ing.should_wait_for_frame(t0 + ms(6)), "not due yet"); + assert!(ing.should_wait_for_frame(t0 + ms(20))); + // A redraw that never comes stops holding parsing up. + assert!(!ing.should_wait_for_frame(t0 + ms(5) + MAX_FRAME_WAIT)); + ing.redraw_handled(); + assert!(!ing.should_wait_for_frame(t0 + ms(20))); + } + + #[test] + fn frame_requested_keeps_the_earliest_request() { + let mut ing = Ingest::default(); + let t0 = Instant::now(); + ing.frame_requested(t0); + ing.frame_requested(t0 + ms(50)); + assert!(!ing.should_wait_for_frame(t0 + MAX_FRAME_WAIT)); + } + + #[test] + fn slice_to_esu_ends_after_the_first_esu() { + let s = b"\x1b[?2026hAAA\x1b[?2026l\x1b[?2026hBBB\x1b[?2026l"; + let n = slice_to_esu(s); + assert_eq!(&s[..n], b"\x1b[?2026hAAA\x1b[?2026l"); + assert_eq!(slice_to_esu(b"no esu here"), 11); + assert_eq!(slice_to_esu(b""), 0); + } +} diff --git a/crates/toastty/src/lib.rs b/crates/toastty/src/lib.rs index 29cff98c..9a96e64d 100644 --- a/crates/toastty/src/lib.rs +++ b/crates/toastty/src/lib.rs @@ -17,6 +17,7 @@ pub mod focus; pub mod font_zoom; pub mod geometry; pub mod image_cmd; +pub mod ingest; pub mod keyboard; pub mod links; pub mod markdown; diff --git a/crates/toastty/src/main.rs b/crates/toastty/src/main.rs index 4ff66905..3cad560d 100644 --- a/crates/toastty/src/main.rs +++ b/crates/toastty/src/main.rs @@ -8,7 +8,7 @@ //! `Term`, and `Renderer`. A background mio thread reads bytes from //! the PTY master into a bounded buffer and wakes the loop with //! `Event::PtyReady`; the binary drains that buffer into the parser -//! and requests a redraw. +//! (paced by `toastty::ingest` during floods) and requests a redraw. #![deny(unsafe_op_in_unsafe_fn)] @@ -42,6 +42,7 @@ use toastty::font_zoom::{FontZoom, ZoomAction, zoom_action}; use toastty::geometry::{ content_rect_from_padding, effective_font_size_px, grid_dims_from_pixels, scaled_pads, }; +use toastty::ingest::{self, Ingest}; use toastty::keyboard::{self, encode_key_with_modes}; use toastty::mouse::{ MouseEventKind, classify_button_event, drag_autoscroll_dir, encode_mouse, pixel_to_cell, @@ -283,6 +284,8 @@ struct Toastty { /// far, drained on `Event::PtyReady`. The thread exits on its own when /// the PTY closes or the proxy goes away. reader: Option, + /// Paces `reader` draining against frames during floods. + ingest: Ingest, /// Background writer that drains queued bytes to the PTY master in /// order. Routing writes through here (instead of a direct /// `Pty::write`) guarantees the whole payload lands even when the @@ -494,6 +497,7 @@ impl Toastty { term, pty: None, reader: None, + ingest: Ingest::default(), pty_writer: None, physical_size: initial_size, // Real value is read from the window in `init_impl`; 1.0 is a @@ -946,22 +950,43 @@ impl Toastty { } } - /// Drain the PTY reader into the parser (`Event::PtyReady`). Bytes - /// the reader buffers past its cap wait in the kernel's buffer, which - /// blocks the child: memory stays flat however fast it writes. + /// Drain the PTY reader into the parser (`Event::PtyReady`). + /// + /// Parses in slices of at most [`ingest::SLICE`]. When output keeps + /// coming for longer than a frame interval, stops once a frame is due + /// and answers `Yield`, so the loop draws and then calls back here + /// (see `toastty::ingest`). Bytes the reader buffers meanwhile wait + /// in its bounded buffer, and past that in the kernel's, which blocks + /// the child: memory stays flat however fast it writes. fn handle_pty_ready(&mut self) -> ControlSignal { + let parse_start = Instant::now(); + if self.ingest.should_wait_for_frame(parse_start) { + return ControlSignal::Yield; + } let Some(mut reader) = self.reader.take() else { return ControlSignal::Continue; }; let mut parsed = false; + let mut yielded = false; loop { - let avail = reader.peek(usize::MAX); + let now = Instant::now(); + let frame_due = self.ingest.should_yield(now, parse_start); + if frame_due && !self.term.pause_rendering() { + yielded = true; + break; + } + let avail = reader.peek(ingest::SLICE); if avail.is_empty() { break; } - let now = Instant::now(); - let n = avail.len(); - self.handle_pty_bytes(avail); + // A frame is due but a synchronized update holds it: parse + // only to the end of that update, then check again. + let n = if frame_due { + ingest::slice_to_esu(avail) + } else { + avail.len() + }; + self.handle_pty_bytes(&avail[..n]); self.stats.pty_batch(now, n, now.elapsed()); reader.consume(n); parsed = true; @@ -981,6 +1006,10 @@ impl Toastty { if self.term.link_hover().is_some() { self.refresh_link_hover(); } + if yielded { + self.ingest.frame_requested(parse_start); + return ControlSignal::Yield; + } if self.term.pause_rendering() { // Mid synchronized update: `render_term` would just skip, so // instead of redrawing now, wake at the BSU watchdog deadline. @@ -995,6 +1024,7 @@ impl Toastty { // wake immediately. On macOS, plain `request_redraw()` doesn't // wake from Wait reliably; RedrawIn sets ControlFlow::Poll, // which does. + self.ingest.frame_requested(parse_start); ControlSignal::RedrawIn(Duration::ZERO) } @@ -2826,11 +2856,14 @@ impl App for Toastty { self.term.sync_rich_blocks(); self.refresh_rich_chip(); let mut next_blink: Option = None; + self.ingest.redraw_handled(); if let Some(r) = self.renderer.as_mut() { let frame_start = Instant::now(); match r.render_term(&mut self.term) { Ok(RenderOutcome::Rendered) => { - self.stats.frame(frame_start, Instant::now()); + let frame_end = Instant::now(); + self.stats.frame(frame_start, frame_end); + self.ingest.frame_drawn(frame_start, frame_end); // Consume the per-cell damage signal so the // next render only re-emits cells that // changed. Clear the BSU timeout-flush diff --git a/docs/decisions/pty-event-loop.md b/docs/decisions/pty-event-loop.md index a6ee57fe..27eb5234 100644 --- a/docs/decisions/pty-event-loop.md +++ b/docs/decisions/pty-event-loop.md @@ -249,3 +249,61 @@ That's the whole pattern. The rest of the variant-C file is PTY setup, CPU measu - **Streaming APC parser.** Variant C feeds vte 0.15 directly, which buffers entire APC payloads. Kitty graphics and RGP can carry MiB of data; the architecture doc already flags this as an open question. Decision belongs in its own RFC. - **Backpressure on render-thread overload.** If `parser.advance` ever becomes slow (it isn't today), we need to stop the read loop draining into vte and let the kernel pipe absorb. mio's level-triggered readiness means the poll will simply keep firing, which is fine — but if we adopt `Interest::READABLE` with edge-triggered mode we need an explicit re-arm. Document at integration time. - **Windows ConPTY** integration. Per architecture doc this is a v2 milestone — recommendation here does not block on it. mio's `NamedPipe` is the integration point when we get there. + +## Update 2026-10-22: bounded buffer, one wake in flight, paced ingest + +What shipped differed from the sketch above: the mio thread posted every +4 KiB read as its own `UserEvent::PtyBytes(Vec)` through winit's +unbounded proxy channel, and the main thread parsed each one in full. Under +a flood (`cat` of a 64 MB file) that had three costs: + +- **No frames.** winit hands out every queued user event before it gets to + `RedrawRequested` (X11 and macOS drain the channel with `try_recv`), and + the reader refilled the queue faster than the parser emptied it, so not + one frame was drawn until the flood ended. +- **Unbounded memory.** The reader ran ahead of the parser by the whole + file: peak RSS 400–650 MB, and a Ctrl-C only took effect after toastty + had parsed everything already queued. +- **Per-event overhead.** ~16k events per 64 MB, each with a `Vec` + allocation and a run of per-batch work in `handle_pty_bytes`, including a + foreground-CWD probe (`tcgetpgrp` + `readlink /proc//cwd`) that + cost ~8 % of flood CPU. + +The follow-up "Backpressure on render-thread overload" above is what this +section does. + +**Reader → main thread** (`toastty-io`). The reader appends to a buffer +shared with the main thread's `PtyReader` (a `Mutex>`; the main +thread swaps the whole `Vec` out, so there is no copy and, in steady state, +no allocation). It posts a payload-less `UserEvent::PtyReady` only when no +wake is already in flight, so the queue never holds more than one. Once +`BUFFER_CAP` (256 KiB) is waiting it stops reading until the main thread +takes it; the kernel PTY buffer then fills and the child blocks in +`write`. The close is a flag in the same state, so it is seen only after +every byte read before it has been consumed. The reader collects up to +64 KiB of (≤ 4 KiB, on Linux) reads before it takes the lock. mio is +edge-triggered, so the reader still drains to `WouldBlock` before polling +again; while it waits for space it isn't polling at all. + +**Main thread** (`crates/toastty/src/ingest.rs`, `handle_pty_ready` in +`main.rs`). `PtyReady` parses in slices of at most 16 KiB. Output that is +drained before a frame interval passes is drawn at once, as before +(`RedrawIn(ZERO)`; small writes see no added latency). When output keeps +coming, the handler stops once a frame is due (16.7 ms after the last +frame started, and after parsing at least as long as that frame took to +render) and answers the new `ControlSignal::Yield`. The `toastty-window` +runner treats it as `RedrawIn(ZERO)`, drops reader wakes until +`about_to_wait`, and there delivers `Event::PtyReady` again. Every +platform winit supports runs `about_to_wait` after that iteration's +redraws, so each loop iteration is one parse slice plus one frame. A +second check at the top of the handler (a redraw it requested is still +queued and a frame is due) covers output that arrives about as fast as it +parses, where each pass drains the buffer and a fresh wake follows +immediately. + +Mode 2026 is unchanged: the handler never yields while rendering is +paused. Once a frame is due during a synchronized update it parses only up +to the next ESU (`CSI ? 2026 l`), so the frame goes out between updates +even when the app sends ESU and the next BSU back to back. A batch that +ends mid-update asks for a wake at the BSU watchdog deadline instead of a +redraw that `render_term` would skip. From 8a59f27bb2b7519795b96fb72c60ff7860e712cb Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 01:55:21 -0700 Subject: [PATCH 03/10] main: probe the foreground CWD only for slices that open an APC handle_pty_bytes ran tcgetpgrp + readlink /proc//cwd (and built a String) for every PTY batch, ~8% of flood CPU. The CWD's only reader is RGP's r;path= registration, which is an APC, so probe only when the slice contains ESC _ (tracking an ESC that ends the previous slice). The probe still runs before the slice is parsed, so a relative path is resolved against the CWD of the process that is emitting it. --- crates/toastty/src/ingest.rs | 32 ++++++++++++++++++++++++++++++++ crates/toastty/src/main.rs | 11 +++++++---- docs/decisions/pty-event-loop.md | 4 ++++ 3 files changed, 43 insertions(+), 4 deletions(-) diff --git a/crates/toastty/src/ingest.rs b/crates/toastty/src/ingest.rs index 4b830cfb..1f9ab50a 100644 --- a/crates/toastty/src/ingest.rs +++ b/crates/toastty/src/ingest.rs @@ -56,6 +56,9 @@ pub struct Ingest { last_render: Duration, /// When the ingest asked for a redraw that hasn't run yet. frame_requested: Option, + /// The last slice ended in ESC, so a `_` opening the next one starts + /// an APC. + trailing_esc: bool, } impl Ingest { @@ -100,6 +103,19 @@ impl Ingest { let min_parse = self.last_render.clamp(MIN_PARSE, FRAME_INTERVAL); self.frame_due(now) && now.saturating_duration_since(parse_start) >= min_parse } + + /// Whether `bytes` (the next slice, in stream order) opens an APC: + /// `ESC _`, possibly split across slices. RGP's `r;path=` registration + /// is an APC and resolves relative paths against the foreground + /// process's CWD; probing that CWD costs an ioctl plus a `readlink`, + /// so `handle_pty_bytes` does it only for slices where this is true. + pub fn opens_apc(&mut self, bytes: &[u8]) -> bool { + let Some(&last) = bytes.last() else { + return false; + }; + let split = std::mem::replace(&mut self.trailing_esc, last == 0x1B) && bytes[0] == b'_'; + split || memmem::find(bytes, b"\x1b_").is_some() + } } /// How much of `avail` to parse next once a frame is due but mode 2026 @@ -188,4 +204,20 @@ mod tests { assert_eq!(slice_to_esu(b"no esu here"), 11); assert_eq!(slice_to_esu(b""), 0); } + + #[test] + fn opens_apc_finds_esc_underscore_even_across_slices() { + let mut ing = Ingest::default(); + assert!(!ing.opens_apc(b"plain output\r\n")); + assert!(ing.opens_apc(b"x\x1b_Gr;path=a.glb\x1b\\")); + // Split between ESC and `_`. + assert!(!ing.opens_apc(b"abc\x1b")); + assert!(ing.opens_apc(b"_r;path=a\x1b\\")); + // ESC at the end of one slice, something else next. + assert!(!ing.opens_apc(b"\x1b")); + assert!(!ing.opens_apc(b"[0m")); + // tmux passthrough doubles the ESC; still contains `ESC _`. + assert!(ing.opens_apc(b"\x1bPtmux;\x1b\x1b_Ga=q\x1b\x1b\\\x1b\\")); + assert!(!ing.opens_apc(b"")); + } } diff --git a/crates/toastty/src/main.rs b/crates/toastty/src/main.rs index 3cad560d..ad21fc42 100644 --- a/crates/toastty/src/main.rs +++ b/crates/toastty/src/main.rs @@ -1043,12 +1043,15 @@ impl Toastty { /// it queued (title, clipboard, replies, BSU watchdog). fn handle_pty_bytes(&mut self, bytes: &[u8]) { self.pty_log.log(Direction::FromApp, bytes); - // Refresh the foreground-process CWD before parsing this - // batch — if an RGP `r;path=...` arrives in these bytes, + // Refresh the foreground-process CWD before parsing a slice that + // opens an APC — if it is an RGP `r;path=...`, // `Term::register_asset_by_path` needs to know the current // shell-side CWD to resolve relative paths. macOS uses - // `proc_pidinfo`; Linux uses `/proc//cwd`. - if let Some(pty) = self.pty.as_ref() + // `proc_pidinfo`; Linux uses `/proc//cwd`. Nothing else + // reads it, and the probe (an ioctl plus a readlink) was ~8% of + // flood CPU when it ran on every read. + if self.ingest.opens_apc(bytes) + && let Some(pty) = self.pty.as_ref() && let Some(cwd) = toastty_pty::pty_foreground_cwd(pty.master_fd()) { let cwd_str = cwd.to_string_lossy(); diff --git a/docs/decisions/pty-event-loop.md b/docs/decisions/pty-event-loop.md index 27eb5234..71826bd3 100644 --- a/docs/decisions/pty-event-loop.md +++ b/docs/decisions/pty-event-loop.md @@ -307,3 +307,7 @@ to the next ESU (`CSI ? 2026 l`), so the frame goes out between updates even when the app sends ESU and the next BSU back to back. A batch that ends mid-update asks for a wake at the BSU watchdog deadline instead of a redraw that `render_term` would skip. + +The CWD probe now runs only for slices that open an APC (`ESC _`, also +across a slice boundary): RGP's `r;path=` registration, the only reader of +the CWD, is an APC. From 965b652cc06b406bab72bb5be47aa02f8ca8bb56 Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 02:07:27 -0700 Subject: [PATCH 04/10] ingest: give slow frames twice their render time in parsing With the parse floor equal to the last frame's render time, a flood of frames that each cost ~10 ms (CJK text, whose glyphs are new to the atlas) spent half the main thread rendering: unicode throughput fell to ~0.45x of no-frames. Parse at least twice the render time (capped at 33 ms), so rendering takes at most a third; frames up to ~5 ms keep the 60 Hz cadence. --- crates/toastty/src/ingest.rs | 33 ++++++++++++++++++++++---------- docs/decisions/pty-event-loop.md | 5 +++-- 2 files changed, 26 insertions(+), 12 deletions(-) diff --git a/crates/toastty/src/ingest.rs b/crates/toastty/src/ingest.rs index 1f9ab50a..00ee01b2 100644 --- a/crates/toastty/src/ingest.rs +++ b/crates/toastty/src/ingest.rs @@ -31,11 +31,17 @@ pub const SLICE: usize = 16 * 1024; pub const FRAME_INTERVAL: Duration = Duration::from_micros(16_667); /// Least time to parse before yielding again, so a flood always makes -/// progress. The longer of this and the last frame's render time is -/// used, which keeps parsing at >= half the main thread when frames are -/// slow (e.g. a huge window). +/// progress. The longer of this and twice the last frame's render time +/// is used (capped at [`MAX_MIN_PARSE`]): when frames are slow (a huge +/// window, or a frame full of glyphs new to the atlas) the frame rate +/// drops rather than rendering eating more than a third of the main +/// thread. Frames of up to ~5 ms still go out at 60 Hz. const MIN_PARSE: Duration = Duration::from_millis(1); +/// Upper bound on the render-scaled parse time, so even very slow frames +/// come at least every ~33 ms of parsing. +const MAX_MIN_PARSE: Duration = Duration::from_micros(33_333); + /// How long a requested frame may hold up parsing. Redraws arrive within /// the same loop iteration on every platform winit supports, so this only /// guards against one that never comes: parsing then carries on as if no @@ -100,7 +106,7 @@ impl Ingest { /// since that frame would be skipped. #[must_use] pub fn should_yield(&self, now: Instant, parse_start: Instant) -> bool { - let min_parse = self.last_render.clamp(MIN_PARSE, FRAME_INTERVAL); + let min_parse = (self.last_render * 2).clamp(MIN_PARSE, MAX_MIN_PARSE); self.frame_due(now) && now.saturating_duration_since(parse_start) >= min_parse } @@ -160,16 +166,23 @@ mod tests { } #[test] - fn slow_frames_get_at_least_as_much_parse_time() { + fn slow_frames_get_twice_their_time_in_parsing() { let mut ing = Ingest::default(); let t0 = Instant::now(); - // A 30 ms frame: one is due as soon as it ends, but parse for as - // long as it took first. + // A 10 ms frame: one is due 6.7 ms after it ends, but parse for + // twice as long as it took first. + ing.frame_drawn(t0, t0 + ms(10)); + let parse_start = t0 + ms(10); + assert!(!ing.should_yield(parse_start + ms(19), parse_start)); + assert!(ing.should_yield(parse_start + ms(20), parse_start)); + // Fast frames don't change the 60 Hz cadence. + ing.frame_drawn(t0, t0 + ms(5)); + assert!(ing.should_yield(t0 + ms(17), t0 + ms(5))); + // Capped: a 30 ms frame gets 33 ms, not 60. ing.frame_drawn(t0, t0 + ms(30)); let parse_start = t0 + ms(30); - assert!(!ing.should_yield(t0 + ms(40), parse_start)); - // Capped at one frame interval. - assert!(ing.should_yield(parse_start + FRAME_INTERVAL, parse_start)); + assert!(!ing.should_yield(parse_start + ms(33), parse_start)); + assert!(ing.should_yield(parse_start + MAX_MIN_PARSE, parse_start)); } #[test] diff --git a/docs/decisions/pty-event-loop.md b/docs/decisions/pty-event-loop.md index 71826bd3..158720fd 100644 --- a/docs/decisions/pty-event-loop.md +++ b/docs/decisions/pty-event-loop.md @@ -290,8 +290,9 @@ again; while it waits for space it isn't polling at all. drained before a frame interval passes is drawn at once, as before (`RedrawIn(ZERO)`; small writes see no added latency). When output keeps coming, the handler stops once a frame is due (16.7 ms after the last -frame started, and after parsing at least as long as that frame took to -render) and answers the new `ControlSignal::Yield`. The `toastty-window` +frame started, and after parsing at least twice as long as that frame +took to render, so slow frames lower the frame rate rather than take more +than a third of the main thread) and answers the new `ControlSignal::Yield`. The `toastty-window` runner treats it as `RedrawIn(ZERO)`, drops reader wakes until `about_to_wait`, and there delivers `Event::PtyReady` again. Every platform winit supports runs `about_to_wait` after that iteration's From df653a62f2168d99cd3f8807057f868d2d542ec9 Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 02:11:42 -0700 Subject: [PATCH 05/10] docs: pty-event-loop measurements for the bounded, paced ingest --- docs/decisions/pty-event-loop.md | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/docs/decisions/pty-event-loop.md b/docs/decisions/pty-event-loop.md index 158720fd..dc739c6c 100644 --- a/docs/decisions/pty-event-loop.md +++ b/docs/decisions/pty-event-loop.md @@ -312,3 +312,21 @@ redraw that `render_term` would skip. The CWD probe now runs only for slices that open an APC (`ESC _`, also across a slice boundary): RGP's `r;path=` registration, the only reader of the CWD, is an APC. + +Measured with `bench/run.py` (Xvfb, 128×42 cells), interleaved single +runs of the old and new binaries, median of 4 each, on a shared and busy +machine (absolute MB/s swing ±40 % between runs; compare within a row): + +| workload | MB/s old → new | frames/s during flood | peak RSS MB | +|---|---|---|---| +| `cat` 64 MB ascii | 22 → 39 | 0 → 60 | 336 → 104 | +| `cat` 32 MB SGR | 34 → 56 | 0 → 60 | 194 → 103 | +| `cat` 16 MB unicode | 17 → 19 | 0 → 51 | 161 → 106 | +| `cat` 23 MB `seq` | 10 → 21 | 0 → 59 | 289 → 103 | +| `dense_cells` (mode 2026 frames) | 64 → 100 | 0 → 58 | 124 → 104 | + +Key → frame latency was unchanged (p50 0.89 → 0.63 ms); idle RSS is +~98–102 MB. Output → frame during a flood is now one frame interval +(~17 ms) instead of the length of the flood. Unicode is the case where +drawing frames costs throughput: its frames take ~6 ms (vs < 1 ms for +ASCII), so the render share is close to the one-third cap. From 00a6a359d99e8d2572dc61ec31da3326d3f191a5 Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 02:26:48 -0700 Subject: [PATCH 06/10] ingest: never stop a PtyReady pass before its first slice If a frame was already due when the loop first checked (the thread was descheduled for >= min_parse after the pass started), it stopped without peeking, and the 'nothing parsed' check answered Continue before the yield could. Nothing had taken the reader's buffer, so its wake stayed armed and it never posted PtyReady again: output froze for good (reviewer repro: one 40 ms stall froze `seq 1 3000000`). The loop's decisions move into ingest::Ingest::step and ingest::outcome, pure and unit-tested: step never stops before a slice is parsed (that slice also takes the buffer and re-arms the wake), and outcome answers any stop with Yield whatever was parsed. Also time the slice after peek() so a buffer swap isn't counted as parse time. --- crates/toastty/src/ingest.rs | 125 +++++++++++++++++++++++++++++++++-- crates/toastty/src/main.rs | 86 +++++++++++++----------- 2 files changed, 168 insertions(+), 43 deletions(-) diff --git a/crates/toastty/src/ingest.rs b/crates/toastty/src/ingest.rs index 00ee01b2..8eaef446 100644 --- a/crates/toastty/src/ingest.rs +++ b/crates/toastty/src/ingest.rs @@ -101,11 +101,37 @@ impl Ingest { && self.frame_due(now) } - /// Between slices: stop parsing (begun at `parse_start`) to draw a - /// frame. The caller also holds off while mode 2026 pauses rendering, - /// since that frame would be skipped. + /// What the `PtyReady` loop does next, at `now`, in a pass that began + /// at `parse_start` and has (`any_parsed`) or hasn't parsed a slice yet, + /// while mode 2026 does (`sync_paused`) or doesn't pause rendering. + /// + /// Never stops before the first slice. Besides guaranteeing progress + /// when the thread was descheduled between the pass starting and this + /// check, that first slice is what takes the reader's buffer and + /// re-arms its wake; see [`outcome`] for the other half. #[must_use] - pub fn should_yield(&self, now: Instant, parse_start: Instant) -> bool { + pub fn step( + &self, + now: Instant, + parse_start: Instant, + any_parsed: bool, + sync_paused: bool, + ) -> Step { + let frame_due = self.should_yield(now, parse_start); + if !frame_due { + Step::Parse + } else if any_parsed && !sync_paused { + Step::Stop(Stop::Yield) + } else { + // Due, but a synchronized update holds the frame back (or + // nothing is parsed yet): parse to the end of the update. + Step::ParseToEsu + } + } + + /// Whether a frame is due and the pass begun at `parse_start` has + /// parsed long enough to let it through. + fn should_yield(&self, now: Instant, parse_start: Instant) -> bool { let min_parse = (self.last_render * 2).clamp(MIN_PARSE, MAX_MIN_PARSE); self.frame_due(now) && now.saturating_duration_since(parse_start) >= min_parse } @@ -124,6 +150,59 @@ impl Ingest { } } +/// The next move of the `PtyReady` parse loop ([`Ingest::step`]). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Step { + /// Parse the next slice, up to [`SLICE`] bytes. + Parse, + /// Parse the next slice only up to the end of its first ESU + /// ([`slice_to_esu`]). + ParseToEsu, + /// Leave the loop with bytes still buffered. + Stop(Stop), +} + +/// Why the parse loop left with bytes still buffered. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Stop { + /// To draw a frame: answer `ControlSignal::Yield`. + Yield, +} + +/// What a finished `PtyReady` pass answers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + /// The PTY closed and every byte is parsed: reap and exit. + Closed, + /// Nothing new to parse (a wake for bytes an earlier pass took). + Idle, + /// Redraw, then resume parsing (`ControlSignal::Yield`). + Yield, + /// Drained mid synchronized update: wake at the BSU watchdog deadline + /// rather than redraw a frame `render_term` would skip. + AwaitEsu, + /// Drained: redraw now. + Redraw, +} + +/// The answer to a `PtyReady` pass that ended with `stop` (or by draining +/// the reader when `None`). +/// +/// A stop must always be answered with one that schedules a resume, even +/// if nothing was parsed: the stop left bytes in the reader, whose wake +/// stays armed until they are taken, so no further `PtyReady` would +/// arrive and output would freeze. +#[must_use] +pub fn outcome(finished: bool, any_parsed: bool, stop: Option, sync_paused: bool) -> Outcome { + match stop { + _ if finished => Outcome::Closed, + Some(Stop::Yield) => Outcome::Yield, + None if !any_parsed => Outcome::Idle, + None if sync_paused => Outcome::AwaitEsu, + None => Outcome::Redraw, + } +} + /// How much of `avail` to parse next once a frame is due but mode 2026 /// is holding it back: up to and including the first ESU, if there is /// one. Ending the slice right after the ESU lets the frame go out before @@ -209,6 +288,44 @@ mod tests { assert!(!ing.should_wait_for_frame(t0 + MAX_FRAME_WAIT)); } + #[test] + fn a_frame_due_on_entry_still_parses_a_slice_first() { + // Regression: the thread was descheduled between the pass + // starting and the first check, so a frame was already due. The + // loop stopped without peeking, the pass answered "idle", and the + // reader, whose wake was still armed, never woke us again. + let ing = Ingest::default(); + let parse_start = Instant::now(); + let late = parse_start + ms(40); + assert_eq!(ing.step(late, parse_start, false, false), Step::ParseToEsu); + assert_eq!( + ing.step(late, parse_start, true, false), + Step::Stop(Stop::Yield) + ); + // And a stop is never answered as idle, parsed or not. + assert_eq!( + outcome(false, false, Some(Stop::Yield), false), + Outcome::Yield + ); + } + + #[test] + fn step_parses_to_the_esu_while_the_frame_is_held() { + let ing = Ingest::default(); + let t0 = Instant::now(); + assert_eq!(ing.step(t0, t0, true, true), Step::Parse); + assert_eq!(ing.step(t0 + ms(5), t0, true, true), Step::ParseToEsu); + } + + #[test] + fn outcome_after_draining() { + assert_eq!(outcome(true, true, None, false), Outcome::Closed); + assert_eq!(outcome(true, false, None, true), Outcome::Closed); + assert_eq!(outcome(false, false, None, false), Outcome::Idle); + assert_eq!(outcome(false, true, None, true), Outcome::AwaitEsu); + assert_eq!(outcome(false, true, None, false), Outcome::Redraw); + } + #[test] fn slice_to_esu_ends_after_the_first_esu() { let s = b"\x1b[?2026hAAA\x1b[?2026l\x1b[?2026hBBB\x1b[?2026l"; diff --git a/crates/toastty/src/main.rs b/crates/toastty/src/main.rs index ad21fc42..79631e38 100644 --- a/crates/toastty/src/main.rs +++ b/crates/toastty/src/main.rs @@ -961,18 +961,26 @@ impl Toastty { fn handle_pty_ready(&mut self) -> ControlSignal { let parse_start = Instant::now(); if self.ingest.should_wait_for_frame(parse_start) { + // Nothing taken from the reader, so its wake stays armed and + // it won't post again; the runner's `about_to_wait` resume + // that `Yield` schedules is what brings us back. return ControlSignal::Yield; } + // No reader (spawn failed): there is nothing to wake us anyway. let Some(mut reader) = self.reader.take() else { return ControlSignal::Continue; }; - let mut parsed = false; - let mut yielded = false; + let mut any_parsed = false; + let mut stopped = None; loop { - let now = Instant::now(); - let frame_due = self.ingest.should_yield(now, parse_start); - if frame_due && !self.term.pause_rendering() { - yielded = true; + let step = self.ingest.step( + Instant::now(), + parse_start, + any_parsed, + self.term.pause_rendering(), + ); + if let ingest::Step::Stop(why) = step { + stopped = Some(why); break; } let avail = reader.peek(ingest::SLICE); @@ -981,51 +989,51 @@ impl Toastty { } // A frame is due but a synchronized update holds it: parse // only to the end of that update, then check again. - let n = if frame_due { + let n = if step == ingest::Step::ParseToEsu { ingest::slice_to_esu(avail) } else { avail.len() }; + // After `peek`, so a buffer swap doesn't count as parsing. + let at = Instant::now(); self.handle_pty_bytes(&avail[..n]); - self.stats.pty_batch(now, n, now.elapsed()); + self.stats.pty_batch(at, n, at.elapsed()); reader.consume(n); - parsed = true; + any_parsed = true; } let finished = reader.is_finished(); self.reader = Some(reader); - if finished { - // Every byte read before the close is parsed by now. - return self.handle_pty_closed(); - } - if !parsed { - // A wake for bytes an earlier pass already took. - return ControlSignal::Continue; - } - // New output may have rewritten or moved the hovered link; - // re-resolve it (clears when it's gone). - if self.term.link_hover().is_some() { + let outcome = ingest::outcome(finished, any_parsed, stopped, self.term.pause_rendering()); + if any_parsed && outcome != ingest::Outcome::Closed && self.term.link_hover().is_some() { + // New output may have rewritten or moved the hovered link; + // re-resolve it (clears when it's gone). self.refresh_link_hover(); } - if yielded { - self.ingest.frame_requested(parse_start); - return ControlSignal::Yield; - } - if self.term.pause_rendering() { - // Mid synchronized update: `render_term` would just skip, so - // instead of redrawing now, wake at the BSU watchdog deadline. - // `handle_pty_bytes` fires the watchdog when more bytes - // arrive after it; this covers BSU-then-silence. - let left = self.term.sync_output_started_at().map_or(BSU_TIMEOUT, |t| { - (t + BSU_TIMEOUT).saturating_duration_since(Instant::now()) - }); - return ControlSignal::RedrawIn(left); + match outcome { + // Every byte read before the close is any_parsed by now. + ingest::Outcome::Closed => self.handle_pty_closed(), + ingest::Outcome::Idle => ControlSignal::Continue, + ingest::Outcome::Yield => { + self.ingest.frame_requested(parse_start); + ControlSignal::Yield + } + ingest::Outcome::AwaitEsu => { + // `handle_pty_bytes` fires the watchdog when more bytes + // arrive after it; this covers BSU-then-silence. + let left = self.term.sync_output_started_at().map_or(BSU_TIMEOUT, |t| { + (t + BSU_TIMEOUT).saturating_duration_since(Instant::now()) + }); + ControlSignal::RedrawIn(left) + } + ingest::Outcome::Redraw => { + // RedrawIn(ZERO) — not Continue — to force the event loop + // to wake immediately. On macOS, plain `request_redraw()` + // doesn't wake from Wait reliably; RedrawIn sets + // ControlFlow::Poll, which does. + self.ingest.frame_requested(parse_start); + ControlSignal::RedrawIn(Duration::ZERO) + } } - // RedrawIn(ZERO) — not Continue — to force the event loop to - // wake immediately. On macOS, plain `request_redraw()` doesn't - // wake from Wait reliably; RedrawIn sets ControlFlow::Poll, - // which does. - self.ingest.frame_requested(parse_start); - ControlSignal::RedrawIn(Duration::ZERO) } /// The PTY closed and its last bytes are parsed: reap and exit. From a21dba02ad77b3a08a649d7e57085e7387446543 Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 02:27:41 -0700 Subject: [PATCH 07/10] ingest: return to the event loop during long synchronized updates While mode 2026 holds frames back the parse loop never yields (the frame would be skipped), so a flood inside one long BSU kept input, resize and close waiting for up to the 1 s watchdog. After 33 ms of parsing in that state, stop with the new ControlSignal::Defer: like Yield (the runner drops reader wakes and resumes from about_to_wait) but without requesting a redraw. --- crates/toastty-window/src/event.rs | 4 ++++ crates/toastty-window/src/lib.rs | 35 ++++++++++++++++++++++------ crates/toastty/src/ingest.rs | 37 ++++++++++++++++++++++++++++++ crates/toastty/src/main.rs | 1 + 4 files changed, 70 insertions(+), 7 deletions(-) diff --git a/crates/toastty-window/src/event.rs b/crates/toastty-window/src/event.rs index 15eea3ee..e7571809 100644 --- a/crates/toastty-window/src/event.rs +++ b/crates/toastty-window/src/event.rs @@ -250,4 +250,8 @@ pub enum ControlSignal { /// frame (winit hands out every queued user event before it gets to /// `RedrawRequested`). Yield, + /// Like [`Yield`](Self::Yield) without the redraw: let the loop + /// handle input, then deliver [`Event::PtyReady`] again. Returned + /// while mode 2026 holds frames back, when a redraw would be skipped. + Defer, } diff --git a/crates/toastty-window/src/lib.rs b/crates/toastty-window/src/lib.rs index 73b37c4e..7987e44b 100644 --- a/crates/toastty-window/src/lib.rs +++ b/crates/toastty-window/src/lib.rs @@ -317,8 +317,9 @@ fn schedule(sig: ControlSignal, is_redraw: bool, redraw_at: Option, now redraw_at, } } - // Yield redraws now exactly like RedrawIn(0); the resumed - // PtyReady is the Runner's business (see `Runner::resume_pty`). + // Yield redraws now exactly like RedrawIn(0), Defer polls + // without a redraw; the resumed PtyReady is the Runner's business + // (see `Runner::resume_pty`). // Duration::ZERO means "wake immediately" — under // ControlFlow::WaitUntil(now), winit's EventLoopWaker re-arms // the CFRunLoopTimer every iteration because Instant::now() @@ -332,6 +333,11 @@ fn schedule(sig: ControlSignal, is_redraw: bool, redraw_at: Option, now redraw_now: true, redraw_at, }, + ControlSignal::Defer => Plan { + flow: Flow::Poll, + redraw_now: false, + redraw_at, + }, ControlSignal::RedrawIn(d) if d.is_zero() => Plan { flow: Flow::Poll, redraw_now: true, @@ -364,10 +370,11 @@ struct Runner<'a, A: App> { /// When the next scheduled redraw is due (from a /// `ControlSignal::RedrawIn`), if any. redraw_at: Option, - /// The app answered `ControlSignal::Yield`: it stopped parsing PTY - /// output so a frame could be drawn. Until `about_to_wait` (which - /// every platform runs after the iteration's `RedrawRequested`) - /// delivers the resuming `Event::PtyReady`, reader wakes are dropped. + /// The app answered `ControlSignal::Yield` (or `Defer`): it stopped + /// parsing PTY output so a frame could be drawn (or input handled). + /// Until `about_to_wait` (which every platform runs after the + /// iteration's `RedrawRequested`) delivers the resuming + /// `Event::PtyReady`, reader wakes are dropped. resume_pty: bool, } @@ -376,7 +383,7 @@ impl Runner<'_, A> { let is_redraw = matches!(ev, Event::Redraw); let sig = self.app.event(ev); self.last_status = sig; - if sig == ControlSignal::Yield { + if matches!(sig, ControlSignal::Yield | ControlSignal::Defer) { self.resume_pty = true; } self.apply_control(sig, is_redraw, event_loop); @@ -1086,6 +1093,20 @@ mod tests { assert_eq!(plan.redraw_at, pending); } + #[test] + fn defer_polls_without_redrawing() { + let now = Instant::now(); + let pending = Some(now + ms(300)); + assert_eq!( + schedule(ControlSignal::Defer, false, pending, now), + Plan { + flow: Flow::Poll, + redraw_now: false, + redraw_at: pending, + } + ); + } + #[test] fn control_signal_variants_are_distinct() { assert_ne!(ControlSignal::Continue, ControlSignal::Exit); diff --git a/crates/toastty/src/ingest.rs b/crates/toastty/src/ingest.rs index 8eaef446..fc84eaeb 100644 --- a/crates/toastty/src/ingest.rs +++ b/crates/toastty/src/ingest.rs @@ -42,6 +42,12 @@ const MIN_PARSE: Duration = Duration::from_millis(1); /// come at least every ~33 ms of parsing. const MAX_MIN_PARSE: Duration = Duration::from_micros(33_333); +/// Longest stretch of parsing while mode 2026 holds frames back before +/// returning to the event loop anyway (without a redraw, which would be +/// skipped). A flood inside one long synchronized update otherwise keeps +/// input, resize and close waiting until the 1 s BSU watchdog. +const MAX_HELD_PARSE: Duration = Duration::from_micros(33_333); + /// How long a requested frame may hold up parsing. Redraws arrive within /// the same loop iteration on every platform winit supports, so this only /// guards against one that never comes: parsing then carries on as if no @@ -117,6 +123,10 @@ impl Ingest { any_parsed: bool, sync_paused: bool, ) -> Step { + if any_parsed && sync_paused && now.saturating_duration_since(parse_start) >= MAX_HELD_PARSE + { + return Step::Stop(Stop::Defer); + } let frame_due = self.should_yield(now, parse_start); if !frame_due { Step::Parse @@ -167,6 +177,9 @@ pub enum Step { pub enum Stop { /// To draw a frame: answer `ControlSignal::Yield`. Yield, + /// To let input through while mode 2026 holds the frame: answer + /// `ControlSignal::Defer`. + Defer, } /// What a finished `PtyReady` pass answers. @@ -178,6 +191,9 @@ pub enum Outcome { Idle, /// Redraw, then resume parsing (`ControlSignal::Yield`). Yield, + /// Resume parsing after the loop's other events, without a redraw + /// (`ControlSignal::Defer`). + Defer, /// Drained mid synchronized update: wake at the BSU watchdog deadline /// rather than redraw a frame `render_term` would skip. AwaitEsu, @@ -197,6 +213,7 @@ pub fn outcome(finished: bool, any_parsed: bool, stop: Option, sync_paused match stop { _ if finished => Outcome::Closed, Some(Stop::Yield) => Outcome::Yield, + Some(Stop::Defer) => Outcome::Defer, None if !any_parsed => Outcome::Idle, None if sync_paused => Outcome::AwaitEsu, None => Outcome::Redraw, @@ -317,6 +334,26 @@ mod tests { assert_eq!(ing.step(t0 + ms(5), t0, true, true), Step::ParseToEsu); } + #[test] + fn a_long_synchronized_update_returns_to_the_loop_without_a_frame() { + let ing = Ingest::default(); + let t0 = Instant::now(); + assert_eq!(ing.step(t0 + ms(20), t0, true, true), Step::ParseToEsu); + assert_eq!( + ing.step(t0 + MAX_HELD_PARSE, t0, true, true), + Step::Stop(Stop::Defer) + ); + // Still never before the first slice. + assert_eq!( + ing.step(t0 + MAX_HELD_PARSE, t0, false, true), + Step::ParseToEsu + ); + assert_eq!( + outcome(false, false, Some(Stop::Defer), true), + Outcome::Defer + ); + } + #[test] fn outcome_after_draining() { assert_eq!(outcome(true, true, None, false), Outcome::Closed); diff --git a/crates/toastty/src/main.rs b/crates/toastty/src/main.rs index 79631e38..2cdaf0b1 100644 --- a/crates/toastty/src/main.rs +++ b/crates/toastty/src/main.rs @@ -1017,6 +1017,7 @@ impl Toastty { self.ingest.frame_requested(parse_start); ControlSignal::Yield } + ingest::Outcome::Defer => ControlSignal::Defer, ingest::Outcome::AwaitEsu => { // `handle_pty_bytes` fires the watchdog when more bytes // arrive after it; this covers BSU-then-silence. From 0f0777461d06891e2605ec6886f421034c8f6c6c Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 02:28:28 -0700 Subject: [PATCH 08/10] io/window: document Yield/Defer, note the pre_present_notify hazard, share test glue - App::event documents ControlSignal::Yield and Defer. - Runner::about_to_wait notes that Wayland pre_present_notify() would defer RedrawRequested past the resume and make it spin under Poll. - PtyReader::has_unconsumed is test-only now. - ChannelSink/recv_batch move to a hidden toastty_io::test_support module shared by the toastty-io and end_to_end tests. - pty-event-loop.md: Defer and the never-stop-before-a-slice invariant. --- crates/toastty-io/src/lib.rs | 66 ++++++++++++++++++++++++++- crates/toastty-io/tests/common/mod.rs | 54 ---------------------- crates/toastty-io/tests/pty_reader.rs | 4 +- crates/toastty-io/tests/pty_writer.rs | 4 +- crates/toastty-window/src/lib.rs | 30 ++++++++---- crates/toastty/tests/end_to_end.rs | 46 +------------------ docs/decisions/pty-event-loop.md | 12 +++++ 7 files changed, 103 insertions(+), 113 deletions(-) delete mode 100644 crates/toastty-io/tests/common/mod.rs diff --git a/crates/toastty-io/src/lib.rs b/crates/toastty-io/src/lib.rs index ddbafb2c..e7f81ab0 100644 --- a/crates/toastty-io/src/lib.rs +++ b/crates/toastty-io/src/lib.rs @@ -230,7 +230,8 @@ impl PtyReader { /// Whether unconsumed bytes are already in hand (not counting any /// the reader thread buffered since the last take). - pub fn has_unconsumed(&self) -> bool { + #[cfg(test)] + fn has_unconsumed(&self) -> bool { self.pos < self.buf.len() } @@ -464,6 +465,69 @@ fn is_eio(e: &io::Error) -> bool { e.raw_os_error() == Some(libc::EIO) } +// ============================================================================ +// Test support +// ============================================================================ + +/// Glue for tests that drive a real reader without winit: a +/// channel-backed [`EventSink`] and [`recv_batch`], which turns the wake + +/// shared-buffer API back into "the next batch of bytes, or the close". +/// Public for the integration tests here and in the binary crate. +#[doc(hidden)] +pub mod test_support { + use std::sync::{Mutex, mpsc}; + use std::time::Duration; + + use super::{EventSink, PtyReader, SinkClosed, UserEvent}; + + /// Forwards each [`UserEvent`] into an mpsc channel. + #[derive(Debug)] + pub struct ChannelSink(pub Mutex>); + + impl EventSink for ChannelSink { + fn send(&self, ev: UserEvent) -> Result<(), SinkClosed> { + let g = self.0.lock().map_err(|_| SinkClosed)?; + g.send(ev).map_err(|_| SinkClosed) + } + } + + /// One step of what the reader delivered. + #[derive(Debug)] + pub enum Batch { + Bytes(Vec), + Closed, + } + + /// Everything buffered right now, or the close once every byte is + /// out. Waits up to `timeout` for a wake when there is neither; + /// `None` on timeout. + pub fn recv_batch( + reader: &mut PtyReader, + rx: &mpsc::Receiver, + timeout: Duration, + ) -> Option { + loop { + let mut out = Vec::new(); + loop { + let chunk = reader.peek(usize::MAX); + if chunk.is_empty() { + break; + } + out.extend_from_slice(chunk); + let n = chunk.len(); + reader.consume(n); + } + if !out.is_empty() { + return Some(Batch::Bytes(out)); + } + if reader.is_finished() { + return Some(Batch::Closed); + } + rx.recv_timeout(timeout).ok()?; + } + } +} + // ============================================================================ // PTY writer // ============================================================================ diff --git a/crates/toastty-io/tests/common/mod.rs b/crates/toastty-io/tests/common/mod.rs deleted file mode 100644 index bb31afbe..00000000 --- a/crates/toastty-io/tests/common/mod.rs +++ /dev/null @@ -1,54 +0,0 @@ -//! Shared test glue: a channel-backed [`EventSink`] and a helper that -//! turns the reader's wake + shared-buffer API back into "the next batch -//! of bytes, or the close", which is what these tests want to assert on. - -use std::sync::Mutex; -use std::sync::mpsc; -use std::time::Duration; - -use toastty_io::{EventSink, PtyReader, SinkClosed, UserEvent}; - -/// Test sink: forwards `UserEvent` into an mpsc channel. -pub(crate) struct ChannelSink(pub(crate) Mutex>); - -impl EventSink for ChannelSink { - fn send(&self, ev: UserEvent) -> Result<(), SinkClosed> { - let g = self.0.lock().map_err(|_| SinkClosed)?; - g.send(ev).map_err(|_| SinkClosed) - } -} - -/// One step of what the reader delivered. -pub(crate) enum Batch { - Bytes(Vec), - Closed, -} - -/// Everything buffered right now, or the close once every byte is out. -/// Waits up to `timeout` for a wake when there is neither; `None` on -/// timeout. -pub(crate) fn recv_batch( - reader: &mut PtyReader, - rx: &mpsc::Receiver, - timeout: Duration, -) -> Option { - loop { - let mut out = Vec::new(); - loop { - let chunk = reader.peek(usize::MAX); - if chunk.is_empty() { - break; - } - out.extend_from_slice(chunk); - let n = chunk.len(); - reader.consume(n); - } - if !out.is_empty() { - return Some(Batch::Bytes(out)); - } - if reader.is_finished() { - return Some(Batch::Closed); - } - rx.recv_timeout(timeout).ok()?; - } -} diff --git a/crates/toastty-io/tests/pty_reader.rs b/crates/toastty-io/tests/pty_reader.rs index 637f5deb..6ae0c77f 100644 --- a/crates/toastty-io/tests/pty_reader.rs +++ b/crates/toastty-io/tests/pty_reader.rs @@ -6,13 +6,11 @@ //! winit-side wiring is one trait impl and is exercised end-to-end by //! the binary's smoke run. -mod common; - use std::sync::Mutex; use std::sync::mpsc; use std::time::{Duration, Instant}; -use common::{Batch, ChannelSink, recv_batch}; +use toastty_io::test_support::{Batch, ChannelSink, recv_batch}; use toastty_io::{PtyReader, UserEvent, spawn_pty_reader_with_sink}; use toastty_pty::{Pty, PtySpec}; diff --git a/crates/toastty-io/tests/pty_writer.rs b/crates/toastty-io/tests/pty_writer.rs index 0f043da9..0dcf6c41 100644 --- a/crates/toastty-io/tests/pty_writer.rs +++ b/crates/toastty-io/tests/pty_writer.rs @@ -8,14 +8,12 @@ //! the foreground app in paste-collect mode. `PtyWriter` must deliver every //! byte in order regardless of ring size. -mod common; - use std::os::fd::{AsRawFd, BorrowedFd}; use std::sync::Mutex; use std::sync::mpsc; use std::time::{Duration, Instant}; -use common::{Batch, ChannelSink, recv_batch}; +use toastty_io::test_support::{Batch, ChannelSink, recv_batch}; use toastty_io::{PtyReader, UserEvent, spawn_pty_reader_with_sink, spawn_pty_writer}; use toastty_pty::{Pty, PtySpec}; diff --git a/crates/toastty-window/src/lib.rs b/crates/toastty-window/src/lib.rs index 7987e44b..d7a4c41b 100644 --- a/crates/toastty-window/src/lib.rs +++ b/crates/toastty-window/src/lib.rs @@ -158,6 +158,11 @@ pub trait App { /// Called for every event. Returning [`ControlSignal::Exit`] stops /// the loop; [`ControlSignal::RedrawIn`] schedules a redraw deadline; /// [`ControlSignal::Continue`] waits for the next event. + /// [`ControlSignal::Yield`] (redraw now) and [`ControlSignal::Defer`] + /// (no redraw) are answers to [`Event::PtyReady`] that stopped with + /// PTY output still buffered: the loop handles its other pending + /// events, drops reader wakes, and delivers `Event::PtyReady` again + /// from `about_to_wait`. fn event(&mut self, event: Event) -> ControlSignal; } @@ -598,14 +603,23 @@ impl ApplicationHandler for Runner<'_, A> { } fn about_to_wait(&mut self, event_loop: &ActiveEventLoop) { - // Resume PTY parsing after a `Yield`. winit runs this after the - // iteration's user events and redraws on X11, Wayland and macOS - // alike, so the frame the app yielded for is out by now. Calling - // the app directly (rather than posting a wake) keeps it to one - // parse slice per iteration without a channel round trip; the - // `Poll` flow `Yield` set means the loop doesn't sleep before the - // next frame. Otherwise we rely on the `ControlFlow` set in - // `apply_control`. + // Resume PTY parsing after a `Yield` / `Defer`. winit runs this + // after the iteration's user events and redraws on X11, Wayland + // and macOS alike, so the frame the app yielded for is out by + // now. Calling the app directly (rather than posting a wake) + // keeps it to one parse slice per iteration without a channel + // round trip; the `Poll` flow they set means the loop doesn't + // sleep before the next frame. Otherwise we rely on the + // `ControlFlow` set in `apply_control`. + // + // That ordering is load-bearing. If someone adds + // `Window::pre_present_notify()` (Wayland frame-callback + // throttling), winit holds `RedrawRequested` back until the + // compositor's frame callback, so the frame is NOT out by now: + // this resume would run before it, the app's + // `should_wait_for_frame` would `Yield` again, and under `Poll` + // the loop would spin until the callback (or the app's 100 ms + // cutoff). Revisit this before adding it. if std::mem::take(&mut self.resume_pty) { self.dispatch(Event::PtyReady, event_loop); } diff --git a/crates/toastty/tests/end_to_end.rs b/crates/toastty/tests/end_to_end.rs index 2ac7cb25..b5f95b58 100644 --- a/crates/toastty/tests/end_to_end.rs +++ b/crates/toastty/tests/end_to_end.rs @@ -8,7 +8,8 @@ use std::sync::Mutex; use std::sync::mpsc; use std::time::{Duration, Instant}; -use toastty_io::{EventSink, PtyReader, SinkClosed, UserEvent, spawn_pty_reader_with_sink}; +use toastty_io::spawn_pty_reader_with_sink; +use toastty_io::test_support::{Batch, ChannelSink, recv_batch}; use toastty_parser::Parser; use toastty_pty::{Pty, PtySpec, WinSize}; use toastty_term::Term; @@ -19,49 +20,6 @@ use toastty::shell::resolve_shell; use toastty_config::ShellConfig; use toastty_window::{KeyState, LogicalKey, Modifiers, NamedKey}; -struct ChannelSink(Mutex>); -impl EventSink for ChannelSink { - fn send(&self, ev: UserEvent) -> Result<(), SinkClosed> { - let g = self.0.lock().map_err(|_| SinkClosed)?; - g.send(ev).map_err(|_| SinkClosed) - } -} - -/// One step of what the reader delivered. -enum Batch { - Bytes(Vec), - Closed, -} - -/// Everything buffered right now, or the close once every byte is out. -/// Waits up to `timeout` for a wake when there is neither; `None` on -/// timeout. -fn recv_batch( - reader: &mut PtyReader, - rx: &mpsc::Receiver, - timeout: Duration, -) -> Option { - loop { - let mut out = Vec::new(); - loop { - let chunk = reader.peek(usize::MAX); - if chunk.is_empty() { - break; - } - out.extend_from_slice(chunk); - let n = chunk.len(); - reader.consume(n); - } - if !out.is_empty() { - return Some(Batch::Bytes(out)); - } - if reader.is_finished() { - return Some(Batch::Closed); - } - rx.recv_timeout(timeout).ok()?; - } -} - #[test] fn pty_echo_drives_term_through_parser() { // Spawn `echo hello` under a PTY at a known grid size. diff --git a/docs/decisions/pty-event-loop.md b/docs/decisions/pty-event-loop.md index dc739c6c..f8f687b1 100644 --- a/docs/decisions/pty-event-loop.md +++ b/docs/decisions/pty-event-loop.md @@ -308,6 +308,18 @@ to the next ESU (`CSI ? 2026 l`), so the frame goes out between updates even when the app sends ESU and the next BSU back to back. A batch that ends mid-update asks for a wake at the BSU watchdog deadline instead of a redraw that `render_term` would skip. +After 33 ms of parsing with frames held back, the handler answers +`ControlSignal::Defer` (resume from `about_to_wait`, no redraw), so a +flood inside one long update doesn't hold input, resize and close for +the whole 1 s watchdog. + +One invariant matters for both: the handler never stops before it has +parsed a slice, and any stop answers `Yield`/`Defer`. A stop that left +bytes untaken and answered `Continue` would freeze output for good, since +the reader's wake stays armed until its buffer is taken (an early version +did exactly that when the thread was descheduled at the top of a pass). +The decisions are pure functions in `ingest` (`Ingest::step`, `outcome`) +with unit tests. The CWD probe now runs only for slices that open an APC (`ESC _`, also across a slice boundary): RGP's `r;path=` registration, the only reader of From d3d9d8b862b3cd801309382b7cbb124e1e721836 Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 02:41:38 -0700 Subject: [PATCH 09/10] ingest: an empty pass inside a synchronized update re-arms the watchdog (review) --- crates/toastty/src/ingest.rs | 15 ++++++++++++++- crates/toastty/src/main.rs | 2 +- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/crates/toastty/src/ingest.rs b/crates/toastty/src/ingest.rs index fc84eaeb..aa1f9166 100644 --- a/crates/toastty/src/ingest.rs +++ b/crates/toastty/src/ingest.rs @@ -214,8 +214,13 @@ pub fn outcome(finished: bool, any_parsed: bool, stop: Option, sync_paused _ if finished => Outcome::Closed, Some(Stop::Yield) => Outcome::Yield, Some(Stop::Defer) => Outcome::Defer, - None if !any_parsed => Outcome::Idle, + // Still inside a synchronized update: (re)arm the watchdog wake even + // when this pass parsed nothing. The pass before it may have taken + // the update's last bytes and stopped with a `Defer`, which arms no + // deadline; answering `Idle` here would then leave the frame stale + // until something unrelated redraws. None if sync_paused => Outcome::AwaitEsu, + None if !any_parsed => Outcome::Idle, None => Outcome::Redraw, } } @@ -241,6 +246,14 @@ mod tests { Duration::from_millis(n) } + #[test] + fn an_empty_pass_inside_a_synchronized_update_rearms_the_watchdog() { + // A `Defer` pass took the last bytes of an update; the resumed pass + // finds nothing but must still schedule the BSU watchdog. + assert_eq!(outcome(false, false, None, true), Outcome::AwaitEsu); + assert_eq!(outcome(false, false, None, false), Outcome::Idle); + } + #[test] fn the_first_frame_is_due_at_once() { let ing = Ingest::default(); diff --git a/crates/toastty/src/main.rs b/crates/toastty/src/main.rs index 2cdaf0b1..a6856476 100644 --- a/crates/toastty/src/main.rs +++ b/crates/toastty/src/main.rs @@ -1010,7 +1010,7 @@ impl Toastty { self.refresh_link_hover(); } match outcome { - // Every byte read before the close is any_parsed by now. + // Every byte read before the close is parsed by now. ingest::Outcome::Closed => self.handle_pty_closed(), ingest::Outcome::Idle => ControlSignal::Continue, ingest::Outcome::Yield => { From 7c325637fa84817fb0b91e579e5638b7fcc7d4be Mon Sep 17 00:00:00 2001 From: EDToaster Date: Thu, 22 Oct 2026 02:54:05 -0700 Subject: [PATCH 10/10] ingest: cap drained-pass redraws at the frame interval --- crates/toastty/src/ingest.rs | 23 +++++++++++++++++++++++ crates/toastty/src/main.rs | 12 +++++++----- 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/crates/toastty/src/ingest.rs b/crates/toastty/src/ingest.rs index aa1f9166..51c2cd88 100644 --- a/crates/toastty/src/ingest.rs +++ b/crates/toastty/src/ingest.rs @@ -90,6 +90,18 @@ impl Ingest { self.frame_requested.get_or_insert(now); } + /// How long a drained pass waits before its redraw: until a frame is + /// due again. Zero after a quiet spell, so an echo or a prompt draws at + /// once; while output keeps arriving (a flood the parser now outruns, + /// so every pass drains) it caps the redraws at [`FRAME_INTERVAL`] + /// instead of drawing once per pass, which reached 360 FPS. + #[must_use] + pub fn redraw_delay(&self, now: Instant) -> Duration { + self.last_frame.map_or(Duration::ZERO, |t| { + (t + FRAME_INTERVAL).saturating_duration_since(now) + }) + } + fn frame_due(&self, now: Instant) -> bool { self.last_frame .is_none_or(|t| now.saturating_duration_since(t) >= FRAME_INTERVAL) @@ -246,6 +258,17 @@ mod tests { Duration::from_millis(n) } + #[test] + fn a_drained_pass_redraws_at_once_only_when_a_frame_is_due() { + let mut ing = Ingest::default(); + let t0 = Instant::now(); + assert_eq!(ing.redraw_delay(t0), Duration::ZERO); + ing.frame_drawn(t0, t0 + ms(1)); + assert_eq!(ing.redraw_delay(t0 + ms(4)) + ms(4), FRAME_INTERVAL); + assert_eq!(ing.redraw_delay(t0 + FRAME_INTERVAL), Duration::ZERO); + assert_eq!(ing.redraw_delay(t0 + ms(100)), Duration::ZERO); + } + #[test] fn an_empty_pass_inside_a_synchronized_update_rearms_the_watchdog() { // A `Defer` pass took the last bytes of an update; the resumed pass diff --git a/crates/toastty/src/main.rs b/crates/toastty/src/main.rs index a6856476..8e5f5fc7 100644 --- a/crates/toastty/src/main.rs +++ b/crates/toastty/src/main.rs @@ -1027,12 +1027,14 @@ impl Toastty { ControlSignal::RedrawIn(left) } ingest::Outcome::Redraw => { - // RedrawIn(ZERO) — not Continue — to force the event loop - // to wake immediately. On macOS, plain `request_redraw()` - // doesn't wake from Wait reliably; RedrawIn sets - // ControlFlow::Poll, which does. + // RedrawIn — not Continue — to force the event loop to + // wake. On macOS, plain `request_redraw()` doesn't wake + // from Wait reliably; RedrawIn(ZERO) sets ControlFlow::Poll + // and a later deadline WaitUntil, which both do. The delay + // is zero unless a frame was drawn under one frame + // interval ago (see `Ingest::redraw_delay`). self.ingest.frame_requested(parse_start); - ControlSignal::RedrawIn(Duration::ZERO) + ControlSignal::RedrawIn(self.ingest.redraw_delay(Instant::now())) } } }