Non-MPI backends (e.g. Slurm)

[lgarrison]

Since a "major makeover" was mentioned for dask-mpi in dask/distributed#7192, I thought I would ask if support for non-MPI backends was a possibility. That is, use something like Slurm's srun to launch the ranks even if MPI is not available.

The reason I think this may be possible is because as far as I can tell, MPI is only being used to broadcast the address of the scheduler to the workers. Afterwards, it seems to me that any communication is done via direct TCP. If this is correct, could the startup be achieved instead by, e.g., writing the address to a file in a shared location?

The benefit of this is the elimination of a fairly heavyweight MPI dependency. When users start mixing, e.g. a conda install of MPI with a Lmod install of MPI, their jobs often get confused and fail. It seems like dask-mpi has a chance to sidestep these issues completely.

[mrocklin]

Perhaps you are looking for https://jobqueue.dask.org/en/latest/ ?

…

On Fri, Oct 28, 2022 at 12:52 PM Lehman Garrison ***@***.***> wrote: Since a "major makeover" was mentioned for dask-mpi in dask/distributed#7192 <dask/distributed#7192>, I thought I would ask if support for non-MPI backends was a possibility. That is, use something like Slurm's srun to launch the ranks even if MPI is not available. The reason I think this may be possible is because as far as I can tell, MPI is only being used to broadcast the address of the scheduler to the workers. Afterwards, it seems to me that any communication is done via direct TCP. If this is correct, could the startup be achieved instead by, e.g., writing the address to a file in a shared location? The benefit of this is the elimination of a fairly heavyweight MPI dependency. When users start mixing, e.g. a conda install of MPI with a Lmod install of MPI, their jobs often get confused and fail. It seems like dask-mpi has a chance to sidestep these issues completely. — Reply to this email directly, view it on GitHub <#97>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTCLSPD77RTT7YJGZUDWFQHEBANCNFSM6AAAAAARRJF5NE> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[lgarrison]

I'm actually looking for something that can run a dask cluster within an existing Slurm allocation. My understanding is that jobqueue submits Slurm jobs itself; my use-case is that I already have an allocation, and I'd like to start up dask within it.

Taking a step back, I'd like something that can dynamically schedule generic Python tasks within a given allocation of resources. For the most part, I'm interested in completely independent jobs, although task graph capabilities might be useful in the future too. I think dask-mpi is the closest thing I've found so far, but I'm open to suggestions.

[kmpaul]

@lgarrison: Yes. Dask-MPI is just a convenience function. As you point out, MPI is not used for client-scheduler-worker communication. MPI is only used to communicate the scheduler address to the worker processes (and the client process, if used with the initialize function). Earlier implementations of Dask-MPI actually required a shared filesystem because the scheduler shared its address with the other processes through use of the --scheduler-file option. But we ran into problems with this because some users had MPI-enabled clusters without shared filesystems. It seems to me that we need to support both of those use cases going forward (and it currently does).

It seems to me that the feature you are looking for would only require that Dask-MPI not require mpi4py. And if mpi4py was not found, then to restrict the input parameters to make it possible to run with non-MPI-communication mechanisms (i.e., require --scheduler-file or similar).

Does that accurately describe what you are looking for?

[lgarrison]

It seems to me that the feature you are looking for would only require that Dask-MPI not require mpi4py

Yes, I think that's exactly right. I'll test out the --scheduler-file mechanism to confirm that this is the case. Thanks!

[jacobtomlinson]

I attempted a pull out a base class from dask-mpi a while ago to make it easier to build alternative implementations with different backends.

distributed #4710
dask-mpi #69

The core difference is that folks want to take an existing allocation and fill it with a Dask cluster like dask-mpi does, rather than create the whole thing from scratch like dask-jobqueue does.

However, that draft implementation met some friction from other maintainers and I didn't pursue it any further. But maybe this is a signal that there is demand for this and the distinction is valuable?

[lgarrison]

I did realize that one other piece is needed to get this to work, besides the scheduler-file option, which is a way for processes to identify their rank. In a Slurm context, this means that processes need to read the SLURM_PROCID environment variable. Easy enough, but there will need to be an abstraction where different bootstrap contexts can define how they determine the rank.

@jacobtomlinson, it looks like you've already worked on related abstractions. Are we on the right track here? Is it okay to implement "Slurm-awareness" at the dask-mpi level, or is this something that belongs higher up in distributed?

[jacobtomlinson]

