Support multiple workers per VM

[eric-czech]

It may be more efficient to run many workers on larger VMs as opposed to larger numbers of worker processes on small VMs, presumably to avoid inter-VM communication. This was a suggestion that resulted from this thread on optimizing some dask array workflows: https://github.com/pystatgen/sgkit/issues/390.

@quasiben mentioned that threads per worker can be controlled with worker_options={"nthreads": 2 }, but there appears to be no way to run more than one worker on a single VM.

[quasiben]

I think for this to work we need to change:

dask-cloudprovider/dask_cloudprovider/gcp/instances.py

Lines 318 to 330 in d5dfd99

	"--spec",
	"''%s''" # in yaml double single quotes escape the single quote
	% json.dumps(
	{
	"cls": self.worker_class,
	"opts": {
	**worker_options,
	"name": self.name,
	},
	}
	),
	]
	)

To handle a spec which looks like the following:

workers_options = {
    'worker-1': {"cls": "dask.distributed.Nanny", "opts": {"nthreads": 1}},
    'worker-2': {"cls": "dask.distributed.Nanny", "opts": {"nthreads": 2}},
}

[jacobtomlinson]

When running on a local machine LocalCluster optimises this for you right? On my laptop with 12 cores it runs three workers with four threads each.

Perhaps we can reuse this logic?

[quasiben]

You mean, have dask-cloudprovider inspect the system and assign worker/threads accordingly ?

[jacobtomlinson]

Yeah. We do this for LocalCluster here.

[jacobtomlinson]

dask/distributed#4377 allows us to detect and create worker/threads automatically.

dask-worker --nprocs=auto

Once a release is out we can update things here to use this new option.

[eric-valente]

any progress here?

[jacobtomlinson]

@eric-valente with the latest release of dask and distributed you should be able to set worker_options={"nprocs": "auto"}. Although I haven't managed to come back here and test this yet.

If you could give it a go and report back it would be much appreciated!

[eric-valente]

Hi @jacobtomlinson Tried it with packages and no luck. The workers will turn on in EC2 and won't connect to the scheduler and then proceed to die. When I remove the nprocs it works fine but again only has 1 worker per VM with threads = # of cores

worker_options={'nprocs':'auto'}

dask 2021.4.1 pyhd8ed1ab_0 conda-forge
dask-cloudprovider 2021.3.1 pyhd8ed1ab_0 conda-forge
dask-core 2021.4.1 pyhd8ed1ab_0 conda-forge
distributed 2021.4.1 py38h578d9bd_0 conda-forge

[jacobtomlinson]

Strange! Are you able to get any logs from the EC2 instances? They tend to get dumped in var/log/cloud-init-output.log.

[eric-valente]

@jacobtomlinson

This error then worker dies:

Here are my settings:

Note, this works if i simply remove nprocs from the worker_options. Thanks for your help!

[jacobtomlinson]

Thanks @eric-valente. You also need to set the worker_class kwarg to "dask.distributed.Nanny".

The traceback you got above looks like a bug, I'll raise that back in distributed.

@quasiben what do you think about making dask.distributed.Nanny the default instead of dask.distributed.Worker?

[eric-valente]

Thanks again for your help here @jacobtomlinson

Tried setting the worker class to Nanny but still same issue: