Client.run on specific workers by name

When running a command on specific workers, I expected I would be able to use worker names to specify which workers to use.

But only worker addresses seem to work.

It'd be nice to make names work, but if not I'd suggest at least cleaning up the docstring so that it's clear what you need in the `workers=` arg. Currently it just says:

```
workers : list
Workers on which to run the function. Defaults to all known
workers.
```

LocalCluster example

In [1]: from dask.distributed import Client, LocalCluster


In [2]: c = LocalCluster()

client

In [3]: client = Client(c)

In [4]: address = list(client.scheduler_info()["workers"])[0]

In [5]: name = client.scheduler_info()["workers"][address]["name"]

In [6]: client.run(lambda: None, workers=[address])

Out[6]: {'tcp://127.0.0.1:61556': None}

In [7]: client.run(lambda: None, workers=[name]) 

distributed.scheduler - ERROR - broadcast to 3 failed: TypeError: expected str, got 'int'

---------------------------------------------------------------------------

TypeError                                 Traceback (most recent call last)

Input In [7], in 

----> 1 client.run(lambda: None, workers=[name])

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/client.py:2752, in Client.run(self, function, workers, wait, nanny, on_error, *args, **kwargs)

   2669 def run(

   2670     self,

   2671     function,

   (...)

   2677     **kwargs,

   2678 ):

   2679     """

   2680     Run a function on all workers outside of task scheduling system

   2681

   (...)

   2750     >>> c.run(print_state, wait=False)  # doctest: +SKIP

   2751     """

-> 2752     return self.sync(

   2753         self._run,

   2754         function,

   2755         *args,

   2756         workers=workers,

   2757         wait=wait,

   2758         nanny=nanny,

   2759         on_error=on_error,

   2760         **kwargs,

   2761     )

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/utils.py:309, in SyncMethodMixin.sync(self, func, asynchronous, callback_timeout, *args, **kwargs)

    307     return future

    308 else:

--> 309     return sync(

    310         self.loop, func, *args, callback_timeout=callback_timeout, **kwargs

    311     )

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/utils.py:376, in sync(loop, func, callback_timeout, *args, **kwargs)

    374 if error:

    375     typ, exc, tb = error

--> 376     raise exc.with_traceback(tb)

    377 else:

    378     return result

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/utils.py:349, in sync..f()

    347         future = asyncio.wait_for(future, callback_timeout)

    348     future = asyncio.ensure_future(future)

--> 349     result = yield future

    350 except Exception:

    351     error = sys.exc_info()

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/tornado/gen.py:762, in Runner.run(self)

    759 exc_info = None

    761 try:

--> 762     value = future.result()

    763 except Exception:

    764     exc_info = sys.exc_info()

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/client.py:2657, in Client._run(self, function, nanny, workers, wait, on_error, *args, **kwargs)

   2654     continue

   2656 if on_error == "raise":

-> 2657     raise exc

   2658 elif on_error == "return":

   2659     results[key] = exc

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/scheduler.py:6061, in send_message()

   6059 async def send_message(addr):

   6060     try:

-> 6061         comm = await self.rpc.connect(addr)

   6062         comm.name = "Scheduler Broadcast"

   6063         try:

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/core.py:1072, in connect()

   1063 fut = asyncio.create_task(

   1064     connect(

   1065         addr,

   (...)

   1069     )

   1070 )

   1071 self._connecting.add(fut)

-> 1072 comm = await fut

   1073 comm.name = "ConnectionPool"

   1074 comm._pool = weakref.ref(self)

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/comm/core.py:268, in connect()

    265     timeout = dask.config.get("distributed.comm.timeouts.connect")

    266 timeout = parse_timedelta(timeout, default="seconds")

--> 268 scheme, loc = parse_address(addr)

    269 backend = registry.get_backend(scheme)

    270 connector = backend.get_connector()

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/comm/addressing.py:21, in parse_address()

     12 """

     13 Split address into its scheme and scheme-dependent location string.

     14

   (...)

     18 If strict is set to true the address must have a scheme.

     19 """

     20 if not isinstance(addr, str):

