Repository: tokio-rs/mio
Pull request #1130 (2019-11-08 05:23:33)

Wrap kqueue::Events so that it can be Send and Sync
## Motivation

Both `Poll` and `Registry` are `Send` and `Sync`. `Events` is passed as an
argument to `Poll::poll`, and is intended to be created once--at the same time
as `Poll`--and reused on each call. It would be expected that `Events` is
therefore `Send` and `Sync` but that is currently not the case in kqueue.

#983 changed the "events" type (`KeventList` -> `Events`) to have a named field,
but also [dropped](https://github.com/tokio-rs/mio/pull/983/files#diff-1b9651542d08c6eca04e6025b1c6fd53L330) the `Send` and `Sync` impls.

For epoll events this has been okay because these marker traits can be derived
from the fields within `epoll_event`, but `kevent` has a `*mut T` field making
this not possible.

## Solution

`kqueue::Events` is now a smart pointer wrapping a `Vec<libc::kevent>` and impls
`Send` and `Sync`. `Deref` and `DerefMut` traits have also been implemented
which reduce changes in the rest of `sys`.

Assertions for `Events` have been added to the `assert_send`/`assert_sync` test
as well.

Signed-off-by: Kevin Leimkuhler <kleimkuhler@icloud.com>


Patch:
diff --git a/src/net/tcp/listener.rs b/src/net/tcp/listener.rs
index 7e0e815..d3ab37f 100644
--- a/src/net/tcp/listener.rs
+++ b/src/net/tcp/listener.rs
@@ -5,9 +5,12 @@ use crate::{event, sys, Interests, Registry, Token};
 
 use std::fmt;
 use std::io;
+use std::net;
 use std::net::SocketAddr;
 #[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+#[cfg(windows)]
+use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
 
 /// A structure representing a socket server
 ///
@@ -58,6 +61,21 @@ impl TcpListener {
         })
     }
 
+    /// Creates a new `TcpListener` from a standard `net::TcpListener`.
+    ///
+    /// This function is intended to be used to wrap a TCP listener from the
+    /// standard library in the Mio equivalent. The conversion assumes nothing
+    /// about the underlying listener; ; it is left up to the user to set it
+    /// in non-blocking mode.
+    pub fn from_std(listener: net::TcpListener) -> TcpListener {
+        let sys = sys::TcpListener::from_std(listener);
+        TcpListener {
+            sys,
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+
     /// Accepts a new `TcpStream`.
     ///
     /// This may return an `Err(e)` where `e.kind()` is
@@ -168,3 +186,28 @@ impl FromRawFd for TcpListener {
         }
     }
 }
+
+#[cfg(windows)]
+impl AsRawSocket for TcpListener {
+    fn as_raw_socket(&self) -> RawSocket {
+        self.sys.as_raw_socket()
+    }
+}
+
+#[cfg(windows)]
+impl FromRawSocket for TcpListener {
+    unsafe fn from_raw_socket(socket: RawSocket) -> TcpListener {
+        TcpListener {
+            sys: FromRawSocket::from_raw_socket(socket),
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+}
+
+#[cfg(windows)]
+impl IntoRawSocket for TcpListener {
+    fn into_raw_socket(self) -> RawSocket {
+        self.sys.into_raw_socket()
+    }
+}
diff --git a/src/net/tcp/stream.rs b/src/net/tcp/stream.rs
index d718401..dc0a6c5 100644
--- a/src/net/tcp/stream.rs
+++ b/src/net/tcp/stream.rs
@@ -1,8 +1,11 @@
 use std::fmt;
 use std::io::{self, IoSlice, IoSliceMut, Read, Write};
+use std::net;
 use std::net::SocketAddr;
 #[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+#[cfg(windows)]
+use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
 
 #[cfg(debug_assertions)]
 use crate::poll::SelectorId;
@@ -65,6 +68,27 @@ impl TcpStream {
         })
     }
 
