Replace socket2 calls with rustix

Cargo.toml

@@ -30,12 +30,14 @@ futures-io = { version = "0.3.28", default-features = false, features = ["std"]
 futures-lite = { version = "1.11.0", default-features = false }
 parking = "2.0.0"
 polling = "3.0.0"
-rustix = { version = "0.38.2", default-features = false, features = ["std", "fs"] }
+rustix = { version = "0.38.2", default-features = false, features = ["fs", "net", "std"] }
 slab = "0.4.2"
-socket2 = { version = "0.5.3", features = ["all"] }
 tracing = { version = "0.1.37", default-features = false }
 waker-fn = "1.1.0"
 
+[target.'cfg(windows)'.dependencies]
+windows-sys = { version = "0.48.0", features = ["Win32_Foundation"] }
+
 [dev-dependencies]
 async-channel = "1"
 async-net = "1"

src/lib.rs

@@ -83,7 +83,9 @@ use std::os::windows::io::{AsRawSocket, AsSocket, BorrowedSocket, OwnedSocket, R
 use futures_io::{AsyncRead, AsyncWrite};
 use futures_lite::stream::{self, Stream};
 use futures_lite::{future, pin, ready};
-use socket2::{Domain, Protocol, SockAddr, Socket, Type};
+
+use rustix::io as rio;
+use rustix::net as rn;
 
 use crate::reactor::{Reactor, Registration, Source};
 
@@ -656,23 +658,7 @@ impl<T: AsFd> Async<T> {
     pub fn new(io: T) -> io::Result<Async<T>> {
         // Put the file descriptor in non-blocking mode.
         let fd = io.as_fd();
-        cfg_if::cfg_if! {
-            // ioctl(FIONBIO) sets the flag atomically, but we use this only on Linux
-            // for now, as with the standard library, because it seems to behave
-            // differently depending on the platform.
-            // https://github.com/rust-lang/rust/commit/efeb42be2837842d1beb47b51bb693c7474aba3d
-            // https://github.com/libuv/libuv/blob/e9d91fccfc3e5ff772d5da90e1c4a24061198ca0/src/unix/poll.c#L78-L80
-            // https://github.com/tokio-rs/mio/commit/0db49f6d5caf54b12176821363d154384357e70a
-            if #[cfg(target_os = "linux")] {
-                rustix::io::ioctl_fionbio(fd, true)?;
-            } else {
-                let previous = rustix::fs::fcntl_getfl(fd)?;
-                let new = previous | rustix::fs::OFlags::NONBLOCK;
-                if new != previous {
-                    rustix::fs::fcntl_setfl(fd, new)?;
-                }
-            }
-        }
+        set_nonblocking(fd)?;
 
         // SAFETY: It is impossible to drop the I/O source while it is registered through
         // this type.
@@ -1487,10 +1473,15 @@ impl Async<TcpStream> {
     /// # std::io::Result::Ok(()) });
     /// ```
     pub async fn connect<A: Into<SocketAddr>>(addr: A) -> io::Result<Async<TcpStream>> {
-        // Begin async connect.
+        // Figure out how to handle this address.
         let addr = addr.into();
-        let domain = Domain::for_address(addr);
-        let socket = connect(addr.into(), domain, Some(Protocol::TCP))?;
+        let (domain, sock_addr) = match addr {
+            SocketAddr::V4(v4) => (rn::AddressFamily::INET, rn::SocketAddrAny::V4(v4)),
+            SocketAddr::V6(v6) => (rn::AddressFamily::INET6, rn::SocketAddrAny::V6(v6)),
+        };
+
+        // Begin async connect.
+        let socket = connect(sock_addr, domain, Some(rn::ipproto::TCP))?;
         let stream = Async::new(TcpStream::from(socket))?;
 
         // The stream becomes writable when connected.
@@ -1819,7 +1810,11 @@ impl Async<UnixStream> {
     /// ```
     pub async fn connect<P: AsRef<Path>>(path: P) -> io::Result<Async<UnixStream>> {
         // Begin async connect.
-        let socket = connect(SockAddr::unix(path)?, Domain::UNIX, None)?;
+        let socket = connect(
+            rn::SocketAddrUnix::new(path.as_ref())?.into(),
+            rn::AddressFamily::UNIX,
+            None,
+        )?;
         let stream = Async::new(UnixStream::from(socket))?;
 
         // The stream becomes writable when connected.
@@ -2029,8 +2024,14 @@ async fn optimistic(fut: impl Future<Output = io::Result<()>>) -> io::Result<()>
     .await
 }
 
-fn connect(addr: SockAddr, domain: Domain, protocol: Option<Protocol>) -> io::Result<Socket> {
-    let sock_type = Type::STREAM;
+fn connect(
+    addr: rn::SocketAddrAny,
+    domain: rn::AddressFamily,
+    protocol: Option<rn::Protocol>,
+) -> io::Result<rustix::fd::OwnedFd> {
+    #[cfg(windows)]
+    use rustix::fd::AsFd;
+
     #[cfg(any(
         target_os = "android",
         target_os = "dragonfly",
@@ -2041,10 +2042,13 @@ fn connect(addr: SockAddr, domain: Domain, protocol: Option<Protocol>) -> io::Re
         target_os = "netbsd",
         target_os = "openbsd"
     ))]
-    // If we can, set nonblocking at socket creation for unix
-    let sock_type = sock_type.nonblocking();
-    // This automatically handles cloexec on unix, no_inherit on windows and nosigpipe on macos
-    let socket = Socket::new(domain, sock_type, protocol)?;
+    let socket = rn::socket_with(
+        domain,
+        rn::SocketType::STREAM,
+        rn::SocketFlags::CLOEXEC | rn::SocketFlags::NONBLOCK,
+        protocol,
+    )?;
+
     #[cfg(not(any(
         target_os = "android",
         target_os = "dragonfly",
@@ -2055,14 +2059,100 @@ fn connect(addr: SockAddr, domain: Domain, protocol: Option<Protocol>) -> io::Re
         target_os = "netbsd",
         target_os = "openbsd"
     )))]
-    // If the current platform doesn't support nonblocking at creation, enable it after creation
-    socket.set_nonblocking(true)?;
-    match socket.connect(&addr) {
+    let socket = {
+        #[cfg(not(any(
+            target_os = "macos",
+            target_os = "ios",
+            target_os = "tvos",
+            target_os = "watchos",
+            target_os = "espidf",
+            windows,
+        )))]
+        let flags = rn::SocketFlags::CLOEXEC;
+        #[cfg(any(
+            target_os = "macos",
+            target_os = "ios",
+            target_os = "tvos",
+            target_os = "watchos",
+            target_os = "espidf",
+            windows,
+        ))]
+        let flags = rn::SocketFlags::empty();
+
+        // Create the socket.
+        let socket = rn::socket_with(domain, rn::SocketType::STREAM, flags, protocol)?;
+
+        // Set cloexec if necessary.
+        #[cfg(any(
+            target_os = "macos",
+            target_os = "ios",
+            target_os = "tvos",
+            target_os = "watchos",
+        ))]
+        rio::fcntl_setfd(&socket, rio::fcntl_getfd(&socket)? | rio::FdFlags::CLOEXEC)?;
+
+        // Set non-blocking mode.
+        set_nonblocking(socket.as_fd())?;
+
+        socket
+    };
+
+    // Set nosigpipe if necessary.
+    #[cfg(any(
+        target_os = "macos",
+        target_os = "ios",
+        target_os = "tvos",
+        target_os = "watchos",
+        target_os = "freebsd"
+    ))]
+    rn::sockopt::set_socket_nosigpipe(&socket, true)?;
+
+    // Set the handle information to HANDLE_FLAG_INHERIT.
+    #[cfg(windows)]
+    unsafe {
+        if windows_sys::Win32::Foundation::SetHandleInformation(
+            socket.as_raw_socket() as _,
+            windows_sys::Win32::Foundation::HANDLE_FLAG_INHERIT,
+            windows_sys::Win32::Foundation::HANDLE_FLAG_INHERIT,
+        ) == 0
+        {
+            return Err(io::Error::last_os_error());
+        }
+    }
+
+    #[allow(unreachable_patterns)]
+    match rn::connect_any(&socket, &addr) {
         Ok(_) => {}
         #[cfg(unix)]
-        Err(err) if err.raw_os_error() == Some(rustix::io::Errno::INPROGRESS.raw_os_error()) => {}
-        Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
-        Err(err) => return Err(err),
+        Err(rio::Errno::INPROGRESS) => {}
+        Err(rio::Errno::AGAIN) | Err(rio::Errno::WOULDBLOCK) => {}
+        Err(err) => return Err(err.into()),
     }
     Ok(socket)
 }
