Merge pull request #108 from MaxVerevkin/msgread

read the enitre message in one go without copies

rustbus/benches/marshal_benchmark.rs

@@ -12,10 +12,10 @@ fn marsh(msg: &rustbus::message_builder::MarshalledMessage, buf: &mut Vec<u8>) {
 }
 
 fn unmarshal(buf: &[u8]) {
-    let (hdrbytes, header) = unmarshal_header(&buf, 0).unwrap();
-    let (dynhdrbytes, dynheader) = unmarshal_dynamic_header(&header, &buf, hdrbytes).unwrap();
+    let (hdrbytes, header) = unmarshal_header(buf, 0).unwrap();
+    let (dynhdrbytes, dynheader) = unmarshal_dynamic_header(&header, buf, hdrbytes).unwrap();
     let (_, _unmarshed_msg) =
-        unmarshal_next_message(&header, dynheader, &buf, hdrbytes + dynhdrbytes).unwrap();
+        unmarshal_next_message(&header, dynheader, buf.to_vec(), hdrbytes + dynhdrbytes).unwrap();
 }
 
 fn criterion_benchmark(c: &mut Criterion) {

rustbus/examples/conn.rs

@@ -35,10 +35,7 @@ fn main() -> Result<(), rustbus::connection::Error> {
     println!("\n");
 
     let reqname_serial = rpc_con
-        .send_message(&mut standard_messages::request_name(
-            "io.killing.spark".into(),
-            0,
-        ))?
+        .send_message(&mut standard_messages::request_name("io.killing.spark", 0))?
         .write_all()
         .unwrap();
 
@@ -73,7 +70,7 @@ fn main() -> Result<(), rustbus::connection::Error> {
     println!("\n");
     println!("\n");
 
-    let mut sig_listen_msg = standard_messages::add_match("type='signal'".into());
+    let mut sig_listen_msg = standard_messages::add_match("type='signal'");
 
     //println!("Send message: {:?}", sig_listen_msg);
     rpc_con

rustbus/examples/dispatch.rs

@@ -70,10 +70,10 @@ fn main() {
     con.send_hello(rustbus::connection::Timeout::Infinite)
         .unwrap();
 
-    if std::env::args().find(|arg| "server".eq(arg)).is_some() {
+    if std::env::args().any(|arg| "server".eq(&arg)) {
         con.send
             .send_message(&mut rustbus::standard_messages::request_name(
-                "killing.spark.io".into(),
+                "killing.spark.io",
                 rustbus::standard_messages::DBUS_NAME_FLAG_REPLACE_EXISTING,
             ))
             .unwrap()

rustbus/examples/fd.rs

@@ -20,7 +20,7 @@ fn main() -> Result<(), rustbus::connection::Error> {
             .write_all()
             .unwrap();
 
-        con.send_message(&mut standard_messages::add_match("type='signal'".into()))?
+        con.send_message(&mut standard_messages::add_match("type='signal'"))?
             .write_all()
             .unwrap();
 
@@ -31,10 +31,9 @@ fn main() -> Result<(), rustbus::connection::Error> {
                 .dynheader
                 .interface
                 .eq(&Some("io.killing.spark".to_owned()))
+                && signal.dynheader.member.eq(&Some("TestSignal".to_owned()))
             {
-                if signal.dynheader.member.eq(&Some("TestSignal".to_owned())) {
-                    break signal;
-                }
+                break signal;
             }
         };
 

rustbus/examples/server.rs

@@ -32,10 +32,7 @@ fn main() -> Result<(), rustbus::connection::Error> {
     let mut rpc_con = RpcConn::session_conn(Timeout::Infinite)?;
 
     let namereq_serial = rpc_con
-        .send_message(&mut standard_messages::request_name(
-            "io.killing.spark".into(),
-            0,
-        ))?
+        .send_message(&mut standard_messages::request_name("io.killing.spark", 0))?
         .write_all()
         .unwrap();
     let resp = rpc_con.wait_response(namereq_serial, Timeout::Infinite)?;

rustbus/fuzz/fuzz_targets/fuzz_unmarshal.rs

@@ -17,7 +17,7 @@ fuzz_target!(|data: &[u8]| {
     let (_bytes_used, msg) = match rustbus::wire::unmarshal::unmarshal_next_message(
         &header,
         dynheader,
-        data,
+        data.to_vec(),
         hdrbytes + dynhdrbytes,
     ) {
         Ok(msg) => msg,

rustbus/src/bin/fuzz_artifact.rs

@@ -34,7 +34,7 @@ fn run_artifact(path: &str) {
     let (_bytes_used, msg) = match rustbus::wire::unmarshal::unmarshal_next_message(
         &header,
         dynheader,
-        data,
+        data.clone(),
         hdrbytes + dynhdrbytes,
     ) {
         Ok(msg) => msg,

rustbus/src/connection/ll_conn.rs

@@ -33,7 +33,9 @@ pub struct RecvConn {
     stream: UnixStream,
 
     msg_buf_in: Vec<u8>,
+    msg_buf_filled: usize,
     cmsgs_in: Vec<ControlMessageOwned>,
+    cmsgspace: Vec<u8>,
 }
 
 pub struct DuplexConn {
@@ -56,14 +58,13 @@ impl RecvConn {
     /// Reads from the source once but takes care that the internal buffer only reaches at maximum max_buffer_size
     /// so we can process messages separatly and avoid leaking file descriptors to wrong messages
     fn refill_buffer(&mut self, max_buffer_size: usize, timeout: Timeout) -> Result<()> {
-        let bytes_to_read = max_buffer_size - self.msg_buf_in.len();
-
-        const BUFSIZE: usize = 512;
-        let mut tmpbuf = [0u8; BUFSIZE];
+        if self.msg_buf_in.len() != max_buffer_size {
+            self.msg_buf_in.resize(max_buffer_size, 0);
+        }
 
-        let iovec = IoSliceMut::new(&mut tmpbuf[..usize::min(bytes_to_read, BUFSIZE)]);
+        let iovec = IoSliceMut::new(&mut self.msg_buf_in[self.msg_buf_filled..max_buffer_size]);
 
-        let mut cmsgspace = cmsg_space!([RawFd; 10]);
+        self.cmsgspace.clear();
         let flags = MsgFlags::empty();
 
         let old_timeout = self.stream.read_timeout()?;
@@ -82,7 +83,7 @@ impl RecvConn {
         let msg = recvmsg::<SockaddrStorage>(
             self.stream.as_raw_fd(),
             iovec_mut,
-            Some(&mut cmsgspace),
+            Some(&mut self.cmsgspace),
             flags,
         )
         .map_err(|e| match e {
@@ -101,19 +102,18 @@ impl RecvConn {
 
         self.cmsgs_in.extend(msg.cmsgs());
         let bytes = msg.bytes;
-        self.msg_buf_in.extend_from_slice(&tmpbuf[..bytes]);
+        self.msg_buf_filled += bytes;
         Ok(())
     }
 
     pub fn bytes_needed_for_current_message(&self) -> Result<usize> {
-        if self.msg_buf_in.len() < 16 {
+        if self.msg_buf_filled < 16 {
             return Ok(16);
         }
-        let (_, header) = unmarshal::unmarshal_header(&self.msg_buf_in, 0)?;
-        let (_, header_fields_len) = crate::wire::util::parse_u32(
-            &self.msg_buf_in[unmarshal::HEADER_LEN..],
-            header.byteorder,
-        )?;
+        let msg_buf_in = &self.msg_buf_in[..self.msg_buf_filled];
+        let (_, header) = unmarshal::unmarshal_header(msg_buf_in, 0)?;
+        let (_, header_fields_len) =
+            crate::wire::util::parse_u32(&msg_buf_in[unmarshal::HEADER_LEN..], header.byteorder)?;
         let complete_header_size = unmarshal::HEADER_LEN + header_fields_len as usize + 4; // +4 because the length of the header fields does not count
 
         let padding_between_header_and_body = 8 - ((complete_header_size) % 8);
@@ -130,7 +130,7 @@ impl RecvConn {
 
     // Checks if the internal buffer currently holds a complete message
     pub fn buffer_contains_whole_message(&self) -> Result<bool> {
-        if self.msg_buf_in.len() < 16 {
+        if self.msg_buf_filled < 16 {
             return Ok(false);
         }
         let bytes_needed = self.bytes_needed_for_current_message();
@@ -142,7 +142,7 @@ impl RecvConn {
                     Err(e)
                 }
             }
-            Ok(bytes_needed) => Ok(self.msg_buf_in.len() >= bytes_needed),
+            Ok(bytes_needed) => Ok(self.msg_buf_filled >= bytes_needed),
         }
     }
     /// Blocks until a message has been read from the conn or the timeout has been reached
@@ -170,21 +170,22 @@ impl RecvConn {
     /// Blocks until a message has been read from the conn or the timeout has been reached
     pub fn get_next_message(&mut self, timeout: Timeout) -> Result<MarshalledMessage> {
         self.read_whole_message(timeout)?;
+        debug_assert_eq!(self.msg_buf_filled, self.msg_buf_in.len());
         let (hdrbytes, header) = unmarshal::unmarshal_header(&self.msg_buf_in, 0)?;
         let (dynhdrbytes, dynheader) =
             unmarshal::unmarshal_dynamic_header(&header, &self.msg_buf_in, hdrbytes)?;
 
         let (bytes_used, mut msg) = unmarshal::unmarshal_next_message(
             &header,
             dynheader,
-            &self.msg_buf_in,
+            std::mem::take(&mut self.msg_buf_in),
             hdrbytes + dynhdrbytes,
         )?;
 
-        if self.msg_buf_in.len() != bytes_used + hdrbytes + dynhdrbytes {
+        if self.msg_buf_filled != bytes_used + hdrbytes + dynhdrbytes {
             return Err(Error::UnmarshalError(UnmarshalError::NotAllBytesUsed));
         }
-        self.msg_buf_in.clear();
+        self.msg_buf_filled = 0;
 
         for cmsg in &self.cmsgs_in {
             match cmsg {
@@ -481,7 +482,9 @@ impl DuplexConn {
             },
             recv: RecvConn {
                 msg_buf_in: Vec::new(),
+                msg_buf_filled: 0,
                 cmsgs_in: Vec::new(),
+                cmsgspace: cmsg_space!([RawFd; 10]),
                 stream,
             },
         })

rustbus/src/tests.rs

@@ -48,7 +48,8 @@ fn test_marshal_unmarshal() {
     let headers_plus_padding = hdrbytes + dynhdrbytes + (8 - ((hdrbytes + dynhdrbytes) % 8));
     assert_eq!(headers_plus_padding, buf.len());
 
-    let (_, unmarshed_msg) = unmarshal_next_message(&header, dynheader, msg.get_buf(), 0).unwrap();
+    let (_, unmarshed_msg) =
+        unmarshal_next_message(&header, dynheader, msg.get_buf().to_vec(), 0).unwrap();
 
     let msg = unmarshed_msg.unmarshall_all().unwrap();
 

rustbus/src/tests/dbus_send.rs

@@ -33,7 +33,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
     // Request name
     let reqname_serial = rpc_con
         .send_message(&mut standard_messages::request_name(
-            "io.killing.spark.dbustest".into(),
+            "io.killing.spark.dbustest",
             0,
         ))?
         .write_all()
@@ -44,7 +44,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
     )?;
 
     let sig_serial = rpc_con
-        .send_message(&mut standard_messages::add_match("type='signal'".into()))?
+        .send_message(&mut standard_messages::add_match("type='signal'"))?
         .write_all()
         .map_err(force_finish_on_error)?;
     let _msg = rpc_con.wait_response(
@@ -53,7 +53,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
     )?;
 
     std::process::Command::new("dbus-send")
-        .args(&[
+        .args([
             "--dest=io.killing.spark.dbustest",
             "/",
             "io.killing.spark.dbustest.Member",
@@ -64,7 +64,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
         .unwrap();
 
     std::process::Command::new("dbus-send")
-        .args(&[
+        .args([
             "--dest=io.killing.spark.dbustest",
             "/",
             "io.killing.spark.dbustest.Member",
@@ -76,7 +76,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
         .unwrap();
 
     std::process::Command::new("dbus-send")
-        .args(&[
+        .args([
             "--dest=io.killing.spark.dbustest",
             "/",
             "io.killing.spark.dbustest.Member",
@@ -88,7 +88,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
         .unwrap();
 
     std::process::Command::new("dbus-send")
-        .args(&[
+        .args([
             "--dest=io.killing.spark.dbustest",
             "/",
             "io.killing.spark.dbustest.Member",
@@ -100,7 +100,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
         .unwrap();
 
     std::process::Command::new("dbus-send")
-        .args(&[
+        .args([
             "--dest=io.killing.spark.dbustest",
             "/",
             "io.killing.spark.dbustest.Member",
@@ -116,7 +116,7 @@ fn test_dbus_send_comp() -> Result<(), crate::connection::Error> {
         .unwrap();
 
     std::process::Command::new("dbus-send")
-        .args(&[
+        .args([
             "--dest=io.killing.spark.dbustest",
             "/",
             "io.killing.spark.dbustest.Member",

rustbus/src/tests/fdpassing.rs

@@ -20,12 +20,10 @@ fn test_fd_passing() {
         .unwrap()
         .write_all()
         .unwrap();
-    con2.send_message(&mut crate::standard_messages::add_match(
-        "type='signal'".into(),
-    ))
-    .unwrap()
-    .write_all()
-    .unwrap();
+    con2.send_message(&mut crate::standard_messages::add_match("type='signal'"))
+        .unwrap()
+        .write_all()
+        .unwrap();
 
     std::thread::sleep(std::time::Duration::from_secs(1));
 
@@ -39,10 +37,9 @@ fn test_fd_passing() {
             .dynheader
             .interface
             .eq(&Some("io.killing.spark".to_owned()))
+            && signal.dynheader.member.eq(&Some("TestSignal".to_owned()))
         {
-            if signal.dynheader.member.eq(&Some("TestSignal".to_owned())) {
-                break signal;
-            }
+            break signal;
         }
     };
 

rustbus/src/tests/verify_marshalling.rs

@@ -244,7 +244,7 @@ fn verify_dict_marshalling() {
         byteorder: ByteOrder::LittleEndian,
     };
     let ctx = &mut ctx;
-    (&map).marshal(ctx).unwrap();
+    map.marshal(ctx).unwrap();
     assert_eq!(
         ctx.buf,
         // Note the longer \0 chain after the length. This is the needed padding after the u32 length and the dict-entry

rustbus/src/wire/unmarshal.rs

@@ -148,11 +148,11 @@ pub fn unmarshal_body<'a, 'e>(
 pub fn unmarshal_next_message(
     header: &Header,
     dynheader: DynamicHeader,
-    buf: &[u8],
+    mut buf: Vec<u8>,
     offset: usize,
 ) -> UnmarshalResult<MarshalledMessage> {
     let sig = dynheader.signature.clone().unwrap_or_else(|| "".to_owned());
-    let padding = align_offset(8, buf, offset)?;
+    let padding = align_offset(8, &buf, offset)?;
 
     if header.body_len == 0 {
         let msg = MarshalledMessage {
@@ -169,14 +169,12 @@ pub fn unmarshal_next_message(
             return Err(UnmarshalError::NotEnoughBytes);
         }
 
+        // TODO: keep the offset around instead of shifting the bytes.
+        drop(buf.drain(..offset));
+
         let msg = MarshalledMessage {
             dynheader,
-            body: MarshalledMessageBody::from_parts(
-                buf[offset..].to_vec(),
-                vec![],
-                sig,
-                header.byteorder,
-            ),
+            body: MarshalledMessageBody::from_parts(buf, vec![], sig, header.byteorder),
             typ: header.typ,
             flags: header.flags,
         };