From 72b8bd94e8cd52af9a5e23a1f927742add841395 Mon Sep 17 00:00:00 2001 From: Rui Carmo Date: Thu, 8 Oct 2026 22:33:50 +0100 Subject: [PATCH 1/5] fix(egfx): respect client frame acknowledgement limits --- crates/ironrdp-egfx/src/client.rs | 19 +- crates/ironrdp-egfx/src/server.rs | 457 ++++++++++++------ crates/ironrdp-server/src/gfx.rs | 34 ++ crates/ironrdp-server/src/server.rs | 56 ++- .../tests/egfx/server.rs | 243 +++++++++- 5 files changed, 615 insertions(+), 194 deletions(-) diff --git a/crates/ironrdp-egfx/src/client.rs b/crates/ironrdp-egfx/src/client.rs index 59eb0482ea..5282b72274 100644 --- a/crates/ironrdp-egfx/src/client.rs +++ b/crates/ironrdp-egfx/src/client.rs @@ -1201,10 +1201,10 @@ impl GraphicsPipelineClient { self.handler.on_frame_complete(frame_id); // Per [3.3.5.12]: client MUST send FrameAcknowledge after EndFrame. - // We send the actual queue depth (not Unavailable / 0xFFFFFFFF as FreeRDP does); - // the real value gives the server backpressure information for frame pacing. + // QueueDepth counts queued bytes, not decoded/queued frames. This client + // does not measure queued bytes, so report Unavailable (not Suspend). let ack = GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { - queue_depth: QueueDepth::from_u32(self.frames_queued), + queue_depth: QueueDepth::Unavailable, frame_id, total_frames_decoded: self.total_frames_decoded, }); @@ -2683,4 +2683,17 @@ mod tests { a destRect-relative reading would take (28, 32) instead (issue #2042)" ); } + #[test] + fn frame_ack_queue_depth_is_not_a_frame_count() { + let mut handler = GraphicsPipelineClient::new(Box::new(TestHandler), None); + handler.total_frames_decoded = 3; + let responses = handler.handle_end_frame(7).unwrap(); + let bytes = ironrdp_core::encode_vec(responses[0].as_ref()).unwrap(); + let GfxPdu::FrameAcknowledge(ack) = ironrdp_core::decode::(&bytes).unwrap() else { + panic!("expected frame ack") + }; + assert_eq!(ack.frame_id, 7); + assert_eq!(ack.total_frames_decoded, 4); + assert_eq!(ack.queue_depth, QueueDepth::Unavailable); + } } diff --git a/crates/ironrdp-egfx/src/server.rs b/crates/ironrdp-egfx/src/server.rs index 5096607ffc..60cddf57e9 100644 --- a/crates/ironrdp-egfx/src/server.rs +++ b/crates/ironrdp-egfx/src/server.rs @@ -56,6 +56,7 @@ //! ``` use std::collections::{HashMap, VecDeque}; +use std::num::NonZeroU32; use std::time::Instant; use ironrdp_core::{Encode, EncodeResult, WriteCursor, decode, impl_as_any}; @@ -511,8 +512,12 @@ pub struct FrameTracker { ack_suspended: bool, /// Next frame ID to assign next_frame_id: u32, - /// Maximum frames in flight before backpressure - max_in_flight: u32, + /// First frame ID issued in the current channel generation. + generation_first_frame_id: u32, + /// Operator-configured maximum frames in flight (zero means unlimited). + configured_max_in_flight: u32, + /// Optional non-zero ceiling advertised by the active client. + client_max_in_flight: Option, /// Total frames sent total_sent: u64, /// Total frames acknowledged @@ -540,7 +545,9 @@ impl FrameTracker { client_queue_depth: 0, ack_suspended: false, next_frame_id: 0, - max_in_flight: DEFAULT_MAX_FRAMES_IN_FLIGHT, + generation_first_frame_id: 0, + configured_max_in_flight: DEFAULT_MAX_FRAMES_IN_FLIGHT, + client_max_in_flight: None, total_sent: 0, total_acked: 0, last_backpressure_state: false, @@ -562,7 +569,7 @@ impl FrameTracker { if now_backpressure { debug!( in_flight, - max_in_flight = self.max_in_flight, + max_in_flight = self.max_in_flight(), ack_suspended = self.ack_suspended, client_queue_depth = self.client_queue_depth, "EGFX backpressure ENGAGED, encoder must wait for FrameAcknowledge" @@ -570,7 +577,7 @@ impl FrameTracker { } else { debug!( in_flight, - max_in_flight = self.max_in_flight, + max_in_flight = self.max_in_flight(), "EGFX backpressure RELEASED, encoder may proceed" ); } @@ -593,9 +600,29 @@ impl FrameTracker { } } - /// Set maximum frames in flight + /// Set the operator-configured maximum frames in flight. + /// + /// Zero means unlimited, while a negotiated client ceiling still applies. pub fn set_max_in_flight(&mut self, max: u32) { - self.max_in_flight = max; + self.configured_max_in_flight = max; + } + + /// Store a non-zero client-advertised ceiling and return the effective limit. + fn clamp_max_in_flight(&mut self, client_max: NonZeroU32) -> u32 { + self.client_max_in_flight = Some(client_max); + self.max_in_flight() + } + + /// Effective maximum number of frames permitted in flight. + /// + /// Zero means unlimited only when no client ceiling is present. + pub fn max_in_flight(&self) -> u32 { + match (self.configured_max_in_flight, self.client_max_in_flight) { + (0, None) => 0, + (0, Some(client)) => client.get(), + (configured, None) => configured, + (configured, Some(client)) => configured.min(client.get()), + } } /// Allocate a new frame ID and track it @@ -632,8 +659,11 @@ impl FrameTracker { } } - /// Handle frame acknowledgment from client + /// Handle a tracked frame acknowledgment from the client. pub fn acknowledge(&mut self, frame_id: u32, queue_depth: u32) -> Option { + // Validate the acknowledgement before it can alter flow-control state. + // A stale acknowledgement must not suspend the current generation. + let info = self.unacknowledged.remove(&frame_id)?; let suspending = queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH; if suspending { self.ack_suspended = true; @@ -643,37 +673,17 @@ impl FrameTracker { self.client_queue_depth = queue_depth; } - // A suspending Frame Acknowledge still acknowledges: take the frame - // out before the clear below, since it carries the round-trip sample - // and byte count the QoE report is built from. - let info = self.unacknowledged.remove(&frame_id); - if info.is_some() { - self.total_acked += 1; - } + self.total_acked += 1; if suspending { - // [MS-RDPEGFX] 3.2.5.13: "the server MUST clear the Unacknowledged - // Frames ADM element and MUST NOT expect any further - // RDPGFX_FRAME_ACKNOWLEDGE_PDU messages from the client". The one - // frame this PDU acknowledges was already removed above, so its QoE - // sample survives. + // [MS-RDPEGFX] 3.2.5.13: a client that suspends acknowledgements + // makes the remaining unacknowledged frame state unavailable. self.unacknowledged.clear(); } // Edge-trigger after the remove/clear so an ack that releases // backpressure logs. self.emit_state_transitions(); - info - } - - /// Whether `frame_id` was ever handed out by [`begin_frame`]. - /// - /// Frame IDs are assigned sequentially, so any ID below the next one to - /// assign has been issued at some point. A `u32` frame counter does not - /// wrap within a session. Used to tell a benign acknowledgement for a - /// frame we no longer track (a resume after suspension cleared the map, or - /// a stale duplicate) from an acknowledgement for an ID never sent. - pub(crate) fn was_issued(&self, frame_id: u32) -> bool { - frame_id < self.next_frame_id + Some(info) } /// Number of frames in flight @@ -688,7 +698,8 @@ impl FrameTracker { /// Check if backpressure should be applied pub fn should_backpressure(&self) -> bool { - !self.ack_suspended && self.in_flight() >= self.max_in_flight + let max_in_flight = self.max_in_flight(); + !self.ack_suspended && max_in_flight != 0 && self.in_flight() >= max_in_flight } /// Get client queue depth @@ -711,14 +722,76 @@ impl FrameTracker { self.total_acked } - /// Clear all tracking state + /// Resume tracking only for a non-suspend ACK naming a frame issued in this generation. + /// Frames sent during suspension are intentionally absent from the tracked map. + fn resume_from_untracked_ack(&mut self, frame_id: u32, queue_depth: u32) -> bool { + let issued = frame_id.wrapping_sub(self.generation_first_frame_id) + < self.next_frame_id.wrapping_sub(self.generation_first_frame_id); + if !self.ack_suspended || queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH || !issued { + return false; + } + self.ack_suspended = false; + self.client_queue_depth = queue_depth; + self.emit_state_transitions(); + true + } + + /// Clear all per-channel tracking state while preserving operator config. pub fn clear(&mut self) { self.unacknowledged.clear(); self.client_queue_depth = 0; self.ack_suspended = false; + self.generation_first_frame_id = self.next_frame_id; + self.last_backpressure_state = false; + self.last_ack_suspended_state = false; } } +/// Typed reason why an encoded frame could not be submitted to EGFX. +/// +/// Backpressure is an expected transient flow-control condition. Callers must +/// not treat it as an encoder or channel failure. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum FrameSubmissionError { + /// EGFX capability negotiation has not completed or the channel is closed. + NotReady, + /// The negotiated client capability does not include AVC420. + Avc420Unsupported, + /// The negotiated client capability does not include AVC444. + Avc444Unsupported, + /// The acknowledgement window is full. + Backpressured { + frames_in_flight: u32, + max_frames_in_flight: u32, + }, + /// The requested surface does not exist. + UnknownSurface { surface_id: u16 }, + /// The requested frame does not satisfy the EGFX framing rules. + InvalidFrame, +} + +impl core::fmt::Display for FrameSubmissionError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::NotReady => f.write_str("EGFX channel is not ready"), + Self::Avc420Unsupported => f.write_str("AVC420 is not supported by the client"), + Self::Avc444Unsupported => f.write_str("AVC444 is not supported by the client"), + Self::Backpressured { + frames_in_flight, + max_frames_in_flight, + } => write!( + f, + "EGFX acknowledgement window is full ({frames_in_flight}/{max_frames_in_flight})" + ), + Self::UnknownSurface { surface_id } => write!(f, "unknown EGFX surface {surface_id}"), + Self::InvalidFrame => f.write_str("EGFX frame does not satisfy the negotiated framing rules"), + } + } +} + +impl core::error::Error for FrameSubmissionError {} + // ============================================================================ // Capability Negotiation // ============================================================================ @@ -1159,6 +1232,23 @@ impl GraphicsPipelineServer { } } + /// Reset all state owned by one DVC channel generation. + fn reset_channel_state(&mut self) { + self.state = ServerState::WaitingForCapabilities; + self.negotiated_caps = None; + self.codec_caps = CodecCapabilities::default(); + self.surfaces = Surfaces::new(); + self.frames.clear(); + self.frames.client_max_in_flight = None; + self.output_width = 0; + self.output_height = 0; + self.reset_graphics_sent = false; + self.output_queue.clear(); + self.channel_id = None; + self.zgfx_compressor = Compressor::new(); + self.qoe.clear(); + } + /// Create a server with ZGFX compression enabled for output. /// /// When `compression_mode` is `Auto` or `Always`, `drain_output()` will @@ -1416,11 +1506,25 @@ impl GraphicsPipelineServer { self.frames.client_queue_depth() } - /// Set the maximum frames in flight before backpressure + /// Set the maximum frames in flight before backpressure. pub fn set_max_frames_in_flight(&mut self, max: u32) { self.frames.set_max_in_flight(max); } + /// Clamp the configured frame window to a non-zero client-advertised limit. + /// + /// A configured value of zero remains unlimited only until the client + /// advertises a non-zero ceiling. Returns the effective limit. + pub fn clamp_max_frames_in_flight(&mut self, client_max: NonZeroU32) -> u32 { + self.frames.clamp_max_in_flight(client_max) + } + + /// Effective maximum number of frames in flight. + #[must_use] + pub fn max_frames_in_flight(&self) -> u32 { + self.frames.max_in_flight() + } + // ======================================================================== // QoE Statistics // ======================================================================== @@ -1507,40 +1611,50 @@ impl GraphicsPipelineServer { rect.left < rect.right && rect.top < rect.bottom && rect.right <= surface.width && rect.bottom <= surface.height } - /// Queue an H.264 AVC420 frame for transmission - /// - /// Returns `Some(frame_id)` if queued, `None` if backpressure is active, - /// server not ready, or AVC420 not supported. - pub fn send_avc420_frame( - &mut self, - surface_id: u16, - h264_data: &[u8], - regions: &[Avc420Region], - timestamp_ms: u32, - ) -> Option { + /// Check whether an AVC420 frame can be submitted without encoding it. + pub fn avc420_submission_state(&self, surface_id: u16) -> Result<(), FrameSubmissionError> { if !self.is_ready() { - return None; + return Err(FrameSubmissionError::NotReady); } if !self.supports_avc420() { - return None; + return Err(FrameSubmissionError::Avc420Unsupported); } if self.should_backpressure() { - self.qoe.record_backpressure(); - return None; + return Err(FrameSubmissionError::Backpressured { + frames_in_flight: self.frames_in_flight(), + max_frames_in_flight: self.max_frames_in_flight(), + }); + } + if self.surfaces.get(surface_id).is_none() { + return Err(FrameSubmissionError::UnknownSurface { surface_id }); } + Ok(()) + } - let surface = self.surfaces.get(surface_id)?; + /// Queue an H.264 AVC420 frame for transmission with a typed rejection. + pub fn try_send_avc420_frame( + &mut self, + surface_id: u16, + h264_data: &[u8], + regions: &[Avc420Region], + timestamp_ms: u32, + ) -> Result { + if let Err(error) = self.avc420_submission_state(surface_id) { + if matches!(error, FrameSubmissionError::Backpressured { .. }) { + self.qoe.record_backpressure(); + } + return Err(error); + } + let surface = self.surfaces.get(surface_id).expect("surface existence checked above"); let timestamp = Self::make_timestamp(timestamp_ms); let frame_id = self.frames.begin_frame(timestamp); - let encoded_stream = encode_avc420_bitmap_stream(regions, h264_data); let target_rect = Self::compute_dest_rect(regions, surface.width, surface.height); - // MS-RDPEGFX requires three-PDU sequence per frame + // MS-RDPEGFX requires three-PDU sequence per frame. self.output_queue .push_back(GfxPdu::StartFrame(StartFramePdu { timestamp, frame_id })); - self.output_queue.push_back(GfxPdu::WireToSurface1(WireToSurface1Pdu { surface_id, codec_id: Codec1Type::Avc420, @@ -1548,25 +1662,48 @@ impl GraphicsPipelineServer { destination_rectangle: target_rect, bitmap_data: encoded_stream, })); - self.output_queue.push_back(GfxPdu::EndFrame(EndFramePdu { frame_id })); - Some(frame_id) + Ok(frame_id) + } + + /// Compatibility wrapper returning `None` for any submission rejection. + pub fn send_avc420_frame( + &mut self, + surface_id: u16, + h264_data: &[u8], + regions: &[Avc420Region], + timestamp_ms: u32, + ) -> Option { + self.try_send_avc420_frame(surface_id, h264_data, regions, timestamp_ms) + .ok() + } + + /// Check whether an AVC444 frame can be submitted without encoding it. + pub fn avc444_submission_state(&self, surface_id: u16) -> Result<(), FrameSubmissionError> { + if !self.is_ready() { + return Err(FrameSubmissionError::NotReady); + } + if !self.supports_avc444() { + return Err(FrameSubmissionError::Avc444Unsupported); + } + if self.should_backpressure() { + return Err(FrameSubmissionError::Backpressured { + frames_in_flight: self.frames_in_flight(), + max_frames_in_flight: self.max_frames_in_flight(), + }); + } + if self.surfaces.get(surface_id).is_none() { + return Err(FrameSubmissionError::UnknownSurface { surface_id }); + } + Ok(()) } - /// Queue an H.264 AVC444 frame for transmission + /// Queue an H.264 AVC444 frame for transmission with a typed rejection. /// /// AVC444 uses two streams: luma (Y) and chroma (UV). Set `chroma_data` to /// `None` for luma-only transmission. - /// - /// Unlike [`Self::send_mixed_frame()`], this does not check the regions. Each - /// must be non-empty and inside the surface, with a QP of at most 51 and a - /// quality of at most 100 (MS-RDPEGFX 2.2.4.4.2). A `None` from this sender has - /// only meant not ready, not supported or backpressure, and callers read it that - /// way. - /// - /// Returns `Some(frame_id)` if queued, `None` if not supported or backpressured. - pub fn send_avc444_frame( + pub fn try_send_avc444_frame( &mut self, surface_id: u16, luma_data: &[u8], @@ -1574,7 +1711,7 @@ impl GraphicsPipelineServer { chroma_data: Option<&[u8]>, chroma_regions: Option<&[Avc420Region]>, timestamp_ms: u32, - ) -> Option { + ) -> Result { let (encoding, stream2_data, stream2_regions) = if let (Some(chroma_data), Some(chroma_regions)) = (chroma_data, chroma_regions) { (Encoding::LUMA_AND_CHROMA, Some(chroma_data), Some(chroma_regions)) @@ -1582,7 +1719,7 @@ impl GraphicsPipelineServer { (Encoding::LUMA, None, None) }; - self.send_avc444_frame_with_encoding( + self.try_send_avc444_frame_with_encoding( Codec1Type::Avc444, surface_id, encoding, @@ -1594,22 +1731,32 @@ impl GraphicsPipelineServer { ) } - /// Queue an H.264 AVC444v2 frame for transmission. + /// Compatibility wrapper returning `None` for any submission rejection. + pub fn send_avc444_frame( + &mut self, + surface_id: u16, + luma_data: &[u8], + luma_regions: &[Avc420Region], + chroma_data: Option<&[u8]>, + chroma_regions: Option<&[Avc420Region]>, + timestamp_ms: u32, + ) -> Option { + self.try_send_avc444_frame( + surface_id, + luma_data, + luma_regions, + chroma_data, + chroma_regions, + timestamp_ms, + ) + .ok() + } + + /// Compatibility wrapper returning `None` for any AVC444v2 rejection. /// /// `encoding` selects the AVC444v2 stream layout: LC=0 carries luma in /// stream 1 and chroma in stream 2, LC=1 carries luma in stream 1, and LC=2 /// carries chroma in stream 1. - /// - /// Unlike [`Self::send_mixed_frame()`], this does not check the regions. Each - /// must be non-empty and inside the surface, with a QP of at most 51 and a - /// quality of at most 100 (MS-RDPEGFX 2.2.4.4.2). A `None` from this sender has - /// only meant not ready, not supported or backpressure, and callers read it that - /// way. - /// - /// Returns `Some(frame_id)` if queued, `None` if the stream shape is - /// invalid, AVC444 is not supported, or backpressure is active. - /// - /// [2.2.4.6 RFX_AVC444V2_BITMAP_STREAM]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpegfx/3b337b87-f478-4786-a63b-97794aa72075 #[expect( clippy::too_many_arguments, reason = "the public API mirrors the two AVC444v2 streams" @@ -1624,7 +1771,34 @@ impl GraphicsPipelineServer { stream2_regions: Option<&[Avc420Region]>, timestamp_ms: u32, ) -> Option { - self.send_avc444_frame_with_encoding( + self.try_send_avc444v2_frame( + surface_id, + encoding, + stream1_data, + stream1_regions, + stream2_data, + stream2_regions, + timestamp_ms, + ) + .ok() + } + + /// Queue an AVC444v2 frame with a typed submission rejection. + #[expect( + clippy::too_many_arguments, + reason = "the public API mirrors the two AVC444v2 streams" + )] + pub fn try_send_avc444v2_frame( + &mut self, + surface_id: u16, + encoding: Encoding, + stream1_data: &[u8], + stream1_regions: &[Avc420Region], + stream2_data: Option<&[u8]>, + stream2_regions: Option<&[Avc420Region]>, + timestamp_ms: u32, + ) -> Result { + self.try_send_avc444_frame_with_encoding( Codec1Type::Avc444v2, surface_id, encoding, @@ -1640,7 +1814,7 @@ impl GraphicsPipelineServer { clippy::too_many_arguments, reason = "the shared implementation receives both AVC444 streams" )] - fn send_avc444_frame_with_encoding( + fn try_send_avc444_frame_with_encoding( &mut self, codec_id: Codec1Type, surface_id: u16, @@ -1650,34 +1824,27 @@ impl GraphicsPipelineServer { stream2_data: Option<&[u8]>, stream2_regions: Option<&[Avc420Region]>, timestamp_ms: u32, - ) -> Option { + ) -> Result { // A second sub-stream is its regions and its data together; one without the // other is never valid. let stream2 = match (stream2_regions, stream2_data) { (Some(regions), Some(data)) => Some((regions, data)), (None, None) => None, - _ => return None, + _ => return Err(FrameSubmissionError::InvalidFrame), }; if !Self::avc444_stream_shape_is_valid(encoding, stream2.is_some()) { - return None; - } - - if !self.is_ready() { - return None; + return Err(FrameSubmissionError::InvalidFrame); } - if !self.supports_avc444() { - return None; - } - if self.should_backpressure() { - self.qoe.record_backpressure(); - return None; + if let Err(error) = self.avc444_submission_state(surface_id) { + if matches!(error, FrameSubmissionError::Backpressured { .. }) { + self.qoe.record_backpressure(); + } + return Err(error); } - let surface = self.surfaces.get(surface_id)?; - + let surface = self.surfaces.get(surface_id).expect("surface existence checked above"); let timestamp = Self::make_timestamp(timestamp_ms); let frame_id = self.frames.begin_frame(timestamp); - let wire_pdu = Self::avc444_wire_pdu(codec_id, surface, encoding, (stream1_regions, stream1_data), stream2); self.output_queue @@ -1685,7 +1852,7 @@ impl GraphicsPipelineServer { self.output_queue.push_back(GfxPdu::WireToSurface1(wire_pdu)); self.output_queue.push_back(GfxPdu::EndFrame(EndFramePdu { frame_id })); - Some(frame_id) + Ok(frame_id) } /// Whether the second AVC444 sub-stream is present exactly when the LC @@ -2344,48 +2511,40 @@ impl GraphicsPipelineServer { } fn handle_frame_acknowledge(&mut self, pdu: FrameAcknowledgePdu) { - let queue_depth = pdu.queue_depth.to_u32(); - let suspended = queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH; - - if let Some(info) = self.frames.acknowledge(pdu.frame_id, queue_depth) { - let rtt = info.sent_at.elapsed(); - self.qoe.record_rtt(rtt); - self.qoe.record_frame_ack(info.size_bytes); - // DEBUG: visible at default log levels. FrameAcknowledge is the - // single most important signal in the EGFX flow-control loop — - // without it the server cannot decide when to release backpressure. - // Logging at TRACE only made the loop invisible under normal - // operator-friendly log levels. - debug!( - frame_id = pdu.frame_id, - latency_us = u64::try_from(rtt.as_micros()).unwrap_or(u64::MAX), - queue_depth, - suspended, - in_flight_after = self.frames.in_flight(), - "EGFX FrameAcknowledge received" - ); - } else if self.frames.was_issued(pdu.frame_id) { - // Issued but no longer tracked. This is the normal resume path: - // while acknowledgements are suspended the server stops tracking - // frames ([MS-RDPEGFX] 2.2.2.13), and the client opts back in by - // acknowledging the END_FRAME it last decoded -- which may be one - // of those untracked frames -- or this is a stale/duplicate ack. - // Either way the frame is genuinely done, not a protocol violation. - debug!( + if !self.is_ready() { + warn!( frame_id = pdu.frame_id, - queue_depth, suspended, "EGFX FrameAcknowledge for an untracked issued frame (resume or stale ack)" + "Ignoring frame acknowledgement outside active EGFX generation" ); - } else { - // PROTOCOL COMPLIANCE: per MS-RDPEGFX 2.2.4.3 the client MUST only - // acknowledge frame_ids the server has sent. An ack for a frame_id - // the server never issued indicates client misbehavior or a session - // out of sync. + return; + } + + let queue_depth = pdu.queue_depth.to_u32(); + let suspended = queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH; + let Some(info) = self.frames.acknowledge(pdu.frame_id, queue_depth) else { + if self.frames.resume_from_untracked_ack(pdu.frame_id, queue_depth) { + self.handler + .on_frame_ack(pdu.frame_id, queue_depth, pdu.total_frames_decoded); + return; + } warn!( frame_id = pdu.frame_id, - queue_depth, "EGFX FrameAcknowledge for UNKNOWN frame_id, protocol violation" + queue_depth, "Ignoring unknown or stale frame acknowledgement" ); - } + return; + }; + let rtt = info.sent_at.elapsed(); + self.qoe.record_rtt(rtt); + self.qoe.record_frame_ack(info.size_bytes); + debug!( + frame_id = pdu.frame_id, + latency_us = u64::try_from(rtt.as_micros()).unwrap_or(u64::MAX), + queue_depth, + suspended, + in_flight_after = self.frames.in_flight(), + "EGFX FrameAcknowledge received" + ); self.handler .on_frame_ack(pdu.frame_id, queue_depth, pdu.total_frames_decoded); } @@ -2418,6 +2577,9 @@ impl DvcProcessor for GraphicsPipelineServer { } fn start(&mut self, channel_id: u32) -> PduResult> { + // The core capability clamp is applied before the dynamic channel opens, + // so preserve it here. close() already clears the prior generation. + self.state = ServerState::WaitingForCapabilities; self.channel_id = Some(channel_id); debug!(channel_id, "EGFX channel started"); Ok(vec![]) @@ -2425,8 +2587,8 @@ impl DvcProcessor for GraphicsPipelineServer { fn close(&mut self, _channel_id: u32) { debug!("EGFX channel closed"); + self.reset_channel_state(); self.state = ServerState::Closed; - self.reset_graphics_sent = false; self.handler.on_close(); } @@ -2665,29 +2827,6 @@ mod capability_negotiation_tests { mod tests { use super::*; - #[test] - fn was_issued_separates_stale_resume_acks_from_never_sent_ids() { - let ts = Timestamp { - milliseconds: 0, - seconds: 0, - minutes: 0, - hours: 0, - }; - let mut frames = FrameTracker::new(); - let a = frames.begin_frame(ts); - let b = frames.begin_frame(ts); - - // Both were handed out, so both are "issued" even after the map is - // cleared -- a resume ack for either is benign, not a violation. - frames.clear(); - assert!(frames.was_issued(a)); - assert!(frames.was_issued(b)); - - // An id the server never assigned is a real unknown frame. - assert!(!frames.was_issued(b.wrapping_add(1))); - assert!(!frames.was_issued(u32::MAX)); - } - struct DefaultsHandler; impl GraphicsPipelineHandler for DefaultsHandler { diff --git a/crates/ironrdp-server/src/gfx.rs b/crates/ironrdp-server/src/gfx.rs index 2347edd595..71a9c5a3ff 100644 --- a/crates/ironrdp-server/src/gfx.rs +++ b/crates/ironrdp-server/src/gfx.rs @@ -41,6 +41,17 @@ pub trait GfxServerFactory: ServerEventSender + Send { } } +/// Build an EGFX bridge while always retaining a shared server handle. +pub(crate) fn build_server_and_handle(factory: &dyn GfxServerFactory) -> (GfxDvcBridge, GfxServerHandle) { + if let Some((bridge, handle)) = factory.build_server_with_handle() { + return (bridge, handle); + } + + let handler = factory.build_gfx_handler(); + let server = Arc::new(Mutex::new(GraphicsPipelineServer::new(handler))); + (GfxDvcBridge::new(Arc::clone(&server)), server) +} + /// DVC bridge wrapping a shared `GraphicsPipelineServer`. /// /// Delegates all `DvcProcessor` methods to the inner server through a mutex, @@ -130,6 +141,29 @@ mod tests { fn on_ready(&mut self, _negotiated: &CapabilitySet) {} } + struct LegacyFactory; + + impl ServerEventSender for LegacyFactory { + fn set_sender(&mut self, _sender: tokio::sync::mpsc::UnboundedSender) {} + } + + impl GfxServerFactory for LegacyFactory { + fn build_gfx_handler(&self) -> Box { + Box::new(Handler) + } + } + + #[test] + fn legacy_factory_path_retains_clampable_handle() { + let (_bridge, handle) = build_server_and_handle(&LegacyFactory); + let mut server = handle.lock().expect("GfxServerHandle mutex poisoned"); + server.set_max_frames_in_flight(3); + assert_eq!( + server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")), + 1 + ); + } + #[test] fn egfx_is_found_whichever_way_it_was_registered() { let server = Arc::new(Mutex::new(GraphicsPipelineServer::new(Box::new(Handler)))); diff --git a/crates/ironrdp-server/src/server.rs b/crates/ironrdp-server/src/server.rs index 37af847670..37e5392158 100644 --- a/crates/ironrdp-server/src/server.rs +++ b/crates/ironrdp-server/src/server.rs @@ -1,6 +1,8 @@ use core::cell::RefCell; use core::fmt; use core::net::{IpAddr, SocketAddr}; +#[cfg(feature = "egfx")] +use core::num::NonZeroU32; use core::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use core::time::Duration; #[cfg(feature = "usb")] @@ -76,6 +78,14 @@ use ironrdp_rdpeusb::{InterfaceAlloc, server::UrbdrcControlServer, server::Urbdr /// TCP listen backlog size for the RDP server socket. const LISTENER_BACKLOG: u32 = 1024; + +#[cfg(feature = "egfx")] +fn client_core_frame_ack_limit(capabilities: &[CapabilitySet]) -> Option { + capabilities.iter().find_map(|capability| match capability { + CapabilitySet::FrameAcknowledge(capability) => NonZeroU32::new(capability.max_unacknowledged_frame_count), + _ => None, + }) +} const AUTO_RECONNECT_COOKIE_UPDATE_INTERVAL: Duration = Duration::from_secs(60 * 60); /// How long a single [`ironrdp_acceptor::accept_finalize`] pass may take before @@ -2082,14 +2092,9 @@ impl RdpServer { let dvc = { let mut dvc = dvc; if let Some(gfx_factory) = self.gfx_factory.as_deref() { - if let Some((bridge, handle)) = gfx_factory.build_server_with_handle() { - self.gfx_handle = Some(handle); - dvc = dvc.with_dynamic_channel(bridge); - } else { - let handler = gfx_factory.build_gfx_handler(); - let gfx_server = ironrdp_egfx::server::GraphicsPipelineServer::new(handler); - dvc = dvc.with_dynamic_channel(gfx_server); - } + let (processor, handle) = crate::gfx::build_server_and_handle(gfx_factory); + self.gfx_handle = Some(handle); + dvc = dvc.with_dynamic_channel(processor); } dvc }; @@ -4262,6 +4267,15 @@ impl RdpServer { .await?; } + #[cfg(feature = "egfx")] + if let Some(client_limit) = client_core_frame_ack_limit(&result.capabilities) + && let Some(gfx_handle) = self.gfx_handle.as_ref() + { + let mut server = gfx_handle + .lock() + .map_err(|_| ServerError::reason("graphics server", "GfxServerHandle mutex poisoned"))?; + server.clamp_max_frames_in_flight(client_limit); + } conn.static_channels = result.static_channels; conn.client_supports_heartbeat = result .client_early_capability_flags @@ -5969,3 +5983,29 @@ mod cliprdr_error_tests { assert!(writer.0.is_empty(), "a refused message has nothing to put on the wire"); } } + +#[cfg(all(test, feature = "egfx"))] +mod frame_ack_limit_tests { + use super::*; + use ironrdp_pdu::rdp::capability_sets::FrameAcknowledge; + #[test] + fn absent_and_zero_core_limit_are_unconstrained() { + assert_eq!(client_core_frame_ack_limit(&[]), None); + assert_eq!( + client_core_frame_ack_limit(&[CapabilitySet::FrameAcknowledge(FrameAcknowledge { + max_unacknowledged_frame_count: 0 + })]), + None + ); + } + #[test] + fn nonzero_core_limit_is_preserved() { + assert_eq!( + client_core_frame_ack_limit(&[CapabilitySet::FrameAcknowledge(FrameAcknowledge { + max_unacknowledged_frame_count: 1 + })]) + .map(NonZeroU32::get), + Some(1) + ); + } +} diff --git a/crates/ironrdp-testsuite-core/tests/egfx/server.rs b/crates/ironrdp-testsuite-core/tests/egfx/server.rs index aeac46b60f..cfed2f3f49 100644 --- a/crates/ironrdp-testsuite-core/tests/egfx/server.rs +++ b/crates/ironrdp-testsuite-core/tests/egfx/server.rs @@ -6,7 +6,8 @@ use ironrdp_egfx::pdu::{ Codec1Type, Codec2Type, Encoding, FrameAcknowledgePdu, GfxPdu, PixelFormat, QueueDepth, }; use ironrdp_egfx::server::{ - Avc444SubStream, Avc444Tile, GraphicsPipelineHandler, GraphicsPipelineServer, MixedTilePayload, QoeMetrics, Surface, + Avc444SubStream, Avc444Tile, FrameSubmissionError, GraphicsPipelineHandler, GraphicsPipelineServer, + MixedTilePayload, QoeMetrics, Surface, }; use ironrdp_graphics::zgfx::Decompressor; use ironrdp_pdu::geometry::ExclusiveRectangle; @@ -149,8 +150,11 @@ fn test_server_not_ready_before_capabilities() { let h264_data = vec![0x00, 0x00, 0x00, 0x01, 0x67]; let regions = vec![Avc420Region::full_frame(1920, 1080, 22)]; - let result = server.send_avc420_frame(0, &h264_data, ®ions, 0); - assert!(result.is_none()); + assert_eq!( + server.try_send_avc420_frame(0, &h264_data, ®ions, 0), + Err(FrameSubmissionError::NotReady) + ); + assert!(server.send_avc420_frame(0, &h264_data, ®ions, 0).is_none()); } #[test] @@ -815,6 +819,39 @@ fn mixed_frame_of_avc444_tiles_alone_is_allowed_before_10_4() { // Planar Frame Tests // ============================================================================ +#[test] +fn avc444v2_submission_preserves_typed_rejections() { + let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + let caps = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V10 { + flags: CapabilitiesV10Flags::empty(), + }])); + server.process(0, &encode_pdu(&caps)).unwrap(); + let surface = server.create_surface(64, 64).unwrap(); + server.drain_output(); + let regions = vec![Avc420Region::full_frame(64, 64, 22)]; + assert_eq!( + server.try_send_avc444v2_frame(surface, Encoding::LUMA_AND_CHROMA, &[1], ®ions, None, None, 1), + Err(FrameSubmissionError::InvalidFrame) + ); + assert!(!server.has_pending_output()); + assert_eq!(server.frames_in_flight(), 0); + server.set_max_frames_in_flight(1); + assert!( + server + .try_send_avc444v2_frame(surface, Encoding::LUMA, &[1], ®ions, None, None, 2) + .is_ok() + ); + server.drain_output(); + assert!(matches!( + server.try_send_avc444v2_frame(surface, Encoding::LUMA, &[2], ®ions, None, None, 3), + Err(FrameSubmissionError::Backpressured { + frames_in_flight: 1, + max_frames_in_flight: 1 + }) + )); + assert!(!server.has_pending_output()); +} + #[test] fn test_planar_frame_round_trip() { let handler = Box::new(TestHandler::new()); @@ -967,41 +1004,195 @@ fn test_resize() { fn test_frame_flow_control() { let handler = Box::new(TestHandler::new()); let mut server = GraphicsPipelineServer::new(handler); - server.set_max_frames_in_flight(2); + server.set_max_frames_in_flight(3); + assert_eq!(server.max_frames_in_flight(), 3); + assert_eq!( + server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")), + 1 + ); - // Negotiate capabilities with AVC420 let client_caps_pdu = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V8_1 { flags: CapabilitiesV81Flags::AVC420_ENABLED, }])); - let payload = encode_pdu(&client_caps_pdu); - let _output = server.process(0, &payload).expect("process failed"); + server + .process(0, &encode_pdu(&client_caps_pdu)) + .expect("process capabilities"); + let surface_id = server.create_surface(1920, 1080).expect("create surface"); + server.drain_output(); - // Create surface - let surface_id = server.create_surface(1920, 1080).unwrap(); - server.drain_output(); // Clear setup PDUs + let h264_data = vec![0x00, 0x00, 0x00, 0x01, 0x67]; + let regions = vec![Avc420Region::full_frame(1920, 1080, 22)]; + + let frame1 = server + .try_send_avc420_frame(surface_id, &h264_data, ®ions, 0) + .expect("first frame should be accepted"); + assert!(server.should_backpressure()); + assert_eq!(server.frames_in_flight(), 1); + assert_eq!( + server.avc420_submission_state(surface_id), + Err(FrameSubmissionError::Backpressured { + frames_in_flight: 1, + max_frames_in_flight: 1, + }) + ); + assert_eq!( + server.try_send_avc420_frame(surface_id, &h264_data, ®ions, 16), + Err(FrameSubmissionError::Backpressured { + frames_in_flight: 1, + max_frames_in_flight: 1, + }) + ); + + let ack_pdu = GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { + frame_id: frame1, + queue_depth: QueueDepth::AvailableBytes(1), + total_frames_decoded: 1, + }); + server.process(0, &encode_pdu(&ack_pdu)).expect("process ACK"); + assert_eq!(server.frames_in_flight(), 0); + assert_eq!(server.avc420_submission_state(surface_id), Ok(())); + assert!( + server + .try_send_avc420_frame(surface_id, &h264_data, ®ions, 33) + .is_ok() + ); +} + +#[test] +fn resize_retains_client_frame_ack_ceiling() { + let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + server.set_max_frames_in_flight(3); + server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).unwrap()); + server.resize(64, 64); + assert_eq!(server.max_frames_in_flight(), 1); +} + +#[test] +fn test_zero_configured_window_is_unlimited_until_client_clamp() { + let handler = Box::new(TestHandler::new()); + let mut server = GraphicsPipelineServer::new(handler); + server.set_max_frames_in_flight(0); + assert_eq!(server.max_frames_in_flight(), 0); + assert!(!server.should_backpressure()); + + assert_eq!( + server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")), + 1 + ); + server.set_max_frames_in_flight(3); + assert_eq!( + server.max_frames_in_flight(), + 1, + "a later setter must not exceed the negotiated client ceiling" + ); +} + +#[test] +fn test_close_reopen_clears_stale_frame_window_and_output() { + let handler = Box::new(TestHandler::new()); + let mut server = GraphicsPipelineServer::new(handler); + server.set_max_frames_in_flight(3); + server.start(7).expect("initial DVC start"); + server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")); + + let client_caps_pdu = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V8_1 { + flags: CapabilitiesV81Flags::AVC420_ENABLED, + }])); + server + .process(7, &encode_pdu(&client_caps_pdu)) + .expect("initial capabilities"); + let surface = server.create_surface(1920, 1080).expect("initial surface"); + server.drain_output(); let h264_data = vec![0x00, 0x00, 0x00, 0x01, 0x67]; let regions = vec![Avc420Region::full_frame(1920, 1080, 22)]; + server + .try_send_avc420_frame(surface, &h264_data, ®ions, 0) + .expect("initial frame"); + assert_eq!(server.frames_in_flight(), 1); + assert!(server.has_pending_output()); - // First two frames should succeed - let frame1 = server.send_avc420_frame(surface_id, &h264_data, ®ions, 0); - assert!(frame1.is_some()); + server.close(7); + assert_eq!(server.frames_in_flight(), 0); + assert!(!server.has_pending_output()); + assert_eq!(server.max_frames_in_flight(), 3); - let frame2 = server.send_avc420_frame(surface_id, &h264_data, ®ions, 16); - assert!(frame2.is_some()); + server.start(8).expect("reopened DVC start"); + server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")); + server + .process(8, &encode_pdu(&client_caps_pdu)) + .expect("reopened capabilities"); + let surface = server.create_surface(1920, 1080).expect("reopened surface"); + server.drain_output(); - // Check backpressure is active - assert!(server.should_backpressure()); - assert_eq!(server.frames_in_flight(), 2); + let stale_ack = GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { + frame_id: u32::MAX, + queue_depth: QueueDepth::Suspend, + total_frames_decoded: 1, + }); + server + .process(8, &encode_pdu(&stale_ack)) + .expect("stale ACK processing"); + assert_eq!(server.client_queue_depth(), 0); - // Third frame should fail due to backpressure - let frame3 = server.send_avc420_frame(surface_id, &h264_data, ®ions, 33); - assert!(frame3.is_none()); + assert!(server.try_send_avc420_frame(surface, &h264_data, ®ions, 16).is_ok()); + assert!( + server.should_backpressure(), + "stale suspend ACK must not disable the current frame ceiling" + ); } -// ============================================================================ -// QoE Statistics Tests -// ============================================================================ +#[test] +fn closed_generation_cannot_resume_or_suspend_current_tracking() { + let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + let caps = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V8_1 { + flags: CapabilitiesV81Flags::AVC420_ENABLED, + }])); + server.start(7).unwrap(); + server.process(7, &encode_pdu(&caps)).unwrap(); + let surface = server.create_surface(64, 64).unwrap(); + let regions = [Avc420Region::full_frame(64, 64, 22)]; + let old_frame = server.send_avc420_frame(surface, &[1], ®ions, 1).unwrap(); + let ack = |frame_id, queue_depth| { + GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { + frame_id, + queue_depth, + total_frames_decoded: 1, + }) + }; + server + .process(7, &encode_pdu(&ack(old_frame, QueueDepth::Unavailable))) + .unwrap(); + assert!(server.qoe_snapshot().is_some()); + server.close(7); + assert!(server.qoe_snapshot().is_none()); + server.start(8).unwrap(); + server.process(8, &encode_pdu(&caps)).unwrap(); + let surface = server.create_surface(64, 64).unwrap(); + server.set_max_frames_in_flight(1); + let current = server.send_avc420_frame(surface, &[2], ®ions, 2).unwrap(); + server + .process(8, &encode_pdu(&ack(old_frame, QueueDepth::Suspend))) + .unwrap(); + assert!(server.should_backpressure()); + server + .process(8, &encode_pdu(&ack(current, QueueDepth::Suspend))) + .unwrap(); + server + .process(8, &encode_pdu(&ack(old_frame, QueueDepth::Unavailable))) + .unwrap(); + let during_suspend = server.send_avc420_frame(surface, &[3], ®ions, 3).unwrap(); + assert_eq!( + server.frames_in_flight(), + 0, + "old ACK must not resume the current generation" + ); + server + .process(8, &encode_pdu(&ack(during_suspend, QueueDepth::Unavailable))) + .unwrap(); + server.send_avc420_frame(surface, &[4], ®ions, 4).unwrap(); + assert!(server.should_backpressure()); +} #[test] fn test_qoe_snapshot_none_before_data() { @@ -1268,6 +1459,8 @@ fn test_frames_sent_while_acknowledgement_is_suspended_do_not_hold_backpressure( "frames counted during the pause hold backpressure on for good: with no \ frame going out there is no End Frame left for the client to acknowledge" ); + assert_eq!(server.frames_in_flight(), 1, "resume must restore tracking"); + assert!(server.should_backpressure()); } #[test] @@ -1329,6 +1522,8 @@ fn test_suspension_does_not_shrink_the_window_for_the_rest_of_the_connection() { "frame {index} after resume was held back by a frame stranded in the pause" ); } + assert_eq!(server.frames_in_flight(), 3); + assert!(server.should_backpressure()); } #[test] From b2b0602c2ba5e24fddced57099b4fd577825f1e8 Mon Sep 17 00:00:00 2001 From: Rui Carmo Date: Sat, 10 Oct 2026 13:04:22 +0100 Subject: [PATCH 2/5] fix(egfx): avoid panicking surface lookups in typed submission --- crates/ironrdp-egfx/src/server.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/crates/ironrdp-egfx/src/server.rs b/crates/ironrdp-egfx/src/server.rs index 60cddf57e9..3c7e2bb27f 100644 --- a/crates/ironrdp-egfx/src/server.rs +++ b/crates/ironrdp-egfx/src/server.rs @@ -55,8 +55,8 @@ //! let server = GraphicsPipelineServer::new(Box::new(MyHandler)); //! ``` +use core::num::NonZeroU32; use std::collections::{HashMap, VecDeque}; -use std::num::NonZeroU32; use std::time::Instant; use ironrdp_core::{Encode, EncodeResult, WriteCursor, decode, impl_as_any}; @@ -1646,7 +1646,10 @@ impl GraphicsPipelineServer { return Err(error); } - let surface = self.surfaces.get(surface_id).expect("surface existence checked above"); + let surface = self + .surfaces + .get(surface_id) + .ok_or(FrameSubmissionError::UnknownSurface { surface_id })?; let timestamp = Self::make_timestamp(timestamp_ms); let frame_id = self.frames.begin_frame(timestamp); let encoded_stream = encode_avc420_bitmap_stream(regions, h264_data); @@ -1842,7 +1845,10 @@ impl GraphicsPipelineServer { return Err(error); } - let surface = self.surfaces.get(surface_id).expect("surface existence checked above"); + let surface = self + .surfaces + .get(surface_id) + .ok_or(FrameSubmissionError::UnknownSurface { surface_id })?; let timestamp = Self::make_timestamp(timestamp_ms); let frame_id = self.frames.begin_frame(timestamp); let wire_pdu = Self::avc444_wire_pdu(codec_id, surface, encoding, (stream1_regions, stream1_data), stream2); From ebc27e173fa45be6a585981607c28445b276f261 Mon Sep 17 00:00:00 2001 From: Rui Carmo Date: Sat, 10 Oct 2026 14:16:10 +0100 Subject: [PATCH 3/5] fix(egfx): distinguish ACK suspension from resume eligibility --- crates/ironrdp-egfx/src/client.rs | 7 - crates/ironrdp-egfx/src/server.rs | 155 ++++++++++++++---- crates/ironrdp-server/src/server.rs | 4 +- .../tests/egfx/server.rs | 70 +++++++- 4 files changed, 190 insertions(+), 46 deletions(-) diff --git a/crates/ironrdp-egfx/src/client.rs b/crates/ironrdp-egfx/src/client.rs index 5282b72274..35d9f03d35 100644 --- a/crates/ironrdp-egfx/src/client.rs +++ b/crates/ironrdp-egfx/src/client.rs @@ -419,7 +419,6 @@ pub struct GraphicsPipelineClient { surfaces: BTreeMap, compositor: Compositor, current_frame_id: Option, - frames_queued: u32, total_frames_decoded: u32, pending_output_reset: Option<(u16, u16)>, } @@ -446,7 +445,6 @@ impl GraphicsPipelineClient { surfaces: BTreeMap::new(), compositor: Compositor::default(), current_frame_id: None, - frames_queued: 0, total_frames_decoded: 0, pending_output_reset: None, } @@ -535,7 +533,6 @@ impl GraphicsPipelineClient { } GfxPdu::StartFrame(start) => { self.current_frame_id = Some(start.frame_id); - self.frames_queued = self.frames_queued.saturating_add(1); self.progressive_decoder.begin_frame(); trace!(frame_id = start.frame_id, "StartFrame"); Ok(vec![]) @@ -704,7 +701,6 @@ impl GraphicsPipelineClient { // per spec, capabilities are negotiated via CapabilitiesConfirm before // ResetGraphics, and a ResetGraphics does not re-negotiate capabilities. self.current_frame_id = None; - self.frames_queued = 0; // Reset decoder state for new stream if let Some(ref mut decoder) = self.h264_decoder { @@ -1191,7 +1187,6 @@ impl GraphicsPipelineClient { fn handle_end_frame(&mut self, frame_id: u32) -> PduResult> { self.total_frames_decoded = self.total_frames_decoded.wrapping_add(1); self.current_frame_id = None; - self.frames_queued = self.frames_queued.saturating_sub(1); self.progressive_decoder.end_frame(); @@ -1819,7 +1814,6 @@ mod tests { frame_id: 42, })); assert!(client.current_frame_id.is_some()); - assert_eq!(client.frames_queued, 1); let _ = client.handle_pdu(GfxPdu::ResetGraphics(crate::pdu::ResetGraphicsPdu { width: 1920, @@ -1829,7 +1823,6 @@ mod tests { assert!(client.surfaces.is_empty(), "surfaces should be cleared"); assert!(client.current_frame_id.is_none(), "frame_id should be reset"); - assert_eq!(client.frames_queued, 0, "frame queue should be reset"); } #[test] diff --git a/crates/ironrdp-egfx/src/server.rs b/crates/ironrdp-egfx/src/server.rs index 3c7e2bb27f..f748bffa8e 100644 --- a/crates/ironrdp-egfx/src/server.rs +++ b/crates/ironrdp-egfx/src/server.rs @@ -56,7 +56,7 @@ //! ``` use core::num::NonZeroU32; -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::time::Instant; use ironrdp_core::{Encode, EncodeResult, WriteCursor, decode, impl_as_any}; @@ -512,8 +512,12 @@ pub struct FrameTracker { ack_suspended: bool, /// Next frame ID to assign next_frame_id: u32, - /// First frame ID issued in the current channel generation. + /// First frame ID issued in this generation, used to validate suspend ACKs. generation_first_frame_id: u32, + /// Frames still outstanding when acknowledgements were suspended. + suspension_outstanding: HashSet, + /// First frame ID issued during the current suspension. + suspension_first_frame_id: Option, /// Operator-configured maximum frames in flight (zero means unlimited). configured_max_in_flight: u32, /// Optional non-zero ceiling advertised by the active client. @@ -546,6 +550,8 @@ impl FrameTracker { ack_suspended: false, next_frame_id: 0, generation_first_frame_id: 0, + suspension_outstanding: HashSet::new(), + suspension_first_frame_id: None, configured_max_in_flight: DEFAULT_MAX_FRAMES_IN_FLIGHT, client_max_in_flight: None, total_sent: 0, @@ -666,20 +672,13 @@ impl FrameTracker { let info = self.unacknowledged.remove(&frame_id)?; let suspending = queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH; if suspending { - self.ack_suspended = true; - self.client_queue_depth = 0; + self.suspend_acknowledgements(); } else { self.ack_suspended = false; self.client_queue_depth = queue_depth; } self.total_acked += 1; - if suspending { - // [MS-RDPEGFX] 3.2.5.13: a client that suspends acknowledgements - // makes the remaining unacknowledged frame state unavailable. - self.unacknowledged.clear(); - } - // Edge-trigger after the remove/clear so an ack that releases // backpressure logs. self.emit_state_transitions(); @@ -722,25 +721,55 @@ impl FrameTracker { self.total_acked } - /// Resume tracking only for a non-suspend ACK naming a frame issued in this generation. - /// Frames sent during suspension are intentionally absent from the tracked map. - fn resume_from_untracked_ack(&mut self, frame_id: u32, queue_depth: u32) -> bool { + fn suspend_acknowledgements(&mut self) { + if !self.ack_suspended { + self.suspension_outstanding = self.unacknowledged.keys().copied().collect(); + self.suspension_first_frame_id = Some(self.next_frame_id); + } + self.ack_suspended = true; + self.client_queue_depth = 0; + self.unacknowledged.clear(); + } + + /// A valid suspend request is independent of whether an RTT sample is tracked. + fn suspend_from_untracked_ack(&mut self, frame_id: u32) -> bool { let issued = frame_id.wrapping_sub(self.generation_first_frame_id) < self.next_frame_id.wrapping_sub(self.generation_first_frame_id); - if !self.ack_suspended || queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH || !issued { + if !issued { + return false; + } + self.suspend_acknowledgements(); + self.emit_state_transitions(); + true + } + + /// Resume only for a frame outstanding at suspension or issued during it. + /// Already-acknowledged frames cannot change the current flow-control state. + fn resume_from_untracked_ack(&mut self, frame_id: u32, queue_depth: u32) -> bool { + if !self.ack_suspended || queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH { return false; } + let issued_during_suspension = self + .suspension_first_frame_id + .is_some_and(|first| frame_id.wrapping_sub(first) < self.next_frame_id.wrapping_sub(first)); + if !self.suspension_outstanding.contains(&frame_id) && !issued_during_suspension { + return false; + } + self.suspension_outstanding.clear(); + self.suspension_first_frame_id = None; self.ack_suspended = false; self.client_queue_depth = queue_depth; self.emit_state_transitions(); true } - /// Clear all per-channel tracking state while preserving operator config. + /// Clear per-channel tracking while preserving operator config and the connection ceiling. pub fn clear(&mut self) { self.unacknowledged.clear(); self.client_queue_depth = 0; self.ack_suspended = false; + self.suspension_outstanding.clear(); + self.suspension_first_frame_id = None; self.generation_first_frame_id = self.next_frame_id; self.last_backpressure_state = false; self.last_ack_suspended_state = false; @@ -1234,12 +1263,10 @@ impl GraphicsPipelineServer { /// Reset all state owned by one DVC channel generation. fn reset_channel_state(&mut self) { - self.state = ServerState::WaitingForCapabilities; self.negotiated_caps = None; self.codec_caps = CodecCapabilities::default(); self.surfaces = Surfaces::new(); self.frames.clear(); - self.frames.client_max_in_flight = None; self.output_width = 0; self.output_height = 0; self.reset_graphics_sent = false; @@ -1519,6 +1546,15 @@ impl GraphicsPipelineServer { self.frames.clamp_max_in_flight(client_max) } + /// Replace the current connection's core Frame Acknowledge ceiling. + /// + /// The ceiling survives DVC close/reopen and resize. Apply this for every + /// new connection; `None` clears a previous connection's constraint. + pub fn set_client_frame_ack_limit(&mut self, client_max: Option) -> u32 { + self.frames.client_max_in_flight = client_max; + self.frames.max_in_flight() + } + /// Effective maximum number of frames in flight. #[must_use] pub fn max_frames_in_flight(&self) -> u32 { @@ -1639,12 +1675,8 @@ impl GraphicsPipelineServer { regions: &[Avc420Region], timestamp_ms: u32, ) -> Result { - if let Err(error) = self.avc420_submission_state(surface_id) { - if matches!(error, FrameSubmissionError::Backpressured { .. }) { - self.qoe.record_backpressure(); - } - return Err(error); - } + let state = self.avc420_submission_state(surface_id); + self.record_submission_state(state)?; let surface = self .surfaces @@ -1670,6 +1702,13 @@ impl GraphicsPipelineServer { Ok(frame_id) } + fn record_submission_state(&mut self, state: Result<(), FrameSubmissionError>) -> Result<(), FrameSubmissionError> { + if matches!(state, Err(FrameSubmissionError::Backpressured { .. })) { + self.qoe.record_backpressure(); + } + state + } + /// Compatibility wrapper returning `None` for any submission rejection. pub fn send_avc420_frame( &mut self, @@ -1838,12 +1877,8 @@ impl GraphicsPipelineServer { if !Self::avc444_stream_shape_is_valid(encoding, stream2.is_some()) { return Err(FrameSubmissionError::InvalidFrame); } - if let Err(error) = self.avc444_submission_state(surface_id) { - if matches!(error, FrameSubmissionError::Backpressured { .. }) { - self.qoe.record_backpressure(); - } - return Err(error); - } + let state = self.avc444_submission_state(surface_id); + self.record_submission_state(state)?; let surface = self .surfaces @@ -2517,23 +2552,28 @@ impl GraphicsPipelineServer { } fn handle_frame_acknowledge(&mut self, pdu: FrameAcknowledgePdu) { - if !self.is_ready() { - warn!( + let queue_depth = pdu.queue_depth.to_u32(); + let suspended = queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH; + if !self.is_ready() && !suspended { + debug!( frame_id = pdu.frame_id, "Ignoring frame acknowledgement outside active EGFX generation" ); return; } - let queue_depth = pdu.queue_depth.to_u32(); - let suspended = queue_depth == SUSPEND_FRAME_ACK_QUEUE_DEPTH; let Some(info) = self.frames.acknowledge(pdu.frame_id, queue_depth) else { - if self.frames.resume_from_untracked_ack(pdu.frame_id, queue_depth) { + let accepted = if suspended { + self.frames.suspend_from_untracked_ack(pdu.frame_id) + } else { + self.frames.resume_from_untracked_ack(pdu.frame_id, queue_depth) + }; + if accepted { self.handler .on_frame_ack(pdu.frame_id, queue_depth, pdu.total_frames_decoded); return; } - warn!( + debug!( frame_id = pdu.frame_id, queue_depth, "Ignoring unknown or stale frame acknowledgement" ); @@ -2833,6 +2873,51 @@ mod capability_negotiation_tests { mod tests { use super::*; + #[test] + fn valid_suspend_ack_is_honoured_without_a_tracked_sample_or_ready_state() { + struct Handler; + impl GraphicsPipelineHandler for Handler { + fn capabilities_advertise(&mut self, _: &CapabilitiesAdvertisePdu) {} + fn on_ready(&mut self, _: &CapabilitySet) {} + } + let mut server = GraphicsPipelineServer::new(Box::new(Handler)); + let frame_id = server.frames.begin_frame(GraphicsPipelineServer::make_timestamp(0)); + server.frames.acknowledge(frame_id, 0).unwrap(); + server.state = ServerState::Resizing; + server.handle_frame_acknowledge(FrameAcknowledgePdu { + frame_id, + queue_depth: crate::pdu::QueueDepth::Suspend, + total_frames_decoded: 1, + }); + assert!(server.frames.is_ack_suspended()); + assert_eq!(server.frames.in_flight(), 0); + assert!(!server.frames.should_backpressure()); + assert_eq!( + server.frames.total_acked(), + 1, + "suspension must not fabricate another RTT/ACK sample" + ); + } + + #[test] + fn suspension_resume_eligibility_handles_wrapping_ids_and_repeated_suspend() { + let mut tracker = FrameTracker::new(); + tracker.next_frame_id = u32::MAX; + tracker.generation_first_frame_id = u32::MAX; + let suspended_frame = tracker.begin_frame(GraphicsPipelineServer::make_timestamp(0)); + tracker + .acknowledge(suspended_frame, SUSPEND_FRAME_ACK_QUEUE_DEPTH) + .unwrap(); + let during_pause = tracker.begin_frame(GraphicsPipelineServer::make_timestamp(0)); + assert_eq!(during_pause, 0); + assert!(tracker.suspend_from_untracked_ack(suspended_frame)); + assert!(tracker.resume_from_untracked_ack(during_pause, 0)); + assert!(!tracker.is_ack_suspended()); + assert!(!tracker.resume_from_untracked_ack(during_pause, 99)); + tracker.clear(); + assert!(!tracker.suspend_from_untracked_ack(suspended_frame)); + } + struct DefaultsHandler; impl GraphicsPipelineHandler for DefaultsHandler { diff --git a/crates/ironrdp-server/src/server.rs b/crates/ironrdp-server/src/server.rs index 1172a4ac76..e1cc52f5b3 100644 --- a/crates/ironrdp-server/src/server.rs +++ b/crates/ironrdp-server/src/server.rs @@ -4687,13 +4687,13 @@ impl RdpServer { } #[cfg(feature = "egfx")] - if let Some(client_limit) = client_core_frame_ack_limit(&result.capabilities) + if !result.reactivation && let Some(gfx_handle) = self.gfx_handle.as_ref() { let mut server = gfx_handle .lock() .map_err(|_| ServerError::reason("graphics server", "GfxServerHandle mutex poisoned"))?; - server.clamp_max_frames_in_flight(client_limit); + server.set_client_frame_ack_limit(client_core_frame_ack_limit(&result.capabilities)); } conn.static_channels = result.static_channels; conn.client_supports_heartbeat = result diff --git a/crates/ironrdp-testsuite-core/tests/egfx/server.rs b/crates/ironrdp-testsuite-core/tests/egfx/server.rs index cfed2f3f49..4f753cb50e 100644 --- a/crates/ironrdp-testsuite-core/tests/egfx/server.rs +++ b/crates/ironrdp-testsuite-core/tests/egfx/server.rs @@ -1087,6 +1087,19 @@ fn test_zero_configured_window_is_unlimited_until_client_clamp() { ); } +#[test] +fn new_connection_replaces_the_previous_client_ceiling() { + let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + server.set_max_frames_in_flight(3); + assert_eq!(server.set_client_frame_ack_limit(core::num::NonZeroU32::new(1)), 1); + server.close(7); + assert_eq!(server.set_client_frame_ack_limit(core::num::NonZeroU32::new(2)), 2); + assert_eq!(server.set_client_frame_ack_limit(None), 3); + server.set_max_frames_in_flight(0); + assert_eq!(server.set_client_frame_ack_limit(core::num::NonZeroU32::new(1)), 1); + assert_eq!(server.set_client_frame_ack_limit(None), 0); +} + #[test] fn test_close_reopen_clears_stale_frame_window_and_output() { let handler = Box::new(TestHandler::new()); @@ -1115,10 +1128,11 @@ fn test_close_reopen_clears_stale_frame_window_and_output() { server.close(7); assert_eq!(server.frames_in_flight(), 0); assert!(!server.has_pending_output()); - assert_eq!(server.max_frames_in_flight(), 3); + assert_eq!(server.max_frames_in_flight(), 1); server.start(8).expect("reopened DVC start"); - server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")); + // No new core capability exchange occurs when only the DVC reopens. + assert_eq!(server.max_frames_in_flight(), 1); server .process(8, &encode_pdu(&client_caps_pdu)) .expect("reopened capabilities"); @@ -1142,6 +1156,58 @@ fn test_close_reopen_clears_stale_frame_window_and_output() { ); } +#[test] +fn acknowledged_duplicates_cannot_resume_a_later_suspension() { + let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + let caps = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V8_1 { + flags: CapabilitiesV81Flags::AVC420_ENABLED, + }])); + server.process(0, &encode_pdu(&caps)).unwrap(); + let surface = server.create_surface(64, 64).unwrap(); + let regions = [Avc420Region::full_frame(64, 64, 22)]; + let ack = |frame_id, queue_depth| { + GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { + frame_id, + queue_depth, + total_frames_decoded: 1, + }) + }; + let outstanding = server.send_avc420_frame(surface, &[1], ®ions, 1).unwrap(); + let already_acked = server.send_avc420_frame(surface, &[2], ®ions, 2).unwrap(); + let suspending = server.send_avc420_frame(surface, &[3], ®ions, 3).unwrap(); + server + .process(0, &encode_pdu(&ack(already_acked, QueueDepth::AvailableBytes(12)))) + .unwrap(); + server + .process(0, &encode_pdu(&ack(suspending, QueueDepth::Suspend))) + .unwrap(); + // This duplicate is newer than an outstanding frame, so a broad ID range is insufficient. + server + .process(0, &encode_pdu(&ack(already_acked, QueueDepth::AvailableBytes(99)))) + .unwrap(); + server + .process(0, &encode_pdu(&ack(suspending, QueueDepth::AvailableBytes(99)))) + .unwrap(); + assert_eq!(server.client_queue_depth(), 0); + server.send_avc420_frame(surface, &[4], ®ions, 4).unwrap(); + assert_eq!(server.frames_in_flight(), 0); + // A frame outstanding at suspension is still allowed to resume. + server + .process(0, &encode_pdu(&ack(outstanding, QueueDepth::AvailableBytes(8)))) + .unwrap(); + assert_eq!(server.client_queue_depth(), 8); + let next = server.send_avc420_frame(surface, &[5], ®ions, 5).unwrap(); + assert_eq!(server.frames_in_flight(), 1); + server.process(0, &encode_pdu(&ack(next, QueueDepth::Suspend))).unwrap(); + // Eligibility was cleared on the previous resume; it must not carry into this pause. + server + .process(0, &encode_pdu(&ack(outstanding, QueueDepth::AvailableBytes(99)))) + .unwrap(); + assert_eq!(server.client_queue_depth(), 0); + server.send_avc420_frame(surface, &[6], ®ions, 6).unwrap(); + assert_eq!(server.frames_in_flight(), 0); +} + #[test] fn closed_generation_cannot_resume_or_suspend_current_tracking() { let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); From 53cfcf9a02ba5fb1fa7fbd58f8594355c9cf9931 Mon Sep 17 00:00:00 2001 From: Rui Carmo Date: Sat, 10 Oct 2026 15:40:37 +0100 Subject: [PATCH 4/5] refactor(egfx): share AVC submission gates and refresh negotiated limits --- crates/ironrdp-egfx/src/server.rs | 93 +++++++------------ crates/ironrdp-server/src/gfx.rs | 2 +- crates/ironrdp-server/src/server.rs | 4 +- .../tests/egfx/server.rs | 19 +++- 4 files changed, 48 insertions(+), 70 deletions(-) diff --git a/crates/ironrdp-egfx/src/server.rs b/crates/ironrdp-egfx/src/server.rs index f748bffa8e..1abe16d72c 100644 --- a/crates/ironrdp-egfx/src/server.rs +++ b/crates/ironrdp-egfx/src/server.rs @@ -613,12 +613,6 @@ impl FrameTracker { self.configured_max_in_flight = max; } - /// Store a non-zero client-advertised ceiling and return the effective limit. - fn clamp_max_in_flight(&mut self, client_max: NonZeroU32) -> u32 { - self.client_max_in_flight = Some(client_max); - self.max_in_flight() - } - /// Effective maximum number of frames permitted in flight. /// /// Zero means unlimited only when no client ceiling is present. @@ -1538,18 +1532,10 @@ impl GraphicsPipelineServer { self.frames.set_max_in_flight(max); } - /// Clamp the configured frame window to a non-zero client-advertised limit. - /// - /// A configured value of zero remains unlimited only until the client - /// advertises a non-zero ceiling. Returns the effective limit. - pub fn clamp_max_frames_in_flight(&mut self, client_max: NonZeroU32) -> u32 { - self.frames.clamp_max_in_flight(client_max) - } - /// Replace the current connection's core Frame Acknowledge ceiling. /// /// The ceiling survives DVC close/reopen and resize. Apply this for every - /// new connection; `None` clears a previous connection's constraint. + /// capability exchange, including reactivation; `None` clears the constraint. pub fn set_client_frame_ack_limit(&mut self, client_max: Option) -> u32 { self.frames.client_max_in_flight = client_max; self.frames.max_in_flight() @@ -1647,24 +1633,41 @@ impl GraphicsPipelineServer { rect.left < rect.right && rect.top < rect.bottom && rect.right <= surface.width && rect.bottom <= surface.height } - /// Check whether an AVC420 frame can be submitted without encoding it. - pub fn avc420_submission_state(&self, surface_id: u16) -> Result<(), FrameSubmissionError> { + fn submission_state(&self, surface_id: u16, avc444: bool) -> Result<&Surface, FrameSubmissionError> { if !self.is_ready() { return Err(FrameSubmissionError::NotReady); } - if !self.supports_avc420() { + if avc444 && !self.supports_avc444() { + return Err(FrameSubmissionError::Avc444Unsupported); + } + if !avc444 && !self.supports_avc420() { return Err(FrameSubmissionError::Avc420Unsupported); } if self.should_backpressure() { return Err(FrameSubmissionError::Backpressured { - frames_in_flight: self.frames_in_flight(), - max_frames_in_flight: self.max_frames_in_flight(), + frames_in_flight: self.frames.in_flight(), + max_frames_in_flight: self.frames.max_in_flight(), }); } - if self.surfaces.get(surface_id).is_none() { - return Err(FrameSubmissionError::UnknownSurface { surface_id }); + self.surfaces + .get(surface_id) + .ok_or(FrameSubmissionError::UnknownSurface { surface_id }) + } + + /// Inspect AVC420 readiness without queueing a frame. + pub fn avc420_submission_state(&self, surface_id: u16) -> Result<(), FrameSubmissionError> { + self.submission_state(surface_id, false).map(|_| ()) + } + + fn prepare_submission(&mut self, surface_id: u16, avc444: bool) -> Result { + // Copy the small descriptor so validation and QoE accounting use one lookup. + match self.submission_state(surface_id, avc444).cloned() { + Err(error @ FrameSubmissionError::Backpressured { .. }) => { + self.qoe.record_backpressure(); + Err(error) + } + result => result, } - Ok(()) } /// Queue an H.264 AVC420 frame for transmission with a typed rejection. @@ -1675,13 +1678,7 @@ impl GraphicsPipelineServer { regions: &[Avc420Region], timestamp_ms: u32, ) -> Result { - let state = self.avc420_submission_state(surface_id); - self.record_submission_state(state)?; - - let surface = self - .surfaces - .get(surface_id) - .ok_or(FrameSubmissionError::UnknownSurface { surface_id })?; + let surface = self.prepare_submission(surface_id, false)?; let timestamp = Self::make_timestamp(timestamp_ms); let frame_id = self.frames.begin_frame(timestamp); let encoded_stream = encode_avc420_bitmap_stream(regions, h264_data); @@ -1702,13 +1699,6 @@ impl GraphicsPipelineServer { Ok(frame_id) } - fn record_submission_state(&mut self, state: Result<(), FrameSubmissionError>) -> Result<(), FrameSubmissionError> { - if matches!(state, Err(FrameSubmissionError::Backpressured { .. })) { - self.qoe.record_backpressure(); - } - state - } - /// Compatibility wrapper returning `None` for any submission rejection. pub fn send_avc420_frame( &mut self, @@ -1721,24 +1711,9 @@ impl GraphicsPipelineServer { .ok() } - /// Check whether an AVC444 frame can be submitted without encoding it. + /// Inspect AVC444 readiness without queueing a frame. pub fn avc444_submission_state(&self, surface_id: u16) -> Result<(), FrameSubmissionError> { - if !self.is_ready() { - return Err(FrameSubmissionError::NotReady); - } - if !self.supports_avc444() { - return Err(FrameSubmissionError::Avc444Unsupported); - } - if self.should_backpressure() { - return Err(FrameSubmissionError::Backpressured { - frames_in_flight: self.frames_in_flight(), - max_frames_in_flight: self.max_frames_in_flight(), - }); - } - if self.surfaces.get(surface_id).is_none() { - return Err(FrameSubmissionError::UnknownSurface { surface_id }); - } - Ok(()) + self.submission_state(surface_id, true).map(|_| ()) } /// Queue an H.264 AVC444 frame for transmission with a typed rejection. @@ -1877,16 +1852,10 @@ impl GraphicsPipelineServer { if !Self::avc444_stream_shape_is_valid(encoding, stream2.is_some()) { return Err(FrameSubmissionError::InvalidFrame); } - let state = self.avc444_submission_state(surface_id); - self.record_submission_state(state)?; - - let surface = self - .surfaces - .get(surface_id) - .ok_or(FrameSubmissionError::UnknownSurface { surface_id })?; + let surface = self.prepare_submission(surface_id, true)?; let timestamp = Self::make_timestamp(timestamp_ms); let frame_id = self.frames.begin_frame(timestamp); - let wire_pdu = Self::avc444_wire_pdu(codec_id, surface, encoding, (stream1_regions, stream1_data), stream2); + let wire_pdu = Self::avc444_wire_pdu(codec_id, &surface, encoding, (stream1_regions, stream1_data), stream2); self.output_queue .push_back(GfxPdu::StartFrame(StartFramePdu { timestamp, frame_id })); diff --git a/crates/ironrdp-server/src/gfx.rs b/crates/ironrdp-server/src/gfx.rs index 71a9c5a3ff..035038a96d 100644 --- a/crates/ironrdp-server/src/gfx.rs +++ b/crates/ironrdp-server/src/gfx.rs @@ -159,7 +159,7 @@ mod tests { let mut server = handle.lock().expect("GfxServerHandle mutex poisoned"); server.set_max_frames_in_flight(3); assert_eq!( - server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")), + server.set_client_frame_ack_limit(Some(core::num::NonZeroU32::new(1).expect("one is non-zero"))), 1 ); } diff --git a/crates/ironrdp-server/src/server.rs b/crates/ironrdp-server/src/server.rs index e1cc52f5b3..bd88c63b7e 100644 --- a/crates/ironrdp-server/src/server.rs +++ b/crates/ironrdp-server/src/server.rs @@ -4687,9 +4687,7 @@ impl RdpServer { } #[cfg(feature = "egfx")] - if !result.reactivation - && let Some(gfx_handle) = self.gfx_handle.as_ref() - { + if let Some(gfx_handle) = self.gfx_handle.as_ref() { let mut server = gfx_handle .lock() .map_err(|_| ServerError::reason("graphics server", "GfxServerHandle mutex poisoned"))?; diff --git a/crates/ironrdp-testsuite-core/tests/egfx/server.rs b/crates/ironrdp-testsuite-core/tests/egfx/server.rs index 4f753cb50e..7a3e8c4c8a 100644 --- a/crates/ironrdp-testsuite-core/tests/egfx/server.rs +++ b/crates/ironrdp-testsuite-core/tests/egfx/server.rs @@ -1007,7 +1007,7 @@ fn test_frame_flow_control() { server.set_max_frames_in_flight(3); assert_eq!(server.max_frames_in_flight(), 3); assert_eq!( - server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")), + server.set_client_frame_ack_limit(Some(core::num::NonZeroU32::new(1).expect("one is non-zero"))), 1 ); @@ -1062,7 +1062,7 @@ fn test_frame_flow_control() { fn resize_retains_client_frame_ack_ceiling() { let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); server.set_max_frames_in_flight(3); - server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).unwrap()); + server.set_client_frame_ack_limit(Some(core::num::NonZeroU32::new(1).unwrap())); server.resize(64, 64); assert_eq!(server.max_frames_in_flight(), 1); } @@ -1076,7 +1076,7 @@ fn test_zero_configured_window_is_unlimited_until_client_clamp() { assert!(!server.should_backpressure()); assert_eq!( - server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")), + server.set_client_frame_ack_limit(Some(core::num::NonZeroU32::new(1).expect("one is non-zero"))), 1 ); server.set_max_frames_in_flight(3); @@ -1087,6 +1087,17 @@ fn test_zero_configured_window_is_unlimited_until_client_clamp() { ); } +#[test] +fn negotiated_ceiling_can_be_refreshed_after_resize() { + let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + server.set_max_frames_in_flight(3); + server.set_client_frame_ack_limit(core::num::NonZeroU32::new(1)); + server.resize(64, 64); + assert_eq!(server.max_frames_in_flight(), 1); + assert_eq!(server.set_client_frame_ack_limit(core::num::NonZeroU32::new(2)), 2); + assert_eq!(server.set_client_frame_ack_limit(None), 3); +} + #[test] fn new_connection_replaces_the_previous_client_ceiling() { let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); @@ -1106,7 +1117,7 @@ fn test_close_reopen_clears_stale_frame_window_and_output() { let mut server = GraphicsPipelineServer::new(handler); server.set_max_frames_in_flight(3); server.start(7).expect("initial DVC start"); - server.clamp_max_frames_in_flight(core::num::NonZeroU32::new(1).expect("one is non-zero")); + server.set_client_frame_ack_limit(Some(core::num::NonZeroU32::new(1).expect("one is non-zero"))); let client_caps_pdu = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V8_1 { flags: CapabilitiesV81Flags::AVC420_ENABLED, From 209db33b6192c3269b1e7bd753c974b11f76cdfe Mon Sep 17 00:00:00 2001 From: Rui Carmo Date: Sat, 10 Oct 2026 16:24:16 +0100 Subject: [PATCH 5/5] fix(egfx): retain suspend ACK validity across graphics reset --- crates/ironrdp-egfx/src/server.rs | 23 ++-- .../tests/egfx/server.rs | 111 +++++++++++------- 2 files changed, 80 insertions(+), 54 deletions(-) diff --git a/crates/ironrdp-egfx/src/server.rs b/crates/ironrdp-egfx/src/server.rs index 1abe16d72c..1b95579a44 100644 --- a/crates/ironrdp-egfx/src/server.rs +++ b/crates/ironrdp-egfx/src/server.rs @@ -757,17 +757,23 @@ impl FrameTracker { true } - /// Clear per-channel tracking while preserving operator config and the connection ceiling. - pub fn clear(&mut self) { + /// Discard pending frames for a graphics reset without ending the DVC generation. + /// Previously issued frames can still be named by delayed suspend ACKs. + fn clear_pending_frames(&mut self) { self.unacknowledged.clear(); self.client_queue_depth = 0; self.ack_suspended = false; self.suspension_outstanding.clear(); self.suspension_first_frame_id = None; - self.generation_first_frame_id = self.next_frame_id; self.last_backpressure_state = false; self.last_ack_suspended_state = false; } + + /// Clear per-channel tracking while preserving operator config and the connection ceiling. + pub fn clear(&mut self) { + self.clear_pending_frames(); + self.generation_first_frame_id = self.next_frame_id; + } } /// Typed reason why an encoded frame could not be submitted to EGFX. @@ -1490,7 +1496,7 @@ impl GraphicsPipelineServer { self.delete_surface(id); } - self.frames.clear(); + self.frames.clear_pending_frames(); self.output_queue.push_back(GfxPdu::ResetGraphics(ResetGraphicsPdu { width: u32::from(width), @@ -2471,7 +2477,7 @@ impl GraphicsPipelineServer { state for re-initialization (no DeleteSurface PDU emitted, surface ID counter reset)" ); self.surfaces.reset_for_reinit(); - self.frames.clear(); + self.frames.clear_pending_frames(); self.reset_graphics_sent = false; } @@ -2844,12 +2850,7 @@ mod tests { #[test] fn valid_suspend_ack_is_honoured_without_a_tracked_sample_or_ready_state() { - struct Handler; - impl GraphicsPipelineHandler for Handler { - fn capabilities_advertise(&mut self, _: &CapabilitiesAdvertisePdu) {} - fn on_ready(&mut self, _: &CapabilitySet) {} - } - let mut server = GraphicsPipelineServer::new(Box::new(Handler)); + let mut server = GraphicsPipelineServer::new(Box::new(DefaultsHandler)); let frame_id = server.frames.begin_frame(GraphicsPipelineServer::make_timestamp(0)); server.frames.acknowledge(frame_id, 0).unwrap(); server.state = ServerState::Resizing; diff --git a/crates/ironrdp-testsuite-core/tests/egfx/server.rs b/crates/ironrdp-testsuite-core/tests/egfx/server.rs index 7a3e8c4c8a..06a39a96e4 100644 --- a/crates/ironrdp-testsuite-core/tests/egfx/server.rs +++ b/crates/ironrdp-testsuite-core/tests/egfx/server.rs @@ -76,6 +76,14 @@ fn encode_pdu(pdu: &T) -> Vec { // Tests // ============================================================================ +fn ack_pdu(frame_id: u32, queue_depth: QueueDepth) -> GfxPdu { + GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { + frame_id, + queue_depth, + total_frames_decoded: 1, + }) +} + #[test] fn test_server_creation() { let handler = Box::new(TestHandler::new()); @@ -822,6 +830,7 @@ fn mixed_frame_of_avc444_tiles_alone_is_allowed_before_10_4() { #[test] fn avc444v2_submission_preserves_typed_rejections() { let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + assert_eq!(server.avc444_submission_state(0), Err(FrameSubmissionError::NotReady)); let caps = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V10 { flags: CapabilitiesV10Flags::empty(), }])); @@ -835,6 +844,11 @@ fn avc444v2_submission_preserves_typed_rejections() { ); assert!(!server.has_pending_output()); assert_eq!(server.frames_in_flight(), 0); + assert_eq!(server.avc444_submission_state(surface), Ok(())); + assert_eq!( + server.avc444_submission_state(u16::MAX), + Err(FrameSubmissionError::UnknownSurface { surface_id: u16::MAX }) + ); server.set_max_frames_in_flight(1); assert!( server @@ -849,6 +863,13 @@ fn avc444v2_submission_preserves_typed_rejections() { max_frames_in_flight: 1 }) )); + assert_eq!( + server.avc444_submission_state(surface), + Err(FrameSubmissionError::Backpressured { + frames_in_flight: 1, + max_frames_in_flight: 1 + }) + ); assert!(!server.has_pending_output()); } @@ -1058,35 +1079,6 @@ fn test_frame_flow_control() { ); } -#[test] -fn resize_retains_client_frame_ack_ceiling() { - let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); - server.set_max_frames_in_flight(3); - server.set_client_frame_ack_limit(Some(core::num::NonZeroU32::new(1).unwrap())); - server.resize(64, 64); - assert_eq!(server.max_frames_in_flight(), 1); -} - -#[test] -fn test_zero_configured_window_is_unlimited_until_client_clamp() { - let handler = Box::new(TestHandler::new()); - let mut server = GraphicsPipelineServer::new(handler); - server.set_max_frames_in_flight(0); - assert_eq!(server.max_frames_in_flight(), 0); - assert!(!server.should_backpressure()); - - assert_eq!( - server.set_client_frame_ack_limit(Some(core::num::NonZeroU32::new(1).expect("one is non-zero"))), - 1 - ); - server.set_max_frames_in_flight(3); - assert_eq!( - server.max_frames_in_flight(), - 1, - "a later setter must not exceed the negotiated client ceiling" - ); -} - #[test] fn negotiated_ceiling_can_be_refreshed_after_resize() { let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); @@ -1176,13 +1168,7 @@ fn acknowledged_duplicates_cannot_resume_a_later_suspension() { server.process(0, &encode_pdu(&caps)).unwrap(); let surface = server.create_surface(64, 64).unwrap(); let regions = [Avc420Region::full_frame(64, 64, 22)]; - let ack = |frame_id, queue_depth| { - GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { - frame_id, - queue_depth, - total_frames_decoded: 1, - }) - }; + let ack = ack_pdu; let outstanding = server.send_avc420_frame(surface, &[1], ®ions, 1).unwrap(); let already_acked = server.send_avc420_frame(surface, &[2], ®ions, 2).unwrap(); let suspending = server.send_avc420_frame(surface, &[3], ®ions, 3).unwrap(); @@ -1219,6 +1205,51 @@ fn acknowledged_duplicates_cannot_resume_a_later_suspension() { assert_eq!(server.frames_in_flight(), 0); } +#[rstest] +#[case(false)] +#[case(true)] +fn delayed_suspend_ack_survives_graphics_reset(#[case] readvertise: bool) { + let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); + let caps = GfxPdu::CapabilitiesAdvertise(CapabilitiesAdvertisePdu::from_typed(&[CapabilitySet::V8_1 { + flags: CapabilitiesV81Flags::AVC420_ENABLED, + }])); + server.start(7).unwrap(); + server.process(7, &encode_pdu(&caps)).unwrap(); + let surface = server.create_surface(64, 64).unwrap(); + let regions = [Avc420Region::full_frame(64, 64, 22)]; + let before_reset = server.send_avc420_frame(surface, &[1], ®ions, 1).unwrap(); + server.set_max_frames_in_flight(1); + if readvertise { + server.process(7, &encode_pdu(&caps)).unwrap(); + } else { + server.resize(64, 64); + } + let surface = server.create_surface(64, 64).unwrap(); + let after_reset = server.send_avc420_frame(surface, &[2], ®ions, 2).unwrap(); + assert!(server.should_backpressure()); + let suspend = GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { + frame_id: before_reset, + queue_depth: QueueDepth::Suspend, + total_frames_decoded: 1, + }); + server.process(7, &encode_pdu(&suspend)).unwrap(); + assert_eq!(server.frames_in_flight(), 0); + assert!( + !server.should_backpressure(), + "graphics reset must not invalidate an issued suspend ACK" + ); + server.send_avc420_frame(surface, &[3], ®ions, 3).unwrap(); + let resume = GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { + frame_id: after_reset, + queue_depth: QueueDepth::Unavailable, + total_frames_decoded: 2, + }); + server.process(7, &encode_pdu(&resume)).unwrap(); + server.send_avc420_frame(surface, &[4], ®ions, 4).unwrap(); + assert_eq!(server.frames_in_flight(), 1); + assert!(server.should_backpressure()); +} + #[test] fn closed_generation_cannot_resume_or_suspend_current_tracking() { let mut server = GraphicsPipelineServer::new(Box::new(TestHandler::new())); @@ -1230,13 +1261,7 @@ fn closed_generation_cannot_resume_or_suspend_current_tracking() { let surface = server.create_surface(64, 64).unwrap(); let regions = [Avc420Region::full_frame(64, 64, 22)]; let old_frame = server.send_avc420_frame(surface, &[1], ®ions, 1).unwrap(); - let ack = |frame_id, queue_depth| { - GfxPdu::FrameAcknowledge(FrameAcknowledgePdu { - frame_id, - queue_depth, - total_frames_decoded: 1, - }) - }; + let ack = ack_pdu; server .process(7, &encode_pdu(&ack(old_frame, QueueDepth::Unavailable))) .unwrap();