Manually set name of Worker [in dask-jobqueue] #7070

jmuchovej · 2021-01-14T23:29:41Z

jmuchovej
Jan 14, 2021

Background

My sys-admin has asked me to use SLURM's Job Arrays, however all but one of my workers will die off in the array due to [what seems to be] a naming error. I'm currently using the "fix" specified here.

Question

Is it possible to manually set the name of a worker, programmatically?

Stacktrace, `SLURMCluster` Config, and Job Script

The trace I get back is the following:

distributed.worker - INFO -       Start worker at:    tcp://172.16.20.8:43811
distributed.worker - INFO -          Listening to:    tcp://172.16.20.8:43811
distributed.worker - INFO -          dashboard at:          172.16.20.8:35261
distributed.worker - INFO - Waiting to connect to:    tcp://172.16.20.5:39090
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -               Threads:                          2
distributed.worker - INFO -                Memory:                    2.00 GB
distributed.worker - INFO -       Local Directory: <redacted>/dask-worker-space/worker-9mfov8xl
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO - Stopping worker at tcp://172.16.20.8:43811
distributed.dask_worker - INFO - End worker
Traceback (most recent call last):
  File "<redacted>/lib/python3.8/site-packages/distributed/core.py", line 295, in _
    await asyncio.wait_for(self.start(), timeout=timeout)
  File "<redacted>/lib/python3.8/asyncio/tasks.py", line 491, in wait_for
    return fut.result()
  File "<redacted>/lib/python3.8/site-packages/distributed/worker.py", line 1091, in start
    await self._register_with_scheduler()
  File "<redacted>/lib/python3.8/site-packages/distributed/worker.py", line 844, in _register_with_scheduler
    raise ValueError("Unexpected response from register: %r" % (response,))
