86: Slurm script template to run Nextflow workflows

Author: rafaelfsilva

docs/source/generating_workflow_benchmarks.rst

@@ -141,6 +141,17 @@ workflow benchmark for running with Nextflow::
     that depend on another instance of the same abstract task. Thus, the translator
     fails when you try to translate a workflow with iterations.
 
+.. note::
+    
+    If you plan to run Nextflow on an HPC system using Slurm, we **strongly
+    recommend** using the `HyperQueue <https://github.com/It4innovations/hyperqueue>`_ 
+    executor. HyperQueue efficiently distributes workflow tasks across all allocated 
+    compute nodes, improving scalability and resource utilization.
+
+    The :class:`~wfcommons.wfbench.translator.nextflow.NextflowTranslator`
+    class includes functionality to automatically generate a Slurm script 
+    template for running the workflow on HPC systems.
+
 Pegasus
 +++++++
 

wfcommons/wfbench/translator/nextflow.py

@@ -26,16 +26,19 @@ class NextflowTranslator(Translator):
     A WfFormat parser for creating Nextflow workflow applications.
 
     :param workflow: Workflow benchmark object or path to the workflow benchmark JSON instance.
-    :type workflow: Union[Workflow, pathlib.Path],
+    :type workflow: Union[Workflow, pathlib.Path]
+    :param slurm: Whether to generate a Slurm template script for workflow submission using :code:`sbatch`.
+    :type slurm: bool
     :param logger: The logger where to log information/warning or errors (optional).
     :type logger: Logger
     """
     def __init__(self,
                  workflow: Union[Workflow, pathlib.Path],
+                 slurm: Optional[bool] = False,
                  logger: Optional[Logger] = None) -> None:
         """Create an object of the translator."""
         super().__init__(workflow, logger)
-
+        self.slurm = slurm
         self.script = ""
         self.out_files = set()
 
@@ -54,6 +57,9 @@ def translate(self, output_folder: pathlib.Path) -> None:
         self._copy_binary_files(output_folder)
         self._generate_input_files(output_folder)
 
+        if self.slurm:
+            shutil.copy(this_dir.joinpath("templates/nextflow/nextflow_hyperqueue_job.sh"), output_folder)
+
         if self.workflow.workflow_id:
             shutil.copy(this_dir.joinpath("templates/flowcept_agent.py"), output_folder.joinpath("bin"))
 

wfcommons/wfbench/translator/templates/nextflow/nextflow_hyperqueue_job.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+#SBATCH --job-name=nextflowjob
+#SBATCH --partition=<PARTITION_NAME>
+#SBATCH --account=<ACCOUNT>
+#SBATCH --nodes=<NUM_NODES>
+#SBATCH --ntasks-per-node=1
+#SBATCH --cpus-per-task=<MAX_NUM_CPU_PER_TASK>
+#SBATCH --time=01:00:00
+
+# Set variables
+export PATH=$PATH:/$PWD
+export NXF_DISABLE_CHECK_LATEST=true
+
+# Set the directory which hyperqueue will use 
+export HQ_SERVER_DIR=${PWD}/.hq-server
+mkdir -p ${HQ_SERVER_DIR}
+
+# Start the server in the background (&) and wait until it has started
+hq server start &
+until hq job list &>/dev/null ; do sleep 1 ; done
+
+# Start the workers in the background and wait for them to start
+srun --overlap --cpu-bind=none --mpi=none hq worker start --cpus=${SLURM_CPUS_PER_TASK} &
+hq worker wait "${SLURM_NTASKS}"
+
+# Ensure Nextflow uses the right executor and knows how many jobs it can submit
+# The `queueSize` can be limited as needed. 
+echo "executor {
+  queueSize = $(( SLURM_CPUS_PER_TASK*SLURM_NNODES ))
+  name = 'hq'
+  cpus = $(( SLURM_CPUS_PER_TASK*SLURM_NNODES ))
+}" > ${PWD}/nextflow.config
+
+# run the Nextflow pipeline 
+nextflow run workflow.nf -c nextflow.config --pwd $PWD &> output.log
+
+# Wait for all jobs to finish, then shut down the workers and server
+hq job wait all
+hq worker stop all
+hq server stop