Add split() and Pipe

Author: DanglingPointer

.github/workflows/ci.yml

@@ -63,4 +63,8 @@ jobs:
     steps:
     - uses: actions/checkout@v4
     - uses: dtolnay/rust-toolchain@stable
-    - run: cargo test --doc
+    - run: cargo test --doc
+    - run: cargo doc --lib --no-deps --all-features --document-private-items
+      env:
+        RUSTDOCFLAGS: -Dwarnings
+        RUSTFLAGS: -Dwarnings

.vscode/settings.json

@@ -19,6 +19,7 @@
     "rust-analyzer.runnables.extraTestBinaryArgs": [
         "--nocapture"
     ],
+    "rust-analyzer.cargo.features": "all",
     "gitlens.codeLens.enabled": false,
     "editor.semanticTokenColorCustomizations": {
         "rules": {

Cargo.toml

@@ -15,11 +15,12 @@ tokio-time = ["dep:tokio"]
 [dependencies]
 futures = "0.3.31"
 log = "0.4.27"
-tokio = { version = "1.45.1", optional = true, features = ["time"] }
+tokio = { version = "1.48.0", optional = true, features = ["time"] }
 
 [dev-dependencies]
 static_assertions = "1.1.0"
 tokio-test = "0.4.4"
+tokio = { version = "1.48.0", features = ["time", "io-util"] }
 
 [profile.dev]
 opt-level = 3

src/lib.rs

@@ -1,5 +1,7 @@
 pub mod sealed;
 pub mod shared;
+#[cfg(feature = "tokio-time")]
+pub mod split;
 pub mod stopwatch;
 pub mod sync;
 mod time;

src/split.rs

@@ -0,0 +1,68 @@
+use std::cell::RefCell;
+use std::io;
+use std::pin::Pin;
+use std::rc::Rc;
+use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+
+/// The readable half of a value returned from [`split`].
+pub struct ReadHalf<T: AsyncRead>(Rc<RefCell<T>>);
+
+/// The writable half of a value returned from [`split`].
+pub struct WriteHalf<T: AsyncWrite>(Rc<RefCell<T>>);
+
+/// Splits a single value implementing `AsyncRead + AsyncWrite` into separate `AsyncRead` and `AsyncWrite` handles.
+/// Non-thread-safe equivalent of [`tokio::io::split`](https://docs.rs/tokio/latest/tokio/io/fn.split.html) without the overhead of a mutex.
+pub fn split<T: AsyncRead + AsyncWrite>(value: T) -> (ReadHalf<T>, WriteHalf<T>) {
+    let shared = Rc::new(RefCell::new(value));
+    (ReadHalf(shared.clone()), WriteHalf(shared))
+}
+
+fn with_pin<T, R>(half: &RefCell<T>, f: impl FnOnce(Pin<&mut T>) -> R) -> R {
+    let mut guard = half.borrow_mut();
+
+    // SAFETY: we do not move the stream
+    let stream = unsafe { Pin::new_unchecked(&mut *guard) };
+
+    f(stream)
+}
+
+impl<T: AsyncRead> AsyncRead for ReadHalf<T> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        with_pin(&self.0, |inner| inner.poll_read(cx, buf))
+    }
+}
+
+impl<T: AsyncWrite> AsyncWrite for WriteHalf<T> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        with_pin(&self.0, |inner| inner.poll_write(cx, buf))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        with_pin(&self.0, |inner| inner.poll_flush(cx))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        with_pin(&self.0, |inner| inner.poll_shutdown(cx))
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[io::IoSlice<'_>],
+    ) -> Poll<Result<usize, io::Error>> {
+        with_pin(&self.0, |inner| inner.poll_write_vectored(cx, bufs))
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.0.borrow().is_write_vectored()
+    }
+}

src/sync/mod.rs

@@ -4,6 +4,8 @@ pub mod bounded;
 pub mod condvar;
 pub mod error;
 pub mod oneshot;
+#[cfg(feature = "tokio-time")]
+pub mod pipe;
 pub mod semaphore;
 mod shared_state;
 pub mod unbounded;