diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 6c68a66..2fc4cb3 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -6560,6 +6560,7 @@ async fn run_terminal( let mut winch = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::window_change()).ok(); let mut frame_buffer = FrameBuffer::default(); + let mut accepted_output_seq = cred.last_rendered_seq; // Resolve the prediction display policy (off / experimental / always). An // env var wins for ad-hoc tuning; otherwise the client config's // `predict_mode` provides the persistent default. Predictions only run in a @@ -6568,6 +6569,7 @@ async fn run_terminal( predict && cred.mode != "view-only" && !forward_only, predict_mode, ); + let mut frame_renderer = TerminalFrameRenderer::new()?; // Non-destructive disconnect status line. Off in forward-only mode (no TTY to // draw on) and respecting the client config (default on, env override). let mut disconnect_status = DisconnectStatus::new(resolve_disconnect_status() && !forward_only); @@ -6673,10 +6675,28 @@ async fn run_terminal( let mut recv_buf = vec![0u8; 65535]; let mut detach_requested = false; loop { + accepted_output_seq = accepted_output_seq.max(cred.last_rendered_seq); + predictor.set_output_backpressured(frame_renderer.has_pending()); if detach_requested { break; } tokio::select! { + rendered = frame_renderer.recv(), if frame_renderer.has_pending() => { + let frame = rendered?; + predictor.set_output_backpressured(frame_renderer.has_pending()); + cred.last_rendered_seq = cred.last_rendered_seq.max(frame.output_seq); + note_snapshot_rendered( + &frame, + &mut disconnect_status, + &mut status_restore_pending, + ); + last_terminal_frame_at = Instant::now(); + wake_repaint_retry_until = None; + send_ack(&socket, addr, &cred, &mut send_seq).await?; + if frame.closed { + return Ok(()); + } + } stdin_msg = stdin_rx.recv() => { match stdin_msg { Some(mut bytes) => { @@ -6700,7 +6720,8 @@ async fn run_terminal( forward_only, input_status_tick_gap, last_packet_at.elapsed(), - ) { + ) && !frame_renderer.has_pending() + { let reconnect_started_at = Instant::now(); if let Some(deadline) = wake_repaint_retry_deadline(reconnect_started_at, input_status_tick_gap) @@ -6764,6 +6785,7 @@ async fn run_terminal( let saw_focus_in = input_contains_focus_in(&bytes); if !forward_only && !refreshed_before_input + && !frame_renderer.has_pending() && saw_focus_in && last_focus_repaint_at.elapsed() >= FOCUS_REPAINT_COOLDOWN { @@ -6997,7 +7019,8 @@ async fn run_terminal( &mut pending_user_input_bytes, bytes, )?; - if let Some(frame) = reconnect( + if !frame_renderer.has_pending() + && let Some(frame) = reconnect( &socket, &mut cred, &mut send_seq, @@ -7075,7 +7098,8 @@ async fn run_terminal( maybe_send_resize(&socket, addr, &cred, &mut send_seq, &mut last_size).await?; } _ = frame_gap_tick.tick() => { - if frame_buffer.resync_due() + if !frame_renderer.has_pending() + && frame_buffer.resync_due() && let Some(frame) = reconnect( &socket, &mut cred, @@ -7138,6 +7162,9 @@ async fn run_terminal( } }); let Ok(plain) = decrypted else { + if frame_renderer.has_pending() { + continue; + } if let Some(frame) = reconnect( &socket, &mut cred, @@ -7185,19 +7212,15 @@ async fn run_terminal( continue; }; last_packet_at = Instant::now(); - let frames = frame_buffer.accept(frame, &mut cred.last_rendered_seq); + let frames = frame_buffer.accept(frame, &mut accepted_output_seq); for frame in frames { predictor.clear_pending()?; if !forward_only { - render_frame(&frame)?; - note_snapshot_rendered( - &frame, - &mut disconnect_status, - &mut status_restore_pending, - ); predictor.observe_output(&frame.bytes); last_terminal_frame_at = Instant::now(); wake_repaint_retry_until = None; + frame_renderer.enqueue(frame)?; + predictor.set_output_backpressured(true); flush_startup_input_if_ready( &socket, addr, @@ -7209,12 +7232,15 @@ async fn run_terminal( ) .await?; } - if frame.closed { + else if frame.closed { send_ack(&socket, addr, &cred, &mut send_seq).await?; return Ok(()); } } - send_ack(&socket, addr, &cred, &mut send_seq).await?; + if forward_only { + cred.last_rendered_seq = accepted_output_seq; + send_ack(&socket, addr, &cred, &mut send_seq).await?; + } } PacketKind::Pong => { if protocol::decrypt_body( @@ -7302,7 +7328,7 @@ async fn run_terminal( } PacketKind::AttachReject => { let reject: AttachReject = protocol::from_body(&packet.body)?; - if reject.reason == "unknown client" { + if reject.reason == "unknown client" && !frame_renderer.has_pending() { if let Some(frame) = reconnect( &socket, &mut cred, @@ -7918,8 +7944,9 @@ async fn run_terminal( } let mut repainted_this_tick = false; let stale = last_packet_at.elapsed(); - if status_restore_pending - || stale >= Duration::from_secs(reconnect_timeout_secs.max(1)) + if !frame_renderer.has_pending() + && (status_restore_pending + || stale >= Duration::from_secs(reconnect_timeout_secs.max(1))) { if let Some(frame) = reconnect( &socket, @@ -7977,7 +8004,9 @@ async fn run_terminal( // a latency spike (or recovery) flips speculation on/off promptly // without waiting for the next keystroke to drive `redraw`. if !forward_only { - predictor.refresh_policy()?; + if !frame_renderer.has_pending() { + predictor.refresh_policy()?; + } flush_startup_input_if_ready( &socket, addr, @@ -7989,7 +8018,7 @@ async fn run_terminal( ) .await?; let now = Instant::now(); - if !repainted_this_tick && should_repaint_idle_terminal( + if !repainted_this_tick && !frame_renderer.has_pending() && should_repaint_idle_terminal( predictor.alternate_screen, last_terminal_frame_at, last_idle_repaint_attempt_at, @@ -8047,7 +8076,7 @@ async fn run_terminal( // on how long the link has been silent (recomputed after any // reconnect attempt above may have reset `last_packet_at`). if !forward_only { - let action = if predictor.alternate_screen { + let action = if predictor.alternate_screen || frame_renderer.has_pending() { disconnect_status.on_suppressed() } else { disconnect_status.on_tick(last_packet_at.elapsed()) @@ -9747,6 +9776,9 @@ struct PredictedCell { struct Predictor { mode: PredictMode, enabled: bool, + /// Suppress local display writes while authoritative output is waiting on a + /// slow terminal renderer. Input prediction state is still maintained. + output_backpressured: bool, /// True while the server is in the alternate screen (a full-screen TUI such /// as vim/htop); we never speculate there because we cannot model arbitrary /// cursor addressing safely. @@ -9803,6 +9835,7 @@ impl Predictor { Self { mode, enabled: enabled && mode != PredictMode::Off, + output_backpressured: false, alternate_screen: false, mouse_tracking: TerminalMouseMode::None, output_parse_tail: Vec::new(), @@ -10067,6 +10100,9 @@ impl Predictor { /// Whether we should *display* predictions right now under the active policy. fn should_display(&self) -> bool { + if self.output_backpressured { + return false; + } match self.mode { PredictMode::Off => false, PredictMode::Always => true, @@ -10085,6 +10121,10 @@ impl Predictor { } } + fn set_output_backpressured(&mut self, value: bool) { + self.output_backpressured = value; + } + /// Update the SRTT/flag hysteresis latches from the current estimate. fn update_triggers(&mut self) { let srtt = self.srtt_ms.unwrap_or(0.0); @@ -10500,6 +10540,80 @@ async fn detach_once(socket: &UdpSocket, cred: &CachedCredential, seq: u64) -> R Ok(()) } +const TERMINAL_RENDER_QUEUE_CAPACITY: usize = 256; + +/// Keeps potentially slow console writes off the UDP event loop. Frame ACKs are +/// emitted only after `recv` reports completion, so the server never retires +/// output that has merely been queued locally rather than displayed. +struct TerminalFrameRenderer { + jobs: Option>, + completed: mpsc::UnboundedReceiver>, + pending: usize, + thread: Option>, +} + +impl TerminalFrameRenderer { + fn new() -> Result { + let (job_tx, mut job_rx) = mpsc::channel::(TERMINAL_RENDER_QUEUE_CAPACITY); + let (completed_tx, completed_rx) = mpsc::unbounded_channel(); + let thread = std::thread::Builder::new() + .name("dosh-terminal-render".to_string()) + .spawn(move || { + while let Some(frame) = job_rx.blocking_recv() { + let result = render_frame(&frame) + .map(|()| frame) + .map_err(|err| format!("render terminal frame: {err:#}")); + if completed_tx.send(result).is_err() { + break; + } + } + })?; + Ok(Self { + jobs: Some(job_tx), + completed: completed_rx, + pending: 0, + thread: Some(thread), + }) + } + + fn enqueue(&mut self, frame: Frame) -> Result<()> { + let Some(jobs) = self.jobs.as_ref() else { + return Err(anyhow!("terminal renderer is closed")); + }; + jobs.try_send(frame).map_err(|err| match err { + mpsc::error::TrySendError::Full(_) => anyhow!( + "terminal renderer exceeded its bounded {}-frame queue", + TERMINAL_RENDER_QUEUE_CAPACITY + ), + mpsc::error::TrySendError::Closed(_) => anyhow!("terminal renderer stopped"), + })?; + self.pending += 1; + Ok(()) + } + + fn has_pending(&self) -> bool { + self.pending > 0 + } + + async fn recv(&mut self) -> Result { + let result = + self.completed.recv().await.ok_or_else(|| { + anyhow!("terminal renderer stopped before completing queued output") + })?; + self.pending = self.pending.saturating_sub(1); + result.map_err(anyhow::Error::msg) + } +} + +impl Drop for TerminalFrameRenderer { + fn drop(&mut self) { + self.jobs.take(); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + fn render_frame(frame: &Frame) -> Result<()> { let mut stdout = std::io::stdout(); stdout.write_all(&render_frame_bytes(frame))?; diff --git a/tests/client_terminal_runtime.rs b/tests/client_terminal_runtime.rs index cfb2fa7..144abe2 100644 --- a/tests/client_terminal_runtime.rs +++ b/tests/client_terminal_runtime.rs @@ -48,6 +48,7 @@ struct CachedCredentialWire { #[derive(Debug)] enum ServerObservation { + BulkSent, Input(Vec), Resize(u16, u16), } @@ -159,6 +160,70 @@ fn native_client_terminal_round_trip_is_platform_complete() { } } +#[test] +fn terminal_output_backpressure_does_not_block_input_transport() { + const PROBE: &[u8] = b"DOSH_BACKPRESSURE_INPUT\r"; + + let dir = tempfile::tempdir().unwrap(); + let home = dir.path().join("home"); + let cache = dir.path().join("credentials"); + fs::create_dir_all(home.join(".config/dosh")).unwrap(); + fs::create_dir_all(&cache).unwrap(); + + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(8))) + .unwrap(); + let port = socket.local_addr().unwrap().port(); + write_client_fixture(&home, &cache, port); + let (observation_tx, observation_rx) = mpsc::channel(); + let server = thread::spawn(move || run_backpressure_server(socket, observation_tx, PROBE)); + + let pty = NativePtySystem::default(); + let pair = pty + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + let mut reader = pair.master.try_clone_reader().unwrap(); + let mut writer = pair.master.take_writer().unwrap(); + let mut command = client_command(dir.path(), port); + command.env("HOME", home.to_string_lossy().to_string()); + command.env("USERPROFILE", home.to_string_lossy().to_string()); + command.env("APPDATA", home.to_string_lossy().to_string()); + command.env("LOCALAPPDATA", home.to_string_lossy().to_string()); + command.env("TERM", "xterm-256color"); + let mut child = pair.slave.spawn_command(command).unwrap(); + drop(pair.slave); + + wait_for_observation(&observation_rx, Duration::from_secs(5), |observation| { + matches!(observation, ServerObservation::BulkSent) + }); + thread::sleep(Duration::from_millis(100)); + writer.write_all(PROBE).unwrap(); + writer.flush().unwrap(); + wait_for_input(&observation_rx, PROBE, Duration::from_secs(2)); + + let reader_thread = thread::spawn(move || { + let mut buf = [0u8; 16 * 1024]; + while let Ok(n) = reader.read(&mut buf) { + if n == 0 { + break; + } + } + }); + let status = child.wait().unwrap(); + drop(writer); + drop(pair.master); + reader_thread.join().unwrap(); + server.join().unwrap(); + + assert!(status.success(), "Dosh client exited with {status:?}"); +} + fn write_client_fixture(home: &Path, cache: &Path, port: u16) { let config = ClientConfig { server: "local".to_string(), @@ -344,6 +409,74 @@ fn run_fake_terminal_server(socket: UdpSocket, observations: mpsc::Sender, + expected_input: &[u8], +) { + let mut peer = None; + let mut bulk_sent = false; + let mut received_input = Vec::new(); + let mut buf = [0u8; 65535]; + loop { + let (n, source) = socket.recv_from(&mut buf).unwrap(); + let packet = protocol::decode(&buf[..n]).unwrap(); + match packet.header.kind { + PacketKind::ResumeRequest => { + peer = Some(source); + send_frame( + &socket, + source, + PacketKind::ResumeOk, + 1, + 10, + b"DOSH_BACKPRESSURE_READY", + true, + false, + ); + } + PacketKind::Ack if !bulk_sent => { + let bulk = vec![b'x'; 60 * 1024]; + send_frame( + &socket, + peer.unwrap_or(source), + PacketKind::Frame, + 2, + 11, + &bulk, + false, + false, + ); + bulk_sent = true; + observations.send(ServerObservation::BulkSent).unwrap(); + } + PacketKind::Input => { + let plain = + protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap(); + let input: Input = protocol::from_body(&plain).unwrap(); + received_input.extend_from_slice(&input.bytes); + observations + .send(ServerObservation::Input(input.bytes)) + .unwrap(); + if contains(&received_input, expected_input) { + send_frame( + &socket, + peer.unwrap_or(source), + PacketKind::Frame, + 3, + 12, + b"DOSH_BACKPRESSURE_DONE", + false, + true, + ); + break; + } + } + _ => {} + } + } +} + #[allow(clippy::too_many_arguments)] fn send_frame( socket: &UdpSocket, @@ -429,6 +562,22 @@ fn wait_for_input( panic!("terminal input was not delivered; got {input:?}"); } +fn wait_for_observation( + observations: &mpsc::Receiver, + timeout: Duration, + matches: impl Fn(&ServerObservation) -> bool, +) { + let deadline = Instant::now() + timeout; + while let Some(remaining) = deadline.checked_duration_since(Instant::now()) { + match observations.recv_timeout(remaining) { + Ok(observation) if matches(&observation) => return, + Ok(_) => {} + Err(err) => panic!("expected server observation was not received: {err}"), + } + } + panic!("expected server observation was not received"); +} + fn contains(haystack: &[u8], needle: &[u8]) -> bool { !needle.is_empty() && haystack.windows(needle.len()).any(|bytes| bytes == needle) }