//! Buffering wrappers for I/O traits use crate::io::prelude::*; use crate::cmp; use crate::error; use crate::fmt; use crate::io::{ self, Error, ErrorKind, Initializer, IoSlice, IoSliceMut, SeekFrom, DEFAULT_BUF_SIZE, }; use crate::memchr; /// The `BufReader` struct adds buffering to any reader. /// /// It can be excessively inefficient to work directly with a [`Read`] instance. /// For example, every call to [`read`][`TcpStream::read`] on [`TcpStream`] /// results in a system call. A `BufReader` performs large, infrequent reads on /// the underlying [`Read`] and maintains an in-memory buffer of the results. /// /// `BufReader` can improve the speed of programs that make *small* and /// *repeated* read calls to the same file or network socket. It does not /// help when reading very large amounts at once, or reading just one or a few /// times. It also provides no advantage when reading from a source that is /// already in memory, like a [`Vec`]``. /// /// When the `BufReader` is dropped, the contents of its buffer will be /// discarded. Creating multiple instances of a `BufReader` on the same /// stream can cause data loss. Reading from the underlying reader after /// unwrapping the `BufReader` with [`BufReader::into_inner`] can also cause /// data loss. /// /// [`TcpStream::read`]: Read::read /// [`TcpStream`]: crate::net::TcpStream /// /// # Examples /// /// ```no_run /// use std::io::prelude::*; /// use std::io::BufReader; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f = File::open("log.txt")?; /// let mut reader = BufReader::new(f); /// /// let mut line = String::new(); /// let len = reader.read_line(&mut line)?; /// println!("First line is {} bytes long", len); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub struct BufReader { inner: R, buf: Box<[u8]>, pos: usize, cap: usize, } impl BufReader { /// Creates a new `BufReader` with a default buffer capacity. The default is currently 8 KB, /// but may change in the future. /// /// # Examples /// /// ```no_run /// use std::io::BufReader; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f = File::open("log.txt")?; /// let reader = BufReader::new(f); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn new(inner: R) -> BufReader { BufReader::with_capacity(DEFAULT_BUF_SIZE, inner) } /// Creates a new `BufReader` with the specified buffer capacity. /// /// # Examples /// /// Creating a buffer with ten bytes of capacity: /// /// ```no_run /// use std::io::BufReader; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f = File::open("log.txt")?; /// let reader = BufReader::with_capacity(10, f); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn with_capacity(capacity: usize, inner: R) -> BufReader { unsafe { let mut buffer = Vec::with_capacity(capacity); buffer.set_len(capacity); inner.initializer().initialize(&mut buffer); BufReader { inner, buf: buffer.into_boxed_slice(), pos: 0, cap: 0 } } } } impl BufReader { /// Gets a reference to the underlying reader. /// /// It is inadvisable to directly read from the underlying reader. /// /// # Examples /// /// ```no_run /// use std::io::BufReader; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f1 = File::open("log.txt")?; /// let reader = BufReader::new(f1); /// /// let f2 = reader.get_ref(); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn get_ref(&self) -> &R { &self.inner } /// Gets a mutable reference to the underlying reader. /// /// It is inadvisable to directly read from the underlying reader. /// /// # Examples /// /// ```no_run /// use std::io::BufReader; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f1 = File::open("log.txt")?; /// let mut reader = BufReader::new(f1); /// /// let f2 = reader.get_mut(); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn get_mut(&mut self) -> &mut R { &mut self.inner } /// Returns a reference to the internally buffered data. /// /// Unlike [`fill_buf`], this will not attempt to fill the buffer if it is empty. /// /// [`fill_buf`]: BufRead::fill_buf /// /// # Examples /// /// ```no_run /// use std::io::{BufReader, BufRead}; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f = File::open("log.txt")?; /// let mut reader = BufReader::new(f); /// assert!(reader.buffer().is_empty()); /// /// if reader.fill_buf()?.len() > 0 { /// assert!(!reader.buffer().is_empty()); /// } /// Ok(()) /// } /// ``` #[stable(feature = "bufreader_buffer", since = "1.37.0")] pub fn buffer(&self) -> &[u8] { &self.buf[self.pos..self.cap] } /// Returns the number of bytes the internal buffer can hold at once. /// /// # Examples /// /// ```no_run /// use std::io::{BufReader, BufRead}; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f = File::open("log.txt")?; /// let mut reader = BufReader::new(f); /// /// let capacity = reader.capacity(); /// let buffer = reader.fill_buf()?; /// assert!(buffer.len() <= capacity); /// Ok(()) /// } /// ``` #[stable(feature = "buffered_io_capacity", since = "1.46.0")] pub fn capacity(&self) -> usize { self.buf.len() } /// Unwraps this `BufReader`, returning the underlying reader. /// /// Note that any leftover data in the internal buffer is lost. Therefore, /// a following read from the underlying reader may lead to data loss. /// /// # Examples /// /// ```no_run /// use std::io::BufReader; /// use std::fs::File; /// /// fn main() -> std::io::Result<()> { /// let f1 = File::open("log.txt")?; /// let reader = BufReader::new(f1); /// /// let f2 = reader.into_inner(); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn into_inner(self) -> R { self.inner } /// Invalidates all data in the internal buffer. #[inline] fn discard_buffer(&mut self) { self.pos = 0; self.cap = 0; } } impl BufReader { /// Seeks relative to the current position. If the new position lies within the buffer, /// the buffer will not be flushed, allowing for more efficient seeks. /// This method does not return the location of the underlying reader, so the caller /// must track this information themselves if it is required. #[unstable(feature = "bufreader_seek_relative", issue = "31100")] pub fn seek_relative(&mut self, offset: i64) -> io::Result<()> { let pos = self.pos as u64; if offset < 0 { if let Some(new_pos) = pos.checked_sub((-offset) as u64) { self.pos = new_pos as usize; return Ok(()); } } else { if let Some(new_pos) = pos.checked_add(offset as u64) { if new_pos <= self.cap as u64 { self.pos = new_pos as usize; return Ok(()); } } } self.seek(SeekFrom::Current(offset)).map(drop) } } #[stable(feature = "rust1", since = "1.0.0")] impl Read for BufReader { fn read(&mut self, buf: &mut [u8]) -> io::Result { // If we don't have any buffered data and we're doing a massive read // (larger than our internal buffer), bypass our internal buffer // entirely. if self.pos == self.cap && buf.len() >= self.buf.len() { self.discard_buffer(); return self.inner.read(buf); } let nread = { let mut rem = self.fill_buf()?; rem.read(buf)? }; self.consume(nread); Ok(nread) } fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> io::Result { let total_len = bufs.iter().map(|b| b.len()).sum::(); if self.pos == self.cap && total_len >= self.buf.len() { self.discard_buffer(); return self.inner.read_vectored(bufs); } let nread = { let mut rem = self.fill_buf()?; rem.read_vectored(bufs)? }; self.consume(nread); Ok(nread) } fn is_read_vectored(&self) -> bool { self.inner.is_read_vectored() } // we can't skip unconditionally because of the large buffer case in read. unsafe fn initializer(&self) -> Initializer { self.inner.initializer() } } #[stable(feature = "rust1", since = "1.0.0")] impl BufRead for BufReader { fn fill_buf(&mut self) -> io::Result<&[u8]> { // If we've reached the end of our internal buffer then we need to fetch // some more data from the underlying reader. // Branch using `>=` instead of the more correct `==` // to tell the compiler that the pos..cap slice is always valid. if self.pos >= self.cap { debug_assert!(self.pos == self.cap); self.cap = self.inner.read(&mut self.buf)?; self.pos = 0; } Ok(&self.buf[self.pos..self.cap]) } fn consume(&mut self, amt: usize) { self.pos = cmp::min(self.pos + amt, self.cap); } } #[stable(feature = "rust1", since = "1.0.0")] impl fmt::Debug for BufReader where R: fmt::Debug, { fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { fmt.debug_struct("BufReader") .field("reader", &self.inner) .field("buffer", &format_args!("{}/{}", self.cap - self.pos, self.buf.len())) .finish() } } #[stable(feature = "rust1", since = "1.0.0")] impl Seek for BufReader { /// Seek to an offset, in bytes, in the underlying reader. /// /// The position used for seeking with [`SeekFrom::Current`]`(_)` is the /// position the underlying reader would be at if the `BufReader` had no /// internal buffer. /// /// Seeking always discards the internal buffer, even if the seek position /// would otherwise fall within it. This guarantees that calling /// [`BufReader::into_inner()`] immediately after a seek yields the underlying reader /// at the same position. /// /// To seek without discarding the internal buffer, use [`BufReader::seek_relative`]. /// /// See [`std::io::Seek`] for more details. /// /// Note: In the edge case where you're seeking with [`SeekFrom::Current`]`(n)` /// where `n` minus the internal buffer length overflows an `i64`, two /// seeks will be performed instead of one. If the second seek returns /// [`Err`], the underlying reader will be left at the same position it would /// have if you called `seek` with [`SeekFrom::Current`]`(0)`. /// /// [`std::io::Seek`]: Seek fn seek(&mut self, pos: SeekFrom) -> io::Result { let result: u64; if let SeekFrom::Current(n) = pos { let remainder = (self.cap - self.pos) as i64; // it should be safe to assume that remainder fits within an i64 as the alternative // means we managed to allocate 8 exbibytes and that's absurd. // But it's not out of the realm of possibility for some weird underlying reader to // support seeking by i64::MIN so we need to handle underflow when subtracting // remainder. if let Some(offset) = n.checked_sub(remainder) { result = self.inner.seek(SeekFrom::Current(offset))?; } else { // seek backwards by our remainder, and then by the offset self.inner.seek(SeekFrom::Current(-remainder))?; self.discard_buffer(); result = self.inner.seek(SeekFrom::Current(n))?; } } else { // Seeking with Start/End doesn't care about our buffer length. result = self.inner.seek(pos)?; } self.discard_buffer(); Ok(result) } } /// Wraps a writer and buffers its output. /// /// It can be excessively inefficient to work directly with something that /// implements [`Write`]. For example, every call to /// [`write`][`TcpStream::write`] on [`TcpStream`] results in a system call. A /// `BufWriter` keeps an in-memory buffer of data and writes it to an underlying /// writer in large, infrequent batches. /// /// `BufWriter` can improve the speed of programs that make *small* and /// *repeated* write calls to the same file or network socket. It does not /// help when writing very large amounts at once, or writing just one or a few /// times. It also provides no advantage when writing to a destination that is /// in memory, like a [`Vec`]`. /// /// It is critical to call [`flush`] before `BufWriter` is dropped. Though /// dropping will attempt to flush the contents of the buffer, any errors /// that happen in the process of dropping will be ignored. Calling [`flush`] /// ensures that the buffer is empty and thus dropping will not even attempt /// file operations. /// /// # Examples /// /// Let's write the numbers one through ten to a [`TcpStream`]: /// /// ```no_run /// use std::io::prelude::*; /// use std::net::TcpStream; /// /// let mut stream = TcpStream::connect("127.0.0.1:34254").unwrap(); /// /// for i in 0..10 { /// stream.write(&[i+1]).unwrap(); /// } /// ``` /// /// Because we're not buffering, we write each one in turn, incurring the /// overhead of a system call per byte written. We can fix this with a /// `BufWriter`: /// /// ```no_run /// use std::io::prelude::*; /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut stream = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// for i in 0..10 { /// stream.write(&[i+1]).unwrap(); /// } /// stream.flush().unwrap(); /// ``` /// /// By wrapping the stream with a `BufWriter`, these ten writes are all grouped /// together by the buffer and will all be written out in one system call when /// the `stream` is flushed. /// /// [`TcpStream::write`]: Write::write /// [`TcpStream`]: crate::net::TcpStream /// [`flush`]: Write::flush #[stable(feature = "rust1", since = "1.0.0")] pub struct BufWriter { inner: Option, buf: Vec, // #30888: If the inner writer panics in a call to write, we don't want to // write the buffered data a second time in BufWriter's destructor. This // flag tells the Drop impl if it should skip the flush. panicked: bool, } /// An error returned by [`BufWriter::into_inner`] which combines an error that /// happened while writing out the buffer, and the buffered writer object /// which may be used to recover from the condition. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut stream = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // do stuff with the stream /// /// // we want to get our `TcpStream` back, so let's try: /// /// let stream = match stream.into_inner() { /// Ok(s) => s, /// Err(e) => { /// // Here, e is an IntoInnerError /// panic!("An error occurred"); /// } /// }; /// ``` #[derive(Debug)] #[stable(feature = "rust1", since = "1.0.0")] pub struct IntoInnerError(W, Error); impl BufWriter { /// Creates a new `BufWriter` with a default buffer capacity. The default is currently 8 KB, /// but may change in the future. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut buffer = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn new(inner: W) -> BufWriter { BufWriter::with_capacity(DEFAULT_BUF_SIZE, inner) } /// Creates a new `BufWriter` with the specified buffer capacity. /// /// # Examples /// /// Creating a buffer with a buffer of a hundred bytes. /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let stream = TcpStream::connect("127.0.0.1:34254").unwrap(); /// let mut buffer = BufWriter::with_capacity(100, stream); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn with_capacity(capacity: usize, inner: W) -> BufWriter { BufWriter { inner: Some(inner), buf: Vec::with_capacity(capacity), panicked: false } } /// Send data in our local buffer into the inner writer, looping as /// necessary until either it's all been sent or an error occurs. /// /// Because all the data in the buffer has been reported to our owner as /// "successfully written" (by returning nonzero success values from /// `write`), any 0-length writes from `inner` must be reported as i/o /// errors from this method. fn flush_buf(&mut self) -> io::Result<()> { /// Helper struct to ensure the buffer is updated after all the writes /// are complete. It tracks the number of written bytes and drains them /// all from the front of the buffer when dropped. struct BufGuard<'a> { buffer: &'a mut Vec, written: usize, } impl<'a> BufGuard<'a> { fn new(buffer: &'a mut Vec) -> Self { Self { buffer, written: 0 } } /// The unwritten part of the buffer fn remaining(&self) -> &[u8] { &self.buffer[self.written..] } /// Flag some bytes as removed from the front of the buffer fn consume(&mut self, amt: usize) { self.written += amt; } /// true if all of the bytes have been written fn done(&self) -> bool { self.written >= self.buffer.len() } } impl Drop for BufGuard<'_> { fn drop(&mut self) { if self.written > 0 { self.buffer.drain(..self.written); } } } let mut guard = BufGuard::new(&mut self.buf); let inner = self.inner.as_mut().unwrap(); while !guard.done() { self.panicked = true; let r = inner.write(guard.remaining()); self.panicked = false; match r { Ok(0) => { return Err(Error::new( ErrorKind::WriteZero, "failed to write the buffered data", )); } Ok(n) => guard.consume(n), Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} Err(e) => return Err(e), } } Ok(()) } /// Buffer some data without flushing it, regardless of the size of the /// data. Writes as much as possible without exceeding capacity. Returns /// the number of bytes written. fn write_to_buf(&mut self, buf: &[u8]) -> usize { let available = self.buf.capacity() - self.buf.len(); let amt_to_buffer = available.min(buf.len()); self.buf.extend_from_slice(&buf[..amt_to_buffer]); amt_to_buffer } /// Gets a reference to the underlying writer. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut buffer = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // we can use reference just like buffer /// let reference = buffer.get_ref(); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn get_ref(&self) -> &W { self.inner.as_ref().unwrap() } /// Gets a mutable reference to the underlying writer. /// /// It is inadvisable to directly write to the underlying writer. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut buffer = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // we can use reference just like buffer /// let reference = buffer.get_mut(); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn get_mut(&mut self) -> &mut W { self.inner.as_mut().unwrap() } /// Returns a reference to the internally buffered data. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let buf_writer = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // See how many bytes are currently buffered /// let bytes_buffered = buf_writer.buffer().len(); /// ``` #[stable(feature = "bufreader_buffer", since = "1.37.0")] pub fn buffer(&self) -> &[u8] { &self.buf } /// Returns the number of bytes the internal buffer can hold without flushing. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let buf_writer = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // Check the capacity of the inner buffer /// let capacity = buf_writer.capacity(); /// // Calculate how many bytes can be written without flushing /// let without_flush = capacity - buf_writer.buffer().len(); /// ``` #[stable(feature = "buffered_io_capacity", since = "1.46.0")] pub fn capacity(&self) -> usize { self.buf.capacity() } /// Unwraps this `BufWriter`, returning the underlying writer. /// /// The buffer is written out before returning the writer. /// /// # Errors /// /// An [`Err`] will be returned if an error occurs while flushing the buffer. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut buffer = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // unwrap the TcpStream and flush the buffer /// let stream = buffer.into_inner().unwrap(); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn into_inner(mut self) -> Result>> { match self.flush_buf() { Err(e) => Err(IntoInnerError(self, e)), Ok(()) => Ok(self.inner.take().unwrap()), } } } #[stable(feature = "rust1", since = "1.0.0")] impl Write for BufWriter { fn write(&mut self, buf: &[u8]) -> io::Result { if self.buf.len() + buf.len() > self.buf.capacity() { self.flush_buf()?; } // FIXME: Why no len > capacity? Why not buffer len == capacity? #72919 if buf.len() >= self.buf.capacity() { self.panicked = true; let r = self.get_mut().write(buf); self.panicked = false; r } else { self.buf.extend_from_slice(buf); Ok(buf.len()) } } fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { // Normally, `write_all` just calls `write` in a loop. We can do better // by calling `self.get_mut().write_all()` directly, which avoids // round trips through the buffer in the event of a series of partial // writes in some circumstances. if self.buf.len() + buf.len() > self.buf.capacity() { self.flush_buf()?; } // FIXME: Why no len > capacity? Why not buffer len == capacity? #72919 if buf.len() >= self.buf.capacity() { self.panicked = true; let r = self.get_mut().write_all(buf); self.panicked = false; r } else { self.buf.extend_from_slice(buf); Ok(()) } } fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> io::Result { let total_len = bufs.iter().map(|b| b.len()).sum::(); if self.buf.len() + total_len > self.buf.capacity() { self.flush_buf()?; } // FIXME: Why no len > capacity? Why not buffer len == capacity? #72919 if total_len >= self.buf.capacity() { self.panicked = true; let r = self.get_mut().write_vectored(bufs); self.panicked = false; r } else { bufs.iter().for_each(|b| self.buf.extend_from_slice(b)); Ok(total_len) } } fn is_write_vectored(&self) -> bool { self.get_ref().is_write_vectored() } fn flush(&mut self) -> io::Result<()> { self.flush_buf().and_then(|()| self.get_mut().flush()) } } #[stable(feature = "rust1", since = "1.0.0")] impl fmt::Debug for BufWriter where W: fmt::Debug, { fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { fmt.debug_struct("BufWriter") .field("writer", &self.inner.as_ref().unwrap()) .field("buffer", &format_args!("{}/{}", self.buf.len(), self.buf.capacity())) .finish() } } #[stable(feature = "rust1", since = "1.0.0")] impl Seek for BufWriter { /// Seek to the offset, in bytes, in the underlying writer. /// /// Seeking always writes out the internal buffer before seeking. fn seek(&mut self, pos: SeekFrom) -> io::Result { self.flush_buf()?; self.get_mut().seek(pos) } } #[stable(feature = "rust1", since = "1.0.0")] impl Drop for BufWriter { fn drop(&mut self) { if self.inner.is_some() && !self.panicked { // dtors should not panic, so we ignore a failed flush let _r = self.flush_buf(); } } } impl IntoInnerError { /// Returns the error which caused the call to [`BufWriter::into_inner()`] /// to fail. /// /// This error was returned when attempting to write the internal buffer. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut stream = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // do stuff with the stream /// /// // we want to get our `TcpStream` back, so let's try: /// /// let stream = match stream.into_inner() { /// Ok(s) => s, /// Err(e) => { /// // Here, e is an IntoInnerError, let's log the inner error. /// // /// // We'll just 'log' to stdout for this example. /// println!("{}", e.error()); /// /// panic!("An unexpected error occurred."); /// } /// }; /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn error(&self) -> &Error { &self.1 } /// Returns the buffered writer instance which generated the error. /// /// The returned object can be used for error recovery, such as /// re-inspecting the buffer. /// /// # Examples /// /// ```no_run /// use std::io::BufWriter; /// use std::net::TcpStream; /// /// let mut stream = BufWriter::new(TcpStream::connect("127.0.0.1:34254").unwrap()); /// /// // do stuff with the stream /// /// // we want to get our `TcpStream` back, so let's try: /// /// let stream = match stream.into_inner() { /// Ok(s) => s, /// Err(e) => { /// // Here, e is an IntoInnerError, let's re-examine the buffer: /// let buffer = e.into_inner(); /// /// // do stuff to try to recover /// /// // afterwards, let's just return the stream /// buffer.into_inner().unwrap() /// } /// }; /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn into_inner(self) -> W { self.0 } } #[stable(feature = "rust1", since = "1.0.0")] impl From> for Error { fn from(iie: IntoInnerError) -> Error { iie.1 } } #[stable(feature = "rust1", since = "1.0.0")] impl error::Error for IntoInnerError { #[allow(deprecated, deprecated_in_future)] fn description(&self) -> &str { error::Error::description(self.error()) } } #[stable(feature = "rust1", since = "1.0.0")] impl fmt::Display for IntoInnerError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { self.error().fmt(f) } } /// Private helper struct for implementing the line-buffered writing logic. /// This shim temporarily wraps a BufWriter, and uses its internals to /// implement a line-buffered writer (specifically by using the internal /// methods like write_to_buf and flush_buf). In this way, a more /// efficient abstraction can be created than one that only had access to /// `write` and `flush`, without needlessly duplicating a lot of the /// implementation details of BufWriter. This also allows existing /// `BufWriters` to be temporarily given line-buffering logic; this is what /// enables Stdout to be alternately in line-buffered or block-buffered mode. #[derive(Debug)] pub(super) struct LineWriterShim<'a, W: Write> { buffer: &'a mut BufWriter, } impl<'a, W: Write> LineWriterShim<'a, W> { pub fn new(buffer: &'a mut BufWriter) -> Self { Self { buffer } } /// Get a mutable reference to the inner writer (that is, the writer /// wrapped by the BufWriter). Be careful with this writer, as writes to /// it will bypass the buffer. fn inner_mut(&mut self) -> &mut W { self.buffer.get_mut() } /// Get the content currently buffered in self.buffer fn buffered(&self) -> &[u8] { self.buffer.buffer() } /// Flush the buffer iff the last byte is a newline (indicating that an /// earlier write only succeeded partially, and we want to retry flushing /// the buffered line before continuing with a subsequent write) fn flush_if_completed_line(&mut self) -> io::Result<()> { match self.buffered().last().copied() { Some(b'\n') => self.buffer.flush_buf(), _ => Ok(()), } } } impl<'a, W: Write> Write for LineWriterShim<'a, W> { /// Write some data into this BufReader with line buffering. This means /// that, if any newlines are present in the data, the data up to the last /// newline is sent directly to the underlying writer, and data after it /// is buffered. Returns the number of bytes written. /// /// This function operates on a "best effort basis"; in keeping with the /// convention of `Write::write`, it makes at most one attempt to write /// new data to the underlying writer. If that write only reports a partial /// success, the remaining data will be buffered. /// /// Because this function attempts to send completed lines to the underlying /// writer, it will also flush the existing buffer if it ends with a /// newline, even if the incoming data does not contain any newlines. fn write(&mut self, buf: &[u8]) -> io::Result { let newline_idx = match memchr::memrchr(b'\n', buf) { // If there are no new newlines (that is, if this write is less than // one line), just do a regular buffered write (which may flush if // we exceed the inner buffer's size) None => { self.flush_if_completed_line()?; return self.buffer.write(buf); } // Otherwise, arrange for the lines to be written directly to the // inner writer. Some(newline_idx) => newline_idx + 1, }; // Flush existing content to prepare for our write. We have to do this // before attempting to write `buf` in order to maintain consistency; // if we add `buf` to the buffer then try to flush it all at once, // we're obligated to return Ok(), which would mean suppressing any // errors that occur during flush. self.buffer.flush_buf()?; // This is what we're going to try to write directly to the inner // writer. The rest will be buffered, if nothing goes wrong. let lines = &buf[..newline_idx]; // Write `lines` directly to the inner writer. In keeping with the // `write` convention, make at most one attempt to add new (unbuffered) // data. Because this write doesn't touch the BufWriter state directly, // and the buffer is known to be empty, we don't need to worry about // self.buffer.panicked here. let flushed = self.inner_mut().write(lines)?; // If buffer returns Ok(0), propagate that to the caller without // doing additional buffering; otherwise we're just guaranteeing // an "ErrorKind::WriteZero" later. if flushed == 0 { return Ok(0); } // Now that the write has succeeded, buffer the rest (or as much of // the rest as possible). If there were any unwritten newlines, we // only buffer out to the last unwritten newline that fits in the // buffer; this helps prevent flushing partial lines on subsequent // calls to LineWriterShim::write. // Handle the cases in order of most-common to least-common, under // the presumption that most writes succeed in totality, and that most // writes are smaller than the buffer. // - Is this a partial line (ie, no newlines left in the unwritten tail) // - If not, does the data out to the last unwritten newline fit in // the buffer? // - If not, scan for the last newline that *does* fit in the buffer let tail = if flushed >= newline_idx { &buf[flushed..] } else if newline_idx - flushed <= self.buffer.capacity() { &buf[flushed..newline_idx] } else { let scan_area = &buf[flushed..]; let scan_area = &scan_area[..self.buffer.capacity()]; match memchr::memrchr(b'\n', scan_area) { Some(newline_idx) => &scan_area[..newline_idx + 1], None => scan_area, } }; let buffered = self.buffer.write_to_buf(tail); Ok(flushed + buffered) } fn flush(&mut self) -> io::Result<()> { self.buffer.flush() } /// Write some vectored data into this BufReader with line buffering. This /// means that, if any newlines are present in the data, the data up to /// and including the buffer containing the last newline is sent directly /// to the inner writer, and the data after it is buffered. Returns the /// number of bytes written. /// /// This function operates on a "best effort basis"; in keeping with the /// convention of `Write::write`, it makes at most one attempt to write /// new data to the underlying writer. /// /// Because this function attempts to send completed lines to the underlying /// writer, it will also flush the existing buffer if it contains any /// newlines. /// /// Because sorting through an array of `IoSlice` can be a bit convoluted, /// This method differs from write in the following ways: /// /// - It attempts to write the full content of all the buffers up to and /// including the one containing the last newline. This means that it /// may attempt to write a partial line, that buffer has data past the /// newline. /// - If the write only reports partial success, it does not attempt to /// find the precise location of the written bytes and buffer the rest. /// /// If the underlying vector doesn't support vectored writing, we instead /// simply write the first non-empty buffer with `write`. This way, we /// get the benefits of more granular partial-line handling without losing /// anything in efficiency fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> io::Result { // If there's no specialized behavior for write_vectored, just use // write. This has the benefit of more granular partial-line handling. if !self.is_write_vectored() { return match bufs.iter().find(|buf| !buf.is_empty()) { Some(buf) => self.write(buf), None => Ok(0), }; } // Find the buffer containing the last newline let last_newline_buf_idx = bufs .iter() .enumerate() .rev() .find_map(|(i, buf)| memchr::memchr(b'\n', buf).map(|_| i)); // If there are no new newlines (that is, if this write is less than // one line), just do a regular buffered write let last_newline_buf_idx = match last_newline_buf_idx { // No newlines; just do a normal buffered write None => { self.flush_if_completed_line()?; return self.buffer.write_vectored(bufs); } Some(i) => i, }; // Flush existing content to prepare for our write self.buffer.flush_buf()?; // This is what we're going to try to write directly to the inner // writer. The rest will be buffered, if nothing goes wrong. let (lines, tail) = bufs.split_at(last_newline_buf_idx + 1); // Write `lines` directly to the inner writer. In keeping with the // `write` convention, make at most one attempt to add new (unbuffered) // data. Because this write doesn't touch the BufWriter state directly, // and the buffer is known to be empty, we don't need to worry about // self.panicked here. let flushed = self.inner_mut().write_vectored(lines)?; // If inner returns Ok(0), propagate that to the caller without // doing additional buffering; otherwise we're just guaranteeing // an "ErrorKind::WriteZero" later. if flushed == 0 { return Ok(0); } // Don't try to reconstruct the exact amount written; just bail // in the event of a partial write let lines_len = lines.iter().map(|buf| buf.len()).sum(); if flushed < lines_len { return Ok(flushed); } // Now that the write has succeeded, buffer the rest (or as much of the // rest as possible) let buffered: usize = tail .iter() .filter(|buf| !buf.is_empty()) .map(|buf| self.buffer.write_to_buf(buf)) .take_while(|&n| n > 0) .sum(); Ok(flushed + buffered) } fn is_write_vectored(&self) -> bool { self.buffer.is_write_vectored() } /// Write some data into this BufReader with line buffering. This means /// that, if any newlines are present in the data, the data up to the last /// newline is sent directly to the underlying writer, and data after it /// is buffered. /// /// Because this function attempts to send completed lines to the underlying /// writer, it will also flush the existing buffer if it contains any /// newlines, even if the incoming data does not contain any newlines. fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { match memchr::memrchr(b'\n', buf) { // If there are no new newlines (that is, if this write is less than // one line), just do a regular buffered write (which may flush if // we exceed the inner buffer's size) None => { self.flush_if_completed_line()?; self.buffer.write_all(buf) } Some(newline_idx) => { let (lines, tail) = buf.split_at(newline_idx + 1); if self.buffered().is_empty() { self.inner_mut().write_all(lines)?; } else { // If there is any buffered data, we add the incoming lines // to that buffer before flushing, which saves us at least // one write call. We can't really do this with `write`, // since we can't do this *and* not suppress errors *and* // report a consistent state to the caller in a return // value, but here in write_all it's fine. self.buffer.write_all(lines)?; self.buffer.flush_buf()?; } self.buffer.write_all(tail) } } } } /// Wraps a writer and buffers output to it, flushing whenever a newline /// (`0x0a`, `'\n'`) is detected. /// /// The [`BufWriter`] struct wraps a writer and buffers its output. /// But it only does this batched write when it goes out of scope, or when the /// internal buffer is full. Sometimes, you'd prefer to write each line as it's /// completed, rather than the entire buffer at once. Enter `LineWriter`. It /// does exactly that. /// /// Like [`BufWriter`], a `LineWriter`’s buffer will also be flushed when the /// `LineWriter` goes out of scope or when its internal buffer is full. /// /// If there's still a partial line in the buffer when the `LineWriter` is /// dropped, it will flush those contents. /// /// # Examples /// /// We can use `LineWriter` to write one line at a time, significantly /// reducing the number of actual writes to the file. /// /// ```no_run /// use std::fs::{self, File}; /// use std::io::prelude::*; /// use std::io::LineWriter; /// /// fn main() -> std::io::Result<()> { /// let road_not_taken = b"I shall be telling this with a sigh /// Somewhere ages and ages hence: /// Two roads diverged in a wood, and I - /// I took the one less traveled by, /// And that has made all the difference."; /// /// let file = File::create("poem.txt")?; /// let mut file = LineWriter::new(file); /// /// file.write_all(b"I shall be telling this with a sigh")?; /// /// // No bytes are written until a newline is encountered (or /// // the internal buffer is filled). /// assert_eq!(fs::read_to_string("poem.txt")?, ""); /// file.write_all(b"\n")?; /// assert_eq!( /// fs::read_to_string("poem.txt")?, /// "I shall be telling this with a sigh\n", /// ); /// /// // Write the rest of the poem. /// file.write_all(b"Somewhere ages and ages hence: /// Two roads diverged in a wood, and I - /// I took the one less traveled by, /// And that has made all the difference.")?; /// /// // The last line of the poem doesn't end in a newline, so /// // we have to flush or drop the `LineWriter` to finish /// // writing. /// file.flush()?; /// /// // Confirm the whole poem was written. /// assert_eq!(fs::read("poem.txt")?, &road_not_taken[..]); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub struct LineWriter { inner: BufWriter, } impl LineWriter { /// Creates a new `LineWriter`. /// /// # Examples /// /// ```no_run /// use std::fs::File; /// use std::io::LineWriter; /// /// fn main() -> std::io::Result<()> { /// let file = File::create("poem.txt")?; /// let file = LineWriter::new(file); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn new(inner: W) -> LineWriter { // Lines typically aren't that long, don't use a giant buffer LineWriter::with_capacity(1024, inner) } /// Creates a new `LineWriter` with a specified capacity for the internal /// buffer. /// /// # Examples /// /// ```no_run /// use std::fs::File; /// use std::io::LineWriter; /// /// fn main() -> std::io::Result<()> { /// let file = File::create("poem.txt")?; /// let file = LineWriter::with_capacity(100, file); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn with_capacity(capacity: usize, inner: W) -> LineWriter { LineWriter { inner: BufWriter::with_capacity(capacity, inner) } } /// Gets a reference to the underlying writer. /// /// # Examples /// /// ```no_run /// use std::fs::File; /// use std::io::LineWriter; /// /// fn main() -> std::io::Result<()> { /// let file = File::create("poem.txt")?; /// let file = LineWriter::new(file); /// /// let reference = file.get_ref(); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn get_ref(&self) -> &W { self.inner.get_ref() } /// Gets a mutable reference to the underlying writer. /// /// Caution must be taken when calling methods on the mutable reference /// returned as extra writes could corrupt the output stream. /// /// # Examples /// /// ```no_run /// use std::fs::File; /// use std::io::LineWriter; /// /// fn main() -> std::io::Result<()> { /// let file = File::create("poem.txt")?; /// let mut file = LineWriter::new(file); /// /// // we can use reference just like file /// let reference = file.get_mut(); /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn get_mut(&mut self) -> &mut W { self.inner.get_mut() } /// Unwraps this `LineWriter`, returning the underlying writer. /// /// The internal buffer is written out before returning the writer. /// /// # Errors /// /// An [`Err`] will be returned if an error occurs while flushing the buffer. /// /// # Examples /// /// ```no_run /// use std::fs::File; /// use std::io::LineWriter; /// /// fn main() -> std::io::Result<()> { /// let file = File::create("poem.txt")?; /// /// let writer: LineWriter = LineWriter::new(file); /// /// let file: File = writer.into_inner()?; /// Ok(()) /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn into_inner(self) -> Result>> { self.inner .into_inner() .map_err(|IntoInnerError(buf, e)| IntoInnerError(LineWriter { inner: buf }, e)) } } #[stable(feature = "rust1", since = "1.0.0")] impl Write for LineWriter { fn write(&mut self, buf: &[u8]) -> io::Result { LineWriterShim::new(&mut self.inner).write(buf) } fn flush(&mut self) -> io::Result<()> { self.inner.flush() } fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> io::Result { LineWriterShim::new(&mut self.inner).write_vectored(bufs) } fn is_write_vectored(&self) -> bool { self.inner.is_write_vectored() } fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { LineWriterShim::new(&mut self.inner).write_all(buf) } fn write_all_vectored(&mut self, bufs: &mut [IoSlice<'_>]) -> io::Result<()> { LineWriterShim::new(&mut self.inner).write_all_vectored(bufs) } fn write_fmt(&mut self, fmt: fmt::Arguments<'_>) -> io::Result<()> { LineWriterShim::new(&mut self.inner).write_fmt(fmt) } } #[stable(feature = "rust1", since = "1.0.0")] impl fmt::Debug for LineWriter where W: fmt::Debug, { fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { fmt.debug_struct("LineWriter") .field("writer", &self.inner.inner) .field( "buffer", &format_args!("{}/{}", self.inner.buf.len(), self.inner.buf.capacity()), ) .finish() } } #[cfg(test)] mod tests { use crate::io::prelude::*; use crate::io::{self, BufReader, BufWriter, ErrorKind, IoSlice, LineWriter, SeekFrom}; use crate::sync::atomic::{AtomicUsize, Ordering}; use crate::thread; /// A dummy reader intended at testing short-reads propagation. pub struct ShortReader { lengths: Vec, } // FIXME: rustfmt and tidy disagree about the correct formatting of this // function. This leads to issues for users with editors configured to // rustfmt-on-save. impl Read for ShortReader { fn read(&mut self, _: &mut [u8]) -> io::Result { if self.lengths.is_empty() { Ok(0) } else { Ok(self.lengths.remove(0)) } } } #[test] fn test_buffered_reader() { let inner: &[u8] = &[5, 6, 7, 0, 1, 2, 3, 4]; let mut reader = BufReader::with_capacity(2, inner); let mut buf = [0, 0, 0]; let nread = reader.read(&mut buf); assert_eq!(nread.unwrap(), 3); assert_eq!(buf, [5, 6, 7]); assert_eq!(reader.buffer(), []); let mut buf = [0, 0]; let nread = reader.read(&mut buf); assert_eq!(nread.unwrap(), 2); assert_eq!(buf, [0, 1]); assert_eq!(reader.buffer(), []); let mut buf = [0]; let nread = reader.read(&mut buf); assert_eq!(nread.unwrap(), 1); assert_eq!(buf, [2]); assert_eq!(reader.buffer(), [3]); let mut buf = [0, 0, 0]; let nread = reader.read(&mut buf); assert_eq!(nread.unwrap(), 1); assert_eq!(buf, [3, 0, 0]); assert_eq!(reader.buffer(), []); let nread = reader.read(&mut buf); assert_eq!(nread.unwrap(), 1); assert_eq!(buf, [4, 0, 0]); assert_eq!(reader.buffer(), []); assert_eq!(reader.read(&mut buf).unwrap(), 0); } #[test] fn test_buffered_reader_seek() { let inner: &[u8] = &[5, 6, 7, 0, 1, 2, 3, 4]; let mut reader = BufReader::with_capacity(2, io::Cursor::new(inner)); assert_eq!(reader.seek(SeekFrom::Start(3)).ok(), Some(3)); assert_eq!(reader.fill_buf().ok(), Some(&[0, 1][..])); assert_eq!(reader.seek(SeekFrom::Current(0)).ok(), Some(3)); assert_eq!(reader.fill_buf().ok(), Some(&[0, 1][..])); assert_eq!(reader.seek(SeekFrom::Current(1)).ok(), Some(4)); assert_eq!(reader.fill_buf().ok(), Some(&[1, 2][..])); reader.consume(1); assert_eq!(reader.seek(SeekFrom::Current(-2)).ok(), Some(3)); } #[test] fn test_buffered_reader_seek_relative() { let inner: &[u8] = &[5, 6, 7, 0, 1, 2, 3, 4]; let mut reader = BufReader::with_capacity(2, io::Cursor::new(inner)); assert!(reader.seek_relative(3).is_ok()); assert_eq!(reader.fill_buf().ok(), Some(&[0, 1][..])); assert!(reader.seek_relative(0).is_ok()); assert_eq!(reader.fill_buf().ok(), Some(&[0, 1][..])); assert!(reader.seek_relative(1).is_ok()); assert_eq!(reader.fill_buf().ok(), Some(&[1][..])); assert!(reader.seek_relative(-1).is_ok()); assert_eq!(reader.fill_buf().ok(), Some(&[0, 1][..])); assert!(reader.seek_relative(2).is_ok()); assert_eq!(reader.fill_buf().ok(), Some(&[2, 3][..])); } #[test] fn test_buffered_reader_invalidated_after_read() { let inner: &[u8] = &[5, 6, 7, 0, 1, 2, 3, 4]; let mut reader = BufReader::with_capacity(3, io::Cursor::new(inner)); assert_eq!(reader.fill_buf().ok(), Some(&[5, 6, 7][..])); reader.consume(3); let mut buffer = [0, 0, 0, 0, 0]; assert_eq!(reader.read(&mut buffer).ok(), Some(5)); assert_eq!(buffer, [0, 1, 2, 3, 4]); assert!(reader.seek_relative(-2).is_ok()); let mut buffer = [0, 0]; assert_eq!(reader.read(&mut buffer).ok(), Some(2)); assert_eq!(buffer, [3, 4]); } #[test] fn test_buffered_reader_invalidated_after_seek() { let inner: &[u8] = &[5, 6, 7, 0, 1, 2, 3, 4]; let mut reader = BufReader::with_capacity(3, io::Cursor::new(inner)); assert_eq!(reader.fill_buf().ok(), Some(&[5, 6, 7][..])); reader.consume(3); assert!(reader.seek(SeekFrom::Current(5)).is_ok()); assert!(reader.seek_relative(-2).is_ok()); let mut buffer = [0, 0]; assert_eq!(reader.read(&mut buffer).ok(), Some(2)); assert_eq!(buffer, [3, 4]); } #[test] fn test_buffered_reader_seek_underflow() { // gimmick reader that yields its position modulo 256 for each byte struct PositionReader { pos: u64, } impl Read for PositionReader { fn read(&mut self, buf: &mut [u8]) -> io::Result { let len = buf.len(); for x in buf { *x = self.pos as u8; self.pos = self.pos.wrapping_add(1); } Ok(len) } } impl Seek for PositionReader { fn seek(&mut self, pos: SeekFrom) -> io::Result { match pos { SeekFrom::Start(n) => { self.pos = n; } SeekFrom::Current(n) => { self.pos = self.pos.wrapping_add(n as u64); } SeekFrom::End(n) => { self.pos = u64::MAX.wrapping_add(n as u64); } } Ok(self.pos) } } let mut reader = BufReader::with_capacity(5, PositionReader { pos: 0 }); assert_eq!(reader.fill_buf().ok(), Some(&[0, 1, 2, 3, 4][..])); assert_eq!(reader.seek(SeekFrom::End(-5)).ok(), Some(u64::MAX - 5)); assert_eq!(reader.fill_buf().ok().map(|s| s.len()), Some(5)); // the following seek will require two underlying seeks let expected = 9223372036854775802; assert_eq!(reader.seek(SeekFrom::Current(i64::MIN)).ok(), Some(expected)); assert_eq!(reader.fill_buf().ok().map(|s| s.len()), Some(5)); // seeking to 0 should empty the buffer. assert_eq!(reader.seek(SeekFrom::Current(0)).ok(), Some(expected)); assert_eq!(reader.get_ref().pos, expected); } #[test] fn test_buffered_reader_seek_underflow_discard_buffer_between_seeks() { // gimmick reader that returns Err after first seek struct ErrAfterFirstSeekReader { first_seek: bool, } impl Read for ErrAfterFirstSeekReader { fn read(&mut self, buf: &mut [u8]) -> io::Result { for x in &mut *buf { *x = 0; } Ok(buf.len()) } } impl Seek for ErrAfterFirstSeekReader { fn seek(&mut self, _: SeekFrom) -> io::Result { if self.first_seek { self.first_seek = false; Ok(0) } else { Err(io::Error::new(io::ErrorKind::Other, "oh no!")) } } } let mut reader = BufReader::with_capacity(5, ErrAfterFirstSeekReader { first_seek: true }); assert_eq!(reader.fill_buf().ok(), Some(&[0, 0, 0, 0, 0][..])); // The following seek will require two underlying seeks. The first will // succeed but the second will fail. This should still invalidate the // buffer. assert!(reader.seek(SeekFrom::Current(i64::MIN)).is_err()); assert_eq!(reader.buffer().len(), 0); } #[test] fn test_buffered_writer() { let inner = Vec::new(); let mut writer = BufWriter::with_capacity(2, inner); writer.write(&[0, 1]).unwrap(); assert_eq!(writer.buffer(), []); assert_eq!(*writer.get_ref(), [0, 1]); writer.write(&[2]).unwrap(); assert_eq!(writer.buffer(), [2]); assert_eq!(*writer.get_ref(), [0, 1]); writer.write(&[3]).unwrap(); assert_eq!(writer.buffer(), [2, 3]); assert_eq!(*writer.get_ref(), [0, 1]); writer.flush().unwrap(); assert_eq!(writer.buffer(), []); assert_eq!(*writer.get_ref(), [0, 1, 2, 3]); writer.write(&[4]).unwrap(); writer.write(&[5]).unwrap(); assert_eq!(writer.buffer(), [4, 5]); assert_eq!(*writer.get_ref(), [0, 1, 2, 3]); writer.write(&[6]).unwrap(); assert_eq!(writer.buffer(), [6]); assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5]); writer.write(&[7, 8]).unwrap(); assert_eq!(writer.buffer(), []); assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8]); writer.write(&[9, 10, 11]).unwrap(); assert_eq!(writer.buffer(), []); assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]); writer.flush().unwrap(); assert_eq!(writer.buffer(), []); assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]); } #[test] fn test_buffered_writer_inner_flushes() { let mut w = BufWriter::with_capacity(3, Vec::new()); w.write(&[0, 1]).unwrap(); assert_eq!(*w.get_ref(), []); let w = w.into_inner().unwrap(); assert_eq!(w, [0, 1]); } #[test] fn test_buffered_writer_seek() { let mut w = BufWriter::with_capacity(3, io::Cursor::new(Vec::new())); w.write_all(&[0, 1, 2, 3, 4, 5]).unwrap(); w.write_all(&[6, 7]).unwrap(); assert_eq!(w.seek(SeekFrom::Current(0)).ok(), Some(8)); assert_eq!(&w.get_ref().get_ref()[..], &[0, 1, 2, 3, 4, 5, 6, 7][..]); assert_eq!(w.seek(SeekFrom::Start(2)).ok(), Some(2)); w.write_all(&[8, 9]).unwrap(); assert_eq!(&w.into_inner().unwrap().into_inner()[..], &[0, 1, 8, 9, 4, 5, 6, 7]); } #[test] fn test_read_until() { let inner: &[u8] = &[0, 1, 2, 1, 0]; let mut reader = BufReader::with_capacity(2, inner); let mut v = Vec::new(); reader.read_until(0, &mut v).unwrap(); assert_eq!(v, [0]); v.truncate(0); reader.read_until(2, &mut v).unwrap(); assert_eq!(v, [1, 2]); v.truncate(0); reader.read_until(1, &mut v).unwrap(); assert_eq!(v, [1]); v.truncate(0); reader.read_until(8, &mut v).unwrap(); assert_eq!(v, [0]); v.truncate(0); reader.read_until(9, &mut v).unwrap(); assert_eq!(v, []); } #[test] fn test_line_buffer() { let mut writer = LineWriter::new(Vec::new()); writer.write(&[0]).unwrap(); assert_eq!(*writer.get_ref(), []); writer.write(&[1]).unwrap(); assert_eq!(*writer.get_ref(), []); writer.flush().unwrap(); assert_eq!(*writer.get_ref(), [0, 1]); writer.write(&[0, b'\n', 1, b'\n', 2]).unwrap(); assert_eq!(*writer.get_ref(), [0, 1, 0, b'\n', 1, b'\n']); writer.flush().unwrap(); assert_eq!(*writer.get_ref(), [0, 1, 0, b'\n', 1, b'\n', 2]); writer.write(&[3, b'\n']).unwrap(); assert_eq!(*writer.get_ref(), [0, 1, 0, b'\n', 1, b'\n', 2, 3, b'\n']); } #[test] fn test_read_line() { let in_buf: &[u8] = b"a\nb\nc"; let mut reader = BufReader::with_capacity(2, in_buf); let mut s = String::new(); reader.read_line(&mut s).unwrap(); assert_eq!(s, "a\n"); s.truncate(0); reader.read_line(&mut s).unwrap(); assert_eq!(s, "b\n"); s.truncate(0); reader.read_line(&mut s).unwrap(); assert_eq!(s, "c"); s.truncate(0); reader.read_line(&mut s).unwrap(); assert_eq!(s, ""); } #[test] fn test_lines() { let in_buf: &[u8] = b"a\nb\nc"; let reader = BufReader::with_capacity(2, in_buf); let mut it = reader.lines(); assert_eq!(it.next().unwrap().unwrap(), "a".to_string()); assert_eq!(it.next().unwrap().unwrap(), "b".to_string()); assert_eq!(it.next().unwrap().unwrap(), "c".to_string()); assert!(it.next().is_none()); } #[test] fn test_short_reads() { let inner = ShortReader { lengths: vec![0, 1, 2, 0, 1, 0] }; let mut reader = BufReader::new(inner); let mut buf = [0, 0]; assert_eq!(reader.read(&mut buf).unwrap(), 0); assert_eq!(reader.read(&mut buf).unwrap(), 1); assert_eq!(reader.read(&mut buf).unwrap(), 2); assert_eq!(reader.read(&mut buf).unwrap(), 0); assert_eq!(reader.read(&mut buf).unwrap(), 1); assert_eq!(reader.read(&mut buf).unwrap(), 0); assert_eq!(reader.read(&mut buf).unwrap(), 0); } #[test] #[should_panic] fn dont_panic_in_drop_on_panicked_flush() { struct FailFlushWriter; impl Write for FailFlushWriter { fn write(&mut self, buf: &[u8]) -> io::Result { Ok(buf.len()) } fn flush(&mut self) -> io::Result<()> { Err(io::Error::last_os_error()) } } let writer = FailFlushWriter; let _writer = BufWriter::new(writer); // If writer panics *again* due to the flush error then the process will // abort. panic!(); } #[test] #[cfg_attr(target_os = "emscripten", ignore)] fn panic_in_write_doesnt_flush_in_drop() { static WRITES: AtomicUsize = AtomicUsize::new(0); struct PanicWriter; impl Write for PanicWriter { fn write(&mut self, _: &[u8]) -> io::Result { WRITES.fetch_add(1, Ordering::SeqCst); panic!(); } fn flush(&mut self) -> io::Result<()> { Ok(()) } } thread::spawn(|| { let mut writer = BufWriter::new(PanicWriter); let _ = writer.write(b"hello world"); let _ = writer.flush(); }) .join() .unwrap_err(); assert_eq!(WRITES.load(Ordering::SeqCst), 1); } #[bench] fn bench_buffered_reader(b: &mut test::Bencher) { b.iter(|| BufReader::new(io::empty())); } #[bench] fn bench_buffered_writer(b: &mut test::Bencher) { b.iter(|| BufWriter::new(io::sink())); } /// A simple `Write` target, designed to be wrapped by `LineWriter` / /// `BufWriter` / etc, that can have its `write` & `flush` behavior /// configured #[derive(Default, Clone)] struct ProgrammableSink { // Writes append to this slice pub buffer: Vec, // Flush sets this flag pub flushed: bool, // If true, writes will always be an error pub always_write_error: bool, // If true, flushes will always be an error pub always_flush_error: bool, // If set, only up to this number of bytes will be written in a single // call to `write` pub accept_prefix: Option, // If set, counts down with each write, and writes return an error // when it hits 0 pub max_writes: Option, // If set, attempting to write when max_writes == Some(0) will be an // error; otherwise, it will return Ok(0). pub error_after_max_writes: bool, } impl Write for ProgrammableSink { fn write(&mut self, data: &[u8]) -> io::Result { if self.always_write_error { return Err(io::Error::new(io::ErrorKind::Other, "test - always_write_error")); } match self.max_writes { Some(0) if self.error_after_max_writes => { return Err(io::Error::new(io::ErrorKind::Other, "test - max_writes")); } Some(0) => return Ok(0), Some(ref mut count) => *count -= 1, None => {} } let len = match self.accept_prefix { None => data.len(), Some(prefix) => data.len().min(prefix), }; let data = &data[..len]; self.buffer.extend_from_slice(data); Ok(len) } fn flush(&mut self) -> io::Result<()> { if self.always_flush_error { Err(io::Error::new(io::ErrorKind::Other, "test - always_flush_error")) } else { self.flushed = true; Ok(()) } } } /// Previously the `LineWriter` could successfully write some bytes but /// then fail to report that it has done so. Additionally, an erroneous /// flush after a successful write was permanently ignored. /// /// Test that a line writer correctly reports the number of written bytes, /// and that it attempts to flush buffered lines from previous writes /// before processing new data /// /// Regression test for #37807 #[test] fn erroneous_flush_retried() { let writer = ProgrammableSink { // Only write up to 4 bytes at a time accept_prefix: Some(4), // Accept the first two writes, then error the others max_writes: Some(2), error_after_max_writes: true, ..Default::default() }; // This should write the first 4 bytes. The rest will be buffered, out // to the last newline. let mut writer = LineWriter::new(writer); assert_eq!(writer.write(b"a\nb\nc\nd\ne").unwrap(), 8); // This write should attempt to flush "c\nd\n", then buffer "e". No // errors should happen here because no further writes should be // attempted against `writer`. assert_eq!(writer.write(b"e").unwrap(), 1); assert_eq!(&writer.get_ref().buffer, b"a\nb\nc\nd\n"); } #[test] fn line_vectored() { let mut a = LineWriter::new(Vec::new()); assert_eq!( a.write_vectored(&[ IoSlice::new(&[]), IoSlice::new(b"\n"), IoSlice::new(&[]), IoSlice::new(b"a"), ]) .unwrap(), 2, ); assert_eq!(a.get_ref(), b"\n"); assert_eq!( a.write_vectored(&[ IoSlice::new(&[]), IoSlice::new(b"b"), IoSlice::new(&[]), IoSlice::new(b"a"), IoSlice::new(&[]), IoSlice::new(b"c"), ]) .unwrap(), 3, ); assert_eq!(a.get_ref(), b"\n"); a.flush().unwrap(); assert_eq!(a.get_ref(), b"\nabac"); assert_eq!(a.write_vectored(&[]).unwrap(), 0); assert_eq!( a.write_vectored(&[ IoSlice::new(&[]), IoSlice::new(&[]), IoSlice::new(&[]), IoSlice::new(&[]), ]) .unwrap(), 0, ); assert_eq!(a.write_vectored(&[IoSlice::new(b"a\nb"),]).unwrap(), 3); assert_eq!(a.get_ref(), b"\nabaca\nb"); } #[test] fn line_vectored_partial_and_errors() { use crate::collections::VecDeque; enum Call { Write { inputs: Vec<&'static [u8]>, output: io::Result }, Flush { output: io::Result<()> }, } #[derive(Default)] struct Writer { calls: VecDeque, } impl Write for Writer { fn write(&mut self, buf: &[u8]) -> io::Result { self.write_vectored(&[IoSlice::new(buf)]) } fn write_vectored(&mut self, buf: &[IoSlice<'_>]) -> io::Result { match self.calls.pop_front().expect("unexpected call to write") { Call::Write { inputs, output } => { assert_eq!(inputs, buf.iter().map(|b| &**b).collect::>()); output } Call::Flush { .. } => panic!("unexpected call to write; expected a flush"), } } fn is_write_vectored(&self) -> bool { true } fn flush(&mut self) -> io::Result<()> { match self.calls.pop_front().expect("Unexpected call to flush") { Call::Flush { output } => output, Call::Write { .. } => panic!("unexpected call to flush; expected a write"), } } } impl Drop for Writer { fn drop(&mut self) { if !thread::panicking() { assert_eq!(self.calls.len(), 0); } } } // partial writes keep going let mut a = LineWriter::new(Writer::default()); a.write_vectored(&[IoSlice::new(&[]), IoSlice::new(b"abc")]).unwrap(); a.get_mut().calls.push_back(Call::Write { inputs: vec![b"abc"], output: Ok(1) }); a.get_mut().calls.push_back(Call::Write { inputs: vec![b"bc"], output: Ok(2) }); a.get_mut().calls.push_back(Call::Write { inputs: vec![b"x", b"\n"], output: Ok(2) }); a.write_vectored(&[IoSlice::new(b"x"), IoSlice::new(b"\n")]).unwrap(); a.get_mut().calls.push_back(Call::Flush { output: Ok(()) }); a.flush().unwrap(); // erroneous writes stop and don't write more a.get_mut().calls.push_back(Call::Write { inputs: vec![b"x", b"\na"], output: Err(err()) }); a.get_mut().calls.push_back(Call::Flush { output: Ok(()) }); assert!(a.write_vectored(&[IoSlice::new(b"x"), IoSlice::new(b"\na")]).is_err()); a.flush().unwrap(); fn err() -> io::Error { io::Error::new(io::ErrorKind::Other, "x") } } /// Test that, in cases where vectored writing is not enabled, the /// LineWriter uses the normal `write` call, which more-correctly handles /// partial lines #[test] fn line_vectored_ignored() { let writer = ProgrammableSink::default(); let mut writer = LineWriter::new(writer); let content = [ IoSlice::new(&[]), IoSlice::new(b"Line 1\nLine"), IoSlice::new(b" 2\nLine 3\nL"), IoSlice::new(&[]), IoSlice::new(&[]), IoSlice::new(b"ine 4"), IoSlice::new(b"\nLine 5\n"), ]; let count = writer.write_vectored(&content).unwrap(); assert_eq!(count, 11); assert_eq!(&writer.get_ref().buffer, b"Line 1\n"); let count = writer.write_vectored(&content[2..]).unwrap(); assert_eq!(count, 11); assert_eq!(&writer.get_ref().buffer, b"Line 1\nLine 2\nLine 3\n"); let count = writer.write_vectored(&content[5..]).unwrap(); assert_eq!(count, 5); assert_eq!(&writer.get_ref().buffer, b"Line 1\nLine 2\nLine 3\n"); let count = writer.write_vectored(&content[6..]).unwrap(); assert_eq!(count, 8); assert_eq!( writer.get_ref().buffer.as_slice(), b"Line 1\nLine 2\nLine 3\nLine 4\nLine 5\n".as_ref() ); } /// Test that, given this input: /// /// Line 1\n /// Line 2\n /// Line 3\n /// Line 4 /// /// And given a result that only writes to midway through Line 2 /// /// That only up to the end of Line 3 is buffered /// /// This behavior is desirable because it prevents flushing partial lines #[test] fn partial_write_buffers_line() { let writer = ProgrammableSink { accept_prefix: Some(13), ..Default::default() }; let mut writer = LineWriter::new(writer); assert_eq!(writer.write(b"Line 1\nLine 2\nLine 3\nLine4").unwrap(), 21); assert_eq!(&writer.get_ref().buffer, b"Line 1\nLine 2"); assert_eq!(writer.write(b"Line 4").unwrap(), 6); assert_eq!(&writer.get_ref().buffer, b"Line 1\nLine 2\nLine 3\n"); } /// Test that, given this input: /// /// Line 1\n /// Line 2\n /// Line 3 /// /// And given that the full write of lines 1 and 2 was successful /// That data up to Line 3 is buffered #[test] fn partial_line_buffered_after_line_write() { let writer = ProgrammableSink::default(); let mut writer = LineWriter::new(writer); assert_eq!(writer.write(b"Line 1\nLine 2\nLine 3").unwrap(), 20); assert_eq!(&writer.get_ref().buffer, b"Line 1\nLine 2\n"); assert!(writer.flush().is_ok()); assert_eq!(&writer.get_ref().buffer, b"Line 1\nLine 2\nLine 3"); } /// Test that, given a partial line that exceeds the length of /// LineBuffer's buffer (that is, without a trailing newline), that that /// line is written to the inner writer #[test] fn long_line_flushed() { let writer = ProgrammableSink::default(); let mut writer = LineWriter::with_capacity(5, writer); assert_eq!(writer.write(b"0123456789").unwrap(), 10); assert_eq!(&writer.get_ref().buffer, b"0123456789"); } /// Test that, given a very long partial line *after* successfully /// flushing a complete line, that that line is buffered unconditionally, /// and no additional writes take place. This assures the property that /// `write` should make at-most-one attempt to write new data. #[test] fn line_long_tail_not_flushed() { let writer = ProgrammableSink::default(); let mut writer = LineWriter::with_capacity(5, writer); // Assert that Line 1\n is flushed, and 01234 is buffered assert_eq!(writer.write(b"Line 1\n0123456789").unwrap(), 12); assert_eq!(&writer.get_ref().buffer, b"Line 1\n"); // Because the buffer is full, this subsequent write will flush it assert_eq!(writer.write(b"5").unwrap(), 1); assert_eq!(&writer.get_ref().buffer, b"Line 1\n01234"); } /// Test that, if an attempt to pre-flush buffered data returns Ok(0), /// this is propagated as an error. #[test] fn line_buffer_write0_error() { let writer = ProgrammableSink { // Accept one write, then return Ok(0) on subsequent ones max_writes: Some(1), ..Default::default() }; let mut writer = LineWriter::new(writer); // This should write "Line 1\n" and buffer "Partial" assert_eq!(writer.write(b"Line 1\nPartial").unwrap(), 14); assert_eq!(&writer.get_ref().buffer, b"Line 1\n"); // This will attempt to flush "partial", which will return Ok(0), which // needs to be an error, because we've already informed the client // that we accepted the write. let err = writer.write(b" Line End\n").unwrap_err(); assert_eq!(err.kind(), ErrorKind::WriteZero); assert_eq!(&writer.get_ref().buffer, b"Line 1\n"); } /// Test that, if a write returns Ok(0) after a successful pre-flush, this /// is propagated as Ok(0) #[test] fn line_buffer_write0_normal() { let writer = ProgrammableSink { // Accept two writes, then return Ok(0) on subsequent ones max_writes: Some(2), ..Default::default() }; let mut writer = LineWriter::new(writer); // This should write "Line 1\n" and buffer "Partial" assert_eq!(writer.write(b"Line 1\nPartial").unwrap(), 14); assert_eq!(&writer.get_ref().buffer, b"Line 1\n"); // This will flush partial, which will succeed, but then return Ok(0) // when flushing " Line End\n" assert_eq!(writer.write(b" Line End\n").unwrap(), 0); assert_eq!(&writer.get_ref().buffer, b"Line 1\nPartial"); } /// LineWriter has a custom `write_all`; make sure it works correctly #[test] fn line_write_all() { let writer = ProgrammableSink { // Only write 5 bytes at a time accept_prefix: Some(5), ..Default::default() }; let mut writer = LineWriter::new(writer); writer.write_all(b"Line 1\nLine 2\nLine 3\nLine 4\nPartial").unwrap(); assert_eq!(&writer.get_ref().buffer, b"Line 1\nLine 2\nLine 3\nLine 4\n"); writer.write_all(b" Line 5\n").unwrap(); assert_eq!( writer.get_ref().buffer.as_slice(), b"Line 1\nLine 2\nLine 3\nLine 4\nPartial Line 5\n".as_ref(), ); } #[test] fn line_write_all_error() { let writer = ProgrammableSink { // Only accept up to 3 writes of up to 5 bytes each accept_prefix: Some(5), max_writes: Some(3), ..Default::default() }; let mut writer = LineWriter::new(writer); let res = writer.write_all(b"Line 1\nLine 2\nLine 3\nLine 4\nPartial"); assert!(res.is_err()); // An error from write_all leaves everything in an indeterminate state, // so there's nothing else to test here } /// Under certain circumstances, the old implementation of LineWriter /// would try to buffer "to the last newline" but be forced to buffer /// less than that, leading to inappropriate partial line writes. /// Regression test for that issue. #[test] fn partial_multiline_buffering() { let writer = ProgrammableSink { // Write only up to 5 bytes at a time accept_prefix: Some(5), ..Default::default() }; let mut writer = LineWriter::with_capacity(10, writer); let content = b"AAAAABBBBB\nCCCCDDDDDD\nEEE"; // When content is written, LineWriter will try to write blocks A, B, // C, and D. Only block A will succeed. Under the old behavior, LineWriter // would then try to buffer B, C and D, but because its capacity is 10, // it will only be able to buffer B and C. We don't want to buffer // partial lines concurrent with whole lines, so the correct behavior // is to buffer only block B (out to the newline) assert_eq!(writer.write(content).unwrap(), 11); assert_eq!(writer.get_ref().buffer, *b"AAAAA"); writer.flush().unwrap(); assert_eq!(writer.get_ref().buffer, *b"AAAAABBBBB\n"); } /// Same as test_partial_multiline_buffering, but in the event NO full lines /// fit in the buffer, just buffer as much as possible #[test] fn partial_multiline_buffering_without_full_line() { let writer = ProgrammableSink { // Write only up to 5 bytes at a time accept_prefix: Some(5), ..Default::default() }; let mut writer = LineWriter::with_capacity(5, writer); let content = b"AAAAABBBBBBBBBB\nCCCCC\nDDDDD"; // When content is written, LineWriter will try to write blocks A, B, // and C. Only block A will succeed. Under the old behavior, LineWriter // would then try to buffer B and C, but because its capacity is 5, // it will only be able to buffer part of B. Because it's not possible // for it to buffer any complete lines, it should buffer as much of B as // possible assert_eq!(writer.write(content).unwrap(), 10); assert_eq!(writer.get_ref().buffer, *b"AAAAA"); writer.flush().unwrap(); assert_eq!(writer.get_ref().buffer, *b"AAAAABBBBB"); } #[derive(Debug, Clone, PartialEq, Eq)] enum RecordedEvent { Write(String), Flush, } #[derive(Debug, Clone, Default)] struct WriteRecorder { pub events: Vec, } impl Write for WriteRecorder { fn write(&mut self, buf: &[u8]) -> io::Result { use crate::str::from_utf8; self.events.push(RecordedEvent::Write(from_utf8(buf).unwrap().to_string())); Ok(buf.len()) } fn flush(&mut self) -> io::Result<()> { self.events.push(RecordedEvent::Flush); Ok(()) } } /// Test that a normal, formatted writeln only results in a single write /// call to the underlying writer. A naive implementation of /// LineWriter::write_all results in two writes: one of the buffered data, /// and another of the final substring in the formatted set #[test] fn single_formatted_write() { let writer = WriteRecorder::default(); let mut writer = LineWriter::new(writer); // Under a naive implementation of LineWriter, this will result in two // writes: "hello, world" and "!\n", because write() has to flush the // buffer before attempting to write the last "!\n". write_all shouldn't // have this limitation. writeln!(&mut writer, "{}, {}!", "hello", "world").unwrap(); assert_eq!(writer.get_ref().events, [RecordedEvent::Write("hello, world!\n".to_string())]); } }