Merge pull request #131 from wfcommons/nextflow-subworkflow

Nextflow subworkflow

Author: rafaelfsilva

docs/source/generating_workflow_benchmarks.rst

@@ -132,8 +132,24 @@ workflow benchmark for running with Nextflow::
     benchmark.create_benchmark(pathlib.Path("/tmp/"), cpu_work=100, data=10, percent_cpu=0.6)
 
     # generate a Nextflow workflow
-    translator = NextflowTranslator(benchmark.workflow)
-    translator.translate(output_folder=pathlib.Path("./nextflow-wf/""))
+    translator = NextflowTranslator(
+        benchmark.workflow,
+        use_subworkflows=False,
+        max_tasks_per_subworkflow=1000,
+    )
+    translator.translate(output_folder=pathlib.Path("./nextflow-wf/"))
+
+If you want to split large workflows across multiple Nextflow module files, enable
+subworkflows and set the maximum number of tasks per module. This produces a
+``modules/`` directory plus a top-level ``workflow.nf`` that includes and runs
+the modules sequentially::
+
+    translator = NextflowTranslator(
+        benchmark.workflow,
+        use_subworkflows=True,
+        max_tasks_per_subworkflow=250,
+    )
+    translator.translate(output_folder=pathlib.Path("./nextflow-wf/"))
 
 .. warning::
 

tests/requirements.txt

@@ -0,0 +1,3 @@
+pytest
+pytest-cov
+docker

tests/translators_loggers/test_translators_loggers.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2025 The WfCommons Team.
+# Copyright (c) 2025-2026 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -105,6 +105,7 @@ def _additional_setup_swiftt(container):
     "dask": noop,
     "parsl": noop,
     "nextflow": noop,
+    "nextflow_subworkflow": noop,
     "airflow": noop,
     "bash": noop,
     "taskvine": _additional_setup_taskvine,
@@ -203,6 +204,7 @@ def run_workflow_swiftt(container, num_tasks, str_dirpath):
     "dask": run_workflow_dask,
     "parsl": run_workflow_parsl,
     "nextflow": run_workflow_nextflow,
+    "nextflow_subworkflow": run_workflow_nextflow,
     "airflow": run_workflow_airflow,
     "bash": run_workflow_bash,
     "taskvine": run_workflow_taskvine,
@@ -216,6 +218,7 @@ def run_workflow_swiftt(container, num_tasks, str_dirpath):
     "dask": DaskTranslator,
     "parsl": ParslTranslator,
     "nextflow": NextflowTranslator,
+    "nextflow_subworkflow": NextflowTranslator,
     "airflow": AirflowTranslator,
     "bash": BashTranslator,
     "taskvine": TaskVineTranslator,
@@ -235,6 +238,7 @@ class TestTranslators:
            "dask",
            "parsl",
            "nextflow",
+           "nextflow_subworkflow",
            "airflow",
            "bash",
            "taskvine",
@@ -256,7 +260,10 @@ def test_translator(self, backend) -> None:
 
         # Perform the translation
         sys.stderr.write(f"\n[{backend}] Translating workflow...\n")
-        translator = translator_classes[backend](benchmark.workflow)
+        if backend == "nextflow_subworkflow":
+            translator = translator_classes[backend](benchmark.workflow, use_subworkflows=True, max_tasks_per_subworkflow=10)
+        else:
+            translator = translator_classes[backend](benchmark.workflow)
         translator.translate(output_folder=dirpath)
 
         # # Make the directory that holds the translation world-writable,
@@ -266,7 +273,7 @@ def test_translator(self, backend) -> None:
         # os.chmod(dirpath, 0o777)
 
         # Start the Docker container
-        container = _start_docker_container(backend, str_dirpath, str_dirpath, str_dirpath + "bin/")
+        container = _start_docker_container(backend if backend != "nextflow_subworkflow" else "nextflow", str_dirpath, str_dirpath, str_dirpath + "bin/")
 
         # Do whatever necessary setup
         additional_setup_methods[backend](container)