You shouldn't create Tasks manually, but instead use asyncio.ensure_future.

Did you mean to include the loop in here?

Yes, ideally we want the worker to use the IOLoop used by the cluster object.

I mean that loop is only added if name is not in opts. Wouldn't you always want to pass it?

Ah, indeed. Looking at this again it looks like we do this in an async def function anyway, so IOLoop.current() should be valid regardless. I'll remove the reference to loop entirely, which should be helpful in reducing the contract too.

What is the cluster weakref for?

There are a lot of weakrefs around now. They're useful when tracking down leaking references to things.

The non running workers are never awaited, what happens to them? They're still added to the workers dict below.

This is again a tornado/asyncio difference. I've removed the running check and made things optimal, I think for both async def and gen.coroutine style functions.

Instead of polling, could have a future for the closing operation (created by the first call to _close), and just wait on that?

Good thought. I'm inclined to wait on this for now though if that's ok.

I think this drops scale requests while a current scale request is processing:

• Call scale
• Spec updated
• correct state task start, task is stored as _correct_state_waiting
• scale returns
• Call scale again
• Spec updated
• since previous call is still in progress, state is not corrected, no new workers are started/stopped. Spec and tasks are now out of sync. Also, since there are multiple await calls in _correct_state_internal, the worker_spec can be different at different points in that function, leading to potential bugs.

One naive solution would be to have a background task that loops forever, waiting on an event:

while self.running:
    await self._spec_updated.wait()
    # update workers to match spec
    # After updating, only clear the event if things are up to date
    # If things aren't up to date, then we loop again
    if self.spec_matches_current_state():
        self._spec_updated.clear()

Then _correct_state would look like:

def _correct_state(self):
    # set the event, it's only ever cleared in the loop
    # We force synchronization here to prevent scheduling tons
    # of tasks all setting the event, this blocks until it's set.
    return self.sync(self._mark_state_updated)

async def _mark_state_updated(self):
    self._state_updated.set()

There are likely other ways to handle this. In dask-gateway I have a task per worker/scheduler. As the spec updates, unfinished tasks are cancelled or new ones are fired. If a previous scale call is still in progress for a cluster, scale will block until that call has finished. Note that this only blocks while we update our internal task state (cancelling/firing new tasks), not until those tasks have completed.

since previous call is still in progress, state is not corrected, no new workers are started/stopped. Spec and tasks are now out of sync. Also, since there are multiple await calls in _correct_state_internal, the worker_spec can be different at different points in that function, leading to potential bugs.

So, the _correct_state_waiting attribute isn't the currently running task, it's the currently enqueued one. Once _correct_state starts running it immediately clears this attribute. After someone calls scale there is a clean, not-yet-run _correct_state_waiting future that will run soon.

Since _correct_state_internal waits on the created workers, this does mean that there's no way to cancel pending workers. This is fine for LocalCluster, but would be problematic if used as a base class for other cluster managers. The following would request and start 100 workers before scaling back down afaict:

cluster.scale(100)
cluster.scale(2)

I think that this depends on what you mean by "waits on".

One approach is that for a cluster manager to reach a correct state it only has to successfully submit a request to the resource manager have received an acknowledgement that the resource manager is handling it. We're not guaranteeing full deployment, merely that we've done our part of the job. I would expect this to almost always be fairly fast.

Separately, there is now a Client.wait_for_workers(n=10) method that might be used for full client <-> scheduler checks.

Same here as closing, could wait on the start task instead of polling.

Does this mean that __enter__ will only complete once the initial n workers have started? What happens if we request 2, 1 worker starts and 1 fails?

Yes, this might hang. I'm not sure we ever had a test in our test suite with this case. I'll add something.

Added a test in 5e94069