unable to set nanny_port programmatically

[MichaelSchreier]

When creating new workers via the .start_worker() method of a cluster instance it is possible to supply the worker_port argument to pin down which port is used by that worker.
However, the nanny_port argument is not supported by this method and only accessible when launching workers via the CLI.

[mrocklin]

You can pass any keyword argument to start_worker that the Nanny expects. I think that port= may be one of them.

[MichaelSchreier]

I see that what you describe was added in #2667 a couple of days. I'll be waiting for the next release then. Thanks for the hard work !

[MichaelSchreier]

I just played around with the latest release and noticed that cluster.start_worker() has been deprecated. The suggested alternative via cluster.scale(), however, appears to lack functionality to assign each worker a different set of properties (worker_port and port in my case).
I tried various methods of setting cluster.new_spec and cluster.worker_spec manually and iteratively scaling the cluster but to no avail.

Am I missing something obvious or has the new way of starting workers made it impossible to set individual worker's properties?

[TomAugspurger]

I suspect you can set this in cluster.new_spec. Not sure if there's an easy API for defining that when creating the LocalCluster. May have to do it after the fact?

In [2]: cluster = LocalCluster(n_workers=1)

In [3]: cluster.new_spec
Out[3]:
{'cls': distributed.nanny.Nanny,
 'options': {'memory_limit': 17179869184,
  'nthreads': 8,
  'services': {},
  'dashboard_address': None,
  'interface': None,
  'protocol': 'tcp://',
  'security': None,
  'silence_logs': 30}}

Can you try that out.

[MichaelSchreier]

After writing a long answer detailing what I've tried and what didn't work I seem to have stumbled on the root cause of the issue and a workaround.
This code doesn't work:

In [2]: cluster = LocalCluster(n_workers=0, threads_per_worker=1, process=True)
In [3]:
for i in range(4):
    spec = cluster.new_spec
    spec['options']['worker_port'] = 9000 + i
    cluster.new_spec = spec
    cluster.scale(i+1)

Here, every other worker appears to try to use the same port as its successor and ends up failing.

Adding a time delay solves the issue but is not particularly elegant:

In [2]: cluster = LocalCluster(n_workers=0, threads_per_worker=1, process=True)
In [3]:
for i in range(4):
    spec = cluster.new_spec
    spec['options']['worker_port'] = 9000 + i
    cluster.new_spec = spec
    cluster.scale(i+1)
    time.sleep(5)

I've also noticed that just by setting cluster.new_spec preexisting entries in cluster.worker_spec are changed to reflect the passed values - no clue why this happens.

spec = cluster.new_spec
spec['options']['worker_port'] = 9000
cluster.new_spec = spec
cluster.scale(1)
cluster.worker_spec
>>{0: {'cls': distributed.nanny.Nanny,
>>  'options': {'nthreads': 1,
>>   'services': {},
>>   'dashboard_address': None,
>>   'interface': None,
>>   'protocol': 'tcp://',
>>   'security': None,
>>   'silence_logs': 30,
>>   'worker_port': 9000}}}

spec = cluster.new_spec
spec['options']['worker_port'] = 9001
cluster.new_spec = spec
cluster.worker_spec
>>{0: {'cls': distributed.nanny.Nanny,
>>  'options': {'nthreads': 1,
>>   'services': {},
>>   'dashboard_address': None,
>>   'interface': None,
>>   'protocol': 'tcp://',
>>   'security': None,
>>   'silence_logs': 30,
>>   'worker_port': 9001}}}

[TomAugspurger]

Would appreciate your thoughts here @mrocklin. I haven't used the new WorkerSpec stuff.

[mrocklin]

One way would be to construct the worker spec explicitly, and then ask the cluster to create that system.

cluster.worker_spec = {i: copy.deepcopy(cluster.new_spec) for i in range(n)}
for k, v in cluster.worker_spec.items():
    v['options']['worker_port'] = ...

await cluster

[MichaelSchreier]

An alternative would be to allow LocalCluster to optionally accept the workers argument and pass it along to SpecCluster instead of the former creating its own dictionary.

[mrocklin]

You could also use SpecCluster on it's own? LocalCluster is designed to handle the common use case and provide sensible API points for most novice users. I think that if you're trying to do something more complex you should probably drop down to SpecCluster. My hope is that the lack of magic and fully explicit nature there is a good match for people who know what they are doing.

…

On Tue, Jul 2, 2019 at 4:44 PM MichaelSchreier ***@***.***> wrote: An alternative would be to allow LocalCluster to optionally accept the workers argument and pass it along to SpecCluster instead of the former creating its own dictionary. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#2685?email_source=notifications&email_token=AACKZTHWMZQCPGC4BHSDLGDP5NZUPA5CNFSM4HMH6BBKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGODZBWNMA#issuecomment-507733680>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AACKZTEUBL6U3RW3PWK2AFDP5NZUPANCNFSM4HMH6BBA> .

[MichaelSchreier]

There still seems to be an issue when trying to set the nanny port:

from dask.distributed import SpecCluster
from dask.distributed import Nanny
from dask.distributed import Scheduler

