HPC Cluster Executor

HPC Cluster Executor#

In contrast to the Single Node Executor and the HPC Job Executor the HPC Submission Executors do not communicate via the zero message queue but instead store the python functions on the file system and uses the job scheduler to handle the dependencies of the Python functions. Consequently, the block allocation block_allocation and the init function init_function are not available in the HPC Cluster Executors. At the same time it is possible to close the Python process which created the Executor, wait until the execution of the submitted Python functions is completed and afterwards reload the results from the cache.

Internally the HPC submission mode is using the Python simple queuing system adatper (pysqa) to connect to HPC job schedulers and the h5py package for serializing the Python functions to store them on the file system. Both packages are optional dependency of executorlib. The installation of the pysqa package and the h5py package are covered in the installation section.

SLURM#

The Simple Linux Utility for Resource Management (SLURM) job scheduler is currently the most commonly used job scheduler for HPC clusters. On shared HPC systems users cannot access compute nodes directly — SLURM acts as the resource controller, accepting job requests, managing the queue, and assigning work to nodes when resources become free.

In the HPC submission mode executorlib internally uses the sbatch command, this is in contrast to the HPC Job Executor which internally uses the srun command.

The connection to the job scheduler is based on the Python simple queuing system adatper (pysqa). It provides a default configuration for most commonly used job schedulers including SLURM, in addition it is also possible to provide the submission template as part of the resource dictionary resource_dict for a single script, or persist it once in a configuration directory referenced via the pysqa_config_directory parameter, as covered in the “Reusing the Configuration” section below. All three options are covered in more detail on the pysqa documentation.

Background#

Three commands cover the day-to-day SLURM workflow:

sbatch submits a batch script from the login node. The script carries resource directives on lines starting with #SBATCH:

#!/bin/bash
#SBATCH --job-name=my_job
#SBATCH --output=my_job.out
#SBATCH --ntasks=4           # total MPI ranks
#SBATCH --cpus-per-task=1    # CPU threads per rank
#SBATCH --time=00:30:00      # wall-clock limit (HH:MM:SS)
#SBATCH --partition=regular

srun --mpi=pmix python my_script.py

sbatch job.sh   # submit — returns a job ID immediately

Key #SBATCH directives:

Directive	Meaning
`--ntasks=N`	Total MPI ranks (processes)
`--cpus-per-task=N`	CPU threads available to each rank
`--mem=NG`	Memory per node (e.g. `8G`)
`--time=HH:MM:SS`	Maximum wall-clock time
`--partition=name`	Queue / partition to target
`--dependency=afterok:JOBID`	Run only after another job succeeds

srun launches parallel tasks inside an existing allocation. Multiple srun calls can run concurrently using shell backgrounding:

srun --mpi=pmix -n 4 python task_a.py &
srun --mpi=pmix -n 4 python task_b.py &
wait   # block until both finish

squeue and sacct let you inspect the queue and verify resource assignments:

squeue --me                                              # your running/pending jobs
sacct -j 12345 --format=JobID,State,AllocCPUS,Elapsed   # accounting for job 12345

Common squeue state codes: PD (pending), R (running), CG (completing).

MPI-parallel Python#

The Message Passing Interface (MPI) is the dominant parallelisation standard on HPC systems. mpi4py provides Python bindings. A minimal example:

# script.py
from mpi4py import MPI
comm = MPI.COMM_WORLD
print(f"rank {comm.Get_rank()} of {comm.Get_size()}")

srun --mpi=pmix -n 4 python script.py

When multiple independent groups of ranks need to run inside one allocation there are two approaches:

Approach	How	Cross-group communication
Multiple `srun` calls	Each `srun` gets its own communicator	Not possible
`MPI_Comm_split`	One `srun`, split in Python	Possible via `MPI.COMM_WORLD`

