Add worker name as prefix to ThreadPoolExecutor name (#9120)

Author: maneesh29s

distributed/tests/test_worker.py

@@ -2148,6 +2148,31 @@ def get_thread_name():
         assert "Dask-Foo-Threads" in gpu_result
 
 
+@gen_cluster(client=True, nthreads=[])
+async def test_executor_inherit_threadname_from_worker(c, s):
+    def get_thread_name():
+        return threading.current_thread().name
+
+    async with Worker(
+        s.address,
+        nthreads=1,
+        name="WorkerName",
+    ):
+        result = await c.gather(c.submit(get_thread_name, pure=False))
+        assert "WorkerName-Dask-Default-Threads" in result
+
+    async with Worker(
+        s.address,
+        nthreads=1,
+        name="ALongWorkerNameThatNoOneWillProbablyEverAssignButThisTestsTheRobustnessOfLogic",
+    ):
+        result = await c.gather(c.submit(get_thread_name, pure=False))
+        assert (
+            "ALongWorkerNameThatNoOneWillProbablyEverAssignButThisTestsTheRobustnessOfLogic-Dask-Default-Threads"
+            in result
+        )
+
+
 @gen_cluster(client=True)
 async def test_bad_executor_annotation(c, s, a, b):
     with dask.annotate(executor="bad"):

distributed/worker.py

@@ -647,10 +647,15 @@ def __init__(
         if scheduler_sni:
             self.connection_args["server_hostname"] = scheduler_sni
 
+        self.name = name
+
+        executor_pool_prefix = f"{self.name}-" if self.name else ""
         # Common executors always available
         self.executors = {
             "offload": utils._offload_executor,
-            "actor": ThreadPoolExecutor(1, thread_name_prefix="Dask-Actor-Threads"),
+            "actor": ThreadPoolExecutor(
+                1, thread_name_prefix=f"{executor_pool_prefix}Dask-Actor-Threads"
+            ),
         }
 
         # Find the default executor
@@ -660,13 +665,14 @@ def __init__(
             self.executors.update(executor)
         elif executor is not None:
             self.executors["default"] = executor
+
         if "default" not in self.executors:
             self.executors["default"] = ThreadPoolExecutor(
-                nthreads, thread_name_prefix="Dask-Default-Threads"
+                nthreads,
+                thread_name_prefix=f"{executor_pool_prefix}Dask-Default-Threads",
             )
 
         self.batched_stream = BatchedSend(interval="2ms", loop=self.loop)
-        self.name = name
         self.scheduler_delay = 0
         self.stream_comms = {}