Merge pull request #206 from cgat-developers/AC-pipes

Ac pipes

Author: Acribbs

CHANGELOG.md

@@ -1,3 +1,7 @@
+# Version 0.6.20
+
+- Fix pipeline parallelism: pass `multiprocess` to `ruffus.pipeline_run()` so multiple jobs run in parallel again
+
 # Version 0.6.19
 
 - Update GitHub Actions workflows to use latest action versions

cgatcore/pipeline/control.py

@@ -44,7 +44,8 @@
 
 import cgatcore.experiment as E
 import cgatcore.iotools as iotools
-from cgatcore.pipeline.parameters import input_validation, get_params, get_parameters
+from cgatcore.pipeline.parameters import (
+    input_validation, get_params, get_parameters, write_params_init_file)
 from cgatcore.experiment import get_header, MultiLineFormatter
 from cgatcore.pipeline.utils import get_caller, get_caller_locals, is_test
 from cgatcore.pipeline.execution import execute, start_session, \
@@ -1251,6 +1252,9 @@ def initialize(argv=None, caller=None, defaults=None, optparse=True, **kwargs):
     logger.info("changing directory to {}".format(work_dir))
     os.chdir(work_dir)
 
+    # Write params snapshot for spawned worker processes (multiprocess with spawn)
+    write_params_init_file(work_dir)
+
     logger.info("pipeline has been initialized")
 
     return args
@@ -1302,7 +1306,8 @@ def run_workflow(args, argv=None, pipeline=None):
                     logger=logger, verbose=args.loglevel, log_exceptions=args.log_exceptions,
                     exceptions_terminate_immediately=args.exceptions_terminate_immediately,
                     checksum_level=args.ruffus_checksums_level, pipeline=pipeline,
-                    one_second_per_job=False
+                    one_second_per_job=False,
+                    multiprocess=args.multiprocess
                 )
 
             except ruffus.ruffus_exceptions.RethrownJobError as ex:

cgatcore/pipeline/parameters.py

@@ -7,6 +7,7 @@
 """
 
 import collections
+import json
 import os
 import sys
 import platform
@@ -561,8 +562,58 @@ def check_parameter(param):
         raise ValueError("need `%s` to be set" % param)
 
 
+def _get_serializable_params():
+    """Return a dict of PARAMS entries that are JSON-serializable (for worker init file)."""
+    out = {}
+    for k, v in PARAMS.items():
+        try:
+            json.dumps(v)
+            out[k] = v
+        except (TypeError, ValueError):
+            pass
+    return out
+
+
+def _load_params_in_worker():
+    """Load PARAMS from init file in a spawned worker process (multiprocess spawn)."""
+    global HAVE_INITIALIZED
+    path = os.environ.get("CGAT_PARAMS_INIT_FILE")
+    if not path or not os.path.exists(path):
+        return
+    try:
+        with open(path, "rt", encoding="utf8") as f:
+            loaded = json.load(f)
+        PARAMS.update(loaded)
+        HAVE_INITIALIZED = True
+    except Exception:
+        pass
+
+
+def write_params_init_file(work_dir):
+    """Write a JSON snapshot of PARAMS for spawned worker processes.
+    Call from the main process after initialize(); set CGAT_PARAMS_INIT_FILE in env.
+    """
+    if not HAVE_INITIALIZED:
+        return
+    try:
+        data = _get_serializable_params()
+        path = os.path.join(os.path.abspath(work_dir), "cgat_params_init_{}.json".format(os.getpid()))
+        with open(path, "wt", encoding="utf8") as f:
+            json.dump(data, f, indent=0)
+        os.environ["CGAT_PARAMS_INIT_FILE"] = path
+    except Exception:
+        pass
+
+
 def get_params():
     """return handle to global parameter dictionary"""
+    if not HAVE_INITIALIZED:
+        try:
+            import multiprocessing
+            if multiprocessing.current_process().name != "MainProcess":
+                _load_params_in_worker()
+        except ImportError:
+            pass
     return PARAMS
 
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "cgatcore"
-version = "0.6.19"
+version = "0.6.20"
 description = "cgatcore: the Computational Genomics Analysis Toolkit"
 authors = [
     { name = "Adam Cribbs", email = "adam.cribbs@ndorms.ox.ac.uk" }

tests/template.yml

@@ -0,0 +1,5 @@
+# Default parameters for template_pipeline tests
+min_value: 0.0
+num_samples: 1000
+mu: 0.0
+sigma: 1.0

tests/template_pipeline.py

@@ -76,17 +76,18 @@ def main(argv=None):
     if argv is None:
         argv = sys.argv
 
-    # Get the args first
-    args = P.initialize(argv, config_file="template.yml")
-
-    # Then use them in the defaults dictionary
-    P.get_params().update({
-        "min_value": 0.0,
-        "num_samples": 1000,
-        "mu": 0.0,
-        "sigma": 1.0,
-        "to_cluster": not args.without_cluster  # Now args is defined
-    })
+    # Pass defaults so params are set before any parallel task runs (workers need them too)
+    args = P.initialize(
+        argv,
+        config_file="template.yml",
+        defaults={
+            "min_value": 0.0,
+            "num_samples": 1000,
+            "mu": 0.0,
+            "sigma": 1.0,
+        },
+    )
+    P.get_params().update({"to_cluster": not args.without_cluster})
 
     pipeline = ruffus.Pipeline("template_pipeline")