From 2f91c85ad80c1311225ef482f6feb3c18b441edc Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Fri, 30 Aug 2019 20:46:07 -0700 Subject: [PATCH] io: bring back `split` utility (#1521) Bring back `split` utility as a free fn instead of a method on `AsyncRead`. This utility wraps the `stream` in an `Arc` and uses mutual exclusion to ensure correct access. Additionally, the specialized `split_mut` fn on TcpStream and UdsStream is promoted to `split`. --- tokio-io/src/lib.rs | 3 + tokio-io/src/split.rs | 173 ++++++++++++++++++++++++++++++++ tokio-io/tests/split.rs | 57 +++++++++++ tokio-net/src/tcp/split.rs | 171 ++++--------------------------- tokio-net/src/tcp/stream.rs | 16 +-- tokio-net/src/uds/split.rs | 131 ++++-------------------- tokio-net/src/uds/stream.rs | 16 +-- tokio-net/tests/tcp_echo.rs | 2 +- tokio-net/tests/tcp_shutdown.rs | 2 +- tokio-net/tests/tcp_split.rs | 26 +---- tokio-net/tests/uds_split.rs | 4 +- tokio/examples/connect.rs | 3 +- tokio/examples/proxy.rs | 4 +- tokio/src/io.rs | 1 + 14 files changed, 284 insertions(+), 325 deletions(-) create mode 100644 tokio-io/src/split.rs create mode 100644 tokio-io/tests/split.rs diff --git a/tokio-io/src/lib.rs b/tokio-io/src/lib.rs index dd009043a..cebe00975 100644 --- a/tokio-io/src/lib.rs +++ b/tokio-io/src/lib.rs @@ -22,6 +22,9 @@ mod async_write; #[cfg(feature = "util")] mod io; +#[cfg(feature = "util")] +pub mod split; + pub use self::async_buf_read::AsyncBufRead; pub use self::async_read::AsyncRead; pub use self::async_write::AsyncWrite; diff --git a/tokio-io/src/split.rs b/tokio-io/src/split.rs new file mode 100644 index 000000000..d13f9882b --- /dev/null +++ b/tokio-io/src/split.rs @@ -0,0 +1,173 @@ +//! Split a single value implementing `AsyncRead + AsyncWrite` into separate +//! `AsyncRead` and `AsyncWrite` handles. +//! +//! To restore this read/write object from its `split::ReadHalf` and +//! `split::WriteHalf` use `unsplit`. + +use crate::{AsyncRead, AsyncWrite}; + +use bytes::{Buf, BufMut}; +use futures_core::ready; +use std::cell::UnsafeCell; +use std::fmt; +use std::io; +use std::pin::Pin; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering::{Acquire, Release}; +use std::sync::Arc; +use std::task::{Context, Poll}; + +/// The readable half of a value returned from `split`. +pub struct ReadHalf { + inner: Arc>, +} + +/// The writable half of a value returned from `split`. +pub struct WriteHalf { + inner: Arc>, +} + +struct Inner { + locked: AtomicBool, + stream: UnsafeCell, +} + +struct Guard<'a, T> { + inner: &'a Inner, +} + +/// Split a single value implementing `AsyncRead + AsyncWrite` into separate +/// `AsyncRead` and `AsyncWrite` handles. +/// +/// To restore this read/write object from its `split::ReadHalf` and +/// `split::WriteHalf` use `unsplit`. +pub fn split(stream: T) -> (ReadHalf, WriteHalf) +where + T: AsyncRead + AsyncWrite, +{ + let inner = Arc::new(Inner { + locked: AtomicBool::new(false), + stream: UnsafeCell::new(stream), + }); + + let rd = ReadHalf { + inner: inner.clone(), + }; + + let wr = WriteHalf { inner }; + + (rd, wr) +} + +impl ReadHalf { + /// Reunite with a previously split `WriteHalf`. + /// + /// # Panics + /// + /// If this `ReadHalf` and the given `WriteHalf` do not originate from the + /// same `split` operation this method will panic. + pub fn unsplit(self, wr: WriteHalf) -> T { + if Arc::ptr_eq(&self.inner, &wr.inner) { + drop(wr); + + let inner = Arc::try_unwrap(self.inner) + .ok() + .expect("Arc::try_unwrap failed"); + + inner.stream.into_inner() + } else { + panic!("Unrelated `split::Write` passed to `split::Read::unsplit`.") + } + } +} + +impl AsyncRead for ReadHalf { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + let mut inner = ready!(self.inner.poll_lock(cx)); + inner.stream_pin().poll_read(cx, buf) + } + + fn poll_read_buf( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut B, + ) -> Poll> { + let mut inner = ready!(self.inner.poll_lock(cx)); + inner.stream_pin().poll_read_buf(cx, buf) + } +} + +impl AsyncWrite for WriteHalf { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let mut inner = ready!(self.inner.poll_lock(cx)); + inner.stream_pin().poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let mut inner = ready!(self.inner.poll_lock(cx)); + inner.stream_pin().poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let mut inner = ready!(self.inner.poll_lock(cx)); + inner.stream_pin().poll_shutdown(cx) + } + + fn poll_write_buf( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut B, + ) -> Poll> { + let mut inner = ready!(self.inner.poll_lock(cx)); + inner.stream_pin().poll_write_buf(cx, buf) + } +} + +impl Inner { + fn poll_lock(&self, cx: &mut Context<'_>) -> Poll> { + if !self.locked.compare_and_swap(false, true, Acquire) { + Poll::Ready(Guard { inner: self }) + } else { + // Spin... but investigate a better strategy + + ::std::thread::yield_now(); + cx.waker().wake_by_ref(); + + Poll::Pending + } + } +} + +impl Guard<'_, T> { + fn stream_pin(&mut self) -> Pin<&mut T> { + // safety: the stream is pinned in `Arc` and the `Guard` ensures mutual + // exclusion. + unsafe { Pin::new_unchecked(&mut *self.inner.stream.get()) } + } +} + +impl Drop for Guard<'_, T> { + fn drop(&mut self) { + self.inner.locked.store(false, Release); + } +} + +impl fmt::Debug for ReadHalf { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.debug_struct("split::ReadHalf").finish() + } +} + +impl fmt::Debug for WriteHalf { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.debug_struct("split::WriteHalf").finish() + } +} diff --git a/tokio-io/tests/split.rs b/tokio-io/tests/split.rs new file mode 100644 index 000000000..f9c0c07f3 --- /dev/null +++ b/tokio-io/tests/split.rs @@ -0,0 +1,57 @@ +use tokio::io::{split, AsyncRead, AsyncWrite}; + +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +struct RW; + +impl AsyncRead for RW { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &mut [u8], + ) -> Poll> { + Poll::Ready(Ok(1)) + } +} + +impl AsyncWrite for RW { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &[u8], + ) -> Poll> { + Poll::Ready(Ok(1)) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } +} + +#[test] +fn unsplit_ok() { + let (r, w) = split(RW); + r.unsplit(w); +} + +#[test] +#[should_panic] +fn unsplit_err1() { + let (r, _) = split(RW); + let (_, w) = split(RW); + r.unsplit(w); +} + +#[test] +#[should_panic] +fn unsplit_err2() { + let (_, w) = split(RW); + let (r, _) = split(RW); + r.unsplit(w); +} diff --git a/tokio-net/src/tcp/split.rs b/tokio-net/src/tcp/split.rs index fa6f92212..db937267d 100644 --- a/tokio-net/src/tcp/split.rs +++ b/tokio-net/src/tcp/split.rs @@ -1,137 +1,39 @@ //! `TcpStream` split support. //! -//! A `TcpStream` can be split into a `TcpStreamReadHalf` and a -//! `TcpStreamWriteHalf` with the `TcpStream::split` method. `TcpStreamReadHalf` -//! implements `AsyncRead` while `TcpStreamWriteHalf` implements `AsyncWrite`. -//! The two halves can be used concurrently, even from multiple tasks. +//! A `TcpStream` can be split into a `ReadHalf` and a +//! `WriteHalf` with the `TcpStream::split` method. `ReadHalf` +//! implements `AsyncRead` while `WriteHalf` implements `AsyncWrite`. //! //! Compared to the generic split of `AsyncRead + AsyncWrite`, this specialized -//! split gives read and write halves that are faster and smaller, because they -//! do not use locks. They also provide access to the underlying `TcpStream` -//! after split, implementing `AsRef`. This allows you to call -//! `TcpStream` methods that takes `&self`, e.g., to get local and peer -//! addresses, to get and set socket options, and to shutdown the sockets. +//! split has no associated overhead and enforces all invariants at the type +//! level. use super::TcpStream; use tokio_io::{AsyncRead, AsyncWrite}; use bytes::{Buf, BufMut}; -use std::error::Error; -use std::fmt; use std::io; use std::net::Shutdown; use std::pin::Pin; -use std::sync::Arc; use std::task::{Context, Poll}; /// Read half of a `TcpStream`. #[derive(Debug)] -pub struct TcpStreamReadHalf(Arc); +pub struct ReadHalf<'a>(&'a TcpStream); /// Write half of a `TcpStream`. /// /// Note that in the `AsyncWrite` implemenation of `TcpStreamWriteHalf`, /// `poll_shutdown` actually shuts down the TCP stream in the write direction. #[derive(Debug)] -pub struct TcpStreamWriteHalf(Arc); +pub struct WriteHalf<'a>(&'a TcpStream); -pub(crate) fn split(stream: TcpStream) -> (TcpStreamReadHalf, TcpStreamWriteHalf) { - let shared = Arc::new(stream); - ( - TcpStreamReadHalf(shared.clone()), - TcpStreamWriteHalf(shared), - ) +pub(crate) fn split(stream: &mut TcpStream) -> (ReadHalf<'_>, WriteHalf<'_>) { + (ReadHalf(&*stream), WriteHalf(&*stream)) } -/// Read half of a `TcpStream`. -#[derive(Debug)] -pub struct TcpStreamReadHalfMut<'a>(&'a TcpStream); - -/// Write half of a `TcpStream`. -/// -/// Note that in the `AsyncWrite` implemenation of `TcpStreamWriteHalf`, -/// `poll_shutdown` actually shuts down the TCP stream in the write direction. -#[derive(Debug)] -pub struct TcpStreamWriteHalfMut<'a>(&'a TcpStream); - -pub(crate) fn split_mut( - stream: &mut TcpStream, -) -> (TcpStreamReadHalfMut<'_>, TcpStreamWriteHalfMut<'_>) { - ( - TcpStreamReadHalfMut(&*stream), - TcpStreamWriteHalfMut(&*stream), - ) -} - -/// Error indicating two halves were not from the same stream, and thus could -/// not be `reunite`d. -#[derive(Debug)] -pub struct ReuniteError(pub TcpStreamReadHalf, pub TcpStreamWriteHalf); - -impl fmt::Display for ReuniteError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!( - f, - "tried to reunite halves that are not from the same stream" - ) - } -} - -impl Error for ReuniteError {} - -impl TcpStreamReadHalf { - /// Attempts to put the two "halves" of a `TcpStream` back together and - /// recover the original stream. Succeeds only if the two "halves" - /// originated from the same call to `TcpStream::split`. - pub fn reunite(self, other: TcpStreamWriteHalf) -> Result { - if Arc::ptr_eq(&self.0, &other.0) { - drop(other); - // Only two instances of the `Arc` are ever created, one for the - // reader and one for the writer, and those `Arc`s are never exposed - // externally. And so when we drop one here, the other one must be - // the only remaining one. - Ok(Arc::try_unwrap(self.0).expect("tcp: try_unwrap failed in reunite")) - } else { - Err(ReuniteError(self, other)) - } - } -} - -impl TcpStreamWriteHalf { - /// Attempts to put the two "halves" of a `TcpStream` back together and - /// recover the original stream. Succeeds only if the two "halves" - /// originated from the same call to `TcpStream::split`. - pub fn reunite(self, other: TcpStreamReadHalf) -> Result { - other.reunite(self) - } -} - -impl AsRef for TcpStreamReadHalf { - fn as_ref(&self) -> &TcpStream { - &self.0 - } -} - -impl AsRef for TcpStreamWriteHalf { - fn as_ref(&self) -> &TcpStream { - &self.0 - } -} - -impl AsRef for TcpStreamReadHalfMut<'_> { - fn as_ref(&self) -> &TcpStream { - self.0 - } -} - -impl AsRef for TcpStreamWriteHalfMut<'_> { - fn as_ref(&self) -> &TcpStream { - self.0 - } -} - -impl AsyncRead for TcpStreamReadHalf { +impl AsyncRead for ReadHalf<'_> { unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { false } @@ -153,7 +55,7 @@ impl AsyncRead for TcpStreamReadHalf { } } -impl AsyncWrite for TcpStreamWriteHalf { +impl AsyncWrite for WriteHalf<'_> { fn poll_write( self: Pin<&mut Self>, cx: &mut Context<'_>, @@ -182,53 +84,14 @@ impl AsyncWrite for TcpStreamWriteHalf { } } -impl AsyncRead for TcpStreamReadHalfMut<'_> { - unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { - false - } - - fn poll_read( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { - self.0.poll_read_priv(cx, buf) - } - - fn poll_read_buf( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll> { - self.0.poll_read_buf_priv(cx, buf) +impl AsRef for ReadHalf<'_> { + fn as_ref(&self) -> &TcpStream { + self.0 } } -impl AsyncWrite for TcpStreamWriteHalfMut<'_> { - fn poll_write( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - self.0.poll_write_priv(cx, buf) - } - - #[inline] - fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { - // tcp flush is a no-op - Poll::Ready(Ok(())) - } - - // `poll_shutdown` on a write half shutdowns the stream in the "write" direction. - fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { - self.0.shutdown(Shutdown::Write).into() - } - - fn poll_write_buf( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll> { - self.0.poll_write_buf_priv(cx, buf) +impl AsRef for WriteHalf<'_> { + fn as_ref(&self) -> &TcpStream { + self.0 } } diff --git a/tokio-net/src/tcp/stream.rs b/tokio-net/src/tcp/stream.rs index 7996d285c..b5368127f 100644 --- a/tokio-net/src/tcp/stream.rs +++ b/tokio-net/src/tcp/stream.rs @@ -1,7 +1,4 @@ -use super::split::{ - split, split_mut, TcpStreamReadHalf, TcpStreamReadHalfMut, TcpStreamWriteHalf, - TcpStreamWriteHalfMut, -}; +use super::split::{split, ReadHalf, WriteHalf}; use crate::driver::Handle; use crate::util::PollEvented; use crate::ToSocketAddrs; @@ -584,19 +581,10 @@ impl TcpStream { /// /// See the module level documenation of [`split`](super::split) for more /// details. - pub fn split(self) -> (TcpStreamReadHalf, TcpStreamWriteHalf) { + pub fn split(&mut self) -> (ReadHalf<'_>, WriteHalf<'_>) { split(self) } - /// Split a `TcpStream` into a read half and a write half, which can be used - /// to read and write the stream concurrently. - /// - /// See the module level documenation of [`split`](super::split) for more - /// details. - pub fn split_mut(&mut self) -> (TcpStreamReadHalfMut<'_>, TcpStreamWriteHalfMut<'_>) { - split_mut(self) - } - // == Poll IO functions that takes `&self` == // // They are not public because (taken from the doc of `PollEvented`): diff --git a/tokio-net/src/uds/split.rs b/tokio-net/src/uds/split.rs index 43f0c5d5e..1f0e9c1d2 100644 --- a/tokio-net/src/uds/split.rs +++ b/tokio-net/src/uds/split.rs @@ -1,15 +1,12 @@ //! `UnixStream` split support. //! -//! A `UnixStream` can be split into a read half and a write half with `UnixStream::split` -//! and `UnixStream::split_mut` methods. The read half implements `AsyncRead` while -//! the write half implements `AsyncWrite`. The two halves can be used concurrently. +//! A `UnixStream` can be split into a read half and a write half with +//! `UnixStream::split`. The read half implements `AsyncRead` while the write +//! half implements `AsyncWrite`. //! //! Compared to the generic split of `AsyncRead + AsyncWrite`, this specialized -//! split gives read and write halves that are faster and smaller, because they -//! do not use locks. They also provide access to the underlying `UnixStream` -//! after split, implementing `AsRef`. This allows you to call -//! `UnixStream` methods that takes `&self`, e.g., to get local and peer -//! addresses, to get and set socket options, and to shutdown the sockets. +//! split has no associated overhead and enforces all invariants at the type +//! level. use super::UnixStream; @@ -19,73 +16,21 @@ use bytes::{Buf, BufMut}; use std::io; use std::net::Shutdown; use std::pin::Pin; -use std::sync::Arc; use std::task::{Context, Poll}; /// Read half of a `UnixStream`. #[derive(Debug)] -pub struct UnixStreamReadHalf(Arc); +pub struct ReadHalf<'a>(&'a UnixStream); /// Write half of a `UnixStream`. -/// -/// Note that in the `AsyncWrite` implementation of `UnixStreamWriteHalf`, -/// `poll_shutdown` actually shuts down the stream in the write direction. #[derive(Debug)] -pub struct UnixStreamWriteHalf(Arc); +pub struct WriteHalf<'a>(&'a UnixStream); -/// Read half of a `UnixStream`. -#[derive(Debug)] -pub struct UnixStreamReadHalfMut<'a>(&'a UnixStream); - -/// Write half of a `UnixStream`. -/// -/// Note that in the `AsyncWrite` implementation of `UnixStreamWriteHalfMut`, -/// `poll_shutdown` actually shuts down the stream in the write direction. -#[derive(Debug)] -pub struct UnixStreamWriteHalfMut<'a>(&'a UnixStream); - -pub(crate) fn split(stream: UnixStream) -> (UnixStreamReadHalf, UnixStreamWriteHalf) { - let shared = Arc::new(stream); - ( - UnixStreamReadHalf(shared.clone()), - UnixStreamWriteHalf(shared), - ) +pub(crate) fn split(stream: &mut UnixStream) -> (ReadHalf<'_>, WriteHalf<'_>) { + (ReadHalf(stream), WriteHalf(stream)) } -pub(crate) fn split_mut( - stream: &mut UnixStream, -) -> (UnixStreamReadHalfMut<'_>, UnixStreamWriteHalfMut<'_>) { - ( - UnixStreamReadHalfMut(stream), - UnixStreamWriteHalfMut(stream), - ) -} - -impl AsRef for UnixStreamReadHalf { - fn as_ref(&self) -> &UnixStream { - &self.0 - } -} - -impl AsRef for UnixStreamWriteHalf { - fn as_ref(&self) -> &UnixStream { - &self.0 - } -} - -impl AsRef for UnixStreamReadHalfMut<'_> { - fn as_ref(&self) -> &UnixStream { - self.0 - } -} - -impl AsRef for UnixStreamWriteHalfMut<'_> { - fn as_ref(&self) -> &UnixStream { - self.0 - } -} - -impl AsyncRead for UnixStreamReadHalf { +impl AsyncRead for ReadHalf<'_> { unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { false } @@ -107,29 +52,7 @@ impl AsyncRead for UnixStreamReadHalf { } } -impl AsyncRead for UnixStreamReadHalfMut<'_> { - unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { - false - } - - fn poll_read( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { - self.0.poll_read_priv(cx, buf) - } - - fn poll_read_buf( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll> { - self.0.poll_read_buf_priv(cx, buf) - } -} - -impl AsyncWrite for UnixStreamWriteHalf { +impl AsyncWrite for WriteHalf<'_> { fn poll_write( self: Pin<&mut Self>, cx: &mut Context<'_>, @@ -155,28 +78,14 @@ impl AsyncWrite for UnixStreamWriteHalf { } } -impl AsyncWrite for UnixStreamWriteHalfMut<'_> { - fn poll_write( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - self.0.poll_write_priv(cx, buf) - } - - fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } - - fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { - self.0.shutdown(Shutdown::Write).into() - } - - fn poll_write_buf( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll> { - self.0.poll_write_buf_priv(cx, buf) +impl AsRef for ReadHalf<'_> { + fn as_ref(&self) -> &UnixStream { + self.0 + } +} + +impl AsRef for WriteHalf<'_> { + fn as_ref(&self) -> &UnixStream { + self.0 } } diff --git a/tokio-net/src/uds/stream.rs b/tokio-net/src/uds/stream.rs index 9858184a7..1164390d0 100644 --- a/tokio-net/src/uds/stream.rs +++ b/tokio-net/src/uds/stream.rs @@ -1,7 +1,4 @@ -use super::split::{ - split, split_mut, UnixStreamReadHalf, UnixStreamReadHalfMut, UnixStreamWriteHalf, - UnixStreamWriteHalfMut, -}; +use super::split::{split, ReadHalf, WriteHalf}; use super::ucred::{self, UCred}; use crate::driver::Handle; use crate::util::PollEvented; @@ -112,18 +109,9 @@ impl UnixStream { /// /// See the module level documenation of [`split`](super::split) for more /// details. - pub fn split(self) -> (UnixStreamReadHalf, UnixStreamWriteHalf) { + pub fn split(&mut self) -> (ReadHalf<'_>, WriteHalf<'_>) { split(self) } - - /// Split a `UnixStream` into a read half and a write half, which can be used - /// to read and write the stream concurrently. - /// - /// See the module level documenation of [`split`](super::split) for more - /// details. - pub fn split_mut(&mut self) -> (UnixStreamReadHalfMut<'_>, UnixStreamWriteHalfMut<'_>) { - split_mut(self) - } } impl TryFrom for mio_uds::UnixStream { diff --git a/tokio-net/tests/tcp_echo.rs b/tokio-net/tests/tcp_echo.rs index bbcd02cde..0325ecbb9 100644 --- a/tokio-net/tests/tcp_echo.rs +++ b/tokio-net/tests/tcp_echo.rs @@ -31,7 +31,7 @@ async fn echo_server() { assert_ok!(tx.send(())); }); - let (stream, _) = assert_ok!(srv.accept().await); + let (mut stream, _) = assert_ok!(srv.accept().await); let (mut rd, mut wr) = stream.split(); let n = assert_ok!(rd.copy(&mut wr).await); diff --git a/tokio-net/tests/tcp_shutdown.rs b/tokio-net/tests/tcp_shutdown.rs index f9570ff9a..7e0597c97 100644 --- a/tokio-net/tests/tcp_shutdown.rs +++ b/tokio-net/tests/tcp_shutdown.rs @@ -20,7 +20,7 @@ async fn shutdown() { assert_eq!(n, 0); }); - let (stream, _) = assert_ok!(srv.accept().await); + let (mut stream, _) = assert_ok!(srv.accept().await); let (mut rd, mut wr) = stream.split(); let n = assert_ok!(rd.copy(&mut wr).await); diff --git a/tokio-net/tests/tcp_split.rs b/tokio-net/tests/tcp_split.rs index 269dd895f..ae5f249c0 100644 --- a/tokio-net/tests/tcp_split.rs +++ b/tokio-net/tests/tcp_split.rs @@ -1,25 +1 @@ -use tokio_net::tcp::{TcpListener, TcpStream}; - -#[tokio::test] -async fn split_reunite() -> std::io::Result<()> { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let addr = listener.local_addr()?; - let stream = TcpStream::connect(&addr).await?; - - let (r, w) = stream.split(); - assert!(r.reunite(w).is_ok()); - Ok(()) -} - -#[tokio::test] -async fn split_reunite_error() -> std::io::Result<()> { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let addr = listener.local_addr()?; - let stream = TcpStream::connect(&addr).await?; - let stream1 = TcpStream::connect(&addr).await?; - - let (r, _) = stream.split(); - let (_, w) = stream1.split(); - assert!(r.reunite(w).is_err()); - Ok(()) -} +// TODO: write tests using TcpStream::split() diff --git a/tokio-net/tests/uds_split.rs b/tokio-net/tests/uds_split.rs index be50f773f..46602d5da 100644 --- a/tokio-net/tests/uds_split.rs +++ b/tokio-net/tests/uds_split.rs @@ -11,10 +11,10 @@ use tokio::prelude::*; /// writing by reading to the end of stream on the other side of the connection. #[tokio::test] async fn split() -> std::io::Result<()> { - let (a, mut b) = UnixStream::pair()?; + let (mut a, mut b) = UnixStream::pair()?; let (mut a_read, mut a_write) = a.split(); - let (mut b_read, mut b_write) = b.split_mut(); + let (mut b_read, mut b_write) = b.split(); let (a_response, b_response) = futures::future::try_join( send_recv_all(&mut a_read, &mut a_write, b"A"), diff --git a/tokio/examples/connect.rs b/tokio/examples/connect.rs index 0ba8a6f2d..c53471d89 100644 --- a/tokio/examples/connect.rs +++ b/tokio/examples/connect.rs @@ -93,7 +93,8 @@ mod tcp { stdin: impl Stream, io::Error>> + Unpin, mut stdout: impl Sink, Error = io::Error> + Unpin, ) -> Result<(), Box> { - let (r, w) = TcpStream::connect(addr).await?.split(); + let mut stream = TcpStream::connect(addr).await?; + let (r, w) = stream.split(); let sink = FramedWrite::new(w, codec::Bytes); let mut stream = FramedRead::new(r, codec::Bytes).filter_map(|i| match i { Ok(i) => future::ready(Some(i)), diff --git a/tokio/examples/proxy.rs b/tokio/examples/proxy.rs index 7946be8a9..6886a813b 100644 --- a/tokio/examples/proxy.rs +++ b/tokio/examples/proxy.rs @@ -52,8 +52,8 @@ async fn main() -> Result<(), Box> { Ok(()) } -async fn transfer(inbound: TcpStream, proxy_addr: String) -> Result<(), Box> { - let outbound = TcpStream::connect(proxy_addr).await?; +async fn transfer(mut inbound: TcpStream, proxy_addr: String) -> Result<(), Box> { + let mut outbound = TcpStream::connect(proxy_addr).await?; let (mut ri, mut wi) = inbound.split(); let (mut ro, mut wo) = outbound.split(); diff --git a/tokio/src/io.rs b/tokio/src/io.rs index 0183d28bf..f45906e1e 100644 --- a/tokio/src/io.rs +++ b/tokio/src/io.rs @@ -39,6 +39,7 @@ // standard input, output, and error #[cfg(feature = "fs")] pub use tokio_fs::{stderr, stdin, stdout, Stderr, Stdin, Stdout}; +pub use tokio_io::split::split; pub use tokio_io::{ AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter,