Multiple cores per process/thread

[dgasmith]

We have a use case where we would like to dask.distributed to parallelize a python-bound C++ program that would work best if it could consume 8-32 threads depending on the problem size and will manage threading internally. Normally, a Dask worker is run on a node that has 32 cores with the worker using 2 processes at 1 thread each so that we can give each process 16 cores.

Looking at the code currently, it seems that nthreads = ncores / nprocesses without exceptions, is there a canonical way to change this so that we can orchestrate our normal Dask worker operation with dask-jobqueue?

[guillaumeeb]

Funnily enough, I think we just had a conversation on this point in #179.

As you point out cores/processes kwargs are used both to configure dask-worker command, and to reserve jobqueue system resources. However, the second part is just by default guessed from this kwargs. In every JobQueueCluster implementation, there should be another kwarg allowing you to force different resource reservation on the underlying scheduling system. For example with PBSCluster:

cluster = PBSCluster(cores=2, processes=2, memory='100GB', 
                     resource_spec='select=1:ncpus=32:mem=100GB')

or LSFCluster:

cluster = LSFCluster(cores=2, processes=2, ncpus=32)

It seems we should better document this though, and this seems useful, we were previously thinking of removing this possibility...

[dgasmith]

saw on LSF you could select cpus, but I wasn't aware of the PBSCluster abilities to write out resource specs. To make things simpler, would it be possible to set the number of threads manually so that you would have nthreads = None which would default to current behavior.

It is worth considering this in a larger context where one might wish to start heterogeneous workers with labels where expanding def start_workers(self, n=1): to allow a user to override the internal settings would be very useful. Very happy to code and document here, mostly trying to figure out the current bounds of the project.

[guillaumeeb]

I think there is something to do with the cores/processes/threads/ncpus/ressources_spec.. At least some standardisation between JobQueueCluster implementations. I'm not sure how it should look like though, we previously removed the use of threads in order to replace it by cores. But maybe we should remove ncpus and alike and add backs threads...

It would be more than welcome if you could propose some simple design choices here so we can discuss it, and then if we agree on something, propose a PR!

For the second part on the larger context, I'm not sure I get your point, but maybe it has another scope and should be discussed in another issue?

[dgasmith]

I think my proposal would be similar to the discussed. Scanning through core.py I see the following points:

JobQueueCluster(..., threads_per_process=None, ...):

    self.threads_per_process = threads_per_process or int(self.worker_cores / self.worker_processes)

...

    @property
    def worker_threads(self):
        return self.threads_per_process

    def __repr__(self):
        running_workers = self._count_active_workers()
        running_cores = running_workers * self.worker_threads # maybe changes this to self.cores?

I am not entirely sure what the mental model you want to convey so this project remains relatively easy for new users while being mutable for complex scenarios. But the above while cover our primary use case.

I will hold off the second part for later.

[guillaumeeb]

Ping @mrocklin @jhamman @willirath @lesteve for opinion on how to best handle this.

running_cores = running_workers * self.worker_threads # maybe changes this to self.cores?

No it's thread, or maybe cores divided per process. We've got one running worker per worker process actually.

[lesteve]

It feels like what you want is an advanced use case, so I don't think resource_spec should be automatically inferred from cores and processes. I would recommend:

• setting cores and processes in terms of how many cores/processes/threads you want each dask-worker to have
• specifying explicitly your scheduler-specific resource_spec to say that each job would get the number of threads you want your C++ code to use

[louisabraham]

Replying to the above : I think resource_spec is by worker, not by task and it is specific to some queue systems.

IMO we have three levels:

• the real number of cores that will be allocated, which should always be specified by design
• the number of processes, ie the number of GIL bound tasks that can be executed, and the number of threads per process, ie how many tasks each process can execute
• the number of threads by task

The problem is that the number of cores is supposed to be the number of tasks in our current model, ie threads_by_task=1.

I propose two alternatives:

• a new parameter threads_by_task that divides threads
• a boolean one_task_per_process defaulting to False

I suppose that when handling parallelism manually, one doesn't want to share a process with another task.
one_task_per_process=True sets threads=1 in the dask worker options and still allocates cores/processes threads per process. Since we now have only one task per process, the handcrafted parallelism can allocate up to threads without lying to the scheduler.

I would be more in favor of a boolean than threads_by_task that adds complexity to the system and might break existing parallelism mechanisms if the handcrafted code is executed by a thread (not sure about that).

[lesteve]

@louisabraham it is not completely straightforward to follow what you are saying, and from what I understand it feels like dask.distributed is the right level to address some of your points (for example, I guess by task you mean a function that is used in client.submit right?)

Granted there are plenty of things that could be added/improved in dask-jobqueue and suggestions are always more than welcome! I would suggest we start from precise use cases like the one in the original post and see how we can tackle them.