+    /// Creates a new `TcpStream` from a standard `net::TcpStream`.
+    ///
+    /// This function is intended to be used to wrap a TCP stream from the
+    /// standard library in the Mio equivalent. The conversion assumes nothing
+    /// about the underlying stream; it is left up to the user to set it in
+    /// non-blocking mode.
+    ///
+    /// # Note
+    ///
+    /// The TCP stream here will not have `connect` called on it, so it
+    /// should already be connected via some other means (be it manually, or
+    /// the standard library).
+    pub fn from_std(stream: net::TcpStream) -> TcpStream {
+        let sys = sys::TcpStream::from_std(stream);
+        TcpStream {
+            sys,
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+
     /// Returns the socket address of the remote peer of this TCP connection.
     pub fn peer_addr(&self) -> io::Result<SocketAddr> {
         self.sys.peer_addr()
@@ -278,3 +302,28 @@ impl FromRawFd for TcpStream {
         }
     }
 }
+
+#[cfg(windows)]
+impl AsRawSocket for TcpStream {
+    fn as_raw_socket(&self) -> RawSocket {
+        self.sys.as_raw_socket()
+    }
+}
+
+#[cfg(windows)]
+impl FromRawSocket for TcpStream {
+    unsafe fn from_raw_socket(socket: RawSocket) -> TcpStream {
+        TcpStream {
+            sys: FromRawSocket::from_raw_socket(socket),
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+}
+
+#[cfg(windows)]
+impl IntoRawSocket for TcpStream {
+    fn into_raw_socket(self) -> RawSocket {
+        self.sys.into_raw_socket()
+    }
+}
diff --git a/src/net/udp.rs b/src/net/udp.rs
index bb56c2d..6e89f3a 100644
--- a/src/net/udp.rs
+++ b/src/net/udp.rs
@@ -13,9 +13,12 @@ use crate::{event, sys, Interests, Registry, Token};
 
 use std::fmt;
 use std::io;
+use std::net;
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 #[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+#[cfg(windows)]
+use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
 
 /// A User Datagram Protocol socket.
 ///
@@ -124,6 +127,21 @@ impl UdpSocket {
         })
     }
 
+    /// Creates a new `UdpSocket` from a standard `net::UdpSocket`.
+    ///
+    /// This function is intended to be used to wrap a UDP socket from the
+    /// standard library in the Mio equivalent. The conversion assumes nothing
+    /// about the underlying socket; it is left up to the user to set it in
+    /// non-blocking mode.
+    pub fn from_std(socket: net::UdpSocket) -> UdpSocket {
+        let sys = sys::UdpSocket::from_std(socket);
+        UdpSocket {
+            sys,
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+
     /// Returns the socket address that this socket was created from.
     ///
     /// # Examples
@@ -556,3 +574,28 @@ impl FromRawFd for UdpSocket {
         }
     }
 }
+
+#[cfg(windows)]
+impl AsRawSocket for UdpSocket {
+    fn as_raw_socket(&self) -> RawSocket {
+        self.sys.as_raw_socket()
+    }
+}
+
+#[cfg(windows)]
+impl FromRawSocket for UdpSocket {
+    unsafe fn from_raw_socket(socket: RawSocket) -> UdpSocket {
+        UdpSocket {
+            sys: FromRawSocket::from_raw_socket(socket),
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+}
+
+#[cfg(windows)]
+impl IntoRawSocket for UdpSocket {
+    fn into_raw_socket(self) -> RawSocket {
+        self.sys.into_raw_socket()
+    }
+}
diff --git a/src/net/uds/datagram.rs b/src/net/uds/datagram.rs
index df330cf..36f3822 100644
--- a/src/net/uds/datagram.rs
+++ b/src/net/uds/datagram.rs
@@ -6,6 +6,7 @@ use crate::{sys, Interests, Registry, Token};
 use std::io;
 use std::net::Shutdown;
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+use std::os::unix::net;
 use std::path::Path;
 
 /// A Unix datagram socket.
@@ -31,6 +32,21 @@ impl UnixDatagram {
         Ok(UnixDatagram::new(sys))
     }
 
+    /// Creates a new `UnixDatagram` from a standard `net::UnixDatagram`.
+    ///
+    /// This function is intended to be used to wrap a Unix datagram from the
+    /// standard library in the Mio equivalent. The conversion assumes nothing
+    /// about the underlying datagram; ; it is left up to the user to set it
+    /// in non-blocking mode.
+    pub fn from_std(datagram: net::UnixDatagram) -> UnixDatagram {
+        let sys = sys::UnixDatagram::from_std(datagram);
+        UnixDatagram {
+            sys,
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+
     /// Connects the socket to the specified address.
     pub fn connect<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
         self.sys.connect(path)
diff --git a/src/net/uds/listener.rs b/src/net/uds/listener.rs
index 5cd20a8..4773c13 100644
--- a/src/net/uds/listener.rs
+++ b/src/net/uds/listener.rs
@@ -7,6 +7,7 @@ use crate::{sys, Interests, Registry, Token};
 
 use std::io;
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+use std::os::unix::net;
 use std::path::Path;
 
 /// A non-blocking Unix domain socket server.
@@ -32,6 +33,21 @@ impl UnixListener {
         Ok(UnixListener::new(sys))
     }
 
+    /// Creates a new `UnixListener` from a standard `net::UnixListener`.
+    ///
+    /// This function is intended to be used to wrap a Unix listener from the
+    /// standard library in the Mio equivalent. The conversion assumes nothing
+    /// about the underlying listener; it is left up to the user to set it in
+    /// non-blocking mode.
+    pub fn from_std(listener: net::UnixListener) -> UnixListener {
+        let sys = sys::UnixListener::from_std(listener);
+        UnixListener {
+            sys,
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+
     /// Accepts a new incoming connection to this listener.
     ///
     /// The call is responsible for ensuring that the listening socket is in
diff --git a/src/net/uds/stream.rs b/src/net/uds/stream.rs
index 79c8e00..d7a3a72 100644
--- a/src/net/uds/stream.rs
+++ b/src/net/uds/stream.rs
@@ -6,6 +6,7 @@ use crate::{sys, Interests, Registry, Token};
 use std::io::{self, IoSlice, IoSliceMut};
 use std::net::Shutdown;
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+use std::os::unix::net;
 use std::path::Path;
 
 /// A non-blocking Unix stream socket.
@@ -31,6 +32,27 @@ impl UnixStream {
         Ok(UnixStream::new(sys))
     }
 
+    /// Creates a new `UnixStream` from a standard `net::UnixStream`.
+    ///
+    /// This function is intended to be used to wrap a Unix stream from the
+    /// standard library in the Mio equivalent. The conversion assumes nothing
+    /// about the underlying stream; it is left up to the user to set it in
+    /// non-blocking mode.
+    ///
+    /// # Note
+    ///
+    /// The Unix stream here will not have `connect` called on it, so it
+    /// should already be connected via some other means (be it manually, or
+    /// the standard library).
+    pub fn from_std(stream: net::UnixStream) -> UnixStream {
+        let sys = sys::UnixStream::from_std(stream);
+        UnixStream {
+            sys,
+            #[cfg(debug_assertions)]
+            selector_id: SelectorId::new(),
+        }
+    }
+
     /// Creates an unnamed pair of connected sockets.
     ///
     /// Returns two `UnixStream`s which are connected to each other.
diff --git a/src/sys/unix/selector/kqueue.rs b/src/sys/unix/selector/kqueue.rs
index 0675d55..58da03d 100644
--- a/src/sys/unix/selector/kqueue.rs
+++ b/src/sys/unix/selector/kqueue.rs
@@ -2,6 +2,7 @@ use crate::{Interests, Token};
 
 use log::error;
 use std::mem::MaybeUninit;
+use std::ops::{Deref, DerefMut};
 use std::os::unix::io::{AsRawFd, RawFd};
 #[cfg(debug_assertions)]
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -304,7 +305,35 @@ impl Drop for Selector {
 }
 
 pub type Event = libc::kevent;
-pub type Events = Vec<Event>;
+pub struct Events(Vec<libc::kevent>);
+
+impl Events {
+    pub fn with_capacity(capacity: usize) -> Events {
+        Events(Vec::with_capacity(capacity))
+    }
+}
+
+impl Deref for Events {
+    type Target = Vec<libc::kevent>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl DerefMut for Events {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+// `Events` cannot derive `Send` or `Sync` because of the
+// `udata: *mut ::c_void` field in `libc::kevent`. However, `Events`'s public
+// API treats the `udata` field as a `uintptr_t` which is `Send`. `Sync` is
+// safe because with a `events: &Events` value, the only access to the `udata`
+// field is through `fn token(event: &Event)` which cannot mutate the field.
+unsafe impl Send for Events {}
+unsafe impl Sync for Events {}
 
 pub mod event {
     use crate::sys::Event;
diff --git a/src/sys/unix/tcp/listener.rs b/src/sys/unix/tcp/listener.rs
index b7a6e97..5a4b1b0 100644
--- a/src/sys/unix/tcp/listener.rs
+++ b/src/sys/unix/tcp/listener.rs
@@ -40,6 +40,10 @@ impl TcpListener {
         })
     }
 
+    pub fn from_std(inner: net::TcpListener) -> TcpListener {
+        TcpListener { inner }
+    }
+
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
         self.inner.local_addr()
     }
diff --git a/src/sys/unix/tcp/stream.rs b/src/sys/unix/tcp/stream.rs
index b446266..6e03dbb 100644
--- a/src/sys/unix/tcp/stream.rs
+++ b/src/sys/unix/tcp/stream.rs
@@ -39,6 +39,10 @@ impl TcpStream {
             })
     }
 
+    pub fn from_std(inner: net::TcpStream) -> TcpStream {
+        TcpStream { inner }
+    }
+
     pub fn peer_addr(&self) -> io::Result<SocketAddr> {
         self.inner.peer_addr()
     }
diff --git a/src/sys/unix/udp.rs b/src/sys/unix/udp.rs
index 99d4610..beb16ac 100644
--- a/src/sys/unix/udp.rs
+++ b/src/sys/unix/udp.rs
@@ -44,6 +44,10 @@ impl UdpSocket {
         })
     }
 
+    pub fn from_std(io: net::UdpSocket) -> UdpSocket {
+        UdpSocket { io }
+    }
+
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
         self.io.local_addr()
     }
diff --git a/src/sys/unix/uds/datagram.rs b/src/sys/unix/uds/datagram.rs
index 22255fc..1e4b0a3 100644
--- a/src/sys/unix/uds/datagram.rs
+++ b/src/sys/unix/uds/datagram.rs
@@ -29,6 +29,10 @@ impl UnixDatagram {
         Ok(unsafe { UnixDatagram::from_raw_fd(socket) })
     }
 
+    pub fn from_std(inner: net::UnixDatagram) -> UnixDatagram {
+        UnixDatagram { inner }
+    }
+
     pub(crate) fn connect<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
         self.inner.connect(path)
     }
diff --git a/src/sys/unix/uds/listener.rs b/src/sys/unix/uds/listener.rs
index d5f6554..d0ca11b 100644
--- a/src/sys/unix/uds/listener.rs
+++ b/src/sys/unix/uds/listener.rs
@@ -112,6 +112,10 @@ impl UnixListener {
             .map(|_| unsafe { UnixListener::from_raw_fd(socket) })
     }
 
+    pub fn from_std(inner: net::UnixListener) -> UnixListener {
+        UnixListener { inner }
+    }
+
     pub(crate) fn try_clone(&self) -> io::Result<UnixListener> {
         let inner = self.inner.try_clone()?;
         Ok(UnixListener::new(inner))
diff --git a/src/sys/unix/uds/stream.rs b/src/sys/unix/uds/stream.rs
index 41492a4..c15ba22 100644
--- a/src/sys/unix/uds/stream.rs
+++ b/src/sys/unix/uds/stream.rs
@@ -40,6 +40,10 @@ impl UnixStream {
         Ok(unsafe { UnixStream::from_raw_fd(socket) })
     }
 
+    pub fn from_std(inner: net::UnixStream) -> UnixStream {
+        UnixStream { inner }
+    }
+
     pub(crate) fn pair() -> io::Result<(UnixStream, UnixStream)> {
         let fds = [-1; 2];
         let flags = libc::SOCK_STREAM;
diff --git a/src/sys/windows/tcp.rs b/src/sys/windows/tcp.rs
index bb5fbd0..46f5730 100644
--- a/src/sys/windows/tcp.rs
+++ b/src/sys/windows/tcp.rs
@@ -61,6 +61,13 @@ impl TcpStream {
             })
     }
 
+    pub fn from_std(inner: net::TcpStream) -> TcpStream {
+        TcpStream {
+            internal: Box::new(Mutex::new(None)),
+            inner,
+        }
+    }
+
     pub fn peer_addr(&self) -> io::Result<SocketAddr> {
         self.inner.peer_addr()
     }
@@ -316,6 +323,13 @@ impl TcpListener {
         })
     }
 
