From ef477d0760079d63e87be3b95ede60559c3d6437 Mon Sep 17 00:00:00 2001 From: "luka.pavageau" Date: Fri, 6 Mar 2026 15:27:19 +0100 Subject: [PATCH 01/19] proposed fix for https://github.com/snakemake/snakemake-executor-plugin-slurm/issues/419 --- snakemake_executor_plugin_slurm/__init__.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index 7d3aaf83..db0616de 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -355,6 +355,18 @@ def __post_init__(self): auto_deploy_default_storage_provider=False, ) +def _select_logdir(workflow) : + """Selects where slurm_logdir should be created""" + logdir = workflow.executor_settings.logdir + # logdir is defined as absolute, keep "as is" + if logdir and str(logdir).startswith("/"): + return Path(logdir) + # logdir is relative, we ensure it stays relative to the wokflow's workdir + elif logdir : + return Path(workflow.workdir_init) / workflow.executor_settings.logdir + # logdir is unset + else : + return Path(".snakemake/slurm_logs").resolve() # Required: # Implementation of your executor @@ -398,11 +410,7 @@ def __post_init__(self, test_mode: bool = False): self._status_query_min_seconds = None self._status_query_max_seconds = 0.0 self._status_query_cycle_rows = [] - self.slurm_logdir = ( - Path(self.workflow.executor_settings.logdir) - if self.workflow.executor_settings.logdir - else Path(".snakemake/slurm_logs").resolve() - ) + self.slurm_logdir = _select_logdir(self.workflow) # Check the environment variable "SNAKEMAKE_SLURM_PARTITIONS", # if set, read the partitions from the given file. Let the CLI # option override this behavior. From 847e264fd560e06783a867db1706c65074f68e58 Mon Sep 17 00:00:00 2001 From: meesters Date: Mon, 8 Jun 2026 15:16:26 +0200 Subject: [PATCH 02/19] feat: added flag for node local stage-in directory --- snakemake_executor_plugin_slurm/__init__.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index 8c6092c9..b53025e2 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -229,6 +229,20 @@ class ExecutorSettings(ExecutorSettingsBase): }, ) + node_local_dir: Optional[str] = field( + default=None, + metadata={ + "help": "A cluster specific node local local global directory. " + "Similar to '.remote_job_local_storage_prefix' of the fs-storage " + "plugin. Used to ensure stage-in when input is flagged for random " + "or mixed access patterns - see https://jgu.to/kwenq ." + "Other than the fs-storage plugin, this function only works to " + "stage-in input files (no stage-out) but supports transferring " + "onto all nodes of the SLURM job. For big files (> 2GB)'ssh' from " + "node to node must be enabled." + }, + ) + init_seconds_before_status_checks: Optional[int] = field( default=40, metadata={ @@ -1468,12 +1482,14 @@ async def check_active_jobs( ) elif status == "PREEMPTED" and not self._preemption_warning: self._preemption_warning = True - self.logger.warning(""" + self.logger.warning( + """ ===== A Job preemption occured! ===== Leave Snakemake running, if possible. Otherwise Snakemake needs to restart this job upon a Snakemake restart. -We leave it to SLURM to resume your job(s)""") +We leave it to SLURM to resume your job(s)""" + ) yield j elif status == "UNKNOWN": # the job probably does not exist anymore, but 'sacct' did not work From 7d15c33c49e2e8f94d764f9303ab0b450f218a4e Mon Sep 17 00:00:00 2001 From: meesters Date: Mon, 15 Jun 2026 11:45:48 +0200 Subject: [PATCH 03/19] feat: escaping optional env vars existing only on remote nodes --- snakemake_executor_plugin_slurm/__init__.py | 8 +++++++- snakemake_executor_plugin_slurm/utils.py | 20 ++++++++++++++++++++ tests/test_node_local_prefix.py | 20 ++++++++++++++++++++ 3 files changed, 47 insertions(+), 1 deletion(-) create mode 100644 tests/test_node_local_prefix.py diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index b53025e2..0a5f2e2d 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -44,6 +44,7 @@ from .utils import ( get_max_array_size, get_job_wildcards, + encode_deferred_envvars, pending_jobs_for_rule, delete_slurm_environment, delete_empty_dirs, @@ -229,7 +230,7 @@ class ExecutorSettings(ExecutorSettingsBase): }, ) - node_local_dir: Optional[str] = field( + node_local_prefix: Optional[str] = field( default=None, metadata={ "help": "A cluster specific node local local global directory. " @@ -653,6 +654,11 @@ def additional_general_args(self): # need to pass if self.workflow.executor_settings.pass_command_as_script: general_args += " --slurm-jobstep-pass-command-as-script" + if self.workflow.executor_settings.node_local_prefix: + ld = self.workflow.executor_settings.node_local_prefix + # self.logger.debug(f"Using node local directory prefix for staging: {ld}") + ld = encode_deferred_envvars(ld) + general_args += f" --slurm-jobstep-node-local-prefix={shlex.quote(ld)}" return general_args def run_jobs(self, jobs: List[JobExecutorInterface]): diff --git a/snakemake_executor_plugin_slurm/utils.py b/snakemake_executor_plugin_slurm/utils.py index 4de29a4b..29d4168a 100644 --- a/snakemake_executor_plugin_slurm/utils.py +++ b/snakemake_executor_plugin_slurm/utils.py @@ -71,6 +71,26 @@ def get_job_wildcards(job: JobExecutorInterface) -> str: return wildcard_str +_DEFERRED_ENVVAR_PATTERN = re.compile( + r"(?[A-Za-z_][A-Za-z0-9_]*)\}|(?P[A-Za-z_][A-Za-z0-9_]*))" +) + + +def encode_deferred_envvars(value: str) -> str: + """Replace shell-style env variables with neutral markers. + + The marker keeps submit-time shell quoting safe. The matching jobstep-side + code can expand ``__ENV_NAME__`` back from ``os.environ`` once the job is + running in its real job context. + """ + + def replace(match: re.Match[str]) -> str: + name = match.group("braced") or match.group("bare") + return f"__ENV_{name}__" + + return _DEFERRED_ENVVAR_PATTERN.sub(replace, value) + + def pending_jobs_for_rule(dag: DAGExecutorInterface, rule_name: str) -> int: """Count jobs of a rule that are currently eligible for scheduling. diff --git a/tests/test_node_local_prefix.py b/tests/test_node_local_prefix.py new file mode 100644 index 00000000..1ed8c510 --- /dev/null +++ b/tests/test_node_local_prefix.py @@ -0,0 +1,20 @@ +from snakemake_executor_plugin_slurm.utils import encode_deferred_envvars + + +def test_encode_deferred_envvars_rewrites_simple_env_vars(): + assert ( + encode_deferred_envvars("/localscratch/$SLURM_JOB_ID") + == "/localscratch/__ENV_SLURM_JOB_ID__" + ) + assert ( + encode_deferred_envvars("/localscratch/${SLURM_JOB_ID}/run") + == "/localscratch/__ENV_SLURM_JOB_ID__/run" + ) + + +def test_encode_deferred_envvars_leaves_literal_text_alone(): + assert encode_deferred_envvars("/localscratch/job-1") == "/localscratch/job-1" + assert ( + encode_deferred_envvars(r"/localscratch/\$SLURM_JOB_ID") + == r"/localscratch/\$SLURM_JOB_ID" + ) From cc6b6c579da15a122d17cc5d1cdc2e8fe4f0d6ec Mon Sep 17 00:00:00 2001 From: meesters Date: Mon, 15 Jun 2026 11:48:58 +0200 Subject: [PATCH 04/19] fix: formatting --- snakemake_executor_plugin_slurm/utils.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/snakemake_executor_plugin_slurm/utils.py b/snakemake_executor_plugin_slurm/utils.py index 29d4168a..66585fc7 100644 --- a/snakemake_executor_plugin_slurm/utils.py +++ b/snakemake_executor_plugin_slurm/utils.py @@ -72,7 +72,10 @@ def get_job_wildcards(job: JobExecutorInterface) -> str: _DEFERRED_ENVVAR_PATTERN = re.compile( - r"(?[A-Za-z_][A-Za-z0-9_]*)\}|(?P[A-Za-z_][A-Za-z0-9_]*))" + r"(?[A-Za-z_][A-Za-z0-9_]*)\}" + r"|(?P[A-Za-z_][A-Za-z0-9_]*)" + r")" ) From 1bf35b2d8c63b337ec76003bbe91e05b32137972 Mon Sep 17 00:00:00 2001 From: meesters Date: Mon, 15 Jun 2026 11:53:04 +0200 Subject: [PATCH 05/19] fix: black-stable formatting? --- snakemake_executor_plugin_slurm/utils.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/snakemake_executor_plugin_slurm/utils.py b/snakemake_executor_plugin_slurm/utils.py index 66585fc7..f5425abc 100644 --- a/snakemake_executor_plugin_slurm/utils.py +++ b/snakemake_executor_plugin_slurm/utils.py @@ -72,10 +72,14 @@ def get_job_wildcards(job: JobExecutorInterface) -> str: _DEFERRED_ENVVAR_PATTERN = re.compile( - r"(?[A-Za-z_][A-Za-z0-9_]*)\}" - r"|(?P[A-Za-z_][A-Za-z0-9_]*)" - r")" + r""" + (?[A-Za-z_][A-Za-z0-9_]*)\} + |(?P[A-Za-z_][A-Za-z0-9_]*) + ) + """, + re.VERBOSE, ) From 69d62af6d212a777448b651694297fa2215a4293 Mon Sep 17 00:00:00 2001 From: meesters Date: Mon, 15 Jun 2026 14:24:10 +0200 Subject: [PATCH 06/19] feat: added test files --- tests/test_stagein.py | 123 +++++++++++++++++++++++ tests/testcases/stagein_sbcast/Snakefile | 25 +++++ tests/testcases/stagein_ssh/Snakefile | 33 ++++++ 3 files changed, 181 insertions(+) create mode 100644 tests/test_stagein.py create mode 100644 tests/testcases/stagein_sbcast/Snakefile create mode 100644 tests/testcases/stagein_ssh/Snakefile diff --git a/tests/test_stagein.py b/tests/test_stagein.py new file mode 100644 index 00000000..864c8a2f --- /dev/null +++ b/tests/test_stagein.py @@ -0,0 +1,123 @@ +from pathlib import Path +from typing import Optional + +import snakemake.common.tests +import snakemake.settings.types as settings +from snakemake import api +from snakemake.settings.types import ConfigSettings +from snakemake_interface_executor_plugins.settings import ExecutorSettingsBase + +from snakemake_executor_plugin_slurm import ExecutorSettings + + +class LocalStageinTestcasesBase(snakemake.common.tests.TestWorkflowsLocalStorageBase): + """Resolve stage-in testcases from this plugin's tests/testcases directory.""" + + def get_config_settings(self) -> Optional[ConfigSettings]: + return None + + def run_workflow(self, test_name, tmp_path, deployment_method=frozenset()): + test_path = Path(__file__).parent / "testcases" / test_name + if not test_path.exists(): + return super().run_workflow(test_name, tmp_path, deployment_method) + + if self.omit_tmp: + tmp_path = test_path + else: + tmp_path = Path(tmp_path) / test_name + self._copy_test_files(test_path, tmp_path) + + resource_settings = self.get_resource_settings() + + if self._common_settings().local_exec: + resource_settings.cores = 3 + resource_settings.nodes = None + else: + resource_settings.cores = 1 + resource_settings.nodes = 3 + + with api.SnakemakeApi( + settings.OutputSettings( + verbose=True, + show_failed_logs=True, + ), + ) as snakemake_api: + workflow_api = snakemake_api.workflow( + config_settings=self.get_config_settings(), + resource_settings=resource_settings, + storage_settings=settings.StorageSettings( + default_storage_provider=self.get_default_storage_provider(), + default_storage_prefix=self.get_default_storage_prefix(), + shared_fs_usage=( + settings.SharedFSUsage.all() + if self.get_assume_shared_fs() + else frozenset() + ), + ), + deployment_settings=self.get_deployment_settings(deployment_method), + storage_provider_settings=self.get_default_storage_provider_settings(), + workdir=Path(tmp_path), + snakefile=tmp_path / "Snakefile", + ) + + dag_api = workflow_api.dag() + dag_api.execute_workflow( + executor=self.get_executor(), + executor_settings=self.get_executor_settings(), + execution_settings=settings.ExecutionSettings( + latency_wait=self.latency_wait, + ), + remote_execution_settings=self.get_remote_execution_settings(), + ) + + +class TestStageInSbcast(LocalStageinTestcasesBase): + """Integration test for sbcast stage-in with a small synthetic input.""" + + __test__ = True + + def get_executor(self) -> str: + return "slurm" + + def get_executor_settings(self) -> Optional[ExecutorSettingsBase]: + return ExecutorSettings( + init_seconds_before_status_checks=2, + node_local_prefix="/tmp/snakemake-stagein", + ) + + def test_stagein_sbcast(self, tmp_path): + self.run_workflow("stagein_sbcast", tmp_path) + + run_dir = Path(tmp_path) / "stagein_sbcast" + count_file = run_dir / "counted" / "count.txt" + sbcast_log = run_dir / "log" / "sbcast.log" + + assert count_file.exists() + assert sbcast_log.exists() + assert count_file.read_text().strip().startswith("3 ") + + +class TestStageInSSH(LocalStageinTestcasesBase): + """Optional integration test for SSH stage-in with a sparse large input.""" + + __test__ = True + + def get_executor(self) -> str: + return "slurm" + + def get_executor_settings(self) -> Optional[ExecutorSettingsBase]: + return ExecutorSettings( + init_seconds_before_status_checks=2, + node_local_prefix="/tmp/snakemake-stagein", + ) + + def test_stagein_ssh(self, tmp_path): + self.run_workflow("stagein_ssh", tmp_path) + + run_dir = Path(tmp_path) / "stagein_ssh" + size_file = run_dir / "measured" / "size.txt" + ssh_log = run_dir / "log" / "ssh.log" + + assert size_file.exists() + assert ssh_log.exists() + assert "data/big" in size_file.read_text() diff --git a/tests/testcases/stagein_sbcast/Snakefile b/tests/testcases/stagein_sbcast/Snakefile new file mode 100644 index 00000000..36a41678 --- /dev/null +++ b/tests/testcases/stagein_sbcast/Snakefile @@ -0,0 +1,25 @@ +inputflags: + access.sequential + + +rule all: + input: + "counted/count.txt" + + +rule make_words: + output: + "data/words" + shell: + "printf 'alpha\nbeta\ngamma\n' > {output}" + + +rule use_sbcast: + input: + access.random("data/words") + output: + "counted/count.txt" + log: + "log/sbcast.log" + shell: + "wc -l {input} > {output}" diff --git a/tests/testcases/stagein_ssh/Snakefile b/tests/testcases/stagein_ssh/Snakefile new file mode 100644 index 00000000..e5432263 --- /dev/null +++ b/tests/testcases/stagein_ssh/Snakefile @@ -0,0 +1,33 @@ +import os + + +inputflags: + access.sequential + + +rule all: + input: + "measured/size.txt" + + +rule make_big_sparse: + output: + "data/big" + run: + size_bytes = int( + os.environ.get("STAGEIN_TEST_BIG_BYTES", str(2 * 1024 * 1024 * 1024 + 1)) + ) + with open(output[0], "wb") as f: + f.seek(size_bytes - 1) + f.write(b"\0") + + +rule use_ssh: + input: + access.multi("data/big") + output: + "measured/size.txt" + log: + "log/ssh.log" + shell: + "ls -lh {input} > {output}" From 181e9a68beb57c88895feb1c1313edfbf8946cc7 Mon Sep 17 00:00:00 2001 From: meesters Date: Mon, 15 Jun 2026 14:44:32 +0200 Subject: [PATCH 07/19] test: added debug info to the ci.yml --- .github/workflows/ci.yml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f5954074..55d03549 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,6 +24,11 @@ jobs: run: | poetry install --sync + - name: Debug black environment + run: | + poetry run black --version + poetry run black --check --diff . + - name: Check formatting run: poetry run black --check . From 698d0d0acf8b52b1c1aa8fabf61da24fd6d96666 Mon Sep 17 00:00:00 2001 From: meesters Date: Mon, 15 Jun 2026 14:49:50 +0200 Subject: [PATCH 08/19] fix: formatting --- snakemake_executor_plugin_slurm/__init__.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index 0a5f2e2d..ca3ebe90 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -1488,14 +1488,12 @@ async def check_active_jobs( ) elif status == "PREEMPTED" and not self._preemption_warning: self._preemption_warning = True - self.logger.warning( - """ + self.logger.warning(""" ===== A Job preemption occured! ===== Leave Snakemake running, if possible. Otherwise Snakemake needs to restart this job upon a Snakemake restart. -We leave it to SLURM to resume your job(s)""" - ) +We leave it to SLURM to resume your job(s)""") yield j elif status == "UNKNOWN": # the job probably does not exist anymore, but 'sacct' did not work From 6cb2b6d472a47b19d72af076e43e3365fdbd6df3 Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 10:47:29 +0200 Subject: [PATCH 09/19] docs: added doc section on auto stage-in --- docs/further.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/docs/further.md b/docs/further.md index b9629518..d77bb7df 100644 --- a/docs/further.md +++ b/docs/further.md @@ -190,6 +190,30 @@ The scoring algorithm calculates a score by summing the ratios of requested reso If no suitable partition is found based on the job's resource requirements, the plugin falls back to the default SLURM behavior, which typically uses the cluster's default partition or any partition specified explicitly in the job's resources. +#### Automatic Stage-In + +In HPC we stage-in files from global parallel file system onto node local storage. This serves a number of purposes: + +- avoiding random I/O which throttles performance tremendously and may occur, for instance, when several SLURM jobs with the same program attempt to use the same reference files (e.g. with alignment programs) +- minimizing latency for repeated file access, since local storage (e.g., SSD or NVMe) typically offers much lower access times than network-attached parallel filesystems. +- improving fault tolerance and resilience during job execution — if the parallel filesystem experiences transient issues, locally staged files remain accessible. + +to name a few. + +To enable automatic stage-in of files a workflow rule needs to designate either the [`random` access file pattern or the `multi` access pattern](https://snakemake.readthedocs.io/en/stable/snakefiles/storage.html#access-pattern-annotation). Furthermore a flag `--slurm-node-local-prefix` can be used to indicate a node local directory available for usage within a SLURM job. As always with Snakemake, we may use the flag on the command line and in a configuration file, e.g.: + +```yaml +slurm-node-local-prefix: /localscratch/$SLURM_JOB_ID +``` + +Environment variables may just be written as literals - even if the environment variable in only set within a cluster job. + +If the automatic stage-in process is not desirable, `--slurm-suppress-auto-stagein` will turn it off. + +Stage-in processes will be performed on all nodes of a job. Using `sbcast` for files <= 4GB and `scp` for files > 4GB. Note that stage-in processes with `scp` cross-node `ssh` needs to be working on your cluster. + +.. note:: Currently, there is no distinction between different node local files systems, e.g. `/tmpfs`, `nvme`s on SSD or dedicated ramdisks. Only one node local directory prefix is allowed. + #### Standard Resources From ad51e21d2626a1cf4fffb28f27ac11f22c8b51a4 Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 10:52:14 +0200 Subject: [PATCH 10/19] fix: formatting --- snakemake_executor_plugin_slurm/utils.py | 26 ++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/snakemake_executor_plugin_slurm/utils.py b/snakemake_executor_plugin_slurm/utils.py index f5425abc..43e3690d 100644 --- a/snakemake_executor_plugin_slurm/utils.py +++ b/snakemake_executor_plugin_slurm/utils.py @@ -433,3 +433,29 @@ def set_gres_string(job: JobExecutorInterface) -> str: gres += f" --gpus={gpu_string}" return gres + + +def get_file_size(job: JobExecutorInterface) -> int: + """ + Function to get the file size of the input files for a job. This is used + to determine the resources needed for the job. + + Args: + job: The JobExecutorInterface instance representing the job + Returns: + The total file size of the input files for the job, in GB. + """ + total_size_bytes = 0 + for input_file in job.input_files: + if os.path.isfile(input_file): + total_size_bytes += os.path.getsize(input_file) + elif os.path.isdir(input_file): + for dirpath, _, filenames in os.walk(input_file): + for filename in filenames: + file_path = os.path.join(dirpath, filename) + if os.path.isfile(file_path): + total_size_bytes += os.path.getsize(file_path) + + # Convert bytes to GB + total_size_gb = total_size_bytes / (1024**3) + return round_half_up(total_size_gb) # implicit conversion to int From 0336581fdab6a12b5e9682745c3880e3d5be34d9 Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 10:53:02 +0200 Subject: [PATCH 11/19] feat: allowing to disable auto stage-in and added a warning for big files --- snakemake_executor_plugin_slurm/__init__.py | 57 +++++++++++++++++++-- 1 file changed, 52 insertions(+), 5 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index ca3ebe90..5f5bc5a2 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -34,6 +34,10 @@ from snakemake_interface_executor_plugins.jobs import ( JobExecutorInterface, ) +from snakemake.io.flags.access_patterns import ( + AccessPattern, + STORE_KEY, +) from snakemake_interface_common.exceptions import WorkflowError from .accounts import ( @@ -49,6 +53,7 @@ delete_slurm_environment, delete_empty_dirs, set_gres_string, + get_file_size, ) from .job_status_query import ( get_min_job_age, @@ -243,6 +248,18 @@ class ExecutorSettings(ExecutorSettingsBase): "node to node must be enabled." }, ) + suppress_auto_stagein: bool = field( + default=False, + metadata={ + "help": "By default, when node_local_prefix is set, Snakemake will " + "automatically stage in input files with random or mixed access " + "patterns to the node local directory. Setting this flag will " + "suppress this behavior, so that no automatic staging will be " + "performed.", + "env_var": False, + "required": False, + }, + ) init_seconds_before_status_checks: Optional[int] = field( default=40, @@ -651,12 +668,15 @@ def additional_general_args(self): passed to `exec_job`. """ general_args = "--executor slurm-jobstep --jobs 1" - # need to pass + # switch for passing a command as script vs using `sbatch --wrap` if self.workflow.executor_settings.pass_command_as_script: general_args += " --slurm-jobstep-pass-command-as-script" - if self.workflow.executor_settings.node_local_prefix: + # attempt auto stage-in if not suppressed and node_local_prefix is set + if ( + self.workflow.executor_settings.node_local_prefix + and not self.workflow.executor_settings.suppress_auto_stagein + ): ld = self.workflow.executor_settings.node_local_prefix - # self.logger.debug(f"Using node local directory prefix for staging: {ld}") ld = encode_deferred_envvars(ld) general_args += f" --slurm-jobstep-node-local-prefix={shlex.quote(ld)}" return general_args @@ -668,6 +688,31 @@ def run_jobs(self, jobs: List[JobExecutorInterface]): - `run_array_jobs` for array job submission, or - `run_pool_jobs` for pool job submission (to be implemented in the future). """ + # check whether current jobs are using random or multi access patters + # and node_local_prefix is set + if ( + not self.workflow.executor_settings.suppress_auto_stagein + and self.workflow.executor_settings.node_local_prefix + ): + for job in jobs: + for inp in job.input: + if isinstance( + inp.flags.get(STORE_KEY), AccessPattern + ) and inp.flags[STORE_KEY] in { + AccessPattern.RANDOM, + AccessPattern.MIXED, + }: + size = get_file_size(inp.path) + if size is not None and size > 100: + self.logger.warning( + f"Job '{job.name}' has input file '{inp.path}' with " + f"random or mixed access pattern and size {size} bytes. " + "Snakemake will attempt to stage in this file to the node " + "local directory specified by " + f"{self.workflow.executor_settings.node_local_prefix}. " + "However, for files of this size the process might be slow." + ) + if self._main_event_loop is None: try: self._main_event_loop = asyncio.get_running_loop() @@ -1488,12 +1533,14 @@ async def check_active_jobs( ) elif status == "PREEMPTED" and not self._preemption_warning: self._preemption_warning = True - self.logger.warning(""" + self.logger.warning( + """ ===== A Job preemption occured! ===== Leave Snakemake running, if possible. Otherwise Snakemake needs to restart this job upon a Snakemake restart. -We leave it to SLURM to resume your job(s)""") +We leave it to SLURM to resume your job(s)""" + ) yield j elif status == "UNKNOWN": # the job probably does not exist anymore, but 'sacct' did not work From be09f3399c6c604f2ebea7509dc240c0d85210ee Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 11:04:14 +0200 Subject: [PATCH 12/19] fix: formatting --- snakemake_executor_plugin_slurm/__init__.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index 5f5bc5a2..1a340ad1 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -706,11 +706,12 @@ def run_jobs(self, jobs: List[JobExecutorInterface]): if size is not None and size > 100: self.logger.warning( f"Job '{job.name}' has input file '{inp.path}' with " - f"random or mixed access pattern and size {size} bytes. " - "Snakemake will attempt to stage in this file to the node " - "local directory specified by " + f"random or mixed access pattern and size {size} " + "bytes. Snakemake will attempt to stage in this file " + "to the node local directory specified by " f"{self.workflow.executor_settings.node_local_prefix}. " - "However, for files of this size the process might be slow." + "However, for files of this size the process might be " + "slow." ) if self._main_event_loop is None: From 037495e004e8cfbdf354b860f767d5db21a79297 Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 11:23:48 +0200 Subject: [PATCH 13/19] fix: formatting --- snakemake_executor_plugin_slurm/__init__.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index 1a340ad1..7962952f 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -1534,14 +1534,12 @@ async def check_active_jobs( ) elif status == "PREEMPTED" and not self._preemption_warning: self._preemption_warning = True - self.logger.warning( - """ + self.logger.warning(""" ===== A Job preemption occured! ===== Leave Snakemake running, if possible. Otherwise Snakemake needs to restart this job upon a Snakemake restart. -We leave it to SLURM to resume your job(s)""" - ) +We leave it to SLURM to resume your job(s)""") yield j elif status == "UNKNOWN": # the job probably does not exist anymore, but 'sacct' did not work From 6f533e1cbcf8b25a61f77e97d658c841496cad8a Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 11:57:13 +0200 Subject: [PATCH 14/19] fix: adjusted testcase threshold to 4GB as is the current code standard --- tests/testcases/stagein_ssh/Snakefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/testcases/stagein_ssh/Snakefile b/tests/testcases/stagein_ssh/Snakefile index e5432263..769774e3 100644 --- a/tests/testcases/stagein_ssh/Snakefile +++ b/tests/testcases/stagein_ssh/Snakefile @@ -15,7 +15,7 @@ rule make_big_sparse: "data/big" run: size_bytes = int( - os.environ.get("STAGEIN_TEST_BIG_BYTES", str(2 * 1024 * 1024 * 1024 + 1)) + os.environ.get("STAGEIN_TEST_BIG_BYTES", str(4 * 1024 * 1024 * 1024 + 1)) ) with open(output[0], "wb") as f: f.seek(size_bytes - 1) From 99e3c6965e9912310ae36979fcc600532cb7c97f Mon Sep 17 00:00:00 2001 From: Christian Meesters Date: Wed, 17 Jun 2026 11:57:39 +0200 Subject: [PATCH 15/19] Update .github/workflows/ci.yml Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 55d03549..a8143cf8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,6 +25,7 @@ jobs: poetry install --sync - name: Debug black environment + continue-on-error: true run: | poetry run black --version poetry run black --check --diff . From 024803e70f817a888ab5ed6cc7ca060d7902fcc4 Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 13:31:13 +0200 Subject: [PATCH 16/19] fix: potetial issue with checking file size --- snakemake_executor_plugin_slurm/__init__.py | 42 +++++++++++---------- 1 file changed, 22 insertions(+), 20 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index 7962952f..0af74650 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -695,24 +695,24 @@ def run_jobs(self, jobs: List[JobExecutorInterface]): and self.workflow.executor_settings.node_local_prefix ): for job in jobs: - for inp in job.input: - if isinstance( - inp.flags.get(STORE_KEY), AccessPattern - ) and inp.flags[STORE_KEY] in { - AccessPattern.RANDOM, - AccessPattern.MIXED, - }: - size = get_file_size(inp.path) - if size is not None and size > 100: - self.logger.warning( - f"Job '{job.name}' has input file '{inp.path}' with " - f"random or mixed access pattern and size {size} " - "bytes. Snakemake will attempt to stage in this file " - "to the node local directory specified by " - f"{self.workflow.executor_settings.node_local_prefix}. " - "However, for files of this size the process might be " - "slow." - ) + has_random_or_mixed = any( + isinstance(inp.flags.get(STORE_KEY), AccessPattern) + and inp.flags[STORE_KEY] + in {AccessPattern.RANDOM, AccessPattern.MIXED} + for inp in job.input + ) + if has_random_or_mixed: + size = get_file_size(inp.path) + if size is not None and size > 100: + self.logger.warning( + f"Job '{job.name}' has input file '{inp.path}' with " + f"random or mixed access pattern and size {size} " + "bytes. Snakemake will attempt to stage in this file " + "to the node local directory specified by " + f"{self.workflow.executor_settings.node_local_prefix}. " + "However, for files of this size the process might be " + "slow." + ) if self._main_event_loop is None: try: @@ -1534,12 +1534,14 @@ async def check_active_jobs( ) elif status == "PREEMPTED" and not self._preemption_warning: self._preemption_warning = True - self.logger.warning(""" + self.logger.warning( + """ ===== A Job preemption occured! ===== Leave Snakemake running, if possible. Otherwise Snakemake needs to restart this job upon a Snakemake restart. -We leave it to SLURM to resume your job(s)""") +We leave it to SLURM to resume your job(s)""" + ) yield j elif status == "UNKNOWN": # the job probably does not exist anymore, but 'sacct' did not work From 6869eb5cda6b611df1d87aefa17868afb7156d0c Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 14:31:40 +0200 Subject: [PATCH 17/19] fix: gnarf - syntax error by ommitted line fixed --- snakemake_executor_plugin_slurm/__init__.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index 0af74650..ef5cb655 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -695,12 +695,13 @@ def run_jobs(self, jobs: List[JobExecutorInterface]): and self.workflow.executor_settings.node_local_prefix ): for job in jobs: - has_random_or_mixed = any( - isinstance(inp.flags.get(STORE_KEY), AccessPattern) - and inp.flags[STORE_KEY] - in {AccessPattern.RANDOM, AccessPattern.MIXED} - for inp in job.input - ) + for inp in job.input: + has_random_or_mixed = any( + isinstance(inp.flags.get(STORE_KEY), AccessPattern) + and inp.flags[STORE_KEY] + in {AccessPattern.RANDOM, AccessPattern.MIXED} + for inp in job.input + ) if has_random_or_mixed: size = get_file_size(inp.path) if size is not None and size > 100: From 145d545a13e5d655709be0f81a3d7f9f583ccd9b Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 15:01:28 +0200 Subject: [PATCH 18/19] fix: requiring one python target version for black --- .github/workflows/ci.yml | 4 ++-- pyproject.toml | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a8143cf8..a47bac2b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,10 +28,10 @@ jobs: continue-on-error: true run: | poetry run black --version - poetry run black --check --diff . + poetry run black --check --diff --target-version py311 . - name: Check formatting - run: poetry run black --check . + run: poetry run black --check --target-version py311 . linting: runs-on: ubuntu-latest diff --git a/pyproject.toml b/pyproject.toml index 4dc84588..163eedfe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ omit = [".*", "*/site-packages/*", "Snakefile"] [tool.black] line-length = 88 +target-version = ["py311"] extend-exclude = 'tests/test_scontrol_parsing\.py' [tool.pytest.ini_options] From 411d053ea7b8c566d9daaec4bd101053363ca45d Mon Sep 17 00:00:00 2001 From: meesters Date: Wed, 17 Jun 2026 15:08:41 +0200 Subject: [PATCH 19/19] fix: formatting --- snakemake_executor_plugin_slurm/__init__.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py index ef5cb655..c7b5b6e1 100644 --- a/snakemake_executor_plugin_slurm/__init__.py +++ b/snakemake_executor_plugin_slurm/__init__.py @@ -1535,14 +1535,12 @@ async def check_active_jobs( ) elif status == "PREEMPTED" and not self._preemption_warning: self._preemption_warning = True - self.logger.warning( - """ + self.logger.warning(""" ===== A Job preemption occured! ===== Leave Snakemake running, if possible. Otherwise Snakemake needs to restart this job upon a Snakemake restart. -We leave it to SLURM to resume your job(s)""" - ) +We leave it to SLURM to resume your job(s)""") yield j elif status == "UNKNOWN": # the job probably does not exist anymore, but 'sacct' did not work