From f9a8fddf865af3d24009b530ad95a996d8289091 Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Sat, 28 Mar 2026 17:15:09 +0530 Subject: [PATCH 1/5] fix some asyncio cancellation bugs --- Lib/asyncio/base_subprocess.py | 17 ++- Lib/asyncio/unix_events.py | 2 +- Lib/asyncio/windows_events.py | 2 +- Lib/test/test_asyncio/test_subprocess.py | 131 +++++++++++++++++++++++ 4 files changed, 147 insertions(+), 5 deletions(-) diff --git a/Lib/asyncio/base_subprocess.py b/Lib/asyncio/base_subprocess.py index 98e72f212aa203e..353d844c2857829 100644 --- a/Lib/asyncio/base_subprocess.py +++ b/Lib/asyncio/base_subprocess.py @@ -22,7 +22,7 @@ def __init__(self, loop, protocol, args, shell, self._proc = None self._pid = None self._returncode = None - self._exit_waiters = [] + self._exit_waiters = set() self._pending_calls = collections.deque() self._pipes = {} self._finished = False @@ -211,6 +211,11 @@ async def _connect_pipes(self, waiter): except (SystemExit, KeyboardInterrupt): raise except BaseException as exc: + # Close any pipes that were already connected before the + # error/cancellation to avoid leaking file descriptors. + for proto in self._pipes.values(): + if proto is not None: + proto.pipe.close() if waiter is not None and not waiter.cancelled(): waiter.set_exception(exc) else: @@ -260,8 +265,14 @@ async def _wait(self): return self._returncode waiter = self._loop.create_future() - self._exit_waiters.append(waiter) - return await waiter + self._exit_waiters.add(waiter) + try: + return await waiter + finally: + # _process_exited() sets _exit_waiters to None after waking + # all waiters; only discard if cancelled before process exit. + if self._exit_waiters is not None: + self._exit_waiters.discard(waiter) def _try_finish(self): assert not self._finished diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py index d436512d6c45fc5..1a7d534261ed720 100644 --- a/Lib/asyncio/unix_events.py +++ b/Lib/asyncio/unix_events.py @@ -213,7 +213,7 @@ async def _make_subprocess_transport(self, protocol, args, shell, raise except BaseException: transp.close() - await transp._wait() + await tasks.shield(transp._wait()) raise return transp diff --git a/Lib/asyncio/windows_events.py b/Lib/asyncio/windows_events.py index 8241c10b32d105c..efc7c0c158d3905 100644 --- a/Lib/asyncio/windows_events.py +++ b/Lib/asyncio/windows_events.py @@ -406,7 +406,7 @@ async def _make_subprocess_transport(self, protocol, args, shell, raise except BaseException: transp.close() - await transp._wait() + await tasks.shield(transp._wait()) raise return transp diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py index ce127e73f50d8a3..46b1f96a5df65a4 100644 --- a/Lib/test/test_asyncio/test_subprocess.py +++ b/Lib/test/test_asyncio/test_subprocess.py @@ -1028,6 +1028,137 @@ async def main(): self.loop.run_until_complete(main()) + def test_communicate_cancellation_kills_process(self): + async def run(): + proc = await asyncio.create_subprocess_exec( + *PROGRAM_BLOCKED, + stdout=subprocess.PIPE, + ) + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(proc.communicate(), 0.1) + + returncode = await asyncio.wait_for(proc.wait(), 5.0) + self.assertIsNotNone(returncode) + + self.loop.run_until_complete(run()) + + def test_communicate_cancellation_closes_stdout_transport(self): + async def run(): + proc = await asyncio.create_subprocess_exec( + *PROGRAM_BLOCKED, + stdout=subprocess.PIPE, + ) + try: + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(proc.communicate(), 0.1) + + await asyncio.sleep(0) + + stdout_transport = proc._transport.get_pipe_transport(1) + self.assertTrue( + stdout_transport is None or stdout_transport.is_closing(), + "stdout pipe transport not closed after cancellation") + finally: + if proc.returncode is None: + proc.kill() + await proc.wait() + + self.loop.run_until_complete(run()) + + def test_communicate_cancellation_closes_stdin(self): + async def run(): + proc = await asyncio.create_subprocess_exec( + *PROGRAM_BLOCKED, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + ) + try: + large_input = b'x' * (1024 * 1024) + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for( + proc.communicate(large_input), 0.5) + + await asyncio.sleep(0) + + stdin_transport = proc._transport.get_pipe_transport(0) + self.assertTrue( + stdin_transport is None or stdin_transport.is_closing(), + "stdin pipe transport not closed after cancellation") + finally: + if proc.returncode is None: + proc.kill() + await proc.wait() + + self.loop.run_until_complete(run()) + + def test_communicate_cancellation_closes_stderr_transport(self): + async def run(): + proc = await asyncio.create_subprocess_exec( + *PROGRAM_BLOCKED, + stderr=subprocess.PIPE, + ) + try: + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(proc.communicate(), 0.1) + + await asyncio.sleep(0) + + stderr_transport = proc._transport.get_pipe_transport(2) + self.assertTrue( + stderr_transport is None or stderr_transport.is_closing(), + "stderr pipe transport not closed after cancellation") + finally: + if proc.returncode is None: + proc.kill() + await proc.wait() + + self.loop.run_until_complete(run()) + + def test_wait_cancellation_removes_exit_waiters(self): + async def run(): + proc = await asyncio.create_subprocess_exec(*PROGRAM_BLOCKED) + try: + for _ in range(5): + task = self.loop.create_task(proc.wait()) + self.loop.call_soon(task.cancel) + try: + await task + except asyncio.CancelledError: + pass + + self.assertEqual(len(proc._transport._exit_waiters), 0) + finally: + proc.kill() + await proc.wait() + + self.loop.run_until_complete(run()) + + def test_communicate_cancellation_all_pipes(self): + async def run(): + proc = await asyncio.create_subprocess_exec( + *PROGRAM_BLOCKED, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + large_input = b'x' * (1024 * 1024) + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for( + proc.communicate(large_input), 0.5) + + await asyncio.sleep(0) + + for fd, name in [(0, 'stdin'), (1, 'stdout'), (2, 'stderr')]: + transport = proc._transport.get_pipe_transport(fd) + self.assertTrue( + transport is None or transport.is_closing(), + f"{name} pipe transport not closed after cancellation") + + returncode = await asyncio.wait_for(proc.wait(), 5.0) + self.assertIsNotNone(returncode) + + self.loop.run_until_complete(run()) + @warnings_helper.ignore_warnings(category=ResourceWarning) def test_subprocess_read_pipe_cancelled(self): async def main(): From e4926af43c268e8f1d720095001748340bf8cc0d Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Sat, 28 Mar 2026 17:33:43 +0530 Subject: [PATCH 2/5] fix resource warnings --- Lib/asyncio/base_subprocess.py | 3 +++ Lib/test/test_asyncio/test_subprocess.py | 5 +---- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Lib/asyncio/base_subprocess.py b/Lib/asyncio/base_subprocess.py index 353d844c2857829..56a6cd71ee37408 100644 --- a/Lib/asyncio/base_subprocess.py +++ b/Lib/asyncio/base_subprocess.py @@ -216,6 +216,9 @@ async def _connect_pipes(self, waiter): for proto in self._pipes.values(): if proto is not None: proto.pipe.close() + for raw_pipe in (proc.stdin, proc.stdout, proc.stderr): + if raw_pipe is not None and not raw_pipe.closed: + raw_pipe.close() if waiter is not None and not waiter.cancelled(): waiter.set_exception(exc) else: diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py index 46b1f96a5df65a4..5314fddecb86dd7 100644 --- a/Lib/test/test_asyncio/test_subprocess.py +++ b/Lib/test/test_asyncio/test_subprocess.py @@ -12,7 +12,7 @@ from asyncio import subprocess from test.test_asyncio import utils as test_utils from test import support -from test.support import os_helper, warnings_helper, gc_collect +from test.support import os_helper, gc_collect if not support.has_subprocess_support: raise unittest.SkipTest("test module requires subprocess") @@ -1159,7 +1159,6 @@ async def run(): self.loop.run_until_complete(run()) - @warnings_helper.ignore_warnings(category=ResourceWarning) def test_subprocess_read_pipe_cancelled(self): async def main(): loop = asyncio.get_running_loop() @@ -1170,7 +1169,6 @@ async def main(): asyncio.run(main()) gc_collect() - @warnings_helper.ignore_warnings(category=ResourceWarning) def test_subprocess_write_pipe_cancelled(self): async def main(): loop = asyncio.get_running_loop() @@ -1181,7 +1179,6 @@ async def main(): asyncio.run(main()) gc_collect() - @warnings_helper.ignore_warnings(category=ResourceWarning) def test_subprocess_read_write_pipe_cancelled(self): async def main(): loop = asyncio.get_running_loop() From 2424a20a31499354b9857ba37f2c3514b368870f Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Sat, 28 Mar 2026 17:56:33 +0530 Subject: [PATCH 3/5] fix windows --- Lib/asyncio/base_subprocess.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/asyncio/base_subprocess.py b/Lib/asyncio/base_subprocess.py index 56a6cd71ee37408..22a60f0a16852f0 100644 --- a/Lib/asyncio/base_subprocess.py +++ b/Lib/asyncio/base_subprocess.py @@ -217,7 +217,7 @@ async def _connect_pipes(self, waiter): if proto is not None: proto.pipe.close() for raw_pipe in (proc.stdin, proc.stdout, proc.stderr): - if raw_pipe is not None and not raw_pipe.closed: + if raw_pipe is not None: raw_pipe.close() if waiter is not None and not waiter.cancelled(): waiter.set_exception(exc) From 1c40f5a84b4b472dd22b0a23e48dc9686909689e Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Tue, 21 Jul 2026 23:08:03 +0530 Subject: [PATCH 4/5] remove tests --- Lib/test/test_asyncio/test_subprocess.py | 131 ----------------------- 1 file changed, 131 deletions(-) diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py index 5314fddecb86dd7..f9b4fe53acb1343 100644 --- a/Lib/test/test_asyncio/test_subprocess.py +++ b/Lib/test/test_asyncio/test_subprocess.py @@ -1028,137 +1028,6 @@ async def main(): self.loop.run_until_complete(main()) - def test_communicate_cancellation_kills_process(self): - async def run(): - proc = await asyncio.create_subprocess_exec( - *PROGRAM_BLOCKED, - stdout=subprocess.PIPE, - ) - with self.assertRaises(asyncio.TimeoutError): - await asyncio.wait_for(proc.communicate(), 0.1) - - returncode = await asyncio.wait_for(proc.wait(), 5.0) - self.assertIsNotNone(returncode) - - self.loop.run_until_complete(run()) - - def test_communicate_cancellation_closes_stdout_transport(self): - async def run(): - proc = await asyncio.create_subprocess_exec( - *PROGRAM_BLOCKED, - stdout=subprocess.PIPE, - ) - try: - with self.assertRaises(asyncio.TimeoutError): - await asyncio.wait_for(proc.communicate(), 0.1) - - await asyncio.sleep(0) - - stdout_transport = proc._transport.get_pipe_transport(1) - self.assertTrue( - stdout_transport is None or stdout_transport.is_closing(), - "stdout pipe transport not closed after cancellation") - finally: - if proc.returncode is None: - proc.kill() - await proc.wait() - - self.loop.run_until_complete(run()) - - def test_communicate_cancellation_closes_stdin(self): - async def run(): - proc = await asyncio.create_subprocess_exec( - *PROGRAM_BLOCKED, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - ) - try: - large_input = b'x' * (1024 * 1024) - with self.assertRaises(asyncio.TimeoutError): - await asyncio.wait_for( - proc.communicate(large_input), 0.5) - - await asyncio.sleep(0) - - stdin_transport = proc._transport.get_pipe_transport(0) - self.assertTrue( - stdin_transport is None or stdin_transport.is_closing(), - "stdin pipe transport not closed after cancellation") - finally: - if proc.returncode is None: - proc.kill() - await proc.wait() - - self.loop.run_until_complete(run()) - - def test_communicate_cancellation_closes_stderr_transport(self): - async def run(): - proc = await asyncio.create_subprocess_exec( - *PROGRAM_BLOCKED, - stderr=subprocess.PIPE, - ) - try: - with self.assertRaises(asyncio.TimeoutError): - await asyncio.wait_for(proc.communicate(), 0.1) - - await asyncio.sleep(0) - - stderr_transport = proc._transport.get_pipe_transport(2) - self.assertTrue( - stderr_transport is None or stderr_transport.is_closing(), - "stderr pipe transport not closed after cancellation") - finally: - if proc.returncode is None: - proc.kill() - await proc.wait() - - self.loop.run_until_complete(run()) - - def test_wait_cancellation_removes_exit_waiters(self): - async def run(): - proc = await asyncio.create_subprocess_exec(*PROGRAM_BLOCKED) - try: - for _ in range(5): - task = self.loop.create_task(proc.wait()) - self.loop.call_soon(task.cancel) - try: - await task - except asyncio.CancelledError: - pass - - self.assertEqual(len(proc._transport._exit_waiters), 0) - finally: - proc.kill() - await proc.wait() - - self.loop.run_until_complete(run()) - - def test_communicate_cancellation_all_pipes(self): - async def run(): - proc = await asyncio.create_subprocess_exec( - *PROGRAM_BLOCKED, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) - large_input = b'x' * (1024 * 1024) - with self.assertRaises(asyncio.TimeoutError): - await asyncio.wait_for( - proc.communicate(large_input), 0.5) - - await asyncio.sleep(0) - - for fd, name in [(0, 'stdin'), (1, 'stdout'), (2, 'stderr')]: - transport = proc._transport.get_pipe_transport(fd) - self.assertTrue( - transport is None or transport.is_closing(), - f"{name} pipe transport not closed after cancellation") - - returncode = await asyncio.wait_for(proc.wait(), 5.0) - self.assertIsNotNone(returncode) - - self.loop.run_until_complete(run()) - def test_subprocess_read_pipe_cancelled(self): async def main(): loop = asyncio.get_running_loop() From 8f8161793e37e2076cd6efc546b33d0aa8d4634d Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Tue, 21 Jul 2026 23:11:29 +0530 Subject: [PATCH 5/5] remove comment --- Lib/asyncio/base_subprocess.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/Lib/asyncio/base_subprocess.py b/Lib/asyncio/base_subprocess.py index 22a60f0a16852f0..0423bc100b2d932 100644 --- a/Lib/asyncio/base_subprocess.py +++ b/Lib/asyncio/base_subprocess.py @@ -272,8 +272,6 @@ async def _wait(self): try: return await waiter finally: - # _process_exited() sets _exit_waiters to None after waking - # all waiters; only discard if cancelled before process exit. if self._exit_waiters is not None: self._exit_waiters.discard(waiter)