Use local executor again

Otherwise we let a lingering thread start, which makes some tests
unhappy.

There is some chance that the worker might have an executor which
doesn't work well with asyncio.  That seems rare enough and this feature
seems fringe enough that I'm totally willing to take that chance.

distributed/worker.py

@@ -3699,11 +3699,11 @@ def _notify_plugins(self, method_name, *args, **kwargs):
 
     async def benchmark_disk(self) -> dict[str, float]:
         return await self.loop.run_in_executor(
-            None, benchmark_disk, self.local_directory
+            self.executor, benchmark_disk, self.local_directory
         )
 
     async def benchmark_memory(self) -> dict[str, float]:
-        return await self.loop.run_in_executor(None, benchmark_memory)
+        return await self.loop.run_in_executor(self.executor, benchmark_memory)
 
     async def benchmark_network(self, address: str) -> dict[str, float]:
         return await benchmark_network(rpc=self.rpc, address=address)