+
+#[inline]
+fn set_nonblocking(
+    #[cfg(unix)] fd: BorrowedFd<'_>,
+    #[cfg(windows)] fd: BorrowedSocket<'_>,
+) -> io::Result<()> {
+    cfg_if::cfg_if! {
+        // ioctl(FIONBIO) sets the flag atomically, but we use this only on Linux
+        // for now, as with the standard library, because it seems to behave
+        // differently depending on the platform.
+        // https://github.com/rust-lang/rust/commit/efeb42be2837842d1beb47b51bb693c7474aba3d
+        // https://github.com/libuv/libuv/blob/e9d91fccfc3e5ff772d5da90e1c4a24061198ca0/src/unix/poll.c#L78-L80
+        // https://github.com/tokio-rs/mio/commit/0db49f6d5caf54b12176821363d154384357e70a
+        if #[cfg(any(windows, target_os = "linux"))] {
+            rustix::io::ioctl_fionbio(fd, true)?;
+        } else {
+            let previous = rustix::fs::fcntl_getfl(fd)?;
+            let new = previous | rustix::fs::OFlags::NONBLOCK;
+            if new != previous {
+                rustix::fs::fcntl_setfl(fd, new)?;
+            }
+        }
+    }
+
+    Ok(())
+}

tests/async.rs

@@ -178,7 +178,8 @@ fn udp_connect() -> io::Result<()> {
     })
 }
 
-#[cfg(unix)]
+// This test is broken for now on OpenBSD: https://github.com/rust-lang/rust/issues/116523
+#[cfg(all(unix, not(target_os = "openbsd")))]
 #[test]
 fn uds_connect() -> io::Result<()> {
     future::block_on(async {