Run Scheduler and Client from the same MPI Rank

[guillaumeeb]

I believe that in some simple cases, we don't need to have a rank dedicated to the Scheduler, and a rank dedicated to the Client. We should provide a way to start both within the same MPI rank in https://github.com/dask/dask-mpi/blob/master/dask_mpi/core.py#L62-L71.

Thoughts?

[mrocklin]

Seems fine to me.

…

On Mon, Mar 11, 2019 at 1:25 PM Guillaume Eynard-Bontemps < ***@***.***> wrote: I believe than in some simple case, we don't need to have a rank dedicated to the Scheduler, and a rank dedicated to the Client. We should provide a way to start both within the same MPI rank in https://github.com/dask/dask-mpi/blob/master/dask_mpi/core.py#L62-L71. Thoughts? — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#29>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszK-hXbWnE093kNxNhEOH1LLXd9Roks5vVru5gaJpZM4bpfR9> .

[kmpaul]

@guillaumeeb Sorry for the delay!

I think this is entirely true. And it's an excellent idea! In fact, I would like to go a little further than this. I would like to make the launch algorithm work for any size MPI communicator, not just > 2!
I am considering an optional parameter like separate_scheduler_rank to place the scheduler on a rank different from the client rank.

Here's what I'm considering

1. If the MPI communicator size is 1, then place the scheduler and a single worker on the same rank as the Client (rank 0). This is the "LocalCluster" option. The separate_scheduler_rank option is ignored.
2. If the MPI communicator size is 2, then place the scheduler on the same rank as the client (rank 0) and start a worker on the other rank (rank 1). Again, the separate_scheduler_rank option is ignored.
3. If the MPI communicator size is 3 or more, then place the client on rank 0, the scheduler on rank 0 or rank 1 (depending on separate_scheduler_rank), and the workers on all of the other ranks.

Is this reasonable?

[guillaumeeb]

If you think about use case where the MPI rank would be less than 3, then yes. I assumed that there was not really such a use case... But maybe for algorithm testing or debug purpose?

Otherwise I think the idea of an optional parameter is perfectly fine!

[betatim]

I'd like to reboot this issue :D

Is there a reason to not place a worker on each MPI rank? For example if the MPI size is 1: place worker, scheduler and client on rank 0. If the MPI size is >=2: scheduler on rank 0, client on rank 1, workers on all ranks.

If that isn't a fundamentally bad idea I'd propose adding a flag (maybe called workers_on_all_ranks=True?) that leads to this new behaviour. With a default of False so that nothing changes for current users.

[kmpaul]

Thanks for rebooting this, @betatim!

I have some cycles to devote to this today, so the timing is great!

I've been thinking about how to give the user more explicit control about placement for client, scheduler and workers within the MPI cluster. To aid with this, I've been thinking about adding the CLI options:

• scheduler_rank: The MPI rank of the scheduler. The default value is 0.
• client_rank: The MPI rank of the client, if being used in a stand-alone MPI script. If the MPI cluster size is 1, then the client rank will be 0. If the MPI cluster size is 2 or more, then the default value is 1.

They could be the same, and they could be different from 0/1, but they have sensible defaults (chosen for backwards compatibility).

I think the client and scheduler are the easy bits, though, because there are only ever 1 of each of them.

The workers are the harder bits. I think the default behavior (again, backward compatibility) is for workers to "fill" the remaining MPI cluster, one worker per MPI rank. But it's the idea of more than one client/scheduler/worker per MPI process that we are discussing, I think. And if we allow the client and scheduler to run on the same MPI rank, then why not multiple workers?

In terms of implementation, most MPI implementations that I am aware of will not allow multiple processes to be associated with the same MPI rank (and things like forking/spawning new processes from an MPI process can actually be blocked). I do not know if some MPI implementations will allow it, or if it is safe even if possible, but regardless, the general approach to MPI is that each rank should (ideally) be a single process. That means that to get multiple client/scheduler/workers to run on a single MPI process will require running them asynchronously in a single process, which is very doable from a "I know how to code that up" perspective... but I don't know if it will be a stable solution.

Does anyone listening in on this conversation have thoughts on this? In this time zone, maybe @guillaumeeb @jacobtomlinson?

Of course, I don't have a problem with implementing something that gives the user the ability to create an unstable instantiation of a Dask-MPI cluster if it also gives the user more flexibility. However, I don't want to create general instability for other users who aren't aware of the complexity and subtlety.

[kmpaul]

In terms of a user CLI to allow complete control over worker placement, the complete solution would involve asking the user to specify a list of one int per MPI rank, with the value of the int indicating the number of workers to launch for the corresponding rank.

That is, something like:

workers_per_rank = [0,0,1,1,1,2]

