Repository: dask/dask-jobqueue
Pull request #680 (2025-01-30 17:16:28)

Set -N=1/--nodes=1 for SLURM workers
The SLURM worker hardcodes --ntasks=1/-n=1, but on the SLURM cluster that I have experience with (SDSC Expanse), you also have to set -N=1/--nodes=1.


Patch:
diff --git a/dask_jobqueue/core.py b/dask_jobqueue/core.py
index fe70ed5..8fcf4ca 100644
--- a/dask_jobqueue/core.py
+++ b/dask_jobqueue/core.py
@@ -25,9 +25,12 @@ logger = logging.getLogger(__name__)
 
 job_parameters = """
     cores : int
-        Total number of cores per job
+        Total number of CPU cores on which all worker threads inside a job will run.
+        The number of threads per worker process are determined using the formula ``cores / processes``.
+        Used by job queuing system by default as amount of CPUs per job.
     memory: str
-        Total amount of memory per job
+        Total amount of memory to be used by all workers inside a job.
+        Used by job queuing system by default as amount of memory per job.
     processes : int
         Cut the job up into this many processes. Good for GIL workloads or for
         nodes with many cores.
@@ -55,6 +58,8 @@ job_parameters = """
         Deprecated: use ``job_script_prologue`` instead. This parameter will be removed in a future version.
     job_script_prologue : list
         Other commands to add to script before launching worker.
+    job_script_epilogue : list
+        Commands to add to script which will run after the worker command has exited.
     header_skip : list
         Deprecated: use ``job_directives_skip`` instead. This parameter will be removed in a future version.
     job_directives_skip : list
@@ -144,6 +149,7 @@ class Job(ProcessInterface, abc.ABC):
 %(job_header)s
 %(job_script_prologue)s
 %(worker_command)s
