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-io/src/lib.rs b/crates/toastty-io/src/lib.rs index f3e9d0bd..e7f81ab0 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,179 @@ 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). + #[cfg(test)] + 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 +290,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 +314,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 +345,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 +375,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,41 +403,131 @@ 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) } +// ============================================================================ +// 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 // ============================================================================ @@ -446,11 +727,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/pty_reader.rs b/crates/toastty-io/tests/pty_reader.rs index 7510c5c5..6ae0c77f 100644 --- a/crates/toastty-io/tests/pty_reader.rs +++ b/crates/toastty-io/tests/pty_reader.rs @@ -10,23 +10,17 @@ 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::test_support::{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 +29,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 +48,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 +65,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 +77,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 +104,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..0dcf6c41 100644 --- a/crates/toastty-io/tests/pty_writer.rs +++ b/crates/toastty-io/tests/pty_writer.rs @@ -13,20 +13,10 @@ 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 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}; -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 +37,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 +50,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 +79,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 +87,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 +114,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 +131,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..e7571809 100644 --- a/crates/toastty-window/src/event.rs +++ b/crates/toastty-window/src/event.rs @@ -207,13 +207,14 @@ 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. + /// + /// 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 /// (e.g. fcitx5 Hangul/CJK composition), NOT through `Event::Key`. @@ -240,4 +241,17 @@ 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, + /// 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 a641cbe8..d7a4c41b 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, } } } @@ -162,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; } @@ -266,6 +267,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); @@ -320,6 +322,9 @@ fn schedule(sig: ControlSignal, is_redraw: bool, redraw_at: Option, now redraw_at, } } + // 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() @@ -328,6 +333,16 @@ 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::Defer => Plan { + flow: Flow::Poll, + redraw_now: false, + redraw_at, + }, ControlSignal::RedrawIn(d) if d.is_zero() => Plan { flow: Flow::Poll, redraw_now: true, @@ -360,6 +375,12 @@ 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` (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, } impl Runner<'_, A> { @@ -367,6 +388,9 @@ impl Runner<'_, A> { let is_redraw = matches!(ev, Event::Redraw); let sig = self.app.event(ev); self.last_status = sig; + if matches!(sig, ControlSignal::Yield | ControlSignal::Defer) { + self.resume_pty = true; + } self.apply_control(sig, is_redraw, event_loop); } @@ -559,8 +583,9 @@ 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, + // 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); } @@ -577,8 +602,27 @@ 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` / `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); + } } } @@ -1049,6 +1093,34 @@ 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 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/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..51c2cd88 --- /dev/null +++ b/crates/toastty/src/ingest.rs @@ -0,0 +1,426 @@ +//! 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 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); + +/// 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 +/// 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, + /// The last slice ended in ESC, so a `_` opening the next one starts + /// an APC. + trailing_esc: bool, +} + +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); + } + + /// 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) + } + + /// 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) + } + + /// 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 step( + &self, + now: Instant, + parse_start: Instant, + 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 + } 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 + } + + /// 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() + } +} + +/// 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, + /// To let input through while mode 2026 holds the frame: answer + /// `ControlSignal::Defer`. + Defer, +} + +/// 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, + /// 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, + /// 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, + Some(Stop::Defer) => Outcome::Defer, + // 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, + } +} + +/// 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 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 + // 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(); + 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_twice_their_time_in_parsing() { + let mut ing = Ingest::default(); + let t0 = Instant::now(); + // 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(parse_start + ms(33), parse_start)); + assert!(ing.should_yield(parse_start + MAX_MIN_PARSE, 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 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 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); + 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"; + 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); + } + + #[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/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 4560b4c8..8e11821f 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 +//! (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, @@ -271,9 +272,12 @@ 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, + /// 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 @@ -299,7 +303,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 @@ -504,6 +508,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 @@ -964,19 +969,119 @@ 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`). + /// + /// 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) { + // 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 any_parsed = false; + let mut stopped = None; + loop { + 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); + if avail.is_empty() { + break; + } + // A frame is due but a synchronized update holds it: parse + // only to the end of that update, then check again. + 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(at, n, at.elapsed()); + reader.consume(n); + any_parsed = true; + } + let finished = reader.is_finished(); + self.reader = Some(reader); + 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(); + } + match outcome { + // 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 => { + 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. + 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 — 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(self.ingest.redraw_delay(Instant::now())) + } + } + } + + /// 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, + // 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(); @@ -1025,11 +1130,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() } } @@ -1183,7 +1283,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, @@ -1391,7 +1491,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) } @@ -2953,11 +3053,14 @@ impl 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 @@ -3143,43 +3246,7 @@ impl 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 @@ -3578,6 +3645,12 @@ mod tests { // against, so this checks the guard's effect on a hand cursor // that was showing: under the prompt, Ctrl and new output only // ever clear it. (Checked against a real link on Xvfb.) + struct Wake(std::sync::mpsc::Sender<()>); + impl toastty_io::EventSink for Wake { + fn send(&self, _: toastty_io::UserEvent) -> Result<(), toastty_io::SinkClosed> { + self.0.send(()).map_err(|_| toastty_io::SinkClosed) + } + } let mut app = headless_app(); let ctrl = if cfg!(target_os = "macos") { Modifiers::SUPER @@ -3590,7 +3663,18 @@ mod tests { app.event(Event::ModifiersChanged(ctrl)); assert!(!app.link_pointer && app.term.link_hover().is_none()); app.link_pointer = true; - app.event(Event::PtyBytes(b"x".to_vec())); + // New output arrives the real way: through a reader on a pipe, + // drained on `PtyReady`. + let (rx, mut tx) = std::io::pipe().unwrap(); + let (wake_tx, wake_rx) = std::sync::mpsc::channel(); + let reader = + toastty_io::spawn_pty_reader_with_sink(std::os::fd::AsFd::as_fd(&rx), Wake(wake_tx)) + .unwrap(); + std::io::Write::write_all(&mut tx, b"x").unwrap(); + wake_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + app.reader = Some(reader); + app.event(Event::PtyReady); + assert_eq!(app.term.cursor().col, 1, "the output was parsed"); assert!(app.term.link_hover().is_none()); } diff --git a/crates/toastty/tests/end_to_end.rs b/crates/toastty/tests/end_to_end.rs index d23e56a9..b5f95b58 100644 --- a/crates/toastty/tests/end_to_end.rs +++ b/crates/toastty/tests/end_to_end.rs @@ -1,14 +1,15 @@ //! 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::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,14 +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) - } -} - #[test] fn pty_echo_drives_term_through_parser() { // Spawn `echo hello` under a PTY at a known grid size. @@ -42,7 +35,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 +43,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 +84,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 +123,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 +132,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 +155,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 +177,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 +211,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 +250,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 +295,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 +340,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 +349,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); diff --git a/docs/decisions/pty-event-loop.md b/docs/decisions/pty-event-loop.md index a6ee57fe..f8f687b1 100644 --- a/docs/decisions/pty-event-loop.md +++ b/docs/decisions/pty-event-loop.md @@ -249,3 +249,96 @@ 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 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 +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. +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 +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.