feat(server): move the encoder to blocking task

Since it is a CPU intesive task, we should spawn a task to avoid
blocking other async tasks.

The UpdateFragmenterOwned is quite a gross hack to allow returning the result
from a the task lifetime. I don't know how to accomplish this better, or
else we have to add some Arc<Mutex> stuff.

Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
This commit is contained in:
Marc-André Lureau
2024-09-23 02:25:59 -04:00
committed by Benoît Cortier
parent b7164ecc68
commit 3c6b2ef2e2
2 changed files with 36 additions and 6 deletions
+22
View File
@@ -131,6 +131,14 @@ impl UpdateEncoder {
update(self, bitmap)
}
pub(crate) fn fragmenter_from_owned(&self, res: UpdateFragmenterOwned) -> UpdateFragmenter<'_> {
UpdateFragmenter {
code: res.code,
index: res.index,
data: &self.buffer[0..res.len],
}
}
fn bitmap_update(&mut self, bitmap: BitmapUpdate) -> Result<UpdateFragmenter<'_>> {
let len = loop {
match self.bitmap.encode(&bitmap, self.buffer.as_mut_slice()) {
@@ -208,6 +216,12 @@ impl UpdateEncoder {
}
}
pub(crate) struct UpdateFragmenterOwned {
code: UpdateCode,
index: usize,
len: usize,
}
pub(crate) struct UpdateFragmenter<'a> {
code: UpdateCode,
index: usize,
@@ -219,6 +233,14 @@ impl<'a> UpdateFragmenter<'a> {
Self { code, index: 0, data }
}
pub(crate) fn into_owned(self) -> UpdateFragmenterOwned {
UpdateFragmenterOwned {
code: self.code,
index: self.index,
len: self.data.len(),
}
}
pub(crate) fn size_hint(&self) -> usize {
FASTPATH_HEADER_SIZE + cmp::min(self.data.len(), MAX_FASTPATH_UPDATE_SIZE)
}
+14 -6
View File
@@ -376,13 +376,21 @@ impl RdpServer {
user_channel_id: u16,
io_channel_id: u16,
buffer: &mut Vec<u8>,
encoder: &mut UpdateEncoder,
) -> Result<RunState>
mut encoder: UpdateEncoder,
) -> Result<(RunState, UpdateEncoder)>
where
W: FramedWrite,
{
let mut fragmenter = match update {
DisplayUpdate::Bitmap(bitmap) => encoder.bitmap(bitmap),
DisplayUpdate::Bitmap(bitmap) => {
let (enc, res) = task::spawn_blocking(move || {
let res = encoder.bitmap(bitmap).map(|r| r.into_owned());
(encoder, res)
})
.await?;
encoder = enc;
res.map(|r| encoder.fragmenter_from_owned(r))
}
DisplayUpdate::PointerPosition(pos) => encoder.pointer_position(pos),
DisplayUpdate::Resize(desktop_size) => {
debug!(?desktop_size, "Display resize");
@@ -400,7 +408,7 @@ impl RdpServer {
};
let msg = encode_vec(&X224(pdu))?;
writer.write_all(&msg).await?;
return Ok(RunState::DeactivationReactivation { desktop_size });
return Ok((RunState::DeactivationReactivation { desktop_size }, encoder));
}
DisplayUpdate::RGBAPointer(ptr) => encoder.rgba_pointer(ptr),
DisplayUpdate::ColorPointer(ptr) => encoder.color_pointer(ptr),
@@ -420,7 +428,7 @@ impl RdpServer {
.context("failed to write display update")?;
}
Ok(RunState::Continue)
Ok((RunState::Continue, encoder))
}
async fn dispatch_server_events<W>(
@@ -526,7 +534,7 @@ impl RdpServer {
},
Some(update) = display_updates.next_update() => {
state = self.dispatch_display_update(update, writer, user_channel_id, io_channel_id, &mut buffer, &mut encoder).await?;
(state, encoder) = self.dispatch_display_update(update, writer, user_channel_id, io_channel_id, &mut buffer, encoder).await?;
}
nevents = self.ev_receiver.recv_many(&mut events, 100) => {