scheduler = {
    'cls': Scheduler,
    'options': {
        'host': '127.0.0.1',
        'port': 9000,
        'dashboard_address': 9001
    }
}

workers = dict()
for i in range(4):
    workers[i] = {
        'cls': Nanny,
        'options': {
            'worker_port': 9100 + i, # Process port
            # 'port': 9200 + i, # Nanny port
        }
    }
    
cluster = SpecCluster(
    scheduler=scheduler,
    workers=workers
)

This works with the following console output:

distributed.scheduler - INFO - Clear task state
distributed.scheduler - INFO -   Scheduler at:      tcp://127.0.0.1:9000
distributed.scheduler - INFO -   dashboard at:            127.0.0.1:9001
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17910'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17911'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17912'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17913'
distributed.scheduler - INFO - Register tcp://127.0.0.1:9100
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9100
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Register tcp://127.0.0.1:9102
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9102
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Register tcp://127.0.0.1:9101
distributed.scheduler - INFO - Register tcp://127.0.0.1:9103
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9101
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9103
distributed.core - INFO - Starting established connection
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17911'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17913'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17912'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:17910'

When I remove the comment from the line # 'port': 9200 + i, # Nanny port, however, initialization fails with the following log:

distributed.scheduler - INFO - Clear task state
distributed.scheduler - INFO -   Scheduler at:      tcp://127.0.0.1:9000
distributed.scheduler - INFO -   dashboard at:            127.0.0.1:9001
distributed.nanny - INFO -         Start Nanny at: 'tcp://10.50.128.79:9200'
distributed.nanny - INFO -         Start Nanny at: 'tcp://10.50.128.79:9202'
distributed.nanny - INFO -         Start Nanny at: 'tcp://10.50.128.79:9201'
distributed.nanny - INFO -         Start Nanny at: 'tcp://10.50.128.79:9203'
distributed.scheduler - INFO - Register tcp://10.50.128.79:9102
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.50.128.79:9102
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Register tcp://10.50.128.79:9103
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.50.128.79:9103
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Register tcp://10.50.128.79:9101
distributed.scheduler - INFO - Register tcp://10.50.128.79:9100
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.50.128.79:9101
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.50.128.79:9100
distributed.core - INFO - Starting established connection

---------------------------------------------------------------------------
OSError                                   Traceback (most recent call last)
<ipython-input-1-dbfae83ea64f> in <module>()
     24 cluster = SpecCluster(
     25     scheduler=scheduler,
---> 26     workers=workers
     27 )

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\deploy\spec.py in __init__(self, workers, scheduler, worker, asynchronous, loop, silence_logs)
    143             self._loop_runner.start()
    144             self.sync(self._start)
--> 145             self.sync(self._correct_state)
    146             self.sync(self._wait_for_workers)
    147 

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\deploy\cluster.py in sync(self, func, *args, **kwargs)
    245             return future
    246         else:
--> 247             return sync(self.loop, func, *args, **kwargs)

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\utils.py in sync(loop, func, *args, **kwargs)
    329             e.wait(10)
    330     if error[0]:
--> 331         six.reraise(*error[0])
    332     else:
    333         return result[0]

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\six.py in reraise(tp, value, tb)
    691             if value.__traceback__ is not tb:
    692                 raise value.with_traceback(tb)
--> 693             raise value
    694         finally:
    695             value = None

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\utils.py in f()
    314             if timeout is not None:
    315                 future = gen.with_timeout(timedelta(seconds=timeout), future)
--> 316             result[0] = yield future
    317         except Exception as exc:
    318             error[0] = sys.exc_info()

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\tornado\gen.py in run(self)
    727 
    728                     try:
--> 729                         value = future.result()
    730                     except Exception:
    731                         exc_info = sys.exc_info()

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\deploy\spec.py in _correct_state_internal(self)
    198                 for w in workers:
    199                     w._cluster = weakref.ref(self)
--> 200                     await w  # for tornado gen.coroutine support
    201             self.workers.update(dict(zip(to_open, workers)))
    202 

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\tornado\gen.py in wrapper(*args, **kwargs)
    207                 # performance penalty for the synchronous case.
    208                 try:
--> 209                     yielded = next(result)
    210                 except (StopIteration, Return) as e:
    211                     future_set_result_unless_cancelled(

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\nanny.py in _start(self, addr_or_port)
    218             self.listen((self.ip, addr_or_port), listen_args=self.listen_args)
    219         else:
--> 220             self.listen(addr_or_port, listen_args=self.listen_args)
    221             self.ip = get_address_host(self.address)
    222 

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\core.py in listen(self, port_or_addr, listen_args)
    320             connection_args=listen_args,
    321         )
--> 322         self.listener.start()
    323 
    324     @gen.coroutine

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\distributed\comm\tcp.py in start(self)
    419                 # is large enough not to lose any.
    420                 sockets = netutil.bind_sockets(
--> 421                     self.port, address=self.ip, backlog=backlog
    422                 )
    423             except EnvironmentError as e:

C:\Local Applications\WPy-3702\python-3.7.0.amd64\lib\site-packages\tornado\netutil.py in bind_sockets(port, address, family, backlog, flags, reuse_port)
    172 
    173         sock.setblocking(False)
