diff --git a/Cargo.lock b/Cargo.lock index a65f1e7..2392622 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "addr2line" @@ -142,9 +142,9 @@ dependencies = [ [[package]] name = "bytes" -version = "1.7.1" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8318a53db07bb3f8dca91a600466bdb3f2eaadeedfdbcf02e1accbad9271ba50" +checksum = "325918d6fe32f23b19878fe4b34794ae41fc19ddbe53b10571a4874d44ffd39b" [[package]] name = "cc" @@ -892,6 +892,7 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" name = "windlass" version = "0.1.0" dependencies = [ + "bytes", "flate2", "futures", "paste", diff --git a/crates/windlass/Cargo.toml b/crates/windlass/Cargo.toml index dedf006..898062b 100644 --- a/crates/windlass/Cargo.toml +++ b/crates/windlass/Cargo.toml @@ -12,6 +12,7 @@ serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } tokio = { optional = true, workspace = true, features = ["sync", "io-util", "time", "macros", "rt", "process", "fs"] } tracing = { workspace = true } +bytes = "1.9.0" [dev-dependencies] test-log = { version = "0.2.16", features = ["trace"], default-features = false } diff --git a/crates/windlass/src/application.rs b/crates/windlass/src/application.rs index dc884ac..57e06dc 100644 --- a/crates/windlass/src/application.rs +++ b/crates/windlass/src/application.rs @@ -2,6 +2,7 @@ use std::{io::Write, time::Instant}; +use bytes::BytesMut; use thiserror::Error; use tracing::{debug, trace}; @@ -23,7 +24,7 @@ impl Connection { self.transport.receive_data(buf, receive_time); while let Some(frame) = self.transport.pop_frame() { - let mut rest = frame.payload.as_slice(); + let mut rest = &*frame.payload; while !rest.is_empty() { self.receive_next_message(&mut rest, frame.receive_time)?; } @@ -73,12 +74,12 @@ impl Default for Connection { let mut transport = transport::Connection::default(); // Immediately send request for first part of dictionary. - transport.push_message(Identify::encode(0, 40).payload.content); + transport.push_message(&Identify::encode(0, 40).payload.content); Self { transport, dictionary: DictionaryReceiveState::InProgress { start: Instant::now(), - so_far: Vec::default(), + so_far: BytesMut::with_capacity(800), }, } } @@ -86,7 +87,7 @@ impl Default for Connection { #[derive(Debug)] enum DictionaryReceiveState { - InProgress { start: Instant, so_far: Vec }, + InProgress { start: Instant, so_far: BytesMut }, Done(Dictionary), } @@ -127,7 +128,7 @@ impl DictionaryReceiveState { "received more of dictionary" ); transport.push_message( - Identify::encode( + &Identify::encode( u32::try_from(so_far.len()).expect("identify data length overflowed u32!"), 40, ) @@ -140,7 +141,7 @@ impl DictionaryReceiveState { // done receiving dictionary trace!("done receiving dictionary, attempting decode"); - let dict = Dictionary::from_complete_lookup_resp(so_far.as_slice())?; + let dict = Dictionary::from_complete_lookup_resp(so_far)?; trace!(dict = ?dict, "dictionary in place"); *self = Self::Done(dict); diff --git a/crates/windlass/src/transport/frame.rs b/crates/windlass/src/transport/frame.rs index 11ca9e2..b7480b3 100644 --- a/crates/windlass/src/transport/frame.rs +++ b/crates/windlass/src/transport/frame.rs @@ -1,29 +1,29 @@ +use bytes::{BufMut, Bytes, BytesMut}; use std::{cmp::min, collections::VecDeque, mem, time::Instant}; use tracing::trace; use crate::{ encoding::crc16, transport::{ - MESSAGE_DEST, MESSAGE_HEADER_SIZE, MESSAGE_LENGTH_MAX, MESSAGE_LENGTH_MIN, + Error, MESSAGE_DEST, MESSAGE_HEADER_SIZE, MESSAGE_LENGTH_MAX, MESSAGE_LENGTH_MIN, MESSAGE_POSITION_SEQ, MESSAGE_SEQ_MASK, MESSAGE_TRAILER_CRC, MESSAGE_TRAILER_SIZE, MESSAGE_VALUE_SYNC, }, }; -use super::Error; - /// A link-level frame. See [klipper docs](https://www.klipper3d.org/Protocol.html#message-blocks) #[derive(Debug, PartialEq, Eq)] pub struct Received { pub receive_time: Instant, pub sequence: u8, - pub payload: Vec, + pub payload: Bytes, } /// Buffers and reads link-level frames. See [`ReceivedFrame`]. #[derive(Debug, Default)] pub struct Reader { buf: VecDeque, + byte_buf: BytesMut, partial: ReceiveState, } @@ -31,7 +31,10 @@ impl Reader { /// Process new data from `b` pub fn receive_data(&mut self, mut b: &[u8], receive_time: Instant) { while !b.is_empty() { - if let Some((frame, remaining)) = self.partial.receive_data(b, receive_time) { + if let Some((frame, remaining)) = + self.partial + .receive_data(b, &mut self.byte_buf, receive_time) + { trace!(frame = ?frame, "receivestate returned frame"); self.buf.push_back(frame); b = remaining; @@ -62,9 +65,6 @@ pub enum ReceiveState { /// The total length of this frame, including header and footer. len: usize, - /// Accumulated data so far. - so_far: Vec, - /// When the packet started being received. receive_time: Instant, }, @@ -74,14 +74,15 @@ impl ReceiveState { pub fn receive_data<'a>( &mut self, mut b: &'a [u8], + byte_buf: &mut BytesMut, receive_time: Instant, ) -> Option<(Received, &'a [u8])> { while !b.is_empty() { match self { Self::Unsynced => self.attempt_sync(&mut b), - Self::Synced => self.attempt_start_message(&mut b, receive_time), + Self::Synced => self.attempt_start_message(&mut b, byte_buf, receive_time), Self::Receiving { .. } => { - if let Some(f) = self.attempt_receive_more(&mut b) { + if let Some(f) = self.attempt_receive_more(&mut b, byte_buf) { return Some((f, b)); } } @@ -102,7 +103,12 @@ impl ReceiveState { } } - fn attempt_start_message(&mut self, b: &mut &[u8], receive_time: Instant) { + fn attempt_start_message( + &mut self, + b: &mut &[u8], + byte_buf: &mut BytesMut, + receive_time: Instant, + ) { // Attempt to start a new frame let len = b[0] as usize; if !(MESSAGE_LENGTH_MIN..=MESSAGE_LENGTH_MAX).contains(&len) { @@ -112,34 +118,29 @@ impl ReceiveState { } trace!(len = len, receiving = true); - let mut so_far = Vec::with_capacity(len); - so_far.push(b[0]); - *self = Self::Receiving { - len, - so_far, - receive_time, - }; + byte_buf.put_u8(b[0]); + *self = Self::Receiving { len, receive_time }; *b = &b[min(1, b.len())..]; } #[allow(clippy::cognitive_complexity)] // it will be more confusing to split it up tbh - fn attempt_receive_more(&mut self, b: &mut &[u8]) -> Option { - let Self::Receiving { so_far, len, .. } = self else { + fn attempt_receive_more(&mut self, b: &mut &[u8], byte_buf: &mut BytesMut) -> Option { + let Self::Receiving { len, .. } = self else { unreachable!() }; // Continue to receive data for frame let len = *len; - let take = min(len - so_far.len(), b.len()); - so_far.extend_from_slice(&b[..take]); + let take = min(len - byte_buf.len(), b.len()); + byte_buf.extend_from_slice(&b[..take]); trace!( total_len = len, added = take, - so_far = so_far.len(), + so_far = byte_buf.len(), receiving = true ); *b = &b[min(take, b.len())..]; - if so_far.len() < len { + if byte_buf.len() < len { // Frame not yet done, most likely b is empty now. return None; } @@ -147,7 +148,8 @@ impl ReceiveState { // Check validity of frame // We checked that len >=MESSAGE_LEN_MIN when we started receiving, so // all of these are guaranteed to be in bounds - let seq = so_far[MESSAGE_POSITION_SEQ]; + let frame = byte_buf.split().freeze(); + let seq = frame[MESSAGE_POSITION_SEQ]; trace!(received = true, seq = seq); if seq & !MESSAGE_SEQ_MASK != MESSAGE_DEST { @@ -156,9 +158,9 @@ impl ReceiveState { return None; } - let actual_crc = crc16(&so_far[..len - MESSAGE_TRAILER_CRC]); - let frame_crc = (u16::from(so_far[len - MESSAGE_TRAILER_CRC])) << 8 - | (u16::from(so_far[len - MESSAGE_TRAILER_CRC + 1])); + let actual_crc = crc16(&frame[..len - MESSAGE_TRAILER_CRC]); + let frame_crc = (u16::from(frame[len - MESSAGE_TRAILER_CRC])) << 8 + | (u16::from(frame[len - MESSAGE_TRAILER_CRC + 1])); if frame_crc != actual_crc { trace!( actual_crc = actual_crc, @@ -171,19 +173,14 @@ impl ReceiveState { } // Set current state back to synced - let Self::Receiving { - so_far, - receive_time, - .. - } = mem::replace(self, Self::Synced) - else { + let Self::Receiving { receive_time, .. } = mem::replace(self, Self::Synced) else { unreachable!() }; Some(Received { receive_time, sequence: seq & MESSAGE_SEQ_MASK, - payload: so_far[MESSAGE_HEADER_SIZE..len - MESSAGE_TRAILER_SIZE].into(), + payload: frame.slice(MESSAGE_HEADER_SIZE..len - MESSAGE_TRAILER_SIZE), }) } } @@ -196,7 +193,7 @@ pub struct Sent { pub at: Instant, /// The payload of the message, including frame header - pub payload: Vec, + pub payload: Bytes, /// True iff the frame has been retransmitted at least once pub is_retransmit: bool, @@ -209,24 +206,24 @@ impl Sent { return Err(Error::MessageTooLong); } - let mut payload = Vec::with_capacity(len); + let mut payload = BytesMut::with_capacity(len); let Ok(len) = u8::try_from(len) else { unreachable!() }; - payload.push(len); - payload.push( + payload.put_u8(len); + payload.put_u8( MESSAGE_DEST | ((u8::try_from(sequence).expect("sequence number overflow")) & MESSAGE_SEQ_MASK), ); payload.extend_from_slice(data); let crc = crc16(&payload[0..len as usize - MESSAGE_TRAILER_SIZE]); - payload.push(((crc >> 8) & 0xFF) as u8); - payload.push((crc & 0xFF) as u8); - payload.push(MESSAGE_VALUE_SYNC); + payload.put_u8(((crc >> 8) & 0xFF) as u8); + payload.put_u8((crc & 0xFF) as u8); + payload.put_u8(MESSAGE_VALUE_SYNC); Ok(Self { at: Instant::now(), - payload, + payload: payload.freeze(), is_retransmit: false, }) } @@ -234,6 +231,7 @@ impl Sent { #[cfg(test)] mod tests { + use bytes::Bytes; use std::time::{Duration, Instant}; use test_log::test; @@ -297,7 +295,6 @@ mod tests { rdr.partial, ReceiveState::Receiving { len: 40, - so_far: vec![40], receive_time: now }, "now in middle of receiving" @@ -355,7 +352,7 @@ mod tests { Some(ReceivedFrame { receive_time: now, sequence: 7, - payload: vec![] + payload: Bytes::new() }), "received message at front of buffer" ); @@ -385,7 +382,7 @@ mod tests { Some(ReceivedFrame { receive_time: now + Duration::from_secs(1), sequence: 9, - payload: vec![] + payload: Bytes::new() }), "received message at front of buffer" ); @@ -418,7 +415,7 @@ mod tests { Some(ReceivedFrame { receive_time: now, sequence: 2, - payload: vec![1, 2, 3, 4, 5] + payload: Bytes::from_static(&[1, 2, 3, 4, 5]) }), "received message at front of buffer" ); @@ -453,7 +450,7 @@ mod tests { Some(ReceivedFrame { receive_time: now + Duration::from_secs(1), sequence: 4, - payload: vec![1, 2, 3, 4, 5] + payload: Bytes::from_static(&[1, 2, 3, 4, 5]) }), "received message at front of buffer" ); diff --git a/crates/windlass/src/transport/mod.rs b/crates/windlass/src/transport/mod.rs index f2c4a97..75bd141 100644 --- a/crates/windlass/src/transport/mod.rs +++ b/crates/windlass/src/transport/mod.rs @@ -3,6 +3,7 @@ mod frame; mod rtt; +use bytes::{Bytes, BytesMut}; use rtt::RttState; use std::{collections::VecDeque, io::Write, time::Instant}; use tracing::trace; @@ -35,7 +36,8 @@ pub struct Connection { inflight_messages: VecDeque, received_frames: VecDeque, - pending_messages: VecDeque>, + pending_buf: BytesMut, + pending_messages: VecDeque, } impl Connection { @@ -45,9 +47,11 @@ impl Connection { } /// Queue a new message to be transmitted - pub fn push_message(&mut self, msg: Vec) { + pub fn push_message(&mut self, msg: &[u8]) { trace!(msg = ?msg, "queued outgoing message"); - self.pending_messages.push_back(msg); + self.pending_buf.extend_from_slice(msg); + self.pending_messages + .push_back(self.pending_buf.split().freeze()); } /// Process received data and update internal state @@ -188,19 +192,18 @@ impl Connection { /// Send as many more frames as possible from [`self.pending_messages`] fn send_more_frames(&mut self, w: &mut impl Write) -> Result<(), Error> { while self.can_send() { - let mut buf = Vec::new(); while let Some(next) = self.pending_messages.front() { - if buf.len() + next.len() <= MESSAGE_LENGTH_PAYLOAD_MAX { + if self.pending_buf.len() + next.len() <= MESSAGE_LENGTH_PAYLOAD_MAX { // Add to the end of the frame. Unwrap is safe because we already peeked. - let mut next = self.pending_messages.pop_front().unwrap(); + let next = self.pending_messages.pop_front().unwrap(); trace!(added_len = next.len(), "queued pending message"); - buf.append(&mut next); + self.pending_buf.extend_from_slice(&next); } else { break; } } - let frame = frame::Sent::new(self.next_send_seq, &buf)?; + let frame = frame::Sent::new(self.next_send_seq, &self.pending_buf.split())?; trace!(frame = ?frame, "new frame in flight"); self.inflight_messages.push_back(frame); self.next_send_seq += 1; @@ -268,7 +271,7 @@ mod tests { fn send_single_message() { let mut conn = Connection::default(); - conn.push_message(vec![1, 2, 3]); + conn.push_message(&[1, 2, 3]); let mut buf = vec![]; conn.write_data(&mut buf).unwrap(); @@ -293,8 +296,8 @@ mod tests { fn send_consolidates_block() { let mut conn = Connection::default(); - conn.push_message(vec![1, 2, 3]); - conn.push_message(vec![4, 5, 6]); + conn.push_message(&[1, 2, 3]); + conn.push_message(&[4, 5, 6]); let mut buf = vec![]; conn.write_data(&mut buf).unwrap(); @@ -332,9 +335,9 @@ mod tests { let mut conn = Connection::default(); let big_msg = vec![2; MESSAGE_LENGTH_PAYLOAD_MAX]; - conn.push_message(big_msg.clone()); - conn.push_message(vec![1, 2, 3]); - conn.push_message(vec![4, 5, 6]); + conn.push_message(&big_msg); + conn.push_message(&[1, 2, 3]); + conn.push_message(&[4, 5, 6]); let mut buf = vec![]; conn.write_data(&mut buf).unwrap(); @@ -389,8 +392,8 @@ mod tests { let mut conn = Connection::default(); let big_msg = vec![2; MESSAGE_LENGTH_PAYLOAD_MAX]; - conn.push_message(vec![1, 2, 3]); - conn.push_message(big_msg.clone()); + conn.push_message(&[1, 2, 3]); + conn.push_message(&big_msg); let mut buf = vec![]; conn.write_data(&mut buf).unwrap(); @@ -433,7 +436,7 @@ mod tests { fn send_ack_removes_pending() { let mut conn = Connection::default(); - conn.push_message(vec![1, 2, 3]); + conn.push_message(&[1, 2, 3]); let mut buf = vec![]; conn.write_data(&mut buf).unwrap(); @@ -467,7 +470,7 @@ mod tests { fn send_retransmits_on_timeout() { let mut conn = Connection::default(); - conn.push_message(vec![1, 2, 3]); + conn.push_message(&[1, 2, 3]); let mut buf = vec![]; let msg_sent = Instant::now(); @@ -511,7 +514,7 @@ mod tests { // both our and klipper's implementation can't deal with nacks at the start of the connection stream // so send at least one successfull message first. - conn.push_message(vec![1, 2, 3]); + conn.push_message(&[1, 2, 3]); conn.write_data(&mut vec![]).unwrap(); conn.receive_data( &[ @@ -525,7 +528,7 @@ mod tests { Instant::now(), ); - conn.push_message(vec![1, 2, 3]); + conn.push_message(&[1, 2, 3]); let mut buf = vec![]; conn.write_data(&mut buf).unwrap();