return transp
def _child_watcher_callback(self, pid, returncode, transp):
- self.call_soon_threadsafe(transp._process_exited, returncode)
+ transp._process_exited(returncode)
async def create_unix_connection(
self, protocol_factory, path=None, *,
def _do_waitpid(self, loop, expected_pid, callback, args):
assert expected_pid > 0
+ if hasattr(os, 'waitid'):
+ # Wait for the child process using waitid() on platforms which support it.
+ # WNOWAIT is used to avoid reaping the child process, allowing the event loop to
+ # reap the child process with waitpid() later in event loop thread.
+ # This makes the reaping of the child and notification of the return code
+ # atomic with respect to the event loop thread.
+ try:
+ os.waitid(os.P_PID, expected_pid, os.WEXITED | os.WNOWAIT)
+ except ChildProcessError:
+ # The child process is already reaped
+ pass
+ if loop.is_closed():
+ # loop is already closed, reap the zombie here so that it is not leaked.
+ pid, _ = self._reap(loop, expected_pid)
+ logger.warning("Loop %r that handles pid %r is closed",
+ loop, pid)
+ else:
+ try:
+ loop.call_soon_threadsafe(
+ self._reap_and_notify, loop, expected_pid,
+ callback, args)
+ except RuntimeError:
+ # The event loop was closed concurrently.
+ pid, _ = self._reap(loop, expected_pid)
+ logger.warning("Loop %r that handles pid %r is closed",
+ loop, pid)
+ else:
+ # Fallback for platforms that don't support waitid(): we have to
+ # reap the child here, which is racy with respect to send_signal()
+ pid, returncode = self._reap(loop, expected_pid)
+ if loop.is_closed():
+ logger.warning("Loop %r that handles pid %r is closed",
+ loop, pid)
+ else:
+ loop.call_soon_threadsafe(callback, pid, returncode, *args)
+
+ self._threads.pop(expected_pid)
+
+ def _reap_and_notify(self, loop, expected_pid, callback, args):
+ pid, returncode = self._reap(loop, expected_pid)
+ callback(pid, returncode, *args)
+
+ def _reap(self, loop, expected_pid):
try:
pid, status = os.waitpid(expected_pid, 0)
except ChildProcessError:
if loop.get_debug():
logger.debug('process %s exited with returncode %s',
expected_pid, returncode)
-
- if loop.is_closed():
- logger.warning("Loop %r that handles pid %r is closed", loop, pid)
- else:
- loop.call_soon_threadsafe(callback, pid, returncode, *args)
-
- self._threads.pop(expected_pid)
+ return pid, returncode
def can_use_pidfd():
if not hasattr(os, 'pidfd_open'):
else:
self.assertIsInstance(watcher, unix_events._ThreadedChildWatcher)
+ @unittest.skipUnless(hasattr(os, 'waitid'), 'needs os.waitid()')
+ def test_send_signal_never_targets_reaped_pid(self):
+ # gh-127049: there must be no window between the child watcher
+ # reaping the child and asyncio publishing the exit in which
+ # send_signal() signals the freed (possibly recycled) PID.
+ reaped_pids = set()
+ stale_kills = []
+ orig_waitpid = os.waitpid
+ orig_kill = os.kill
+
+ def waitpid(pid, options):
+ res = orig_waitpid(pid, options)
+ if res[0] != 0:
+ reaped_pids.add(res[0])
+ return res
+
+ def kill(pid, sig):
+ if pid in reaped_pids:
+ stale_kills.append(pid)
+ return
+ orig_kill(pid, sig)
+
+ async def run():
+ with mock.patch('os.waitpid', waitpid), \
+ mock.patch('os.kill', kill):
+ proc = await asyncio.create_subprocess_exec(
+ *PROGRAM_BLOCKED)
+ proc.kill()
+ deadline = self.loop.time() + support.SHORT_TIMEOUT
+ while proc.returncode is None:
+ if self.loop.time() > deadline:
+ self.fail('child exit was not published in time')
+ proc.kill()
+ await asyncio.sleep(0)
+ await proc.wait()
+ self.assertEqual(stale_kills, [])
+
+ self.loop.run_until_complete(run())
+
class SubprocessThreadedWatcherTests(SubprocessWatcherMixin,
test_utils.TestCase):
unix_events.can_use_pidfd = self._original_can_use_pidfd
return super().tearDown()
+ @unittest.skipUnless(hasattr(os, 'waitid'), 'needs os.waitid()')
+ def test_pid_not_reaped_before_exit_published(self):
+ # gh-127049: the watcher thread must wait for the child
+ # without reaping it: the PID must stay reserved until the
+ # event loop thread reaps it and publishes the exit as one
+ # atomic step.
+ async def run():
+ proc = await asyncio.create_subprocess_exec(
+ *PROGRAM_BLOCKED)
+ thread = self.loop._watcher._threads.get(proc.pid)
+ self.assertIsNotNone(thread)
+ proc.kill()
+ # Wait for the watcher thread to observe the exit while
+ # the event loop cannot process the notification yet.
+ thread.join(support.SHORT_TIMEOUT)
+ self.assertFalse(thread.is_alive())
+ # The exit has not been published yet...
+ self.assertIsNone(proc.returncode)
+ # ...so the child must still be an unreaped zombie and
+ # signalling its PID must still be safe. waitid() raises
+ # ChildProcessError if the PID was already reaped (and
+ # possibly recycled by the kernel).
+ os.waitid(os.P_PID, proc.pid,
+ os.WEXITED | os.WNOWAIT | os.WNOHANG)
+ proc.kill()
+ self.assertEqual(await proc.wait(), -signal.SIGKILL)
+
+ self.loop.run_until_complete(run())
+
@unittest.skipUnless(
unix_events.can_use_pidfd(),
"operating system does not support pidfds",
projects / thirdparty / Python / cpython.git / commitdiff
