From fd8b4ff14cd4352d1fca2a0abf7a5ed7eed7c20e Mon Sep 17 00:00:00 2001 From: l0ng-ai <24760907+l0ng-ai@users.noreply.github.com> Date: Mon, 9 Nov 2026 09:17:30 +0800 Subject: [PATCH] fix(terminal): resume pasted input as soon as the PTY accepts more (#2220) --- crates/ide-core/src/input.rs | 138 ++++++++++++++++++++++++++++++--- crates/ide-dap/src/session.rs | 45 +++++++++-- crates/ide-run/src/process.rs | 124 +++++++++++++++++++++++------ crates/ide-terminal/src/lib.rs | 53 ++++++++++--- 4 files changed, 308 insertions(+), 52 deletions(-) diff --git a/crates/ide-core/src/input.rs b/crates/ide-core/src/input.rs index 7d4a24961..be61d8330 100644 --- a/crates/ide-core/src/input.rs +++ b/crates/ide-core/src/input.rs @@ -4,10 +4,12 @@ use std::collections::VecDeque; use std::io::{self, Write}; use std::sync::{Arc, Condvar, Mutex}; use std::thread::JoinHandle; +#[cfg(any(not(unix), test))] use std::time::Duration; const MAX_BYTES: usize = 4 * 1024 * 1024; const MAX_WRITES: usize = 1024; +#[cfg(any(not(unix), test))] const RETRY: Duration = Duration::from_millis(4); #[derive(Default)] @@ -28,10 +30,11 @@ pub struct InputWriter { } impl InputWriter { - /// Unix writers must first be made nonblocking with `nonblocking`. + /// On Unix, pass the readiness hook returned by `nonblocking` for this writer. pub fn new( name: String, mut writer: Box, + #[cfg(unix)] mut wait_writable: impl FnMut() -> io::Result<()> + Send + 'static, on_error: impl FnOnce(io::Error) + Send + 'static, ) -> io::Result { let shared = Arc::new((Mutex::new(State::default()), Condvar::new())); @@ -74,11 +77,19 @@ impl InputWriter { Ok(()) => {} Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, Err(e) if e.kind() == io::ErrorKind::WouldBlock => { - let state = lock.lock().unwrap(); - if state.aborted { - return Ok(()); + #[cfg(unix)] + match wait_writable() { + Err(e) if e.kind() == io::ErrorKind::Interrupted => {} + result => result?, + } + #[cfg(not(unix))] + { + let state = lock.lock().unwrap(); + if state.aborted { + return Ok(()); + } + drop(wake.wait_timeout(state, RETRY).unwrap()); } - drop(wake.wait_timeout(state, RETRY).unwrap()); } Err(e) => return Err(e), } @@ -168,14 +179,46 @@ impl Drop for InputWriter { } } -/// PTY clones share file status flags, so readers must also handle WouldBlock. +/// Make a writer nonblocking and return its readiness hook. PTY clones share file +/// status flags, so readers must also handle WouldBlock. #[cfg(unix)] -pub fn nonblocking(fd: std::os::fd::RawFd) -> io::Result<()> { +pub fn nonblocking(fd: std::os::fd::RawFd) -> io::Result io::Result<()> + Send> { + let wait = readiness(fd, libc::POLLOUT)?; let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; if flags == -1 || unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } == -1 { return Err(io::Error::last_os_error()); } - Ok(()) + Ok(wait) +} + +/// Wait for output from a nonblocking PTY without delaying the child's next read. +#[cfg(unix)] +pub fn readable(fd: std::os::fd::RawFd) -> io::Result io::Result<()> + Send> { + readiness(fd, libc::POLLIN) +} + +#[cfg(unix)] +fn readiness(fd: std::os::fd::RawFd, events: libc::c_short) -> io::Result io::Result<()> + Send> { + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + + // Own a duplicate so the hook remains valid if the caller closes its fd. + let duplicate = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) }; + if duplicate == -1 { + return Err(io::Error::last_os_error()); + } + // SAFETY: fcntl returned a new, owned descriptor. + let fd = unsafe { OwnedFd::from_raw_fd(duplicate) }; + Ok(move || { + let mut pollfd = libc::pollfd { fd: fd.as_raw_fd(), events, revents: 0 }; + // Readiness wakes immediately; the timeout only bounds abort latency. + if unsafe { libc::poll(&mut pollfd, 1, 50) } == -1 { + return Err(io::Error::last_os_error()); + } + if pollfd.revents & libc::POLLNVAL != 0 { + return Err(io::Error::from_raw_os_error(libc::EBADF)); + } + Ok(()) + }) } #[cfg(test)] @@ -221,6 +264,11 @@ mod tests { let writer = InputWriter::new( "test-input".into(), Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }), + #[cfg(unix)] + || { + std::thread::sleep(RETRY); + Ok(()) + }, |e| panic!("{e}"), ) .unwrap(); @@ -243,6 +291,11 @@ mod tests { let writer = InputWriter::new( "test-input".into(), Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }), + #[cfg(unix)] + || { + std::thread::sleep(RETRY); + Ok(()) + }, |e| panic!("{e}"), ) .unwrap(); @@ -256,6 +309,26 @@ mod tests { assert_eq!(*output.lock().unwrap(), format!("{paste}tail").as_bytes()); } + #[cfg(unix)] + #[test] + fn readiness_errors_are_reported_and_reject_following_writes() { + let (dropped, _rx) = mpsc::channel(); + let (errors, error) = mpsc::channel(); + let writer = InputWriter::new( + "test-error-input".into(), + Box::new(SlowWriter { ready: Arc::new(AtomicBool::new(false)), output: Arc::default(), dropped }), + || Err(io::Error::from_raw_os_error(libc::EBADF)), + move |e| errors.send(e.raw_os_error()).unwrap(), + ) + .unwrap(); + writer.write(b"input").unwrap(); + assert_eq!(error.recv_timeout(Duration::from_secs(1)).unwrap(), Some(libc::EBADF)); + assert_eq!(writer.write(b"later").unwrap_err().kind(), io::ErrorKind::BrokenPipe); + let state = writer.shared.0.lock().unwrap(); + assert!(state.queue.is_empty()); + assert_eq!(state.bytes, 0); + } + #[test] fn tiny_writes_are_also_bounded() { let shared = Arc::new((Mutex::new(State::default()), Condvar::new())); @@ -266,14 +339,59 @@ mod tests { assert_eq!(writer.write(b"x").unwrap_err().kind(), io::ErrorKind::WouldBlock); } + #[cfg(unix)] + #[test] + fn writable_input_resumes_without_timer_delays() { + use std::io::Read; + use std::os::fd::AsRawFd; + use std::os::unix::net::UnixStream; + + let (pipe, mut peer) = UnixStream::pair().unwrap(); + let capacity: libc::c_int = 1024; + // Keep the OS buffer small enough that a timer per buffer takes seconds. + assert_eq!( + unsafe { + libc::setsockopt( + pipe.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_SNDBUF, + (&capacity as *const libc::c_int).cast(), + std::mem::size_of_val(&capacity) as libc::socklen_t, + ) + }, + 0 + ); + peer.set_read_timeout(Some(Duration::from_secs(3))).unwrap(); + let wait_writable = nonblocking(pipe.as_raw_fd()).unwrap(); + let writer = + InputWriter::new("test-ready-input".into(), Box::new(pipe), wait_writable, |e| panic!("{e}")).unwrap(); + let bytes = vec![b'x'; MAX_BYTES]; + writer.write(&bytes).unwrap(); + writer.finish(); + // Start with a full buffer, then make the fd writable from another thread. + std::thread::sleep(Duration::from_millis(50)); + let (tx, rx) = mpsc::channel(); + let started = Instant::now(); + let reader = std::thread::spawn(move || { + let mut output = Vec::new(); + peer.read_to_end(&mut output).unwrap(); + tx.send(output).unwrap(); + }); + let output = rx.recv_timeout(Duration::from_secs(1)).expect("input resumes as soon as the fd is writable"); + assert_eq!(output, bytes); + reader.join().unwrap(); + eprintln!("4 MiB input drained in {:?}", started.elapsed()); + } + #[cfg(unix)] #[test] fn close_interrupts_a_full_os_pipe_while_the_peer_stays_open() { use std::os::fd::AsRawFd; use std::os::unix::net::UnixStream; let (pipe, _peer) = UnixStream::pair().unwrap(); - nonblocking(pipe.as_raw_fd()).unwrap(); - let mut writer = InputWriter::new("test-pipe-input".into(), Box::new(pipe), |e| panic!("{e}")).unwrap(); + let wait_writable = nonblocking(pipe.as_raw_fd()).unwrap(); + let mut writer = + InputWriter::new("test-pipe-input".into(), Box::new(pipe), wait_writable, |e| panic!("{e}")).unwrap(); writer.write(&vec![b'x'; MAX_BYTES]).unwrap(); std::thread::sleep(Duration::from_millis(50)); assert!(!writer.worker.as_ref().unwrap().is_finished()); diff --git a/crates/ide-dap/src/session.rs b/crates/ide-dap/src/session.rs index 614823ae1..cfb4bfb1d 100644 --- a/crates/ide-dap/src/session.rs +++ b/crates/ide-dap/src/session.rs @@ -630,9 +630,15 @@ pub(crate) fn run_in_terminal(s: &Arc, args: &Value) -> Result, args: &Value) -> Result, args: &Value) -> Result) { +fn forward_terminal_output( + s: &Session, + mut reader: Box, + #[cfg(unix)] mut wait_readable: impl FnMut() -> std::io::Result<()> + Send + 'static, +) { let session = s.id; let sink = s.owner.sink.clone(); let _ = std::thread::Builder::new().name(format!("openidea-debuggee-{}", session.0)).spawn(move || { @@ -692,6 +713,14 @@ fn forward_terminal_output(s: &Session, mut reader: Box pending.extend_from_slice(&buf[..n]), Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + #[cfg(unix)] + if let Err(e) = wait_readable() + && e.kind() != std::io::ErrorKind::Interrupted + { + tracing::warn!(%session, "debug terminal read: {e}"); + break; + } + #[cfg(not(unix))] std::thread::sleep(std::time::Duration::from_millis(4)); continue; } diff --git a/crates/ide-run/src/process.rs b/crates/ide-run/src/process.rs index 3c955eadc..e48b3074e 100644 --- a/crates/ide-run/src/process.rs +++ b/crates/ide-run/src/process.rs @@ -426,11 +426,16 @@ impl ProcessManager { Ok(match child.stdin.take() { Some(pipe) if !redirected => { #[cfg(unix)] - { + let wait_writable = { use std::os::fd::AsRawFd; - ide_core::input::nonblocking(pipe.as_raw_fd())?; - } - Some(input_writer(id, Box::new(pipe))?) + ide_core::input::nonblocking(pipe.as_raw_fd())? + }; + Some(input_writer( + id, + Box::new(pipe), + #[cfg(unix)] + wait_writable, + )?) } _ => None, }) @@ -462,10 +467,24 @@ impl ProcessManager { let stderr = child.stderr.take(); let result = (|| -> std::io::Result<()> { if let Some(out) = stdout { - spawn_reader(id, Stream::Stdout, out, tx.clone())?; + spawn_reader( + id, + Stream::Stdout, + out, + tx.clone(), + #[cfg(unix)] + None, + )?; } if let Some(err) = stderr { - spawn_reader(id, Stream::Stderr, err, tx.clone())?; + spawn_reader( + id, + Stream::Stderr, + err, + tx.clone(), + #[cfg(unix)] + None, + )?; } spawn_pump(rx, on_output.clone(), entry.clone(), sink, options.encoding.clone(), notice)?; Ok(()) @@ -538,9 +557,13 @@ impl ProcessManager { cmd.env(k, v); } #[cfg(unix)] - if let Some(fd) = pair.master.as_raw_fd() { - ide_core::input::nonblocking(fd)?; - } + let wait_writable = ide_core::input::nonblocking( + pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?, + )?; + #[cfg(unix)] + let wait_readable = ide_core::input::readable( + pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?, + )?; let mut child = pair.slave.spawn_command(cmd).map_err(|e| { let text = e.to_string(); if text.contains("No such file") || text.contains("not found") { @@ -555,7 +578,13 @@ impl ProcessManager { let writer = pair.master.take_writer().map_err(|e| IdeError::internal(format!("pty writer: {e}")))?; let id = RunId::next(); - let writer = input_writer(id, writer).inspect_err(|_| { + let writer = input_writer( + id, + writer, + #[cfg(unix)] + wait_writable, + ) + .inspect_err(|_| { let _ = child.kill(); let _ = child.wait(); })?; @@ -577,8 +606,15 @@ impl ProcessManager { on_output(OutputChunk::System { text: format!("{}\n", inv.display) }); let (tx, rx) = mpsc::channel(); - let result = spawn_reader(id, Stream::Stdout, reader, tx.clone()) - .and_then(|_| spawn_pump(rx, on_output, entry.clone(), sink, options.encoding.clone(), None)); + let result = spawn_reader( + id, + Stream::Stdout, + reader, + tx.clone(), + #[cfg(unix)] + Some(Box::new(wait_readable)), + ) + .and_then(|_| spawn_pump(rx, on_output, entry.clone(), sink, options.encoding.clone(), None)); if let Err(e) = result { let _ = child.kill(); let _ = child.wait(); @@ -789,11 +825,16 @@ fn run_followup( Ok(match child.stdin.take() { Some(pipe) => { #[cfg(unix)] - { + let wait_writable = { use std::os::fd::AsRawFd; - ide_core::input::nonblocking(pipe.as_raw_fd())?; - } - Some(input_writer(entry.id, Box::new(pipe))?) + ide_core::input::nonblocking(pipe.as_raw_fd())? + }; + Some(input_writer( + entry.id, + Box::new(pipe), + #[cfg(unix)] + wait_writable, + )?) } None => None, }) @@ -805,10 +846,24 @@ fn run_followup( drop(pid); let readers = (|| { if let Some(out) = child.stdout.take() { - spawn_reader(entry.id, Stream::Stdout, out, tx.clone())?; + spawn_reader( + entry.id, + Stream::Stdout, + out, + tx.clone(), + #[cfg(unix)] + None, + )?; } if let Some(err) = child.stderr.take() { - spawn_reader(entry.id, Stream::Stderr, err, tx.clone())?; + spawn_reader( + entry.id, + Stream::Stderr, + err, + tx.clone(), + #[cfg(unix)] + None, + )?; } Ok::<_, std::io::Error>(()) })(); @@ -982,6 +1037,7 @@ fn spawn_reader( stream: Stream, mut pipe: impl Read + Send + 'static, tx: Sender, + #[cfg(unix)] mut wait_readable: Option std::io::Result<()> + Send>>, ) -> std::io::Result<()> { let label = if stream == Stream::Stdout { "out" } else { "err" }; std::thread::Builder::new().name(format!("openidea-run-{label}-{}", id.0)).spawn(move || { @@ -995,7 +1051,19 @@ fn spawn_reader( } } Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL), + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + #[cfg(unix)] + if let Some(wait) = &mut wait_readable { + if let Err(e) = wait() + && e.kind() != std::io::ErrorKind::Interrupted + { + tracing::warn!(%id, "run read: {e}"); + break; + } + continue; + } + std::thread::sleep(BATCH_INTERVAL); + } #[cfg(unix)] Err(e) if e.raw_os_error() == Some(libc::EIO) => break, Err(e) => { @@ -1218,10 +1286,20 @@ fn signal_number(_description: &str) -> Option { None } -fn input_writer(id: RunId, writer: Box) -> std::io::Result { - ide_core::input::InputWriter::new(format!("openidea-run-input-{}", id.0), writer, move |e| { - tracing::warn!(%id, "run input: {e}"); - }) +fn input_writer( + id: RunId, + writer: Box, + #[cfg(unix)] wait_writable: impl FnMut() -> std::io::Result<()> + Send + 'static, +) -> std::io::Result { + ide_core::input::InputWriter::new( + format!("openidea-run-input-{}", id.0), + writer, + #[cfg(unix)] + wait_writable, + move |e| { + tracing::warn!(%id, "run input: {e}"); + }, + ) } #[cfg(test)] diff --git a/crates/ide-terminal/src/lib.rs b/crates/ide-terminal/src/lib.rs index d5c70ffa8..75cb14f13 100644 --- a/crates/ide-terminal/src/lib.rs +++ b/crates/ide-terminal/src/lib.rs @@ -163,9 +163,13 @@ impl TerminalManager { apply_powershell_env(&shell, &mut cmd); #[cfg(unix)] - if let Some(fd) = pair.master.as_raw_fd() { - ide_core::input::nonblocking(fd)?; - } + let wait_writable = ide_core::input::nonblocking( + pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?, + )?; + #[cfg(unix)] + let wait_readable = ide_core::input::readable( + pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?, + )?; let mut child = pair .slave @@ -180,9 +184,15 @@ impl TerminalManager { let id = TerminalId::next(); let info = TerminalInfo { id, title: shell_title(&shell), pid: child.process_id() }; let exited = Arc::new(AtomicBool::new(false)); - let writer = ide_core::input::InputWriter::new(format!("openidea-term-write-{}", id.0), writer, move |e| { - tracing::warn!(%id, "terminal input: {e}"); - }) + let writer = ide_core::input::InputWriter::new( + format!("openidea-term-write-{}", id.0), + writer, + #[cfg(unix)] + wait_writable, + move |e| { + tracing::warn!(%id, "terminal input: {e}"); + }, + ) .inspect_err(|_| { let _ = child.kill(); let _ = child.wait(); @@ -200,9 +210,15 @@ impl TerminalManager { self.sessions.lock().insert(id, session); let (tx, rx) = mpsc::channel::(); - let spawned = spawn_reader(id, reader, tx.clone()) - .and_then(|_| spawn_waiter(id, child, tx)) - .and_then(|_| spawn_pump(id, rx, on_output, exited, self.sessions.clone(), sink, Instant::now())); + let spawned = spawn_reader( + id, + reader, + tx.clone(), + #[cfg(unix)] + wait_readable, + ) + .and_then(|_| spawn_waiter(id, child, tx)) + .and_then(|_| spawn_pump(id, rx, on_output, exited, self.sessions.clone(), sink, Instant::now())); if let Err(e) = spawned { if let Some(s) = self.sessions.lock().remove(&id) { let _ = s.killer.lock().kill(); @@ -407,7 +423,12 @@ fn terminate(session: &Session) { } } -fn spawn_reader(id: TerminalId, mut reader: Box, tx: mpsc::Sender) -> std::io::Result<()> { +fn spawn_reader( + id: TerminalId, + mut reader: Box, + tx: mpsc::Sender, + #[cfg(unix)] mut wait_readable: impl FnMut() -> std::io::Result<()> + Send + 'static, +) -> std::io::Result<()> { std::thread::Builder::new().name(format!("openidea-term-read-{}", id.0)).spawn(move || { let mut buf = vec![0u8; 16 * 1024]; loop { @@ -419,7 +440,17 @@ fn spawn_reader(id: TerminalId, mut reader: Box, tx: mpsc::Send } } Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL), + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + #[cfg(unix)] + if let Err(e) = wait_readable() + && e.kind() != std::io::ErrorKind::Interrupted + { + tracing::warn!(%id, "terminal read: {e}"); + break; + } + #[cfg(not(unix))] + std::thread::sleep(BATCH_INTERVAL); + } // EIO is how Linux/macOS report "slave side closed". #[cfg(unix)] Err(e) if e.raw_os_error() == Some(libc::EIO) => break,