From ed2dcaaff9b7bfabcba1a4ed6526d80d306c9b8e Mon Sep 17 00:00:00 2001 From: Philipp Oppermann Date: Thu, 22 Feb 2018 07:33:11 +0100 Subject: [PATCH] Use separate metadata and payload buffers for UDP sockets. Co-authored-by: Dan Robertson --- examples/server.rs | 6 +- src/iface/ethernet.rs | 6 +- src/socket/mod.rs | 2 +- src/socket/udp.rs | 372 +++++++++++++++++++++++++++---------- src/storage/ring_buffer.rs | 57 +++++- 5 files changed, 328 insertions(+), 115 deletions(-) diff --git a/examples/server.rs b/examples/server.rs index aa5f91c..0f7229a 100644 --- a/examples/server.rs +++ b/examples/server.rs @@ -14,7 +14,7 @@ use smoltcp::phy::wait as phy_wait; use smoltcp::wire::{EthernetAddress, IpAddress, IpCidr}; use smoltcp::iface::{NeighborCache, EthernetInterfaceBuilder}; use smoltcp::socket::SocketSet; -use smoltcp::socket::{UdpSocket, UdpSocketBuffer, UdpPacketBuffer}; +use smoltcp::socket::{UdpSocket, UdpSocketBuffer, UdpPacketMetadata}; use smoltcp::socket::{TcpSocket, TcpSocketBuffer}; use smoltcp::time::{Duration, Instant}; @@ -32,8 +32,8 @@ fn main() { let neighbor_cache = NeighborCache::new(BTreeMap::new()); - let udp_rx_buffer = UdpSocketBuffer::new(vec![UdpPacketBuffer::new(vec![0; 64])]); - let udp_tx_buffer = UdpSocketBuffer::new(vec![UdpPacketBuffer::new(vec![0; 128])]); + let udp_rx_buffer = UdpSocketBuffer::new(vec![UdpPacketMetadata::default()], vec![0; 64]); + let udp_tx_buffer = UdpSocketBuffer::new(vec![UdpPacketMetadata::default()], vec![0; 128]); let udp_socket = UdpSocket::new(udp_rx_buffer, udp_tx_buffer); let tcp1_rx_buffer = TcpSocketBuffer::new(vec![0; 64]); diff --git a/src/iface/ethernet.rs b/src/iface/ethernet.rs index 0a82adc..09dc13a 100644 --- a/src/iface/ethernet.rs +++ b/src/iface/ethernet.rs @@ -1365,15 +1365,15 @@ mod test { #[test] #[cfg(all(feature = "socket-udp", feature = "proto-ipv4"))] fn test_handle_udp_broadcast() { - use socket::{UdpPacketBuffer, UdpSocket, UdpSocketBuffer}; + use socket::{UdpSocket, UdpSocketBuffer}; use wire::IpEndpoint; static UDP_PAYLOAD: [u8; 5] = [0x48, 0x65, 0x6c, 0x6c, 0x6f]; let (iface, mut socket_set) = create_loopback(); - let rx_buffer = UdpSocketBuffer::new(vec![UdpPacketBuffer::new(vec![0; 15])]); - let tx_buffer = UdpSocketBuffer::new(vec![UdpPacketBuffer::new(vec![0; 15])]); + let rx_buffer = UdpSocketBuffer::new(vec![Default::default()], vec![0; 15]); + let tx_buffer = UdpSocketBuffer::new(vec![Default::default()], vec![0; 15]); let udp_socket = UdpSocket::new(rx_buffer, tx_buffer); diff --git a/src/socket/mod.rs b/src/socket/mod.rs index 7ae2b99..32a3e74 100644 --- a/src/socket/mod.rs +++ b/src/socket/mod.rs @@ -40,7 +40,7 @@ pub use self::icmp::{PacketBuffer as IcmpPacketBuffer, IcmpSocket}; #[cfg(feature = "socket-udp")] -pub use self::udp::{PacketBuffer as UdpPacketBuffer, +pub use self::udp::{PacketMetadata as UdpPacketMetadata, SocketBuffer as UdpSocketBuffer, UdpSocket}; diff --git a/src/socket/udp.rs b/src/socket/udp.rs index 3637fcb..f3657e4 100644 --- a/src/socket/udp.rs +++ b/src/socket/udp.rs @@ -1,58 +1,79 @@ use core::cmp::min; -use managed::Managed; +use managed::ManagedSlice; use {Error, Result}; use socket::{Socket, SocketMeta, SocketHandle}; -use storage::{Resettable, RingBuffer}; +use storage::RingBuffer; use time::Instant; use wire::{IpProtocol, IpRepr, IpEndpoint, UdpRepr}; -/// A buffered UDP packet. -#[derive(Debug)] -pub struct PacketBuffer<'a> { +/// Endpoint and size of an UDP packet. +#[derive(Debug, Clone, Copy, Default)] +pub struct PacketMetadata { endpoint: IpEndpoint, - size: usize, - payload: Managed<'a, [u8]> -} - -impl<'a> PacketBuffer<'a> { - /// Create a buffered packet. - pub fn new(payload: T) -> PacketBuffer<'a> - where T: Into> { - PacketBuffer { - endpoint: IpEndpoint::default(), - size: 0, - payload: payload.into() - } - } - - fn as_ref<'b>(&'b self) -> &'b [u8] { - &self.payload[..self.size] - } - - fn as_mut<'b>(&'b mut self) -> &'b mut [u8] { - &mut self.payload[..self.size] - } - - fn resize<'b>(&'b mut self, size: usize) -> Result<&'b mut Self> { - if self.payload.len() >= size { - self.size = size; - Ok(self) - } else { - Err(Error::Truncated) - } - } -} - -impl<'a> Resettable for PacketBuffer<'a> { - fn reset(&mut self) { - self.endpoint = Default::default(); - self.size = 0; - } + size: usize, + /// Padding packets can be used to avoid wrap-arounds of packets in the payload buffer + padding: bool, } /// An UDP packet ring buffer. -pub type SocketBuffer<'a, 'b: 'a> = RingBuffer<'a, PacketBuffer<'b>>; +#[derive(Debug)] +pub struct SocketBuffer<'a, 'b> { + metadata_buffer: RingBuffer<'a, PacketMetadata>, + payload_buffer: RingBuffer<'b, u8>, +} + +impl<'a, 'b> SocketBuffer<'a, 'b> { + /// Create a new socket buffer with the provided metadata and payload storage. + /// + /// Metadata storage limits the maximum _number_ of UDP packets in the buffer and payload + /// storage limits the maximum _cumulated size_ of UDP packets. + pub fn new(metadata_storage: MS, payload_storage: PS) -> SocketBuffer<'a, 'b> + where MS: Into>, PS: Into>, + { + SocketBuffer { + metadata_buffer: RingBuffer::new(metadata_storage), + payload_buffer: RingBuffer::new(payload_storage), + } + } + + fn is_full(&self) -> bool { + self.metadata_buffer.is_full() || self.payload_buffer.is_full() + } + + fn is_empty(&self) -> bool { + self.metadata_buffer.is_empty() + } + + fn enqueue(&mut self, required_size: usize, endpoint: IpEndpoint) -> Result<&mut [u8]> { + let window = self.payload_buffer.window(); + let contig_window = self.payload_buffer.contiguous_window(); + + if self.metadata_buffer.is_full() || self.payload_buffer.window() < required_size { + return Err(Error::Exhausted); + } + + if contig_window < required_size { + // we reached the end of buffer, so the data does not fit without wrap-around + // -> insert padding and try again + self.payload_buffer.enqueue_many(required_size); + let metadata_buf = self.metadata_buffer.enqueue_one()?; + metadata_buf.padding = true; + metadata_buf.size = required_size; + metadata_buf.endpoint = IpEndpoint::default(); + if window - contig_window < required_size { + return Err(Error::Exhausted); + } + } + + let metadata_buf = self.metadata_buffer.enqueue_one()?; + metadata_buf.endpoint = endpoint; + metadata_buf.size = required_size; + metadata_buf.padding = false; + + Ok(self.payload_buffer.enqueue_many(required_size)) + } +} /// An User Datagram Protocol socket. /// @@ -156,19 +177,19 @@ impl<'a, 'b> UdpSocket<'a, 'b> { /// Enqueue a packet to be sent to a given remote endpoint, and return a pointer /// to its payload. /// - /// This function returns `Err(Error::Exhausted)` if the transmit buffer is full, - /// `Err(Error::Truncated)` if the requested size is larger than the packet buffer - /// size, and `Err(Error::Unaddressable)` if local or remote port, or remote address, - /// are unspecified. + /// This function returns `Err(Error::Exhausted)` if the transmit buffer is full and + /// `Err(Error::Unaddressable)` if local or remote port, or remote address are unspecified. pub fn send(&mut self, size: usize, endpoint: IpEndpoint) -> Result<&mut [u8]> { if self.endpoint.port == 0 { return Err(Error::Unaddressable) } if !endpoint.is_specified() { return Err(Error::Unaddressable) } - let packet_buf = self.tx_buffer.enqueue_one_with(|buf| buf.resize(size))?; - packet_buf.endpoint = endpoint; + let payload_buf = self.tx_buffer.enqueue(size, endpoint)?; + + debug_assert_eq!(payload_buf.len(), size); + net_trace!("{}:{}:{}: buffer to send {} octets", - self.meta.handle, self.endpoint, packet_buf.endpoint, size); - Ok(&mut packet_buf.as_mut()[..size]) + self.meta.handle, self.endpoint, endpoint, size); + Ok(payload_buf) } /// Enqueue a packet to be sent to a given remote endpoint, and fill it from a slice. @@ -184,11 +205,21 @@ impl<'a, 'b> UdpSocket<'a, 'b> { /// /// This function returns `Err(Error::Exhausted)` if the receive buffer is empty. pub fn recv(&mut self) -> Result<(&[u8], IpEndpoint)> { - let packet_buf = self.rx_buffer.dequeue_one()?; + let mut metadata_buf = *self.rx_buffer.metadata_buffer.dequeue_one()?; + if metadata_buf.padding { + // packet is padding packet -> drop it and try again + self.rx_buffer.payload_buffer.dequeue_many(metadata_buf.size); + metadata_buf = *self.rx_buffer.metadata_buffer.dequeue_one()?; + } + + debug_assert!(!metadata_buf.padding); + let payload_buf = self.rx_buffer.payload_buffer.dequeue_many(metadata_buf.size); + debug_assert_eq!(metadata_buf.size, payload_buf.len()); // ensured by inserting logic + net_trace!("{}:{}:{}: receive {} buffered octets", self.meta.handle, self.endpoint, - packet_buf.endpoint, packet_buf.size); - Ok((&packet_buf.as_ref(), packet_buf.endpoint)) + metadata_buf.endpoint, metadata_buf.size); + Ok((payload_buf, metadata_buf.endpoint)) } /// Dequeue a packet received from a remote endpoint, copy the payload into the given slice, @@ -213,12 +244,16 @@ impl<'a, 'b> UdpSocket<'a, 'b> { pub(crate) fn process(&mut self, ip_repr: &IpRepr, repr: &UdpRepr) -> Result<()> { debug_assert!(self.accepts(ip_repr, repr)); - let packet_buf = self.rx_buffer.enqueue_one_with(|buf| buf.resize(repr.payload.len()))?; - packet_buf.as_mut().copy_from_slice(repr.payload); - packet_buf.endpoint = IpEndpoint { addr: ip_repr.src_addr(), port: repr.src_port }; + let size = repr.payload.len(); + + let endpoint = IpEndpoint { addr: ip_repr.src_addr(), port: repr.src_port }; + let payload_buf = self.rx_buffer.enqueue(size, endpoint)?; + assert_eq!(payload_buf.len(), size); + payload_buf.copy_from_slice(repr.payload); + net_trace!("{}:{}:{}: receiving {} octets", self.meta.handle, self.endpoint, - packet_buf.endpoint, packet_buf.size); + endpoint, size); Ok(()) } @@ -227,24 +262,47 @@ impl<'a, 'b> UdpSocket<'a, 'b> { let handle = self.handle(); let endpoint = self.endpoint; let hop_limit = self.hop_limit.unwrap_or(64); - self.tx_buffer.dequeue_one_with(|packet_buf| { - net_trace!("{}:{}:{}: sending {} octets", - handle, endpoint, - packet_buf.endpoint, packet_buf.size); - let repr = UdpRepr { - src_port: endpoint.port, - dst_port: packet_buf.endpoint.port, - payload: &packet_buf.as_ref()[..] - }; - let ip_repr = IpRepr::Unspecified { - src_addr: endpoint.addr, - dst_addr: packet_buf.endpoint.addr, - protocol: IpProtocol::Udp, - payload_len: repr.buffer_len(), - hop_limit: hop_limit, - }; - emit((ip_repr, repr)) + let SocketBuffer { ref mut metadata_buffer, ref mut payload_buffer } = self.tx_buffer; + + // dequeue potential padding packet + let result = metadata_buffer.dequeue_one_with(|metadata_buf| { + if metadata_buf.padding { + Ok(metadata_buf.size) // dequeue metadata + } else { + Err(Error::Exhausted) // don't dequeue metadata + } + }); + if let Ok(size) = result { + payload_buffer.dequeue_many(size); // dequeue padding payload + } + + metadata_buffer.dequeue_one_with(move |metadata_buf| { + debug_assert!(!metadata_buf.padding); + payload_buffer.dequeue_many_with(|payload_buf| { + let payload_buf = &payload_buf[..metadata_buf.size]; + + net_trace!("{}:{}:{}: sending {} octets", + handle, endpoint, + metadata_buf.endpoint, metadata_buf.size); + + let repr = UdpRepr { + src_port: endpoint.port, + dst_port: metadata_buf.endpoint.port, + payload: payload_buf, + }; + let ip_repr = IpRepr::Unspecified { + src_addr: endpoint.addr, + dst_addr: metadata_buf.endpoint.addr, + protocol: IpProtocol::Udp, + payload_len: repr.buffer_len(), + hop_limit: hop_limit, + }; + match emit((ip_repr, repr)) { + Ok(ret) => (metadata_buf.size, Ok(ret)), + Err(ret) => (0, Err(ret)), + } + }).1 }) } @@ -274,11 +332,7 @@ mod test { use super::*; fn buffer(packets: usize) -> SocketBuffer<'static, 'static> { - let mut storage = vec![]; - for _ in 0..packets { - storage.push(PacketBuffer::new(vec![0; 16])) - } - SocketBuffer::new(storage) + SocketBuffer::new(vec![Default::default(); packets], vec![0; 16 * packets]) } fn socket(rx_buffer: SocketBuffer<'static, 'static>, @@ -369,14 +423,6 @@ mod test { assert_eq!(socket.send_slice(b"abcdef", REMOTE_END), Ok(())); } - #[test] - fn test_send_truncated() { - let mut socket = socket(buffer(0), buffer(1)); - assert_eq!(socket.bind(LOCAL_END), Ok(())); - - assert_eq!(socket.send_slice(&[0; 32][..], REMOTE_END), Err(Error::Truncated)); - } - #[test] fn test_send_dispatch() { let mut socket = socket(buffer(0), buffer(1)); @@ -439,17 +485,6 @@ mod test { assert_eq!(&slice, b"abcd"); } - #[test] - fn test_recv_truncated_packet() { - let mut socket = socket(buffer(1), buffer(0)); - assert_eq!(socket.bind(LOCAL_PORT), Ok(())); - - let udp_repr = UdpRepr { payload: &[0; 100][..], ..REMOTE_UDP_REPR }; - assert!(socket.accepts(&remote_ip_repr(), &udp_repr)); - assert_eq!(socket.process(&remote_ip_repr(), &udp_repr), - Err(Error::Truncated)); - } - #[test] fn test_set_hop_limit() { let mut s = socket(buffer(0), buffer(1)); @@ -512,4 +547,141 @@ mod test { assert_eq!(ip_bound_socket.bind(LOCAL_END), Ok(())); assert!(!ip_bound_socket.accepts(&generate_bad_repr(), &REMOTE_UDP_REPR)); } + + #[test] + fn test_send_large_packet() { + // buffer(4) creates a payload buffer of size 16*4 + let mut socket = socket(buffer(0), buffer(4)); + assert_eq!(socket.bind(LOCAL_END), Ok(())); + + let too_large = b"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdefx"; + assert_eq!(socket.send_slice(too_large, REMOTE_END), Err(Error::Exhausted)); + assert_eq!(socket.send_slice(&too_large[..16*4], REMOTE_END), Ok(())); + } + + #[test] + fn test_send_wraparound_1() { + let mut socket = socket(buffer(0), buffer(3)); + assert_eq!(socket.bind(LOCAL_END), Ok(())); + + let large = b"0123456789abcdef0123456789abcdef0123456789abcdef"; + + assert_eq!(socket.send_slice(&large[..15], REMOTE_END), Ok(())); + assert_eq!(socket.send_slice(&large[..16*2], REMOTE_END), Ok(())); + // no padding should be inserted because capacity does not suffice + assert_eq!(socket.send_slice(b"12", REMOTE_END), Err(Error::Exhausted)); + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 2); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 16*3-1); + + assert_eq!(socket.dispatch(|_| Ok(())), Ok(())); + // insert padding + assert_eq!(socket.send_slice(&large[..16], REMOTE_END), Err(Error::Exhausted)); + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 2); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 16*3-15); + + assert_eq!(socket.dispatch(|_| Ok(())), Ok(())); + // packet dequed, but padding is still there + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 1); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 1); + + assert_eq!(socket.dispatch(|_| Ok(())), Err(Error::Exhausted)); + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 0); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 0); + } + + #[test] + fn test_send_wraparound_2() { + let mut socket = socket(buffer(0), buffer(3)); + assert_eq!(socket.bind(LOCAL_END), Ok(())); + + let large = b"0123456789abcdef0123456789abcdef0123456789abcdef"; + + assert_eq!(socket.send_slice(&large[..16*2], REMOTE_END), Ok(())); + assert_eq!(socket.send_slice(&large[..15], REMOTE_END), Ok(())); + // no padding should be inserted because capacity does not suffice + assert_eq!(socket.send_slice(b"12", REMOTE_END), Err(Error::Exhausted)); + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 2); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 16*3-1); + + assert_eq!(socket.dispatch(|_| Ok(())), Ok(())); + // insert padding and slice + assert_eq!(socket.send_slice(&large[..16*2], REMOTE_END), Ok(())); + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 3); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 16*3); + + assert_eq!(socket.dispatch(|_| Ok(())), Ok(())); + // packet dequed, but padding is still there + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 2); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 16*3-15); + + assert_eq!(socket.dispatch(|_| Ok(())), Ok(())); + // padding and packet dequeued + assert_eq!(socket.tx_buffer.metadata_buffer.len(), 0); + assert_eq!(socket.tx_buffer.payload_buffer.len(), 0); + } + + #[test] + fn test_process_wraparound() { + // every packet will be 6 bytes + let recv_buffer = SocketBuffer::new(vec![Default::default(); 4], vec![0; 6*3 + 2]); + let mut socket = socket(recv_buffer, buffer(0)); + assert_eq!(socket.bind(LOCAL_PORT), Ok(())); + + assert_eq!(socket.process(&remote_ip_repr(), &REMOTE_UDP_REPR), Ok(())); + assert_eq!(socket.process(&remote_ip_repr(), &REMOTE_UDP_REPR), Ok(())); + assert_eq!(socket.process(&remote_ip_repr(), &REMOTE_UDP_REPR), Ok(())); + assert_eq!(socket.rx_buffer.metadata_buffer.len(), 3); + assert_eq!(socket.rx_buffer.payload_buffer.len(), 6*3); + + assert_eq!(socket.process(&remote_ip_repr(), &REMOTE_UDP_REPR), + Err(Error::Exhausted)); + // no padding inserted because capacity does not suffice + assert_eq!(socket.rx_buffer.metadata_buffer.len(), 3); + assert_eq!(socket.rx_buffer.payload_buffer.len(), 6*3); + + assert_eq!(socket.recv(), Ok((&b"abcdef"[..], REMOTE_END))); + assert_eq!(socket.process(&remote_ip_repr(), &REMOTE_UDP_REPR), Ok(())); + // padding inserted + assert_eq!(socket.rx_buffer.metadata_buffer.len(), 4); + assert_eq!(socket.rx_buffer.payload_buffer.len(), 6*3 + 2); + + assert_eq!(socket.recv(), Ok((&b"abcdef"[..], REMOTE_END))); + assert_eq!(socket.recv(), Ok((&b"abcdef"[..], REMOTE_END))); + // two packets dequed, last packet and padding still there + assert_eq!(socket.rx_buffer.metadata_buffer.len(), 2); + assert_eq!(socket.rx_buffer.payload_buffer.len(), 6 + 2); + + assert_eq!(socket.recv(), Ok((&b"abcdef"[..], REMOTE_END))); + // everything dequed + assert_eq!(socket.rx_buffer.metadata_buffer.len(), 0); + assert_eq!(socket.rx_buffer.payload_buffer.len(), 0); + } + + #[test] + fn test_process_empty_payload() { + // every packet will be 6 bytes + let recv_buffer = SocketBuffer::new(vec![Default::default(); 1], vec![]); + let mut socket = socket(recv_buffer, buffer(0)); + assert_eq!(socket.bind(LOCAL_PORT), Ok(())); + + let repr = UdpRepr { + src_port: REMOTE_PORT, + dst_port: LOCAL_PORT, + payload: &[] + }; + + assert_eq!(socket.process(&remote_ip_repr(), &repr), Ok(())); + assert_eq!(socket.rx_buffer.metadata_buffer.len(), 1); + assert_eq!(socket.rx_buffer.payload_buffer.len(), 0); + + // The metatdata has been queued into the metadata buffer + assert!(!socket.rx_buffer.metadata_buffer.is_empty()); + // The no payload data has been queued into the payload buffer + assert!(socket.rx_buffer.payload_buffer.is_empty()); + // The received packets buffer is not empty and we can recv + assert!(socket.can_recv()); + assert_eq!(socket.recv(), Ok((&[][..], REMOTE_END))); + assert_eq!(socket.process(&remote_ip_repr(), &repr), Ok(())); + assert_eq!(socket.recv(), Ok((&[][..], REMOTE_END))); + } } diff --git a/src/storage/ring_buffer.rs b/src/storage/ring_buffer.rs index 67f6d66..d6df51c 100644 --- a/src/storage/ring_buffer.rs +++ b/src/storage/ring_buffer.rs @@ -72,6 +72,12 @@ impl<'a, T: 'a> RingBuffer<'a, T> { self.capacity() - self.len() } + /// Return the largest number of elements that can be added to the buffer + /// without wrapping around (i.e. in a single `enqueue_many` call). + pub fn contiguous_window(&self) -> usize { + cmp::min(self.window(), self.capacity() - self.get_idx(self.length)) + } + /// Query whether the buffer is empty. pub fn is_empty(&self) -> bool { self.len() == 0 @@ -81,6 +87,23 @@ impl<'a, T: 'a> RingBuffer<'a, T> { pub fn is_full(&self) -> bool { self.window() == 0 } + + /// Shorthand for `(self.read + idx) % self.capacity()` with an + /// additional check to ensure that the capacity is not zero. + fn get_idx(&self, idx: usize) -> usize { + let len = self.capacity(); + if len > 0 { + (self.read_at + idx) % len + } else { + 0 + } + } + + /// Shorthand for `(self.read + idx) % self.capacity()` with no + /// additional checks to ensure the capacity is not zero. + fn get_idx_unchecked(&self, idx: usize) -> usize { + (self.read_at + idx) % self.capacity() + } } /// This is the "discrete" ring buffer interface: it operates with single elements, @@ -92,7 +115,7 @@ impl<'a, T: 'a> RingBuffer<'a, T> { where F: FnOnce(&'b mut T) -> Result { if self.is_full() { return Err(Error::Exhausted) } - let index = (self.read_at + self.length) % self.capacity(); + let index = self.get_idx_unchecked(self.length); match f(&mut self.storage[index]) { Ok(result) => { self.length += 1; @@ -116,7 +139,7 @@ impl<'a, T: 'a> RingBuffer<'a, T> { where F: FnOnce(&'b mut T) -> Result { if self.is_empty() { return Err(Error::Exhausted) } - let next_at = (self.read_at + 1) % self.capacity(); + let next_at = self.get_idx_unchecked(1); match f(&mut self.storage[self.read_at]) { Ok(result) => { self.length -= 1; @@ -147,8 +170,8 @@ impl<'a, T: 'a> RingBuffer<'a, T> { /// than the size of the slice passed into it. pub fn enqueue_many_with<'b, R, F>(&'b mut self, f: F) -> (usize, R) where F: FnOnce(&'b mut [T]) -> (usize, R) { - let write_at = (self.read_at + self.length) % self.capacity(); - let max_size = cmp::min(self.window(), self.capacity() - write_at); + let write_at = self.get_idx(self.length); + let max_size = self.contiguous_window(); let (size, result) = f(&mut self.storage[write_at..write_at + max_size]); assert!(size <= max_size); self.length += size; @@ -198,7 +221,11 @@ impl<'a, T: 'a> RingBuffer<'a, T> { let max_size = cmp::min(self.len(), capacity - self.read_at); let (size, result) = f(&mut self.storage[self.read_at..self.read_at + max_size]); assert!(size <= max_size); - self.read_at = (self.read_at + size) % capacity; + self.read_at = if capacity > 0 { + (self.read_at + size) % capacity + } else { + 0 + }; self.length -= size; (size, result) } @@ -242,7 +269,7 @@ impl<'a, T: 'a> RingBuffer<'a, T> { /// at the given offset past the last allocated element, and up to the given size. // #[must_use] pub fn get_unallocated(&mut self, offset: usize, mut size: usize) -> &mut [T] { - let start_at = (self.read_at + self.length + offset) % self.capacity(); + let start_at = self.get_idx(self.length + offset); // We can't access past the end of unallocated data. if offset > self.window() { return &mut [] } // We can't enqueue more than there is free space. @@ -289,7 +316,7 @@ impl<'a, T: 'a> RingBuffer<'a, T> { /// at the given offset past the first allocated element, and up to the given size. // #[must_use] pub fn get_allocated(&self, offset: usize, mut size: usize) -> &[T] { - let start_at = (self.read_at + offset) % self.capacity(); + let start_at = self.get_idx(offset); // We can't read past the end of the allocated data. if offset > self.length { return &mut [] } // We can't read more than we have allocated. @@ -328,7 +355,7 @@ impl<'a, T: 'a> RingBuffer<'a, T> { pub fn dequeue_allocated(&mut self, count: usize) { assert!(count <= self.len()); self.length -= count; - self.read_at = (self.read_at + count) % self.capacity(); + self.read_at = self.get_idx(count); } } @@ -680,4 +707,18 @@ mod test { assert_eq!(&data[..], b"mno\x00\x00\x00"); } + + #[test] + fn test_buffer_with_no_capacity() { + let mut no_capacity: RingBuffer = RingBuffer::new(vec![]); + + // Call all functions that calculate the remainder against rx_buffer.capacity() + // with a backing storage with a length of 0. + assert_eq!(no_capacity.get_unallocated(0, 0), &[]); + assert_eq!(no_capacity.get_allocated(0, 0), &[]); + no_capacity.dequeue_allocated(0); + assert_eq!(no_capacity.enqueue_many(0), &[]); + assert_eq!(no_capacity.enqueue_one(), Err(Error::Exhausted)); + assert_eq!(no_capacity.contiguous_window(), 0); + } }