code: def test_workspace_process_crash(tmp_path):
      # WorkSpace should be able to clean up stale contents left by
      # crashed process
      base_dir = str(tmp_path)
      assert_contents = functools.partial(
          assert_directory_contents, base_dir
      )
  
      ws = WorkSpace(base_dir)
  
      code = """if 1:
          import signal
          import sys
          import time
  
          from distributed.diskutils import WorkSpace
  
          ws = WorkSpace(%(base_dir)r)
          a = ws.new_work_dir(name='aa')
          b = ws.new_work_dir(prefix='foo-')
          print((a.dir_path, b.dir_path))
          sys.stdout.flush()
  
          time.sleep(100)
          """ % dict(
          base_dir=base_dir
      )
  
      p = subprocess.Popen(
          [sys.executable, "-c", code],
          stdin=subprocess.PIPE,
          stdout=subprocess.PIPE,
          universal_newlines=True,
      )
      line = p.stdout.readline()
      assert p.poll() is None
      a_path, b_path = eval(line)
      assert_contents(
          [a_path, a_path + ".dirlock", b_path, b_path + ".dirlock"]
      )
  
      # The child process holds a lock so the work dirs shouldn't be removed
      ws._purge_leftovers()
      assert_contents(
          [a_path, a_path + ".dirlock", b_path, b_path + ".dirlock"]
      )
  
      # Kill the process so it's unable to clear the work dirs itself
      p.kill()
      assert p.wait()  # process returned with non-zero code
      assert_contents(
          [a_path, a_path + ".dirlock", b_path, b_path + ".dirlock"]
      )
  
      with captured_logger(
          "distributed.diskutils", "INFO", propagate=False
      ) as sio:
          ws._purge_leftovers()
      assert_contents([])
      # One log line per purged directory
      lines = sio.getvalue().splitlines()
      assert len(lines) == 2
      for p in (a_path, b_path):
          assert any(repr(p) in line for line in lines)
  
imports: from __future__ import annotations
  import functools
  import gc
  import os
  import queue
  import shutil
  import subprocess
  import sys
  from time import sleep
  from unittest import mock
  import pytest
  import dask
  from distributed import profile
  from distributed.compatibility import WINDOWS
  from distributed.diskutils import WorkSpace
  from distributed.metrics import time
  from distributed.utils import get_mp_context
  from distributed.utils_test import captured_logger
label_desc: Use image node_id 4 for calling a global function with example usage: test_workspace_process_crash(tmp_path) without return types