This sounds like a good track but I don't think it belongs in dask-mpi given that it isn't MPI related. It also isn't best for it to go in distributed. Maybe dask-jobqueue is the right location given that it is job queue specific functionality (cc @guillaumeeb)? If not then a separate package.

This is why I tried to abstract things out in distributed so that other projects could reuse things.

[lgarrison]

I agree it's a bit odd to have a package named dask-mpi support non-MPI bootstrapping, but the amount of Slurm stuff that would have to go in dask-mpi is so minimal that it seems overkill to loop in another package. The entire abstraction probably looks like:

class SlurmBootstrapper(Bootstrapper):
    def get_rank(self, comm=None):
        return os.environ['SLURM_PROCID']
    def supports_bcast(self):
        '''Do we have MPI broadcast support, or do we have to use scheduler-file?'''
        return False

class MPIBootstrapper(Bootstrapper):
    def get_rank(self, comm=None):
        return comm.get_rank()
    def supports_bcast(self):
        return True
    def bcast(self, comm, *args, **kwargs):
        comm.bcast(*args, **kwargs)

This could also be an interesting use case for entry points, which would allow an external package to register a bootstrapping interface. But it still seems like a shame to split off another package since the packaging overhead would be so much larger than the useful code itself.

Plus, an important part of this (for my use case, at least) is to have the mpi4py dependency be optional, since it's not such a lightweight library and other mechanisms could do its job. I realize this needs more discussion, but if dask-mpi is willing to forgo a mandatory MPI dependency (as I think @kmpaul was suggesting), then natively supporting common job runners might not be so out-of-place.

[jacobtomlinson]

I agree it's a shame for such a small amount of code to cause this discussion, but I'm really not sure this proposal makes sense.

Given that this whole library is like ~200 lines of code I don't think there would be much pushback to making a PR into dask-jobqueue that implements a slurm runner and copy/pastes a chunk from here (although ideally that common chunk would be upstreamed into distributed). Both dask-jobqueue and distributed already exist and seem like the obvious places to put this logic, so there is no packaging overhead.

if dask-mpi is willing to forgo a mandatory MPI dependency

This seems like a pretty unusual ask. Making MPI optional in the dask-mpi package so that non-MPI things can be put here doesn't make sense to me.

[lgarrison]

dask-jobqueue seems to have minimal dependencies, so on that front I'd be happy enough if the dask-mpi framework were copy-pasted over there. It just seemed orthogonal to what dask-jobqueue wants to do, which is launch and manage its own Slurm jobs.

Perhaps more importantly, I can see many users upon being directed to the dask-jobqueue package getting confused and using the job-submission functionality, rather than the dask-mpi-like functionality. The pattern I'm trying to encourage is to request a single multi-node allocation, then do dynamic dispatch inside that. I don't think this is so crazy; many HPC schedulers won't even consider more than a handful of queued jobs at a time for scheduling (and have a jobs-per-user max), and dask-jobqueue only submits single-node jobs, so multi-node allocations can be far more efficient for both the user and the scheduler. Indeed, this has been discussed in dask-jobqueue, and it seems they are pretty much set on single-node submissions: dask/dask-jobqueue#364 (comment). In that same comment, they point users to dask-mpi for this multi-node allocation case.

This seems like a pretty unusual ask. Making MPI optional in the dask-mpi package so that non-MPI things can be put here doesn't make sense to me.

I know this seems unusual, but I think it's more because dask-mpi happens to have "MPI" in the name rather than anything fundamental about the package. And maybe that's a good enough reason---I agree it would cause some confusion. Hopefully nothing some informative error messages couldn't smooth over, though.

But if we are at the point of copy-pasting dask-mpi, then maybe it makes the most sense for me to do that in my own repo, rather than try to shoehorn it into dask-jobqueue. Happy to hear what others think.

[guillaumeeb]

Hi everyone,

Yes, things like that have been discussed in the past. In the HPC world, between dask-jobqueue and dask-mpi functionalities, there are multi nodes allocations (subject of this discussion), and job arrays possibilities. There are several other threads in dask-jobqueue: dask/dask-jobqueue#459, dask/dask-jobqueue#196, dask/dask-jobqueue#84 (comment). In dask-jobqueue, we've reached to the conclusion that it would introduce to much complexity and specific code for each job scheduling system.

