aboutsummaryrefslogtreecommitdiff
path: root/src/pty/actor.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/pty/actor.rs')
-rw-r--r--src/pty/actor.rs321
1 files changed, 282 insertions, 39 deletions
diff --git a/src/pty/actor.rs b/src/pty/actor.rs
index 6fde5051..39b1d29f 100644
--- a/src/pty/actor.rs
+++ b/src/pty/actor.rs
@@ -43,6 +43,24 @@ mod windows {
pub on_reader_exit: Option<ReaderExitCallback>,
}
+ enum PtyIoDataCommand {
+ WriteUserInput(Bytes),
+ SubmitUserInput {
+ text: Bytes,
+ enter: Bytes,
+ delay: Duration,
+ reply: std_mpsc::Sender<std::io::Result<()>>,
+ },
+ }
+
+ enum PtyIoWriteCommand {
+ Write(Bytes),
+ SubmissionPart {
+ bytes: Bytes,
+ reply: std_mpsc::Sender<std::io::Result<()>>,
+ },
+ }
+
enum PtyIoControlCommand {
Resize(PtyResizeRequest),
Shutdown,
@@ -50,40 +68,77 @@ mod windows {
#[derive(Clone)]
pub(crate) struct PtyIoActorHandle {
- data_tx: mpsc::Sender<Bytes>,
+ data_tx: mpsc::Sender<PtyIoDataCommand>,
control_tx: std_mpsc::Sender<PtyIoControlCommand>,
- write_tx: std_mpsc::Sender<Bytes>,
+ write_tx: std_mpsc::Sender<PtyIoWriteCommand>,
response_order: Arc<Mutex<()>>,
accepting: Arc<Mutex<bool>>,
}
impl PtyIoActorHandle {
- pub(crate) async fn write_user_input(
+ pub(crate) fn try_write_user_input(
&self,
bytes: Bytes,
- ) -> Result<(), mpsc::error::SendError<Bytes>> {
+ ) -> Result<(), mpsc::error::TrySendError<Bytes>> {
if !*self
.accepting
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
{
- return Err(mpsc::error::SendError(bytes));
+ return Err(mpsc::error::TrySendError::Closed(bytes));
}
- self.data_tx.send(bytes).await
+ self.data_tx
+ .try_send(PtyIoDataCommand::WriteUserInput(bytes))
+ .map_err(|err| match err {
+ mpsc::error::TrySendError::Full(command) => {
+ let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+ unreachable!("queued write returned another command")
+ };
+ mpsc::error::TrySendError::Full(bytes)
+ }
+ mpsc::error::TrySendError::Closed(command) => {
+ let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+ unreachable!("queued write returned another command")
+ };
+ mpsc::error::TrySendError::Closed(bytes)
+ }
+ })
}
- pub(crate) fn try_write_user_input(
+ pub(crate) fn queue_user_input_submission(
&self,
- bytes: Bytes,
- ) -> Result<(), mpsc::error::TrySendError<Bytes>> {
- if !*self
+ text: Bytes,
+ enter: Bytes,
+ delay: Duration,
+ ) -> std::io::Result<std_mpsc::Receiver<std::io::Result<()>>> {
+ let accepting = self
.accepting
.lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner())
- {
- return Err(mpsc::error::TrySendError::Closed(bytes));
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ if !*accepting {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::BrokenPipe,
+ "pty actor closed",
+ ));
}
- self.data_tx.try_send(bytes)
+ let (reply_tx, reply_rx) = std_mpsc::channel();
+ self.data_tx
+ .try_send(PtyIoDataCommand::SubmitUserInput {
+ text,
+ enter,
+ delay,
+ reply: reply_tx,
+ })
+ .map_err(|err| match err {
+ mpsc::error::TrySendError::Full(_) => std::io::Error::new(
+ std::io::ErrorKind::WouldBlock,
+ "pty input queue is full",
+ ),
+ mpsc::error::TrySendError::Closed(_) => {
+ std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed")
+ }
+ })?;
+ Ok(reply_rx)
}
pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option<Bytes>) {
@@ -92,7 +147,7 @@ mod windows {
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(bytes) = response().filter(|bytes| !bytes.is_empty()) {
- let _ = self.write_tx.send(bytes);
+ let _ = self.write_tx.send(PtyIoWriteCommand::Write(bytes));
}
}
@@ -143,29 +198,22 @@ mod windows {
let mut writer = master
.take_writer()
.map_err(|err| std::io::Error::other(err.to_string()))?;
- let (data_tx, mut data_rx) = mpsc::channel::<Bytes>(1024);
+ let (data_tx, mut data_rx) = mpsc::channel::<PtyIoDataCommand>(1024);
let (control_tx, control_rx) = std_mpsc::channel::<PtyIoControlCommand>();
- let (write_tx, write_rx) = std_mpsc::channel::<Bytes>();
+ let (write_tx, write_rx) = std_mpsc::channel::<PtyIoWriteCommand>();
let response_order = Arc::new(Mutex::new(()));
let accepting = Arc::new(Mutex::new(!initially_quiesced));
std::thread::spawn(move || {
- for bytes in write_rx {
- if writer.write_all(&bytes).is_err() || writer.flush().is_err() {
- break;
- }
- }
+ run_writer(&mut writer, write_rx);
debug!(pane_id, "windows pty writer thread exiting");
});
{
let write_tx = write_tx.clone();
+ let accepting = Arc::clone(&accepting);
std::thread::spawn(move || {
- while let Some(bytes) = data_rx.blocking_recv() {
- if write_tx.send(bytes).is_err() {
- break;
- }
- }
+ run_input_forwarder(&mut data_rx, write_tx, accepting);
debug!(pane_id, "windows pty input thread exiting");
});
}
@@ -183,11 +231,9 @@ mod windows {
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let result = on_read(&buf[..n]);
- if result
- .terminal_responses
- .into_iter()
- .any(|response| write_tx.send(response).is_err())
- {
+ if result.terminal_responses.into_iter().any(|response| {
+ write_tx.send(PtyIoWriteCommand::Write(response)).is_err()
+ }) {
break;
}
}
@@ -219,11 +265,9 @@ mod windows {
}) {
warn!(pane_id, err = %err, "windows pty resize failed");
}
- if request
- .terminal_responses
- .into_iter()
- .any(|response| write_tx.send(response).is_err())
- {
+ if request.terminal_responses.into_iter().any(|response| {
+ write_tx.send(PtyIoWriteCommand::Write(response)).is_err()
+ }) {
break;
}
}
@@ -244,8 +288,207 @@ mod windows {
}
}
- #[allow(dead_code)]
- fn _assert_duration_send(_: Duration) {}
+ fn run_writer(writer: &mut impl Write, write_rx: std_mpsc::Receiver<PtyIoWriteCommand>) {
+ for command in write_rx {
+ let result = match command {
+ PtyIoWriteCommand::Write(bytes) => write_and_flush(writer, &bytes),
+ PtyIoWriteCommand::SubmissionPart { bytes, reply } => {
+ let result = write_and_flush(writer, &bytes);
+ let failed = result.is_err();
+ let _ = reply.send(result);
+ if failed {
+ break;
+ }
+ continue;
+ }
+ };
+ if result.is_err() {
+ break;
+ }
+ }
+ }
+
+ fn run_input_forwarder(
+ data_rx: &mut mpsc::Receiver<PtyIoDataCommand>,
+ write_tx: std_mpsc::Sender<PtyIoWriteCommand>,
+ accepting: Arc<Mutex<bool>>,
+ ) {
+ while let Some(command) = data_rx.blocking_recv() {
+ match command {
+ PtyIoDataCommand::WriteUserInput(bytes) => {
+ if write_tx.send(PtyIoWriteCommand::Write(bytes)).is_err() {
+ break;
+ }
+ }
+ PtyIoDataCommand::SubmitUserInput {
+ text,
+ enter,
+ delay,
+ reply,
+ } => {
+ let result = write_submission_part(&write_tx, text).and_then(|()| {
+ std::thread::sleep(delay);
+ let accepting = accepting
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ if !*accepting {
+ return Err(pty_actor_closed());
+ }
+ write_submission_part(&write_tx, enter)
+ });
+ let failed = result.is_err();
+ let _ = reply.send(result);
+ if failed {
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ fn write_submission_part(
+ write_tx: &std_mpsc::Sender<PtyIoWriteCommand>,
+ bytes: Bytes,
+ ) -> std::io::Result<()> {
+ let (reply, completion) = std_mpsc::channel();
+ write_tx
+ .send(PtyIoWriteCommand::SubmissionPart { bytes, reply })
+ .map_err(|_| pty_actor_closed())?;
+ completion
+ .recv()
+ .unwrap_or_else(|_| Err(pty_actor_closed()))
+ }
+
+ fn pty_actor_closed() -> std::io::Error {
+ std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed")
+ }
+
+ fn write_and_flush(writer: &mut impl Write, bytes: &[u8]) -> std::io::Result<()> {
+ writer.write_all(bytes)?;
+ writer.flush()
+ }
+
+ #[cfg(test)]
+ mod tests {
+ use super::*;
+ use std::time::Instant;
+
+ struct RecordingWriter {
+ writes: Vec<(Vec<u8>, Instant)>,
+ flushes: Vec<Instant>,
+ fail_after: Option<usize>,
+ flushed: std_mpsc::Sender<()>,
+ }
+
+ impl Write for RecordingWriter {
+ fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+ if self.fail_after == Some(self.writes.len()) {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::BrokenPipe,
+ "writer closed",
+ ));
+ }
+ self.writes.push((bytes.to_vec(), Instant::now()));
+ Ok(bytes.len())
+ }
+
+ fn flush(&mut self) -> std::io::Result<()> {
+ self.flushes.push(Instant::now());
+ let _ = self.flushed.send(());
+ Ok(())
+ }
+ }
+
+ fn run_recorded_submission(
+ fail_after: Option<usize>,
+ delay: Duration,
+ during_delay: impl FnOnce(&std_mpsc::Sender<PtyIoWriteCommand>, &Arc<Mutex<bool>>),
+ ) -> (RecordingWriter, std::io::Result<()>) {
+ let (flushed_tx, flushed_rx) = std_mpsc::channel();
+ let mut writer = RecordingWriter {
+ writes: Vec::new(),
+ flushes: Vec::new(),
+ fail_after,
+ flushed: flushed_tx,
+ };
+ let (data_tx, mut data_rx) = mpsc::channel(2);
+ let (write_tx, write_rx) = std_mpsc::channel();
+ let (reply_tx, reply_rx) = std_mpsc::channel();
+ let accepting = Arc::new(Mutex::new(true));
+ data_tx
+ .try_send(PtyIoDataCommand::SubmitUserInput {
+ text: Bytes::from_static(b"prompt"),
+ enter: Bytes::from_static(b"\r"),
+ delay,
+ reply: reply_tx,
+ })
+ .unwrap();
+ data_tx
+ .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
+ b"user",
+ )))
+ .unwrap();
+ let writer_thread = std::thread::spawn(move || {
+ run_writer(&mut writer, write_rx);
+ writer
+ });
+ let input_write_tx = write_tx.clone();
+ let input_accepting = Arc::clone(&accepting);
+ let input_thread = std::thread::spawn(move || {
+ run_input_forwarder(&mut data_rx, input_write_tx, input_accepting)
+ });
+ flushed_rx.recv().expect("prompt was flushed");
+ during_delay(&write_tx, &accepting);
+ let result = reply_rx.recv().expect("writer reports submission");
+ drop(data_tx);
+ input_thread.join().expect("input thread joins");
+ drop(write_tx);
+ (writer_thread.join().expect("writer thread joins"), result)
+ }
+
+ #[test]
+ fn submission_sequences_user_input_but_allows_terminal_responses() {
+ let delay = Duration::from_millis(30);
+ let (writer, result) = run_recorded_submission(None, delay, |write_tx, _| {
+ write_tx
+ .send(PtyIoWriteCommand::Write(Bytes::from_static(b"response")))
+ .unwrap();
+ });
+ result.expect("submission succeeds");
+
+ assert_eq!(writer.writes[0].0, b"prompt");
+ assert_eq!(writer.writes[1].0, b"response");
+ assert_eq!(writer.writes[2].0, b"\r");
+ assert_eq!(writer.writes[3].0, b"user");
+ assert!(writer.writes[2].1.duration_since(writer.flushes[0]) >= delay);
+ }
+
+ #[test]
+ fn submission_returns_enter_write_failure() {
+ let (_writer, result) = run_recorded_submission(Some(1), Duration::ZERO, |_, _| {});
+ let err = result.expect_err("enter failure reaches caller");
+
+ assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+ }
+
+ #[test]
+ fn shutdown_during_submission_delay_cancels_enter() {
+ let (writer, result) =
+ run_recorded_submission(None, Duration::from_millis(30), |_, accepting| {
+ *accepting.lock().unwrap() = false;
+ });
+ let err = result.expect_err("shutdown cancels enter");
+ assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+ assert_eq!(
+ writer
+ .writes
+ .iter()
+ .map(|write| write.0.as_slice())
+ .collect::<Vec<_>>(),
+ vec![b"prompt"]
+ );
+ }
+ }
}
#[cfg(windows)]
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-07 07:43:42 +0000