Add check to scheduler creation in SpecCluster

[jacobtomlinson]

While working on cluster discovery support I found that when reconstructing a cluster scheduler and worker objects may have already been created before start is called.

This small PR adds a check and only creates the scheduler if it doesn't already exist.

This is the first step towards making SpecCluster declarative rather than imperative.

• Tests added / passed
• Passes black distributed / flake8 distributed

[jrbourbeau]

Thanks @jacobtomlinson! In principle this check seems fine, though I'm wondering how we will encounter a case where self.scheduler isn't None in practice. Are you doing something like

cluster = SpecCluster(...)
cluster.scheduler = <an-existing-scheduler>
await cluster.start()

during cluster discovery?

[jacobtomlinson]

Are you doing something like ... during cluster discovery?

Kinda, self.scheduler is being set in KubeCluster.start() if a cluster with the same name already exists. So when it calls super().start() I'm running into problems.

I had already tried setting the status to running which would skip over the contents of SpecCluster.start but then I had trouble because I needed the rpc to be created and for Cluster.start to be called in order to correctly set up status listeners.

[jacobtomlinson]

The _lock also wasn't being created when I was short-circuiting SpecCluster._start by setting the status to running, hence me opening #4596. But this approach seems better.

[jrbourbeau]

Ah, I see. Would you mind adding a comment which mentions this, or a test, to make sure we don't accidentally revert this if self.scheduler is None check?

[jrbourbeau]

Thanks @jacobtomlinson!