Inconsistent use of workers parameter between client submit and scatter.
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
It appears that workers parameter between the client submit and scatter methods are not consistent with each other. The submit worker can take ip address, hostnames, (and other other formats, like tcp:// prefix?). The scatter method requires a worker port to be added. The documentation states this needs to be a pair but I found that tcp://127.0.0.1:3454 syntax works as well.
Ideally the same format for defining workers should be the same across the client api.
Here is an example of the failure
```
import dask
from distributed import Client
def doSomething(x):
print(x)
if __name__ == '__main__':
client = Client()
workers = ['127.0.0.1']
dictionary = {1: 2}
client.gather(client.submit(doSomething, dictionary,
pure=False, workers=workers))
future = client.scatter(dictionary, broadcast=True, workers=workers)
client.gather(client.submit(doSomething, dictionary,
pure=False, workers=workers))
```
The client scatter line fails with the error
```
File "d:\mpi\dask-scatter.py", line 18, in
future = client.scatter(dictionary, broadcast=True, workers=workers)
File "c:\users\ddeforest\documents\distributed\distributed\client.py", line 1717, in scatter
asynchronous=asynchronous, hash=hash)
File "c:\users\ddeforest\documents\distributed\distributed\client.py", line 615, in sync
return sync(self.loop, func, *args, **kwargs)
File "c:\users\ddeforest\documents\distributed\distributed\utils.py", line 253, in sync
six.reraise(*error[0])
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\six.py", line 693, in reraise
raise value
File "c:\users\ddeforest\documents\distributed\distributed\utils.py", line 238, in f
result[0] = yield make_coro()
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1099, in run
value = future.result()
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1107, in run
yielded = self.gen.throw(*exc_info)
File "c:\users\ddeforest\documents\distributed\distributed\client.py", line 1518, in _scatter
d = yield self._scatter(keymap(tokey, data), workers, broadcast)
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1099, in run
value = future.result()
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1107, in run
yielded = self.gen.throw(*exc_info)
File "c:\users\ddeforest\documents\distributed\distributed\client.py", line 1585, in _scatter
broadcast=broadcast)
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1099, in run
value = future.result()
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1107, in run
yielded = self.gen.throw(*exc_info)
File "c:\users\ddeforest\documents\distributed\distributed\core.py", line 475, in send_recv_from_rpc
result = yield send_recv(comm=comm, op=key, **kwargs)
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1099, in run
value = future.result()
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1113, in run
yielded = self.gen.send(value)
File "c:\users\ddeforest\documents\distributed\distributed\core.py", line 375, in send_recv
six.reraise(*clean_exception(**response))
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\six.py", line 692, in reraise
raise value.with_traceback(tb)
File "c:\users\ddeforest\documents\distributed\distributed\core.py", line 297, in handle_comm
result = yield result
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 1099, in run
value = future.result()
File "c:\ProgramData\Anaconda3\envs\build\lib\site-packages\tornado\gen.py", line 315, in wrapper
yielded = next(result)
File "c:\users\ddeforest\documents\distributed\distributed\scheduler.py", line 2227, in scatter
workers = [self.coerce_address(w) for w in workers]
File "c:\users\ddeforest\documents\distributed\distributed\scheduler.py", line 2227, in
workers = [self.coerce_address(w) for w in workers]
File "c:\users\ddeforest\documents\distributed\distributed\scheduler.py", line 3987, in coerce_address
addr = resolve_address(addr)
File "c:\users\ddeforest\documents\distributed\distributed\comm\addressing.py", line 170, in resolve_address
return unparse_address(scheme, backend.resolve_address(loc))
File "c:\users\ddeforest\documents\distributed\distributed\comm\tcp.py", line 493, in resolve_address
host, port = parse_host_port(loc)
File "c:\users\ddeforest\documents\distributed\distributed\comm\addressing.py", line 87, in parse_host_port
port = _default()
File "c:\users\ddeforest\documents\distributed\distributed\comm\addressing.py", line 68, in _default
raise ValueError("missing port number in address %r" % (address,))
ValueError: missing port number in address '127.0.0.1'
```
I have tested this with using the dask-worker ```--name mel``` and changing ```workers = ['mel']``` and this works which contradicts the scatter documentaiton
```
workers: list of tuples (optional)
Optionally constrain locations of data. Specify workers as hostname/port pairs, e.g. ('127.0.0.1', 8787).
```
Contributor guide
Assessment
This issue has not been assessed yet.