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-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);

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);

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()

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"),