Skip to content

fix(session)!: always own a bulk decompressor for FastPath updates #1255

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Greg Lamberson (glamberson) wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(session)!: always own a bulk decompressor for FastPath updates #1255

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 3 additions & 3 deletions crates/ironrdp-bulk/benches/bulk_compression.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ fn bench_compress_decompress(c: &mut Criterion, ct: CompressionType, data: &[u8]
let name = algo_name(ct);

// Verify data actually compresses with this algorithm
let mut test_comp = BulkCompressor::new(ct).expect("bulk compressor should initialize");
let mut test_comp = BulkCompressor::new(ct);
let (test_size, test_flags) = test_comp.compress(data).expect("bulk compression should succeed");
let is_compressed = test_flags & flags::PACKET_COMPRESSED != 0;

Expand All @@ -85,7 +85,7 @@ fn bench_compress_decompress(c: &mut Criterion, ct: CompressionType, data: &[u8]

group.bench_function(BenchmarkId::new("compress", data.len()), |b| {
b.iter_batched(
|| BulkCompressor::new(ct).expect("bulk compressor should initialize"),
|| BulkCompressor::new(ct),
|mut compressor| {
black_box(
compressor
Expand All @@ -107,7 +107,7 @@ fn bench_compress_decompress(c: &mut Criterion, ct: CompressionType, data: &[u8]

group.bench_function(BenchmarkId::new("decompress", data.len()), |b| {
b.iter_batched(
|| BulkCompressor::new(ct).expect("bulk compressor should initialize"),
|| BulkCompressor::new(ct),
|mut decompressor| {
black_box(
decompressor
Expand Down
54 changes: 27 additions & 27 deletions crates/ironrdp-bulk/src/bulk.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,12 +72,12 @@ impl BulkCompressor {
/// - `Rdp6` (0x02): NCRUSH (Huffman-based)
/// - `Rdp61` (0x03): XCRUSH (two-level: chunk matching + MPPC)
///
pub fn new(compression_level: CompressionType) -> Result<Self, BulkError> {
pub fn new(compression_level: CompressionType) -> Self {
// MPPC contexts are created with level 1 by default and adjusted dynamically.
let mppc_send = MppcContext::new(1);
let mppc_recv = MppcContext::new(1);
let ncrush_send = NCrushContext::new()?;
let ncrush_recv = NCrushContext::new()?;
let ncrush_send = NCrushContext::new();
let ncrush_recv = NCrushContext::new();
let xcrush_send = XCrushContext::new();
let xcrush_recv = XCrushContext::new();

Expand All @@ -88,7 +88,7 @@ impl BulkCompressor {
v.into_boxed_slice().try_into().unwrap_or_else(|_| unreachable!())
};

Ok(Self {
Self {
compression_level,
mppc_send,
mppc_recv,
Expand All @@ -99,7 +99,7 @@ impl BulkCompressor {
output_buffer,
total_uncompressed_bytes: 0,
total_compressed_bytes: 0,
})
}
}

/// Returns the configured compression level.
Expand Down Expand Up @@ -271,25 +271,25 @@ mod tests {

#[test]
fn test_bulk_compressor_new_rdp4() {
let bulk = BulkCompressor::new(CompressionType::Rdp4).unwrap();
let bulk = BulkCompressor::new(CompressionType::Rdp4);
assert_eq!(bulk.compression_level(), CompressionType::Rdp4);
}

#[test]
fn test_bulk_compressor_new_rdp5() {
let bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let bulk = BulkCompressor::new(CompressionType::Rdp5);
assert_eq!(bulk.compression_level(), CompressionType::Rdp5);
}

#[test]
fn test_bulk_compressor_new_rdp6() {
let bulk = BulkCompressor::new(CompressionType::Rdp6).unwrap();
let bulk = BulkCompressor::new(CompressionType::Rdp6);
assert_eq!(bulk.compression_level(), CompressionType::Rdp6);
}

#[test]
fn test_bulk_compressor_new_rdp61() {
let bulk = BulkCompressor::new(CompressionType::Rdp61).unwrap();
let bulk = BulkCompressor::new(CompressionType::Rdp61);
assert_eq!(bulk.compression_level(), CompressionType::Rdp61);
}

Expand All @@ -314,14 +314,14 @@ mod tests {

#[test]
fn test_bulk_compressor_reset() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp61).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp61);
// Should not panic
bulk.reset();
}

#[test]
fn test_bulk_compressor_contexts_independent() {
let bulk = BulkCompressor::new(CompressionType::Rdp6).unwrap();
let bulk = BulkCompressor::new(CompressionType::Rdp6);
// Send and receive NCRUSH contexts should be separate instances
// (we can only verify they exist and the struct was created)
assert_eq!(bulk.compression_level(), CompressionType::Rdp6);
Expand All @@ -333,7 +333,7 @@ mod tests {

#[test]
fn test_bulk_compress_skip_small_input() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp5);
let data = b"tiny"; // 4 bytes, below threshold
let (size, flags) = bulk.compress(data).unwrap();
assert_eq!(size, data.len());
Expand All @@ -342,7 +342,7 @@ mod tests {

#[test]
fn test_bulk_compress_skip_empty() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp5);
let data = b"";
let (size, flags) = bulk.compress(data).unwrap();
assert_eq!(size, 0);
Expand All @@ -355,15 +355,15 @@ mod tests {

#[test]
fn test_bulk_decompress_no_flags() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp5);
let data = b"uncompressed data";
let result = bulk.decompress(data, 0x00).unwrap();
assert_eq!(result, data);
}

#[test]
fn test_bulk_decompress_unsupported_type() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp5);
// flags = PACKET_COMPRESSED | type 0x0F (invalid)
let result = bulk.decompress(b"data", 0x2F);
assert!(result.is_err());
Expand All @@ -376,8 +376,8 @@ mod tests {
/// Helper: compress with one BulkCompressor (sender) and decompress
/// with another (receiver). Returns the decompressed data as a Vec.
fn bulk_roundtrip(compression_level: CompressionType, input: &[u8]) -> Vec<u8> {
let mut sender = BulkCompressor::new(compression_level).unwrap();
let mut receiver = BulkCompressor::new(compression_level).unwrap();
let mut sender = BulkCompressor::new(compression_level);
let mut receiver = BulkCompressor::new(compression_level);

let (comp_size, flags) = sender.compress(input).unwrap();

Expand Down Expand Up @@ -451,7 +451,7 @@ mod tests {

#[test]
fn test_bulk_compress_rdp5_sets_type_bits() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp5);
let input = b"Some data that should compress with MPPC level 1 algorithm!!";
let (_size, flags) = bulk.compress(input).unwrap();

Expand All @@ -464,7 +464,7 @@ mod tests {

#[test]
fn test_bulk_compress_rdp6_sets_type_bits() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp6).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp6);
let input = b"for.whom.the.bell.tolls,.the.bell.tolls.for.thee!xx";
let (_size, flags) = bulk.compress(input).unwrap();

Expand All @@ -481,15 +481,15 @@ mod tests {

#[test]
fn test_metrics_start_at_zero() {
let bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let bulk = BulkCompressor::new(CompressionType::Rdp5);
assert_eq!(bulk.total_compressed_bytes(), 0);
assert_eq!(bulk.total_uncompressed_bytes(), 0);
assert!(bulk.compression_ratio().abs() < f64::EPSILON);
}

#[test]
fn test_metrics_accumulate_on_compress() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp5);
let input = b"Hello world! Hello world! Hello world! Hello world! x";
let (comp_size, flags) = bulk.compress(input).unwrap();

Expand All @@ -509,8 +509,8 @@ mod tests {

#[test]
fn test_metrics_accumulate_on_decompress() {
let mut sender = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut receiver = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut sender = BulkCompressor::new(CompressionType::Rdp5);
let mut receiver = BulkCompressor::new(CompressionType::Rdp5);

let input = b"Hello world! Hello world! Hello world! Hello world! x";
let (comp_size, flags) = sender.compress(input).unwrap();
Expand All @@ -533,8 +533,8 @@ mod tests {

#[test]
fn test_metrics_accumulate_across_multiple_calls() {
let mut sender = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut receiver = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut sender = BulkCompressor::new(CompressionType::Rdp5);
let mut receiver = BulkCompressor::new(CompressionType::Rdp5);

let inputs: &[&[u8]] = &[
b"Hello world! Hello world! Hello world! Hello world! x",
Expand All @@ -557,7 +557,7 @@ mod tests {

#[test]
fn test_metrics_not_reset_by_context_reset() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp5).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp5);
let input = b"Hello world! Hello world! Hello world! Hello world! x";
let _ = bulk.compress(input).unwrap();
let before = bulk.total_uncompressed_bytes();
Expand All @@ -569,7 +569,7 @@ mod tests {

#[test]
fn test_bulk_compress_rdp61_sets_type_bits() {
let mut bulk = BulkCompressor::new(CompressionType::Rdp61).unwrap();
let mut bulk = BulkCompressor::new(CompressionType::Rdp61);
let input = b"XCRUSH test data with repeated XCRUSH patterns for compression!!";
let (_size, flags) = bulk.compress(input).unwrap();

Expand Down
4 changes: 2 additions & 2 deletions crates/ironrdp-bulk/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@
//! use ironrdp_bulk::{BulkCompressor, CompressionType, flags};
//!
//! // Create sender (compressor) and receiver (decompressor)
//! let mut sender = BulkCompressor::new(CompressionType::Rdp5).unwrap();
//! let mut receiver = BulkCompressor::new(CompressionType::Rdp5).unwrap();
//! let mut sender = BulkCompressor::new(CompressionType::Rdp5);
//! let mut receiver = BulkCompressor::new(CompressionType::Rdp5);
//!
//! let input = b"Hello world! Hello world! Hello world! Hello world! x";
//!
Expand Down
Loading
Loading