renet
/
renet
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Use separate metadata and payload buffers for UDP sockets. 

Co-authored-by: Dan Robertson <danlrobertson89@gmail.com>
v0.7.x
Philipp Oppermann 2018-02-22 07:33:11 +01:00 committed by whitequark
parent 785637957c
commit ed2dcaaff9
5 changed files with 328 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
examples/server.rs
View File

@ -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]);

6
src/iface/ethernet.rs
View File

@ -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);

2
src/socket/mod.rs
View File

@ -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};

372
src/socket/udp.rs
View File

@ -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<T>(payload: T) -> PacketBuffer<'a>
where T: Into<Managed<'a, [u8]>> {
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<MS, PS>(metadata_storage: MS, payload_storage: PS) -> SocketBuffer<'a, 'b>
where MS: Into<ManagedSlice<'a, PacketMetadata>>, PS: Into<ManagedSlice<'b, u8>>,
{
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)));
}
}

57
src/storage/ring_buffer.rs
View File

@ -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<R> {
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<R> {
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<u8> = 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);
}
}