TST: Use safer context for ProcessPoolExecutor

When running `test_process_executor_kills_process` just by itself, I get
a warning that threads have been used and that the `fork`
multiprocessing context is unsafe. If I run the full test suite, then
the same test hangs.

I don't know where the threads are coming from (as I didn't use xdist),
but explicitly use the safer `spawn` multiprocess context to avoid these
issues.

distributed/tests/test_worker.py

@@ -5,6 +5,7 @@
 import importlib
 import itertools
 import logging
+import multiprocessing
 import os
 import random
 import sys
@@ -2199,7 +2200,7 @@ async def test_bad_executor_annotation(c, s, a, b):
 
 @gen_cluster(client=True)
 async def test_process_executor(c, s, a, b):
-    with ProcessPoolExecutor() as e:
+    with ProcessPoolExecutor(mp_context=multiprocessing.get_context("spawn")) as e:
         a.executors["processes"] = e
         b.executors["processes"] = e
 
@@ -2231,7 +2232,7 @@ def kill_process():
 
 @gen_cluster(nthreads=[("127.0.0.1", 1)], client=True)
 async def test_process_executor_kills_process(c, s, a):
-    with ProcessPoolExecutor() as e:
+    with ProcessPoolExecutor(mp_context=multiprocessing.get_context("spawn")) as e:
         a.executors["processes"] = e
         with dask.annotate(executor="processes", retries=1):
             future = c.submit(kill_process)
@@ -2254,7 +2255,7 @@ def raise_exc():
 
 @gen_cluster(client=True)
 async def test_process_executor_raise_exception(c, s, a, b):
-    with ProcessPoolExecutor() as e:
+    with ProcessPoolExecutor(mp_context=multiprocessing.get_context("spawn")) as e:
         a.executors["processes"] = e
         b.executors["processes"] = e
         with dask.annotate(executor="processes", retries=1):

distributed/tests/test_worker_metrics.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import multiprocessing
 import time
 from collections.abc import Collection, Hashable
 from concurrent.futures import ProcessPoolExecutor
@@ -141,7 +142,7 @@ async def test_custom_executor(c, s, a):
     """Don't try to acquire in-thread metrics when the executor is a ProcessPoolExecutor
     or a custom, arbitrary executor.
     """
-    with ProcessPoolExecutor(1) as e:
+    with ProcessPoolExecutor(1, mp_context=multiprocessing.get_context("spawn")) as e:
         # Warm up executor - this can take up to 2s in Windows and MacOSX
         e.submit(inc, 1).result()