# communicator splitting — 8 ranks split into two groups of 4
comm = MPI.COMM_WORLD
color = comm.Get_rank() // 4   # group 0 or group 1
sub_comm = comm.Split(color)

SlurmClusterExecutor#

from executorlib import SlurmClusterExecutor

In comparison to the SingleNodeExecutor, the only parameter which is changed in the SlurmClusterExecutor is the requirement to specify the cache directory using the cache_directory="./cache". The rest of the syntax remains exactly the same, to simplify the up-scaling of simulation workflows.

with SlurmClusterExecutor(cache_directory="./cache") as exe:
    future_lst = [exe.submit(sum, [i, i]) for i in range(1, 4)]
    print([f.result() for f in future_lst])

Specific parameters for SlurmClusterExecutor like the maximum run time "run_time_max", the maximum memory "memory_max" or the submission template for the job submission script "submission_template" can be specified as part of the resource dictionary. Again it is possible to specify the resource dictonary resource_dicionary either for each function in the submit() function or during the initialization of the SlurmClusterExecutor.

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 --ntasks={{cores}}

{{command}}
"""

with SlurmClusterExecutor(cache_directory="./cache") as exe:
    future = exe.submit(
        sum, [4, 4], 
        resource_dict={
            "submission_template": submission_template, 
            "run_time_max": 180,  # in seconds  
            "partition": "s.cmfe",
        })
    print(future.result())

The template uses Jinja2 syntax. executorlib fills {{cores}} into --ntasks, so cores=1 requests one serial process and cores=4 requests four MPI ranks. With pmi_mode="pmix" the executor additionally wraps the function call in srun --mpi=pmix -n <cores>:

def mpi_calc(i):
    from mpi4py import MPI
    comm = MPI.COMM_WORLD
    return {"rank": comm.Get_rank(), "size": comm.Get_size(), "input": i}

with SlurmClusterExecutor(pmi_mode="pmix", cache_directory="./cache") as exe:
    future = exe.submit(
        mpi_calc, 42,
        resource_dict={"submission_template": submission_template, "cores": 4, "partition": "regular", "run_time_max": 120})
    print(future.result())

With these options executorlib in combination with the SLURM job scheduler provides a lot flexibility to configure the submission of Python functions depending on the specific configuration of the job scheduler.

In case the submission goes wrong, for example when no submission template is defined and executorlib falls back to the default submission template a CalledProcessError is raised:

def echo(i):
    return i

with SlurmClusterExecutor() as exe:
    f1 = exe.submit(echo)
    print(f1.result())

The error message would be something like:

CalledProcessError: Command '['sbatch', '--parsable', '/home/jovyan/executorlib_cache/echoe5873521b7330f831bed941744079e83/run_queue.sh']' returned non-zero exit status 1.

This is not really helpful, as it primarily says the sbatch command was not successful, but the error message which sbatch raised on the command line is not directly accessible from Python. To access the error message you can access the exception of the future object:

excep = f1.exception()
print(excep.output)

This gives:

sbatch: error: invalid partition specified: (null)
sbatch: error: Batch job submission failed: Invalid partition name specified

So the submission failed because no partition was provided, which can be corrected by adding the --partition flag in the submission template, as demonstrated above.

Verifying the Resource Assignment#

After the submission it is often useful to confirm that the job scheduler actually assigned the requested resources. For the SlurmClusterExecutor the SLURM job identifier is stored in the cache and can be retrieved with the get_cache_data() function. This job identifier queue_id can then be passed to the SLURM sacct command to inspect the accounting record of the job:

from executorlib import get_cache_data
import subprocess

for entry in get_cache_data(cache_directory="./cache"):
    if "calc" in str(entry["function"]):
        job_id = entry["queue_id"]
        print(subprocess.check_output(
            ["sacct", "-j", str(job_id), "--format=JobID,State,AllocCPUS,Elapsed"],
            universal_newlines=True,
        ))

The AllocCPUS column reports the number of CPU cores SLURM allocated for the function. In addition to the queue_id each cache entry also contains the full resource_dict, the runtime and the path of the result file, which together provide a complete audit trail of the submission - this is the same information returned by the get_cache_data() function demonstrated for the Single Node Executor.

Reusing the Configuration with `pysqa_config_directory`#

The submission_template shown above is convenient for a single script, but has to be repeated in every Python session which submits jobs. For a configuration which is set up once and reused by any number of scripts and users, executorlib can instead read the submission template from a configuration directory via the pysqa_config_directory parameter, which is passed through to the pysqa package. This is the same mechanism pyiron uses to manage its HPC configuration, so an existing pysqa configuration directory can be reused directly with executorlib.

A pysqa configuration directory contains a queue.yaml file which lists the available queues, together with a Jinja2 submission template for each of them:

~/.queues/
├── queue.yaml
└── slurm.sh

queue.yaml defines the queue type, the primary queue and the individual queues - cores_min and cores_max restrict the number of cores which can be requested, run_time_max caps the maximum run time in seconds and script points to the Jinja2 template file used to generate the submission script:

queue_type: SLURM
queue_primary: slurm
queues:
  slurm: {cores_min: 1, cores_max: 100, run_time_max: 259200, script: slurm.sh}

slurm.sh is the same kind of Jinja2 template already introduced above as the submission_template, just stored in a file rather than a Python string:

#!/bin/bash
#SBATCH --output=time.out
#SBATCH --job-name={{job_name}}
#SBATCH --chdir={{working_directory}}
#SBATCH --get-user-env=L
#SBATCH --partition=my_partition
{%- if run_time_max %}
#SBATCH --time={{ [1, run_time_max // 60]|max }}
{%- endif %}
{%- if dependency_list %}
#SBATCH --dependency=afterok:{{ dependency_list | join(',') }}
{%- endif %}
{%- if memory_max %}
#SBATCH --mem={{memory_max}}G
{%- endif %}
#SBATCH --ntasks={{cores}}

{{command}}

Once the directory is set up it can be reused by any SlurmClusterExecutor, simply by pointing pysqa_config_directory at it - no submission_template needs to be specified anymore:

with SlurmClusterExecutor(cache_directory="./cache", pysqa_config_directory="~/.queues") as exe:
    future = exe.submit(sum, [4, 4], resource_dict={"run_time_max": 180, "cores": 4})
    print(future.result())

If pysqa_config_directory is not provided, as in the earlier examples, executorlib falls back to a generic, built-in submission template for the selected job scheduler - this is the default submission template referenced in the CalledProcessError example above. Providing an explicit pysqa_config_directory (or submission_template) becomes necessary as soon as the target HPC system requires mandatory options like --partition.

For clusters which are only reachable via SSH, or setups spanning multiple job schedulers, the configuration directory can instead contain a clusters.yaml file referencing several such queue.yaml configurations. This and further configuration options, like SSH-based remote submission, are covered in the pysqa documentation.

Setup with Claude Code#

Writing a submission_template for a new HPC cluster requires knowing its partitions, memory defaults, GPU resources, submit filters and job limits. To automate this step, executorlib provides the executorlib-cluster-setup skill for Claude Code. Started on the login node of a SLURM cluster, it analyses the cluster with sinfo, scontrol and sacctmgr, checks the generated submission templates with sbatch --test-only, runs short smoke tests for serial functions, GPUs, MPI binaries and dependencies, and finally writes a cluster-specific skill executorlib-<cluster>. This cluster-specific skill contains the tested submission_template, helper functions to construct the resource_dict and the pitfalls found during testing, so later Claude Code sessions can submit Python functions with the SlurmClusterExecutor on this cluster.

To install the skill for your user, download it to the Claude Code skills directory on the login node:

mkdir -p ~/.claude/skills/executorlib-cluster-setup
curl -o ~/.claude/skills/executorlib-cluster-setup/SKILL.md https://raw.githubusercontent.com/pyiron/executorlib/main/.claude/skills/executorlib-cluster-setup/SKILL.md

Afterwards start claude on the login node, in the Python environment with executorlib and pysqa installed, and call the skill with /executorlib-cluster-setup or ask Claude Code to set up executorlib for this cluster. Claude Code asks for the information the cluster cannot provide itself, like the codes you want to run, the account to charge and the URL of the site documentation. As the smoke tests submit short jobs, preferably to a debug or test partition, they consume a small amount of your computing time.

Flux#

Flux is a modern HPC resource manager developed at Lawrence Livermore National Laboratory (LLNL). On many systems it runs as a secondary scheduler inside a SLURM allocation, enabling fine-grained hierarchical task distribution. Unlike SLURM, Flux can also be installed locally via conda — making it especially suitable for demonstrations, testing, and continuous integration:

conda install -c conda-forge flux-core
flux start   # launch a local Flux instance

This simple installation is explained in the installation section. The features demonstrated below using Flux apply equally to SLURM.

Background#

The key Flux commands map closely onto their SLURM equivalents:

Flux command	SLURM equivalent	Description
`flux resource list`	`sinfo`	Show available nodes, cores, and GPUs
`flux jobs -a`	`squeue`	List all jobs (running and completed)
`flux submit`	`sbatch`	Submit a non-blocking job; returns a job ID immediately
`flux run`	`srun`	Launch a blocking job; waits for completion
`flux job attach <ID>`	—	Stream output of a previously submitted job

# submit a non-blocking 4-rank MPI job
flux submit -o pmi=pmix --ntasks=4 python script.py

# run a blocking 4-rank MPI job (waits for output)
flux run -o pmi=pmix -n 4 python script.py

The -o pmi=pmix flag matches what SLURM’s --mpi=pmix provides — the same mpi4py scripts run unchanged under both schedulers.

Dependencies#

As already demonstrated for the SingleNodeExecutor the Executor classes from executorlib are capable of resolving the dependencies of serial functions, when concurrent futures Future objects are used as inputs for subsequent function calls. For the case of the HPC submission these dependencies are communicated to the job scheduler, which allows to stop the Python process which created the Executor class, wait until the execution of the submitted Python functions is completed and afterwards restart the Python process for the Executor class and reload the calculation results from the cache defined by the cache_directory parameter.

def add_funct(a, b):
    return a + b

from executorlib import FluxClusterExecutor

with FluxClusterExecutor(cache_directory="./file") as exe:
    future = 0
    for i in range(4, 8):
        future = exe.submit(add_funct, i, future)
    print(future.result())

Resource Assignment#

In analogy to the SingleNodeExecutor the resource assignment for the FluxClusterExecutor is handled by either including the resource dictionary parameter resource_dict in the initialization of the FluxClusterExecutor class or in every call of the submit() function.

Below this is demonstrated once for the assignment of multiple CPU cores for the execution of a Python function which internally uses the message passing interface (MPI) via the mpi4py package.

def calc(i):
    from mpi4py import MPI

    size = MPI.COMM_WORLD.Get_size()
    rank = MPI.COMM_WORLD.Get_rank()
    return i, size, rank

with FluxClusterExecutor(cache_directory="./file") as exe:
    fs = exe.submit(calc, 3, resource_dict={"cores": 2})
    print(fs.result())

(3, 1, 0)

Beyond CPU cores and threads which were previously also introduced for the Single Node Executor the HPC Cluster Executors also provide the option to select the available accelerator cards or GPUs, by specifying the "gpus_per_core" parameter in the resource dictionary resource_dict. For demonstration we create a Python function which reads the GPU device IDs and submit it to the FluxClusterExecutor class:

def get_available_gpus():
    import socket
    from tensorflow.python.client import device_lib
    local_device_protos = device_lib.list_local_devices()
    return [
        (x.name, x.physical_device_desc, socket.gethostname()) 
        for x in local_device_protos if x.device_type == 'GPU'
    ]

with FluxClusterExecutor(
    cache_directory="./cache",
    resource_dict={"gpus_per_core": 1}
) as exe:
    fs_1 = exe.submit(get_available_gpus)
    fs_2 = exe.submit(get_available_gpus)
    print(fs_1.result(), fs_2.result())

Block Allocation#

In analogy to the SingleNodeExecutor also the FluxClusterExecutor and SlurmClusterExecutor support the block allocation mode. This starts a number of workers by submitting jobs to the HPC cluster. After these jobs are started, functions are submitted using socket based communication and the same Executor is used to execute multiple functions. This scheme is commonly known as a worker approach, which enables off-loading compute heavy tasks. Below the same example is repeated, which was previously introduced for the SingleNodeExecutor:

def init_function():
    return {"j": 4, "k": 3, "l": 2}

def calc_with_preload(i, j, k):
    return i + j + k

The function calc_with_preload() requires three inputs i, j and k. But when the function is submitted to the executor only two inputs are provided fs = exe.submit(calc, 2, j=5). In this case the first input parameter is mapped to i=2, the second input parameter is specified explicitly j=5 but the third input parameter k is not provided. So the FluxClusterExecutor automatically checks the keys set in the init_function() function. In this case the returned dictionary {"j": 4, "k": 3, "l": 2} defines j=4, k=3 and l=2. For this specific call of the calc_with_preload() function, i and j are already provided so j is not required, but k=3 is used from the init_function() and as the calc_with_preload() function does not define the l parameter this one is also ignored.

with FluxClusterExecutor(
    cache_directory="./file", 
    max_workers=2, 
    init_function=init_function, 
    block_allocation=True
) as exe:
    fs = exe.submit(calc_with_preload, 2, j=5)
    print(fs.result())

The result is 2+5+3=10 as i=2 and j=5 are provided during the submission and k=3 is defined in the init_function() function.

Disconnecting and Reconnecting#

A key advantage of the HPC Cluster Executors over the HPC Job Executors is that the Python process which created the executor does not need to stay alive while the submitted functions are running. As the functions are submitted as individual scheduler jobs and the results are stored on the file system, the Python process can be closed after the submission - for example to log out of the login node overnight - and the results can be reloaded later. This is controlled with the shutdown() method of the executor, which provides the same wait and cancel_futures parameters as the Executor interface of the Python standard library.

To submit a set of functions and disconnect without waiting, the executor is created without a with statement and shutdown() is called with wait=False and cancel_futures=False:

from executorlib import SlurmClusterExecutor

exe = SlurmClusterExecutor(cache_directory="./cache")
future_lst = [exe.submit(sum, [i, i]) for i in range(1, 4)]
exe.shutdown(wait=False, cancel_futures=False)

The submitted jobs remain in the queue of the job scheduler and continue to run. At a later point - even from a new Python process - the same functions are submitted again using the same cache_directory. executorlib recognises the cached results and returns them immediately instead of submitting the functions a second time:

from executorlib import SlurmClusterExecutor

with SlurmClusterExecutor(cache_directory="./cache") as exe:
    future_lst = [exe.submit(sum, [i, i]) for i in range(1, 4)]
    print([f.result() for f in future_lst])

The behaviour of the shutdown() method is summarized in the following table:

`shutdown()` call	Effect
`shutdown(wait=True)`	Wait until all submitted functions are finished before continuing - this is the default and also what the `with` statement does.
`shutdown(wait=False, cancel_futures=False)`	Return immediately and leave the submitted jobs running in the scheduler queue - used to disconnect from a running workflow.
`shutdown(wait=False, cancel_futures=True)`	Cancel the submitted jobs which have not started or finished yet.

This disconnect-and-reconnect capability is available for both the SlurmClusterExecutor and the FluxClusterExecutor, as both communicate via the file system rather than via sockets.

Combine both#

While SlurmClusterExecutor submits each Python function as a separate sbatch job from the login node, the HPC Job Executor (SlurmJobExecutor) runs inside an already-running SLURM allocation and dispatches tasks as srun steps — no new jobs enter the queue.

from executorlib import SlurmJobExecutor

# This code runs inside an existing SLURM job
with SlurmJobExecutor(max_workers=4) as exe:
    futures = [exe.submit(sum, [i, i], resource_dict={"cores": 1}) for i in range(4)]
    print([f.result() for f in futures])

The two executor types can be nested: a function submitted via SlurmClusterExecutor (running as an sbatch job) can itself create a SlurmJobExecutor to parallelise sub-tasks as srun steps within the same allocation:

from executorlib import SlurmClusterExecutor, SlurmJobExecutor

def parallel_workflow(n):
    with SlurmJobExecutor(max_workers=n) as inner:
        futures = [inner.submit(sum, [i, i], resource_dict={"cores": 1}) for i in range(n)]
        return [f.result() for f in futures]

with SlurmClusterExecutor(cache_directory="./cache") as outer:
    future = outer.submit(
        parallel_workflow, 4,
        resource_dict={"cores": 1, "partition": "regular", "run_time_max": 300})
    print(future.result())

The sacct output for such a job will show the outer sbatch job together with numbered srun steps (e.g. 12345.0, 12345.1, …), all completing within the same allocation.

Executor	Scheduler command	Typical use
`SlurmClusterExecutor`	`sbatch` (one job per function)	Submit from login node
`SlurmJobExecutor`	`srun` steps within current job	Inside an existing allocation
`FluxClusterExecutor`	`flux submit`	Flux-managed allocation or local testing

Cleaning Cache#

Finally, as the HPC Cluster Executors leverage the file system to communicate serialized Python functions, it is important to clean up the cache directory specified by the cache_directory parameter once the results of the submitted Python functions are no longer needed. The serialized Python functions are stored in binary format using the cloudpickle library for serialization. This format is design for caching but not for long-term storage. The user is responsible for the long-term storage of their data.

import os
import shutil

cache_dir = "./file"
if os.path.exists(cache_dir):
    print(os.listdir(cache_dir))
    try:
        shutil.rmtree(cache_dir)
    except OSError:
        pass

['add_funct144ecf19b5020fccad214df3f4bdabd0_o.h5', 'add_funct67b8245bf71c3c6dcb2018663939c72d', 'add_funct4a5d1f06cca57f0ffbaa3116d39ecc6a', 'add_functcb272924f36cbaa9ac79f8d42b6771c8_o.h5', 'calcd732cdcbd2c8e68145d99f0be814e667', '393f99f890d39348c6248689096e6efb', 'add_funct67b8245bf71c3c6dcb2018663939c72d_o.h5', 'add_funct4a5d1f06cca57f0ffbaa3116d39ecc6a_o.h5', 'calcd732cdcbd2c8e68145d99f0be814e667_o.h5', '3888b889a896c43b68f0bd924ee5285c', 'add_funct144ecf19b5020fccad214df3f4bdabd0', 'add_functcb272924f36cbaa9ac79f8d42b6771c8', 'calc_with_preloade920bec0b69f103b2579175e37aa3528_o.h5']

HPC Cluster Executor

Contents

HPC Cluster Executor#

SLURM#

Background#

MPI-parallel Python#

SlurmClusterExecutor#

Verifying the Resource Assignment#

Reusing the Configuration with pysqa_config_directory#

Setup with Claude Code#

Flux#

Background#

Dependencies#

Resource Assignment#

Block Allocation#

Disconnecting and Reconnecting#

Combine both#

Cleaning Cache#

Reusing the Configuration with `pysqa_config_directory`#