code: def test_workspace_concurrency(tmp_path):
      """WorkSpace concurrency test. We merely check that no exception or
      deadlock happens.
      """
      base_dir = str(tmp_path)
  
      err_q = get_mp_context().Queue()
      purged_q = get_mp_context().Queue()
      stop_evt = get_mp_context().Event()
      ws = WorkSpace(base_dir)
      # Make sure purging only happens in the child processes
      ws._purge_leftovers = lambda: None
  
      # Windows (or at least Windows GitHub CI) has been observed to be exceptionally
      # slow. Don't stress it too much.
      max_procs = 2 if WINDOWS else 16
  
      # Run a bunch of child processes that will try to purge concurrently
      barrier = get_mp_context().Barrier(parties=max_procs + 1)
      processes = [
          get_mp_context().Process(
              target=_workspace_concurrency,
              args=(base_dir, purged_q, err_q, stop_evt, barrier),
          )
          for _ in range(max_procs)
      ]
      for p in processes:
          p.start()
      barrier.wait()
      n_created = 0
      n_purged = 0
      t1 = time()
      try:
          # On Linux, you will typically end with n_created > 10.000
          # On Windows, it can take 60 seconds to create 50 locks!
          while time() - t1 < 10:
              # Add a bunch of locks and simulate forgetting them.
              # The concurrent processes should try to purge them.
              for _ in range(100):
                  d = ws.new_work_dir(prefix="workspace-concurrency-")
                  d._finalizer.detach()
                  n_created += 1
  
      finally:
          stop_evt.set()
          for p in processes:
              p.join()
  
      # Any errors?
      try:
          err = err_q.get_nowait()
      except queue.Empty:
          pass
      else:
          raise err
  
      try:
          while True:
              n_purged += purged_q.get_nowait()
      except queue.Empty:
          pass
  
      # We attempted to purge most directories at some point
      assert n_purged >= 0.5 * n_created > 0
  
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 8 for calling a global function with example usage: test_workspace_concurrency(tmp_path) without return types