ValueError: Unexpected response from register: {'status': 'error', 'message': 'name taken, 0', 'time': 1610665861.915631}

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<redacted>/lib/python3.8/runpy.py", line 194, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "<redacted>/lib/python3.8/runpy.py", line 87, in _run_code
    exec(code, run_globals)
  File "<redacted>/lib/python3.8/site-packages/distributed/cli/dask_worker.py", line 460, in <module>
    go()
  File "<redacted>/lib/python3.8/site-packages/distributed/cli/dask_worker.py", line 456, in go
    main()
  File "<redacted>/lib/python3.8/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "<redacted>/lib/python3.8/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "<redacted>/lib/python3.8/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "<redacted>/lib/python3.8/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "<redacted>/lib/python3.8/site-packages/distributed/cli/dask_worker.py", line 442, in main
    loop.run_sync(run)
  File "<redacted>/lib/python3.8/site-packages/tornado/ioloop.py", line 530, in run_sync
    return future_cell[0].result()
  File "<redacted>/lib/python3.8/site-packages/distributed/cli/dask_worker.py", line 436, in run
    await asyncio.gather(*nannies)
  File "<redacted>/lib/python3.8/asyncio/tasks.py", line 692, in _wrap_awaitable
    return (yield from awaitable.__await__())
  File "<redacted>/lib/python3.8/site-packages/distributed/core.py", line 299, in _
    raise TimeoutError(
asyncio.exceptions.TimeoutError: Worker failed to start in 60 seconds

My SLURMCluster parameters (using gin-config, but that's not the issue, to my knowledge):

#: dask_jobqueue.SLURMCluster ---------------------------------------------
###: Values empirically determined
SLURMCluster.cores = 2
SLURMCluster.processes = 1
SLURMCluster.log_directory = "dask"
SLURMCluster.walltime = "1-00:00:00"
SLURMCluster.shebangstr = "#!/usr/bin/zsh"
SLURMCluster.memory = "2GB"
SLURMCluster.scheduler_options = {
    "dashboard_address": "0.0.0.0:36601",
}
SLURMCluster.python = """
source ~/.zshenv
conda activate <redacted>
python"""
SLURMCluster.job_extra = [
    # "-p <redacted>",
    "--exclude=node[055-059]",
    "--requeue",
    "--array=0-50",
]
SLURMCluster.nanny = False
SLURMCluster.env_extra = [
    "JOB_ID=${SLURM_ARRAY_JOB_ID%;*}_${SLURM_ARRAY_TASK_ID%;*}",
    "SLURM_JOB_ID=${SLURM_ARRAY_JOB_ID%;*}_${SLURM_ARRAY_TASK_ID%;*}",
]

My cluster.job_script() output:

#!/usr/bin/env bash

#SBATCH -J dask-worker
#SBATCH -e dask/dask-worker-%J.err
#SBATCH -o dask/dask-worker-%J.out
#SBATCH -n 1
#SBATCH --cpus-per-task=2
#SBATCH --mem=2G
#SBATCH -t 1-00:00:00
#SBATCH --exclude=node[055-059]
#SBATCH --requeue
#SBATCH --array=0-50
JOB_ID=${SLURM_ARRAY_JOB_ID%;*}_${SLURM_ARRAY_TASK_ID%;*}
SLURM_JOB_ID=${SLURM_ARRAY_JOB_ID%;*}_${SLURM_ARRAY_TASK_ID%;*}

source ~/.zshenv
conda activate <redacted>
python -m distributed.cli.dask_worker tcp://18.13.53.52:43019 --nthreads 2 --memory-limit 2.00GB --name name --no-nanny --death-timeout 60

Dask Version Info

𝙭 conda list | grep -E "dask|distributed"
dask                      2.30.0                     py_0    conda-forge
dask-core                 2.30.0                     py_0    conda-forge
dask-jobqueue             0.7.1                    pypi_0    pypi
distributed               2.30.1           py38h578d9bd_0    conda-forge

Answered by jmuchovej

Jan 15, 2021

NOTE: Following dask/dask-jobqueue#480, the following will work: #7070 (reply in thread).

To be clear, the solution I found was the following:

Create a custom class based on JobQueueCluster, like the following:

 class JobArrayCluster(JobQueueCluster):
    __doc__ = <Batch>Cluster.__doc__
    job_cls = <Batch>Cluster.job_cls

    def _new_worker_name(self, worker_number):
        """Overriding ``SpecCluster._new_worker_name`` to provide unique names across `n_jobs * len(jobArray)`."""
        return f"{self._name}-{worker_number}"

Pass a name into JobArrayCluster like the following: project-${JOB_ID} (the ${JOB_ID} is the important bit).
Modify your job_extra to include the syntax for s…

View full answer

jrbourbeau · 2021-01-15T03:27:39Z

jrbourbeau
Jan 15, 2021
Maintainer

cc @dask/dask-jobqueue-slurm if anyone gets a moment

Is it possible to manually set the name of a worker, programmatically?

I don't have experience using SLURMCluster, but FWIW after briefly looking through the SLURMCluster code, since SLURMCluster is a subclass of distributed.SpecCluster, I would expect overriding the SpecCluster._new_worker_name method would be a good place to start trying to programmatically set worker names.

2 replies

jmuchovej Jan 15, 2021
Author

I actually just stumbled upon that a few minutes ago! 😁 I'm considering submitting a PR at this point regarding that, as I think that would enable Job Array support for JobQueue.

jmuchovej Jan 15, 2021
Author

For those in dask-jobqueue-slurm, would the solution I found be a reasonable patch for dask/dask-jobqueue#196?

It's not quite full-fledged support for Job Arrays (as you need to manually pass --array in job_extra), but it seems to do the trick to allow for folks wanting Job Arrays to use them, at least on SLURM clusters. I don't have access to others to properly test this workload, but I can test my current code on different systems if given access (can't release this code just yet as it's part of publication pending research).

jmuchovej · 2021-01-15T04:45:49Z

jmuchovej
Jan 15, 2021
Author

NOTE: Following dask/dask-jobqueue#480, the following will work: #7070 (reply in thread).

To be clear, the solution I found was the following:

Create a custom class based on JobQueueCluster, like the following:

 class JobArrayCluster(JobQueueCluster):
    __doc__ = <Batch>Cluster.__doc__
    job_cls = <Batch>Cluster.job_cls

    def _new_worker_name(self, worker_number):
        """Overriding ``SpecCluster._new_worker_name`` to provide unique names across `n_jobs * len(jobArray)`."""
        return f"{self._name}-{worker_number}"

Pass a name into JobArrayCluster like the following: project-${JOB_ID} (the ${JOB_ID} is the important bit).
Modify your job_extra to include the syntax for selecting arrays in submissions, e.g. for SLURM: --array=0-{N}.
Modify your env_extra to include the equivalent of: JOB_ID=${SLURM_ARRAY_JOB_ID}_${SLURM_ARRAY_TASK_ID}.

So calling the following should work:

cluster = JobArrayCluster(..., name="project-${JOB_ID}", job_extra=["--array=0-50"], env_extra=[
    "JOB_ID=${SLURM_ARRAY_JOB_ID}_${SLURM_ARRAY_TASK_ID}",
])

NOTE: I've only tested this on a SLURM cluster as it's the only Batch Manager I have access to.

4 replies

lesteve Jan 15, 2021

Thanks for this, that is indeed very similar to what I had in mind in dask/dask-jobqueue#196

jmuchovej Jan 15, 2021
Author

No problem! I don't know how widespread naming collisions are outside of my use-case with SLURM's Job Arrays, but this might also make sense to be implemented in SpecCluster._new_worker_name? (Largely considering that my solution relies on a Cluster being given a name [which, to my knowledge, is eventually given a default value anyways].)

Obviously I don't have the background to make claims about how it won't affect other Dask add-ons, but that seems to provide a more specific worker name, which ought to reduce naming collisions that cause workers to die.

lesteve Jan 15, 2021

IIRC I originally wanted to have the cluster name used automatically in the woker names in SpecCluster._new_worker_name in dask/distributed#3855 but then it was deemed preferrable not to change the SpecCluster behaviour.

I want to reiterate here (on top of the linked issue above dask/dask-jobqueue#196 (comment)) that using job arrays inside dask-jobqueue is a complete hack with plenty of caveats.

lesteve Jan 20, 2021

Thanks to @ionlights PR dask/dask-jobqueue#480, this is a bit simpler now in the DaskJobqueue development version, you can directly use SlurmCluster (no need to create your own JobArrayCluster class). So something like this should work in the development version:

cluster = SLURMCluster(
    ...,
    name="project-${JOB_ID}",
    job_extra=["--array=0-50"],
    env_extra=[
        "JOB_ID=${SLURM_ARRAY_JOB_ID}_${SLURM_ARRAY_TASK_ID}",
])

lesteve · 2021-01-20T09:51:38Z

lesteve
Jan 20, 2021

My sys-admin has asked me to use SLURM's Job Arrays

By the way @ionlights I would be very curious if you can give a rough estimate of the number of very similar jobs that is acceptable to launch on your cluster without job arrays?

In other words, on your cluster and without job arrays, is it OK to use Dask-Jobqueue and do

.scale(jobs=2)?
.scale(jobs=5)?
.scale(jobs=10)?
.scale(jobs=100)?

Also I would be curious if you can give the configuration of your SLURMCluster (cores and memory especially) and the number of total jobs in your Dask cluster to give me an idea of the total resources that you used for your Dask-Jobqueue use case

6 replies

lesteve Jan 20, 2021

I meant more the parameters of your SLURMCluster object sorry for the ambiguity, i.e. basically you replied : 2-core 2GB RAM.

I am curious if there is a reason why you are not using less jobs but bigger jobs. Maybe small jobs start running faster on your cluster ? Maybe some other reason ?

jmuchovej Jan 20, 2021
Author

Maybe I have a broken mental model of dask-jobqueue, but I understood jobqueue to handle "parallelization" strictly a the job level – so adding bigger jobs would have required an additional layer of parallelization (within and between jobs). So I opted for many small jobs as the task was single-threaded, I just needed lots of workers to get it done in a reasonable time.

There's certainly a priority towards low-resource, low-request-time jobs as well. I often found that requesting 1-2 cores for 16-24 hours would get approved within seconds, rather than 5-10+ min waits for longer, bigger jobs. Additionally, I was attempting to make use of remaining resources that other's didn't use (e.g. if all the GPUs in a node were used and the total consumed CPU count was like 60/64, then the jobs I added would likely pack into those 4 remaining CPUs).

lesteve Jan 20, 2021

"broken" is a bit harsh, after all your mental model is good enough to make Dask-Jobqueue useful for use case which is already a clear achievement 😉

Each job launches a dask-worker command (look at cluster.job_script()) and a dask-worker command can have multiple threads or launch multiple processes to have some parallelism. Look at https://docs.dask.org/en/latest/setup/cli.html for example especially the parameters --nthreads and --nprocs

jmuchovej Jan 21, 2021
Author

Hehe, thanks for the kind words. Mostly meant it as, I assumed jobqueue could only handle jobs, not the processes within jobs.

Do you mean to say that if I had a series of input (e.g. list of 10K ints) and used jobqueue to request a SLURM job, but had say 100 cores in that worker, dask would handle distributing the 10K ints across each of the cores in the same way vanilla dask does? (Lol, seems kinda obvious when I lay it out like this, but would be good to know if there's any intermediate steps. I didn't see any on the docs you linked. 😕)

lesteve Jan 21, 2021

Dask-Jobqueue handles conveniently configuring and launching dask-worker jobs through a HPC scheduler. Other than this, this is vanilla Dask.

By the way since each of your job use two cores, you are already doing parallelism inside jobs without knowing it 😉

And sorry I am not sure which part of the Dask doc explains this in details, maybe other Dask developers can chime in about this. You can play with something like this:

In a first terminal

dask-scheduler

In a second terminal

dask-worker --nprocs 4 <scheduler-address:scheduler-port>

In a Python session (Jupyter notebook/IPython whatever)

from distributed import Client
client = Client('<scheduler-address'>)
# submit your task heres
client.submit(...)

Look at the dashboard

client.dashboard_link will tell you the dashboard address

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Manually set name of Worker [in dask-jobqueue] #7070

{{title}}

Replies: 3 comments 12 replies

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Select a reply

Manually set name of Worker [in dask-jobqueue] #7070

Background

Question

Stacktrace, SLURMCluster Config, and Job Script

Dask Version Info

NOTE: Following dask/dask-jobqueue#480, the following will work: #7070 (reply in thread).

Replies: 3 comments · 12 replies

jrbourbeau Jan 15, 2021 Maintainer

jmuchovej Jan 15, 2021 Author

jmuchovej Jan 15, 2021 Author

jmuchovej Jan 15, 2021 Author

NOTE: Following dask/dask-jobqueue#480, the following will work: #7070 (reply in thread).

jmuchovej Jan 15, 2021 Author

jmuchovej Jan 20, 2021 Author

jmuchovej Jan 21, 2021 Author

In a first terminal

In a second terminal

In a Python session (Jupyter notebook/IPython whatever)

Look at the dashboard

Stacktrace, `SLURMCluster` Config, and Job Script

Replies: 3 comments 12 replies

jrbourbeau
Jan 15, 2021
Maintainer

jmuchovej Jan 15, 2021
Author

jmuchovej Jan 15, 2021
Author

jmuchovej
Jan 15, 2021
Author

jmuchovej Jan 15, 2021
Author

jmuchovej Jan 20, 2021
Author

jmuchovej Jan 21, 2021
Author