twine_core/terminal/
pty.rs

1use std::io::{Read, Write};
2use std::path::PathBuf;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::{Arc, Mutex};
5use std::thread::JoinHandle;
6
7use portable_pty::{CommandBuilder, MasterPty, PtySize, native_pty_system};
8use tracing::debug;
9
10use super::launcher::verify_working_directory;
11use super::process::{SharedChild, shared_child, terminate_unobserved_child};
12use super::{TerminalError, TerminalId, TerminalSize, TerminalStream};
13
14pub(super) const READ_CHUNK_BYTES: usize = 16 * 1024;
15const READER_POLL_MILLISECONDS: libc::c_int = 50;
16
17#[cfg(unix)]
18pub(super) type TerminalReaderDescriptor = libc::c_int;
19
20#[cfg(not(unix))]
21pub(super) type TerminalReaderDescriptor = ();
22
23impl From<TerminalSize> for PtySize {
24    fn from(size: TerminalSize) -> Self {
25        Self {
26            rows: size.rows,
27            cols: size.columns,
28            pixel_width: size.pixel_width,
29            pixel_height: size.pixel_height,
30        }
31    }
32}
33
34pub(super) struct StartingTerminal {
35    pub(super) master: Box<dyn MasterPty + Send>,
36    pub(super) reader: Box<dyn Read + Send>,
37    pub(super) writer: Arc<Mutex<Box<dyn Write + Send>>>,
38    pub(super) child: SharedChild,
39    pub(super) reader_descriptor: TerminalReaderDescriptor,
40}
41
42pub(super) fn spawn_terminal_process(
43    command: CommandBuilder,
44    size: TerminalSize,
45    working_directory_handshake: Option<PathBuf>,
46) -> Result<StartingTerminal, TerminalError> {
47    let pair = native_pty_system()
48        .openpty(size.validate()?.into())
49        .map_err(|error| TerminalError::Pty {
50            operation: "open PTY",
51            message: error.to_string(),
52        })?;
53    let reader_descriptor = terminal_reader_descriptor(pair.master.as_ref())?;
54    let child = pair
55        .slave
56        .spawn_command(command)
57        .map_err(|error| TerminalError::Pty {
58            operation: "spawn shell",
59            message: error.to_string(),
60        })?;
61    drop(pair.slave);
62    let child = shared_child(child);
63    let mut reader = pair.master.try_clone_reader().map_err(|error| {
64        terminate_unobserved_child(&child);
65        TerminalError::Pty {
66            operation: "clone PTY reader",
67            message: error.to_string(),
68        }
69    })?;
70    if let Some(working_directory) = working_directory_handshake
71        && let Err(error) = verify_working_directory(&mut reader, &working_directory)
72    {
73        terminate_unobserved_child(&child);
74        return Err(error);
75    }
76    let writer = pair.master.take_writer().map_err(|error| {
77        terminate_unobserved_child(&child);
78        TerminalError::Pty {
79            operation: "take PTY writer",
80            message: error.to_string(),
81        }
82    })?;
83    Ok(StartingTerminal {
84        master: pair.master,
85        reader,
86        writer: Arc::new(Mutex::new(writer)),
87        child,
88        reader_descriptor,
89    })
90}
91
92#[cfg(unix)]
93fn terminal_reader_descriptor(
94    master: &(dyn MasterPty + Send),
95) -> Result<TerminalReaderDescriptor, TerminalError> {
96    master.as_raw_fd().ok_or_else(|| TerminalError::Pty {
97        operation: "inspect PTY reader",
98        message: "native PTY did not expose a file descriptor".to_owned(),
99    })
100}
101
102#[cfg(not(unix))]
103fn terminal_reader_descriptor(
104    _master: &(dyn MasterPty + Send),
105) -> Result<TerminalReaderDescriptor, TerminalError> {
106    Ok(())
107}
108
109pub(super) fn spawn_terminal_reader(
110    output: Arc<TerminalStream>,
111    terminal_id: TerminalId,
112    mut reader: Box<dyn Read + Send>,
113    reader_descriptor: TerminalReaderDescriptor,
114    cancelled: Arc<AtomicBool>,
115) -> Result<JoinHandle<()>, TerminalError> {
116    crate::blocking_worker::spawn(
117        format!("terminal-{}-reader", terminal_id.value()),
118        move || {
119            let mut buffer = vec![0; READ_CHUNK_BYTES];
120            loop {
121                match wait_for_terminal_output(reader_descriptor, &cancelled) {
122                    Ok(true) => {}
123                    Ok(false) => break,
124                    Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
125                    Err(error) => {
126                        debug!(terminal_id = terminal_id.value(), %error, "terminal reader poll stopped");
127                        break;
128                    }
129                }
130                match reader.read(&mut buffer) {
131                    Ok(0) => break,
132                    Ok(count) => {
133                        if output
134                            .publish_blocking(terminal_id, buffer[..count].to_vec())
135                            .is_err()
136                        {
137                            break;
138                        }
139                    }
140                    Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
141                    Err(error) => {
142                        debug!(terminal_id = terminal_id.value(), %error, "terminal reader stopped");
143                        break;
144                    }
145                }
146            }
147            let _ = output.finish(terminal_id);
148        },
149    )
150        .map_err(|error| TerminalError::Thread {
151            operation: "start PTY reader",
152            message: error.to_string(),
153        })
154}
155
156#[cfg(unix)]
157fn wait_for_terminal_output(
158    descriptor: TerminalReaderDescriptor,
159    cancelled: &AtomicBool,
160) -> std::io::Result<bool> {
161    loop {
162        if cancelled.load(Ordering::Acquire) {
163            return Ok(false);
164        }
165        let mut poll_descriptor = libc::pollfd {
166            fd: descriptor,
167            events: libc::POLLIN,
168            revents: 0,
169        };
170        // SAFETY: `poll_descriptor` is valid for one entry for the duration of the call. The PTY
171        // master stays owned by the session until this reader thread has joined.
172        let result = unsafe { libc::poll(&raw mut poll_descriptor, 1, READER_POLL_MILLISECONDS) };
173        if result > 0 {
174            return Ok(!cancelled.load(Ordering::Acquire));
175        }
176        if result < 0 {
177            return Err(std::io::Error::last_os_error());
178        }
179    }
180}
181
182#[cfg(not(unix))]
183fn wait_for_terminal_output(
184    _descriptor: TerminalReaderDescriptor,
185    cancelled: &AtomicBool,
186) -> std::io::Result<bool> {
187    Ok(!cancelled.load(Ordering::Acquire))
188}
twine_core/terminal/pty.rs

twine_core/terminal/
pty.rs
