call exit callback even if AsyncProcess is reaped elsewhere

Author: graingert

distributed/process.py

@@ -232,9 +232,11 @@ def _start():
     def _watch_process(cls, selfref, process, state, q):
         r = repr(selfref())
         process.join()
-        exitcode = process.exitcode
-        assert exitcode is not None
-        logger.debug("[%s] process %r exited with code %r", r, state.pid, exitcode)
+        exitcode = original_exit_code = process.exitcode
+        if exitcode is None:
+            # The child process is already reaped
+            # (may happen if waitpid() is called elsewhere).
+            exitcode = 255
         state.is_alive = False
         state.exitcode = exitcode
         # Make sure the process is removed from the global list
@@ -247,6 +249,16 @@ def _watch_process(cls, selfref, process, state, q):
         finally:
             self = None  # lose reference
 
+        # logging may fail - defer calls to after the callback is added
+        if original_exit_code is None:
+            logger.warning(
+                "[%s] process %r exit status was already read will report exitcode 255",
+                r,
+                state.pid,
+            )
+        else:
+            logger.debug("[%s] process %r exited with code %r", r, state.pid, exitcode)
+
     def start(self):
         """
         Start the child process.

distributed/tests/test_asyncprocess.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
 import asyncio
+import concurrent.futures
 import gc
+import multiprocessing
 import os
 import signal
+import socket
 import sys
 import threading
 import weakref
@@ -12,11 +15,9 @@
 
 import psutil
 import pytest
-from tornado import gen
 from tornado.ioloop import IOLoop
-from tornado.locks import Event
 
-from distributed.compatibility import LINUX, MACOS, WINDOWS
+from distributed.compatibility import LINUX, MACOS, WINDOWS, to_thread
 from distributed.metrics import time
 from distributed.process import AsyncProcess
 from distributed.utils import get_mp_context
@@ -238,13 +239,9 @@ async def test_close():
 async def test_exit_callback():
     to_child = get_mp_context().Queue()
     from_child = get_mp_context().Queue()
-    evt = Event()
+    evt = asyncio.Event()
 
-    # FIXME: this breaks if changed to async def...
-    @gen.coroutine
     def on_stop(_proc):
-        assert _proc is proc
-        yield gen.moment
         evt.set()
 
     # Normal process exit
@@ -275,10 +272,69 @@ def on_stop(_proc):
     assert not evt.is_set()
 
     await proc.terminate()
-    await evt.wait(timedelta(seconds=5))
+    await asyncio.wait_for(evt.wait(), 5)
     assert evt.is_set()
 
 
+def _run_and_close_tornado(async_fn, /, *args, **kwargs):
+    tornado_loop = None
+
+    async def inner_fn():
+        nonlocal tornado_loop
+        tornado_loop = IOLoop.current()
+        return await async_fn(*args, **kwargs)
+
+    try:
+        return asyncio.run(inner_fn())
+    finally:
+        tornado_loop.close(all_fds=True)
+
+
+def _write_byte_wait_closed(sock):
+    with sock:
+        sock.send(b"\x00")
+        sock.recv(1)
+
+
+async def _check_process_reaped_elsewhere():
+    loop = asyncio.get_running_loop()
+
+    def psutil_terminate(pid):
+        proc = psutil.Process(pid)
+        proc.terminate()
+        proc.wait()
+
+    a, b = socket.socketpair()
+    with a:
+        with b:
+            proc = AsyncProcess(
+                target=_write_byte_wait_closed, args=(b,), loop=IOLoop.current()
+            )
+            await proc.start()
+
+        a.setblocking(False)
+        assert await loop.sock_recv(a, 1) == b"\00"
+        await to_thread(psutil_terminate, proc.pid)
+        await proc.join()
+        return proc.exitcode
+
+
+def test_process_reaped_elsewhere(cleanup):
+    with concurrent.futures.ProcessPoolExecutor(
+        max_workers=1, mp_context=multiprocessing.get_context("spawn")
+    ) as pool:
+        # this needs to be run in a process pool because reaping a process
+        # outside multiprocessing causes it to remain in
+        # multiprocessing.active_children() forever - which blocks cleanup for
+        # 40 seconds
+        assert (
+            pool.submit(
+                _run_and_close_tornado, _check_process_reaped_elsewhere
+            ).result()
+            == 255
+        )
+
+
 @gen_test()
 async def test_child_main_thread():
     """

setup.cfg

@@ -79,8 +79,6 @@ filterwarnings =
     ignore:Scheduler already contains a plugin with name nonidempotentplugin. overwriting:UserWarning
     ignore:Increasing number of chunks by factor of 20:dask.array.core.PerformanceWarning
     ignore::distributed.versions.VersionMismatchWarning
-    ignore:(?s)Exception in thread AsyncProcess Dask Worker process \(from Nanny\) watch process join.*assert exitcode is not None:pytest.PytestUnhandledThreadExceptionWarning
-    ignore:(?s)Exception in thread AsyncProcess SpawnProcess-\d+ watch process join.*assert exitcode is not None:pytest.PytestUnhandledThreadExceptionWarning
     ignore:(?s)Exception in thread.*old_ssh.*channel\.send\(b"\\x03"\).*Socket is closed:pytest.PytestUnhandledThreadExceptionWarning
     ignore:(?s)Exception in thread.*paramiko\.ssh_exception\.NoValidConnectionsError:pytest.PytestUnhandledThreadExceptionWarning
     ignore:(?s)Exception ignored in. <Finalize object, dead>.*sem_unlink.*FileNotFoundError:pytest.PytestUnraisableExceptionWarning