net: fix TcpSocket bugs introduced in #4270

Author: taiki-e

.github/workflows/ci.yml

@@ -31,6 +31,7 @@ jobs:
       - valgrind
       - loom-compile
       - check-readme
+      - test-hyper
     steps:
       - run: exit 0
 
@@ -355,3 +356,36 @@ jobs:
       - name: Verify that Tokio version is up to date in README
         working-directory: tokio
         run: grep -q "$(sed '/^version = /!d' Cargo.toml | head -n1)" README.md
+
+  test-hyper:
+    name: Test hyper
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        cmd:
+          - test
+          - ubuntu-latest
+          - bench
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions-rs/toolchain@v1
+        with:
+          toolchain: ${{ env.nightly }}
+          override: true
+      - uses: Swatinem/rust-cache@v1
+      - name: Test hyper
+        run: |
+          set -x
+          git clone https://github.com/hyperium/hyper.git
+          cd hyper
+          # checkout the latest release because master maybe contains breakage.
+          tag=$(curl -fsSL https://api.github.com/repos/hyperium/hyper/releases/latest | jq -r '.tag_name')
+          git checkout "${tag}"
+          echo "[workspace]" >>Cargo.toml
+          echo "[patch.crates-io]" >>Cargo.toml
+          echo "tokio = { path = "../tokio" }" >>Cargo.toml
+          echo "tokio-util = { path = "../tokio-util" }" >>Cargo.toml
+          echo "tokio-stream = { path = "../tokio-stream" }" >>Cargo.toml
+          echo "tokio-test = { path = "../tokio-test" }" >>Cargo.toml
+          cargo ${{ matrix.cmd }} --features full --no-fail-fast

tokio/src/net/tcp/socket.rs

@@ -125,6 +125,7 @@ impl TcpSocket {
             socket2::Type::STREAM,
             Some(socket2::Protocol::TCP),
         )?;
+        inner.set_nonblocking(true)?;
         Ok(TcpSocket { inner })
     }
 
@@ -163,6 +164,7 @@ impl TcpSocket {
             socket2::Type::STREAM,
             Some(socket2::Protocol::TCP),
         )?;
+        inner.set_nonblocking(true)?;
         Ok(TcpSocket { inner })
     }
 
@@ -534,7 +536,14 @@ impl TcpSocket {
     /// }
     /// ```
     pub async fn connect(self, addr: SocketAddr) -> io::Result<TcpStream> {
-        self.inner.connect(&addr.into())?;
+        match self.inner.connect(&addr.into()) {
+            // https://github.com/tokio-rs/mio/blob/d400ddfc97212b4a2844d741b095dab2c6d15543/src/sys/unix/tcp.rs#L31
+            #[cfg(unix)]
+            Err(e) if e.raw_os_error() != Some(libc::EINPROGRESS) => return Err(e),
+            #[cfg(not(unix))]
+            Err(e) => return Err(e),
+            _ => {}
+        }
 
         #[cfg(windows)]
         let mio = unsafe { mio::net::TcpStream::from_raw_socket(self.inner.into_raw_socket()) };

tokio/tests/tcp_socket.rs

@@ -10,6 +10,8 @@ async fn basic_usage_v4() {
     // Create server
     let addr = assert_ok!("127.0.0.1:0".parse());
     let srv = assert_ok!(TcpSocket::new_v4());
+    assert_non_blocking(&srv);
+    assert_close_on_exec(&srv);
     assert_ok!(srv.bind(addr));
 
     let srv = assert_ok!(srv.listen(128));
@@ -28,6 +30,8 @@ async fn basic_usage_v6() {
     // Create server
     let addr = assert_ok!("[::1]:0".parse());
     let srv = assert_ok!(TcpSocket::new_v6());
+    assert_non_blocking(&srv);
+    assert_close_on_exec(&srv);
     assert_ok!(srv.bind(addr));
 
     let srv = assert_ok!(srv.listen(128));
@@ -46,6 +50,8 @@ async fn bind_before_connect() {
     // Create server
     let any_addr = assert_ok!("127.0.0.1:0".parse());
     let srv = assert_ok!(TcpSocket::new_v4());
+    assert_non_blocking(&srv);
+    assert_close_on_exec(&srv);
     assert_ok!(srv.bind(any_addr));
 
     let srv = assert_ok!(srv.listen(128));
@@ -65,10 +71,29 @@ async fn basic_linger() {
     // Create server
     let addr = assert_ok!("127.0.0.1:0".parse());
     let srv = assert_ok!(TcpSocket::new_v4());
+    assert_non_blocking(&srv);
+    assert_close_on_exec(&srv);
     assert_ok!(srv.bind(addr));
 
     assert!(srv.linger().unwrap().is_none());
 
     srv.set_linger(Some(Duration::new(0, 0))).unwrap();
     assert_eq!(srv.linger().unwrap(), Some(Duration::new(0, 0)));
 }
+
+fn assert_non_blocking(socket: &TcpSocket) {
+    #[cfg(unix)]
+    {
+        use std::os::unix::io::AsRawFd;
+        let flags = unsafe { libc::fcntl(socket.as_raw_fd(), libc::F_GETFL) };
+        assert!(flags & libc::O_NONBLOCK != 0, "O_NONBLOCK not set");
+    }
+}
+fn assert_close_on_exec(socket: &TcpSocket) {
+    #[cfg(unix)]
+    {
+        use std::os::unix::io::AsRawFd;
+        let flags = unsafe { libc::fcntl(socket.as_raw_fd(), libc::F_GETFD) };
+        assert!(flags & libc::FD_CLOEXEC != 0, "FD_CLOEXEC not set");
+    }
+}