diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index f26fba175b63cd7..57d205ee782bd9e 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -1497,10 +1497,15 @@ async def create_datagram_endpoint(self, protocol_factory, else: raise exceptions[0] - protocol = protocol_factory() - waiter = self.create_future() - transport = self._make_datagram_transport( - sock, protocol, r_addr, waiter) + try: + protocol = protocol_factory() + waiter = self.create_future() + transport = self._make_datagram_transport( + sock, protocol, r_addr, waiter) + except: + # gh-156400: close the socket if the transport is never created. + sock.close() + raise if self._debug: if local_addr: logger.info("Datagram endpoint local_addr=%r remote_addr=%r " @@ -1714,9 +1719,14 @@ async def connect_accepted_socket( return transport, protocol async def connect_read_pipe(self, protocol_factory, pipe): - protocol = protocol_factory() - waiter = self.create_future() - transport = self._make_read_pipe_transport(pipe, protocol, waiter) + try: + protocol = protocol_factory() + waiter = self.create_future() + transport = self._make_read_pipe_transport(pipe, protocol, waiter) + except: + # gh-156400: close the pipe if the transport is never created. + pipe.close() + raise try: await waiter @@ -1730,9 +1740,14 @@ async def connect_read_pipe(self, protocol_factory, pipe): return transport, protocol async def connect_write_pipe(self, protocol_factory, pipe): - protocol = protocol_factory() - waiter = self.create_future() - transport = self._make_write_pipe_transport(pipe, protocol, waiter) + try: + protocol = protocol_factory() + waiter = self.create_future() + transport = self._make_write_pipe_transport(pipe, protocol, waiter) + except: + # gh-156400: close the pipe if the transport is never created. + pipe.close() + raise try: await waiter diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py index 18afdca23163a1e..e11f77ef10c0c96 100644 --- a/Lib/test/test_asyncio/test_base_events.py +++ b/Lib/test/test_asyncio/test_base_events.py @@ -2041,6 +2041,43 @@ def test_create_datagram_endpoint_sock(self): self.loop.run_until_complete(protocol.done) self.assertEqual('CLOSED', protocol.state) + def test_create_datagram_endpoint_transport_error_closes_sock(self): + # gh-156400: the socket is closed if the transport is never created. + sock = mock.Mock() + sock.type = socket.SOCK_DGRAM + + def factory(): + raise ZeroDivisionError + + coro = self.loop.create_datagram_endpoint(factory, sock=sock) + with self.assertRaises(ZeroDivisionError): + self.loop.run_until_complete(coro) + self.assertTrue(sock.close.called) + + def test_connect_read_pipe_transport_error_closes_pipe(self): + # gh-156400: the pipe is closed if the transport is never created. + pipe = mock.Mock() + + def factory(): + raise ZeroDivisionError + + coro = self.loop.connect_read_pipe(factory, pipe) + with self.assertRaises(ZeroDivisionError): + self.loop.run_until_complete(coro) + self.assertTrue(pipe.close.called) + + def test_connect_write_pipe_transport_error_closes_pipe(self): + # gh-156400: the pipe is closed if the transport is never created. + pipe = mock.Mock() + + def factory(): + raise ZeroDivisionError + + coro = self.loop.connect_write_pipe(factory, pipe) + with self.assertRaises(ZeroDivisionError): + self.loop.run_until_complete(coro) + self.assertTrue(pipe.close.called) + @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets') def test_create_datagram_endpoint_sock_unix(self): fut = self.loop.create_datagram_endpoint( diff --git a/Misc/NEWS.d/next/Library/2026-08-26-13-58-01.gh-issue-156400.dGrmP1.rst b/Misc/NEWS.d/next/Library/2026-08-26-13-58-01.gh-issue-156400.dGrmP1.rst new file mode 100644 index 000000000000000..bad5109d59cfac1 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-08-26-13-58-01.gh-issue-156400.dGrmP1.rst @@ -0,0 +1,6 @@ +Fix socket and pipe leaks in :mod:`asyncio` when ``protocol_factory()`` or +transport creation fails in :meth:`loop.create_datagram_endpoint +`, :meth:`loop.connect_read_pipe +`, and :meth:`loop.connect_write_pipe +`. The socket or pipe is now closed instead +of leaking until garbage collection.