Start MPI Dynamically from Dask

[mrocklin]

I would like to explore the possibility of Dask starting MPI. This is sort of the reverse behavior of what the dask-mpi package does today.

To clarify the situation I'm talking about, consider the situation where we are running Dask with some other system like Kubernetes, Yarn, or SLURM, and are doing Dask's normal dynamic computations. We scale up and down, load data from disk, do some preprocessing. Then, we want to run some MPI code on the data we have in memory in the Dask worker processes. We don't currently have an MPI world set up (our workers were not started with mpirun or anything) but would like to create one. Is this possible?

To do this, Dask will have to go through whatever process mpirun/mpiexec goes through to set up an MPI communicator. What is this process?

Ideally it would be able to do this without launching new processes. Ideally the same processes currently running the Dask workers would initialize some MPI code, be told about each other, then run some MPI program, then shut down MPI and continue on with normal Dask work.

We'll need to become careful about what to do if a worker goes away during this process. We'll probably have to restart the MPI job, which will be fine. I think that I can handle that on the scheduling/resiliency side.

I think that the people who know the answer to these questions will be people who have experience not only in using MPI, but also in deploying it.

[kmpaul]

I do not know how to initialize an MPI environment without starting a new process. Every MPI implementation is different, and so every mpirun/mpiexec does something different when executed. Its a strange thing that the launching process is not part of the MPI specification, which means there is no standardization.

