io: add io::duplex() as bidirectional reader/writer (#2661)

`duplex` returns a pair of connected `DuplexStream`s.

`DuplexStream` is a bidirectional type that can be used to simulate IO,
but over an in-process piece of memory.

tokio/tests/io_mem_stream.rs

@@ -0,0 +1,83 @@
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
+
+#[tokio::test]
+async fn ping_pong() {
+    let (mut a, mut b) = duplex(32);
+
+    let mut buf = [0u8; 4];
+
+    a.write_all(b"ping").await.unwrap();
+    b.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, b"ping");
+
+    b.write_all(b"pong").await.unwrap();
+    a.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, b"pong");
+}
+
+#[tokio::test]
+async fn across_tasks() {
+    let (mut a, mut b) = duplex(32);
+
+    let t1 = tokio::spawn(async move {
+        a.write_all(b"ping").await.unwrap();
+        let mut buf = [0u8; 4];
+        a.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"pong");
+    });
+
+    let t2 = tokio::spawn(async move {
+        let mut buf = [0u8; 4];
+        b.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"ping");
+        b.write_all(b"pong").await.unwrap();
+    });
+
+    t1.await.unwrap();
+    t2.await.unwrap();
+}
+
+#[tokio::test]
+async fn disconnect() {
+    let (mut a, mut b) = duplex(32);
+
+    let t1 = tokio::spawn(async move {
+        a.write_all(b"ping").await.unwrap();
+        // and dropped
+    });
+
+    let t2 = tokio::spawn(async move {
+        let mut buf = [0u8; 32];
+        let n = b.read(&mut buf).await.unwrap();
+        assert_eq!(&buf[..n], b"ping");
+
+        let n = b.read(&mut buf).await.unwrap();
+        assert_eq!(n, 0);
+    });
+
+    t1.await.unwrap();
+    t2.await.unwrap();
+}
+
+#[tokio::test]
+async fn max_write_size() {
+    let (mut a, mut b) = duplex(32);
+
+    let t1 = tokio::spawn(async move {
+        let n = a.write(&[0u8; 64]).await.unwrap();
+        assert_eq!(n, 32);
+        let n = a.write(&[0u8; 64]).await.unwrap();
+        assert_eq!(n, 4);
+    });
+
+    let t2 = tokio::spawn(async move {
+        let mut buf = [0u8; 4];
+        b.read_exact(&mut buf).await.unwrap();
+    });
+
+    t1.await.unwrap();
+    t2.await.unwrap();
+}