aboutsummaryrefslogtreecommitdiff
path: root/src/pty
diff options
context:
space:
mode:
Diffstat (limited to 'src/pty')
-rw-r--r--src/pty/actor.rs321
-rw-r--r--src/pty/actor/unix.rs607
2 files changed, 725 insertions, 203 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)]
diff --git a/src/pty/actor/unix.rs b/src/pty/actor/unix.rs
index 599f02cb..d7a8bb78 100644
--- a/src/pty/actor/unix.rs
+++ b/src/pty/actor/unix.rs
@@ -73,6 +73,12 @@ pub(crate) struct PtyIoActorConfig {
enum PtyIoDataCommand {
WriteUserInput(Bytes),
+ SubmitUserInput {
+ text: Bytes,
+ enter: Bytes,
+ delay: Duration,
+ reply: std_mpsc::Sender<std::io::Result<()>>,
+ },
}
enum PtyIoControlCommand {
@@ -100,37 +106,6 @@ struct UserWriteGate {
}
impl PtyIoActorHandle {
- pub(crate) async fn write_user_input(
- &self,
- bytes: Bytes,
- ) -> Result<(), mpsc::error::SendError<Bytes>> {
- {
- let user_writes = self
- .user_writes
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- if !user_writes.accepting {
- return Err(mpsc::error::SendError(bytes));
- }
- }
-
- let permit = match self.data_tx.reserve().await {
- Ok(permit) => permit,
- Err(_) => return Err(mpsc::error::SendError(bytes)),
- };
-
- let user_writes = self
- .user_writes
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- if !user_writes.accepting {
- return Err(mpsc::error::SendError(bytes));
- }
- permit.send(PtyIoDataCommand::WriteUserInput(bytes));
- self.wake_actor();
- Ok(())
- }
-
pub(crate) fn try_write_user_input(
&self,
bytes: Bytes,
@@ -150,15 +125,57 @@ impl PtyIoActorHandle {
self.wake_actor();
Ok(())
}
- Err(mpsc::error::TrySendError::Full(PtyIoDataCommand::WriteUserInput(bytes))) => {
+ Err(mpsc::error::TrySendError::Full(command)) => {
+ let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+ unreachable!("queued write returned another command")
+ };
Err(mpsc::error::TrySendError::Full(bytes))
}
- Err(mpsc::error::TrySendError::Closed(PtyIoDataCommand::WriteUserInput(bytes))) => {
+ Err(mpsc::error::TrySendError::Closed(command)) => {
+ let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+ unreachable!("queued write returned another command")
+ };
Err(mpsc::error::TrySendError::Closed(bytes))
}
}
}
+ pub(crate) fn queue_user_input_submission(
+ &self,
+ text: Bytes,
+ enter: Bytes,
+ delay: Duration,
+ ) -> std::io::Result<std_mpsc::Receiver<std::io::Result<()>>> {
+ let user_writes = self
+ .user_writes
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ if !user_writes.accepting {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::BrokenPipe,
+ "pty actor closed",
+ ));
+ }
+ 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")
+ }
+ })?;
+ self.wake_actor();
+ Ok(reply_rx)
+ }
+
pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option<Bytes>) {
let _order = self
.response_order
@@ -232,6 +249,12 @@ impl PtyIoActorHandle {
.user_writes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
+ if !user_writes.accepting {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::WouldBlock,
+ "PTY handoff is already in progress",
+ ));
+ }
user_writes.accepting = false;
if self
.control_tx
@@ -391,6 +414,8 @@ impl PtyIoActor {
},
pending_writes: VecDeque::new(),
current_write_offset: 0,
+ active_submission: None,
+ pending_handoff: None,
wake_read_fd: wake_pipe.read_fd,
controls,
response_order,
@@ -421,8 +446,10 @@ struct PtyIoActorRunner {
data_rx: mpsc::Receiver<PtyIoDataCommand>,
control_rx: std_mpsc::Receiver<PtyIoControlCommand>,
state: ActorState,
- pending_writes: VecDeque<Bytes>,
+ pending_writes: VecDeque<PendingWrite>,
current_write_offset: usize,
+ active_submission: Option<ActiveSubmission>,
+ pending_handoff: Option<std_mpsc::Sender<std::io::Result<()>>>,
wake_read_fd: OwnedFd,
controls: Arc<Mutex<SharedPtyControls>>,
response_order: Arc<Mutex<()>>,
@@ -431,10 +458,47 @@ struct PtyIoActorRunner {
poll_observer: Option<std_mpsc::Sender<()>>,
}
+struct ActiveSubmission {
+ enter: Bytes,
+ delay: Duration,
+ phase: SubmissionPhase,
+ reply: std_mpsc::Sender<std::io::Result<()>>,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct PendingWrite {
+ bytes: Bytes,
+ boundary: Option<SubmissionBoundary>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum SubmissionBoundary {
+ Text,
+ Enter,
+}
+
+enum SubmissionPhase {
+ WritingText,
+ WaitingUntil(Instant),
+ WritingEnter,
+}
+
impl PtyIoActorRunner {
fn enqueue_write(&mut self, bytes: Bytes) {
if !bytes.is_empty() {
- self.pending_writes.push_back(bytes);
+ self.pending_writes.push_back(PendingWrite {
+ bytes,
+ boundary: None,
+ });
+ }
+ }
+
+ fn enqueue_submission_write(&mut self, bytes: Bytes, boundary: SubmissionBoundary) {
+ if !bytes.is_empty() {
+ self.pending_writes.push_back(PendingWrite {
+ bytes,
+ boundary: Some(boundary),
+ });
}
}
@@ -449,7 +513,18 @@ impl PtyIoActorRunner {
self.apply_pending_controls();
if !self.pending_writes.is_empty() {
- self.flush_pending_writes_once();
+ match self.flush_pending_writes_once() {
+ Ok(Some(boundary)) => self.complete_submission_boundary(boundary),
+ Ok(None) => {}
+ Err(err) => {
+ self.fail_active_submission(err);
+ break;
+ }
+ }
+ }
+ self.schedule_submission_enter();
+ if self.active_submission.is_none() && self.pending_handoff.is_some() {
+ continue;
}
if let Some(poll_observer) = &self.poll_observer {
@@ -461,7 +536,7 @@ impl PtyIoActorRunner {
self.wake_read_fd.as_raw_fd(),
self.state == ActorState::Running,
!self.pending_writes.is_empty(),
- ACTOR_IDLE_POLL_MS,
+ self.poll_timeout_ms(),
) {
Ok(readiness) => {
if readiness.wake_ready {
@@ -478,7 +553,14 @@ impl PtyIoActorRunner {
break;
}
if readiness.pty_write_ready && !self.pending_writes.is_empty() {
- self.flush_pending_writes_once();
+ match self.flush_pending_writes_once() {
+ Ok(Some(boundary)) => self.complete_submission_boundary(boundary),
+ Ok(None) => {}
+ Err(err) => {
+ self.fail_active_submission(err);
+ break;
+ }
+ }
}
}
Err(err) => {
@@ -488,6 +570,10 @@ impl PtyIoActorRunner {
}
}
+ self.fail_active_submission(std::io::Error::new(
+ std::io::ErrorKind::BrokenPipe,
+ "PTY actor closed during input submission",
+ ));
if let Some(on_reader_exit) = self.on_reader_exit.take() {
on_reader_exit();
}
@@ -498,6 +584,13 @@ impl PtyIoActorRunner {
if self.drain_control_commands() {
return true;
}
+ if self.active_submission.is_some() {
+ return false;
+ }
+ if let Some(reply) = self.pending_handoff.take() {
+ self.defer_or_begin_handoff(reply);
+ return false;
+ }
self.drain_data_commands()
}
@@ -530,6 +623,9 @@ impl PtyIoActorRunner {
should_exit = true;
break;
}
+ if self.active_submission.is_some() {
+ break;
+ }
}
Err(DataTryRecvError::Empty) => break,
Err(DataTryRecvError::Disconnected) => {
@@ -548,6 +644,32 @@ impl PtyIoActorRunner {
self.enqueue_write(bytes);
}
}
+ PtyIoDataCommand::SubmitUserInput {
+ text,
+ enter,
+ delay,
+ reply,
+ } => {
+ if self.state == ActorState::Running {
+ let phase = if text.is_empty() {
+ SubmissionPhase::WaitingUntil(Instant::now() + delay)
+ } else {
+ self.enqueue_submission_write(text, SubmissionBoundary::Text);
+ SubmissionPhase::WritingText
+ };
+ self.active_submission = Some(ActiveSubmission {
+ enter,
+ delay,
+ phase,
+ reply,
+ });
+ } else {
+ let _ = reply.send(Err(std::io::Error::new(
+ std::io::ErrorKind::BrokenPipe,
+ "pty actor is not accepting input",
+ )));
+ }
+ }
}
false
}
@@ -555,8 +677,7 @@ impl PtyIoActorRunner {
fn handle_control_command(&mut self, command: PtyIoControlCommand) -> bool {
match command {
PtyIoControlCommand::BeginHandoff(reply) => {
- let result = self.begin_handoff();
- let _ = reply.send(result);
+ self.defer_or_begin_handoff(reply);
}
PtyIoControlCommand::DuplicateForHandoff(reply) => {
let result = if self.state == ActorState::Quiesced {
@@ -574,6 +695,7 @@ impl PtyIoActorRunner {
let _ = reply.send(result);
}
PtyIoControlCommand::RollbackHandoff(reply) => {
+ self.pending_handoff.take();
let result = if self.state == ActorState::Released {
Err(std::io::Error::new(
std::io::ErrorKind::BrokenPipe,
@@ -596,8 +718,26 @@ impl PtyIoActorRunner {
false
}
+ fn defer_or_begin_handoff(&mut self, reply: std_mpsc::Sender<std::io::Result<()>>) {
+ if self.active_submission.is_none() {
+ self.drain_pre_quiesce_commands();
+ }
+ if self.active_submission.is_some() {
+ self.pending_handoff = Some(reply);
+ } else {
+ let result = self.begin_handoff();
+ let _ = reply.send(result);
+ }
+ }
+
fn begin_handoff(&mut self) -> std::io::Result<()> {
self.drain_pre_quiesce_commands();
+ if self.active_submission.is_some() {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::WouldBlock,
+ "PTY input submission is still in progress",
+ ));
+ }
self.apply_pending_controls();
if self.state == ActorState::Released {
return Err(std::io::Error::new(
@@ -606,7 +746,7 @@ impl PtyIoActorRunner {
));
}
let deadline = Instant::now() + HANDOFF_DRAIN_TIMEOUT;
- self.flush_pending_writes_once();
+ let _ = self.flush_pending_writes_once()?;
while !self.pending_writes.is_empty() {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
@@ -633,7 +773,7 @@ impl PtyIoActorRunner {
));
}
if readiness.pty_write_ready {
- self.flush_pending_writes_once();
+ let _ = self.flush_pending_writes_once()?;
}
}
self.state = ActorState::Quiesced;
@@ -641,9 +781,12 @@ impl PtyIoActorRunner {
}
fn drain_pre_quiesce_commands(&mut self) {
- while let Ok(PtyIoDataCommand::WriteUserInput(bytes)) = self.data_rx.try_recv() {
- if self.state != ActorState::Released {
- self.enqueue_write(bytes);
+ while let Ok(command) = self.data_rx.try_recv() {
+ if self.handle_data_command(command) {
+ break;
+ }
+ if self.active_submission.is_some() {
+ break;
}
}
}
@@ -717,32 +860,100 @@ impl PtyIoActorRunner {
}
}
- fn flush_pending_writes_once(&mut self) {
- while let Some(bytes) = self.pending_writes.front() {
- let chunk = &bytes[self.current_write_offset..];
+ fn complete_submission_boundary(&mut self, boundary: SubmissionBoundary) {
+ match boundary {
+ SubmissionBoundary::Text => {
+ let Some(submission) = self.active_submission.as_mut() else {
+ return;
+ };
+ debug_assert!(matches!(submission.phase, SubmissionPhase::WritingText));
+ submission.phase = SubmissionPhase::WaitingUntil(Instant::now() + submission.delay);
+ }
+ SubmissionBoundary::Enter => {
+ let Some(submission) = self.active_submission.take() else {
+ return;
+ };
+ debug_assert!(matches!(submission.phase, SubmissionPhase::WritingEnter));
+ let _ = submission.reply.send(Ok(()));
+ }
+ }
+ }
+
+ fn schedule_submission_enter(&mut self) {
+ let Some(ActiveSubmission {
+ enter,
+ phase: SubmissionPhase::WaitingUntil(deadline),
+ ..
+ }) = self.active_submission.as_ref()
+ else {
+ return;
+ };
+ if Instant::now() >= *deadline {
+ let enter = enter.clone();
+ if enter.is_empty() {
+ let submission = self.active_submission.take().unwrap();
+ let _ = submission.reply.send(Ok(()));
+ } else {
+ self.active_submission.as_mut().unwrap().phase = SubmissionPhase::WritingEnter;
+ self.enqueue_submission_write(enter, SubmissionBoundary::Enter);
+ }
+ }
+ }
+
+ fn poll_timeout_ms(&self) -> i32 {
+ let Some(ActiveSubmission {
+ phase: SubmissionPhase::WaitingUntil(deadline),
+ ..
+ }) = self.active_submission.as_ref()
+ else {
+ return ACTOR_IDLE_POLL_MS;
+ };
+ deadline
+ .saturating_duration_since(Instant::now())
+ .as_millis()
+ .max(1)
+ .min(ACTOR_IDLE_POLL_MS as u128) as i32
+ }
+
+ fn fail_active_submission(&mut self, err: std::io::Error) {
+ if let Some(submission) = self.active_submission.take() {
+ let _ = submission.reply.send(Err(err));
+ }
+ }
+
+ fn flush_pending_writes_once(&mut self) -> std::io::Result<Option<SubmissionBoundary>> {
+ while let Some(write) = self.pending_writes.front() {
+ let chunk = &write.bytes[self.current_write_offset..];
match self.file.write(chunk) {
Ok(0) => {
- warn!(pane = self.pane_id, "PTY actor write returned zero bytes");
- return;
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::WriteZero,
+ "PTY actor write returned zero bytes",
+ ));
}
Ok(written) => {
self.current_write_offset += written;
- if self.current_write_offset >= bytes.len() {
- self.pending_writes.pop_front();
+ if self.current_write_offset >= write.bytes.len() {
+ let completed = self.pending_writes.pop_front().unwrap();
self.current_write_offset = 0;
+ if let Some(boundary) = completed.boundary {
+ self.file.flush()?;
+ return Ok(Some(boundary));
+ }
}
}
- Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => return,
- Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return,
+ Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => return Ok(None),
+ Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return Ok(None),
Err(err) => {
warn!(pane = self.pane_id, err = %err, "PTY actor write failed");
self.pending_writes.clear();
self.current_write_offset = 0;
- return;
+ return Err(err);
}
}
}
- let _ = self.file.flush();
+ self.file.flush()?;
+ Ok(None)
}
fn resize(&self, resize: PtyResize) {
@@ -879,6 +1090,8 @@ mod tests {
state: ActorState::Running,
pending_writes: VecDeque::new(),
current_write_offset: 0,
+ active_submission: None,
+ pending_handoff: None,
wake_read_fd: wake_pipe.read_fd,
controls: Arc::new(Mutex::new(SharedPtyControls::default())),
response_order: Arc::new(Mutex::new(())),
@@ -899,6 +1112,22 @@ mod tests {
}
#[test]
+ fn submission_boundary_does_not_wait_for_following_protocol_write() {
+ let (mut runner, _peer) = actor_runner_for_unit_test();
+ runner.enqueue_submission_write(Bytes::from_static(b"prompt"), SubmissionBoundary::Text);
+ runner.enqueue_write(Bytes::from_static(b"response"));
+
+ assert_eq!(
+ runner.flush_pending_writes_once().unwrap(),
+ Some(SubmissionBoundary::Text)
+ );
+ assert_eq!(
+ runner.pending_writes[0].bytes,
+ Bytes::from_static(b"response")
+ );
+ }
+
+ #[test]
fn actor_writes_user_input_to_owned_fd() {
let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
@@ -913,6 +1142,127 @@ mod tests {
}
#[test]
+ fn actor_delays_enter_from_completed_prompt_write() {
+ let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+ let text = Bytes::from(vec![b'x'; 4 * 1024 * 1024]);
+ let text_len = text.len();
+ let delay = Duration::from_millis(200);
+ let reader = std::thread::spawn(move || {
+ std::thread::sleep(delay);
+ let mut received = vec![0; text_len];
+ peer.read_exact(&mut received)
+ .expect("peer receives prompt");
+ let prompt_completed = Instant::now();
+ let mut enter = [0; 1];
+ peer.read_exact(&mut enter).expect("peer receives enter");
+ let enter_received = Instant::now();
+ let mut user = [0; 4];
+ peer.read_exact(&mut user)
+ .expect("peer receives queued input");
+ (prompt_completed, enter_received, enter, user)
+ });
+
+ let completion = handle
+ .queue_user_input_submission(text, Bytes::from_static(b"\r"), delay)
+ .expect("submission queues");
+ handle
+ .try_write_user_input(Bytes::from_static(b"user"))
+ .expect("ordinary input queues behind submission");
+ completion
+ .recv()
+ .expect("actor reports submission")
+ .expect("submission completes");
+ let (prompt_completed, enter_received, enter, user) = reader.join().expect("reader joins");
+
+ assert_eq!(enter, *b"\r");
+ assert_eq!(user, *b"user");
+ assert!(enter_received.duration_since(prompt_completed) >= delay / 2);
+
+ let err = match handle.queue_user_input_submission(
+ Bytes::from_static(b"prompt"),
+ Bytes::from_static(b"\r"),
+ Duration::ZERO,
+ ) {
+ Ok(completion) => completion
+ .recv()
+ .expect("actor reports submission")
+ .expect_err("closed PTY rejects submission"),
+ Err(err) => err,
+ };
+
+ assert!(matches!(
+ err.kind(),
+ std::io::ErrorKind::BrokenPipe
+ | std::io::ErrorKind::ConnectionReset
+ | std::io::ErrorKind::WriteZero
+ ));
+ }
+
+ #[test]
+ fn actor_completes_empty_submission_parts() {
+ let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+ peer.set_read_timeout(Some(Duration::from_secs(1)))
+ .expect("peer timeout");
+
+ let completion = handle
+ .queue_user_input_submission(Bytes::new(), Bytes::from_static(b"\r"), Duration::ZERO)
+ .expect("empty prompt submission queues");
+ let mut enter = [0; 1];
+ peer.read_exact(&mut enter)
+ .expect("peer receives enter for empty prompt");
+ assert_eq!(enter, *b"\r");
+ completion
+ .recv_timeout(Duration::from_secs(1))
+ .expect("actor reports empty prompt submission")
+ .expect("empty prompt submission completes");
+
+ let completion = handle
+ .queue_user_input_submission(
+ Bytes::from_static(b"prompt"),
+ Bytes::new(),
+ Duration::from_millis(40),
+ )
+ .expect("empty enter submission queues");
+ let handoff_handle = handle.clone();
+ let handoff =
+ std::thread::spawn(move || handoff_handle.begin_handoff(Duration::from_millis(250)));
+ let mut prompt = [0; 6];
+ peer.read_exact(&mut prompt)
+ .expect("peer receives prompt before empty enter");
+ assert_eq!(&prompt, b"prompt");
+ completion
+ .recv_timeout(Duration::from_secs(1))
+ .expect("actor reports empty enter submission")
+ .expect("empty enter submission completes");
+ handoff
+ .join()
+ .expect("handoff thread joins")
+ .expect("handoff resumes without an idle poll after submission");
+ handle.shutdown();
+ }
+
+ #[test]
+ fn actor_reports_peer_closure_during_submission_delay() {
+ let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+ let completion = handle
+ .queue_user_input_submission(
+ Bytes::from_static(b"prompt"),
+ Bytes::from_static(b"\r"),
+ Duration::from_secs(1),
+ )
+ .expect("submission queues");
+ let mut prompt = [0; 6];
+ peer.read_exact(&mut prompt).expect("peer receives prompt");
+ drop(peer);
+
+ let err = completion
+ .recv_timeout(Duration::from_secs(1))
+ .expect("actor reports peer closure")
+ .expect_err("peer closure fails the active submission");
+ assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+ }
+
+ #[test]
fn actor_wakes_idle_poll_for_user_input() {
let (poll_tx, poll_rx) = std_mpsc::channel();
let (handle, mut peer, _read_rx) =
@@ -976,9 +1326,9 @@ mod tests {
.expect("actor spawn");
let marker = Bytes::from_static(b"queued-input");
- handle
- .try_write_user_input(marker.clone())
- .expect("write command accepted");
+ let completion = handle
+ .queue_user_input_submission(marker.clone(), Bytes::from_static(b"\r"), Duration::ZERO)
+ .expect("submission accepted");
const OUTPUT_LEN: usize = 128 * 1024;
let mut peer_writer = peer.try_clone().expect("clone peer writer");
@@ -1004,11 +1354,27 @@ mod tests {
assert_eq!(output_len, OUTPUT_LEN);
output_writer.join().expect("output writer joins");
- let mut received_input = vec![0; prefilled + marker.len()];
+ let handoff_handle = handle.clone();
+ let handoff =
+ std::thread::spawn(move || handoff_handle.begin_handoff(Duration::from_secs(1)));
+
+ let mut received_input = vec![0; prefilled + marker.len() + 1];
peer.read_exact(&mut received_input)
.expect("peer receives prefill and queued input");
assert!(received_input[..prefilled].iter().all(|byte| *byte == 0xAA));
- assert_eq!(&received_input[prefilled..], marker.as_ref());
+ assert_eq!(
+ &received_input[prefilled..prefilled + marker.len()],
+ marker.as_ref()
+ );
+ assert_eq!(received_input.last(), Some(&b'\r'));
+ completion
+ .recv_timeout(Duration::from_secs(1))
+ .expect("actor reports submission")
+ .expect("submission completes");
+ handoff
+ .join()
+ .expect("handoff thread joins")
+ .expect("handoff waits for submission");
handle.shutdown();
}
@@ -1080,6 +1446,10 @@ mod tests {
handle
.begin_handoff(Duration::from_secs(1))
.expect("handoff quiesced");
+ let err = handle
+ .begin_handoff(Duration::from_secs(1))
+ .expect_err("concurrent handoff rejected");
+ assert_eq!(err.kind(), std::io::ErrorKind::WouldBlock);
assert!(handle
.try_write_user_input(Bytes::from_static(b"blocked"))
.is_err());
@@ -1205,6 +1575,8 @@ mod tests {
state: ActorState::Running,
pending_writes: VecDeque::new(),
current_write_offset: 0,
+ active_submission: None,
+ pending_handoff: None,
wake_read_fd: wake_pipe.read_fd,
controls: Arc::clone(&controls),
response_order: Arc::clone(&response_order),
@@ -1251,8 +1623,14 @@ mod tests {
assert_eq!(
runner.pending_writes,
VecDeque::from([
- Bytes::from_static(b"live-light"),
- Bytes::from_static(b"query-light"),
+ PendingWrite {
+ bytes: Bytes::from_static(b"live-light"),
+ boundary: None,
+ },
+ PendingWrite {
+ bytes: Bytes::from_static(b"query-light"),
+ boundary: None,
+ },
])
);
}
@@ -1271,107 +1649,6 @@ mod tests {
handle.shutdown();
}
- #[tokio::test]
- async fn async_user_input_waits_for_queue_capacity() {
- let (data_tx, mut data_rx) = mpsc::channel(1);
- let (control_tx, _control_rx) = std_mpsc::channel();
- data_tx
- .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
- b"fill",
- )))
- .expect("fill data queue");
- let (wake, _wake_read_fd) = test_wake_pair();
- let handle = PtyIoActorHandle {
- data_tx,
- control_tx,
- wake,
- user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
- controls: Arc::new(Mutex::new(SharedPtyControls::default())),
- response_order: Arc::new(Mutex::new(())),
- };
-
- let write = tokio::spawn(async move {
- handle
- .write_user_input(Bytes::from_static(b"wait-for-capacity"))
- .await
- });
- tokio::time::sleep(Duration::from_millis(50)).await;
- assert!(
- !write.is_finished(),
- "async input should wait for queue capacity"
- );
-
- assert!(matches!(
- data_rx.recv().await,
- Some(PtyIoDataCommand::WriteUserInput(_))
- ));
- write
- .await
- .expect("write task joins")
- .expect("write succeeds after capacity opens");
- match data_rx.recv().await {
- Some(PtyIoDataCommand::WriteUserInput(bytes)) => {
- assert_eq!(bytes, Bytes::from_static(b"wait-for-capacity"));
- }
- _ => panic!("expected queued user input"),
- }
- }
-
- #[tokio::test]
- async fn async_user_input_waiting_for_capacity_is_rejected_after_handoff_begins() {
- let (data_tx, mut data_rx) = mpsc::channel(1);
- let (control_tx, control_rx) = std_mpsc::channel();
- data_tx
- .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
- b"fill",
- )))
- .expect("fill data queue");
- let (wake, _wake_read_fd) = test_wake_pair();
- let handle = PtyIoActorHandle {
- data_tx,
- control_tx,
- wake,
- user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
- controls: Arc::new(Mutex::new(SharedPtyControls::default())),
- response_order: Arc::new(Mutex::new(())),
- };
- let write_handle = handle.clone();
- let write = tokio::spawn(async move {
- write_handle
- .write_user_input(Bytes::from_static(b"after-handoff-start"))
- .await
- });
- tokio::time::sleep(Duration::from_millis(50)).await;
-
- let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1)));
- match control_rx
- .recv_timeout(Duration::from_secs(1))
- .expect("handoff control command")
- {
- PtyIoControlCommand::BeginHandoff(reply) => {
- reply.send(Ok(())).expect("handoff waiter alive");
- }
- _ => panic!("expected begin handoff command"),
- }
- handoff
- .join()
- .expect("handoff thread joins")
- .expect("handoff succeeds");
- assert!(matches!(
- data_rx.recv().await,
- Some(PtyIoDataCommand::WriteUserInput(_))
- ));
-
- let err = write.await.expect("write task joins").expect_err(
- "write waiting for capacity must be rejected after handoff closes the input gate",
- );
- assert_eq!(err.0, Bytes::from_static(b"after-handoff-start"));
- match tokio::time::timeout(Duration::from_millis(50), data_rx.recv()).await {
- Err(_) | Ok(None) => {}
- Ok(Some(_)) => panic!("rejected write must not be queued"),
- }
- }
-
#[test]
fn handoff_control_is_not_blocked_by_full_data_queue() {
let (data_tx, _data_rx) = mpsc::channel(1);
@@ -1431,6 +1708,8 @@ mod tests {
state: ActorState::Running,
pending_writes: VecDeque::new(),
current_write_offset: 0,
+ active_submission: None,
+ pending_handoff: None,
wake_read_fd: fd::create_wake_pipe().expect("wake pipe").read_fd,
controls: Arc::new(Mutex::new(SharedPtyControls::default())),
response_order: Arc::new(Mutex::new(())),
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-06 22:17:24 +0000