The closest thing I can think of to launching an MPI job from within python is my personal mpirical package (https://github.com/NCAR/mpirical) which I developed for running MPI tests. However, it still uses subprocess to launch the MPI process with mpirun.

If starting another process is a non-starter (bad pun?), then the only thing I can think of is tailoring a solution to a particular implementation of MPI, such as MPICH or OpenMPI.

[mrocklin]

OK, so lets say we pin ourselves to something like OpenMPI (or whatever is most common). Does this become possible? If so, what is the path to doing this?

[kmpaul]

I think it becomes possible, but I think we would need some OpenMPI or MPICH developers to chime in. Maybe an issue here: https://github.com/open-mpi/ompi?

[jrhemstad]

I don't suggest limiting yourself to only a single MPI implementation.

@kmpaul I can't speak for every MPI implementation, but usually mpirun just does the work of setting up your MPI_COMM_WORLD for you.

It is possible, but ugly, to do this manually. I've done this before in a previous life.

It requires using the MPI_Open_port, MPI_Comm_accept, MPI_Comm_connect, and MPI_Intercomm_merge APIs documented here: https://www.mpi-forum.org/docs/mpi-2.2/mpi22-report/node212.htm#Node212

You end up having to build up your communicator a process at a time.

However, you should note that some MPI implementations flat out do not implement these APIs. I believe OpenMPI and Intel MPI both do, but MVAPICH does not.

MPI_COMM_JOIN is an additional API that allows you to do this that assumes you have an existing socket connection between your two processes. Last I heard the MPI Forum was trying to deprecate this API, so I definitely don't suggest using this one. https://www.mpi-forum.org/docs/mpi-2.0/mpi-20-html/node115.htm

[kmpaul]

@jrhemstad I agree that we shouldn't limit ourselves to a single MPI implementation, but you also point out a problem that finding a solution that works for all implementations may not be possible...easily. (This API is supported in the MPI 3.1 spec, too.)

[jrhemstad]

I was suggesting that one avoids doing anything implementation specific. The APIs I suggest are part of the MPI standard, they're just so infrequently used that some implementations just neglect to implement them.

[spotluri]

MPI libraries use an interface called Process Management Interface (PMI) to interact with the job launcher. Different MPI libraries use different versions of PMI that are incompatible with one another. There has been efforts for standardization and compatibility but AFAIK this is far from complete.

In current state, the job launcher/manager should provide the same PMI version that the MPI library is using. Hence, the tie-in between MPICH MPI library + hydra launcher, OpenMPI library + orte launcher. Slurm provides its own variant of PMI and both MPICH and OpenMPI have configuration option to use PMI interface compatible with SLURM.

It is a non-trivial effort for DASK to replace the MPI launcher. It will have to

1. to provide implementation for the PMI interface use by a particular MPI library
2. provide the backed for exchanging control information that happens through the launcher/deamon processes

And it will have redo 1 for each MPI library it has to be compatible with.

@jrhemstad, dynamic process model (MPI_Comm_spawn, MPI_Comm_accept, MPI_Comm_connect) are for a way for independent MPI applications to launch, discover and connect with each other. The assumption is that you are in an MPI environment already. So it does not obviate the need for the above, especially if you want to use existing MPI libraries.

[jrhemstad]

@jrhemstad, dynamic process model (MPI_Comm_spawn, MPI_Comm_accept, MPI_Comm_connect) are for a way for independent MPI applications to launch, discover and connect with each other. The assumption is that you are in an MPI environment already. So it does not obviate the need for the above, especially if you want to use existing MPI libraries.

This is not correct.

It is certainly possible to launch a process w/o mpirun and then bring it into an MPI communicator---I have done it. You just need to call MPI_Init in the process.

See SO post here: https://stackoverflow.com/questions/15578009/difference-between-running-a-program-with-and-without-mpirun

[spotluri]

This is not correct.

It is certainly possible to launch a process w/o mpirun and then bring it into an MPI communicator---I have done it. You just need to call MPI_Init in the process.

@jrhemstad, my bad. I agree with you. at least according to the specification, one should be able to call these from a singleton process which was not launched through a process manger.

I am skeptical about the current state of implementation for these in MPI libraries though. Like you said, not all libraries support these APIs and even if they do, probably not on all system configs.

Also, this is not the most performant way of initializing MPI. And it can be quite ugly like you point out.

[datametrician]

Given this seems ugly in general, and I hate to say let's create yet another standard, but should we start with OpenMPI and then slowly add other variants as time goes on that follow the same standard?

[jrhemstad]

I am skeptical about the current state of implementation for these in MPI libraries though. Like you said, not all libraries support these APIs and even if they do, probably not on all system configs.

Also, this is not the most performant way of initializing MPI. And it can be quite ugly like you point out.

Absolutely! It's definitely not going to be as easy as just doing mpirun -n N ./a.out and definitely not going to be as performant, but does that performance really matter? It's a one-time, upfront cost.

I think a good place to start is to create a simple client/server C++ code that exercises the MPI_Open_port, MPI_Comm_accept, MPI_Comm_connect, and MPI_Intercomm_merge APIs and it can be tested across various MPIs to get confirmation of which implementations actually support it.

[jrhemstad]

Given this seems ugly in general, and I hate to say let's create yet another standard, but should we start with OpenMPI and then slowly add other variants as time goes on that follow the same standard?

To be clear, the APIs I described are part of the MPI standard that all MPI implementations should implement, it's not specific to OpenMPI.

From what I recall when I've played with this in the past, OpenMPI is one implementation that does support these APIs, so it's a good place to start.

[datametrician]

To be clear, the APIs I described are part of the MPI standard that all MPI implementations should implement, it's not specific to OpenMPI.

From what I recall when I've played with this in the past, OpenMPI is one implementation that does support these APIs, so it's a good place to start.

Understood. So should we start with OpenMPI to see if the standard that's already suppose to be there is universally used :)

[samcmill]

mpi4py.futures may be useful for this: https://mpi4py.readthedocs.io/en/stable/mpi4py.futures.html

[mrocklin]

. Absolutely! It's definitely not going to be as easy as just doing mpirun -n N ./a.out and definitely not going to be as performant, but does that performance really matter? It's a one-time, upfront cost.

What's the upfront cost like? Microseconds, milliseconds, seconds?

I think a good place to start is to create a simple client/server C++ code that exercises the MPI_Open_port, MPI_Comm_accept, MPI_Comm_connect, and MPI_Intercomm_merge APIs and it can be tested across various MPIs to get confirmation of which implementations actually support it.

+1