use pin_project_lite::pin_project; use std::fmt; use std::io::{self, IoSlice, SeekFrom, Write}; use std::pin::Pin; use std::task::{Context, Poll};
pin_project! { /// Wraps a writer and buffers its output. /// /// It can be excessively inefficient to work directly with something that /// implements [`AsyncWrite`]. 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<u8>`. /// /// When the `BufWriter` is dropped, the contents of its buffer will be /// discarded. Creating multiple instances of a `BufWriter` on the same /// stream can cause data loss. If you need to write out the contents of its /// buffer, you must manually call flush before the writer is dropped. /// /// [`AsyncWrite`]: AsyncWrite /// [`flush`]: super::AsyncWriteExt::flush /// #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))] pubstruct BufWriter<W> { #[pin] pub(super) inner: W, pub(super) buf: Vec<u8>, pub(super) written: usize, pub(super) seek_state: SeekState,
}
}
impl<W: AsyncWrite> BufWriter<W> { /// Creates a new `BufWriter` with a default buffer capacity. The default is currently 8 KB, /// but may change in the future. pubfn new(inner: W) -> Self { Self::with_capacity(DEFAULT_BUF_SIZE, inner)
}
/// Creates a new `BufWriter` with the specified buffer capacity. pubfn with_capacity(cap: usize, inner: W) -> Self { Self {
inner,
buf: Vec::with_capacity(cap),
written: 0,
seek_state: SeekState::Init,
}
}
let len = me.buf.len(); letmut ret = Ok(()); while *me.written < len { match ready!(me.inner.as_mut().poll_write(cx, &me.buf[*me.written..])) {
Ok(0) => {
ret = Err(io::Error::new(
io::ErrorKind::WriteZero, "failed to write the buffered data",
)); break;
}
Ok(n) => *me.written += n,
Err(e) => {
ret = Err(e); break;
}
}
} if *me.written > 0 {
me.buf.drain(..*me.written);
}
*me.written = 0;
Poll::Ready(ret)
}
/// Gets a reference to the underlying writer. pubfn get_ref(&self) -> &W {
&self.inner
}
/// Gets a mutable reference to the underlying writer. /// /// It is inadvisable to directly write to the underlying writer. pubfn get_mut(&mutself) -> &mut W {
&mutself.inner
}
/// Gets a pinned mutable reference to the underlying writer. /// /// It is inadvisable to directly write to the underlying writer. pubfn get_pin_mut(self: Pin<&mutSelf>) -> Pin<&mut W> { self.project().inner
}
/// Consumes this `BufWriter`, returning the underlying writer. /// /// Note that any leftover data in the internal buffer is lost. pubfn into_inner(self) -> W { self.inner
}
/// Returns a reference to the internally buffered data. pubfn buffer(&self) -> &[u8] {
&self.buf
}
}
let me = self.project(); if buf.len() >= me.buf.capacity() {
me.inner.poll_write(cx, buf)
} else {
Poll::Ready(me.buf.write(buf))
}
}
fn poll_write_vectored( mutself: Pin<&mutSelf>,
cx: &mut Context<'_>, mut bufs: &[IoSlice<'_>],
) -> Poll<io::Result<usize>> { ifself.inner.is_write_vectored() { let total_len = bufs
.iter()
.fold(0usize, |acc, b| acc.saturating_add(b.len())); if total_len > self.buf.capacity() - self.buf.len() {
ready!(self.as_mut().flush_buf(cx))?;
} let me = self.as_mut().project(); if total_len >= me.buf.capacity() { // It's more efficient to pass the slices directly to the // underlying writer than to buffer them. // The case when the total_len calculation saturates at // usize::MAX is also handled here.
me.inner.poll_write_vectored(cx, bufs)
} else {
bufs.iter().for_each(|b| me.buf.extend_from_slice(b));
Poll::Ready(Ok(total_len))
}
} else { // Remove empty buffers at the beginning of bufs. while bufs.first().map(|buf| buf.len()) == Some(0) {
bufs = &bufs[1..];
} if bufs.is_empty() { return Poll::Ready(Ok(0));
} // Flush if the first buffer doesn't fit. let first_len = bufs[0].len(); if first_len > self.buf.capacity() - self.buf.len() {
ready!(self.as_mut().flush_buf(cx))?;
debug_assert!(self.buf.is_empty());
} let me = self.as_mut().project(); if first_len >= me.buf.capacity() { // The slice is at least as large as the buffering capacity, // so it's better to write it directly, bypassing the buffer.
debug_assert!(me.buf.is_empty()); return me.inner.poll_write(cx, &bufs[0]);
} else {
me.buf.extend_from_slice(&bufs[0]);
bufs = &bufs[1..];
} letmut total_written = first_len;
debug_assert!(total_written != 0); // Append the buffers that fit in the internal buffer. for buf in bufs { if buf.len() > me.buf.capacity() - me.buf.len() { break;
} else {
me.buf.extend_from_slice(buf);
total_written += buf.len();
}
}
Poll::Ready(Ok(total_written))
}
}
#[derive(Debug, Clone, Copy)] pub(super) enum SeekState { /// `start_seek` has not been called.
Init, /// `start_seek` has been called, but `poll_complete` has not yet been called.
Start(SeekFrom), /// Waiting for completion of `poll_complete`.
Pending,
}
/// Seek to the offset, in bytes, in the underlying writer. /// /// Seeking always writes out the internal buffer before seeking. impl<W: AsyncWrite + AsyncSeek> AsyncSeek for BufWriter<W> { fn start_seek(self: Pin<&mutSelf>, pos: SeekFrom) -> io::Result<()> { // We need to flush the internal buffer before seeking. // It receives a `Context` and returns a `Poll`, so it cannot be called // inside `start_seek`.
*self.project().seek_state = SeekState::Start(pos);
Ok(())
}
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.