ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Apply terminal backpressure without dropping output

Browse source
This commit is contained in:
ms2k 2026-08-21 02:35:08 +09:00
commit 968ff09eae
1 changed files with 89 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

143
crates/lazaret-client/src/terminal.rs
View file

@ -114,7 +114,8 @@ pub struct PromptOutcome {
const PROMPT_ESCAPE_INTERVAL: Duration = Duration::from_millis(40); const PROMPT_ESCAPE_INTERVAL: Duration = Duration::from_millis(40);
pub struct SystemTerminal { pub struct SystemTerminal {
terminal: File, input: File,
output: File,
original: Termios, original: Termios,
original_flags: OFlags, original_flags: OFlags,
raw: bool, raw: bool,
@ -245,12 +246,14 @@ impl SystemTerminal {
/// # Errors /// # Errors
/// Returns an error when the process has no usable controlling terminal. /// Returns an error when the process has no usable controlling terminal.
pub fn open() -> io::Result<Self> { pub fn open() -> io::Result<Self> {
let terminal = OpenOptions::new().read(true).write(true).open("/dev/tty")?; let input = OpenOptions::new().read(true).write(true).open("/dev/tty")?;
let size = tcgetwinsize(&terminal)?; let output = OpenOptions::new().read(true).write(true).open("/dev/tty")?;
let size = tcgetwinsize(&input)?;
Ok(Self { Ok(Self {
original: tcgetattr(&terminal)?, original: tcgetattr(&input)?,
original_flags: fcntl_getfl(&terminal)?, original_flags: fcntl_getfl(&input)?,
terminal, input,
output,
raw: false, raw: false,
signals: SignalQueue::register()?, signals: SignalQueue::register()?,
events: VecDeque::new(), events: VecDeque::new(),
@ -323,9 +326,9 @@ impl HostTerminal for SystemTerminal {
} }
let mut raw = self.original.clone(); let mut raw = self.original.clone();
raw.make_raw(); raw.make_raw();
tcsetattr(&self.terminal, OptionalActions::Flush, &raw)?; tcsetattr(&self.input, OptionalActions::Flush, &raw)?;
if let Err(error) = fcntl_setfl(&self.terminal, self.original_flags | OFlags::NONBLOCK) { if let Err(error) = fcntl_setfl(&self.input, self.original_flags | OFlags::NONBLOCK) {
let _ = tcsetattr(&self.terminal, OptionalActions::Flush, &self.original); let _ = tcsetattr(&self.input, OptionalActions::Flush, &self.original);
return Err(error.into()); return Err(error.into());
} }
self.raw = true; self.raw = true;
@ -339,14 +342,14 @@ impl HostTerminal for SystemTerminal {
self.leave_for_host()?; self.leave_for_host()?;
self.trusted = None; self.trusted = None;
self.deferred_resize = None; self.deferred_resize = None;
fcntl_setfl(&self.terminal, self.original_flags)?; fcntl_setfl(&self.input, self.original_flags)?;
tcsetattr(&self.terminal, OptionalActions::Flush, &self.original)?; tcsetattr(&self.input, OptionalActions::Flush, &self.original)?;
self.raw = false; self.raw = false;
Ok(()) Ok(())
} }
fn size(&self, session_id: lazaret_core::identity::SessionId) -> io::Result<TerminalResize> { fn size(&self, session_id: lazaret_core::identity::SessionId) -> io::Result<TerminalResize> {
let size = tcgetwinsize(&self.terminal)?; let size = tcgetwinsize(&self.input)?;
Ok(TerminalResize { Ok(TerminalResize {
session_id, session_id,
rows: NonZeroU16::new(size.ws_row) rows: NonZeroU16::new(size.ws_row)
@ -373,7 +376,7 @@ impl HostTerminal for SystemTerminal {
return Ok(event); return Ok(event);
} }
let mut bytes = vec![0_u8; 16 * 1024]; let mut bytes = vec![0_u8; 16 * 1024];
match self.terminal.read(&mut bytes) { match self.input.read(&mut bytes) {
Ok(0) => Ok(HostTerminalEvent::Terminate), Ok(0) => Ok(HostTerminalEvent::Terminate),
Ok(length) => { Ok(length) => {
bytes.truncate(length); bytes.truncate(length);
@ -393,8 +396,8 @@ impl HostTerminal for SystemTerminal {
self.restore_guest_presentation()?; self.restore_guest_presentation()?;
} }
self.presentation.process(bytes); self.presentation.process(bytes);
self.terminal.write_all(bytes)?; self.output.write_all(bytes)?;
self.terminal.flush()?; self.output.flush()?;
self.physical_screen = if self.presentation.alternate_screen() { self.physical_screen = if self.presentation.alternate_screen() {
PhysicalScreen::Alternate PhysicalScreen::Alternate
} else { } else {
@ -434,7 +437,7 @@ impl HostTerminal for SystemTerminal {
self.events.clear(); self.events.clear();
self.escape_pending = false; self.escape_pending = false;
self.key_translator.clear(); self.key_translator.clear();
tcflush(&self.terminal, QueueSelector::IFlush)?; tcflush(&self.input, QueueSelector::IFlush)?;
let mut decoder = PromptInputDecoder::new(); let mut decoder = PromptInputDecoder::new();
self.trusted = Some(TrustedPresentation { self.trusted = Some(TrustedPresentation {
prompt: prompt.clone(), prompt: prompt.clone(),
@ -496,8 +499,8 @@ impl HostTerminal for SystemTerminal {
fn suspend_and_resume(&mut self) -> io::Result<()> { fn suspend_and_resume(&mut self) -> io::Result<()> {
self.leave_for_host()?; self.leave_for_host()?;
fcntl_setfl(&self.terminal, self.original_flags)?; fcntl_setfl(&self.input, self.original_flags)?;
tcsetattr(&self.terminal, OptionalActions::Flush, &self.original)?; tcsetattr(&self.input, OptionalActions::Flush, &self.original)?;
self.raw = false; self.raw = false;
lazaret_linux::process::suspend_current_process()?; lazaret_linux::process::suspend_current_process()?;
self.enter_raw()?; self.enter_raw()?;
@ -519,29 +522,29 @@ impl SystemTerminal {
} }
fn leave_for_host(&mut self) -> io::Result<()> { fn leave_for_host(&mut self) -> io::Result<()> {
self.terminal.write_all(HOST_TERMINAL_MODES)?; self.output.write_all(HOST_TERMINAL_MODES)?;
if self.physical_screen == PhysicalScreen::Alternate { if self.physical_screen == PhysicalScreen::Alternate {
self.terminal.write_all(b"\x1b[?1049l")?; self.output.write_all(b"\x1b[?1049l")?;
self.physical_screen = PhysicalScreen::Primary; self.physical_screen = PhysicalScreen::Primary;
} }
self.terminal.flush() self.output.flush()
} }
fn reconstruct_after_resume(&mut self) -> io::Result<()> { fn reconstruct_after_resume(&mut self) -> io::Result<()> {
if self.trusted.is_some() { if self.trusted.is_some() {
return self.render_trusted_prompt(true); return self.render_trusted_prompt(true);
} }
self.terminal.write_all(b"\x1b[?2026l\x1b[?2026h")?; self.output.write_all(b"\x1b[?2026l\x1b[?2026h")?;
if self.presentation.alternate_screen() { if self.presentation.alternate_screen() {
self.terminal.write_all(b"\x1b[?1049h")?; self.output.write_all(b"\x1b[?1049h")?;
self.physical_screen = PhysicalScreen::Alternate; self.physical_screen = PhysicalScreen::Alternate;
} }
self.terminal self.output
.write_all(&self.presentation.restoration_snapshot())?; .write_all(&self.presentation.restoration_snapshot())?;
self.terminal.write_all(b"\x1b[?2026l")?; self.output.write_all(b"\x1b[?2026l")?;
self.terminal self.output
.write_all(&self.presentation.synchronized_output_mode())?; .write_all(&self.presentation.synchronized_output_mode())?;
self.terminal.flush() self.output.flush()
} }
fn render_trusted_prompt(&mut self, reconstruct: bool) -> io::Result<()> { fn render_trusted_prompt(&mut self, reconstruct: bool) -> io::Result<()> {
@ -552,26 +555,21 @@ impl SystemTerminal {
let result = (|| { let result = (|| {
let size = self.size(trusted.prompt.session_id)?; let size = self.size(trusted.prompt.session_id)?;
let layout = PromptLayout::new(&trusted.prompt, size.columns.get(), size.rows.get()); let layout = PromptLayout::new(&trusted.prompt, size.columns.get(), size.rows.get());
self.terminal.write_all(b"\x1b[?2026l\x1b[?2026h")?; self.output.write_all(b"\x1b[?2026l\x1b[?2026h")?;
self.terminal.write_all(PROMPT_INPUT_MODES_OFF)?; self.output.write_all(PROMPT_INPUT_MODES_OFF)?;
self.terminal.write_all(b"\x1b[?25l\x1b[?7l")?; self.output.write_all(b"\x1b[?25l\x1b[?7l")?;
let entered_alternate = self.physical_screen == PhysicalScreen::Primary; let entered_alternate = self.physical_screen == PhysicalScreen::Primary;
if entered_alternate { if entered_alternate {
self.terminal.write_all(b"\x1b[?1049h")?; self.output.write_all(b"\x1b[?1049h")?;
self.physical_screen = PhysicalScreen::Alternate; self.physical_screen = PhysicalScreen::Alternate;
} }
if reconstruct || entered_alternate { if reconstruct || entered_alternate {
self.terminal self.output
.write_all(&self.presentation.context_snapshot())?; .write_all(&self.presentation.context_snapshot())?;
} }
render_prompt_card( render_prompt_card(&mut self.output, &layout, trusted.choice, &mut trusted.view)?;
&mut self.terminal, self.output.write_all(b"\x1b[?2026l")?;
&layout, self.output.flush()
trusted.choice,
&mut trusted.view,
)?;
self.terminal.write_all(b"\x1b[?2026l")?;
self.terminal.flush()
})(); })();
self.trusted = Some(trusted); self.trusted = Some(trusted);
result result
@ -582,24 +580,24 @@ impl SystemTerminal {
.trusted .trusted
.take() .take()
.expect("guest restoration requires an active trusted presentation"); .expect("guest restoration requires an active trusted presentation");
self.terminal.write_all(b"\x1b[?2026l\x1b[?2026h")?; self.output.write_all(b"\x1b[?2026l\x1b[?2026h")?;
if trusted.guest_alternate { if trusted.guest_alternate {
if self.physical_screen == PhysicalScreen::Primary { if self.physical_screen == PhysicalScreen::Primary {
self.terminal.write_all(b"\x1b[?1049h")?; self.output.write_all(b"\x1b[?1049h")?;
self.physical_screen = PhysicalScreen::Alternate; self.physical_screen = PhysicalScreen::Alternate;
} }
self.terminal self.output
.write_all(&self.presentation.restoration_snapshot())?; .write_all(&self.presentation.restoration_snapshot())?;
} else if self.physical_screen == PhysicalScreen::Alternate { } else if self.physical_screen == PhysicalScreen::Alternate {
self.terminal.write_all(b"\x1b[?1049l")?; self.output.write_all(b"\x1b[?1049l")?;
self.physical_screen = PhysicalScreen::Primary; self.physical_screen = PhysicalScreen::Primary;
} }
self.terminal self.output
.write_all(&self.presentation.restoration_modes())?; .write_all(&self.presentation.restoration_modes())?;
self.terminal.write_all(b"\x1b[?2026l")?; self.output.write_all(b"\x1b[?2026l")?;
self.terminal self.output
.write_all(&self.presentation.synchronized_output_mode())?; .write_all(&self.presentation.synchronized_output_mode())?;
self.terminal.flush() self.output.flush()
} }
} }
@ -1098,17 +1096,24 @@ mod tests {
openpt(OpenptFlags::RDWR | OpenptFlags::NOCTTY | OpenptFlags::CLOEXEC).unwrap(); openpt(OpenptFlags::RDWR | OpenptFlags::NOCTTY | OpenptFlags::CLOEXEC).unwrap();
grantpt(&master).unwrap(); grantpt(&master).unwrap();
unlockpt(&master).unwrap(); unlockpt(&master).unwrap();
let slave = File::from( let input = File::from(
ioctl_tiocgptpeer( ioctl_tiocgptpeer(
&master, &master,
OpenptFlags::RDWR | OpenptFlags::NOCTTY | OpenptFlags::CLOEXEC, OpenptFlags::RDWR | OpenptFlags::NOCTTY | OpenptFlags::CLOEXEC,
) )
.unwrap(), .unwrap(),
); );
let original = tcgetattr(&slave).unwrap(); let output = File::from(
let original_flags = fcntl_getfl(&slave).unwrap(); ioctl_tiocgptpeer(
&master,
OpenptFlags::RDWR | OpenptFlags::NOCTTY | OpenptFlags::CLOEXEC,
)
.unwrap(),
);
let original = tcgetattr(&input).unwrap();
let original_flags = fcntl_getfl(&input).unwrap();
tcsetwinsize( tcsetwinsize(
&slave, &input,
Winsize { Winsize {
ws_row: 40, ws_row: 40,
ws_col: 140, ws_col: 140,
@ -1119,7 +1124,8 @@ mod tests {
.unwrap(); .unwrap();
( (
SystemTerminal { SystemTerminal {
terminal: slave, input,
output,
original, original,
original_flags, original_flags,
raw: false, raw: false,
@ -1177,7 +1183,7 @@ mod tests {
let (mut terminal, mut master) = terminal(); let (mut terminal, mut master) = terminal();
terminal.enter_raw().unwrap(); terminal.enter_raw().unwrap();
assert!( assert!(
!tcgetattr(&terminal.terminal) !tcgetattr(&terminal.input)
.unwrap() .unwrap()
.local_modes .local_modes
.contains(LocalModes::ISIG) .contains(LocalModes::ISIG)
@ -1189,6 +1195,35 @@ mod tests {
); );
} }
#[test]
fn terminal_output_waits_for_a_slow_reader() {
let (mut terminal, mut master) = terminal();
terminal.enter_raw().unwrap();
let output = vec![b'x'; 256 * 1024];
let expected = output.clone();
let reader = thread::spawn(move || {
thread::sleep(Duration::from_millis(100));
let mut received = Vec::new();
let mut buffer = [0_u8; 16 * 1024];
while received.len() < expected.len() {
match master.read(&mut buffer) {
Ok(0) => break,
Ok(length) => received.extend_from_slice(&buffer[..length]),
Err(error) if error.raw_os_error() == Some(nix::libc::EIO) => break,
Err(error) => panic!("slow terminal reader failed: {error}"),
}
}
received
});
let result = terminal.write_output(&output);
drop(terminal);
let received = reader.join().unwrap();
assert!(result.is_ok(), "{result:?}");
assert!(received.starts_with(&output));
}
#[test] #[test]
fn ssh_style_escapes_survive_read_boundaries_and_preserve_literals() { fn ssh_style_escapes_survive_read_boundaries_and_preserve_literals() {
let (mut terminal, mut master) = terminal(); let (mut terminal, mut master) = terminal();