a fix for fd leak

Author: egorsmkv

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "rsloop"
-version = "0.1.12"
+version = "0.1.13"
 edition = "2021"
 description = "An event loop for asyncio written in Rust"
 license = "Apache-2.0"

Cargo.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "rsloop"
-version = "0.1.12"
+version = "0.1.13"
 description = "An event loop for asyncio written in Rust"
 readme = "README.md"
 license = { file = "LICENSE" }

pyproject.toml

@@ -31,13 +31,14 @@ use crate::process_transport::{
 use crate::python_names;
 use crate::stream_transport::{
     create_server as create_py_server, remove_unix_socket_if_present, spawn_read_pipe_transport,
-    spawn_write_pipe_transport, start_tls_transport, tcp_listener_from_socket_fd,
-    tcp_server_listener, transport_from_socket, transport_from_socket_server_tls,
-    transport_from_socket_tls, PyServer, PyStreamTransport, ServerCreateParams,
-    TransportSpawnContext,
+    spawn_write_pipe_transport, start_tls_transport, tcp_server_listener, transport_from_socket,
+    transport_from_socket_server_tls, transport_from_socket_tls, PyServer, PyStreamTransport,
+    ServerCreateParams, TransportSpawnContext,
 };
 #[cfg(unix)]
-use crate::stream_transport::{unix_listener_from_socket_fd, unix_server_listener};
+use crate::stream_transport::{
+    tcp_listener_from_owned_socket_fd, unix_listener_from_owned_socket_fd, unix_server_listener,
+};
 use crate::tls::{client_tls_settings, server_tls_settings};
 
 static ASYNCIO_TASK_CLS: OnceLock<Py<PyAny>> = OnceLock::new();
@@ -658,15 +659,15 @@ fn listener_sources_from_sockets(
         #[cfg(unix)]
         {
             listeners.push(if family == libc::AF_UNIX {
-                unix_server_listener(unix_listener_from_socket_fd(fd)?)
+                unix_server_listener(unix_listener_from_owned_socket_fd(fd)?)
             } else {
-                tcp_server_listener(tcp_listener_from_socket_fd(fd)?)
+                tcp_server_listener(tcp_listener_from_owned_socket_fd(fd)?)
             });
         }
         #[cfg(not(unix))]
         {
             let _ = family;
-            listeners.push(tcp_server_listener(tcp_listener_from_socket_fd(fd)?));
+            listeners.push(tcp_server_listener(tcp_listener_from_owned_socket_fd(fd)?));
         }
     }
     Ok(listeners)

src/python_api.rs

@@ -1546,12 +1546,12 @@ impl StreamTransportCore {
         let family = socket.bind(py).getattr("family")?.extract::<i32>()?;
         #[cfg(unix)]
         let stream = if family == libc::AF_UNIX {
-            StreamKind::Unix(unix_stream_from_socket_fd(fd)?)
+            StreamKind::Unix(unix_stream_from_owned_socket_fd(fd)?)
         } else {
-            StreamKind::Tcp(tcp_stream_from_socket_fd(fd)?)
+            StreamKind::Tcp(tcp_stream_from_owned_socket_fd(fd)?)
         };
         #[cfg(not(unix))]
-        let stream = StreamKind::Tcp(tcp_stream_from_socket_fd(fd)?);
+        let stream = StreamKind::Tcp(tcp_stream_from_owned_socket_fd(fd)?);
 
         Ok((
             TransportSpawnContext {
@@ -2188,11 +2188,21 @@ pub fn transport_from_socket(
     #[cfg(unix)]
     if family == libc::AF_UNIX {
         let fd = detached_socket_handle(py, &socket_obj)?;
-        return spawn_unix_transport(py, spawn_context, unix_stream_from_socket_fd(fd)?, None);
+        return spawn_unix_transport(
+            py,
+            spawn_context,
+            unix_stream_from_owned_socket_fd(fd)?,
+            None,
+        );
     }
 
     let fd = detached_socket_handle(py, &socket_obj)?;
-    spawn_tcp_transport(py, spawn_context, tcp_stream_from_socket_fd(fd)?, None)
+    spawn_tcp_transport(
+        py,
+        spawn_context,
+        tcp_stream_from_owned_socket_fd(fd)?,
+        None,
+    )
 }
 
 pub fn transport_from_socket_tls(
@@ -2220,7 +2230,7 @@ pub fn transport_from_socket_tls(
         return spawn_tls_client_transport(
             py,
             spawn_context,
-            StreamKind::Unix(unix_stream_from_socket_fd(fd)?),
+            StreamKind::Unix(unix_stream_from_owned_socket_fd(fd)?),
             tls,
             None,
             true,
@@ -2231,7 +2241,7 @@ pub fn transport_from_socket_tls(
     spawn_tls_client_transport(
         py,
         spawn_context,
-        StreamKind::Tcp(tcp_stream_from_socket_fd(fd)?),
+        StreamKind::Tcp(tcp_stream_from_owned_socket_fd(fd)?),
         tls,
         None,
         true,
@@ -2263,7 +2273,7 @@ pub fn transport_from_socket_server_tls(
         return spawn_tls_server_transport(
             py,
             spawn_context,
-            StreamKind::Unix(unix_stream_from_socket_fd(fd)?),
+            StreamKind::Unix(unix_stream_from_owned_socket_fd(fd)?),
             tls,
             None,
             true,
@@ -2274,7 +2284,7 @@ pub fn transport_from_socket_server_tls(
     spawn_tls_server_transport(
         py,
         spawn_context,
-        StreamKind::Tcp(tcp_stream_from_socket_fd(fd)?),
+        StreamKind::Tcp(tcp_stream_from_owned_socket_fd(fd)?),
         tls,
         None,
         true,
@@ -2576,28 +2586,26 @@ pub fn spawn_write_pipe_transport(
     Ok(transport)
 }
 
-pub fn tcp_stream_from_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdTcpStream> {
-    duplicate_configured_tcp_stream(fd)
+pub fn tcp_stream_from_owned_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdTcpStream> {
+    configured_tcp_stream_from_owned_fd(fd)
 }
 
 #[cfg(unix)]
-pub fn unix_stream_from_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdUnixStream> {
-    let dup = fd_ops::dup_raw_fd(fd).map_err(|err| PyRuntimeError::new_err(err.to_string()))?;
-    let stream = unsafe { StdUnixStream::from_raw_fd(raw_fd_for_std(dup)?) };
+pub fn unix_stream_from_owned_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdUnixStream> {
+    let stream = unsafe { StdUnixStream::from_raw_fd(raw_fd_for_std(fd)?) };
     stream
         .set_nonblocking(true)
         .map_err(|err| PyRuntimeError::new_err(err.to_string()))?;
     Ok(stream)
 }
 
-pub fn tcp_listener_from_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdTcpListener> {
-    duplicate_configured_tcp_listener(fd)
+pub fn tcp_listener_from_owned_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdTcpListener> {
+    configured_tcp_listener_from_owned_fd(fd)
 }
 
 #[cfg(unix)]
-pub fn unix_listener_from_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdUnixListener> {
-    let dup = fd_ops::dup_raw_fd(fd).map_err(|err| PyRuntimeError::new_err(err.to_string()))?;
-    let listener = unsafe { StdUnixListener::from_raw_fd(raw_fd_for_std(dup)?) };
+pub fn unix_listener_from_owned_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdUnixListener> {
+    let listener = unsafe { StdUnixListener::from_raw_fd(raw_fd_for_std(fd)?) };
     listener
         .set_nonblocking(true)
         .map_err(|err| PyRuntimeError::new_err(err.to_string()))?;
@@ -2606,17 +2614,20 @@ pub fn unix_listener_from_socket_fd(fd: fd_ops::RawFd) -> PyResult<StdUnixListen
 
 fn duplicate_configured_tcp_stream(fd: fd_ops::RawFd) -> PyResult<StdTcpStream> {
     let dup = fd_ops::dup_raw_fd(fd).map_err(|err| PyRuntimeError::new_err(err.to_string()))?;
-    let socket = socket_from_owned_raw(dup)?;
+    configured_tcp_stream_from_owned_fd(dup)
+}
+
+fn configured_tcp_stream_from_owned_fd(fd: fd_ops::RawFd) -> PyResult<StdTcpStream> {
+    let socket = socket_from_owned_raw(fd)?;
     socket
         .set_nonblocking(true)
         .map_err(|err| PyRuntimeError::new_err(err.to_string()))?;
     let _ = socket.set_tcp_nodelay(true);
     Ok(socket.into())
 }
 
-fn duplicate_configured_tcp_listener(fd: fd_ops::RawFd) -> PyResult<StdTcpListener> {
-    let dup = fd_ops::dup_raw_fd(fd).map_err(|err| PyRuntimeError::new_err(err.to_string()))?;
-    let socket = socket_from_owned_raw(dup)?;
+fn configured_tcp_listener_from_owned_fd(fd: fd_ops::RawFd) -> PyResult<StdTcpListener> {
+    let socket = socket_from_owned_raw(fd)?;
     socket
         .set_nonblocking(true)
         .map_err(|err| PyRuntimeError::new_err(err.to_string()))?;

src/stream_transport.rs

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import gc
 import os
 import shlex
 import socket
@@ -36,6 +37,51 @@ async def main() -> str:
 
         self.assertEqual(rsloop.run(main()), "ok")
 
+    @unittest.skipUnless(
+        os.name == "posix" and os.path.isdir("/proc/self/fd"),
+        "requires /proc/self/fd",
+    )
+    def test_create_connection_does_not_leak_file_descriptors(self) -> None:
+        def fd_count() -> int:
+            return len(os.listdir("/proc/self/fd"))
+
+        async def main() -> None:
+            loop = asyncio.get_running_loop()
+
+            class ServerProtocol(asyncio.Protocol):
+                def connection_made(self, transport: asyncio.BaseTransport) -> None:
+                    transport.close()
+
+            server = await loop.create_server(ServerProtocol, "127.0.0.1", 0)
+            try:
+                port = server.sockets[0].getsockname()[1]
+
+                for _ in range(64):
+                    closed = loop.create_future()
+
+                    class ClientProtocol(asyncio.Protocol):
+                        def connection_made(
+                            self, transport: asyncio.BaseTransport
+                        ) -> None:
+                            transport.close()
+
+                        def connection_lost(self, exc: Exception | None) -> None:
+                            if not closed.done():
+                                closed.set_result(None)
+
+                    await loop.create_connection(ClientProtocol, "127.0.0.1", port)
+                    await asyncio.wait_for(closed, 1.0)
+                    await asyncio.sleep(0)
+            finally:
+                server.close()
+                await server.wait_closed()
+
+        before = fd_count()
+        rsloop.run(main())
+        gc.collect()
+        after = fd_count()
+        self.assertLessEqual(after, before + 4)
+
     def test_connect_pipe_round_trip(self) -> None:
         async def main() -> tuple[str, str]:
             loop = asyncio.get_running_loop()