would describe launching 0 workers on ranks 0 and 1, 1 worker on ranks 2, 3, and 4, and 2 workers on rank 5. ...But for large clusters, this is untenable.

One possible solution would be to assume that the last int in the list is "repeating" for all additional ranks to the size of the MPI cluster. For example, one might specify:

workers_per_rank = [0, 0, 1]

which would imply not launching a worker on ranks 0 and 1 and only launching a worker on ranks 2 or higher. That's the default behavior right now.

[kmpaul]

So, currently Dask-MPI in the stand-alone script mode only works if you have an MPI comm size of 3 or larger. This proposed change would make it possible to run Dask-MPI with MPI comm sizes 1 or 2, for which we would need to define sensible defaults. Here are my suggestions:

MPI comm size 1:

There is only 1 possible choice here, so it doesn't really need to be said. However, just to be pedantic: scheduler on rank 0, client on rank 0, 1 worker on rank 0.

MPI comm size 2:

I think the default arrangement should be client on rank 0, scheduler on rank 0, and worker on rank 1. But I'm open for comment here.

MPI comm sizes 3+:

The normal defaults should exist with scheduler on rank 0, client on rank 1, and workers on all other ranks.

[betatim]

My desire to have a worker on the ranks 0 and 1 is that I don't know enough about the scheduler/job submission system I am using to request different types of nodes per rank. So I end up with a gigantic machine with eight GPUs for each rank. Given that I have to queue a while to get the allocation I'd like to use all of it :D Of course an alternative solution would be to figure out if it is possible to receive a smaller node for rank 0 and 1 (a bit more like you'd do in the cloud).

This means for me it is all about "workers on all the ranks". I care less/know nothing about how the scheduler and client are distributed amongst the available ranks.

I also don't use the CLI but call initialize() within my code. But I assume if a CLI option is added there will also be a way to do the same thing via initialize.

I hadn't considered the restrictions on starting new processes and such (can you tell I am a MPI newbie?). I had naively assumed that we'd just add more stuff to the event loop and that would be that. I am +1 on making it hard for people to configure things in a way that makes running unstable. My assumption is that most people "have no clue", so like me they'd just see a wasted node and see that they can configure it to do work and do it. And then have a miserable life because sometimes things work and sometimes they don't. I think it would be better to not give people that rope to hang themselves by and write a short bit of documentation explaining why you can't configure a particular setup.

For "why not more than one worker": I learnt that the word "worker" is overloaded. For example when I instantiate one CUDAWorker I will end up with as many workers (in the sense of entries in the scheduler information) as there are GPUs in the system. From that point of view I think we don't need more than one worker (aka CUDAWorker instance) per rank. From what you said above about "more than one process per rank is often not allowed" it also sounds like in the CPU setting you'd only want one worker per rank?

[kmpaul]

My desire to have a worker on the ranks 0 and 1 is that I don't know enough about the scheduler/job submission system I am using to request different types of nodes per rank. So I end up with a gigantic machine with eight GPUs for each rank. Given that I have to queue a while to get the allocation I'd like to use all of it :D Of course an alternative solution would be to figure out if it is possible to receive a smaller node for rank 0 and 1 (a bit more like you'd do in the cloud).

Understood! Then I would lean towards making the "run workers everywhere" option easy to implement, at least.

I also don't use the CLI but call initialize() within my code. But I assume if a CLI option is added there will also be a way to do the same thing via initialize.

Yes. I think the cli option is actually the easy option. If we find a solution that works for the initialize() workflow, then the CLI implications will be natural.

I hadn't considered the restrictions on starting new processes and such (can you tell I am a MPI newbie?). I had naively assumed that we'd just add more stuff to the event loop and that would be that. I am +1 on making it hard for people to configure things in a way that makes running unstable. My assumption is that most people "have no clue", so like me they'd just see a wasted node and see that they can configure it to do work and do it. And then have a miserable life because sometimes things work and sometimes they don't. I think it would be better to not give people that rope to hang themselves by and write a short bit of documentation explaining why you can't configure a particular setup.

Yes. I think that being explicit here would benefit everyone. Then, at least, people will have a better idea of what to expect before hand.

That said, MPI is always confusing for many people. Dask-MPI was created out of a simple abstraction that attempted to make launching a dask cluster in an MPI environment easy, but I think that just opened the trail to a steep climb for many people. Once you're in the door, MPI can be a labyrinth! (And that's not even going into the complications created by each implementation of MPI being a little different from each other...and from the MPI standard itself!)

[kmpaul]

For "why not more than one worker": I learnt that the word "worker" is overloaded. For example when I instantiate one CUDAWorker I will end up with as many workers (in the sense of entries in the scheduler information) as there are GPUs in the system. From that point of view I think we don't need more than one worker (aka CUDAWorker instance) per rank. From what you said above about "more than one process per rank is often not allowed" it also sounds like in the CPU setting you'd only want one worker per rank?

Yes. You've hit on exactly the issue, I think. That is why the initial implementations of Dask-MPI have always worked in "units of 1". And I think (because of this overloaded "worker" word) it is safe to continue assuming only 1 worker per rank.

If that were the assumption, then I would modify my CLI/parameter suggestion above to replace the workers_per_rank option with a workers_skip_ranks option like so:

workers_skip_ranks = [0, 1, 5]

which would imply that workers would not be launched on ranks 0, 1 and 5.

Changed my mind. I think it actually makes sense to just adopt your notion of whether to allow a worker on the same rank as the client/scheduler. This is the notion of worker exclusivity, or whether a worker has exclusive access to an MPI rank. Would it be sufficient to just have a boolean exclusive_workers option? If exclusive_workers=False then workers would be launched on the same ranks as the client and scheduler. If exclusive_ranks=True, then the default behavior of "no workers on client and scheduler ranks` would apply.

[jacobtomlinson]

I think this makes a load of sense. Especially on large nodes.

Thinking purely from a GPU perspective there are a bunch of different configurations I would love to have (assuming the rank 0 node has 8 GPUs):

• Give the Client and Scheduler their own GPU, populate the other 6 with workers
• Give the Client and scheduler a shared GPU, populate the others with 7 workers
• Start 8 workers, have the client share with a worker and the scheduler share with a different worker

I'm also playing around with https://github.com/jacobtomlinson/dask-hpc-runner which is based on dask/distributed#4710 and #69. My main goal here is to make the dask-mpi deployment paradigm (populate an existing allocation with a Dask cluster) easier to do in non-MPI scenarios.

My motivating use case is SLURM where we have env vars to get the ranks and a shared filesystem to communicate the scheduler address. It would be nice to leave MPI out of things altogether because there can be consequences to importing mpi4py especially in GPU workloads.

There are other places like Databricks where it would also be useful to have this paradigm for side-loading a Dask cluster onto a Spark cluster.

[kmpaul]

@jacobtomlinson: Oooh! Excited to hear about progress on the dask-hpc-runner. Been interested in that for a while. And I agree that the use cases are compelling.

Thanks for chiming in on these suggestions. It sounds like the 3 bulleted configurations you described would be satisfied by the options I suggested above. I'm working on a PR now.

[betatim]

Good point about assigning one of the GPUs (each) to the scheduler and client.

[kmpaul]

Ok. I've been playing around with this today and there are solutions, but I don't think they are as pretty as anyone was hoping. So, I'm sharing my experiences in the hopes that someone might have better ideas than I have.

First, it's easy to launch a worker on the same MPI rank as the scheduler. The scheduler and worker are both asynchronous tasks, and therefore they can both be run on the same asynchronous event loop, which is started with the asyncio.run_until_complete() function.

However, the complication comes from the fact that the client code is synchronous. And launching an asynchronous task inside of synchronous code can only be done by starting an event loop with either asyncio.run_until_complete() or asyncio.run_forever(), both of which are blocking until the asynchronous tasks are canceled or completed. If we were to try to launch the scheduler in the same process as the client, the asyncio.run_until_complete() function would block the execution of the client code indefinitely because the shutdown signal would never be sent (because it is sent by the client code which is blocked).

The only way to resolve this issue is for the event loop that runs in the client process to either: (1) start an event loop in a separate thread or (2) start an event loop in a separate process (i.e., fork or spawn). For option 2, this violates the 1 process per MPI rank rule that I mentioned above, so I think we should exclude it from consideration. For option 1, however, this means starting a thread pool in the client code, starting the event loop in a separate thread, continuing execution of the main thread (i.e., the client code) and then shutting down the thread when the event loop is complete.

Option 1 is technically doable, but I feel like it is messy, and running processes on MPI ranks that randomly grab extra threads without being explicit about it seems like a way of creating problems for users later on. ...But it's possible.

What do people think about this?

[jacobtomlinson]

This is a fun rabbit hole. I spent a ton of time thinking about async-in-sync, sync-in-async, sync-in-async-in-sync, etc recently.

To clarify the client code can either be sync or async, but we have no control over how the user decides to implement things so we have to support both.

If we are in a situation where we want to start more than one component per rank it's probably because each node is big and a single rank refers to a substantial machine. Do you think it's fair to say in these cases that the "one process per rank" rule will not be enforced? It seems bonkers for an HPC admin to enforce a single process on a machine with 8 GPUs. How on earth do they expect to saturate that hardware?