From ae218c284e4604f335cca27401067510e38368ec Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Mon, 25 Mar 2024 16:13:34 +0100 Subject: [PATCH 01/18] update: relax concrete class implementation constraints for QueueContextAdapter --- .../api/queue_context/queue_context_adapter.py | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/junifer/api/queue_context/queue_context_adapter.py b/junifer/api/queue_context/queue_context_adapter.py index fe8315b4c..0387d5cc3 100644 --- a/junifer/api/queue_context/queue_context_adapter.py +++ b/junifer/api/queue_context/queue_context_adapter.py @@ -19,14 +19,6 @@ class QueueContextAdapter(ABC): """ - @abstractmethod - def pre_run(self) -> str: - """Return pre-run commands.""" - raise_error( - msg="Concrete classes need to implement pre_run()", - klass=NotImplementedError, - ) - @abstractmethod def run(self) -> str: """Return run commands.""" @@ -35,14 +27,6 @@ class QueueContextAdapter(ABC): klass=NotImplementedError, ) - @abstractmethod - def pre_collect(self) -> str: - """Return pre-collect commands.""" - raise_error( - msg="Concrete classes need to implement pre_collect()", - klass=NotImplementedError, - ) - @abstractmethod def collect(self) -> str: """Return collect commands.""" -- 2.52.0 From fd6e7a8f2d1a347f1baf89d01568e27c05bac998 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Mon, 25 Mar 2024 16:14:36 +0100 Subject: [PATCH 02/18] feature: introduce GnuParallelLocalAdapter and unit tests --- .../gnu_parallel_local_adapter.py | 174 ++++++++++++++++++ .../tests/test_gnu_parallel_local_adapter.py | 93 ++++++++++ 2 files changed, 267 insertions(+) create mode 100644 junifer/api/queue_context/gnu_parallel_local_adapter.py create mode 100644 junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py new file mode 100644 index 000000000..2950d7edd --- /dev/null +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -0,0 +1,174 @@ +"""Define concrete class for generating GNU Parallel (local) assets.""" + +# Authors: Synchon Mandal +# License: AGPL + +import shutil +from pathlib import Path +from typing import Dict, List, Optional, Tuple, Union + +from ...utils import logger, make_executable, raise_error, run_ext_cmd +from .queue_context_adapter import QueueContextAdapter + + +__all__ = ["GnuParallelLocalAdapter"] + + +class GnuParallelLocalAdapter(QueueContextAdapter): + """Class for generating commands for GNU Parallel (local). + + Parameters + ---------- + job_name : str + The job name. + job_dir : pathlib.Path + The path to the job directory. + yaml_config_path : pathlib.Path + The path to the YAML config file. + elements : list of str or tuple + Element(s) to process. Will be used to index the DataGrabber. + env : dict, optional + The Python environment configuration. If None, will run without a + virtual environment of any kind (default None). + verbose : str, optional + The level of verbosity (default "info"). + submit : bool, optional + Whether to submit the jobs (default False). + + Raises + ------ + ValueError + If``env`` is invalid. + + See Also + -------- + QueueContextAdapter : + The base class for QueueContext. + HTCondorAdapter : + The concrete class for queueing via HTCondor. + + """ + + def __init__( + self, + job_name: str, + job_dir: Path, + yaml_config_path: Path, + elements: List[Union[str, Tuple]], + env: Optional[Dict[str, str]] = None, + verbose: str = "info", + submit: bool = False, + ) -> None: + """Initialize the class.""" + self._job_dir = job_dir + self._yaml_config_path = yaml_config_path + self._elements = elements + self._check_env(env) + self._verbose = verbose + self._submit = submit + + self._run_joblog_path = self._job_dir / f"{job_name}_run_joblog.txt" + self._collect_joblog_path = ( + self._job_dir / f"{job_name}_collect_joblog.txt" + ) + + def _check_env(self, env: Optional[Dict[str, str]]) -> None: + """Check value of env parameter on init. + + Parameters + ---------- + env : dict or None + The value of env parameter. + + Raises + ------ + ValueError + If ``env.kind`` is invalid. + + """ + # Set env related variables + if env is None: + env = {"kind": "local"} + # Check env kind + valid_env_kinds = ["conda", "venv", "local"] + if env["kind"] not in valid_env_kinds: + raise_error( + f"Invalid value for `env.kind`: {env['kind']}, " + f"must be one of {valid_env_kinds}" + ) + else: + # Set variables + if env["kind"] == "local": + # No virtual environment + self._executable = "junifer" + self._arguments = "" + else: + self._executable = f"run_{env['kind']}.sh" + self._arguments = f"{env['name']} junifer" + self._exec_path = self._job_dir / self._executable + + def run(self) -> str: + """Return run commands.""" + junifer_run_cmd = ( + f"parallel --joblog {self._run_joblog_path} " + f"{self._job_dir.resolve()!s}/{self._executable} " + f"{self._arguments} run " + f"{self._yaml_config_path.resolve()!s} " + f"--verbose {self._verbose} " + f"--element :::" + ) + for element in self._elements: + # Stringify elements if tuple for operation + str_element = ( + ",".join(element) if isinstance(element, tuple) else element + ) + junifer_run_cmd += f" '{str_element}'" + + return junifer_run_cmd + + def collect(self) -> str: + """Return collect commands.""" + return ( + f"parallel --joblog {self._collect_joblog_path} " + f"{self._job_dir.resolve()!s}/{self._executable} " + f"{self._arguments} collect " + f"{self._yaml_config_path.resolve()!s} " + f"--verbose {self._verbose}" + ) + + def prepare(self) -> None: + """Prepare assets for submission.""" + logger.info("Preparing for local queue via GNU parallel") + # Copy executable if not local + if hasattr(self, "_exec_path"): + logger.info( + f"Copying {self._executable} to " + f"{self._exec_path.resolve()!s}" + ) + shutil.copy( + src=Path(__file__).parent.parent / "res" / self._executable, + dst=self._exec_path, + ) + make_executable(self._exec_path) + # Generate run command + logger.info(f"`junifer run` command:\n{self.run()}") + # Generate collect command + logger.info(f"`junifer collect` command:\n{self.collect()}") + # Submit if required + if self._submit: + logger.info( + "First the `junifer run` command as shown above will be run. " + "After successful completion of the previous step, run the " + "`junifer collect` command as shown above." + "If you need to retry any step, simply append " + "`--resume-failed` after `--joblog ` option." + ) + run_ext_cmd(name="parallel", cmd=[self.run()]) + else: + logger.info( + "First run the `junifer run` command as shown above. " + "After successful completion of the previous step, run the " + "`junifer collect` command as shown above." + "If you need to retry any step, simply append " + "`--resume-failed` after `--joblog ` option." + ) diff --git a/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py b/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py new file mode 100644 index 000000000..b05beccfb --- /dev/null +++ b/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py @@ -0,0 +1,93 @@ +"""Provide tests for GnuParallelLocalAdapter.""" + +# Authors: Synchon Mandal +# License: AGPL + +import logging +from pathlib import Path +from typing import Dict + +import pytest + +from junifer.api.queue_context import GnuParallelLocalAdapter + + +def test_GnuParallelLocalAdapter_env_error() -> None: + """Test error for invalid env kind.""" + with pytest.raises(ValueError, match="Invalid value for `env.kind`"): + GnuParallelLocalAdapter( + job_name="check_env", + job_dir=Path("."), + yaml_config_path=Path("."), + elements=["sub01"], + env={"kind": "jambalaya"}, + ) + + +def test_GnuParallelLocalAdapter_run() -> None: + """Test HTCondorAdapter run().""" + adapter = GnuParallelLocalAdapter( + job_name="test_run", + job_dir=Path("."), + yaml_config_path=Path("."), + elements=["sub01"], + ) + assert "run" in adapter.run() + + +def test_GnuParallelLocalAdapter_collect() -> None: + """Test HTCondorAdapter collect().""" + adapter = GnuParallelLocalAdapter( + job_name="test_run_collect", + job_dir=Path("."), + yaml_config_path=Path("."), + elements=["sub01"], + ) + assert "collect" in adapter.collect() + + +@pytest.mark.parametrize( + "env", + [ + {"kind": "conda", "name": "junifer"}, + {"kind": "venv", "name": "./junifer"}, + ], +) +def test_GnuParallelLocalAdapter_prepare( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + env: Dict[str, str], +) -> None: + """Test GnuParallelLocalAdapter prepare(). + + Parameters + ---------- + tmp_path : pathlib.Path + The path to the test directory. + monkeypatch : pytest.MonkeyPatch + The pytest.MonkeyPatch object. + caplog : pytest.LogCaptureFixture + The pytest.LogCaptureFixture object. + env : dict + The parametrized Python environment config. + + """ + with monkeypatch.context() as m: + m.chdir(tmp_path) + with caplog.at_level(logging.DEBUG): + adapter = GnuParallelLocalAdapter( + job_name="test_prepare", + job_dir=tmp_path, + yaml_config_path=tmp_path / "config.yaml", + elements=["sub01"], + env=env, + ) + adapter.prepare() + + assert "GNU parallel" in caplog.text + assert f"Copying run_{env['kind']}" in caplog.text + assert "`junifer run`" in caplog.text + assert "`junifer collect`" in caplog.text + + assert adapter._exec_path.stat().st_size != 0 -- 2.52.0 From db4da64bedc45aa1df37d5a0fb7fc39bfa5f7f54 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Mon, 25 Mar 2024 16:16:32 +0100 Subject: [PATCH 03/18] update: integrate GnuParallelLocalAdapter as a queue kind --- junifer/api/functions.py | 14 +++++++++++--- junifer/api/queue_context/__init__.py | 1 + junifer/api/queue_context/htcondor_adapter.py | 5 ++++- 3 files changed, 16 insertions(+), 4 deletions(-) diff --git a/junifer/api/functions.py b/junifer/api/functions.py index 97fff8d06..32eab948f 100644 --- a/junifer/api/functions.py +++ b/junifer/api/functions.py @@ -18,7 +18,7 @@ from ..pipeline.registry import build from ..preprocess.base import BasePreprocessor from ..storage.base import BaseFeatureStorage from ..utils import logger, raise_error -from .queue_context import HTCondorAdapter +from .queue_context import GnuParallelLocalAdapter, HTCondorAdapter from .utils import yaml @@ -220,7 +220,7 @@ def queue( ---------- config : dict The configuration to be used for queueing the job. - kind : {"HTCondor"} + kind : {"HTCondor", "GNUParallelLocal"} The kind of job queue system to use. jobname : str, optional The name of the job (default "junifer_job"). @@ -239,7 +239,7 @@ def queue( if the ``jobdir`` exists and ``overwrite = False``. """ - valid_kind = ["HTCondor"] + valid_kind = ["HTCondor", "GNUParallelLocal"] if kind not in valid_kind: raise_error( f"Invalid value for `kind`: {kind}, " @@ -311,6 +311,14 @@ def queue( elements=elements, **kwargs, # type: ignore ) + elif kind == "GNUParallelLocal": + adapter = GnuParallelLocalAdapter( + job_name=jobname, + job_dir=jobdir, + yaml_config_path=yaml_config, + elements=elements, + **kwargs, # type: ignore + ) adapter.prepare() # type: ignore logger.info("Queue done") diff --git a/junifer/api/queue_context/__init__.py b/junifer/api/queue_context/__init__.py index 3524f4e82..8e1953412 100644 --- a/junifer/api/queue_context/__init__.py +++ b/junifer/api/queue_context/__init__.py @@ -5,3 +5,4 @@ from .queue_context_adapter import QueueContextAdapter from .htcondor_adapter import HTCondorAdapter +from .gnu_parallel_local_adapter import GnuParallelLocalAdapter diff --git a/junifer/api/queue_context/htcondor_adapter.py b/junifer/api/queue_context/htcondor_adapter.py index 63899e13f..9435d0026 100644 --- a/junifer/api/queue_context/htcondor_adapter.py +++ b/junifer/api/queue_context/htcondor_adapter.py @@ -66,7 +66,10 @@ class HTCondorAdapter(QueueContextAdapter): See Also -------- - QueueContextAdapter : The base class for QueueContext. + QueueContextAdapter : + The base class for QueueContext. + GnuParallelLocalAdapter : + The concrete class for queueing via GNU Parallel (local). """ -- 2.52.0 From 4f74f61c2dbdad14cab80dc9b004076a42900c55 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Mon, 25 Mar 2024 16:17:05 +0100 Subject: [PATCH 04/18] update: modify job directory creation and reset to be queue kind specific --- junifer/api/functions.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/junifer/api/functions.py b/junifer/api/functions.py index 32eab948f..7fce5eb73 100644 --- a/junifer/api/functions.py +++ b/junifer/api/functions.py @@ -247,7 +247,7 @@ def queue( ) # Create a folder within the CWD to store the job files / config - jobdir = Path.cwd() / "junifer_jobs" / jobname + jobdir = Path.cwd() / "junifer_jobs" / jobname / kind.lower() logger.info(f"Creating job directory at {jobdir.resolve()!s}") if jobdir.exists(): if not overwrite: @@ -352,6 +352,7 @@ def reset(config: Dict) -> None: Path.cwd() / "junifer_jobs" / (queue.get("jobname") or "junifer_job") + / queue["kind"].lower() ) logger.info(f"Deleting job directory at {job_dir.resolve()!s}") if job_dir.exists(): -- 2.52.0 From 1dbd3d89870848edc64bfbb0878794b74af97066 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Mon, 25 Mar 2024 16:24:24 +0100 Subject: [PATCH 05/18] chore: improve logging for GnuParallelLocalAdapter --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index 2950d7edd..b83040ebd 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -159,7 +159,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter): logger.info( "First the `junifer run` command as shown above will be run. " "After successful completion of the previous step, run the " - "`junifer collect` command as shown above." + "`junifer collect` command as shown above. " "If you need to retry any step, simply append " "`--resume-failed` after `--joblog ` option." ) @@ -168,7 +168,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter): logger.info( "First run the `junifer run` command as shown above. " "After successful completion of the previous step, run the " - "`junifer collect` command as shown above." + "`junifer collect` command as shown above. " "If you need to retry any step, simply append " "`--resume-failed` after `--joblog ` option." ) -- 2.52.0 From 56bdade89a8f5fc19a2db72320268c73fc2fb8ed Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Mon, 25 Mar 2024 16:47:15 +0100 Subject: [PATCH 06/18] update: add --bar for junifer run in GnuParallelLocalAdapter --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index b83040ebd..095a61509 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -110,7 +110,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter): def run(self) -> str: """Return run commands.""" junifer_run_cmd = ( - f"parallel --joblog {self._run_joblog_path} " + f"parallel --bar --joblog {self._run_joblog_path} " f"{self._job_dir.resolve()!s}/{self._executable} " f"{self._arguments} run " f"{self._yaml_config_path.resolve()!s} " -- 2.52.0 From 78482716d7ee8323e55a25b4399ef30ac7160913 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Mon, 25 Mar 2024 16:54:05 +0100 Subject: [PATCH 07/18] update: make parallel halt and kill jobs if single element of junifer run fails --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index 095a61509..8c8bb6746 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -110,7 +110,8 @@ class GnuParallelLocalAdapter(QueueContextAdapter): def run(self) -> str: """Return run commands.""" junifer_run_cmd = ( - f"parallel --bar --joblog {self._run_joblog_path} " + "parallel --bar --halt now,fail=1 " + f"--joblog {self._run_joblog_path} " f"{self._job_dir.resolve()!s}/{self._executable} " f"{self._arguments} run " f"{self._yaml_config_path.resolve()!s} " -- 2.52.0 From 5bf489e3b25180bb486ee99aa7b79c09cdbedd24 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 09:31:10 +0100 Subject: [PATCH 08/18] fix: correct tests in test_functions.py --- junifer/api/tests/test_functions.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/junifer/api/tests/test_functions.py b/junifer/api/tests/test_functions.py index a0043713b..b22e1ee74 100644 --- a/junifer/api/tests/test_functions.py +++ b/junifer/api/tests/test_functions.py @@ -276,7 +276,11 @@ def test_queue_correct_yaml_config( assert "Queue done" in caplog.text generated_config_yaml_path = Path( - tmp_path / "junifer_jobs" / "yaml_config_gen_check" / "config.yaml" + tmp_path + / "junifer_jobs" + / "yaml_config_gen_check" + / "htcondor" + / "config.yaml" ) yaml_config = yaml.load(generated_config_yaml_path) # Check for correct YAML config generation @@ -429,7 +433,11 @@ def test_queue_with_imports( # Check that file is copied assert Path( - tmp_path / "junifer_jobs" / "with_import_check" / "a.py" + tmp_path + / "junifer_jobs" + / "with_import_check" + / "htcondor" + / "a.py" ).is_file() @@ -577,9 +585,9 @@ def test_reset_queue( reset( config={ "storage": storage, - "queue": {"jobname": job_name}, + "queue": {"kind": "HTCondor", "jobname": job_name}, } ) assert not Path(storage["uri"]).exists() - assert not (tmp_path / "junifer_jobs" / job_name).exists() + assert not (tmp_path / "junifer_jobs" / job_name / "htcondor").exists() -- 2.52.0 From ac5481541c3aa2f440829a26681112e2d6567224 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 09:36:09 +0100 Subject: [PATCH 09/18] chore: remove unnecessary parallel invocation for junifer collect in GnuParallelLocalAdapter --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index 8c8bb6746..73bc0ee55 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -68,9 +68,6 @@ class GnuParallelLocalAdapter(QueueContextAdapter): self._submit = submit self._run_joblog_path = self._job_dir / f"{job_name}_run_joblog.txt" - self._collect_joblog_path = ( - self._job_dir / f"{job_name}_collect_joblog.txt" - ) def _check_env(self, env: Optional[Dict[str, str]]) -> None: """Check value of env parameter on init. @@ -130,7 +127,6 @@ class GnuParallelLocalAdapter(QueueContextAdapter): def collect(self) -> str: """Return collect commands.""" return ( - f"parallel --joblog {self._collect_joblog_path} " f"{self._job_dir.resolve()!s}/{self._executable} " f"{self._arguments} collect " f"{self._yaml_config_path.resolve()!s} " -- 2.52.0 From dcae8376f23322adaaa5e4fcc277f4ce40374d35 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 10:45:21 +0100 Subject: [PATCH 10/18] chore: add changelog 306.enh --- docs/changes/newsfragments/306.enh | 1 + 1 file changed, 1 insertion(+) create mode 100644 docs/changes/newsfragments/306.enh diff --git a/docs/changes/newsfragments/306.enh b/docs/changes/newsfragments/306.enh new file mode 100644 index 000000000..d14b06138 --- /dev/null +++ b/docs/changes/newsfragments/306.enh @@ -0,0 +1 @@ +Add support for local ``junifer queue`` via GNU Parallel by `Synchon Mandal`_ -- 2.52.0 From 971a6c47d7d3c250446c5a6dcb3fc45f11f2e084 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 15:46:08 +0100 Subject: [PATCH 11/18] update: revert removed abstract methods for QueueContextAdapter --- .../api/queue_context/queue_context_adapter.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/junifer/api/queue_context/queue_context_adapter.py b/junifer/api/queue_context/queue_context_adapter.py index 0387d5cc3..fe8315b4c 100644 --- a/junifer/api/queue_context/queue_context_adapter.py +++ b/junifer/api/queue_context/queue_context_adapter.py @@ -19,6 +19,14 @@ class QueueContextAdapter(ABC): """ + @abstractmethod + def pre_run(self) -> str: + """Return pre-run commands.""" + raise_error( + msg="Concrete classes need to implement pre_run()", + klass=NotImplementedError, + ) + @abstractmethod def run(self) -> str: """Return run commands.""" @@ -27,6 +35,14 @@ class QueueContextAdapter(ABC): klass=NotImplementedError, ) + @abstractmethod + def pre_collect(self) -> str: + """Return pre-collect commands.""" + raise_error( + msg="Concrete classes need to implement pre_collect()", + klass=NotImplementedError, + ) + @abstractmethod def collect(self) -> str: """Return collect commands.""" -- 2.52.0 From 1f83849b5f083ee4a04f8119ce207e2ec0e61c86 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 15:47:32 +0100 Subject: [PATCH 12/18] update: improve GnuParallelLocalAdapter to handle pre-run, pre-collect and other stuff --- .../gnu_parallel_local_adapter.py | 138 ++++++++++++++---- 1 file changed, 109 insertions(+), 29 deletions(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index 73bc0ee55..8ec320e14 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -4,6 +4,7 @@ # License: AGPL import shutil +import textwrap from pathlib import Path from typing import Dict, List, Optional, Tuple, Union @@ -27,6 +28,10 @@ class GnuParallelLocalAdapter(QueueContextAdapter): The path to the YAML config file. elements : list of str or tuple Element(s) to process. Will be used to index the DataGrabber. + pre_run : str or None, optional + Extra shell commands to source before the run (default None). + pre_collect : str or None, optional + Extra bash commands to source before the collect (default None). env : dict, optional The Python environment configuration. If None, will run without a virtual environment of any kind (default None). @@ -55,19 +60,30 @@ class GnuParallelLocalAdapter(QueueContextAdapter): job_dir: Path, yaml_config_path: Path, elements: List[Union[str, Tuple]], + pre_run: Optional[str] = None, + pre_collect: Optional[str] = None, env: Optional[Dict[str, str]] = None, verbose: str = "info", submit: bool = False, ) -> None: """Initialize the class.""" + self._job_name = job_name self._job_dir = job_dir self._yaml_config_path = yaml_config_path self._elements = elements + self._pre_run = pre_run + self._pre_collect = pre_collect self._check_env(env) self._verbose = verbose self._submit = submit - self._run_joblog_path = self._job_dir / f"{job_name}_run_joblog.txt" + self._log_dir = self._job_dir / "logs" + self._pre_run_path = self._job_dir / "pre_run.sh" + self._pre_collect_path = self._job_dir / "pre_collect.sh" + self._run_path = self._job_dir / f"run_{self._job_name}.sh" + self._collect_path = self._job_dir / f"collect_{self._job_name}.sh" + self._run_joblog_path = self._job_dir / f"run_{self._job_name}_joblog" + self._elements_file_path = self._job_dir / "elements" def _check_env(self, env: Optional[Dict[str, str]]) -> None: """Check value of env parameter on init. @@ -104,29 +120,66 @@ class GnuParallelLocalAdapter(QueueContextAdapter): self._arguments = f"{env['name']} junifer" self._exec_path = self._job_dir / self._executable - def run(self) -> str: - """Return run commands.""" - junifer_run_cmd = ( - "parallel --bar --halt now,fail=1 " - f"--joblog {self._run_joblog_path} " - f"{self._job_dir.resolve()!s}/{self._executable} " - f"{self._arguments} run " - f"{self._yaml_config_path.resolve()!s} " - f"--verbose {self._verbose} " - f"--element :::" - ) + def elements(self) -> str: + """Return elements to run.""" + elements_to_run = [] for element in self._elements: # Stringify elements if tuple for operation str_element = ( ",".join(element) if isinstance(element, tuple) else element ) - junifer_run_cmd += f" '{str_element}'" + elements_to_run.append(str_element) - return junifer_run_cmd + return "\n".join(elements_to_run) + + def pre_run(self) -> str: + """Return pre-run commands.""" + fixed = ( + "#!/usr/bin/env bash\n\n" + "# This script is auto-generated by junifer.\n\n" + "# Force datalad to run in non-interactive mode\n" + "DATALAD_UI_INTERACTIVE=false\n" + ) + var = self._pre_run or "" + return fixed + "\n" + var + + def run(self) -> str: + """Return run commands.""" + return ( + "#!/usr/bin/env bash\n\n" + "# This script is auto-generated by junifer.\n\n" + "# Run pre_run.sh\n" + "sh ./pre_run.sh\n\n" + "# Run `junifer run` using `parallel`\n" + "parallel --bar --resume --resume-failed " + f"--joblog {self._run_joblog_path} " + "--delay 60 " # wait 1 min before next job is spawned + f"--results {self._log_dir}" + "--arg-file elements.txt " + f"{self._job_dir.resolve()!s}/{self._executable} " + f"{self._arguments} run " + f"{self._yaml_config_path.resolve()!s} " + f"--verbose {self._verbose} " + f"--element" + ) + + def pre_collect(self) -> str: + """Return pre-collect commands.""" + fixed = ( + "#!/usr/bin/env bash\n\n" + "# This script is auto-generated by junifer.\n" + ) + var = self._pre_collect or "" + return fixed + "\n" + var def collect(self) -> str: """Return collect commands.""" return ( + "#!/usr/bin/env bash\n\n" + "# This script is auto-generated by junifer.\n\n" + "# Run pre_collect.sh\n" + "sh ./pre_collect.sh\n\n" + "# Run `junifer collect`\n" f"{self._job_dir.resolve()!s}/{self._executable} " f"{self._arguments} collect " f"{self._yaml_config_path.resolve()!s} " @@ -147,25 +200,52 @@ class GnuParallelLocalAdapter(QueueContextAdapter): dst=self._exec_path, ) make_executable(self._exec_path) - # Generate run command - logger.info(f"`junifer run` command:\n{self.run()}") - # Generate collect command - logger.info(f"`junifer collect` command:\n{self.collect()}") + # Create pre run + logger.info( + f"Writing {self._pre_run_path.name} to " + f"{self._job_dir.resolve()!s}" + ) + self._pre_run_path.touch() + self._pre_run_path.write_text(textwrap.dedent(self.pre_run())) + make_executable(self._pre_run_path) + # Create run + logger.info( + f"Writing {self._run_path.name} to " f"{self._job_dir.resolve()!s}" + ) + self._run_path.touch() + self._run_path.write_text(textwrap.dedent(self.run())) + make_executable(self._run_path) + # Create pre collect + logger.info( + f"Writing {self._pre_collect_path.name} to " + f"{self._job_dir.resolve()!s}" + ) + self._pre_collect_path.touch() + self._pre_collect_path.write_text(textwrap.dedent(self.pre_collect())) + make_executable(self._pre_collect_path) + # Create collect + logger.info( + f"Writing {self._collect_path.name} to " + f"{self._job_dir.resolve()!s}" + ) + self._collect_path.touch() + self._collect_path.write_text(textwrap.dedent(self.collect())) + make_executable(self._collect_path) # Submit if required + run_cmd = [f"sh {self._run_path.resolve()!s}"] + collect_cmd = [f"sh {self._collect_path.resolve()!s}"] if self._submit: logger.info( - "First the `junifer run` command as shown above will be run. " - "After successful completion of the previous step, run the " - "`junifer collect` command as shown above. " - "If you need to retry any step, simply append " - "`--resume-failed` after `--joblog ` option." + "Shell scripts created, the following will be run:\n" + f"{run_cmd}\n" + "After successful completion of the previous step, run:\n" + f"{collect_cmd}" ) - run_ext_cmd(name="parallel", cmd=[self.run()]) + run_ext_cmd(name=f"{self._run_path.resolve()!s}", cmd=run_cmd) else: logger.info( - "First run the `junifer run` command as shown above. " - "After successful completion of the previous step, run the " - "`junifer collect` command as shown above. " - "If you need to retry any step, simply append " - "`--resume-failed` after `--joblog ` option." + "Shell scripts created, to start, run:\n" + f"{run_cmd}\n" + "After successful completion of the previous step, run:\n" + f"{collect_cmd}" ) -- 2.52.0 From ae312ce051d28ca6d009ddc92f5e3248fd227e63 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 15:48:20 +0100 Subject: [PATCH 13/18] chore: improve tests for GnuParallelLocalAdapter --- .../tests/test_gnu_parallel_local_adapter.py | 104 +++++++++++++++++- 1 file changed, 101 insertions(+), 3 deletions(-) diff --git a/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py b/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py index b05beccfb..e18a33466 100644 --- a/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py @@ -5,7 +5,7 @@ import logging from pathlib import Path -from typing import Dict +from typing import Dict, List, Optional, Tuple, Union import pytest @@ -24,6 +24,97 @@ def test_GnuParallelLocalAdapter_env_error() -> None: ) +@pytest.mark.parametrize( + "elements, expected_text", + [ + (["sub01", "sub02"], "sub01\nsub02"), + ([("sub01", "ses01"), ("sub02", "ses01")], "sub01,ses01\nsub02,ses01"), + ], +) +def test_GnuParallelLocalAdapter_elements( + elements: List[Union[str, Tuple]], + expected_text: str, +) -> None: + """Test GnuParallelLocalAdapter elements(). + + Parameters + ---------- + elements : list of str or tuple + The parametrized elements. + expected_text : str + The parametrized expected text. + + """ + adapter = GnuParallelLocalAdapter( + job_name="test_elements", + job_dir=Path("."), + yaml_config_path=Path("."), + elements=elements, + ) + assert expected_text in adapter.elements() + + +@pytest.mark.parametrize( + "pre_run, expected_text", + [ + (None, "# Force datalad"), + ("# Check this out\n", "# Check this out"), + ], +) +def test_GnuParallelLocalAdapter_pre_run( + pre_run: Optional[str], expected_text: str +) -> None: + """Test GnuParallelLocalAdapter pre_run(). + + Parameters + ---------- + pre_run : str or None + The parametrized pre run text. + expected_text : str + The parametrized expected text. + + """ + adapter = GnuParallelLocalAdapter( + job_name="test_pre_run", + job_dir=Path("."), + yaml_config_path=Path("."), + elements=["sub01"], + pre_run=pre_run, + ) + assert expected_text in adapter.pre_run() + + +@pytest.mark.parametrize( + "pre_collect, expected_text", + [ + (None, "# This script"), + ("# Check this out\n", "# Check this out"), + ], +) +def test_GnuParallelLocalAdapter_pre_collect( + pre_collect: Optional[str], + expected_text: str, +) -> None: + """Test GnuParallelLocalAdapter pre_collect(). + + Parameters + ---------- + pre_collect : str or None + The parametrized pre collect text. + expected_text : str + The parametrized expected text. + + """ + adapter = GnuParallelLocalAdapter( + job_name="test_pre_collect", + job_dir=Path("."), + yaml_config_path=Path("."), + elements=["sub01"], + pre_collect=pre_collect, + ) + assert expected_text in adapter.pre_collect() + + def test_GnuParallelLocalAdapter_run() -> None: """Test HTCondorAdapter run().""" adapter = GnuParallelLocalAdapter( @@ -87,7 +178,14 @@ def test_GnuParallelLocalAdapter_prepare( assert "GNU parallel" in caplog.text assert f"Copying run_{env['kind']}" in caplog.text - assert "`junifer run`" in caplog.text - assert "`junifer collect`" in caplog.text + assert "Writing pre_run.sh" in caplog.text + assert "Writing run_test_prepare.sh" in caplog.text + assert "Writing pre_collect.sh" in caplog.text + assert "Writing collect_test_prepare.sh" in caplog.text + assert "Shell scripts created" in caplog.text assert adapter._exec_path.stat().st_size != 0 + assert adapter._pre_run_path.stat().st_size != 0 + assert adapter._run_path.stat().st_size != 0 + assert adapter._pre_collect_path.stat().st_size != 0 + assert adapter._collect_path.stat().st_size != 0 -- 2.52.0 From 640939bcd392f44fb75ed53895e1520a1a9eaf71 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 16:06:59 +0100 Subject: [PATCH 14/18] update: use absolute paths for pre_run.sh and pre_collect.sh in GnuParallelLocalAdapter --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index 8ec320e14..87ddaaa3b 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -149,7 +149,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter): "#!/usr/bin/env bash\n\n" "# This script is auto-generated by junifer.\n\n" "# Run pre_run.sh\n" - "sh ./pre_run.sh\n\n" + f"sh {self._pre_run_path.resolve()!s}\n\n" "# Run `junifer run` using `parallel`\n" "parallel --bar --resume --resume-failed " f"--joblog {self._run_joblog_path} " @@ -178,7 +178,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter): "#!/usr/bin/env bash\n\n" "# This script is auto-generated by junifer.\n\n" "# Run pre_collect.sh\n" - "sh ./pre_collect.sh\n\n" + f"sh {self._pre_collect_path.resolve()!s}\n\n" "# Run `junifer collect`\n" f"{self._job_dir.resolve()!s}/{self._executable} " f"{self._arguments} collect " -- 2.52.0 From 5e54e9c4f6c7e96abf6dd35be77468d5a36a5572 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 16:07:44 +0100 Subject: [PATCH 15/18] fix: add missing elements file creation for GnuParallelLocalAdapter --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 7 +++++++ .../queue_context/tests/test_gnu_parallel_local_adapter.py | 1 + 2 files changed, 8 insertions(+) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index 87ddaaa3b..9e9163c65 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -200,6 +200,13 @@ class GnuParallelLocalAdapter(QueueContextAdapter): dst=self._exec_path, ) make_executable(self._exec_path) + # Create elements file + logger.info( + f"Writing {self._elements_file_path.name} to " + f"{self._elements_file_path.resolve()!s}" + ) + self._elements_file_path.touch() + self._elements_file_path.write_text(textwrap.dedent(self.elements())) # Create pre run logger.info( f"Writing {self._pre_run_path.name} to " diff --git a/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py b/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py index e18a33466..7bae705fc 100644 --- a/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py @@ -185,6 +185,7 @@ def test_GnuParallelLocalAdapter_prepare( assert "Shell scripts created" in caplog.text assert adapter._exec_path.stat().st_size != 0 + assert adapter._elements_file_path.stat().st_size != 0 assert adapter._pre_run_path.stat().st_size != 0 assert adapter._run_path.stat().st_size != 0 assert adapter._pre_collect_path.stat().st_size != 0 -- 2.52.0 From afc7969c2719dc78d0fef91449c1a831b7ed1c2f Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 16:08:07 +0100 Subject: [PATCH 16/18] chore: improve command printing to stdout in GnuParallelLocalAdapter --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index 9e9163c65..accc9b3c3 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -239,8 +239,8 @@ class GnuParallelLocalAdapter(QueueContextAdapter): self._collect_path.write_text(textwrap.dedent(self.collect())) make_executable(self._collect_path) # Submit if required - run_cmd = [f"sh {self._run_path.resolve()!s}"] - collect_cmd = [f"sh {self._collect_path.resolve()!s}"] + run_cmd = f"sh {self._run_path.resolve()!s}" + collect_cmd = f"sh {self._collect_path.resolve()!s}" if self._submit: logger.info( "Shell scripts created, the following will be run:\n" @@ -248,7 +248,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter): "After successful completion of the previous step, run:\n" f"{collect_cmd}" ) - run_ext_cmd(name=f"{self._run_path.resolve()!s}", cmd=run_cmd) + run_ext_cmd(name=f"{self._run_path.resolve()!s}", cmd=[run_cmd]) else: logger.info( "Shell scripts created, to start, run:\n" -- 2.52.0 From 44f2e9c7fc61150213bf67ac9a8fdb11b764242f Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Tue, 26 Mar 2024 16:16:42 +0100 Subject: [PATCH 17/18] fix: correct run command for GnuParallelLocalAdapter --- junifer/api/queue_context/gnu_parallel_local_adapter.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/junifer/api/queue_context/gnu_parallel_local_adapter.py b/junifer/api/queue_context/gnu_parallel_local_adapter.py index accc9b3c3..766a21d1c 100644 --- a/junifer/api/queue_context/gnu_parallel_local_adapter.py +++ b/junifer/api/queue_context/gnu_parallel_local_adapter.py @@ -154,8 +154,8 @@ class GnuParallelLocalAdapter(QueueContextAdapter): "parallel --bar --resume --resume-failed " f"--joblog {self._run_joblog_path} " "--delay 60 " # wait 1 min before next job is spawned - f"--results {self._log_dir}" - "--arg-file elements.txt " + f"--results {self._log_dir} " + f"--arg-file {self._elements_file_path.resolve()!s} " f"{self._job_dir.resolve()!s}/{self._executable} " f"{self._arguments} run " f"{self._yaml_config_path.resolve()!s} " -- 2.52.0 From 9ec6835554637123bb6ec7a2ecffdecab05da884 Mon Sep 17 00:00:00 2001 From: Synchon Mandal Date: Wed, 27 Mar 2024 16:28:05 +0100 Subject: [PATCH 18/18] update: revert job asset directory creation according to queueing kind --- junifer/api/functions.py | 3 +-- junifer/api/tests/test_functions.py | 18 +++++------------- 2 files changed, 6 insertions(+), 15 deletions(-) diff --git a/junifer/api/functions.py b/junifer/api/functions.py index 7fce5eb73..32eab948f 100644 --- a/junifer/api/functions.py +++ b/junifer/api/functions.py @@ -247,7 +247,7 @@ def queue( ) # Create a folder within the CWD to store the job files / config - jobdir = Path.cwd() / "junifer_jobs" / jobname / kind.lower() + jobdir = Path.cwd() / "junifer_jobs" / jobname logger.info(f"Creating job directory at {jobdir.resolve()!s}") if jobdir.exists(): if not overwrite: @@ -352,7 +352,6 @@ def reset(config: Dict) -> None: Path.cwd() / "junifer_jobs" / (queue.get("jobname") or "junifer_job") - / queue["kind"].lower() ) logger.info(f"Deleting job directory at {job_dir.resolve()!s}") if job_dir.exists(): diff --git a/junifer/api/tests/test_functions.py b/junifer/api/tests/test_functions.py index b22e1ee74..687ffcded 100644 --- a/junifer/api/tests/test_functions.py +++ b/junifer/api/tests/test_functions.py @@ -276,11 +276,7 @@ def test_queue_correct_yaml_config( assert "Queue done" in caplog.text generated_config_yaml_path = Path( - tmp_path - / "junifer_jobs" - / "yaml_config_gen_check" - / "htcondor" - / "config.yaml" + tmp_path / "junifer_jobs" / "yaml_config_gen_check" / "config.yaml" ) yaml_config = yaml.load(generated_config_yaml_path) # Check for correct YAML config generation @@ -433,11 +429,7 @@ def test_queue_with_imports( # Check that file is copied assert Path( - tmp_path - / "junifer_jobs" - / "with_import_check" - / "htcondor" - / "a.py" + tmp_path / "junifer_jobs" / "with_import_check" / "a.py" ).is_file() @@ -578,16 +570,16 @@ def test_reset_queue( }, "mem": "8G", }, - kind="HTCondor", + kind="GNUParallelLocal", jobname=job_name, ) # Reset operation reset( config={ "storage": storage, - "queue": {"kind": "HTCondor", "jobname": job_name}, + "queue": {"kind": "GNUParallelLocal", "jobname": job_name}, } ) assert not Path(storage["uri"]).exists() - assert not (tmp_path / "junifer_jobs" / job_name / "htcondor").exists() + assert not (tmp_path / "junifer_jobs" / job_name).exists() -- 2.52.0