---> 21     raise TypeError("expected str, got %r" % addr.__class__.__name__)

     22 scheme, sep, loc = addr.rpartition("://")

     23 if strict and not sep:

TypeError: expected str, got 'int'

In [8]:

example with Cluster subclass

In [7]: name = client.scheduler_info()["workers"][address]["name"]

---------------------------------------------------------------------------

NameError                                 Traceback (most recent call last)

Input In [7], in 

----> 1 name = client.scheduler_info()["workers"][address]["name"]


NameError: name 'address' is not defined

In [8]: address = list(client.scheduler_info()["workers"])[0]

In [9]: name = client.scheduler_info()["workers"][address]["name"]

In [10]: client.run(lambda: None, workers=[address])  # works as expected

Out[10]: {'tls://10.0.16.54:43669': None}

In [11]: name

Out[11]: 'david-test-bot-65860152-9-worker-eb87d82d48'

In [12]: client.run(lambda: None, workers=[name])  # doesn't work

---------------------------------------------------------------------------

RuntimeError                              Traceback (most recent call last)

Input In [12], in 

----> 1 client.run(lambda: None, workers=[name])

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/client.py:2752, in Client.run(self, function, workers, wait, nanny, on_error, *args, **kwargs)

   2669 def run(

   2670     self,

   2671     function,

   (...)

   2677     **kwargs,

   2678 ):

   2679     """

   2680     Run a function on all workers outside of task scheduling system

   2681

   (...)

   2750     >>> c.run(print_state, wait=False)  # doctest: +SKIP

   2751     """

-> 2752     return self.sync(

   2753         self._run,

   2754         function,

   2755         *args,

   2756         workers=workers,

   2757         wait=wait,

   2758         nanny=nanny,

   2759         on_error=on_error,

   2760         **kwargs,

   2761     )

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/utils.py:309, in SyncMethodMixin.sync(self, func, asynchronous, callback_timeout, *args, **kwargs)

    307     return future

    308 else:

--> 309     return sync(

    310         self.loop, func, *args, callback_timeout=callback_timeout, **kwargs

    311     )

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/utils.py:376, in sync(loop, func, callback_timeout, *args, **kwargs)

    374 if error:

    375     typ, exc, tb = error

--> 376     raise exc.with_traceback(tb)

    377 else:

    378     return result

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/utils.py:349, in sync..f()

    347         future = asyncio.wait_for(future, callback_timeout)

    348     future = asyncio.ensure_future(future)

--> 349     result = yield future

    350 except Exception:

    351     error = sys.exc_info()

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/tornado/gen.py:762, in Runner.run(self)

    759 exc_info = None

    761 try:

--> 762     value = future.result()

    763 except Exception:

    764     exc_info = sys.exc_info()

File ~/miniconda3/envs/coiled/lib/python3.8/site-packages/distributed/client.py:2657, in Client._run(self, function, nanny, workers, wait, on_error, *args, **kwargs)

   2654     continue

   2656 if on_error == "raise":

-> 2657     raise exc

   2658 elif on_error == "return":

   2659     results[key] = exc

File /opt/conda/envs/coiled/lib/python3.8/site-packages/distributed/scheduler.py:6047, in send_message()

File /opt/conda/envs/coiled/lib/python3.8/site-packages/distributed/core.py:1067, in connect()

File /opt/conda/envs/coiled/lib/python3.8/site-packages/distributed/comm/core.py:289, in connect()

File /opt/conda/envs/coiled/lib/python3.8/asyncio/tasks.py:494, in wait_for()

File /opt/conda/envs/coiled/lib/python3.8/site-packages/distributed/comm/tcp.py:404, in connect()

File /opt/conda/envs/coiled/lib/python3.8/site-packages/distributed/comm/tcp.py:380, in _check_encryption()

RuntimeError: encryption required by Dask configuration, refusing communication from/to 'tcp://david-test-bot-65860152-9-worker-eb87d82d48'