Skip to content

Commit 0530990

Browse files
emilkemilk
committed
Support RRD streams with and without compression. Turn off for SDK comms (#2219)
Closes #2216 The compression for SDK comms was added in #2065 and now removed in this PR. * [x] I have read and agree to [Contributor Guide](https://github.com/rerun-io/rerun/blob/main/CONTRIBUTING.md) and the [Code of Conduct](https://github.com/rerun-io/rerun/blob/main/CODE_OF_CONDUCT.md) <!-- This line will get updated when the PR build summary job finishes. --> PR Build Summary: https://build.rerun.io/pr/2219
1 parent de981bb commit 0530990

File tree

10 files changed

+301
-65
lines changed

10 files changed

+301
-65
lines changed

‎crates/re_data_store/examples/memory_usage.rs

+3-1
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,9 @@ fn log_messages() {
6565

6666
fn encode_log_msg(log_msg: &LogMsg) -> Vec<u8> {
6767
let mut bytes = vec![];
68-
re_log_encoding::encoder::encode(std::iter::once(log_msg), &mut bytes).unwrap();
68+
let encoding_options = re_log_encoding::EncodingOptions::COMPRESSED;
69+
re_log_encoding::encoder::encode(encoding_options, std::iter::once(log_msg), &mut bytes)
70+
.unwrap();
6971
bytes
7072
}
7173

‎crates/re_log_encoding/benches/msg_encode_benchmark.rs

+2-1
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,9 @@ criterion_group!(
2727
criterion_main!(benches);
2828

2929
fn encode_log_msgs(messages: &[LogMsg]) -> Vec<u8> {
30+
let encoding_options = re_log_encoding::EncodingOptions::COMPRESSED;
3031
let mut bytes = vec![];
31-
re_log_encoding::encoder::encode(messages.iter(), &mut bytes).unwrap();
32+
re_log_encoding::encoder::encode(encoding_options, messages.iter(), &mut bytes).unwrap();
3233
assert!(bytes.len() > messages.len());
3334
bytes
3435
}

‎crates/re_log_encoding/src/decoder.rs

+84-20
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22
33
use re_log_types::LogMsg;
44

5+
use crate::{Compression, EncodingOptions, Serializer};
6+
57
// ----------------------------------------------------------------------------
68

79
fn warn_on_version_mismatch(encoded_version: [u8; 4]) {
@@ -29,6 +31,12 @@ pub enum DecodeError {
2931
#[error("Not an .rrd file")]
3032
NotAnRrd,
3133

34+
#[error("Found an .rrd file from a Rerun version from 0.5.1 or earlier")]
35+
OldRrdVersion,
36+
37+
#[error("Failed to decode the options: {0}")]
38+
Options(#[from] crate::OptionsError),
39+
3240
#[error("Failed to read: {0}")]
3341
Read(std::io::Error),
3442

@@ -52,26 +60,70 @@ pub fn decode_bytes(bytes: &[u8]) -> Result<Vec<LogMsg>, DecodeError> {
5260

5361
// ----------------------------------------------------------------------------
5462

63+
enum Decompressor<R: std::io::Read> {
64+
Uncompressed(R),
65+
Lz4(lz4_flex::frame::FrameDecoder<R>),
66+
}
67+
68+
impl<R: std::io::Read> Decompressor<R> {
69+
fn new(compression: Compression, read: R) -> Self {
70+
match compression {
71+
Compression::Off => Self::Uncompressed(read),
72+
Compression::LZ4 => Self::Lz4(lz4_flex::frame::FrameDecoder::new(read)),
73+
}
74+
}
75+
76+
pub fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), DecodeError> {
77+
use std::io::Read as _;
78+
79+
match self {
80+
Decompressor::Uncompressed(read) => read.read_exact(buf).map_err(DecodeError::Read),
81+
Decompressor::Lz4(lz4) => lz4.read_exact(buf).map_err(DecodeError::Lz4),
82+
}
83+
}
84+
}
85+
86+
// ----------------------------------------------------------------------------
87+
5588
pub struct Decoder<R: std::io::Read> {
56-
lz4_decoder: lz4_flex::frame::FrameDecoder<R>,
89+
decompressor: Decompressor<R>,
5790
buffer: Vec<u8>,
5891
}
5992

6093
impl<R: std::io::Read> Decoder<R> {
6194
pub fn new(mut read: R) -> Result<Self, DecodeError> {
6295
crate::profile_function!();
6396

64-
let mut header = [0_u8; 4];
65-
read.read_exact(&mut header).map_err(DecodeError::Read)?;
66-
if &header != b"RRF0" {
67-
return Err(DecodeError::NotAnRrd);
97+
{
98+
let mut header = [0_u8; 4];
99+
read.read_exact(&mut header).map_err(DecodeError::Read)?;
100+
if &header == b"RRF0" {
101+
return Err(DecodeError::OldRrdVersion);
102+
} else if &header != crate::RRD_HEADER {
103+
return Err(DecodeError::NotAnRrd);
104+
}
105+
}
106+
107+
{
108+
let mut version_bytes = [0_u8; 4];
109+
read.read_exact(&mut version_bytes)
110+
.map_err(DecodeError::Read)?;
111+
warn_on_version_mismatch(version_bytes);
112+
}
113+
114+
let options = {
115+
let mut options_bytes = [0_u8; 4];
116+
read.read_exact(&mut options_bytes)
117+
.map_err(DecodeError::Read)?;
118+
EncodingOptions::from_bytes(options_bytes)?
119+
};
120+
121+
match options.serializer {
122+
Serializer::MsgPack => {}
68123
}
69-
read.read_exact(&mut header).map_err(DecodeError::Read)?;
70-
warn_on_version_mismatch(header);
71124

72-
let lz4_decoder = lz4_flex::frame::FrameDecoder::new(read);
73125
Ok(Self {
74-
lz4_decoder,
126+
decompressor: Decompressor::new(options.compression, read),
75127
buffer: vec![],
76128
})
77129
}
@@ -82,18 +134,17 @@ impl<R: std::io::Read> Iterator for Decoder<R> {
82134

83135
fn next(&mut self) -> Option<Self::Item> {
84136
crate::profile_function!();
85-
use std::io::Read as _;
86137

87138
let mut len = [0_u8; 8];
88-
self.lz4_decoder.read_exact(&mut len).ok()?;
139+
self.decompressor.read_exact(&mut len).ok()?;
89140
let len = u64::from_le_bytes(len) as usize;
90141

91142
self.buffer.resize(len, 0);
92143

93144
{
94145
crate::profile_scope!("lz4");
95-
if let Err(err) = self.lz4_decoder.read_exact(&mut self.buffer) {
96-
return Some(Err(DecodeError::Lz4(err)));
146+
if let Err(err) = self.decompressor.read_exact(&mut self.buffer) {
147+
return Some(Err(err));
97148
}
98149
}
99150

@@ -129,13 +180,26 @@ fn test_encode_decode() {
129180
},
130181
})];
131182

132-
let mut file = vec![];
133-
crate::encoder::encode(messages.iter(), &mut file).unwrap();
183+
let options = [
184+
EncodingOptions {
185+
compression: Compression::Off,
186+
serializer: Serializer::MsgPack,
187+
},
188+
EncodingOptions {
189+
compression: Compression::LZ4,
190+
serializer: Serializer::MsgPack,
191+
},
192+
];
193+
194+
for options in options {
195+
let mut file = vec![];
196+
crate::encoder::encode(options, messages.iter(), &mut file).unwrap();
134197

135-
let decoded_messages = Decoder::new(&mut file.as_slice())
136-
.unwrap()
137-
.collect::<Result<Vec<LogMsg>, DecodeError>>()
138-
.unwrap();
198+
let decoded_messages = Decoder::new(&mut file.as_slice())
199+
.unwrap()
200+
.collect::<Result<Vec<LogMsg>, DecodeError>>()
201+
.unwrap();
139202

140-
assert_eq!(messages, decoded_messages);
203+
assert_eq!(messages, decoded_messages);
204+
}
141205
}

‎crates/re_log_encoding/src/encoder.rs

+104-37
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,10 @@ use std::io::Write as _;
44

55
use re_log_types::LogMsg;
66

7+
use crate::{Compression, EncodingOptions};
8+
9+
// ----------------------------------------------------------------------------
10+
711
/// On failure to encode or serialize a [`LogMsg`].
812
#[derive(thiserror::Error, Debug)]
913
pub enum EncodeError {
@@ -26,11 +30,12 @@ pub enum EncodeError {
2630
// ----------------------------------------------------------------------------
2731

2832
pub fn encode_to_bytes<'a>(
33+
options: EncodingOptions,
2934
msgs: impl IntoIterator<Item = &'a LogMsg>,
3035
) -> Result<Vec<u8>, EncodeError> {
3136
let mut bytes: Vec<u8> = vec![];
3237
{
33-
let mut encoder = Encoder::new(std::io::Cursor::new(&mut bytes))?;
38+
let mut encoder = Encoder::new(options, std::io::Cursor::new(&mut bytes))?;
3439
for msg in msgs {
3540
encoder.append(msg)?;
3641
}
@@ -41,14 +46,42 @@ pub fn encode_to_bytes<'a>(
4146

4247
// ----------------------------------------------------------------------------
4348

44-
/// Encode a stream of [`LogMsg`] into an `.rrd` file.
45-
pub struct Encoder<W: std::io::Write> {
46-
/// Set to None when finished.
49+
struct Lz4Compressor<W: std::io::Write> {
50+
/// `None` if finished.
4751
lz4_encoder: Option<lz4_flex::frame::FrameEncoder<W>>,
48-
buffer: Vec<u8>,
4952
}
5053

51-
impl<W: std::io::Write> Drop for Encoder<W> {
54+
impl<W: std::io::Write> Lz4Compressor<W> {
55+
pub fn new(write: W) -> Self {
56+
Self {
57+
lz4_encoder: Some(lz4_flex::frame::FrameEncoder::new(write)),
58+
}
59+
}
60+
61+
pub fn write(&mut self, bytes: &[u8]) -> Result<(), EncodeError> {
62+
if let Some(lz4_encoder) = &mut self.lz4_encoder {
63+
lz4_encoder
64+
.write_all(bytes)
65+
.map_err(EncodeError::Lz4Write)?;
66+
67+
Ok(())
68+
} else {
69+
Err(EncodeError::AlreadyFinished)
70+
}
71+
}
72+
73+
pub fn finish(&mut self) -> Result<(), EncodeError> {
74+
if let Some(lz4_encoder) = self.lz4_encoder.take() {
75+
lz4_encoder.finish().map_err(EncodeError::Lz4Finish)?;
76+
Ok(())
77+
} else {
78+
re_log::warn!("Encoder::finish called twice");
79+
Ok(())
80+
}
81+
}
82+
}
83+
84+
impl<W: std::io::Write> Drop for Lz4Compressor<W> {
5285
fn drop(&mut self) {
5386
if self.lz4_encoder.is_some() {
5487
re_log::warn!("Encoder dropped without calling finish()!");
@@ -59,73 +92,107 @@ impl<W: std::io::Write> Drop for Encoder<W> {
5992
}
6093
}
6194

95+
#[allow(clippy::large_enum_variant)]
96+
enum Compressor<W: std::io::Write> {
97+
Off(W),
98+
Lz4(Lz4Compressor<W>),
99+
}
100+
101+
impl<W: std::io::Write> Compressor<W> {
102+
pub fn new(compression: Compression, write: W) -> Self {
103+
match compression {
104+
Compression::Off => Self::Off(write),
105+
Compression::LZ4 => Self::Lz4(Lz4Compressor::new(write)),
106+
}
107+
}
108+
109+
pub fn write(&mut self, bytes: &[u8]) -> Result<(), EncodeError> {
110+
let len = (bytes.len() as u64).to_le_bytes();
111+
112+
match self {
113+
Compressor::Off(write) => {
114+
write.write_all(&len).map_err(EncodeError::Write)?;
115+
write.write_all(bytes).map_err(EncodeError::Write)
116+
}
117+
Compressor::Lz4(lz4) => {
118+
lz4.write(&len)?;
119+
lz4.write(bytes)
120+
}
121+
}
122+
}
123+
124+
pub fn finish(&mut self) -> Result<(), EncodeError> {
125+
match self {
126+
Compressor::Off(_) => Ok(()),
127+
Compressor::Lz4(lz4) => lz4.finish(),
128+
}
129+
}
130+
}
131+
132+
// ----------------------------------------------------------------------------
133+
134+
/// Encode a stream of [`LogMsg`] into an `.rrd` file.
135+
pub struct Encoder<W: std::io::Write> {
136+
compressor: Compressor<W>,
137+
buffer: Vec<u8>,
138+
}
139+
62140
impl<W: std::io::Write> Encoder<W> {
63-
pub fn new(mut write: W) -> Result<Self, EncodeError> {
141+
pub fn new(options: EncodingOptions, mut write: W) -> Result<Self, EncodeError> {
64142
let rerun_version = re_build_info::CrateVersion::parse(env!("CARGO_PKG_VERSION"));
65143

66-
write.write_all(b"RRF0").map_err(EncodeError::Write)?;
144+
write
145+
.write_all(crate::RRD_HEADER)
146+
.map_err(EncodeError::Write)?;
67147
write
68148
.write_all(&rerun_version.to_bytes())
69149
.map_err(EncodeError::Write)?;
150+
write
151+
.write_all(&options.to_bytes())
152+
.map_err(EncodeError::Write)?;
70153

71-
let lz4_encoder = lz4_flex::frame::FrameEncoder::new(write);
154+
match options.serializer {
155+
crate::Serializer::MsgPack => {}
156+
}
72157

73158
Ok(Self {
74-
lz4_encoder: Some(lz4_encoder),
159+
compressor: Compressor::new(options.compression, write),
75160
buffer: vec![],
76161
})
77162
}
78163

79164
pub fn append(&mut self, message: &LogMsg) -> Result<(), EncodeError> {
80-
let Self {
81-
lz4_encoder,
82-
buffer,
83-
} = self;
165+
let Self { compressor, buffer } = self;
84166

85-
if let Some(lz4_encoder) = lz4_encoder {
86-
buffer.clear();
87-
rmp_serde::encode::write_named(buffer, message)?;
167+
buffer.clear();
168+
rmp_serde::encode::write_named(buffer, message)?;
88169

89-
lz4_encoder
90-
.write_all(&(buffer.len() as u64).to_le_bytes())
91-
.map_err(EncodeError::Lz4Write)?;
92-
lz4_encoder
93-
.write_all(buffer)
94-
.map_err(EncodeError::Lz4Write)?;
95-
96-
Ok(())
97-
} else {
98-
Err(EncodeError::AlreadyFinished)
99-
}
170+
compressor.write(buffer)
100171
}
101172

102173
pub fn finish(&mut self) -> Result<(), EncodeError> {
103-
if let Some(lz4_encoder) = self.lz4_encoder.take() {
104-
lz4_encoder.finish().map_err(EncodeError::Lz4Finish)?;
105-
Ok(())
106-
} else {
107-
re_log::warn!("Encoder::finish called twice");
108-
Ok(())
109-
}
174+
self.compressor.finish()
110175
}
111176
}
112177

113178
pub fn encode<'a>(
179+
options: EncodingOptions,
114180
messages: impl Iterator<Item = &'a LogMsg>,
115181
write: &mut impl std::io::Write,
116182
) -> Result<(), EncodeError> {
117-
let mut encoder = Encoder::new(write)?;
183+
let mut encoder = Encoder::new(options, write)?;
118184
for message in messages {
119185
encoder.append(message)?;
120186
}
121187
encoder.finish()
122188
}
123189

124190
pub fn encode_owned(
191+
options: EncodingOptions,
125192
messages: impl Iterator<Item = LogMsg>,
126193
write: impl std::io::Write,
127194
) -> Result<(), EncodeError> {
128-
let mut encoder = Encoder::new(write)?;
195+
let mut encoder = Encoder::new(options, write)?;
129196
for message in messages {
130197
encoder.append(&message)?;
131198
}

0 commit comments

Comments
 (0)