code: def test_asyncprocess_child_teardown_on_parent_exit():
      r"""Check that a child process started by AsyncProcess exits if its parent
      exits.
  
      The motivation is to ensure that if an AsyncProcess is created and the
      creator process dies unexpectedly (e.g, via Out-of-memory SIGKILL), the
      child process and resources held by it should not be leaked.
  
      The child should monitor its parent and exit promptly if the parent exits.
  
      [test process] -> [parent using AsyncProcess (dies)] -> [worker process]
                   \                                          /
                    \________ <--   child_pipe   <-- ________/
      """
      # When child_pipe is closed, the children_alive pipe unblocks.
      children_alive, child_pipe = get_mp_context().Pipe(duplex=False)
  
      try:
          parent = get_mp_context().Process(
              target=_parent_process, args=(child_pipe,)
          )
          parent.start()
  
          # Close our reference to child_pipe so that the child has the only one.
          child_pipe.close()
  
          # Wait for the parent to exit. By the time join returns, the child
          # process is orphaned, and should be in the process of exiting by
          # itself.
          parent.join()
  
          # By the time we reach here,the parent has exited. The parent only exits
          # when the child is ready to enter the sleep, so all of the slow things
          # (process startup, etc) should have happened by now, even on a busy
          # system. A short timeout should therefore be appropriate.
          short_timeout = 5.0
          # Poll is used to allow other tests to proceed after this one in case of
          # test failure.
          try:
              readable = children_alive.poll(short_timeout)
          except BrokenPipeError:
              assert WINDOWS, "should only raise on windows"
              # Broken pipe implies closed, which is readable.
              readable = True
  
          # If this assert fires, then something went wrong. Either the child
          # should write into the pipe, or it should exit and the pipe should be
          # closed (which makes it become readable).
          assert readable
  
          try:
              # This won't block due to the above 'assert readable'.
              result = children_alive.recv()
          except EOFError:
              pass  # Test passes.
          except BrokenPipeError:
              assert WINDOWS, "should only raise on windows"
              # Test passes.
          else:
              # Oops, children_alive read something. It should be closed. If
              # something was read, it's a message from the child telling us they
              # are still alive!
              raise RuntimeError(f"unreachable: {result}")
  
      finally:
          # Cleanup.
          children_alive.close()
  
imports: from __future__ import annotations
  import asyncio
  import gc
  import os
  import signal
  import sys
  import threading
  import weakref
  from time import sleep
  import psutil
  import pytest
  from tornado.ioloop import IOLoop
  from distributed.compatibility import LINUX, MACOS, WINDOWS, asyncio_run
  from distributed.config import get_loop_factory
  from distributed.metrics import time
  from distributed.process import AsyncProcess
  from distributed.utils import get_mp_context, wait_for
  from distributed.utils_test import gen_test, nodebug
label_desc: Use image node_id 11 for calling a global function with example usage: test_asyncprocess_child_teardown_on_parent_exit() without return types