diff --git a/desktop/src-tauri/crates/buzz-terminal/src/lifecycle.rs b/desktop/src-tauri/crates/buzz-terminal/src/lifecycle.rs index b5fdc1268..dbd76f56e 100644 --- a/desktop/src-tauri/crates/buzz-terminal/src/lifecycle.rs +++ b/desktop/src-tauri/crates/buzz-terminal/src/lifecycle.rs @@ -232,7 +232,13 @@ pub const MAX_LIVE_SESSIONS: usize = 20; /// drained. Join the reader first and every tab close pays that, on the arm /// where the user is already waiting. pub trait DrainingReader { - /// Stops consuming and releases the reader. Called *after* reap. + /// Detach parser work and enter raw-drain mode before child termination. + fn begin_closing(&self); + + /// Wake a reader that remains blocked after the child has been reaped. + fn stop(&self); + + /// Releases the reader thread. Called only after [`DrainingReader::stop`]. fn join(self: Box); } @@ -249,11 +255,13 @@ pub fn shutdown_draining( child: &mut Box, reader: Box, ) -> io::Result { + reader.begin_closing(); // Terminate and reap with the reader still running, so the child never // blocks in a tty write while we are waiting on it. let outcome = shutdown(child); - // Unconditional: the reader must be released whether or not the shutdown - // reported an error, or a failed close leaks the thread and its fd. + // Unconditional: wake and release the reader whether or not shutdown + // reported an error. EOF may already have ended it; stop is idempotent. + reader.stop(); reader.join(); outcome } diff --git a/desktop/src-tauri/crates/buzz-terminal/src/lifecycle_tests.rs b/desktop/src-tauri/crates/buzz-terminal/src/lifecycle_tests.rs index 17068f21a..c2a5e2eda 100644 --- a/desktop/src-tauri/crates/buzz-terminal/src/lifecycle_tests.rs +++ b/desktop/src-tauri/crates/buzz-terminal/src/lifecycle_tests.rs @@ -430,7 +430,8 @@ fn reader_drains_through_termination_and_reap() { let after_close = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)); let closing = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); - let reader = RecordingReader::spawn(&pair, after_close.clone(), closing.clone()); + let order = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let reader = RecordingReader::spawn(&pair, after_close.clone(), closing.clone(), order.clone()); // Let the child get well ahead of the reader before we touch anything. assert!( @@ -461,6 +462,11 @@ fn reader_drains_through_termination_and_reap() { period: the child was blocked writing to an undrained master rather \ than exiting on SIGTERM" ); + assert_eq!( + *order.lock().unwrap(), + ["begin_closing", "stop", "join"], + "reader close must begin before termination and stop/join only after reap" + ); } /// Spawns a child that floods the PTY without pause. @@ -478,6 +484,7 @@ fn spawn_noisy(pair: &PtyPair) -> Box { struct RecordingReader { total: std::sync::Arc, handle: std::thread::JoinHandle<()>, + order: std::sync::Arc>>, } impl RecordingReader { @@ -485,6 +492,7 @@ impl RecordingReader { pair: &PtyPair, after_close: std::sync::Arc, closing: std::sync::Arc, + order: std::sync::Arc>>, ) -> Self { let mut reader = pair.master.try_clone_reader().expect("reader"); let total = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)); @@ -502,7 +510,11 @@ impl RecordingReader { } } }); - Self { total, handle } + Self { + total, + handle, + order, + } } fn total_bytes(&self) -> u64 { @@ -511,7 +523,16 @@ impl RecordingReader { } impl DrainingReader for RecordingReader { + fn begin_closing(&self) { + self.order.lock().unwrap().push("begin_closing"); + } + + fn stop(&self) { + self.order.lock().unwrap().push("stop"); + } + fn join(self: Box) { + self.order.lock().unwrap().push("join"); // Bounded, and that is the whole point. The read loop ends when the // master reports EOF, which only happens once the reaped child has // released the slave -- so joining *before* termination blocks diff --git a/desktop/src-tauri/crates/buzz-terminal/src/shared.rs b/desktop/src-tauri/crates/buzz-terminal/src/shared.rs index 58bf0a76e..17c9f413b 100644 --- a/desktop/src-tauri/crates/buzz-terminal/src/shared.rs +++ b/desktop/src-tauri/crates/buzz-terminal/src/shared.rs @@ -14,7 +14,7 @@ //! metered separately: pooling them would let the reader's millions of fast //! acquires dilute the renderer's tail into a false pass. -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::time::Instant; use alacritty_terminal::sync::FairMutex; @@ -126,6 +126,7 @@ pub struct SharedTerminal { term: FairMutex, reader: AcquireMeter, renderer: AcquireMeter, + closing: AtomicBool, } impl SharedTerminal { @@ -134,6 +135,7 @@ impl SharedTerminal { term: FairMutex::new(term), reader: AcquireMeter::default(), renderer: AcquireMeter::default(), + closing: AtomicBool::new(false), } } @@ -156,13 +158,23 @@ impl SharedTerminal { /// the whole buffer under one acquisition is what an unbounded hold *is*, /// so it is not offered here. pub fn feed(&self, bytes: &[u8]) -> bool { - self.acquire(&self.reader).feed(bytes) + let mut term = self.acquire(&self.reader); + if self.closing.load(Ordering::Acquire) { + false + } else { + term.feed(bytes) + } } /// Parse more of the pending tail under a fresh acquisition. Reader /// plane. Returns whether any remains. pub fn drain(&self) -> bool { - self.acquire(&self.reader).drain() + let mut term = self.acquire(&self.reader); + if self.closing.load(Ordering::Acquire) { + false + } else { + term.drain() + } } /// Feed and pump to completion, re-acquiring between slices. @@ -173,6 +185,17 @@ impl SharedTerminal { } } + /// Atomically enter close mode and discard parser work. Subsequent PTY + /// bytes are raw-drained by the embedder and never reach callbacks. + pub fn begin_closing(&self) -> usize { + self.closing.store(true, Ordering::Release); + self.acquire(&self.reader).abandon_tail() + } + + pub fn is_closing(&self) -> bool { + self.closing.load(Ordering::Acquire) + } + /// Sample damage and encode a frame. Renderer plane. /// /// The lock covers the copy only; `encode` -- hashing, span grouping, @@ -236,3 +259,31 @@ impl SharedTerminal { guard } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{Fences, Size}; + + #[test] + fn closing_abandons_tail_and_permanently_refuses_parser_callbacks() { + let (terminal, _actions) = Terminal::new(Size::default(), Fences::ALL); + let shared = SharedTerminal::new(terminal); + let payload = b"\x1b#8".repeat(10_000); + + assert!(shared.feed(&payload), "fixture must create parser tail"); + let before = shared.lock().stats(); + let abandoned = shared.begin_closing(); + assert!(abandoned > 0, "close must abandon without draining first"); + assert!(shared.is_closing()); + + assert!(!shared.feed(b"parser callback after close")); + assert!(!shared.drain()); + let after = shared.lock().stats(); + assert_eq!(after.completed_units, before.completed_units); + assert_eq!( + after.abandoned_bytes, + before.abandoned_bytes + abandoned as u64 + ); + } +} diff --git a/desktop/src-tauri/src/terminal_runtime.rs b/desktop/src-tauri/src/terminal_runtime.rs index dcf861e23..30d8df6d1 100644 --- a/desktop/src-tauri/src/terminal_runtime.rs +++ b/desktop/src-tauri/src/terminal_runtime.rs @@ -1,6 +1,7 @@ //! Rust-owned PTY sessions and the typed Tauri transport for Buzz Substrate. use std::io::{Read, Write}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::thread::JoinHandle; @@ -202,10 +203,23 @@ fn wire_publication(publication: Publication) -> Result { }) } -struct ReaderThread(Option>); +struct ReaderThread { + handle: Option>, + terminal: Arc, + stopping: Arc, +} + impl buzz_terminal::lifecycle::DrainingReader for ReaderThread { + fn begin_closing(&self) { + self.terminal.begin_closing(); + } + + fn stop(&self) { + self.stopping.store(true, Ordering::Release); + } + fn join(mut self: Box) { - if let Some(handle) = self.0.take() { + if let Some(handle) = self.handle.take() { let _ = handle.join(); } } @@ -253,7 +267,12 @@ impl Session { if let Ok(mut channel) = self.channel.lock() { *channel = None; } - // The slave closes on child reap; the reader continues draining until then. + // Publication is detached before the reader enters close mode; the + // lifecycle helper then abandons parser work and keeps raw-draining + // through child termination and reap. + if let Ok(mut publisher) = self.publisher.lock() { + publisher.close(); + } if let Some(reader) = self.reader.take() { #[cfg(unix)] { @@ -261,12 +280,13 @@ impl Session { } #[cfg(not(unix))] { + reader.begin_closing(); let _ = self.child.kill(); let _ = self.child.wait(); + reader.stop(); reader.join(); } } - self.master.take(); } } @@ -465,6 +485,8 @@ pub(crate) fn terminal_attach( let reader_terminal = Arc::clone(&terminal); let reader_publisher = Arc::clone(&publisher); let reader_channel = Arc::clone(&channel); + let reader_stopping = Arc::new(AtomicBool::new(false)); + let thread_stopping = Arc::clone(&reader_stopping); let reader_handle = std::thread::spawn(move || { let mut buffer = [0u8; 16 * 1024]; let mut encoder = buzz_terminal::damage::Encoder::new(); @@ -474,7 +496,19 @@ pub(crate) fn terminal_attach( Ok(0) | Err(_) => break, Ok(count) => count, }; - reader_terminal.feed(&buffer[..count]); + if thread_stopping.load(Ordering::Acquire) { + break; + } + if reader_terminal.is_closing() { + continue; + } + let mut more = reader_terminal.feed(&buffer[..count]); + while more && !reader_terminal.is_closing() { + more = reader_terminal.drain(); + } + if reader_terminal.is_closing() { + continue; + } let needs_snapshot = reader_publisher .lock() .map(|publisher| publisher.requires_snapshot()) @@ -527,12 +561,16 @@ pub(crate) fn terminal_attach( .map_err(|_| "terminal snapshot rejected".to_string())?; let session = Session { id, - terminal, + terminal: Arc::clone(&terminal), master: Some(pair.master), writer, pty_size: current_pty_size, child, - reader: Some(Box::new(ReaderThread(Some(reader_handle)))), + reader: Some(Box::new(ReaderThread { + handle: Some(reader_handle), + terminal: Arc::clone(&terminal), + stopping: reader_stopping, + })), publisher, channel, }; diff --git a/desktop/src-tauri/src/terminal_transport.rs b/desktop/src-tauri/src/terminal_transport.rs index 68666aaa7..c00a1c006 100644 --- a/desktop/src-tauri/src/terminal_transport.rs +++ b/desktop/src-tauri/src/terminal_transport.rs @@ -186,6 +186,11 @@ impl FramePublisher { } } + /// Permanently detach publication as the session begins closing. + pub(crate) fn close(&mut self) { + self.subscription = None; + } + fn require_current_snapshot(&self, frame: &Frame) -> Result<(), OfferError> { if frame.viewport == self.applied && frame.full { Ok(())