send: minor improvements

Author: github-af

src/lib.rs

@@ -26,6 +26,8 @@ mod sock_utils;
 #[allow(unsafe_code)]
 mod udp;
 
+pub const MAX_MTU: u16 = 9000;
+
 #[derive(Clone, Copy, clap::ValueEnum)]
 pub enum RecvMode {
     #[cfg(feature = "receive-native")]

src/protocol.rs

@@ -257,45 +257,27 @@ impl Block {
         client_id: ClientId,
         data: Option<&[u8]>,
     ) -> Result<Self, Error> {
-        match data {
-            None => {
-                let mut content = vec![
-                    0u8;
-                    usize::try_from(raptorq.transfer_length).map_err(|e| {
-                        Error::Other(format!("transfer_length: {e}"))
-                    })?
-                ];
-                let bytes = client_id.to_le_bytes();
-                content[0] = bytes[0];
-                content[1] = bytes[1];
-                content[2] = bytes[2];
-                content[3] = bytes[3];
-                content[4] = block.serialized();
-                Ok(Self(content))
-            }
-            Some(data) => {
-                let mut content = Vec::with_capacity(
-                    usize::try_from(raptorq.transfer_length)
-                        .map_err(|e| Error::Other(format!("transfer_length: {e}")))?,
-                );
-                content.extend_from_slice(&client_id.to_le_bytes());
-                content.push(block.serialized());
-                content.extend_from_slice(&u32::to_le_bytes(
-                    u32::try_from(data.len())
-                        .map_err(|e| Error::Other(format!("data.len(): {e}")))?,
-                ));
-                content.extend_from_slice(data);
-                if content.len() < content.capacity() {
-                    content.resize(content.capacity(), 0);
-                }
-                Ok(Self(content))
-            }
+        let mut content = vec![
+            0u8;
+            usize::try_from(raptorq.transfer_length)
+                .map_err(|e| Error::Other(format!("transfer_length: {e}")))?
+        ];
+        content[0..4].copy_from_slice(&client_id.to_le_bytes());
+        content[4] = block.serialized();
+
+        if let Some(data) = data {
+            let data_len = data.len();
+            content[5..9].copy_from_slice(&u32::to_le_bytes(
+                u32::try_from(data_len).map_err(|e| Error::Other(format!("data.len(): {e}")))?,
+            ));
+            content[9..9 + data_len].copy_from_slice(data);
         }
+
+        Ok(Self(content))
     }
 
     pub(crate) fn client_id(&self) -> ClientId {
-        let bytes = [self.0[0], self.0[1], self.0[2], self.0[3]];
-        u32::from_le_bytes(bytes)
+        u32::from_le_bytes([self.0[0], self.0[1], self.0[2], self.0[3]])
     }
 
     pub(crate) fn block_type(&self) -> Result<BlockType, Error> {
@@ -310,8 +292,7 @@ impl Block {
     }
 
     fn payload_len(&self) -> u32 {
-        let data_len_bytes = [self.0[5], self.0[6], self.0[7], self.0[8]];
-        u32::from_le_bytes(data_len_bytes)
+        u32::from_le_bytes([self.0[5], self.0[6], self.0[7], self.0[8]])
     }
 
     pub(crate) const fn deserialize(data: Vec<u8>) -> Self {

src/receive/mod.rs

@@ -187,6 +187,10 @@ where
         let multiplex_control = semka::Sem::new(config.max_clients)
             .ok_or(Error::Other("failed to create semaphore".into()))?;
 
+        if config.from_mtu > crate::MAX_MTU {
+            return Err(Error::Other(format!("mtu {} is too large (> {})", config.from_mtu, crate::MAX_MTU)));
+        }
+
         let block_to_dispatch = (sync::Mutex::new(0), sync::Condvar::new());
 
         let (to_reblock, for_reblock) = crossbeam_channel::unbounded();

src/send/mod.rs

@@ -141,6 +141,10 @@ where
         let multiplex_control = semka::Sem::new(config.max_clients)
             .ok_or(Error::Other("failed to create semaphore".into()))?;
 
+        if config.to_mtu > crate::MAX_MTU {
+            return Err(Error::Other(format!("mtu {} is too large (> {})", config.to_mtu, crate::MAX_MTU)));
+        }
+
         let (to_server, for_server) = crossbeam_channel::bounded(1);
         let (to_ordering, for_ordering) = crossbeam_channel::bounded(config.to_ports.len());
         let (to_udp, for_udp) = crossbeam_channel::bounded(config.to_ports.len());

src/udp.rs

@@ -122,17 +122,16 @@ pub struct ReceiveMmsg<'a> {
 #[cfg(feature = "receive-mmsg")]
 impl ReceiveMmsg<'_> {
     fn new(socket: i32, udp_packet_size: u16) -> Self {
-        let batch_size = MAX_BATCH_SIZE as usize;
-        let iovecs = vec![unsafe { mem::zeroed::<libc::iovec>() }; batch_size];
+        let iovecs = vec![unsafe { mem::zeroed::<libc::iovec>() }; MAX_BATCH_SIZE as usize];
         let mut iovecs = pin::Pin::new(iovecs);
 
-        let mut mmsghdr = vec![unsafe { mem::zeroed::<libc::mmsghdr>() }; batch_size];
-        for i in 0..batch_size {
+        let mut mmsghdr = vec![unsafe { mem::zeroed::<libc::mmsghdr>() }; MAX_BATCH_SIZE as usize];
+        for i in 0..MAX_BATCH_SIZE as usize {
             mmsghdr[i].msg_hdr.msg_iov = &raw mut iovecs[i];
             mmsghdr[i].msg_hdr.msg_iovlen = 1;
         }
 
-        let mut buffers = vec![pin::Pin::new(vec![0u8; udp_packet_size as usize]); batch_size];
+        let mut buffers = vec![pin::Pin::new(vec![0u8; udp_packet_size as usize]); MAX_BATCH_SIZE as usize];
 
         for (i, buffer) in buffers.iter_mut().enumerate() {
             iovecs[i].iov_base = buffer.as_mut_ptr().cast::<libc::c_void>();
@@ -339,13 +338,12 @@ impl<'a> Send<'a> {
                     io::Error::new(io::ErrorKind::InvalidData, format!("dest_len: {e}"))
                 })?;
 
-                let batch_size = MAX_BATCH_SIZE as usize;
-                let iovecs = vec![unsafe { mem::zeroed::<libc::iovec>() }; batch_size];
+                let iovecs = vec![unsafe { mem::zeroed::<libc::iovec>() }; MAX_BATCH_SIZE as usize];
                 let mut iovecs = pin::Pin::new(iovecs);
 
-                let mut mmsghdr = vec![unsafe { mem::zeroed::<libc::mmsghdr>() }; batch_size];
+                let mut mmsghdr = vec![unsafe { mem::zeroed::<libc::mmsghdr>() }; MAX_BATCH_SIZE as usize];
 
-                for i in 0..batch_size {
+                for i in 0..MAX_BATCH_SIZE as usize {
                     mmsghdr[i].msg_hdr.msg_name = raw_dest.cast::<libc::c_void>();
                     mmsghdr[i].msg_hdr.msg_namelen = dest_len;
                     mmsghdr[i].msg_hdr.msg_iov = &raw mut iovecs[i];
@@ -364,72 +362,69 @@ impl<'a> Send<'a> {
     }
 
     pub fn send(&mut self, packets: &[raptorq::EncodingPacket]) -> Result<(), io::Error> {
-        let mut datagrams = packets.iter().map(raptorq::EncodingPacket::serialize);
+        let datagrams = packets.iter().map(raptorq::EncodingPacket::serialize);
 
         match self {
             #[cfg(feature = "send-native")]
-            Self::Native { socket, dest } => datagrams.try_for_each(|datagram| {
-                let len = datagram.len();
+            Self::Native { socket, dest } => {
+                for datagram in datagrams {
+                    let len = datagram.len();
 
-                let sent = socket.send_to(&datagram, *dest)?;
+                    let sent = socket.send_to(&datagram, *dest)?;
 
-                if sent == len {
-                    Ok(())
-                } else {
-                    Err(io::Error::other(format!(
-                        "libc::sendmsg failed {sent} != {len}"
-                    )))
+                    if sent != len {
+                        return Err(io::Error::other(format!(
+                            "libc::sendmsg failed {sent} != {len}"
+                        )));
+                    }
                 }
-            }),
+            }
             #[cfg(feature = "send-msg")]
             Self::Msg {
                 socket,
                 msghdr,
                 iovec,
                 ..
-            } => datagrams.try_for_each(|mut datagram| {
-                let len = datagram.len();
+            } => {
+                for mut datagram in datagrams {
+                    let len = datagram.len();
 
-                iovec.iov_base = datagram.as_mut_ptr().cast();
-                iovec.iov_len = len;
+                    iovec.iov_base = datagram.as_mut_ptr().cast();
+                    iovec.iov_len = len;
 
-                let sent = unsafe { libc::sendmsg(*socket, msghdr, 0) };
+                    let sent = unsafe { libc::sendmsg(*socket, msghdr, 0) };
 
-                if sent == len.cast_signed() {
-                    Ok(())
-                } else {
-                    Err(io::Error::other(format!(
-                        "libc::sendmsg failed {sent} != {len}"
-                    )))
+                    if sent != len.cast_signed() {
+                        return Err(io::Error::other(format!(
+                            "libc::sendmsg failed {sent} != {len}"
+                        )));
+                    }
                 }
-            }),
+            }
             #[cfg(feature = "send-mmsg")]
             Self::Mmsg {
                 socket,
                 mmsghdr,
                 iovecs,
                 ..
-            } => datagrams
-                .collect::<Vec<_>>()
-                .chunks_mut(MAX_BATCH_SIZE as usize)
-                .try_for_each(|datagrams| {
+            } => {
+                for datagrams in datagrams
+                    .collect::<Vec<_>>()
+                    .chunks_mut(MAX_BATCH_SIZE as usize)
+                {
                     let to_send = datagrams.len();
 
-                    datagrams
-                        .iter_mut()
-                        .enumerate()
-                        .try_for_each(|(i, datagram)| {
-                            mmsghdr[i].msg_len = u32::try_from(datagram.len())?;
-                            iovecs[i].iov_base = datagram.as_mut_ptr().cast::<libc::c_void>();
-                            iovecs[i].iov_len = datagram.len();
-                            Ok(())
-                        })
-                        .map_err(|e: num::TryFromIntError| {
-                            io::Error::new(
-                                io::ErrorKind::InvalidData,
-                                format!("datagram.len(): {e}"),
-                            )
-                        })?;
+                    for (i, datagram) in datagrams.iter_mut().enumerate() {
+                        mmsghdr[i].msg_len =
+                            u32::try_from(datagram.len()).map_err(|e: num::TryFromIntError| {
+                                io::Error::new(
+                                    io::ErrorKind::InvalidData,
+                                    format!("datagram.len(): {e}"),
+                                )
+                            })?;
+                        iovecs[i].iov_base = datagram.as_mut_ptr().cast::<libc::c_void>();
+                        iovecs[i].iov_len = datagram.len();
+                    }
 
                     let sent = unsafe {
                         libc::sendmmsg(
@@ -442,12 +437,13 @@ impl<'a> Send<'a> {
                         ) as isize
                     };
 
-                    if sent.cast_unsigned() == to_send {
-                        Ok(())
-                    } else {
-                        Err(io::Error::other("libc::sendmmsg"))
+                    if sent.cast_unsigned() != to_send {
+                        return Err(io::Error::other("libc::sendmmsg"));
                     }
-                }),
+                }
+            }
         }
+
+        Ok(())
     }
 }