call exit callback even if AsyncProcess is reaped elsewhere

[graingert]

Might help with #6683 ?

• Tests added / passed
• Passes pre-commit run --all-files

[github-actions]

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

       15 files  ±    0         15 suites  ±0   6h 53m 32s ⏱️ + 21m 33s
  2 989 tests +  58    2 894 ✔️ +  49       89 💤 +  4  6 ❌ +5 
22 165 runs  +454  21 113 ✔️ +394  1 046 💤 +55  6 ❌ +5 

For more details on these failures, see this check.

Results for commit 9dbf154. ± Comparison against base commit ade4266.

♻️ This comment has been updated with latest results.

[graingert]

inspired by:

        try:
            _, status = os.waitpid(pid, 0)
        except ChildProcessError:
            # The child process is already reaped
            # (may happen if waitpid() is called elsewhere).
            returncode = 255
            logger.warning(
                "child process pid %d exit status already read: "
                " will report returncode 255",
                pid)

https://github.com/python/cpython/blob/c5819c1f6c67f04e08f058ac7f24dec86e0e4554/Lib/asyncio/unix_events.py#L944-L952

[graingert]

this breaks because async functions are not supported. set_exit_callback calls synchronous callbacks on the event loop thread, and this is intentional: https://github.com/dask/distributed/pull/6526/files

[graingert]

python/cpython#94661

[graingert]

making a test for this that's not flaky is a bit horrible, because the test call to os.waitpid races with the AsyncProcess._watch_process call to process.join(None) which also calls os.waitpid

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():
    import time

    loop = asyncio.get_running_loop()
    a, b = socket.socketpair()
    event = asyncio.Event()
    lock = threading.Lock()
    lock.acquire()

    real_waitpid = os.waitpid

    def _fake_waitpid(*args, **kwargs):
        with lock:
            return real_waitpid(*args, **kwargs)


    os.waitpid = _fake_waitpid

    def wait_pid_in_thread(pid):
        t_start = time.monotonic()
        try:
            real_waitpid(pid, 0)
        finally:
            lock.release()
            print(f"wote pid {t_start=} {time.monotonic()=}")
            loop.call_soon_threadsafe(event.set)

    with a:
        with b:
            proc = AsyncProcess(
                target=_write_byte_wait_closed, args=(b,), loop=IOLoop.current()
            )
            await proc.start()

        t = threading.Thread(target=wait_pid_in_thread, args=(proc.pid, ))
        t.start()
        a.setblocking(False)
        assert await loop.sock_recv(a, 1) == b"\x00"
        print(f"sock recv {time.monotonic()=}")
        await loop.sock_sendall(a, b"\x00")
        print(f"sock sendall {time.monotonic()=}")
        await proc.join()
        print(f"await proc.join() {time.monotonic()=}")
        await event.wait()
        print(f"event wait {time.monotonic()=}")
        t.join()
        print(f"t join {time.monotonic()=}")
        return proc.exitcode

so I think it's better to leave it untested?

[gjoseph92]

Yeah that test patching os.waitpid is a bit mad. Since we won't be filtering the warning anymore, seems ok to leave untested.

[gjoseph92]

• CI failing with test_get_task_stream_save #6312
• distributed/tests/test_nanny.py::test_environ_plugin is failing, which is not a failure I see on the CI dashboard. I want to make sure this isn't related to this PR before merging.

[fjetter]

what's the status on this @gjoseph92 @graingert ?

[graingert]

I think give this another CI run and see, then land it as is?

[gjoseph92]

• Flaky test_quiet_client_close #6540
• Flaky distributed/diagnostics/tests/test_task_stream.py::test_client_sync #6820
• Flaky tests: ResourceWarning: unclosed cluster SSHCluster #6771
• Flaky tests: OSError: Timed out trying to connect to tcp://127.0.0.1:8786 after 5 s #6731

[graingert]

@gjoseph92 thanks for going through the CI failures for me here! I think this is ready to merge