+%(job_script_epilogue)s
 """.lstrip()
 
     # Following class attributes should be overridden by extending classes.
@@ -173,6 +179,7 @@ class Job(ProcessInterface, abc.ABC):
         job_extra_directives=None,
         env_extra=None,
         job_script_prologue=None,
+        job_script_epilogue=None,
         header_skip=None,
         job_directives_skip=None,
         log_directory=None,
@@ -274,6 +281,10 @@ class Job(ProcessInterface, abc.ABC):
             job_script_prologue = dask.config.get(
                 "jobqueue.%s.job-script-prologue" % self.config_name
             )
+        if job_script_epilogue is None:
+            job_script_epilogue = dask.config.get(
+                "jobqueue.%s.job-script-epilogue" % self.config_name
+            )
         if env_extra is not None:
             warn = (
                 "env_extra has been renamed to job_script_prologue. "
@@ -341,6 +352,7 @@ class Job(ProcessInterface, abc.ABC):
         self.shebang = shebang
 
         self._job_script_prologue = job_script_prologue
+        self._job_script_epilogue = job_script_epilogue
 
         # dask-worker command line build
         dask_worker_command = "%(python)s -m %(worker_command)s" % dict(
@@ -393,6 +405,7 @@ class Job(ProcessInterface, abc.ABC):
             "job_header": self.job_header,
             "job_script_prologue": "\n".join(filter(None, self._job_script_prologue)),
             "worker_command": self._command_template,
+            "job_script_epilogue": "\n".join(filter(None, self._job_script_epilogue)),
         }
         return self._script_template % pieces
 
diff --git a/dask_jobqueue/htcondor.py b/dask_jobqueue/htcondor.py
index 4ab5c07..9ab2636 100644
--- a/dask_jobqueue/htcondor.py
+++ b/dask_jobqueue/htcondor.py
@@ -62,6 +62,12 @@ Queue
                 self._job_script_prologue + [self._command_template]
             )
 
+        if self._job_script_epilogue is not None:
+            # Overwrite command template: append commands from job_script_epilogue separated by semicolon.
+            self._command_template = "; ".join(
+                [self._command_template] + self._job_script_epilogue
+            )
+
         self.job_header_dict = {
             "MY.DaskWorkerName": '"htcondor--$F(MY.JobId)--"',
             "batch_name": self.name,
diff --git a/dask_jobqueue/jobqueue.yaml b/dask_jobqueue/jobqueue.yaml
index f942415..becccd5 100644
--- a/dask_jobqueue/jobqueue.yaml
+++ b/dask_jobqueue/jobqueue.yaml
@@ -23,6 +23,7 @@ jobqueue:
     walltime: '00:30:00'
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     resource-spec: null
     job-extra: null
     job-extra-directives: []
@@ -57,6 +58,7 @@ jobqueue:
     walltime: '00:30:00'
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     resource-spec: null
     job-extra: null
     job-extra-directives: []
@@ -90,6 +92,7 @@ jobqueue:
     walltime: '00:30:00'
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     job-extra: null
     job-extra-directives: []
     job-directives-skip: []
@@ -123,6 +126,7 @@ jobqueue:
     walltime: '00:30:00'
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     job-cpu: null
     job-mem: null
     job-extra: null
@@ -157,6 +161,7 @@ jobqueue:
     walltime: '00:30:00'
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     resource-spec: null
     job-extra: null
     job-extra-directives: []
@@ -190,6 +195,7 @@ jobqueue:
     walltime: '00:30'
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     ncpus: null
     mem: null
     job-extra: null
@@ -223,6 +229,7 @@ jobqueue:
     disk: null                  # Total amount of disk per job
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     job-extra: null             # Extra submit attributes
     job-extra-directives: {}    # Extra submit attributes
     job-directives-skip: []
@@ -252,6 +259,7 @@ jobqueue:
 
     env-extra: null
     job-script-prologue: []
+    job-script-epilogue: []
     job-extra: null
     job-extra-directives: []
     job-directives-skip: []
diff --git a/dask_jobqueue/slurm.py b/dask_jobqueue/slurm.py
index 7af75b1..3660cae 100644
--- a/dask_jobqueue/slurm.py
+++ b/dask_jobqueue/slurm.py
@@ -81,9 +81,10 @@ class SLURMJob(Job):
         if account is not None:
             header_lines.append("#SBATCH -A %s" % account)
 
-        # Init resources, always 1 task,
+        # Init resources, always 1 task, 1 node,
         # and then number of cpu is processes * threads if not set
         header_lines.append("#SBATCH -n 1")
+        header_lines.append("#SBATCH -N 1")
         header_lines.append(
             "#SBATCH --cpus-per-task=%d" % (job_cpu or self.worker_cores)
         )
@@ -166,10 +167,11 @@ class SLURMCluster(JobQueueCluster):
     walltime : str
         Walltime for each worker job.
     job_cpu : int
-        Number of cpu to book in SLURM, if None, defaults to worker `threads * processes`
+        Number of CPUs to request in SLURM for each job.
+        This option might be useful to request more CPUs than total number of worker threads
+        for some complex non-python code. If None, defaults to ``cores``.
     job_mem : str
-        Amount of memory to request in SLURM. If None, defaults to worker
-        processes * memory
+        Amount of memory to request in SLURM for each job, If None, defaults to ``memory``.
     job_extra : list
         Deprecated: use ``job_extra_directives`` instead. This parameter will be removed in a future version.
     job_extra_directives : list
diff --git a/docs/source/clusters-advanced-tips-and-tricks.rst b/docs/source/clusters-advanced-tips-and-tricks.rst
index 257ed09..0d6fa9f 100644
--- a/docs/source/clusters-advanced-tips-and-tricks.rst
+++ b/docs/source/clusters-advanced-tips-and-tricks.rst
@@ -104,6 +104,7 @@ parameter in the submit description file. The relevant lines will look like this
 For other batch systems (``*Cluster`` classes) the additional commands will be inserted as separate lines
 in the submission script.
 
+Similarly, if you need to run some commands after the worker has exited, then use ``job_script_epilogue`` parameter.
 
 How to handle job queueing system walltime killing workers
 ----------------------------------------------------------
@@ -118,7 +119,7 @@ Reaching walltime can be troublesome in several cases:
   killed before reaching the end. In this case, you'll want to use adaptive clusters so 
   that Dask ensures some workers are always up.
 
-If you don't set the proper parameters, you'll run into KilleWorker exception in those two cases.
+If you don't set the proper parameters, you'll run into KilledWorker exception in those two cases.
 
 The solution to this problem is to tell Dask up front that the workers have a finite lifetime:
 


Test patch:
diff --git a/dask_jobqueue/tests/test_htcondor.py b/dask_jobqueue/tests/test_htcondor.py
index 4ee455a..6b07623 100644
--- a/dask_jobqueue/tests/test_htcondor.py
+++ b/dask_jobqueue/tests/test_htcondor.py
@@ -33,6 +33,9 @@ def test_job_script():
             "cd /some/path/",
             "source venv/bin/activate",
         ],
+        job_script_epilogue=[
+            'echo "Job finished"',
+        ],
         job_extra_directives={"+Extra": "True"},
         submit_command_extra=["-verbose"],
         cancel_command_extra=["-forcex"],
@@ -64,6 +67,7 @@ def test_job_script():
         assert f"--memory-limit {formatted_bytes}" in job_script
         assert "--nthreads 2" in job_script
         assert "--nworkers 2" in job_script
+        assert 'echo ""Job finished""' in job_script
 
 
 @pytest.mark.env("htcondor")
@@ -144,6 +148,7 @@ def test_config_name_htcondor_takes_custom_config():
         "worker-extra-args": [],
         "env-extra": None,
         "job-script-prologue": [],
+        "job-script-epilogue": [],
         "log-directory": None,
         "shebang": "#!/usr/bin/env condor_submit",
         "local-directory": "/tmp",
diff --git a/dask_jobqueue/tests/test_lsf.py b/dask_jobqueue/tests/test_lsf.py
index 96fdc60..5f8c743 100644
--- a/dask_jobqueue/tests/test_lsf.py
+++ b/dask_jobqueue/tests/test_lsf.py
@@ -318,6 +318,7 @@ def test_config_name_lsf_takes_custom_config():
         "worker-extra-args": [],
         "env-extra": None,
         "job-script-prologue": [],
+        "job-script-epilogue": [],
         "log-directory": None,
         "shebang": "#!/usr/bin/env bash",
         "use-stdin": None,
diff --git a/dask_jobqueue/tests/test_oar.py b/dask_jobqueue/tests/test_oar.py
index 894ffdd..cd9ff12 100644
--- a/dask_jobqueue/tests/test_oar.py
+++ b/dask_jobqueue/tests/test_oar.py
@@ -96,6 +96,9 @@ def test_job_script():
             'export LANGUAGE="en_US.utf8"',
             'export LC_ALL="en_US.utf8"',
         ],
+        job_script_epilogue=[
+            'echo "Job finished"',
+        ],
     ) as cluster:
         job_script = cluster.job_script()
         assert "#OAR" in job_script
@@ -118,6 +121,7 @@ def test_job_script():
         assert "--nthreads 2" in job_script
         assert "--nworkers 4" in job_script
         assert f"--memory-limit {formatted_bytes}" in job_script
+        assert 'echo "Job finished"' in job_script
 
 
 def test_config_name_oar_takes_custom_config():
@@ -141,6 +145,7 @@ def test_config_name_oar_takes_custom_config():
         "worker-extra-args": [],
         "env-extra": None,
         "job-script-prologue": [],
+        "job-script-epilogue": [],
         "log-directory": None,
         "shebang": "#!/usr/bin/env bash",
         "job-cpu": None,
diff --git a/dask_jobqueue/tests/test_pbs.py b/dask_jobqueue/tests/test_pbs.py
index 89db0f6..ee15046 100644
--- a/dask_jobqueue/tests/test_pbs.py
+++ b/dask_jobqueue/tests/test_pbs.py
@@ -355,6 +355,7 @@ def test_config_name_pbs_takes_custom_config():
         "worker-extra-args": [],
         "env-extra": None,
         "job-script-prologue": [],
+        "job-script-epilogue": [],
         "log-directory": None,
         "shebang": "#!/usr/bin/env bash",
         "job-cpu": None,
diff --git a/dask_jobqueue/tests/test_sge.py b/dask_jobqueue/tests/test_sge.py
index 9e347ed..b250c3e 100644
--- a/dask_jobqueue/tests/test_sge.py
+++ b/dask_jobqueue/tests/test_sge.py
@@ -61,6 +61,7 @@ def test_config_name_sge_takes_custom_config():
         "worker-extra-args": [],
         "env-extra": None,
         "job-script-prologue": [],
+        "job-script-epilogue": [],
         "log-directory": None,
         "shebang": "#!/usr/bin/env bash",
         "job-cpu": None,
@@ -84,6 +85,9 @@ def test_job_script(tmpdir):
         project="my-project",
         walltime="02:00:00",
         job_script_prologue=["export MY_VAR=my_var"],
+        job_script_epilogue=[
+            'echo "Job finished"',
+        ],
         job_extra_directives=["-w e", "-m e"],
         log_directory=log_directory,
         resource_spec="h_vmem=12G,mem_req=12G",
@@ -106,6 +110,7 @@ def test_job_script(tmpdir):
             "-l h_vmem=12G,mem_req=12G",
             "#$ -cwd",
             "#$ -j y",
+            'echo "Job finished"',
         ]:
             assert each in job_script
 
diff --git a/dask_jobqueue/tests/test_slurm.py b/dask_jobqueue/tests/test_slurm.py
index c06ece5..037544e 100644
--- a/dask_jobqueue/tests/test_slurm.py
+++ b/dask_jobqueue/tests/test_slurm.py
@@ -35,6 +35,7 @@ def test_header():
         assert "#SBATCH" in cluster.job_header
         assert "#SBATCH -J dask-worker" in cluster.job_header
         assert "#SBATCH -n 1" in cluster.job_header
+        assert "#SBATCH -N 1" in cluster.job_header
         assert "#SBATCH --cpus-per-task=8" in cluster.job_header
         assert "#SBATCH --mem=27G" in cluster.job_header
         assert "#SBATCH -t 00:02:00" in cluster.job_header
@@ -61,6 +62,7 @@ def test_header():
         assert "#SBATCH" in cluster.job_header
         assert "#SBATCH -J " in cluster.job_header
         assert "#SBATCH -n 1" in cluster.job_header
+        assert "#SBATCH -N 1" in cluster.job_header
         assert "#SBATCH -t " in cluster.job_header
         assert "#SBATCH -p" not in cluster.job_header
         # assert "#SBATCH -A" not in cluster.job_header
@@ -76,6 +78,7 @@ def test_job_script():
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
         assert f"--memory-limit {formatted_bytes}" in job_script
         assert "#SBATCH -n 1" in job_script
+        assert "#SBATCH -N 1" in cluster.job_header
         assert "#SBATCH --cpus-per-task=8" in job_script
         assert "#SBATCH --mem=27G" in job_script
         assert "#SBATCH -t 00:02:00" in job_script
@@ -103,11 +106,15 @@ def test_job_script():
             'export LANGUAGE="en_US.utf8"',
             'export LC_ALL="en_US.utf8"',
         ],
+        job_script_epilogue=[
+            'echo "Job finished"',
+        ],
     ) as cluster:
         job_script = cluster.job_script()
         assert "#SBATCH" in job_script
         assert "#SBATCH -J dask-worker" in job_script
         assert "#SBATCH -n 1" in job_script
+        assert "#SBATCH -N 1" in cluster.job_header
         assert "#SBATCH --cpus-per-task=8" in job_script
         assert "#SBATCH --mem=27G" in job_script
         assert "#SBATCH -t 00:02:00" in job_script
@@ -127,6 +134,8 @@ def test_job_script():
         assert "--nworkers 4" in job_script
         assert f"--memory-limit {formatted_bytes}" in job_script
 
+        assert 'echo "Job finished"' in job_script
+
 
 @pytest.mark.env("slurm")
 def test_basic(loop):
@@ -208,6 +217,7 @@ def test_config_name_slurm_takes_custom_config():
         "worker-extra-args": [],
         "env-extra": None,
         "job-script-prologue": [],
+        "job-script-epilogue": [],
         "log-directory": None,
         "shebang": "#!/usr/bin/env bash",
         "job-cpu": None,