From 5adde38a65fb13871e2c397de169f58e2918fe51 Mon Sep 17 00:00:00 2001 From: Sebastian Thiel Date: Sat, 11 Feb 2017 09:01:07 +0100 Subject: [PATCH 1/3] Allocation in get_mut() always uses 8 * 1024 bytes The previous implementation would always use the capacity of the previous buffer, which would effectively prevent it from ever shrinking. This also means that protocol with greater variance in possible frame sizes would likely be heavily over-allocating. If these implementations use zero-copy, this would imply that even small frames kept alive by the client would use large amounts of memory. The change is motivated by the implementation of the cassandra-protocol, which allows frames of up to 256MB in size, which solely depend on the kind of query. --- src/io/frame.rs | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/src/io/frame.rs b/src/io/frame.rs index 479433723..44fa5a1b9 100644 --- a/src/io/frame.rs +++ b/src/io/frame.rs @@ -8,6 +8,8 @@ use futures::{Async, Poll, Stream, Sink, StartSend, AsyncSink}; use io::Io; +const INITIAL_CAPACITY: usize = 8 * 1024; + /// A reference counted buffer of bytes. /// /// An `EasyBuf` is a representation of a byte buffer where sub-slices of it can @@ -31,7 +33,7 @@ pub struct EasyBufMut<'a> { impl EasyBuf { /// Creates a new EasyBuf with no data and the default capacity. pub fn new() -> EasyBuf { - EasyBuf::with_capacity(8 * 1024) + EasyBuf::with_capacity(INITIAL_CAPACITY) } /// Creates a new EasyBuf with `cap` capacity. @@ -126,8 +128,8 @@ impl EasyBuf { /// If this `EasyBuf` is the only instance pointing at the underlying buffer /// of bytes, a direct mutable reference will be returned. Otherwise the /// contents of this `EasyBuf` will be reallocated in a fresh `Vec` - /// allocation with the same capacity as this allocation, and that - /// allocation will be returned. + /// allocation with the same capacity as an `EasyBuf` created with `EasyBuf::new()`, + /// and that allocation will be returned. /// /// This operation **is not O(1)** as it may clone the entire contents of /// this buffer. @@ -150,7 +152,7 @@ impl EasyBuf { // If we couldn't get access above then we give ourself a new buffer // here. - let mut v = Vec::with_capacity(self.buf.capacity()); + let mut v = Vec::with_capacity(INITIAL_CAPACITY); v.extend_from_slice(self.as_ref()); self.start = 0; self.buf = Arc::new(v); @@ -345,9 +347,10 @@ impl Sink for Framed { fn start_send(&mut self, item: C::Out) -> StartSend { // If the buffer is already over 8KiB, then attempt to flush it. If after flushing it's // *still* over 8KiB, then apply backpressure (reject the send). - if self.wr.len() > 8 * 1024 { + const BACKPRESSURE_BOUNDARY: usize = INITIAL_CAPACITY; + if self.wr.len() > BACKPRESSURE_BOUNDARY { try!(self.poll_complete()); - if self.wr.len() > 8 * 1024 { + if self.wr.len() > BACKPRESSURE_BOUNDARY { return Ok(AsyncSink::NotReady(item)); } } @@ -384,7 +387,7 @@ pub fn framed(io: T, codec: C) -> Framed { eof: false, is_readable: false, rd: EasyBuf::new(), - wr: Vec::with_capacity(8 * 1024), + wr: Vec::with_capacity(INITIAL_CAPACITY), } } @@ -421,7 +424,7 @@ impl Framed { #[cfg(test)] mod tests { - use super::EasyBuf; + use super::{INITIAL_CAPACITY, EasyBuf}; use std::mem; #[test] @@ -469,6 +472,7 @@ mod tests { // Clone to make shared let clone = buf.clone(); assert_eq!(*buf.get_mut(), [3, 4, 5, 6, 7, 8]); + assert_eq!(buf.get_mut().buf.capacity(), INITIAL_CAPACITY); mem::drop(clone); // prevent unused warning } From e11dd06ead68230dde6455367ff3237387e4981e Mon Sep 17 00:00:00 2001 From: Sebastian Thiel Date: Sat, 11 Feb 2017 10:23:02 +0100 Subject: [PATCH 2/3] Assure we don't have to allocate while growing the vector We now have enough capacity to copy the unconsumed portion of the previous frame. --- src/io/frame.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/io/frame.rs b/src/io/frame.rs index 44fa5a1b9..552deb5ec 100644 --- a/src/io/frame.rs +++ b/src/io/frame.rs @@ -1,6 +1,7 @@ use std::fmt; use std::io; use std::mem; +use std::cmp; use std::ops::{Deref, DerefMut}; use std::sync::Arc; @@ -152,7 +153,7 @@ impl EasyBuf { // If we couldn't get access above then we give ourself a new buffer // here. - let mut v = Vec::with_capacity(INITIAL_CAPACITY); + let mut v = Vec::with_capacity(cmp::min(INITIAL_CAPACITY, self.as_ref().len())); v.extend_from_slice(self.as_ref()); self.start = 0; self.buf = Arc::new(v); From 65c5c5241b43dff0d33c417df2e4aa67ba71176c Mon Sep 17 00:00:00 2001 From: Sebastian Thiel Date: Sun, 12 Feb 2017 09:47:03 +0100 Subject: [PATCH 3/3] Test semantics of buffer allocation of get_mut() * if remaining bytes are smaller then 8 * 1024, allocate 8 * 1024 * otherwise allocate as much as needed to hold the remaining bytes without re-allocations. --- src/io/frame.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/io/frame.rs b/src/io/frame.rs index 552deb5ec..26538a364 100644 --- a/src/io/frame.rs +++ b/src/io/frame.rs @@ -153,7 +153,7 @@ impl EasyBuf { // If we couldn't get access above then we give ourself a new buffer // here. - let mut v = Vec::with_capacity(cmp::min(INITIAL_CAPACITY, self.as_ref().len())); + let mut v = Vec::with_capacity(cmp::max(INITIAL_CAPACITY, self.as_ref().len())); v.extend_from_slice(self.as_ref()); self.start = 0; self.buf = Arc::new(v); @@ -465,7 +465,7 @@ mod tests { } #[test] - fn easybuf_get_mut_sliced_allocating() { + fn easybuf_get_mut_sliced_allocating_at_least_initial_capacity() { let vec: Vec = (0u8..10u8).collect(); let mut buf: EasyBuf = vec.into(); buf.split_off(9); @@ -477,6 +477,16 @@ mod tests { mem::drop(clone); // prevent unused warning } + #[test] + fn easybuf_get_mut_sliced_allocating_required_capacity() { + let vec: Vec = (0..INITIAL_CAPACITY * 2).map(|_|0u8).collect(); + let mut buf: EasyBuf = vec.into(); + buf.drain_to(INITIAL_CAPACITY / 2); + let clone = buf.clone(); + assert_eq!(buf.get_mut().buf.capacity(), INITIAL_CAPACITY + INITIAL_CAPACITY / 2); + mem::drop(clone) + } + #[test] fn easybuf_into_vec_simple() { let vec: Vec = (0u8..10u8).collect();