#![allow(unsafe_code)]
use std::io::{Read, Write};
use std::sync::{
Arc, Condvar, Mutex,
atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering},
};
use std::thread::JoinHandle;
use std::time::Duration;
#[cfg(unix)]
use std::fs::File;
#[cfg(unix)]
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
use portable_pty::{CommandBuilder, PtySize, native_pty_system};
use super::events::{TerminalKeyModes, key_event_to_bytes};
use super::screen::TerminalCellSize;
use crate::core::event::KeyEvent;
#[derive(thiserror::Error, Debug)]
pub enum TerminalPtyError {
#[error("pty initialization failed: {0}")]
Setup(String),
#[error("failed to clone pty reader: {0}")]
Reader(String),
#[error("failed to acquire pty writer: {0}")]
Writer(String),
#[error("failed to spawn pty command: {0}")]
Spawn(String),
}
fn default_shell_command() -> String {
#[cfg(windows)]
{
std::env::var("COMSPEC")
.ok()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "cmd.exe".to_string())
}
#[cfg(not(windows))]
{
std::env::var("SHELL")
.ok()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "/bin/sh".to_string())
}
}
#[cfg(windows)]
fn prime_conpty_cursor(writer: &mut dyn Write) -> std::io::Result<()> {
writer.write_all(b"\x1b[1;1R")?;
writer.flush()
}
#[cfg(not(windows))]
fn prime_conpty_cursor(_writer: &mut dyn Write) -> std::io::Result<()> {
Ok(())
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TerminalPtyConfig {
pub(crate) command: Arc<str>,
pub(crate) args: Vec<Arc<str>>,
pub(crate) cols: u16,
pub(crate) rows: u16,
pub(crate) cwd: Option<Arc<str>>,
pub(crate) term: Arc<str>,
pub(crate) env: Vec<(Arc<str>, Arc<str>)>,
pub(crate) cell: TerminalCellSize,
}
impl Default for TerminalPtyConfig {
fn default() -> Self {
let shell = default_shell_command();
Self {
command: shell.into(),
args: Vec::new(),
cols: 120,
rows: 32,
cwd: None,
term: Arc::from("xterm-256color"),
env: vec![(Arc::from("COLORTERM"), Arc::from("truecolor"))],
cell: TerminalCellSize::default(),
}
}
}
impl TerminalPtyConfig {
pub fn cell_size(mut self, cell: TerminalCellSize) -> Self {
self.cell = cell;
self
}
pub fn new(command: impl Into<Arc<str>>) -> Self {
Self {
command: command.into(),
..Self::default()
}
}
pub fn arg(mut self, arg: impl Into<Arc<str>>) -> Self {
self.args.push(arg.into());
self
}
pub fn args<I>(mut self, args: I) -> Self
where
I: IntoIterator<Item = Arc<str>>,
{
self.args = args.into_iter().collect();
self
}
pub fn size(mut self, cols: u16, rows: u16) -> Self {
self.cols = cols.max(1);
self.rows = rows.max(1);
self
}
pub fn cwd(mut self, cwd: impl Into<Arc<str>>) -> Self {
self.cwd = Some(cwd.into());
self
}
pub fn term(mut self, term: impl Into<Arc<str>>) -> Self {
self.term = term.into();
self
}
pub fn env(mut self, key: impl Into<Arc<str>>, value: impl Into<Arc<str>>) -> Self {
self.env.push((key.into(), value.into()));
self
}
}
pub struct TerminalPty {
inner: Arc<TerminalPtyInner>,
}
impl Clone for TerminalPty {
fn clone(&self) -> Self {
self.inner.handle_count.fetch_add(1, Ordering::AcqRel);
Self {
inner: self.inner.clone(),
}
}
}
const EXIT_DRAIN_GRACE: Duration = Duration::from_secs(2);
#[derive(Default)]
struct ExitSync {
state: Mutex<ExitState>,
signal: Condvar,
}
#[derive(Default)]
struct ExitState {
code: Option<i32>,
emitted: bool,
stopped: bool,
}
impl ExitSync {
fn publish(&self, code: i32) {
if let Ok(mut state) = self.state.lock() {
state.code = Some(code);
}
self.signal.notify_all();
}
fn stop(&self) {
if let Ok(mut state) = self.state.lock() {
state.stopped = true;
}
self.signal.notify_all();
}
fn claim(&self) -> Option<i32> {
let mut state = self.state.lock().ok()?;
let code = state.code?;
if state.emitted {
return None;
}
state.emitted = true;
Some(code)
}
fn claim_when_known(&self, timeout: Duration) -> Option<i32> {
self.claim_when(timeout, |state| state.code.is_none())
}
fn claim_after_drain(&self, timeout: Duration) -> Option<i32> {
self.claim_when(timeout, |state| !state.emitted)
}
fn claim_when(
&self,
timeout: Duration,
mut keep_waiting: impl FnMut(&mut ExitState) -> bool,
) -> Option<i32> {
let state = self.state.lock().ok()?;
let (state, _) = self
.signal
.wait_timeout_while(state, timeout, |state| {
!state.stopped && keep_waiting(state)
})
.ok()?;
drop(state);
self.claim()
}
}
struct TerminalPtyInner {
backend: Mutex<TerminalPtyBackend>,
writer: Mutex<Option<Box<dyn Write + Send>>>,
killer: Mutex<Option<Box<dyn portable_pty::ChildKiller + Send + Sync>>>,
reader_thread: Mutex<Option<JoinHandle<()>>>,
exit: ExitSync,
active: AtomicBool,
kill_on_drop: AtomicBool,
handle_count: AtomicUsize,
pid: Option<u32>,
cell: AtomicCellSize,
}
struct AtomicCellSize(AtomicU32);
impl AtomicCellSize {
fn new(cell: TerminalCellSize) -> Self {
let slot = Self(AtomicU32::new(0));
slot.store(cell);
slot
}
fn load(&self) -> TerminalCellSize {
let packed = self.0.load(Ordering::Acquire);
TerminalCellSize::new((packed >> 16) as u16, packed as u16)
}
fn store(&self, cell: TerminalCellSize) {
let packed = (u32::from(cell.width) << 16) | u32::from(cell.height);
self.0.store(packed, Ordering::Release);
}
}
fn pty_size(cols: u16, rows: u16, cell: TerminalCellSize) -> PtySize {
let cols = cols.max(1);
let rows = rows.max(1);
PtySize {
rows,
cols,
pixel_width: cols.saturating_mul(cell.width),
pixel_height: rows.saturating_mul(cell.height),
}
}
enum TerminalPtyBackend {
Portable(Box<dyn portable_pty::MasterPty + Send>),
}
#[cfg(unix)]
pub struct TerminalPtyHandoff {
pub master_fd: RawFd,
pub pid: Option<u32>,
_keepalive: TerminalPty,
}
impl TerminalPty {
pub fn spawn(
config: TerminalPtyConfig,
on_event: impl Fn(TerminalPtyEvent) + Send + Sync + 'static,
) -> Result<Self, TerminalPtyError> {
let pty_system = native_pty_system();
let pair = pty_system
.openpty(pty_size(config.cols, config.rows, config.cell))
.map_err(|err| TerminalPtyError::Setup(err.to_string()))?;
#[cfg(unix)]
let reader = File::from(
unix_dup_master_fd(&*pair.master)
.map_err(|err| TerminalPtyError::Reader(err.to_string()))?,
);
#[cfg(not(unix))]
let reader = pair
.master
.try_clone_reader()
.map_err(|err| TerminalPtyError::Reader(err.to_string()))?;
#[cfg(unix)]
let mut writer = Box::new(File::from(
unix_dup_master_fd(&*pair.master)
.map_err(|err| TerminalPtyError::Writer(err.to_string()))?,
)) as Box<dyn Write + Send>;
#[cfg(not(unix))]
let mut writer = pair
.master
.take_writer()
.map_err(|err| TerminalPtyError::Writer(err.to_string()))?;
prime_conpty_cursor(&mut *writer)
.map_err(|err| TerminalPtyError::Writer(err.to_string()))?;
let mut builder = CommandBuilder::new(config.command.as_ref());
for arg in config.args {
builder.arg(arg.as_ref());
}
builder.env("TERM", config.term.as_ref());
if let Some(cwd) = config.cwd {
builder.cwd(cwd.as_ref());
}
for (key, value) in config.env {
builder.env(key.as_ref(), value.as_ref());
}
let mut child = pair
.slave
.spawn_command(builder)
.map_err(|err| TerminalPtyError::Spawn(err.to_string()))?;
let pid = child.process_id();
let inner = Arc::new(TerminalPtyInner {
backend: Mutex::new(TerminalPtyBackend::Portable(pair.master)),
writer: Mutex::new(Some(writer)),
killer: Mutex::new(Some(child.clone_killer())),
reader_thread: Mutex::new(None),
exit: ExitSync::default(),
active: AtomicBool::new(true),
kill_on_drop: AtomicBool::new(true),
handle_count: AtomicUsize::new(1),
pid,
cell: AtomicCellSize::new(config.cell),
});
let on_event = Arc::new(on_event);
{
let on_event = on_event.clone();
let inner = inner.clone();
let thread_inner = inner.clone();
let reader_thread = std::thread::spawn(move || {
let mut reader = reader;
let mut buffer = [0u8; 8192];
let mut stream_ended = false;
loop {
if !thread_inner.active.load(Ordering::Acquire) {
break;
}
#[cfg(unix)]
match unix_wait_readable(reader.as_raw_fd(), &thread_inner.active) {
PtyReadiness::Readable => {}
PtyReadiness::Idle => {
if let Some(code) = thread_inner.exit.claim() {
on_event(TerminalPtyEvent::Exited(code));
}
continue;
}
PtyReadiness::Stop => break,
}
match reader.read(&mut buffer) {
Ok(0) => {
stream_ended = true;
break;
}
Ok(read) => {
if !thread_inner.active.load(Ordering::Acquire) {
break;
}
on_event(TerminalPtyEvent::Output(Arc::<[u8]>::from(
buffer[..read].to_vec(),
)));
}
Err(err) if err.kind() == std::io::ErrorKind::Interrupted => continue,
Err(err) => {
stream_ended = true;
#[cfg(unix)]
if err.raw_os_error() == Some(libc::EIO) {
break;
}
on_event(TerminalPtyEvent::Error(err.to_string().into()));
break;
}
}
}
if stream_ended
&& let Some(code) = thread_inner.exit.claim_when_known(EXIT_DRAIN_GRACE)
{
on_event(TerminalPtyEvent::Exited(code));
}
});
if let Ok(mut slot) = inner.reader_thread.lock() {
*slot = Some(reader_thread);
}
}
{
let on_event = on_event.clone();
let thread_inner = inner.clone();
std::thread::spawn(move || {
let exit_code = child
.wait()
.ok()
.map(|status| status.exit_code() as i32)
.unwrap_or(-1);
thread_inner.exit.publish(exit_code);
if let Some(code) = thread_inner.exit.claim_after_drain(EXIT_DRAIN_GRACE) {
on_event(TerminalPtyEvent::Exited(code));
}
});
}
Ok(Self { inner })
}
#[cfg(unix)]
pub fn handoff(&self) -> std::io::Result<TerminalPtyHandoff> {
self.inner.active.store(false, Ordering::Release);
self.inner.exit.stop();
self.inner.kill_on_drop.store(false, Ordering::Release);
if let Some(handle) = self
.inner
.reader_thread
.lock()
.map_err(|_| std::io::Error::other("pty reader thread lock poisoned"))?
.take()
{
let _ = handle.join();
}
let mut writer = self
.inner
.writer
.lock()
.map_err(|_| std::io::Error::other("pty writer lock poisoned"))?;
writer.take();
drop(writer);
let backend = self
.inner
.backend
.lock()
.map_err(|_| std::io::Error::other("pty master lock poisoned"))?;
let fd = match &*backend {
TerminalPtyBackend::Portable(master) => master
.as_raw_fd()
.ok_or_else(|| std::io::Error::other("pty master fd unavailable"))?,
};
Ok(TerminalPtyHandoff {
master_fd: fd,
pid: self.inner.pid,
_keepalive: self.clone(),
})
}
pub fn pid(&self) -> Option<u32> {
self.inner.pid
}
#[cfg(unix)]
pub fn foreground_process_group_id(&self) -> Option<i32> {
if !self.inner.active.load(Ordering::Acquire) {
return None;
}
let backend = self.inner.backend.lock().ok()?;
let fd = match &*backend {
TerminalPtyBackend::Portable(master) => master.as_raw_fd()?,
};
let pgid = unsafe { libc::tcgetpgrp(fd) };
(pgid >= 0).then_some(pgid)
}
pub fn write(&self, bytes: &[u8]) -> std::io::Result<()> {
if !self.inner.active.load(Ordering::Acquire) {
return Err(std::io::Error::other("pty has been handed off"));
}
let mut writer = self
.inner
.writer
.lock()
.map_err(|_| std::io::Error::other("pty writer lock poisoned"))?;
let writer = writer
.as_mut()
.ok_or_else(|| std::io::Error::other("pty writer unavailable"))?;
writer.write_all(bytes)?;
writer.flush()
}
pub fn send_key(&self, key: KeyEvent, modes: TerminalKeyModes) -> std::io::Result<bool> {
let Some(bytes) = key_event_to_bytes(key, modes) else {
return Ok(false);
};
self.write(&bytes)?;
Ok(true)
}
pub fn resize(&self, cols: u16, rows: u16) -> std::io::Result<()> {
self.resize_with_cell_size(cols, rows, self.inner.cell.load())
}
pub fn resize_with_cell_size(
&self,
cols: u16,
rows: u16,
cell: TerminalCellSize,
) -> std::io::Result<()> {
if !self.inner.active.load(Ordering::Acquire) {
return Err(std::io::Error::other("pty has been handed off"));
}
let backend = self
.inner
.backend
.lock()
.map_err(|_| std::io::Error::other("pty master lock poisoned"))?;
match &*backend {
TerminalPtyBackend::Portable(master) => {
self.inner.cell.store(cell);
master
.resize(pty_size(cols, rows, cell))
.map_err(|err| std::io::Error::other(err.to_string()))
}
}
}
pub fn kill(&self) -> std::io::Result<()> {
if !self.inner.kill_on_drop.load(Ordering::Acquire) {
return Ok(());
}
self.inner.active.store(false, Ordering::Release);
self.inner.exit.stop();
let mut killer = self
.inner
.killer
.lock()
.map_err(|_| std::io::Error::other("pty killer lock poisoned"))?;
if let Some(killer) = killer.as_mut() {
return killer
.kill()
.map_err(|err| std::io::Error::other(err.to_string()));
}
Ok(())
}
}
#[cfg(unix)]
fn unix_dup_master_fd(master: &dyn portable_pty::MasterPty) -> std::io::Result<OwnedFd> {
let fd = master
.as_raw_fd()
.ok_or_else(|| std::io::Error::other("pty master fd unavailable"))?;
unix_dup_raw_fd(fd)
}
#[cfg(unix)]
fn unix_dup_raw_fd(fd: RawFd) -> std::io::Result<OwnedFd> {
let dup = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) };
if dup < 0 {
return Err(std::io::Error::last_os_error());
}
Ok(unsafe { OwnedFd::from_raw_fd(dup) })
}
#[cfg(unix)]
enum PtyReadiness {
Readable,
Idle,
Stop,
}
#[cfg(unix)]
fn unix_wait_readable(fd: RawFd, active: &AtomicBool) -> PtyReadiness {
if !active.load(Ordering::Acquire) {
return PtyReadiness::Stop;
}
let mut pollfd = libc::pollfd {
fd,
events: libc::POLLIN,
revents: 0,
};
let rc = unsafe { libc::poll(&mut pollfd, 1, 100) };
if rc > 0 {
if pollfd.revents & (libc::POLLIN | libc::POLLHUP | libc::POLLERR) != 0 {
return PtyReadiness::Readable;
}
if pollfd.revents & libc::POLLNVAL != 0 {
return PtyReadiness::Stop;
}
return PtyReadiness::Idle;
}
if rc == 0 {
return PtyReadiness::Idle;
}
if std::io::Error::last_os_error().kind() == std::io::ErrorKind::Interrupted {
return PtyReadiness::Idle;
}
PtyReadiness::Stop
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TerminalPtyEvent {
Output(Arc<[u8]>),
Exited(i32),
Error(Arc<str>),
}
impl Drop for TerminalPty {
fn drop(&mut self) {
if self.inner.handle_count.fetch_sub(1, Ordering::AcqRel) == 1 {
let _ = self.kill();
}
}
}
#[cfg(all(test, unix))]
mod tests {
use super::*;
#[test]
fn dropping_a_clone_does_not_kill_the_shared_pty() {
let pty = TerminalPty::spawn(
TerminalPtyConfig::new("/bin/sh").arg("-c").arg("sleep 5"),
|_event| {},
)
.expect("spawn");
let clone = pty.clone();
assert_eq!(pty.inner.handle_count.load(Ordering::Acquire), 2);
drop(clone);
assert_eq!(pty.inner.handle_count.load(Ordering::Acquire), 1);
assert!(
pty.write(b"").is_ok(),
"pty should still be alive after dropping a clone"
);
drop(pty);
}
#[test]
fn a_fast_command_s_output_arrives_before_its_exit() {
for attempt in 0..40 {
let events = Arc::new(Mutex::new(Vec::new()));
let sink = events.clone();
let mut pty = Some(
TerminalPty::spawn(
TerminalPtyConfig::new("/bin/sh")
.arg("-c")
.arg("printf 'fast output\\n'; exit 3"),
move |event| sink.lock().expect("events").push(event),
)
.expect("spawn"),
);
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while std::time::Instant::now() < deadline {
let exited = events
.lock()
.expect("events")
.iter()
.any(|event| matches!(event, TerminalPtyEvent::Exited(_)));
if exited {
break;
}
std::thread::sleep(Duration::from_millis(5));
}
drop(pty.take());
let events = events.lock().expect("events");
let text: String = events
.iter()
.filter_map(|event| match event {
TerminalPtyEvent::Output(bytes) => Some(String::from_utf8_lossy(bytes)),
_ => None,
})
.collect();
assert!(
text.contains("fast output"),
"attempt {attempt}: the command's output was lost; events: {events:?}"
);
assert!(
matches!(events.last(), Some(TerminalPtyEvent::Exited(3))),
"attempt {attempt}: the exit must come last and carry the real status; \
events: {events:?}"
);
}
}
#[test]
fn killing_a_pty_reports_the_exit_without_waiting_for_a_drain() {
let events = Arc::new(Mutex::new(Vec::new()));
let sink = events.clone();
let pty = TerminalPty::spawn(
TerminalPtyConfig::new("/bin/sh").arg("-c").arg("sleep 30"),
move |event| sink.lock().expect("events").push(event),
)
.expect("spawn");
let started = std::time::Instant::now();
pty.kill().expect("kill");
let deadline = started + EXIT_DRAIN_GRACE;
while std::time::Instant::now() < deadline {
if events
.lock()
.expect("events")
.iter()
.any(|event| matches!(event, TerminalPtyEvent::Exited(_)))
{
return;
}
std::thread::sleep(Duration::from_millis(5));
}
panic!("a killed pty must report its exit without waiting out the drain grace");
}
#[test]
fn foreground_process_group_id_reports_a_value_while_alive() {
let pty = TerminalPty::spawn(
TerminalPtyConfig::new("/bin/sh").arg("-c").arg("sleep 5"),
|_event| {},
)
.expect("spawn");
assert!(pty.foreground_process_group_id().is_some());
drop(pty);
}
}