SLURM with Flux#
executorlib extends the Executor interface from the Python standard library’s concurrent.futures to distribute Python functions as jobs on an HPC cluster. This notebook shows how to run a Flux instance nested inside a SLURM allocation, see the Flux section of the HPC Cluster Executor documentation for background.
Based on the cmti and cmmg clusters hosted at the MPCDF for the MPI for Sustainable Materials.
Unlike the previous notebook, which submits each Python function directly as a SLURM job step via srun, this notebook starts a Flux instance inside the SLURM allocation using flux start. executorlib’s FluxJobExecutor then submits tasks to this Flux instance. Flux acts as a lightweight scheduler nested inside the outer SLURM job, which is particularly useful for running many small tasks without the overhead of separate SLURM job steps.
import executorlib
executorlib.__version__
'0.0.1'
executorlib.__path__
['/u/janj/projects/executorlib/src/executorlib']
Submit Python Function to SLURM#
This is just like using sbatch to submit shell scripts. See the SlurmClusterExecutor documentation for the full set of options.
Compared to the plain SLURM submission template, this template requests --cpus-per-task={{cores}} instead of --ntasks={{cores}}, and wraps the command with srun --cpus-per-task={{cores}} flux start {{command}}. flux start boots a Flux instance that spans the resources allocated by SLURM, and {{command}} (the executorlib worker) is then launched inside that Flux instance rather than directly via srun.
submission_template = """\
#!/bin/bash
#SBATCH --output=time.out
#SBATCH --job-name={{job_name}}
#SBATCH --chdir={{working_directory}}
#SBATCH --get-user-env=L
#SBATCH --partition={{partition}}
{%- if run_time_max %}
#SBATCH --time={{ [1, run_time_max // 60]|max }}
{%- endif %}
{%- if dependency %}
#SBATCH --dependency=afterok:{{ dependency | join(',') }}
{%- endif %}
{%- if memory_max %}
#SBATCH --mem={{memory_max}}G
{%- endif %}
#SBATCH --cpus-per-task={{cores}}
srun --cpus-per-task={{cores}} flux start {{command}}
"""
from executorlib import get_cache_data
import subprocess
get_slurm_parallel_env is the MPI-parallel function submitted to the nested Flux instance. Each MPI rank collects the FLUX_* environment variables set by Flux (rather than the SLURM_* variables used in the previous notebook), then comm.gather collects all per-rank dictionaries onto rank 0, which returns the combined list.
def get_slurm_parallel_env():
import os
from mpi4py import MPI
comm = MPI.COMM_WORLD
size = comm.Get_size()
rank = comm.Get_rank()
env_dict = os.environ
data = {
k:v for k, v in os.environ.items() if "FLUX_" in k
}
data = comm.gather(data, root=0)
return data
Nested Executors#
Finally, to gain the most of the flexibility executorlib offers the option to nest executors within each other. See the HPC Job Executor documentation for more details on FluxJobExecutor and running nested workloads inside an existing Flux instance.
def get_nested_env():
with executorlib.FluxJobExecutor(max_workers=4, pmi_mode="pmix") as exe:
future_lst = []
for i in range(4):
future_lst.append(exe.submit(
get_slurm_parallel_env,
resource_dict={"cores": 2, "threads_per_core": 1},
))
return [f.result() for f in future_lst]
get_nested_env runs inside the Flux instance started by flux start and itself creates a FluxJobExecutor. Unlike SlurmJobExecutor, which dispatches steps via srun into the SLURM allocation, FluxJobExecutor submits jobs to the already-running Flux instance. With max_workers=4, up to four get_slurm_parallel_env calls run concurrently, each requesting cores=2 (two MPI ranks) from the Flux instance’s resource pool.
from time import sleep
cache_dir = "executorlib_cache"
We submit get_nested_env to a single SLURM job via SlurmClusterExecutor. cores=1, threads_per_core=4 requests --cpus-per-task=4, so the outer job is allocated four CPUs on one node. Inside that allocation, flux start boots a Flux instance spanning those four CPUs, and get_nested_env then submits four two-core tasks to it via FluxJobExecutor.
with executorlib.SlurmClusterExecutor(pmi_mode="pmix") as exe:
f1 = exe.submit(
get_nested_env,
resource_dict={
"submission_template": submission_template,
"run_time_max": 180, # in seconds
# "partition": "s.cmfe",
"partition": "s.cmmg",
"cores": 1,
"threads_per_core": 4,
})
sleep(10)
df = get_cache_data(cache_dir)
for d in df:
if "get_nested_env" in str(d["function"]):
print(d['queue_id'])
shell_output = subprocess.check_output(["sacct", "-j", str(d['queue_id'])], shell=False, universal_newlines=True)
result = f1.result()
print(result)
/u/janj/projects/executorlib/src/executorlib/task_scheduler/base.py:156: UserWarning: The following keys are not recognized and cannot be validated: ['partition']
self._validator(resource_dict=resource_dict)
20339083
[[{'FLUX_EXEC_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/libexec/flux/cmd', 'FLUX_CONNECTOR_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/connectors', 'FLUX_MODULE_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/modules', 'FLUX_JOB_TMPDIR': '/tmp/flux-asjoSr/jobtmp-0-ƒoYiTAw', 'FLUX_TERMINUS_SESSION': '0', 'FLUX_TASK_LOCAL_ID': '0', 'FLUX_TASK_RANK': '0', 'FLUX_JOB_SIZE': '2', 'FLUX_JOB_NNODES': '1', 'FLUX_JOB_ID': 'ƒoYiTAw', 'FLUX_URI': 'local:///tmp/flux-asjoSr/local-0', 'FLUX_KVS_NAMESPACE': 'job-30551310336'}], [{'FLUX_EXEC_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/libexec/flux/cmd', 'FLUX_CONNECTOR_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/connectors', 'FLUX_MODULE_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/modules', 'FLUX_JOB_TMPDIR': '/tmp/flux-asjoSr/jobtmp-0-ƒoYiTAx', 'FLUX_TERMINUS_SESSION': '0', 'FLUX_TASK_LOCAL_ID': '1', 'FLUX_TASK_RANK': '1', 'FLUX_JOB_SIZE': '2', 'FLUX_JOB_NNODES': '1', 'FLUX_JOB_ID': 'ƒoYiTAx', 'FLUX_URI': 'local:///tmp/flux-asjoSr/local-0', 'FLUX_KVS_NAMESPACE': 'job-30551310337'}], [{'FLUX_EXEC_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/libexec/flux/cmd', 'FLUX_CONNECTOR_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/connectors', 'FLUX_MODULE_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/modules', 'FLUX_JOB_TMPDIR': '/tmp/flux-asjoSr/jobtmp-0-ƒoYiTAy', 'FLUX_TERMINUS_SESSION': '0', 'FLUX_TASK_LOCAL_ID': '1', 'FLUX_TASK_RANK': '1', 'FLUX_JOB_SIZE': '2', 'FLUX_JOB_NNODES': '1', 'FLUX_JOB_ID': 'ƒoYiTAy', 'FLUX_URI': 'local:///tmp/flux-asjoSr/local-0', 'FLUX_KVS_NAMESPACE': 'job-30551310338'}], [{'FLUX_EXEC_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/libexec/flux/cmd', 'FLUX_CONNECTOR_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/connectors', 'FLUX_MODULE_PATH': '/cmmc/ptmp/pyironhb/mambaforge/envs/pyiron_mpie_cmti_2026-06-08/lib/flux/modules', 'FLUX_JOB_TMPDIR': '/tmp/flux-asjoSr/jobtmp-0-ƒoaCSTH', 'FLUX_TERMINUS_SESSION': '0', 'FLUX_TASK_LOCAL_ID': '1', 'FLUX_TASK_RANK': '1', 'FLUX_JOB_SIZE': '2', 'FLUX_JOB_NNODES': '1', 'FLUX_JOB_ID': 'ƒoaCSTH', 'FLUX_URI': 'local:///tmp/flux-asjoSr/local-0', 'FLUX_KVS_NAMESPACE': 'job-30568087552'}]]
Each entry in the returned result corresponds to one of the four FluxJobExecutor tasks. The FLUX_* environment variables show that each task ran with FLUX_JOB_SIZE=2 (two MPI ranks per task), confirming that cores=2 was honoured by the Flux instance. The sacct output for the outer job shows a single flux step (20339083.0), since all four inner tasks were scheduled by Flux within that one SLURM step rather than appearing as separate SLURM steps.
shell_output.split("\n")
['JobID JobName Partition Account AllocCPUS State ExitCode ',
'------------ ---------- ---------- ---------- ---------- ---------- -------- ',
'20339083 executorl+ s.cmmg mpd 8 RUNNING 0:0 ',
'20339083.ba+ batch mpd 8 RUNNING 0:0 ',
'20339083.ex+ extern mpd 8 RUNNING 0:0 ',
'20339083.0 flux mpd 8 RUNNING 0:0 ',
'']
import pandas
The cache DataFrame gives an audit trail for the outer job: the function object, its arguments, the full resource_dict used for submission (including the Flux submission template), the SLURM queue_id, and the path to the HDF5 file storing the serialised return value. The inner FluxJobExecutor tasks are not tracked individually in this cache, since they are managed by the nested Flux instance.
pandas.DataFrame(df)
| function | input_args | input_kwargs | resource_dict | queue_id | error_log_file | filename | |
|---|---|---|---|---|---|---|---|
| 0 | <function get_nested_env at 0x1458dbd3fce0> | [] | {} | {'submission_template': '#!/bin/bash #SBATCH -... | 20339083 | None | /cmmc/u/janj/notebooks/2026/2026-06-03-prabhat... |
for d in get_cache_data(cache_dir):
if "get_nested_env" in str(d["function"]):
print(d['queue_id'])
shell_output = subprocess.check_output(["sacct", "-j", str(d['queue_id'])], shell=False, universal_newlines=True)
20339083
shell_output.split("\n")
['JobID JobName Partition Account AllocCPUS State ExitCode ',
'------------ ---------- ---------- ---------- ---------- ---------- -------- ',
'20339083 executorl+ s.cmmg mpd 8 RUNNING 0:0 ',
'20339083.ba+ batch mpd 8 RUNNING 0:0 ',
'20339083.ex+ extern mpd 8 RUNNING 0:0 ',
'20339083.0 flux mpd 8 RUNNING 0:0 ',
'']
Clean up#
import os
import shutil
cache_dir = "executorlib_cache"
os.listdir(cache_dir)
['get_nested_enveb3ebea03eb5eea6a134279f9e1e3010_o.h5',
'get_nested_enveb3ebea03eb5eea6a134279f9e1e3010']
shutil.rmtree(cache_dir)