Move back towards Distributed's _retire_workers method

[jakirkham]

Currently our _retire_workers implementation implementation deviates from Distributed's _retire_workers implementation. Would be nice if we could bring them back into correspondence and/or eliminate our copy of _retire_workers entirely in favor of the Adaptive base class implementation.

ref: #63 (review)

[jakirkham]

So just tried to drop this code in PR ( #81 ). After testing it on our cluster, got some pretty nasty looking warnings (one of which is included below). Though all jobs that were suppose to close did.

Traceback (most recent call last):
  File "/opt/conda3/lib/python3.6/site-packages/distributed/utils.py", line 622, in log_errors
    yield
  File "/opt/conda3/lib/python3.6/site-packages/distributed/deploy/adaptive.py", line 237, in _retire_workers
    yield f
  File "/opt/conda3/lib/python3.6/site-packages/tornado/gen.py", line 1099, in run
    value = future.result()
  File "/opt/conda3/lib/python3.6/site-packages/tornado/gen.py", line 1107, in run
    yielded = self.gen.throw(*exc_info)
  File "/groups/dudman/home/kirkhamj/dask-drmaa/dask_drmaa/core.py", line 283, in scale_down
    yield self.scheduler.retire_workers(workers=workers)
  File "/opt/conda3/lib/python3.6/site-packages/tornado/gen.py", line 1099, in run
    value = future.result()
  File "/opt/conda3/lib/python3.6/site-packages/tornado/gen.py", line 315, in wrapper
    yielded = next(result)
  File "/opt/conda3/lib/python3.6/site-packages/distributed/scheduler.py", line 2751, in retire_workers
    workers = {self.workers[w] for w in workers}
  File "/opt/conda3/lib/python3.6/site-packages/distributed/scheduler.py", line 2751, in <setcomp>
    workers = {self.workers[w] for w in workers}
KeyError: 'tcp://XXX.XXX.XXX.XXX:XXXXX

Think this comes down to the fact that in Distributed Adaptive tells the Scheduler to retire_workers and then scale_down the Cluster. This is problematic as we also tell the scheduler to retire_workers in scale_down. So two competing processes are trying to do this work causing this problem. Have raised issue ( dask/distributed#1951 ) to get more context on what Distributed is doing here.

[azjps]

I'm guessing the spirit is that scale_down() should actually be stopping the DRMAA jobs (similar to how remove_workers() is implemented) instead of the current implementation of calling scheduler.retire_workers(). So first Adaptive would communicate to the workers to stop via RPC in scheduler.retire_workers(), and second we would also terminate the jobs directly via DRMAA in scale_down().

Assuming there is no error in either step, these steps I think would be redundant (so we could even have the option to do nothing in scale_down() perhaps), but probably there is no downside in redundancy?

[azjps]

(As a potential reference, KubeCluster.scale_down() does no book-keeping nor any calls to its scheduler, all it does is stop the kubernetes pods for the provided list of workers. Might not be a perfect analogy for dask-drmaa but that's how I formed my mental model.)

[jakirkham]

FWIW this comment suggests your understanding is correct. We could use Session.control for the same purpose.

Edit: Should add we are doing this in stop_workers. So we can port this logic over.

[jakirkham]

Attempting to solve this in PR ( #85 ). Please take a look.