+    pub fn from_std(inner: net::TcpListener) -> TcpListener {
+        TcpListener {
+            internal: Box::new(Mutex::new(None)),
+            inner,
+        }
+    }
+
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
         self.inner.local_addr()
     }
diff --git a/src/sys/windows/udp.rs b/src/sys/windows/udp.rs
index 91b2645..798e2ee 100644
--- a/src/sys/windows/udp.rs
+++ b/src/sys/windows/udp.rs
@@ -38,6 +38,13 @@ impl UdpSocket {
         })
     }
 
+    pub fn from_std(inner: net::UdpSocket) -> UdpSocket {
+        UdpSocket {
+            internal: Box::new(Mutex::new(None)),
+            inner,
+        }
+    }
+
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
         self.inner.local_addr()
     }


Test patch:
diff --git a/tests/poll.rs b/tests/poll.rs
index 25602c1..4bd38ba 100644
--- a/tests/poll.rs
+++ b/tests/poll.rs
@@ -13,6 +13,9 @@ use util::{any_local_address, assert_send, assert_sync, init};
 
 #[test]
 fn is_send_and_sync() {
+    assert_send::<Events>();
+    assert_sync::<Events>();
+
     assert_sync::<Poll>();
     assert_send::<Poll>();
 
diff --git a/tests/tcp_listener.rs b/tests/tcp_listener.rs
index 0530245..4c8416f 100644
--- a/tests/tcp_listener.rs
+++ b/tests/tcp_listener.rs
@@ -1,4 +1,4 @@
-use std::io::Read;
+use std::io::{self, Read};
 use std::net::{self, SocketAddr};
 #[cfg(unix)]
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd};
@@ -26,18 +26,32 @@ fn is_send_and_sync() {
 
 #[test]
 fn tcp_listener() {
-    smoke_test_tcp_listener(any_local_address());
+    smoke_test_tcp_listener(any_local_address(), TcpListener::bind);
 }
 
 #[test]
 fn tcp_listener_ipv6() {
-    smoke_test_tcp_listener(any_local_ipv6_address());
+    smoke_test_tcp_listener(any_local_ipv6_address(), TcpListener::bind);
 }
 
-fn smoke_test_tcp_listener(addr: SocketAddr) {
+#[test]
+fn tcp_listener_std() {
+    smoke_test_tcp_listener(any_local_address(), |addr| {
+        let listener = net::TcpListener::bind(addr).unwrap();
+        // `std::net::TcpListener`s are blocking by default, so make sure it is in
+        // non-blocking mode before wrapping in a Mio equivalent.
+        assert_ok!(listener.set_nonblocking(true));
+        Ok(TcpListener::from_std(listener))
+    });
+}
+
+fn smoke_test_tcp_listener<F>(addr: SocketAddr, make_listener: F)
+where
+    F: FnOnce(SocketAddr) -> io::Result<TcpListener>,
+{
     let (mut poll, mut events) = init_with_poll();
 
-    let listener = TcpListener::bind(addr).unwrap();
+    let listener = make_listener(addr).unwrap();
     let address = listener.local_addr().unwrap();
 
     poll.registry()
diff --git a/tests/tcp_stream.rs b/tests/tcp_stream.rs
index cae707d..7896f8a 100644
--- a/tests/tcp_stream.rs
+++ b/tests/tcp_stream.rs
@@ -36,20 +36,35 @@ fn is_send_and_sync() {
 #[test]
 #[cfg_attr(windows, ignore = "fails on Windows, see #1078")]
 fn tcp_stream_ipv4() {
-    smoke_test_tcp_stream(any_local_address());
+    smoke_test_tcp_stream(any_local_address(), TcpStream::connect);
 }
 
 #[test]
 #[cfg_attr(windows, ignore = "fails on Windows, see #1078")]
 fn tcp_stream_ipv6() {
-    smoke_test_tcp_stream(any_local_ipv6_address());
+    smoke_test_tcp_stream(any_local_ipv6_address(), TcpStream::connect);
 }
 
-fn smoke_test_tcp_stream(addr: SocketAddr) {
+#[test]
+#[cfg_attr(windows, ignore = "fails on Windows, see #1078")]
+fn tcp_stream_std() {
+    smoke_test_tcp_stream(any_local_address(), |addr| {
+        let stream = net::TcpStream::connect(addr).unwrap();
+        // `std::net::TcpStream`s are blocking by default, so make sure it is
+        // in non-blocking mode before wrapping in a Mio equivalent.
+        assert_ok!(stream.set_nonblocking(true));
+        Ok(TcpStream::from_std(stream))
+    });
+}
+
+fn smoke_test_tcp_stream<F>(addr: SocketAddr, make_stream: F)
+where
+    F: FnOnce(SocketAddr) -> io::Result<TcpStream>,
+{
     let (mut poll, mut events) = init_with_poll();
 
-    let (thread_handle, addr) = echo_listener(addr, 1);
-    let mut stream = TcpStream::connect(addr).unwrap();
+    let (handle, addr) = echo_listener(addr, 1);
+    let mut stream = make_stream(addr).unwrap();
 
     poll.registry()
         .register(&stream, ID1, Interests::WRITABLE.add(Interests::READABLE))
@@ -116,8 +131,7 @@ fn smoke_test_tcp_stream(addr: SocketAddr) {
 
     // Close the connection to allow the listener to shutdown.
     drop(stream);
-
-    thread_handle.join().expect("unable to join thread");
+    handle.join().expect("unable to join thread");
 }
 
 #[test]
diff --git a/tests/udp_socket.rs b/tests/udp_socket.rs
index 5e00e87..f0c9f2f 100644
--- a/tests/udp_socket.rs
+++ b/tests/udp_socket.rs
@@ -49,6 +49,21 @@ fn unconnected_udp_socket_ipv6() {
     smoke_test_unconnected_udp_socket(socket1, socket2);
 }
 
+#[test]
+fn unconnected_udp_socket_std() {
+    let socket1 = net::UdpSocket::bind(any_local_address()).unwrap();
+    let socket2 = net::UdpSocket::bind(any_local_address()).unwrap();
+
+    // `std::net::UdpSocket`s are blocking by default, so make sure they are
+    // in non-blocking mode before wrapping in a Mio equivalent.
+    assert_ok!(socket1.set_nonblocking(true));
+    assert_ok!(socket2.set_nonblocking(true));
+
+    let socket1 = UdpSocket::from_std(socket1);
+    let socket2 = UdpSocket::from_std(socket2);
+    smoke_test_unconnected_udp_socket(socket1, socket2);
+}
+
 fn smoke_test_unconnected_udp_socket(socket1: UdpSocket, socket2: UdpSocket) {
     let (mut poll, mut events) = init_with_poll();
 
@@ -158,6 +173,28 @@ fn connected_udp_socket_ipv6() {
     smoke_test_connected_udp_socket(socket1, socket2);
 }
 
+#[test]
+fn connected_udp_socket_std() {
+    let socket1 = net::UdpSocket::bind(any_local_address()).unwrap();
+    let address1 = socket1.local_addr().unwrap();
+
+    let socket2 = net::UdpSocket::bind(any_local_address()).unwrap();
+    let address2 = socket2.local_addr().unwrap();
+
+    socket1.connect(address2).unwrap();
+    socket2.connect(address1).unwrap();
+
+    // `std::net::UdpSocket`s are blocking by default, so make sure they are
+    // in non-blocking mode before wrapping in a Mio equivalent.
+    assert_ok!(socket1.set_nonblocking(true));
+    assert_ok!(socket2.set_nonblocking(true));
+
+    let socket1 = UdpSocket::from_std(socket1);
+    let socket2 = UdpSocket::from_std(socket2);
+
+    smoke_test_connected_udp_socket(socket1, socket2);
+}
+
 fn smoke_test_connected_udp_socket(socket1: UdpSocket, socket2: UdpSocket) {
     let (mut poll, mut events) = init_with_poll();
 
diff --git a/tests/uds.rs b/tests/uds.rs
index c9f7373..70a9929 100644
--- a/tests/uds.rs
+++ b/tests/uds.rs
@@ -9,6 +9,7 @@ use mio::{Interests, Token};
 use std::io::{self, IoSlice, IoSliceMut, Read, Write};
 use std::net::Shutdown;
 use std::os::unix::net;
+use std::path::Path;
 use std::sync::mpsc::{channel, Receiver};
 use std::sync::{Arc, Barrier};
 use std::thread;
@@ -30,14 +31,33 @@ const LOCAL: Token = Token(0);
 const LOCAL_CLONE: Token = Token(1);
 
 #[test]
-fn smoke_test() {
+fn uds_stream() {
+    #[allow(clippy::redundant_closure)]
+    smoke_test(|path| UnixStream::connect(path));
+}
+
+#[test]
+fn uds_stream_std() {
+    smoke_test(|path| {
+        let local = assert_ok!(net::UnixStream::connect(path));
+        // `std::os::unix::net::UnixStream`s are blocking by default, so make sure
+        // it is in non-blocking mode before wrapping in a Mio equivalent.
+        assert_ok!(local.set_nonblocking(true));
+        Ok(UnixStream::from_std(local))
+    })
+}
+
+fn smoke_test<F>(make_stream: F)
+where
+    F: FnOnce(&Path) -> io::Result<UnixStream>,
+{
     let (mut poll, mut events) = init_with_poll();
 
     let (sync_sender, sync_receiver) = channel();
     let (handle, remote_addr) = echo_remote(1, sync_receiver);
 
     let path = remote_addr.as_pathname().expect("not a pathname");
-    let mut local = assert_ok!(UnixStream::connect(path));
+    let mut local = assert_ok!(make_stream(path));
     assert_ok!(sync_sender.send(()));
 
     assert_ok!(poll.registry().register(