Resource requests

Declare CPU, memory, GPU, and package requirements so the dispatcher routes each workflow execution to a capable worker.

Resource requests tell Flux what a workflow needs to run: how many CPU cores, how much memory, how many GPUs, and which Python packages must be present. The dispatcher reads those requirements and only sends an execution to a worker that can satisfy all of them.

How dispatcher matching works

When the server picks a worker for a pending execution, Flux checks the workflow’s requests against each connected worker’s advertised resources and pushes the execution to a match over the worker’s SSE stream (workers receive dispatch via GET /workers/{name}/connect, not by polling). A workflow with no requests runs on any worker. A workflow with requests runs only on a worker where every declared requirement is met at the moment dispatch happens:

If no connected worker matches, the execution stays in CREATED state and is re-evaluated when a worker registers, when worker resource reports change, or when a running execution releases resources. An execution never errors due to resource mismatch — it waits until a capable worker is available.

Declaring requests

Pass a ResourceRequest to workflow.with_options(requests=...):

from flux import workflow, ExecutionContext
from flux.domain.resource_request import ResourceRequest

@workflow.with_options(
    name="data_processing_workflow",
    requests=ResourceRequest(
        cpu=4,
        memory="8Gi",
        packages=["pandas>=1.3.0", "numpy"],
    ),
)
async def data_processing_workflow(ctx: ExecutionContext[dict[str, str]]):
    """Process data with specific CPU and memory requirements."""
    ...

ResourceRequest accepts any combination of the five fields — omit fields you don’t need.

Field reference

cpu — logical core count

An integer. The worker’s available cores at claim time must be ≥ this value. Available cores are calculated from psutil.cpu_count(logical=True) minus current usage percentage.

ResourceRequest(cpu=8)           # at least 8 logical cores free
ResourceRequest.with_cpu(8)      # same, using the factory helper

memory — RAM

A string using standard binary suffixes (Ki, Mi, Gi, Ti) or a plain integer for bytes. "8Gi" becomes 8,589,934,592 bytes internally.

ResourceRequest(memory="8Gi")    # 8 gibibytes
ResourceRequest(memory="512Mi")  # 512 mebibytes
ResourceRequest(memory=4096)     # 4096 bytes (rare; prefer suffixed form)
ResourceRequest.with_memory("16Gi")

The worker reports available memory via psutil.virtual_memory().available.

gpu — GPU count

An integer representing the minimum number of GPUs required. Each GPU counts as available only if it has non-zero free VRAM (checked via GPUtil). Workers without GPUs never match a gpu requirement.

ResourceRequest(gpu=1)           # at least one GPU with free VRAM
ResourceRequest.with_gpu(2)      # at least two GPUs
from flux import workflow, ExecutionContext
from flux.domain.resource_request import ResourceRequest

@workflow.with_options(
    name="model_training_workflow",
    requests=ResourceRequest.with_gpu(1),
)
async def model_training_workflow(ctx: ExecutionContext[dict]):
    """Train ML model with GPU requirements."""
    ...

disk — free disk space

An integer in bytes. The worker’s free disk space on its root filesystem must be ≥ this value.

ResourceRequest(disk=10_000_000_000)     # 10 GB free
ResourceRequest.with_disk(50_000_000_000)

packages — Python package requirements

A list of strings, each a package name with an optional version constraint (>= or ==). The worker’s installed environment (reported at registration) must satisfy all entries.

ResourceRequest(packages=["pandas>=1.3.0", "numpy", "torch==2.0.0"])
ResourceRequest.with_packages(["matplotlib>=3.5.0", "seaborn>=0.11.0"])

Package names are compared case-insensitively. A package listed without a version constraint only requires that the package be installed.

Single-field factories

Each resource dimension has a named factory for the common case where only one field matters:

ResourceRequest.with_cpu(4)
ResourceRequest.with_memory("8Gi")
ResourceRequest.with_disk(10_000_000_000)
ResourceRequest.with_gpu(1)
ResourceRequest.with_packages(["scipy", "scikit-learn>=1.2.0"])

Factories are equivalent to ResourceRequest(cpu=4) etc. — the other fields default to None and are not checked.

Combining requirements

Fields compose freely in the constructor:

@workflow.with_options(
    requests=ResourceRequest(
        cpu=4,
        memory="8Gi",
        gpu=1,
        packages=["torch>=2.0.0", "transformers"],
    ),
)
async def fine_tune(ctx: ExecutionContext[dict]):
    ...

All requirements must be satisfied simultaneously. A worker with the right CPUs and memory but no GPU will not match.

Combining requests with affinity

requests and affinity are independent filters that both apply. A worker must match the label constraints in affinity and meet the resource quantities in requests:

from flux import workflow, ExecutionContext
from flux.domain.resource_request import ResourceRequest

@workflow.with_options(
    affinity={"role": "training", "gpu": "a100"},
    requests=ResourceRequest(gpu=1, memory="16Gi"),
)
async def large_training_run(ctx: ExecutionContext[str]):
    # Only dispatches to workers labeled role=training,gpu=a100
    # AND with >= 1 GPU and >= 16 GiB free memory
    ...

Labels are checked first. Resource quantities are checked against the set of workers that passed the label check.

What happens when no worker matches

The execution stays queued in CREATED state. The server re-evaluates pending executions whenever a worker registers, sends an updated resource report, or finishes an execution that frees capacity. As soon as a capable worker is available, the server pushes the execution to it over SSE.

How the server picks a worker

Among workers that satisfy a pending execution’s affinity labels and requests quantities, the server picks an eligible target based on current load — the least-loaded matching worker receives the dispatch.

What’s next