Skip to content

Commit f252132

Browse files
gh-119710: fix asyncio Process.wait() to finish on process exit and not wait for closing of pipes (#151983)
Co-authored-by: Kumar Aditya <kumaraditya@python.org>
1 parent 5625b18 commit f252132

3 files changed

Lines changed: 53 additions & 47 deletions

File tree

Lib/asyncio/base_subprocess.py

Lines changed: 9 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ def __init__(self, loop, protocol, args, shell,
2626
self._pending_calls = collections.deque()
2727
self._pipes = {}
2828
self._finished = False
29-
self._pipes_connected = False
3029

3130
if stdin == subprocess.PIPE:
3231
self._pipes[0] = None
@@ -214,7 +213,6 @@ async def _connect_pipes(self, waiter):
214213
else:
215214
if waiter is not None and not waiter.cancelled():
216215
waiter.set_result(None)
217-
self._pipes_connected = True
218216

219217
def _call(self, cb, *data):
220218
if self._pending_calls is not None:
@@ -235,6 +233,7 @@ def _process_exited(self, returncode):
235233
if self._loop.get_debug():
236234
logger.info('%r exited with return code %r', self, returncode)
237235
self._returncode = returncode
236+
238237
if self._proc.returncode is None:
239238
# asyncio uses a child watcher: copy the status into the Popen
240239
# object. On Python 3.6, it is required to avoid a ResourceWarning.
@@ -243,6 +242,13 @@ def _process_exited(self, returncode):
243242

244243
self._try_finish()
245244

245+
# gh-119710: Wake up futures waiting for wait() as soon as the process
246+
# exits.
247+
for waiter in self._exit_waiters:
248+
if not waiter.done():
249+
waiter.set_result(returncode)
250+
self._exit_waiters = None
251+
246252
async def _wait(self):
247253
"""Wait until the process exit and return the process return code.
248254
@@ -258,15 +264,7 @@ def _try_finish(self):
258264
assert not self._finished
259265
if self._returncode is None:
260266
return
261-
if not self._pipes_connected:
262-
# self._pipes_connected can be False if not all pipes were connected
263-
# because either the process failed to start or the self._connect_pipes task
264-
# got cancelled. In this broken state we consider all pipes disconnected and
265-
# to avoid hanging forever in self._wait as otherwise _exit_waiters
266-
# would never be woken up, we wake them up here.
267-
for waiter in self._exit_waiters:
268-
if not waiter.done():
269-
waiter.set_result(self._returncode)
267+
270268
if all(p is not None and p.disconnected
271269
for p in self._pipes.values()):
272270
self._finished = True
@@ -276,11 +274,6 @@ def _call_connection_lost(self, exc):
276274
try:
277275
self._protocol.connection_lost(exc)
278276
finally:
279-
# wake up futures waiting for wait()
280-
for waiter in self._exit_waiters:
281-
if not waiter.done():
282-
waiter.set_result(self._returncode)
283-
self._exit_waiters = None
284277
self._loop = None
285278
self._proc = None
286279
self._protocol = None

Lib/test/test_asyncio/test_subprocess.py

Lines changed: 40 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -112,37 +112,6 @@ def test_subprocess_repr(self):
112112
)
113113
transport.close()
114114

115-
def test_proc_exited_no_invalid_state_error_on_exit_waiters(self):
116-
# gh-145541: when _connect_pipes hasn't completed (so
117-
# _pipes_connected is False) and the process exits, _try_finish()
118-
# sets the result on exit waiters. Then _call_connection_lost() must
119-
# not call set_result() again on the same waiters.
120-
self.loop.set_exception_handler(
121-
lambda loop, context: self.fail(
122-
f"unexpected exception: {context}")
123-
)
124-
waiter = self.loop.create_future()
125-
transport, protocol = self.create_transport(waiter)
126-
127-
# Simulate a waiter registered via _wait() before the process exits.
128-
exit_waiter = self.loop.create_future()
129-
transport._exit_waiters.append(exit_waiter)
130-
131-
# _connect_pipes hasn't completed, so _pipes_connected is False.
132-
self.assertFalse(transport._pipes_connected)
133-
134-
# Simulate process exit. _try_finish() will set the result on
135-
# exit_waiter because _pipes_connected is False, and then schedule
136-
# _call_connection_lost() because _pipes is empty (vacuously all
137-
# disconnected). _call_connection_lost() must skip exit_waiter
138-
# because it's already done.
139-
transport._process_exited(6)
140-
self.loop.run_until_complete(waiter)
141-
142-
self.assertEqual(exit_waiter.result(), 6)
143-
144-
transport.close()
145-
146115

147116
class SubprocessMixin:
148117

@@ -436,6 +405,46 @@ async def len_message(message):
436405
self.assertEqual(output.rstrip(), b'3')
437406
self.assertEqual(exitcode, 0)
438407

408+
def test_wait_even_if_pipe_is_open(self):
409+
# gh-119710: Process.wait() must return once the process exits even
410+
# if its stdout pipe is inherited by a grandchild that keeps it open,
411+
# so the pipe never reaches EOF. Otherwise wait() hangs forever
412+
# despite the returncode being known.
413+
414+
async def run():
415+
# The grandchild inherits the child's stdin and stdout pipes and
416+
# keeps both open after the child is killed. It writes "ready"
417+
# so we know it has started, and exits once its stdin hits EOF.
418+
code = textwrap.dedent("""\
419+
import subprocess, sys
420+
subprocess.run([sys.executable, "-c",
421+
"import sys; sys.stdout.write('ready');"
422+
" sys.stdout.flush(); sys.stdin.read()"])
423+
""")
424+
425+
proc = await asyncio.create_subprocess_exec(
426+
sys.executable, "-c", code,
427+
stdin=subprocess.PIPE,
428+
stdout=subprocess.PIPE,
429+
)
430+
try:
431+
wait_proc = asyncio.create_task(proc.wait())
432+
# Wait until the grandchild holds the inherited pipes; this
433+
# also lets the wait() task register its waiter.
434+
await proc.stdout.readexactly(5)
435+
proc.kill()
436+
returncode = await asyncio.wait_for(
437+
wait_proc, timeout=support.SHORT_TIMEOUT)
438+
if sys.platform == 'win32':
439+
self.assertIsInstance(returncode, int)
440+
else:
441+
self.assertEqual(-signal.SIGKILL, returncode)
442+
finally:
443+
proc.stdin.close() # let the grandchild exit
444+
await proc.stdout.read()
445+
446+
self.loop.run_until_complete(run())
447+
439448
def test_empty_input(self):
440449

441450
async def empty_input():
Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
Fix :mod:`asyncio` subprocess :meth:`~asyncio.subprocess.Process.wait`
2+
hanging when the process has exited but one of its pipes is kept open by an
3+
inherited child process (so the pipe never reaches EOF). ``wait()`` now
4+
returns as soon as the process exits, regardless of the pipes' state.

0 commit comments

Comments
 (0)