Flux Executor¶

Overview || Base Executor || MPI Executor || Flux Executor

This module provides a native Flux executor using the Flux Python API.

The FluxExecutor submits jobs directly to Flux using its Python bindings, rather than wrapping flux run as a subprocess. This provides better integration with Flux’s job lifecycle management and is particularly useful when running inside containers where MPI runners may not be available.

Usage:

from libensemble.executors.flux_executor import FluxExecutor

exctr = FluxExecutor()
exctr.register_app(full_path="/path/to/my_app.x", app_name="my_app")

# In your sim function:
task = exctr.submit(app_name="my_app", num_procs=4, num_nodes=1)
task.wait()

Requirements:

flux-core Python bindings must be installed
Must be running inside a Flux instance or provide a Flux URI

Notes

Flux handles are not thread-safe. FluxExecutor is best suited for process-based libEnsemble runs, such as multiprocessing or MPI workers. The executor reconnects lazily after process serialization so each worker process uses its own Flux handle.

Note

The FluxExecutor requires the flux-core Python bindings to be installed and importable (e.g., via conda install -c conda-forge flux-core or Spack), and must either be run inside a Flux instance or be given a Flux URI to connect to. If these bindings aren’t available, FluxExecutor won’t be importable, but the standard MPI Executor can still submit to Flux via mpi_runner="flux" (see below).

Flux Executor

class libensemble.executors.flux_executor.FluxExecutor(uri=None)¶

Bases: Executor

Native Flux executor using the Flux Python API.

This executor submits jobs directly to Flux rather than wrapping flux run as a subprocess. It provides better integration with Flux’s job lifecycle and is suitable for container-based workflows.

Parameters:: uri (str, Optional) – Flux instance URI. If omitted, flux.Flux() connects to the nearest enclosing Flux instance discovered by Flux.
Raises:: ExecutorException – If flux Python bindings are not available or connecting to Flux fails.

Example

from libensemble.executors.flux_executor import FluxExecutor

exctr = FluxExecutor()
exctr.register_app(full_path="/path/to/sim.x", app_name="sim")

# In sim function:
task = exctr.submit(app_name="sim", num_procs=4)
task.wait()

__init__(uri=None)¶

Instantiate a new FluxExecutor instance.

Parameters:: uri (str | None)
Return type:: None

submit(calc_type=None, app_name=None, num_procs=None, num_nodes=None, procs_per_node=None, num_gpus=None, app_args=None, stdout=None, stderr=None, dry_run=False, wait_on_start=False, extra_args=None)¶

Submit a job to Flux.

Returns FluxTask object.

Parameters:

calc_type (str, Optional) – The calculation type: ‘sim’ or ‘gen’
app_name (str, Optional) – The application name.
num_procs (int, Optional) – The total number of processes (MPI ranks)
num_nodes (int, Optional) – The number of nodes
procs_per_node (int, Optional) – The processes per node
num_gpus (int, Optional) – The total number of GPUs
app_args (str, Optional) – Application arguments
stdout (str, Optional) – Standard output filename
stderr (str, Optional) – Standard error filename
dry_run (bool, Optional) – If True, don’t actually submit the job
wait_on_start (bool or int, Optional) – Whether to wait for job to start running. If an integer N is supplied, wait at most N seconds.
extra_args (str, Optional) – Additional arguments (currently not used for native Flux)

Returns:

task – The submitted task object

Return type:

FluxTask

Flux Task

Like the base Task, FluxTask objects are created and returned by FluxExecutor.submit(). poll(), kill(), and wait() are overridden to query and control the job via Flux’s own job-lifecycle API instead of subprocess/signal-based mechanisms.

class libensemble.executors.flux_executor.FluxTask(app=None, app_args=None, workdir=None, stdout=None, stderr=None, workerid=None, dry_run=False)¶

Bases: Task

Task subclass for Flux jobs using native Flux Python API.

Overrides poll() and kill() to use Flux job management instead of subprocess operations.

poll()¶

Polls and updates the status attributes of the task using Flux job state.

Return type:: None

wait(timeout=None)¶

Waits on completion of the Flux job or raises TimeoutExpired exception.

Parameters:: timeout (float | None)
Return type:: None

kill(wait_time=60)¶

Kills/cancels the Flux job.

Parameters:: wait_time (int, Optional) – Time in seconds to wait for cancellation. Note: Flux handles job cancellation internally.
Return type:: None

Two ways to run under Flux¶

libEnsemble offers two independent ways to run applications under Flux:

MPI Executor with the Flux runner. The standard MPI Executor can subprocess flux run like any other MPI launcher. This requires no Python bindings, works with the usual resource-manager and GPU auto-detection, and is a good default choice:
```
from libensemble.executors import MPIExecutor

exctr = MPIExecutor(custom_info={"mpi_runner": "flux"})
```
or, equivalently, by selecting the built-in "flux" platform:
```
libE_specs["platform"] = "flux"
```
Flux Executor. FluxExecutor instead submits jobs directly to Flux through its Python API (flux.job.submit_async), bypassing any launcher subprocess entirely. This is particularly useful inside containers or other environments where a standard MPI launcher isn’t available, and gives more direct access to Flux’s own job states. Since Flux handles aren’t thread-safe, FluxExecutor is best suited to process-based libEnsemble runs (e.g., multiprocessing or MPI workers) rather than threaded workers.

Basic usage

from libensemble import Ensemble
from libensemble.executors.flux_executor import FluxExecutor

exctr = FluxExecutor()
exctr.register_app(full_path="/home/user/forces.x", app_name="forces")
ensemble = Ensemble(executor=exctr)

In user simulation function:

def sim_func(H, persis_info, sim_specs, libE_info):
    exctr = libE_info["executor"]

    task = exctr.submit(
        app_name="forces",
        num_procs=8,
        num_nodes=2,
        stdout="out.txt",
        stderr="err.txt",
    )

    # Wait for task to complete
    task.wait()

GPUs can be requested with num_gpus, and jobs can be submitted without blocking on start-up via wait_on_start (an integer gives a timeout in seconds). See the Forces tutorial for a complete calling script and simulation function using the forces.x application; substituting FluxExecutor for MPIExecutor there requires no other changes.