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.rs1309
1 files changed, 184 insertions, 1125 deletions
diff --git a/src/pty/actor.rs b/src/pty/actor.rs
index 70634475..8c3a6b52 100644
--- a/src/pty/actor.rs
+++ b/src/pty/actor.rs
@@ -1,1172 +1,231 @@
-use std::{
- collections::VecDeque,
- io::{Read, Write},
- os::fd::{AsRawFd, OwnedFd, RawFd},
- sync::{mpsc as std_mpsc, Arc, Mutex},
- time::{Duration, Instant},
-};
+#[cfg(unix)]
+mod unix;
-use bytes::Bytes;
-use tokio::sync::mpsc::{self, error::TryRecvError as DataTryRecvError};
-use tracing::{debug, warn};
+#[cfg(unix)]
+pub(crate) use unix::*;
-use crate::pty::fd;
+#[cfg(windows)]
+mod windows {
+ use std::io::{Read, Write};
+ use std::sync::{mpsc as std_mpsc, Arc, Mutex};
+ use std::time::Duration;
-#[cfg(not(test))]
-const ACTOR_POLL_MS: i32 = 50;
-#[cfg(test)]
-const ACTOR_POLL_MS: i32 = 1000;
-const ACTOR_COMMAND_BUFFER: usize = 1024;
-const HANDOFF_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
+ use bytes::Bytes;
+ use portable_pty::{MasterPty, PtySize};
+ use tokio::sync::mpsc;
+ use tracing::{debug, warn};
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum ActorState {
- Running,
- Quiesced,
- Released,
-}
-
-pub(crate) struct PtyReadResult {
- pub terminal_responses: Vec<Bytes>,
-}
-
-impl PtyReadResult {
- #[cfg(test)]
- pub(crate) fn empty() -> Self {
- Self {
- terminal_responses: Vec::new(),
- }
- }
-}
-
-type ReadCallback = Box<dyn FnMut(&[u8]) -> PtyReadResult + Send + 'static>;
-type ReaderExitCallback = Box<dyn FnOnce() + Send + 'static>;
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-struct PtyResize {
- rows: u16,
- cols: u16,
- cell_width_px: u32,
- cell_height_px: u32,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-struct PtyResizeRequest {
- resize: PtyResize,
- terminal_responses: Vec<Bytes>,
-}
-
-#[derive(Default)]
-struct SharedPtyControls {
- resize: Option<PtyResizeRequest>,
- nudge: Option<PtyResize>,
-}
-
-pub(crate) struct PtyIoActorConfig {
- pub pane_id: u32,
- pub master_fd: OwnedFd,
- pub initially_quiesced: bool,
- pub on_read: ReadCallback,
- pub on_reader_exit: Option<ReaderExitCallback>,
-}
-
-enum PtyIoDataCommand {
- WriteUserInput(Bytes),
-}
-
-enum PtyIoControlCommand {
- BeginHandoff(std_mpsc::Sender<std::io::Result<()>>),
- DuplicateForHandoff(std_mpsc::Sender<std::io::Result<RawFd>>),
- ForegroundProcessGroup(std_mpsc::Sender<Option<u32>>),
- RollbackHandoff(std_mpsc::Sender<std::io::Result<()>>),
- ReleaseAfterCommit(std_mpsc::Sender<std::io::Result<()>>),
- Shutdown,
-}
-
-#[derive(Clone)]
-pub(crate) struct PtyIoActorHandle {
- data_tx: mpsc::Sender<PtyIoDataCommand>,
- control_tx: std_mpsc::Sender<PtyIoControlCommand>,
- wake: fd::WakeWriter,
- user_writes: Arc<Mutex<UserWriteGate>>,
- controls: Arc<Mutex<SharedPtyControls>>,
-}
-
-#[derive(Debug)]
-struct UserWriteGate {
- accepting: bool,
-}
-
-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) struct PtyReadResult {
+ pub terminal_responses: Vec<Bytes>,
}
- pub(crate) fn try_write_user_input(
- &self,
- bytes: Bytes,
- ) -> Result<(), mpsc::error::TrySendError<Bytes>> {
- let user_writes = self
- .user_writes
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- if !user_writes.accepting {
- return Err(mpsc::error::TrySendError::Closed(bytes));
- }
- match self
- .data_tx
- .try_send(PtyIoDataCommand::WriteUserInput(bytes))
- {
- Ok(()) => {
- self.wake_actor();
- Ok(())
- }
- Err(mpsc::error::TrySendError::Full(PtyIoDataCommand::WriteUserInput(bytes))) => {
- Err(mpsc::error::TrySendError::Full(bytes))
- }
- Err(mpsc::error::TrySendError::Closed(PtyIoDataCommand::WriteUserInput(bytes))) => {
- Err(mpsc::error::TrySendError::Closed(bytes))
- }
- }
- }
+ type ReadCallback = Box<dyn FnMut(&[u8]) -> PtyReadResult + Send + 'static>;
+ type ReaderExitCallback = Box<dyn FnOnce() + Send + 'static>;
- pub(crate) fn resize(
- &self,
+ #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+ struct PtyResize {
rows: u16,
cols: u16,
cell_width_px: u32,
cell_height_px: u32,
- terminal_responses: Vec<Bytes>,
- ) {
- {
- let mut controls = self
- .controls
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- controls.resize = Some(PtyResizeRequest {
- resize: PtyResize {
- rows,
- cols,
- cell_width_px,
- cell_height_px,
- },
- terminal_responses,
- });
- }
- self.wake_actor();
- }
-
- pub(crate) fn nudge_child_redraw_after_handoff(
- &self,
- rows: u16,
- cols: u16,
- cell_width_px: u32,
- cell_height_px: u32,
- ) {
- {
- let mut controls = self
- .controls
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- controls.nudge = Some(PtyResize {
- rows,
- cols,
- cell_width_px,
- cell_height_px,
- });
- }
- self.wake_actor();
}
- pub(crate) fn begin_handoff(&self, timeout: Duration) -> std::io::Result<()> {
- let (reply_tx, reply_rx) = std_mpsc::channel();
- {
- let mut user_writes = self
- .user_writes
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- user_writes.accepting = false;
- if self
- .control_tx
- .send(PtyIoControlCommand::BeginHandoff(reply_tx))
- .is_err()
- {
- user_writes.accepting = true;
- return Err(std::io::Error::new(
- std::io::ErrorKind::BrokenPipe,
- "pty actor closed",
- ));
- }
- self.wake_actor();
- }
- match reply_rx.recv_timeout(timeout) {
- Ok(Ok(())) => Ok(()),
- Ok(Err(err)) => {
- let _ = self.rollback_handoff();
- Err(err)
- }
- Err(_) => {
- let _ = self.rollback_handoff();
- Err(std::io::Error::new(
- std::io::ErrorKind::TimedOut,
- "timed out waiting for PTY actor to quiesce",
- ))
- }
- }
+ struct PtyResizeRequest {
+ resize: PtyResize,
+ terminal_responses: Vec<Bytes>,
}
- pub(crate) fn duplicate_for_handoff(&self) -> std::io::Result<RawFd> {
- let (reply_tx, reply_rx) = std_mpsc::channel();
- self.control_tx
- .send(PtyIoControlCommand::DuplicateForHandoff(reply_tx))
- .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed"))?;
- self.wake_actor();
- reply_rx.recv_timeout(Duration::from_secs(1)).map_err(|_| {
- std::io::Error::new(
- std::io::ErrorKind::TimedOut,
- "timed out waiting for PTY handoff duplicate",
- )
- })?
+ pub(crate) struct PtyIoActorConfig {
+ pub pane_id: u32,
+ pub master: Box<dyn MasterPty + Send>,
+ pub initially_quiesced: bool,
+ pub on_read: ReadCallback,
+ pub on_reader_exit: Option<ReaderExitCallback>,
}
- pub(crate) fn foreground_process_group_id(&self) -> Option<u32> {
- let (reply_tx, reply_rx) = std_mpsc::channel();
- self.control_tx
- .send(PtyIoControlCommand::ForegroundProcessGroup(reply_tx))
- .ok()?;
- self.wake_actor();
- reply_rx.recv_timeout(Duration::from_secs(1)).ok()?
+ enum PtyIoControlCommand {
+ Resize(PtyResizeRequest),
+ Shutdown,
}
- pub(crate) fn rollback_handoff(&self) -> std::io::Result<()> {
- let (reply_tx, reply_rx) = std_mpsc::channel();
- self.control_tx
- .send(PtyIoControlCommand::RollbackHandoff(reply_tx))
- .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed"))?;
- self.wake_actor();
- let result = reply_rx.recv_timeout(Duration::from_secs(1)).map_err(|_| {
- std::io::Error::new(
- std::io::ErrorKind::TimedOut,
- "timed out waiting for PTY handoff rollback",
- )
- })?;
- if result.is_ok() {
- let mut user_writes = self
- .user_writes
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- user_writes.accepting = true;
- }
- result
+ #[derive(Clone)]
+ pub(crate) struct PtyIoActorHandle {
+ data_tx: mpsc::Sender<Bytes>,
+ control_tx: std_mpsc::Sender<PtyIoControlCommand>,
+ accepting: Arc<Mutex<bool>>,
}
- pub(crate) fn release_after_commit(&self) -> std::io::Result<()> {
- {
- let mut user_writes = self
- .user_writes
+ impl PtyIoActorHandle {
+ pub(crate) async fn write_user_input(
+ &self,
+ bytes: Bytes,
+ ) -> Result<(), mpsc::error::SendError<Bytes>> {
+ if !*self
+ .accepting
.lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- user_writes.accepting = false;
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ {
+ return Err(mpsc::error::SendError(bytes));
+ }
+ self.data_tx.send(bytes).await
}
- let (reply_tx, reply_rx) = std_mpsc::channel();
- self.control_tx
- .send(PtyIoControlCommand::ReleaseAfterCommit(reply_tx))
- .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed"))?;
- self.wake_actor();
- reply_rx.recv_timeout(Duration::from_secs(1)).map_err(|_| {
- std::io::Error::new(
- std::io::ErrorKind::TimedOut,
- "timed out waiting for PTY actor release",
- )
- })?
- }
- pub(crate) fn shutdown(&self) {
- {
- let mut user_writes = self
- .user_writes
+ pub(crate) fn try_write_user_input(
+ &self,
+ bytes: Bytes,
+ ) -> Result<(), mpsc::error::TrySendError<Bytes>> {
+ if !*self
+ .accepting
.lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- user_writes.accepting = false;
- }
- if self.control_tx.send(PtyIoControlCommand::Shutdown).is_ok() {
- self.wake_actor();
- }
- }
-
- fn wake_actor(&self) {
- if let Err(err) = self.wake.wake() {
- debug!(err = %err, "failed to wake PTY actor");
- }
- }
-}
-
-pub(crate) struct PtyIoActor;
-
-impl PtyIoActor {
- pub(crate) fn spawn(config: PtyIoActorConfig) -> std::io::Result<PtyIoActorHandle> {
- Self::spawn_inner(config, None)
- }
-
- fn spawn_inner(
- config: PtyIoActorConfig,
- poll_observer: Option<std_mpsc::Sender<()>>,
- ) -> std::io::Result<PtyIoActorHandle> {
- fd::set_cloexec(config.master_fd.as_raw_fd())?;
- fd::set_nonblocking(config.master_fd.as_raw_fd())?;
-
- let (data_tx, data_rx) = mpsc::channel(ACTOR_COMMAND_BUFFER);
- let (control_tx, control_rx) = std_mpsc::channel();
- let wake_pipe = fd::create_wake_pipe()?;
- let user_writes = Arc::new(Mutex::new(UserWriteGate {
- accepting: !config.initially_quiesced,
- }));
- let controls = Arc::new(Mutex::new(SharedPtyControls::default()));
- let handle = PtyIoActorHandle {
- data_tx,
- control_tx,
- wake: wake_pipe.writer,
- user_writes,
- controls: Arc::clone(&controls),
- };
-
- let mut runner = PtyIoActorRunner {
- pane_id: config.pane_id,
- file: std::fs::File::from(config.master_fd),
- data_rx,
- control_rx,
- state: if config.initially_quiesced {
- ActorState::Quiesced
- } else {
- ActorState::Running
- },
- pending_writes: VecDeque::new(),
- current_write_offset: 0,
- wake_read_fd: wake_pipe.read_fd,
- controls,
- on_read: config.on_read,
- on_reader_exit: config.on_reader_exit,
- poll_observer,
- };
- std::thread::Builder::new()
- .name(format!("herdr-pty-{}", config.pane_id))
- .spawn(move || runner.run())
- .map_err(|err| std::io::Error::other(err.to_string()))?;
-
- Ok(handle)
- }
-
- #[cfg(test)]
- fn spawn_with_poll_observer(
- config: PtyIoActorConfig,
- poll_observer: std_mpsc::Sender<()>,
- ) -> std::io::Result<PtyIoActorHandle> {
- Self::spawn_inner(config, Some(poll_observer))
- }
-}
-
-struct PtyIoActorRunner {
- pane_id: u32,
- file: std::fs::File,
- data_rx: mpsc::Receiver<PtyIoDataCommand>,
- control_rx: std_mpsc::Receiver<PtyIoControlCommand>,
- state: ActorState,
- pending_writes: VecDeque<Bytes>,
- current_write_offset: usize,
- wake_read_fd: OwnedFd,
- controls: Arc<Mutex<SharedPtyControls>>,
- on_read: ReadCallback,
- on_reader_exit: Option<ReaderExitCallback>,
- poll_observer: Option<std_mpsc::Sender<()>>,
-}
-
-impl PtyIoActorRunner {
- fn run(&mut self) {
- let mut should_exit = false;
- while !should_exit {
- should_exit = self.drain_commands();
- if should_exit || self.state == ActorState::Released {
- break;
- }
-
- self.apply_pending_controls();
-
- if !self.pending_writes.is_empty() {
- self.flush_pending_writes_once();
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ {
+ return Err(mpsc::error::TrySendError::Closed(bytes));
}
+ self.data_tx.try_send(bytes)
+ }
- if let Some(poll_observer) = &self.poll_observer {
- let _ = poll_observer.send(());
- }
+ pub(crate) fn resize(
+ &self,
+ rows: u16,
+ cols: u16,
+ cell_width_px: u32,
+ cell_height_px: u32,
+ terminal_responses: Vec<Bytes>,
+ ) {
+ let _ = self
+ .control_tx
+ .send(PtyIoControlCommand::Resize(PtyResizeRequest {
+ resize: PtyResize {
+ rows,
+ cols,
+ cell_width_px,
+ cell_height_px,
+ },
+ terminal_responses,
+ }));
+ }
+
+ pub(crate) fn shutdown(&self) {
+ if let Ok(mut accepting) = self.accepting.lock() {
+ *accepting = false;
+ }
+ let _ = self.control_tx.send(PtyIoControlCommand::Shutdown);
+ }
+ }
+
+ pub(crate) struct PtyIoActor;
+
+ impl PtyIoActor {
+ pub(crate) fn spawn(config: PtyIoActorConfig) -> std::io::Result<PtyIoActorHandle> {
+ let PtyIoActorConfig {
+ pane_id,
+ master,
+ initially_quiesced,
+ mut on_read,
+ on_reader_exit,
+ } = config;
+
+ let mut reader = master
+ .try_clone_reader()
+ .map_err(|err| std::io::Error::other(err.to_string()))?;
+ let writer = master
+ .take_writer()
+ .map_err(|err| std::io::Error::other(err.to_string()))?;
+ let writer = Arc::new(Mutex::new(writer));
+ let (data_tx, mut data_rx) = mpsc::channel::<Bytes>(1024);
+ let (control_tx, control_rx) = std_mpsc::channel::<PtyIoControlCommand>();
+ let accepting = Arc::new(Mutex::new(!initially_quiesced));
- match fd::poll_pty_and_wake(
- self.file.as_raw_fd(),
- self.wake_read_fd.as_raw_fd(),
- self.state == ActorState::Running,
- !self.pending_writes.is_empty(),
- ACTOR_POLL_MS,
- ) {
- Ok(readiness) => {
- if readiness.wake_ready {
- if let Err(err) = fd::drain_wake_fd(self.wake_read_fd.as_raw_fd()) {
- debug!(pane = self.pane_id, err = %err, "PTY actor wake drain failed");
+ {
+ let writer = Arc::clone(&writer);
+ std::thread::spawn(move || {
+ while let Some(bytes) = data_rx.blocking_recv() {
+ if write_all_locked(&writer, &bytes).is_err() {
break;
}
- continue;
- }
- if readiness.pty_write_ready && !self.pending_writes.is_empty() {
- self.flush_pending_writes_once();
- }
- if self.state == ActorState::Running
- && readiness.pty_read_ready
- && !self.read_once()
- {
- break;
}
- }
- Err(err) => {
- debug!(pane = self.pane_id, err = %err, "PTY actor poll failed");
- break;
- }
+ debug!(pane_id, "windows pty writer thread exiting");
+ });
}
- }
- if let Some(on_reader_exit) = self.on_reader_exit.take() {
- on_reader_exit();
- }
- debug!(pane = self.pane_id, "PTY actor exiting");
- }
-
- fn drain_commands(&mut self) -> bool {
- if self.drain_control_commands() {
- return true;
- }
- self.drain_data_commands()
- }
-
- fn drain_control_commands(&mut self) -> bool {
- let mut should_exit = false;
- loop {
- match self.control_rx.try_recv() {
- Ok(command) => {
- if self.handle_control_command(command) {
- should_exit = true;
- break;
+ {
+ let writer = Arc::clone(&writer);
+ std::thread::spawn(move || {
+ let mut buf = [0u8; 8192];
+ loop {
+ match reader.read(&mut buf) {
+ Ok(0) => break,
+ Ok(n) => {
+ let result = on_read(&buf[..n]);
+ for response in result.terminal_responses {
+ if write_all_locked(&writer, &response).is_err() {
+ break;
+ }
+ }
+ }
+ Err(err) => {
+ debug!(pane_id, err = %err, "windows pty reader failed");
+ break;
+ }
+ }
}
- }
- Err(std_mpsc::TryRecvError::Empty) => break,
- Err(std_mpsc::TryRecvError::Disconnected) => {
- should_exit = true;
- break;
- }
- }
- }
- should_exit
- }
-
- fn drain_data_commands(&mut self) -> bool {
- let mut should_exit = false;
- loop {
- match self.data_rx.try_recv() {
- Ok(command) => {
- if self.handle_data_command(command) {
- should_exit = true;
- break;
+ if let Some(on_reader_exit) = on_reader_exit {
+ on_reader_exit();
}
- }
- Err(DataTryRecvError::Empty) => break,
- Err(DataTryRecvError::Disconnected) => {
- should_exit = true;
- break;
- }
+ debug!(pane_id, "windows pty reader thread exiting");
+ });
}
- }
- should_exit
- }
- fn handle_data_command(&mut self, command: PtyIoDataCommand) -> bool {
- match command {
- PtyIoDataCommand::WriteUserInput(bytes) => {
- if self.state == ActorState::Running {
- self.pending_writes.push_back(bytes);
- }
- }
- }
- false
- }
-
- fn handle_control_command(&mut self, command: PtyIoControlCommand) -> bool {
- match command {
- PtyIoControlCommand::BeginHandoff(reply) => {
- let result = self.begin_handoff();
- let _ = reply.send(result);
- }
- PtyIoControlCommand::DuplicateForHandoff(reply) => {
- let result = if self.state == ActorState::Quiesced {
- fd::duplicate_cloexec_fd(self.file.as_raw_fd())
- } else {
- Err(std::io::Error::other(
- "PTY actor must be quiesced before handoff duplication",
- ))
- };
- let _ = reply.send(result);
- }
- PtyIoControlCommand::ForegroundProcessGroup(reply) => {
- let result =
- crate::platform::foreground_process_group_id_for_tty_fd(self.file.as_raw_fd());
- let _ = reply.send(result);
- }
- PtyIoControlCommand::RollbackHandoff(reply) => {
- let result = if self.state == ActorState::Released {
- Err(std::io::Error::new(
- std::io::ErrorKind::BrokenPipe,
- "PTY actor was released before handoff rollback",
- ))
- } else {
- self.state = ActorState::Running;
- Ok(())
- };
- let _ = reply.send(result);
- }
- PtyIoControlCommand::ReleaseAfterCommit(reply) => {
- self.state = ActorState::Released;
- self.pending_writes.clear();
- let _ = reply.send(Ok(()));
- return true;
- }
- PtyIoControlCommand::Shutdown => return true,
- }
- false
- }
-
- fn begin_handoff(&mut self) -> std::io::Result<()> {
- self.drain_pre_quiesce_commands();
- self.apply_pending_controls();
- if self.state == ActorState::Released {
- return Err(std::io::Error::new(
- std::io::ErrorKind::BrokenPipe,
- "PTY actor was released before handoff quiesce",
- ));
- }
- let deadline = Instant::now() + HANDOFF_DRAIN_TIMEOUT;
- while !self.pending_writes.is_empty() {
- if Instant::now() >= deadline {
- return Err(std::io::Error::new(
- std::io::ErrorKind::TimedOut,
- "timed out draining PTY writes before handoff",
- ));
- }
- self.flush_pending_writes_once();
- }
- self.state = ActorState::Quiesced;
- Ok(())
- }
-
- fn drain_pre_quiesce_commands(&mut self) {
- while let Ok(PtyIoDataCommand::WriteUserInput(bytes)) = self.data_rx.try_recv() {
- if self.state != ActorState::Released {
- self.pending_writes.push_back(bytes);
- }
- }
- }
-
- fn apply_pending_controls(&mut self) {
- let (resize, nudge) = {
- let mut controls = self
- .controls
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner());
- (controls.resize.take(), controls.nudge.take())
- };
- if self.state == ActorState::Released {
- return;
- }
- if let Some(request) = resize {
- self.resize(request.resize);
- self.enqueue_terminal_responses(request.terminal_responses);
- }
- if let Some(nudge) = nudge {
- self.nudge(nudge);
- }
- }
-
- fn read_once(&mut self) -> bool {
- let mut buf = [0u8; 8192];
- match self.file.read(&mut buf) {
- Ok(0) => false,
- Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => true,
- Err(err) if err.kind() == std::io::ErrorKind::Interrupted => true,
- Err(err) => {
- debug!(pane = self.pane_id, err = %err, "PTY actor read failed");
- false
- }
- Ok(n) => {
- let result = (self.on_read)(&buf[..n]);
- self.enqueue_terminal_responses(result.terminal_responses);
- true
- }
- }
- }
-
- fn enqueue_terminal_responses(&mut self, terminal_responses: Vec<Bytes>) {
- if self.state == ActorState::Released {
- return;
- }
- self.pending_writes.extend(terminal_responses);
- }
-
- fn flush_pending_writes_once(&mut self) {
- while let Some(bytes) = self.pending_writes.front() {
- let chunk = &bytes[self.current_write_offset..];
- match self.file.write(chunk) {
- Ok(0) => {
- warn!(pane = self.pane_id, "PTY actor write returned zero bytes");
- return;
- }
- Ok(written) => {
- self.current_write_offset += written;
- if self.current_write_offset >= bytes.len() {
- self.pending_writes.pop_front();
- self.current_write_offset = 0;
+ {
+ let writer = Arc::clone(&writer);
+ std::thread::spawn(move || {
+ for command in control_rx {
+ match command {
+ PtyIoControlCommand::Resize(request) => {
+ let size = request.resize;
+ if let Err(err) = master.resize(PtySize {
+ rows: size.rows,
+ cols: size.cols,
+ pixel_width: size.cell_width_px.min(u16::MAX as u32) as u16,
+ pixel_height: size.cell_height_px.min(u16::MAX as u32) as u16,
+ }) {
+ warn!(pane_id, err = %err, "windows pty resize failed");
+ }
+ for response in request.terminal_responses {
+ if write_all_locked(&writer, &response).is_err() {
+ break;
+ }
+ }
+ }
+ PtyIoControlCommand::Shutdown => break,
+ }
}
- }
- Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
- let _ = fd::poll_write_ready(self.file.as_raw_fd(), ACTOR_POLL_MS);
- return;
- }
- Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return,
- Err(err) => {
- warn!(pane = self.pane_id, err = %err, "PTY actor write failed");
- self.pending_writes.clear();
- self.current_write_offset = 0;
- return;
- }
+ debug!(pane_id, "windows pty control thread exiting");
+ });
}
- }
- let _ = self.file.flush();
- }
- fn resize(&self, resize: PtyResize) {
- self.log_resize_result(fd::resize_pty_fd(
- self.file.as_raw_fd(),
- resize.rows,
- resize.cols,
- resize.cell_width_px,
- resize.cell_height_px,
- ));
- }
-
- fn nudge(&mut self, resize: PtyResize) {
- if self.state == ActorState::Released {
- return;
- }
- let nudge = if resize.rows > 2 {
- (
- resize.rows - 1,
- resize.cols,
- resize.cell_width_px,
- resize.cell_height_px,
- )
- } else {
- (
- resize.rows,
- resize.cols.saturating_sub(1).max(4),
- resize.cell_width_px,
- resize.cell_height_px,
- )
- };
- if nudge
- == (
- resize.rows,
- resize.cols,
- resize.cell_width_px,
- resize.cell_height_px,
- )
- {
- return;
- }
- self.log_resize_result(fd::resize_pty_fd(
- self.file.as_raw_fd(),
- nudge.0,
- nudge.1,
- nudge.2,
- nudge.3,
- ));
- std::thread::sleep(Duration::from_millis(30));
- self.log_resize_result(fd::resize_pty_fd(
- self.file.as_raw_fd(),
- resize.rows,
- resize.cols,
- resize.cell_width_px,
- resize.cell_height_px,
- ));
- }
-
- fn log_resize_result(&self, result: std::io::Result<()>) {
- if let Err(err) = result {
- debug!(pane = self.pane_id, err = %err, "PTY resize failed");
- }
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use std::{
- io::{Read, Write},
- os::fd::{AsRawFd, FromRawFd, IntoRawFd},
- os::unix::net::UnixStream,
- };
-
- fn test_wake_pair() -> (fd::WakeWriter, OwnedFd) {
- let pipe = fd::create_wake_pipe().expect("wake pipe");
- (pipe.writer, pipe.read_fd)
- }
-
- fn actor_with_socket_pair(
- initially_quiesced: bool,
- ) -> (PtyIoActorHandle, UnixStream, std_mpsc::Receiver<Bytes>) {
- actor_with_socket_pair_and_poll_observer(initially_quiesced, None)
- }
-
- fn actor_with_socket_pair_and_poll_observer(
- initially_quiesced: bool,
- poll_observer: Option<std_mpsc::Sender<()>>,
- ) -> (PtyIoActorHandle, UnixStream, std_mpsc::Receiver<Bytes>) {
- let (actor_socket, peer) = UnixStream::pair().expect("socket pair");
- actor_socket
- .set_nonblocking(true)
- .expect("actor socket nonblocking");
- peer.set_read_timeout(Some(Duration::from_secs(1)))
- .expect("peer timeout");
- let owned = unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) };
- let (read_tx, read_rx) = std_mpsc::channel();
- let config = PtyIoActorConfig {
- pane_id: 1,
- master_fd: owned,
- initially_quiesced,
- on_read: Box::new(move |bytes| {
- read_tx
- .send(Bytes::copy_from_slice(bytes))
- .expect("read callback receiver alive");
- PtyReadResult::empty()
- }),
- on_reader_exit: None,
- };
- let handle = if let Some(poll_observer) = poll_observer {
- PtyIoActor::spawn_with_poll_observer(config, poll_observer)
- } else {
- PtyIoActor::spawn(config)
- }
- .expect("actor spawn");
- (handle, peer, read_rx)
- }
-
- #[test]
- fn actor_writes_user_input_to_owned_fd() {
- let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
-
- handle
- .try_write_user_input(Bytes::from_static(b"hello"))
- .expect("write command accepted");
-
- let mut buf = [0u8; 5];
- peer.read_exact(&mut buf).expect("peer receives write");
- assert_eq!(&buf, b"hello");
- handle.shutdown();
- }
-
- #[test]
- fn actor_wakes_idle_poll_for_user_input() {
- let (poll_tx, poll_rx) = std_mpsc::channel();
- let (handle, mut peer, _read_rx) =
- actor_with_socket_pair_and_poll_observer(false, Some(poll_tx));
- peer.set_read_timeout(Some(Duration::from_millis(500)))
- .expect("peer timeout");
- poll_rx
- .recv_timeout(Duration::from_secs(1))
- .expect("actor entered idle poll");
-
- let start = Instant::now();
- handle
- .try_write_user_input(Bytes::from_static(b"x"))
- .expect("write command accepted");
-
- let mut buf = [0u8; 1];
- peer.read_exact(&mut buf)
- .expect("peer receives write without waiting for actor poll timeout");
- assert_eq!(&buf, b"x");
- assert!(
- start.elapsed() < Duration::from_millis(500),
- "actor write should be driven by wake fd, not the idle poll timeout"
- );
- handle.shutdown();
- }
-
- #[test]
- fn poll_ignores_pty_hup_without_pty_interest() {
- let (actor_socket, peer) = UnixStream::pair().expect("socket pair");
- actor_socket
- .set_nonblocking(true)
- .expect("actor socket nonblocking");
- drop(peer);
- let wake_pipe = fd::create_wake_pipe().expect("wake pipe");
-
- let readiness = fd::poll_pty_and_wake(
- actor_socket.as_raw_fd(),
- wake_pipe.read_fd.as_raw_fd(),
- false,
- false,
- 10,
- )
- .expect("poll succeeds");
-
- assert!(!readiness.pty_read_ready);
- assert!(!readiness.pty_write_ready);
- assert!(!readiness.wake_ready);
- }
-
- #[test]
- fn actor_delivers_fd_reads_to_callback() {
- let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
-
- peer.write_all(b"from-peer").expect("peer write");
-
- let read = read_rx
- .recv_timeout(Duration::from_secs(1))
- .expect("actor read callback");
- assert_eq!(read, Bytes::from_static(b"from-peer"));
- handle.shutdown();
- }
-
- #[test]
- fn begin_handoff_stops_reads_and_rejects_user_writes_until_rollback() {
- let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
-
- handle
- .begin_handoff(Duration::from_secs(1))
- .expect("handoff quiesced");
- assert!(handle
- .try_write_user_input(Bytes::from_static(b"blocked"))
- .is_err());
-
- peer.write_all(b"held").expect("peer write during quiesce");
- assert!(
- read_rx.recv_timeout(Duration::from_millis(150)).is_err(),
- "actor must not read while quiesced"
- );
-
- handle.rollback_handoff().expect("rollback resumes actor");
- let read = read_rx
- .recv_timeout(Duration::from_secs(1))
- .expect("actor reads held bytes after rollback");
- assert_eq!(read, Bytes::from_static(b"held"));
-
- handle
- .try_write_user_input(Bytes::from_static(b"after"))
- .expect("write accepted after rollback");
- let mut buf = [0u8; 5];
- peer.read_exact(&mut buf).expect("peer receives after");
- assert_eq!(&buf, b"after");
- handle.shutdown();
- }
-
- #[test]
- fn duplicate_for_handoff_requires_quiesced_actor() {
- let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
-
- assert!(handle.duplicate_for_handoff().is_err());
- handle
- .begin_handoff(Duration::from_secs(1))
- .expect("handoff quiesced");
- let duplicate = handle
- .duplicate_for_handoff()
- .expect("handoff duplicate created");
- assert!(duplicate >= 0);
- unsafe {
- libc::close(duplicate);
- }
- handle.rollback_handoff().expect("rollback resumes actor");
-
- peer.write_all(b"still-live").expect("peer write");
- let read = read_rx
- .recv_timeout(Duration::from_secs(1))
- .expect("actor still reads after duplicate closes");
- assert_eq!(read, Bytes::from_static(b"still-live"));
- handle.shutdown();
- }
-
- #[test]
- fn resize_and_nudge_keep_latest_request_when_command_queue_is_full() {
- let (data_tx, _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 command queue");
- let controls = Arc::new(Mutex::new(SharedPtyControls::default()));
- 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::clone(&controls),
- };
-
- handle.resize(20, 80, 8, 16, vec![Bytes::from_static(b"old")]);
- handle.resize(40, 120, 9, 18, vec![Bytes::from_static(b"new")]);
- handle.nudge_child_redraw_after_handoff(41, 121, 10, 20);
-
- let controls = controls.lock().expect("controls lock");
- assert_eq!(
- controls.resize,
- Some(PtyResizeRequest {
- resize: PtyResize {
- rows: 40,
- cols: 120,
- cell_width_px: 9,
- cell_height_px: 18,
- },
- terminal_responses: vec![Bytes::from_static(b"new")],
+ Ok(PtyIoActorHandle {
+ data_tx,
+ control_tx,
+ accepting,
})
- );
- assert_eq!(
- controls.nudge,
- Some(PtyResize {
- rows: 41,
- cols: 121,
- cell_width_px: 10,
- cell_height_px: 20,
- })
- );
- }
-
- #[test]
- fn resize_writes_terminal_responses_after_applying_resize() {
- let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
- let response = Bytes::from_static(b"\x1B[48;40;100;720;900t");
-
- handle.resize(40, 100, 9, 18, vec![response.clone()]);
-
- let mut buf = vec![0; response.len()];
- peer.read_exact(&mut buf)
- .expect("peer receives resize response");
- assert_eq!(Bytes::from(buf), response);
- 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())),
- };
-
- 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())),
- };
- 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);
- 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())),
- };
-
- 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 despite full data queue");
- }
-
- #[test]
- fn begin_handoff_drains_user_writes_already_in_command_queue() {
- let (actor_socket, mut peer) = UnixStream::pair().expect("socket pair");
- actor_socket
- .set_nonblocking(true)
- .expect("actor socket nonblocking");
- peer.set_read_timeout(Some(Duration::from_secs(1)))
- .expect("peer timeout");
- let (data_tx, data_rx) = mpsc::channel(ACTOR_COMMAND_BUFFER);
- let (_control_tx, control_rx) = std_mpsc::channel();
- data_tx
- .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
- b"queued-before-ack",
- )))
- .expect("queued write");
- let mut runner = PtyIoActorRunner {
- pane_id: 1,
- file: std::fs::File::from(unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) }),
- data_rx,
- control_rx,
- state: ActorState::Running,
- pending_writes: VecDeque::new(),
- current_write_offset: 0,
- wake_read_fd: fd::create_wake_pipe().expect("wake pipe").read_fd,
- controls: Arc::new(Mutex::new(SharedPtyControls::default())),
- on_read: Box::new(|_| PtyReadResult::empty()),
- on_reader_exit: None,
- poll_observer: None,
- };
-
- runner.begin_handoff().expect("handoff drains queued write");
-
- let mut buf = [0u8; 17];
- peer.read_exact(&mut buf)
- .expect("queued write reaches peer before quiesce ack");
- assert_eq!(&buf, b"queued-before-ack");
- assert_eq!(runner.state, ActorState::Quiesced);
+ fn write_all_locked(
+ writer: &Arc<Mutex<Box<dyn Write + Send>>>,
+ bytes: &[u8],
+ ) -> std::io::Result<()> {
+ let mut writer = writer
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ writer.write_all(bytes)?;
+ writer.flush()
}
- #[test]
- fn release_after_commit_prevents_further_io() {
- let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
-
- handle.release_after_commit().expect("actor released");
- assert!(handle
- .try_write_user_input(Bytes::from_static(b"blocked"))
- .is_err());
-
- let _ = peer.write_all(b"ignored");
- assert!(read_rx.recv_timeout(Duration::from_millis(150)).is_err());
- }
+ #[allow(dead_code)]
+ fn _assert_duration_send(_: Duration) {}
}
+
+#[cfg(windows)]
+pub(crate) use windows::*;
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-09 19:09:55 +0000