For anything non-trivial, my opinion that you will have to specify resource_spec, and so you will have to understand a bit the underlying scheduler specificities. Given the variability of the different schedulers (and even sometimes the variability of the installs for a given scheduler), I would be a bit hesitant about providing a layer that allows to forget about scheduler-specific things.

My personal opinion is that being able to easily create heterogeneous clusters would be the right direction in which to extend (but even then maybe some of this work may happen in dask.distributed).

[louisabraham]

it feels like dask.distributed is the right level to address some of your points

I didn't intend it this way.

by task you mean a function that is used in client.submit

Yes.

I would suggest we start from precise use cases like the one in the original post and see how we can tackle them.

Totally agree. To achieve what the author wants with my proposal, the parameters are

cluster = ANYCluster(cores=32, processes=2, one_task_per_process=True)

What disturbs me in @guillaumeeb's immediate fix (it is the only way to do it currently,
which is the real problem)

PBSCluster:

cluster = PBSCluster(cores=2, processes=2, memory='100GB', 
                     resource_spec='select=1:ncpus=32:mem=100GB')

or LSFCluster:

cluster = LSFCluster(cores=2, processes=2, ncpus=32)

is that there is no variable that can track the total number of CPUs allocated and that the code is scheduler specific.

On the other hand, cores totals to the total number of tasks that can be executed concurrently.
If you don't want to break this, threads_by_task would allocate threads_by_task*cores CPUs.

For the OP's problem:

cluster = ANYCluster(cores=2, processes=2, threads_by_task=16)

I think the question we want to ask now is: should cores be the number of CPUs or the number of tasks that can be executed at the same time?

[ocaisa]

We have this use case of heterogeneous clusters at our site. We have modular supercomputing, with pure CPU nodes, nodes with GPUs and KNL nodes all available.

While this is heterogeneous architectures, we also have the use case of heterogeneous resources within clusters. We want the ability to self-manage some of the resources (see #84 (comment) for the use case). At present we simply use job_extra to override the resource request that dask-jobqueue makes

        job_extra=[
            '-N %s' % nodes, 
            '--tasks-per-node=%s' % tasks_per_node,
            '--cpus-per-task=%s' % cpus_per_task,
            # Override the default -n 1 value with the calculated value
            '-n %s' % total_tasks
        ],

This results in a submission script like

#SBATCH -J mpi_tasks
#SBATCH -p booster
#SBATCH -n 1
#SBATCH --cpus-per-task=1
#SBATCH --mem=12G
#SBATCH -t 00:19:00
#SBATCH -N 2
#SBATCH --tasks-per-node=68
#SBATCH --cpus-per-task=1
#SBATCH -n 136

I don't know if the same approach would work with other schedulers.

[dgasmith]

It seem ultimately that you do not want users touching resource_spec or similar where possible as it invalidates all of your internal tracking in addition to being custom per scheduler.

I think the question we want to ask now is: should cores be the number of CPUs or the number of tasks that can be executed at the same time?

I would be hesitant to override the "core" definition with supercomputers as it is well defined there, where "tasks" is well defined in Dask.

Heterogeneous clusters would difficult to tackle in a generic way I think. It might be worth adding a start_worker() that allows full specification of the dask-worker CLI as intermediate step.

PS: Hey @ocaisa!

[louisabraham]

Ok, then my proposal is the following:

A threads_by_task defaulting to 1.

Each dask worker receives the parameters processes and threads like now.

Jobqueue allocates cores * threads_by_task CPUs where cores = processes * threads.

I think heterogeneous clusters are out of the topic.

[guillaumeeb]

I think heterogeneous clusters are out of the topic.

Yep I agree, we should address this in another issue if this is something that interests people. See #103 already open.

For the other part of the rich discussion here, it's a complicated desgin choice, but I thin I'm more in favor of avoiding jobqueue system specific kwargs as much as possible. I think I would better like what @dgasmith proposes: having three kwargs:

• cores: which is the ncpus reserved in jobqueue system, and the default total cores used by dask-worker command. e.g. worker_threads = cores / processes.
• processes: default to one, only useful for dask-worker command.
• threads_per_process or something like that: default to none, only useful for dask-worker command.

For job scheduler specificities, we would rely only on job_extra, as @ocaisa is doing. This would mean removing resources_spec from PBSCluster or ncpus from SlurmCluster.

I've two remaining concerns:

• How should we handle the memory part, which may not be expressed identically between dask and jobqueue systems, can we have only one parameter easilly?
• Would we be able to override resources_spec for example in job_extra if really needed, e.g. if there are for example two #PBS -l select=1:ncpus=24 in the job script, would the second be prioritize over the first? Or should we put job_extra first depending on the scheduler?

Any thought?

[louisabraham]

@guillaumeeb I'm not sure to follow.

In your proposal, isn't cores = processes * threads_per_process?

Maybe your threads_per_process is the same as my threads_by_task, aka the number of threads that execute one task.