--> 174         sock.bind(sockaddr)
    175         bound_port = sock.getsockname()[1]
    176         sock.listen(backlog)

OSError: [WinError 10048] Only one usage of each socket address (protocol/network address/port) is normally permitted

[mrocklin]

for i in range(4):
    workers[i] = {
        'cls': Nanny,
        'options': {
            'worker_port': 9100 + i, # Process port
            # 'port': 9200 + i, # Nanny port
        }
    }

The options here have to match the keywords in the Nanny constructor. I think that you're looking for port, but I would check that code.

[MichaelSchreier]

Both port and worker_port I supply in the example above are the arguments expected by Nanny to set the ports for the Nanny and the actual worker process, respectively. Am I misunderstanding something here?

[mrocklin]

Ah, fair enough. Perhaps you already have some other service listening on one of those ports running in some other process?

Thank you for the minimal example by the way. For what it's worth, your code works well for me (output below):

Details

from dask.distributed import SpecCluster
from dask.distributed import Nanny
from dask.distributed import Scheduler

scheduler = {
    'cls': Scheduler,
    'options': {
        'host': '127.0.0.1',
        'port': 9000,
        'dashboard_address': 9001
    }
}

workers = dict()
for i in range(4):
    workers[i] = {
        'cls': Nanny,
        'options': {
            'worker_port': 9100 + i, # Process port
            # 'port': 9200 + i, # Nanny port
        }
    }
    
cluster = SpecCluster(
    scheduler=scheduler,
    workers=workers
)

distributed.scheduler - INFO - Clear task state
distributed.scheduler - INFO -   Scheduler at:      tcp://127.0.0.1:9000
distributed.scheduler - INFO -   dashboard at:            127.0.0.1:9001
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63059'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63060'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63061'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63062'
distributed.worker - INFO -       Start worker at:       tcp://127.0.0.1:9100
distributed.worker - INFO -          Listening to:       tcp://127.0.0.1:9100
distributed.worker - INFO - Waiting to connect to:       tcp://127.0.0.1:9000
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -               Threads:                         12
distributed.worker - INFO -                Memory:                   17.18 GB
distributed.worker - INFO -       Local Directory: /Users/mrocklin/dask-worker-space/worker-qpmaqiig
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -       Start worker at:       tcp://127.0.0.1:9101
distributed.worker - INFO -          Listening to:       tcp://127.0.0.1:9101
distributed.worker - INFO - Waiting to connect to:       tcp://127.0.0.1:9000
distributed.worker - INFO -       Start worker at:       tcp://127.0.0.1:9103
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -          Listening to:       tcp://127.0.0.1:9103
distributed.worker - INFO -               Threads:                         12
distributed.worker - INFO - Waiting to connect to:       tcp://127.0.0.1:9000
distributed.worker - INFO -       Start worker at:       tcp://127.0.0.1:9102
distributed.worker - INFO -                Memory:                   17.18 GB
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -       Local Directory: /Users/mrocklin/dask-worker-space/worker-eca2b7qa
distributed.worker - INFO -               Threads:                         12
distributed.worker - INFO -                Memory:                   17.18 GB
distributed.worker - INFO -          Listening to:       tcp://127.0.0.1:9102
distributed.worker - INFO -       Local Directory: /Users/mrocklin/dask-worker-space/worker-fh62t8kp
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO - Waiting to connect to:       tcp://127.0.0.1:9000
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -               Threads:                         12
distributed.worker - INFO -                Memory:                   17.18 GB
distributed.worker - INFO -       Local Directory: /Users/mrocklin/dask-worker-space/worker-hg8onobk
distributed.worker - INFO - -------------------------------------------------
distributed.scheduler - INFO - Register tcp://127.0.0.1:9100
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9100
distributed.core - INFO - Starting established connection
distributed.worker - INFO -         Registered to:       tcp://127.0.0.1:9000
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Register tcp://127.0.0.1:9103
distributed.scheduler - INFO - Register tcp://127.0.0.1:9101
distributed.scheduler - INFO - Register tcp://127.0.0.1:9102
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9103
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9101
distributed.worker - INFO -         Registered to:       tcp://127.0.0.1:9000
distributed.core - INFO - Starting established connection
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -         Registered to:       tcp://127.0.0.1:9000
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Starting worker compute stream, tcp://127.0.0.1:9102
distributed.core - INFO - Starting established connection
distributed.worker - INFO -         Registered to:       tcp://127.0.0.1:9000
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63059'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63060'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63062'
distributed.nanny - INFO -         Start Nanny at: 'tcp://127.0.0.1:63061'

In [2]: cluster
Out[2]: SpecCluster('tcp://127.0.0.1:9000', workers=4)

In [3]: cluster.workers
Out[3]:
{0: <Nanny: tcp://127.0.0.1:9100, threads: 12>,
 1: <Nanny: tcp://127.0.0.1:9101, threads: 12>,
 2: <Nanny: tcp://127.0.0.1:9102, threads: 12>,
 3: <Nanny: tcp://127.0.0.1:9103, threads: 12>}