Add futures_util::io::{pipe,Pipe{Reader,Writer}}

Author: oxalica

futures-util/src/io/mod.rs

@@ -91,6 +91,9 @@ pub use self::into_sink::IntoSink;
 mod lines;
 pub use self::lines::Lines;
 
+mod pipe;
+pub use self::pipe::{pipe, PipeReader, PipeWriter};
+
 mod read;
 pub use self::read::Read;
 

futures-util/src/io/pipe.rs

@@ -0,0 +1,302 @@
+use core::pin::Pin;
+use core::ptr::copy_nonoverlapping;
+use core::slice;
+use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+use alloc::boxed::Box;
+use alloc::sync::Arc;
+use futures_core::ready;
+use futures_core::task::{Context, Poll, Waker};
+use futures_io::{AsyncBufRead, AsyncRead, AsyncWrite, Error, ErrorKind, Result};
+
+use crate::task::AtomicWaker;
+
+/// Create a unidirectional bounded pipe for data transfer between asynchronous tasks.
+///
+/// The internal buffer size is given by `buffer`, which must be non zero. The [`PipeWriter`]
+/// returned implements the [`AsyncWrite`] trait, while [`PipeReader`] implements [`AsyncRead`].
+///
+/// # Panics
+///
+/// Panics when `buffer` is zero.
+#[must_use]
+pub fn pipe(buffer: usize) -> (PipeWriter, PipeReader) {
+    assert!(buffer != 0, "pipe buffer size must be non zero and not usize::MAX");
+    // If it is `usize::MAX`, the allocation must fail anyway since Rust forbids allocations larger
+    // than `isize::MAX as usize`. This counts as OOM thus no need to state explicitly.
+    let len = buffer.saturating_add(1);
+    let ptr = Box::into_raw(alloc::vec![0u8; len].into_boxed_slice());
+    let inner = Arc::new(Shared {
+        len,
+        buffer: ptr.cast(),
+        write_pos: AtomicUsize::new(0),
+        read_pos: AtomicUsize::new(0),
+        writer_waker: AtomicWaker::new(),
+        reader_waker: AtomicWaker::new(),
+        closed: AtomicBool::new(false),
+    });
+    (PipeWriter { inner: inner.clone() }, PipeReader { inner })
+}
+
+// `read_pos..write_pos` (loop around, same below) contains the buffered content.
+// `write_pos..(read_pos-1)` is the empty space for further data.
+// Note that index `read_pos-1` is left vacant so that `read_pos == write_pos` if and only if
+// the buffer is empty.
+//
+// Invariants, at any time:
+// 1.  `read_pos` and `buffer[read_pos..write_pos]` is owned by the read-end.
+//     Read-end can increment `read_pos` in that range to transfer
+//     a portion of buffer to the write-end.
+// 2.  `write_pos` and `buffer[writer_pos..(read_pos-1)]` is owned by the write-end.
+//     Write-end can increment `write_pos` in that range to transfer
+//     a portion of buffer to the read-end.
+// 3.  Read-end can only park (returning Pending) when it observed `read_pos == write_pos` after
+//     registered the waker.
+// 4.  Write-end can only park when it observed `write_pos == read_pos-1` after
+//     registered the waker.
+#[derive(Debug)]
+struct Shared {
+    len: usize,
+    buffer: *mut u8,
+    read_pos: AtomicUsize,
+    write_pos: AtomicUsize,
+    reader_waker: AtomicWaker,
+    writer_waker: AtomicWaker,
+    closed: AtomicBool,
+}
+
+unsafe impl Send for Shared {}
+unsafe impl Sync for Shared {}
+
+impl Drop for Shared {
+    fn drop(&mut self) {
+        unsafe {
+            drop(Box::from_raw(slice::from_raw_parts_mut(self.buffer, self.len)));
+        }
+    }
+}
+
+impl Shared {
+    fn poll_read_ready(&self, waker: &Waker) -> Poll<Result<(usize, usize)>> {
+        // Only mutable by us reader. No synchronization for load.
+        let data_start = self.read_pos.load(Ordering::Relaxed);
+        // "Acquire" the bytes for read.
+        let mut data_end = self.write_pos.load(Ordering::Acquire);
+        // Fast path.
+        if data_start == data_end {
+            // Implicit "Acquite" `write_pos` below.
+            self.reader_waker.register(waker);
+            // Double check for readiness.
+            data_end = self.write_pos.load(Ordering::Acquire);
+            if data_start == data_end {
+                // Already "acquire"d by `reader_waker`.
+                if self.closed.load(Ordering::Relaxed) {
+                    return Poll::Ready(Ok((0, 0)));
+                }
+                return Poll::Pending;
+            }
+        }
+        Poll::Ready(Ok((data_start, data_end)))
+    }
+
+    unsafe fn commit_read(&self, new_read_pos: usize) {
+        // "Release" the bytes just read.
+        self.read_pos.store(new_read_pos, Ordering::Release);
+        // Implicit "Release" the `read_pos` change.
+        self.writer_waker.wake();
+    }
+
+    fn poll_write_ready(&self, waker: &Waker) -> Poll<Result<(usize, usize)>> {
+        // Only mutable by us writer. No synchronization for load.
+        let write_start = self.write_pos.load(Ordering::Relaxed);
+        // "Acquire" the bytes for write.
+        let mut write_end =
+            self.read_pos.load(Ordering::Acquire).checked_sub(1).unwrap_or(self.len - 1);
+        if write_start == write_end {
+            // Implicit "Acquite" `read_pos` below.
+            self.writer_waker.register(waker);
+            // Double check for writeness.
+            write_end =
+                self.read_pos.load(Ordering::Acquire).checked_sub(1).unwrap_or(self.len - 1);
+            if write_start == write_end {
+                // Already "acquire"d by `reader_waker`.
+                if self.closed.load(Ordering::Relaxed) {
+                    return Poll::Ready(Err(Error::new(ErrorKind::BrokenPipe, "pipe closed")));
+                }
+                return Poll::Pending;
+            }
+        }
+        Poll::Ready(Ok((write_start, write_end)))
+    }
+
+    unsafe fn commit_write(&self, new_write_pos: usize) {
+        // "Release" the bytes just written.
+        self.write_pos.store(new_write_pos, Ordering::Release);
+        // Implicit "Release" the `write_pos` change.
+        self.reader_waker.wake();
+    }
+}
+
+/// The write end of a bounded pipe.
+///
+/// This value is created by the [`pipe`] function.
+#[derive(Debug)]
+pub struct PipeWriter {
+    inner: Arc<Shared>,
+}
+
+impl Drop for PipeWriter {
+    fn drop(&mut self) {
+        self.inner.closed.store(true, Ordering::Relaxed);
+        // "Release" `closed`.
+        self.inner.reader_waker.wake();
+    }
+}
+
+impl AsyncWrite for PipeWriter {
+    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize>> {
+        if buf.is_empty() {
+            return Poll::Ready(Ok(0));
+        }
+
+        let inner = &*self.inner;
+
+        let (write_start, write_end) = ready!(inner.poll_write_ready(cx.waker()))?;
+
+        let written = if write_start <= write_end {
+            let written = buf.len().min(write_end - write_start);
+            // SAFETY: `buffer[write_pos..read_pos-1]` is owned by us writer.
+            unsafe {
+                copy_nonoverlapping(buf.as_ptr(), inner.buffer.add(write_start), written);
+            }
+            written
+        } else {
+            let written1 = buf.len().min(inner.len - write_start);
+            let written2 = (buf.len() - written1).min(write_end);
+            // SAFETY: `buffer[write_pos..]` and `buffer[..read_pos-1]` are owned by us writer.
+            unsafe {
+                copy_nonoverlapping(buf.as_ptr(), inner.buffer.add(write_start), written1);
+                copy_nonoverlapping(buf.as_ptr().add(written1), inner.buffer, written2);
+            }
+            written1 + written2
+        };
+
+        let mut new_write_pos = write_start + written;
+        if new_write_pos >= inner.len {
+            new_write_pos -= inner.len;
+        }
+
+        unsafe {
+            inner.commit_write(new_write_pos);
+        }
+
+        Poll::Ready(Ok(written))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+/// The read end of a bounded pipe.
+///
+/// This value is created by the [`pipe`] function.
+#[derive(Debug)]
+pub struct PipeReader {
+    inner: Arc<Shared>,
+}
+
+impl Drop for PipeReader {
+    fn drop(&mut self) {
+        self.inner.closed.store(true, Ordering::Relaxed);
+        // "Release" `closed`.
+        self.inner.writer_waker.wake();
+    }
+}
+
+impl AsyncRead for PipeReader {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize>> {
+        if buf.is_empty() {
+            return Poll::Ready(Ok(0));
+        }
+
+        let inner = &*self.inner;
+
+        let (data_start, data_end) = ready!(inner.poll_read_ready(cx.waker()))?;
+
+        let read = if data_start <= data_end {
+            let read = buf.len().min(data_end - data_start);
+            // SAFETY: `buffer[read_pos..write_pos]` are owned by us reader.
+            unsafe {
+                copy_nonoverlapping(inner.buffer.add(data_start), buf.as_mut_ptr(), read);
+            }
+            read
+        } else {
+            let read1 = buf.len().min(inner.len - data_start);
+            let read2 = (buf.len() - read1).min(data_end);
+            // SAFETY: `buffer[read_pos..]` and `buffer[..write_pos]` are owned by us reader.
+            unsafe {
+                copy_nonoverlapping(inner.buffer.add(data_start), buf.as_mut_ptr(), read1);
+                copy_nonoverlapping(inner.buffer, buf.as_mut_ptr().add(read1), read2);
+            }
+            read1 + read2
+        };
+
+        let mut new_read_pos = data_start + read;
+        if new_read_pos >= inner.len {
+            new_read_pos -= inner.len;
+        }
+
+        unsafe {
+            self.inner.commit_read(new_read_pos);
+        }
+
+        Poll::Ready(Ok(read))
+    }
+}
+
+impl AsyncBufRead for PipeReader {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<&[u8]>> {
+        let inner = &*self.inner;
+        let (data_start, mut data_end) = ready!(inner.poll_read_ready(cx.waker()))?;
+        if data_end < data_start {
+            data_end = inner.len;
+        }
+        // SAFETY: `buffer[read_pos..]` is owned by us reader.
+        let data =
+            unsafe { slice::from_raw_parts(inner.buffer.add(data_start), data_end - data_start) };
+        Poll::Ready(Ok(data))
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        let inner = &*self.inner;
+        // Only mutable by us reader. No synchronization for load.
+        let data_start = inner.read_pos.load(Ordering::Relaxed);
+        // Can only go forward since the last `poll_fill_buf` in the same thread.
+        // Does not need to be up-to-date.
+        let data_end = inner.write_pos.load(Ordering::Relaxed);
+
+        let len = if data_start <= data_end {
+            data_end - data_start
+        } else {
+            data_end + inner.len - data_start
+        };
+        assert!(amt <= len, "invalid advance");
+
+        let mut new_read_pos = data_start + amt;
+        if new_read_pos >= inner.len {
+            new_read_pos -= inner.len;
+        }
+        unsafe {
+            inner.commit_read(new_read_pos);
+        }
+    }
+}

futures/tests/io_pipe.rs

@@ -0,0 +1,205 @@
+use futures::future::FutureExt;
+use futures::task::Poll;
+use futures_core::task::Context;
+use futures_executor::block_on;
+use futures_io::ErrorKind;
+use futures_test::future::FutureTestExt;
+use futures_test::task::{new_count_waker, panic_context};
+use futures_util::io::{pipe, PipeReader, PipeWriter};
+use futures_util::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt};
+use static_assertions::assert_impl_all;
+
+trait PollExt<T> {
+    fn expect_pending(self);
+    fn expect_ready(self) -> T;
+}
+
+impl<T> PollExt<T> for Poll<T> {
+    #[track_caller]
+    fn expect_pending(self) {
+        assert!(self.is_pending());
+    }
+
+    #[track_caller]
+    fn expect_ready(self) -> T {
+        match self {
+            Poll::Ready(v) => v,
+            Poll::Pending => panic!("should be ready"),
+        }
+    }
+}
+
+// They have only `Pin<&mut Self>` methods. `&Self` can do nothing. Thus Sync.
+assert_impl_all!(PipeReader: Send, Sync, Unpin);
+assert_impl_all!(PipeWriter: Send, Sync, Unpin);
+
+#[test]
+fn small_write_nonblocking() {
+    let (mut w, mut r) = pipe(8);
+    let mut cx = panic_context();
+    for _ in 0..10 {
+        let mut buf = [0u8; 10];
+        assert_eq!(w.write(b"12345").poll_unpin(&mut cx).expect_ready().unwrap(), 5);
+        assert_eq!(r.read(&mut buf).poll_unpin(&mut cx).expect_ready().unwrap(), 5);
+        assert_eq!(&buf[..5], b"12345");
+    }
+}
+
+#[test]
+fn big_write_nonblocking() {
+    let (mut w, mut r) = pipe(7);
+    let mut cx = panic_context();
+    for _ in 0..10 {
+        let mut buf = [0u8; 10];
+        assert_eq!(w.write(b"1234567890").poll_unpin(&mut cx).expect_ready().unwrap(), 7);
+        assert_eq!(r.read(&mut buf).poll_unpin(&mut cx).expect_ready().unwrap(), 7);
+        assert_eq!(&buf[..7], b"1234567");
+    }
+}
+
+#[test]
+fn reader_blocked() {
+    let (mut w, mut r) = pipe(8);
+    let (waker, cnt) = new_count_waker();
+    let mut cx = Context::from_waker(&waker);
+
+    let mut buf = [0u8; 5];
+    r.read(&mut buf).poll_unpin(&mut cx).expect_pending();
+    assert_eq!(cnt.get(), 0);
+    assert_eq!(w.write(b"12345").poll_unpin(&mut cx).expect_ready().unwrap(), 5);
+    assert_eq!(cnt.get(), 1);
+    assert_eq!(r.read(&mut buf[..3]).poll_unpin(&mut cx).expect_ready().unwrap(), 3);
+    assert_eq!(&buf[..3], b"123");
+    assert_eq!(cnt.get(), 1);
+    assert_eq!(r.read(&mut buf[..3]).poll_unpin(&mut cx).expect_ready().unwrap(), 2);
+    assert_eq!(&buf[..2], b"45");
+    assert_eq!(cnt.get(), 1);
+    r.read(&mut buf[..3]).poll_unpin(&mut cx).expect_pending();
+    assert_eq!(cnt.get(), 1);
+}
+
+#[test]
+fn writer_blocked() {
+    let (mut w, mut r) = pipe(7);
+    let (waker, cnt) = new_count_waker();
+    let mut cx = Context::from_waker(&waker);
+    let mut buf = [0u8; 10];
+
+    assert_eq!(w.write(b"12345").poll_unpin(&mut cx).expect_ready().unwrap(), 5);
+    assert_eq!(w.write(b"67890").poll_unpin(&mut cx).expect_ready().unwrap(), 2);
+    assert_eq!(cnt.get(), 0);
+    w.write(b"xxx").poll_unpin(&mut cx).expect_pending();
+    assert_eq!(cnt.get(), 0);
+    assert_eq!(r.read(&mut buf[..3]).poll_unpin(&mut cx).expect_ready().unwrap(), 3);
+    assert_eq!(&buf[..3], b"123");
+    assert_eq!(cnt.get(), 1);
+    assert_eq!(w.write(b"abcde").poll_unpin(&mut cx).expect_ready().unwrap(), 3);
+    assert_eq!(cnt.get(), 1);
+    w.write(b"xxx").poll_unpin(&mut cx).expect_pending();
+    assert_eq!(cnt.get(), 1);
+    assert_eq!(r.read(&mut buf).poll_unpin(&mut cx).expect_ready().unwrap(), 7);
+    assert_eq!(&buf[..7], b"4567abc");
+    assert_eq!(cnt.get(), 2);
+}
+
+#[test]
+fn reader_closed_notify_writer() {
+    let (mut w, r) = pipe(4);
+    let (waker, cnt) = new_count_waker();
+    let mut cx = Context::from_waker(&waker);
+
+    assert_eq!(cnt.get(), 0);
+    assert_eq!(w.write(b"1234").poll_unpin(&mut cx).expect_ready().unwrap(), 4);
+    w.write(b"xxx").poll_unpin(&mut cx).expect_pending();
+    assert_eq!(cnt.get(), 0);
+    drop(r);
+    assert_eq!(cnt.get(), 1);
+
+    assert_eq!(
+        w.write(b"xxx").poll_unpin(&mut cx).expect_ready().unwrap_err().kind(),
+        ErrorKind::BrokenPipe
+    );
+}
+
+#[test]
+fn writer_closed_notify_reader() {
+    let (w, mut r) = pipe(4);
+    let (waker, cnt) = new_count_waker();
+    let mut cx = Context::from_waker(&waker);
+    let mut buf = [0u8; 10];
+
+    assert_eq!(cnt.get(), 0);
+    r.read(&mut buf).poll_unpin(&mut cx).expect_pending();
+    assert_eq!(cnt.get(), 0);
+    drop(w);
+    assert_eq!(cnt.get(), 1);
+
+    assert_eq!(r.read(&mut [0u8; 10]).poll_unpin(&mut cx).expect_ready().unwrap(), 0);
+}
+
+#[test]
+fn writer_closed_with_data() {
+    let (mut w, mut r) = pipe(4);
+    let mut cx = panic_context();
+    let mut buf = [0u8; 10];
+
+    assert_eq!(w.write(b"1234").poll_unpin(&mut cx).expect_ready().unwrap(), 4);
+    drop(w);
+    assert_eq!(r.read(&mut buf).poll_unpin(&mut cx).expect_ready().unwrap(), 4);
+    assert_eq!(&buf[..4], b"1234");
+    assert_eq!(r.read(&mut [0u8; 10]).poll_unpin(&mut cx).expect_ready().unwrap(), 0);
+}
+
+#[test]
+fn smoke() {
+    let (mut w, mut r) = pipe(128);
+    let data = "hello world".repeat(1024);
+
+    let reader = std::thread::spawn(|| {
+        block_on(async move {
+            let mut buf = String::new();
+            r.read_to_string(&mut buf).interleave_pending().await.unwrap();
+            buf
+        })
+    });
+
+    let writer = std::thread::spawn({
+        let data = data.clone();
+        || {
+            block_on(async move {
+                w.write_all(data.as_bytes()).interleave_pending().await.unwrap();
+            });
+        }
+    });
+
+    writer.join().unwrap();
+    let ret = reader.join().unwrap();
+    assert_eq!(ret, data);
+}
+
+#[test]
+fn smoke_bufread() {
+    let (mut w, mut r) = pipe(128);
+    let data = "hello world\n".repeat(1024);
+
+    let reader = std::thread::spawn(|| {
+        block_on(async move {
+            let mut buf = String::new();
+            while r.read_line(&mut buf).await.unwrap() != 0 {}
+            buf
+        })
+    });
+
+    let writer = std::thread::spawn({
+        let data = data.clone();
+        || {
+            block_on(async move {
+                w.write_all(data.as_bytes()).interleave_pending().await.unwrap();
+            });
+        }
+    });
+
+    writer.join().unwrap();
+    let ret = reader.join().unwrap();
+    assert_eq!(ret, data);
+}