net: move into tokio crate (#1683)

A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).

The `net` implementation is now provided by the main `tokio` crate.
Functionality can be opted out of by using the various net related
feature flags.

7 files changed, 1004 insertions, 0 deletions

diff --git a/tokio/src/net/unix/datagram.rs b/tokio/src/net/unix/datagram.rs
new file mode 100644
index 00000000..f9c47dec
--- /dev/null
+++ b/tokio/src/net/unix/datagram.rs
@@ -0,0 +1,233 @@
+use crate::net::util::PollEvented;
+
+use futures_core::ready;
+use futures_util::future::poll_fn;
+use std::convert::TryFrom;
+use std::fmt;
+use std::io;
+use std::net::Shutdown;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::os::unix::net::{self, SocketAddr};
+use std::path::Path;
+use std::task::{Context, Poll};
+
+/// An I/O object representing a Unix datagram socket.
+pub struct UnixDatagram {
+    io: PollEvented<mio_uds::UnixDatagram>,
+}
+
+impl UnixDatagram {
+    /// Creates a new `UnixDatagram` bound to the specified path.
+    pub fn bind<P>(path: P) -> io::Result<UnixDatagram>
+    where
+        P: AsRef<Path>,
+    {
+        let socket = mio_uds::UnixDatagram::bind(path)?;
+        UnixDatagram::new(socket)
+    }
+
+    /// Creates an unnamed pair of connected sockets.
+    ///
+    /// This function will create a pair of interconnected Unix sockets for
+    /// communicating back and forth between one another. Each socket will
+    /// be associated with the default event loop's handle.
+    pub fn pair() -> io::Result<(UnixDatagram, UnixDatagram)> {
+        let (a, b) = mio_uds::UnixDatagram::pair()?;
+        let a = UnixDatagram::new(a)?;
+        let b = UnixDatagram::new(b)?;
+
+        Ok((a, b))
+    }
+
+    /// Consumes a `UnixDatagram` in the standard library and returns a
+    /// nonblocking `UnixDatagram` from this crate.
+    ///
+    /// The returned datagram will be associated with the given event loop
+    /// specified by `handle` and is ready to perform I/O.
+    pub fn from_std(datagram: net::UnixDatagram) -> io::Result<UnixDatagram> {
+        let socket = mio_uds::UnixDatagram::from_datagram(datagram)?;
+        let io = PollEvented::new(socket)?;
+        Ok(UnixDatagram { io })
+    }
+
+    fn new(socket: mio_uds::UnixDatagram) -> io::Result<UnixDatagram> {
+        let io = PollEvented::new(socket)?;
+        Ok(UnixDatagram { io })
+    }
+
+    /// Creates a new `UnixDatagram` which is not bound to any address.
+    pub fn unbound() -> io::Result<UnixDatagram> {
+        let socket = mio_uds::UnixDatagram::unbound()?;
+        UnixDatagram::new(socket)
+    }
+
+    /// Connects the socket to the specified address.
+    ///
+    /// The `send` method may be used to send data to the specified address.
+    /// `recv` and `recv_from` will only receive data from that address.
+    pub fn connect<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
+        self.io.get_ref().connect(path)
+    }
+
+    /// Sends data on the socket to the socket's peer.
+    pub async fn send(&mut self, buf: &[u8]) -> io::Result<usize> {
+        poll_fn(|cx| self.poll_send_priv(cx, buf)).await
+    }
+
+    // Poll IO functions that takes `&self` are provided for the split API.
+    //
+    // They are not public because (taken from the doc of `PollEvented`):
+    //
+    // While `PollEvented` is `Sync` (if the underlying I/O type is `Sync`), the
+    // caller must ensure that there are at most two tasks that use a
+    // `PollEvented` instance concurrently. One for reading and one for writing.
+    // While violating this requirement is "safe" from a Rust memory model point
+    // of view, it will result in unexpected behavior in the form of lost
+    // notifications and tasks hanging.
+    pub(crate) fn poll_send_priv(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.io.poll_write_ready(cx))?;
+
+        match self.io.get_ref().send(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_write_ready(cx)?;
+                Poll::Pending
+            }
+            x => Poll::Ready(x),
+        }
+    }
+
+    /// Receives data from the socket.
+    pub async fn recv(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        poll_fn(|cx| self.poll_recv_priv(cx, buf)).await
+    }
+
+    pub(crate) fn poll_recv_priv(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
+
+        match self.io.get_ref().recv(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_read_ready(cx, mio::Ready::readable())?;
+                Poll::Pending
+            }
+            x => Poll::Ready(x),
+        }
+    }
+
+    /// Sends data on the socket to the specified address.
+    pub async fn send_to<P>(&mut self, buf: &[u8], target: P) -> io::Result<usize>
+    where
+        P: AsRef<Path> + Unpin,
+    {
+        poll_fn(|cx| self.poll_send_to_priv(cx, buf, target.as_ref())).await
+    }
+
+    pub(crate) fn poll_send_to_priv(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+        target: &Path,
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.io.poll_write_ready(cx))?;
+
+        match self.io.get_ref().send_to(buf, target) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_write_ready(cx)?;
+                Poll::Pending
+            }
+            x => Poll::Ready(x),
+        }
+    }
+
+    /// Receives data from the socket.
+    pub async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
+        poll_fn(|cx| self.poll_recv_from_priv(cx, buf)).await
+    }
+
+    pub(crate) fn poll_recv_from_priv(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<Result<(usize, SocketAddr), io::Error>> {
+        ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
+
+        match self.io.get_ref().recv_from(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_read_ready(cx, mio::Ready::readable())?;
+                Poll::Pending
+            }
+            x => Poll::Ready(x),
+        }
+    }
+
+    /// Returns the local address that this socket is bound to.
+    pub fn local_addr(&self) -> io::Result<SocketAddr> {
+        self.io.get_ref().local_addr()
+    }
+
+    /// Returns the address of this socket's peer.
+    ///
+    /// The `connect` method will connect the socket to a peer.
+    pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+        self.io.get_ref().peer_addr()
+    }
+
+    /// Returns the value of the `SO_ERROR` option.
+    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+        self.io.get_ref().take_error()
+    }
+
+    /// Shut down the read, write, or both halves of this connection.
+    ///
+    /// This function will cause all pending and future I/O calls on the
+    /// specified portions to immediately return with an appropriate value
+    /// (see the documentation of `Shutdown`).
+    pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
+        self.io.get_ref().shutdown(how)
+    }
+}
+
+impl TryFrom<UnixDatagram> for mio_uds::UnixDatagram {
+    type Error = io::Error;
+
+    /// Consumes value, returning the mio I/O object.
+    ///
+    /// See [`PollEvented::into_inner`] for more details about
+    /// resource deregistration that happens during the call.
+    ///
+    /// [`PollEvented::into_inner`]: crate::util::PollEvented::into_inner
+    fn try_from(value: UnixDatagram) -> Result<Self, Self::Error> {
+        value.io.into_inner()
+    }
+}
+
+impl TryFrom<net::UnixDatagram> for UnixDatagram {
+    type Error = io::Error;
+
+    /// Consumes stream, returning the tokio I/O object.
+    ///
+    /// This is equivalent to
+    /// [`UnixDatagram::from_std(stream)`](UnixDatagram::from_std).
+    fn try_from(stream: net::UnixDatagram) -> Result<Self, Self::Error> {
+        Self::from_std(stream)
+    }
+}
+
+impl fmt::Debug for UnixDatagram {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.io.get_ref().fmt(f)
+    }
+}
+
+impl AsRawFd for UnixDatagram {
+    fn as_raw_fd(&self) -> RawFd {
+        self.io.get_ref().as_raw_fd()
+    }
+}
diff --git a/tokio/src/net/unix/incoming.rs b/tokio/src/net/unix/incoming.rs
new file mode 100644
index 00000000..542b5e1d
--- /dev/null
+++ b/tokio/src/net/unix/incoming.rs
@@ -0,0 +1,31 @@
+#![cfg(feature = "async-traits")]
+
+use super::{UnixListener, UnixStream};
+
+use futures_core::ready;
+use futures_core::stream::Stream;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Stream of listeners
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Incoming {
+    inner: UnixListener,
+}
+
+impl Incoming {
+    pub(crate) fn new(listener: UnixListener) -> Incoming {
+        Incoming { inner: listener }
+    }
+}
+
+impl Stream for Incoming {
+    type Item = io::Result<UnixStream>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let (socket, _) = ready!(Pin::new(&mut self.inner).poll_accept(cx))?;
+        Poll::Ready(Some(Ok(socket)))
+    }
+}
diff --git a/tokio/src/net/unix/listener.rs b/tokio/src/net/unix/listener.rs
new file mode 100644
index 00000000..3a36dc90
--- /dev/null
+++ b/tokio/src/net/unix/listener.rs
@@ -0,0 +1,135 @@
+use crate::net::unix::UnixStream;
+use crate::net::util::PollEvented;
+
+use futures_core::ready;
+use futures_util::future::poll_fn;
+use mio::Ready;
+use mio_uds;
+use std::convert::TryFrom;
+use std::fmt;
+use std::io;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::os::unix::net::{self, SocketAddr};
+use std::path::Path;
+use std::task::{Context, Poll};
+
+/// A Unix socket which can accept connections from other Unix sockets.
+pub struct UnixListener {
+    io: PollEvented<mio_uds::UnixListener>,
+}
+
+impl UnixListener {
+    /// Creates a new `UnixListener` bound to the specified path.
+    pub fn bind<P>(path: P) -> io::Result<UnixListener>
+    where
+        P: AsRef<Path>,
+    {
+        let listener = mio_uds::UnixListener::bind(path)?;
+        let io = PollEvented::new(listener)?;
+        Ok(UnixListener { io })
+    }
+
+    /// Consumes a `UnixListener` in the standard library and returns a
+    /// nonblocking `UnixListener` from this crate.
+    ///
+    /// The returned listener will be associated with the given event loop
+    /// specified by `handle` and is ready to perform I/O.
+    pub fn from_std(listener: net::UnixListener) -> io::Result<UnixListener> {
+        let listener = mio_uds::UnixListener::from_listener(listener)?;
+        let io = PollEvented::new(listener)?;
+        Ok(UnixListener { io })
+    }
+
+    /// Returns the local socket address of this listener.
+    pub fn local_addr(&self) -> io::Result<SocketAddr> {
+        self.io.get_ref().local_addr()
+    }
+
+    /// Returns the value of the `SO_ERROR` option.
+    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+        self.io.get_ref().take_error()
+    }
+
+    /// Accepts a new incoming connection to this listener.
+    pub async fn accept(&mut self) -> io::Result<(UnixStream, SocketAddr)> {
+        poll_fn(|cx| self.poll_accept(cx)).await
+    }
+
+    pub(crate) fn poll_accept(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<(UnixStream, SocketAddr)>> {
+        let (io, addr) = ready!(self.poll_accept_std(cx))?;
+
+        let io = mio_uds::UnixStream::from_stream(io)?;
+        Ok((UnixStream::new(io)?, addr)).into()
+    }
+
+    fn poll_accept_std(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<(net::UnixStream, SocketAddr)>> {
+        ready!(self.io.poll_read_ready(cx, Ready::readable()))?;
+
+        match self.io.get_ref().accept_std() {
+            Ok(None) => {
+                self.io.clear_read_ready(cx, Ready::readable())?;
+                Poll::Pending
+            }
+            Ok(Some((sock, addr))) => Ok((sock, addr)).into(),
+            Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_read_ready(cx, Ready::readable())?;
+                Poll::Pending
+            }
+            Err(err) => Err(err).into(),
+        }
+    }
+
+    /// Consumes this listener, returning a stream of the sockets this listener
+    /// accepts.
+    ///
+    /// This method returns an implementation of the `Stream` trait which
+    /// resolves to the sockets the are accepted on this listener.
+    #[cfg(feature = "async-traits")]
+    pub fn incoming(self) -> super::Incoming {
+        super::Incoming::new(self)
+    }
+}
+
+impl TryFrom<UnixListener> for mio_uds::UnixListener {
+    type Error = io::Error;
+
+    /// Consumes value, returning the mio I/O object.
+    ///
+    /// See [`PollEvented::into_inner`] for more details about
+    /// resource deregistration that happens during the call.
+    ///
+    /// [`PollEvented::into_inner`]: crate::util::PollEvented::into_inner
+    fn try_from(value: UnixListener) -> Result<Self, Self::Error> {
+        value.io.into_inner()
+    }
+}
+
+impl TryFrom<net::UnixListener> for UnixListener {
+    type Error = io::Error;
+
+    /// Consumes stream, returning the tokio I/O object.
+    ///
+    /// This is equivalent to
+    /// [`UnixListener::from_std(stream)`](UnixListener::from_std).
+    fn try_from(stream: net::UnixListener) -> io::Result<Self> {
+        Self::from_std(stream)
+    }
+}
+
+impl fmt::Debug for UnixListener {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.io.get_ref().fmt(f)
+    }
+}
+
+impl AsRawFd for UnixListener {
+    fn as_raw_fd(&self) -> RawFd {
+        self.io.get_ref().as_raw_fd()
+    }
+}
diff --git a/tokio/src/net/unix/mod.rs b/tokio/src/net/unix/mod.rs
new file mode 100644
index 00000000..4447ca5c
--- /dev/null
+++ b/tokio/src/net/unix/mod.rs
@@ -0,0 +1,21 @@
+//! Unix Domain Sockets for Tokio.
+//!
+//! This crate provides APIs for using Unix Domain Sockets with Tokio.
+
+mod datagram;
+pub use self::datagram::UnixDatagram;
+
+mod incoming;
+#[cfg(feature = "async-traits")]
+pub use self::incoming::Incoming;
+
+mod listener;
+pub use self::listener::UnixListener;
+
+pub mod split;
+
+mod stream;
+pub use self::stream::UnixStream;
+
+mod ucred;
+pub use self::ucred::UCred;
diff --git a/tokio/src/net/unix/split.rs b/tokio/src/net/unix/split.rs
new file mode 100644
index 00000000..1f0e9c1d
--- /dev/null
+++ b/tokio/src/net/unix/split.rs
@@ -0,0 +1,91 @@
+//! `UnixStream` split support.
+//!
+//! 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 has no associated overhead and enforces all invariants at the type
+//! level.
+
+use super::UnixStream;
+
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use bytes::{Buf, BufMut};
+use std::io;
+use std::net::Shutdown;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Read half of a `UnixStream`.
+#[derive(Debug)]
+pub struct ReadHalf<'a>(&'a UnixStream);
+
+/// Write half of a `UnixStream`.
+#[derive(Debug)]
+pub struct WriteHalf<'a>(&'a UnixStream);
+
+pub(crate) fn split(stream: &mut UnixStream) -> (ReadHalf<'_>, WriteHalf<'_>) {
+    (ReadHalf(stream), WriteHalf(stream))
+}
+
+impl AsyncRead for ReadHalf<'_> {
+    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
+        false
+    }
+
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.0.poll_read_priv(cx, buf)
+    }
+
+    fn poll_read_buf<B: BufMut>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
+        self.0.poll_read_buf_priv(cx, buf)
+    }
+}
+
+impl AsyncWrite for WriteHalf<'_> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.0.poll_write_priv(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.0.shutdown(Shutdown::Write).into()
+    }
+
+    fn poll_write_buf<B: Buf>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
+        self.0.poll_write_buf_priv(cx, buf)
+    }
+}
+
+impl AsRef<UnixStream> for ReadHalf<'_> {
+    fn as_ref(&self) -> &UnixStream {
+        self.0
+    }
+}
+
+impl AsRef<UnixStream> for WriteHalf<'_> {
+    fn as_ref(&self) -> &UnixStream {
+        self.0
+    }
+}
diff --git a/tokio/src/net/unix/stream.rs b/tokio/src/net/unix/stream.rs
new file mode 100644
index 00000000..f2ce7a1f
--- /dev/null
+++ b/tokio/src/net/unix/stream.rs
@@ -0,0 +1,339 @@
+use crate::net::unix::split::{split, ReadHalf, WriteHalf};
+use crate::net::unix::ucred::{self, UCred};
+use crate::net::util::PollEvented;
+
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use bytes::{Buf, BufMut};
+use futures_core::ready;
+use futures_util::future::poll_fn;
+use iovec::IoVec;
+use std::convert::TryFrom;
+use std::fmt;
+use std::io::{self, Read, Write};
+use std::net::Shutdown;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::os::unix::net::{self, SocketAddr};
+use std::path::Path;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A structure representing a connected Unix socket.
+///
+/// This socket can be connected directly with `UnixStream::connect` or accepted
+/// from a listener with `UnixListener::incoming`. Additionally, a pair of
+/// anonymous Unix sockets can be created with `UnixStream::pair`.
+pub struct UnixStream {
+    io: PollEvented<mio_uds::UnixStream>,
+}
+
+impl UnixStream {
+    /// Connects to the socket named by `path`.
+    ///
+    /// This function will create a new Unix socket and connect to the path
+    /// specified, associating the returned stream with the default event loop's
+    /// handle.
+    pub async fn connect<P>(path: P) -> io::Result<UnixStream>
+    where
+        P: AsRef<Path>,
+    {
+        let stream = mio_uds::UnixStream::connect(path)?;
+        let stream = UnixStream::new(stream)?;
+
+        poll_fn(|cx| stream.io.poll_write_ready(cx)).await?;
+        Ok(stream)
+    }
+
+    /// Consumes a `UnixStream` in the standard library and returns a
+    /// nonblocking `UnixStream` from this crate.
+    ///
+    /// The returned stream will be associated with the given event loop
+    /// specified by `handle` and is ready to perform I/O.
+    pub fn from_std(stream: net::UnixStream) -> io::Result<UnixStream> {
+        let stream = mio_uds::UnixStream::from_stream(stream)?;
+        let io = PollEvented::new(stream)?;
+
+        Ok(UnixStream { io })
+    }
+
+    /// Creates an unnamed pair of connected sockets.
+    ///
+    /// This function will create a pair of interconnected Unix sockets for
+    /// communicating back and forth between one another. Each socket will
+    /// be associated with the default event loop's handle.
+    pub fn pair() -> io::Result<(UnixStream, UnixStream)> {
+        let (a, b) = mio_uds::UnixStream::pair()?;
+        let a = UnixStream::new(a)?;
+        let b = UnixStream::new(b)?;
+
+        Ok((a, b))
+    }
+
+    pub(crate) fn new(stream: mio_uds::UnixStream) -> io::Result<UnixStream> {
+        let io = PollEvented::new(stream)?;
+        Ok(UnixStream { io })
+    }
+
+    /// Returns the socket address of the local half of this connection.
+    pub fn local_addr(&self) -> io::Result<SocketAddr> {
+        self.io.get_ref().local_addr()
+    }
+
+    /// Returns the socket address of the remote half of this connection.
+    pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+        self.io.get_ref().peer_addr()
+    }
+
+    /// Returns effective credentials of the process which called `connect` or `pair`.
+    pub fn peer_cred(&self) -> io::Result<UCred> {
+        ucred::get_peer_cred(self)
+    }
+
+    /// Returns the value of the `SO_ERROR` option.
+    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+        self.io.get_ref().take_error()
+    }
+
+    /// Shuts down the read, write, or both halves of this connection.
+    ///
+    /// This function will cause all pending and future I/O calls on the
+    /// specified portions to immediately return with an appropriate value
+    /// (see the documentation of `Shutdown`).
+    pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
+        self.io.get_ref().shutdown(how)
+    }
+
+    /// 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 self) -> (ReadHalf<'_>, WriteHalf<'_>) {
+        split(self)
+    }
+}
+
+impl TryFrom<UnixStream> for mio_uds::UnixStream {
+    type Error = io::Error;
+
+    /// Consumes value, returning the mio I/O object.
+    ///
+    /// See [`PollEvented::into_inner`] for more details about
+    /// resource deregistration that happens during the call.
+    ///
+    /// [`PollEvented::into_inner`]: crate::util::PollEvented::into_inner
+    fn try_from(value: UnixStream) -> Result<Self, Self::Error> {
+        value.io.into_inner()
+    }
+}
+
+impl TryFrom<net::UnixStream> for UnixStream {
+    type Error = io::Error;
+
+    /// Consumes stream, returning the tokio I/O object.
+    ///
+    /// This is equivalent to
+    /// [`UnixStream::from_std(stream)`](UnixStream::from_std).
+    fn try_from(stream: net::UnixStream) -> io::Result<Self> {
+        Self::from_std(stream)
+    }
+}
+
+impl AsyncRead for UnixStream {
+    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
+        false
+    }
+
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.poll_read_priv(cx, buf)
+    }
+
+    fn poll_read_buf<B: BufMut>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
+        self.poll_read_buf_priv(cx, buf)
+    }
+}
+
+impl AsyncWrite for UnixStream {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.poll_write_priv(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_write_buf<B: Buf>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
+        self.poll_write_buf_priv(cx, buf)
+    }
+}
+
+impl UnixStream {
+    // == Poll IO functions that takes `&self` ==
+    //
+    // They are not public because (taken from the doc of `PollEvented`):
+    //
+    // While `PollEvented` is `Sync` (if the underlying I/O type is `Sync`), the
+    // caller must ensure that there are at most two tasks that use a
+    // `PollEvented` instance concurrently. One for reading and one for writing.
+    // While violating this requirement is "safe" from a Rust memory model point
+    // of view, it will result in unexpected behavior in the form of lost
+    // notifications and tasks hanging.
+
+    pub(crate) fn poll_read_priv(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
+
+        match self.io.get_ref().read(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_read_ready(cx, mio::Ready::readable())?;
+                Poll::Pending
+            }
+            x => Poll::Ready(x),
+        }
+    }
+
+    pub(crate) fn poll_read_buf_priv<B: BufMut>(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
+
+        let r = unsafe {
+            // The `IoVec` type can't have a 0-length size, so we create a bunch
+            // of dummy versions on the stack with 1 length which we'll quickly
+            // overwrite.
+            let b1: &mut [u8] = &mut [0];
+            let b2: &mut [u8] = &mut [0];
+            let b3: &mut [u8] = &mut [0];
+            let b4: &mut [u8] = &mut [0];
+            let b5: &mut [u8] = &mut [0];
+            let b6: &mut [u8] = &mut [0];
+            let b7: &mut [u8] = &mut [0];
+            let b8: &mut [u8] = &mut [0];
+            let b9: &mut [u8] = &mut [0];
+            let b10: &mut [u8] = &mut [0];
+            let b11: &mut [u8] = &mut [0];
+            let b12: &mut [u8] = &mut [0];
+            let b13: &mut [u8] = &mut [0];
+            let b14: &mut [u8] = &mut [0];
+            let b15: &mut [u8] = &mut [0];
+            let b16: &mut [u8] = &mut [0];
+            let mut bufs: [&mut IoVec; 16] = [
+                b1.into(),
+                b2.into(),
+                b3.into(),
+                b4.into(),
+                b5.into(),
+                b6.into(),
+                b7.into(),
+                b8.into(),
+                b9.into(),
+                b10.into(),
+                b11.into(),
+                b12.into(),
+                b13.into(),
+                b14.into(),
+                b15.into(),
+                b16.into(),
+            ];
+            let n = buf.bytes_vec_mut(&mut bufs);
+            self.io.get_ref().read_bufs(&mut bufs[..n])
+        };
+
+        match r {
+            Ok(n) => {
+                unsafe {
+                    buf.advance_mut(n);
+                }
+                Poll::Ready(Ok(n))
+            }
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_read_ready(cx, mio::Ready::readable())?;
+                Poll::Pending
+            }
+            Err(e) => Poll::Ready(Err(e)),
+        }
+    }
+
+    pub(crate) fn poll_write_priv(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.io.poll_write_ready(cx))?;
+
+        match self.io.get_ref().write(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_write_ready(cx)?;
+                Poll::Pending
+            }
+            x => Poll::Ready(x),
+        }
+    }
+
+    pub(crate) fn poll_write_buf_priv<B: Buf>(
+        &self,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.io.poll_write_ready(cx))?;
+
+        let r = {
+            // The `IoVec` type can't have a zero-length size, so create a dummy
+            // version from a 1-length slice which we'll overwrite with the
+            // `bytes_vec` method.
+            static DUMMY: &[u8] = &[0];
+            let iovec = <&IoVec>::from(DUMMY);
+            let mut bufs = [iovec; 64];
+            let n = buf.bytes_vec(&mut bufs);
+            self.io.get_ref().write_bufs(&bufs[..n])
+        };
+        match r {
+            Ok(n) => {
+                buf.advance(n);
+                Poll::Ready(Ok(n))
+            }
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                self.io.clear_write_ready(cx)?;
+                Poll::Pending
+            }
+            Err(e) => Poll::Ready(Err(e)),
+        }
+    }
+}
+
+impl fmt::Debug for UnixStream {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.io.get_ref().fmt(f)
+    }
+}
+
+impl AsRawFd for UnixStream {
+    fn as_raw_fd(&self) -> RawFd {
+        self.io.get_ref().as_raw_fd()
+    }
+}
diff --git a/tokio/src/net/unix/ucred.rs b/tokio/src/net/unix/ucred.rs
new file mode 100644
index 00000000..c785e68e
--- /dev/null
+++ b/tokio/src/net/unix/ucred.rs
@@ -0,0 +1,154 @@
+use libc::{gid_t, uid_t};
+
+/// Credentials of a process
+#[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)]
+pub struct UCred {
+    /// UID (user ID) of the process
+    pub uid: uid_t,
+    /// GID (group ID) of the process
+    pub gid: gid_t,
+}
+
+#[cfg(any(target_os = "linux", target_os = "android"))]
+pub(crate) use self::impl_linux::get_peer_cred;
+
+#[cfg(any(
+    target_os = "dragonfly",
+    target_os = "macos",
+    target_os = "ios",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd"
+))]
+pub(crate) use self::impl_macos::get_peer_cred;
+
+#[cfg(any(target_os = "solaris"))]
+pub(crate) use self::impl_solaris::get_peer_cred;
+
+#[cfg(any(target_os = "linux", target_os = "android"))]
+pub(crate) mod impl_linux {
+    use crate::net::unix::UnixStream;
+
+    use libc::{c_void, getsockopt, socklen_t, SOL_SOCKET, SO_PEERCRED};
+    use std::{io, mem};
+
+    use libc::ucred;
+
+    pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
+        use std::os::unix::io::AsRawFd;
+
+        unsafe {
+            let raw_fd = sock.as_raw_fd();
+
+            let mut ucred = ucred {
+                pid: 0,
+                uid: 0,
+                gid: 0,
+            };
+
+            let ucred_size = mem::size_of::<ucred>();
+
+            // These paranoid checks should be optimized-out
+            assert!(mem::size_of::<u32>() <= mem::size_of::<usize>());
+            assert!(ucred_size <= u32::max_value() as usize);
+
+            let mut ucred_size = ucred_size as socklen_t;
+
+            let ret = getsockopt(
+                raw_fd,
+                SOL_SOCKET,
+                SO_PEERCRED,
+                &mut ucred as *mut ucred as *mut c_void,
+                &mut ucred_size,
+            );
+            if ret == 0 && ucred_size as usize == mem::size_of::<ucred>() {
+                Ok(super::UCred {
+                    uid: ucred.uid,
+                    gid: ucred.gid,
+                })
+            } else {
+                Err(io::Error::last_os_error())
+            }
+        }
+    }
+}
+
+#[cfg(any(
+    target_os = "dragonfly",
+    target_os = "macos",
+    target_os = "ios",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd"
+))]
+pub(crate) mod impl_macos {
+    use crate::net::unix::UnixStream;
+
+    use libc::getpeereid;
+    use std::io;
+    use std::mem::MaybeUninit;
+    use std::os::unix::io::AsRawFd;
+
+    pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
+        unsafe {
+            let raw_fd = sock.as_raw_fd();
+
+            let mut uid = MaybeUninit::uninit();
+            let mut gid = MaybeUninit::uninit();
+
+            let ret = getpeereid(raw_fd, uid.as_mut_ptr(), gid.as_mut_ptr());
+
+            if ret == 0 {
+                Ok(super::UCred {
+                    uid: uid.assume_init(),
+                    gid: gid.assume_init(),
+                })
+            } else {
+                Err(io::Error::last_os_error())
+            }
+        }
+    }
+}
+
+#[cfg(any(target_os = "solaris"))]
+pub(crate) mod impl_solaris {
+    use std::io;
+    use std::os::unix::io::AsRawFd;
+    use std::ptr;
+    use UnixStream;
+
+    #[allow(non_camel_case_types)]
+    enum ucred_t {}
+
+    extern "C" {
+        fn ucred_free(cred: