Adds the ability to provide a list of worker options to SSHCluster similar to connect_options

[BENR0]

• Closes SSHCluster multiple worker options #4714
• Tests added / passed
• Passes black distributed / flake8 distributed / isort distributed

I ran tests but they did not finish (they break off on distributed/deploy/tests/test_ssh.py::test_defer_to_old. When skipping the test for the feature does not pass but other tests using SSHCluster fail to with a PermissionDenied error for asyncssh. I am not really familiar with testing async routines so I have no solution.

Black passes, flake8 has one SyntaxError which was not introduced by me.

[jrbourbeau]

Thanks @BENR0! Apologies for all the unrelated test failures, those should be resolved soon. Though the distributed/deploy/tests/test_ssh.py::test_list_of_worker_options failure does look relevant.

cc @jacobtomlinson

[BENR0]

It seems that the dicts in the client.scheduler_info()["workers"] values have no nprocs key even though I think nprocs is a valid argument to worker_options. I don't know if this can not be tested then. I will remove the test for nprocs and hopefully the test will pass then.

[jacobtomlinson]

This looks great! Made a couple of docstrings suggestions but this looks ready to go!

[jacobtomlinson]

Thanks @BENR0! Happy to merge on passing tests.

[jrbourbeau]

It seems that the dicts in the client.scheduler_info()["workers"] values have no nprocs key even though I think nprocs is a valid argument to worker_options. I don't know if this can not be tested then. I will remove the test for nprocs and hopefully the test will pass then.

nprocs is a somewhat special option in that it specifies the number of worker processes, while nthreads specifies the size of the threadpool for each worker. So in the test you've added you have

worker_options=[{"nprocs": 4, "nthreads": 1}, {"nprocs": 2, "nthreads": 1}]

for which I would expect there to be 6 total workers, each of which have a nthreads=1. Would you mind adding an additional assert on the total number of workers. This should be something like

assert len(client.scheduler_info()["workers"]) == 6

though I haven't tested the above snippet at all

[jrbourbeau]

Suggested change

	assert all(v["threads"] == 1 for v in d.values())
	assert all(v["nthreads"] == 1 for v in d.values())

[BENR0]

Sorry for the delay. Added the test.

[jrbourbeau]

I added an extra await client.wait_for_workers(6) as a check to make sure the cluster has 6 workers. Interestingly, we get past that line in the test but things fail later when the cluster attempts to update its current state. @jacobtomlinson any thoughts about what might be causing this?

[jacobtomlinson]

I have spent some time today trying to reproduce this locally. I haven't been able to get the test to pass, but for me it gets stuck at the wait for workers call.

The test is failing consistently on Linux (and skipped on mac and windows).

Having the test fail after wait for workers suggests that maybe the idle-timeout is causing the scheduler to close. Perhaps extending that timeout will avoid a race condition?