In any case, I think it would be really nice to have such Dask cluster deployment method, a lot of HPC system admin (and users) would be glad of this. The thing is, job schedulers are so specific in these matters that it seems complicated to me to find the correct abstraction and place for a new dask-hpc-or-multinode package. Slurm is really powerful in this multi nodes domain, it would almost make sense to have a dask-slurm package, that uses allocations and srun commands!

So in my opinion, you should start in a new repo, I would be happy to follow your development and see if things can be generalized in one way or another! Maybe in the end dask-mpi would become a specific implementation of a dask-multinode package?

[jacobtomlinson]

That all sounds reasonable. I think the key difference between dask-mpi and other deployment packages is that it runs inside an existing multi-node allocation and just fills it with a Dask cluster. I think this deployment paradigm is underserved currently. I'm going to continue referring to this paradigm as a "dask cluster runner" as opposed to the dask-jobqueue paradigm which we typically call a "dask cluster manager".

I have concerns about orthogonality in all the deployment packages we have when we talk about introducing more runners.

The SLURM runner you are proposing is similar to dask_jobqueue.SlurmCluster but different enough that the suggestion of putting it in a new package seems sensible. But I want to push back on that some more.

I also want to have a runner for various cloud environments, I think Azure ML is the example I used in my original PR about this. We already have the cluster manager dask_cloudprovider.azure.AzureMLCluster so this would be similar to that but the user would submit a multi-node job rather than the cluster manager. Is that different enough that maybe it shouldn't live in dask_cloudprovider?

In that case, does it make sense for a SLURM runner and Azure ML runner to live in one new package called dask-multinode? My feeling is no. The Azure ML runner will probably benefit more from being in dask_cloudprovider as it would allow us to share cloud-specific code and utilities.

I assume from @guillaumeeb's response that there may not be as much benefit of colocating the SLURM cluster manager and cluster runner in the same package. Maybe code reuse between them would be low? But I think there are other compelling cases for cloud, kubernetes, etc where this colocation does have benefits, especially around package dependencies.

I don't think putting this in dask-mpi makes sense, and I don't think lumping all the runners in a new package called dask-multinode or similar makes sense.

We also don't want to create a new package for every new runner, that would be a pain.

That leaves colocating them with existing cluster managers that use the same resource schedulers and dependencies, even if that is a shift for the package's remit currently.

[guillaumeeb]

I think the key difference between dask-mpi and other deployment packages is that it runs inside an existing multi-node allocation and just fills it with a Dask cluster

I assume from @guillaumeeb's response that there may not be as much benefit of colocating the SLURM cluster manager and cluster runner in the same package. Maybe code reuse between them would be low?

I didn't thought about all that, given that I've not looked a lot into dask-mpi internals, or even use it much. The most part of dask-jobqueue code is about generating and submitting job scripts. In dask-mpi, and this proposal, the user is responsible for writing the job script, and he just want to use some wrapper code in it to start a Dask cluster. Indeed, that is a big difference.

This is also something that dask-mpi is lacking: being able to call cluster = MPICluster(...) inside a notebook would be a really nice feature, but this is another issue.

Back to the subject here, the proposal is really much closer to dask-mpi paradigm. Maybe we should have a dask-hpc package that host dask-mpi and other runners?

[jacobtomlinson]

I agree the deployment paradigm is a big difference, but I think that's ok. If you're worried about maintaining a SLURM runner in dask-jobqueue falling on your shoulders all I can do if offer to help maintain it.

Maybe we should have a dask-hpc package that host dask-mpi and other runners?

This is tempting, we could just rename this repo and then put the SLURM runner in here too. Packaging would get more complex because MPI would need to be made optional.

However I worry that users would get confused between dask-jobqueue and dask-hpc and what the differences are.

Also what about cloud/kubernetes/hadoop/etc? If we want runners for those it doesn't make sense for those to go in a dask-hpc package.

[guillaumeeb]

If you're worried about maintaining a SLURM runner in dask-jobqueue falling on your shoulders all I can do if offer to help maintain it.

Not at all! You've seen over the years that I'm not always available to maintain things, so I won't feel the responsibility for this 😄.

I just think I'm not seeing yet what really is a runner, or at least this one, and how it would integrate into dask-jobqueue.

Do you think we could share part of the runner code and use it in SlurmCluster internals at some point? This would make a lot of sense then. If not, I'm curious to see how you would organize things.

Maybe the development should still be started in another repository so that we can see how it can fit later on in a existing repo? Maybe you're afraid in that case that we never merge it in an existing package...