Skip to main content

API Reference

This is the auto-generated API reference for the dagster-slurm library. It is generated directly from the docstrings in the Python source code.

Core​

This section covers the main components of the library.

Dagster Slurm Integration.

Run Dagster assets on Slurm clusters with support for:

  • Local dev mode (no SSH/Slurm)
  • Per-asset Slurm submission (staging)
  • Run-scoped Slurm Ray allocation (opt-in)
  • Session mode with operator fusion (production)
  • Multiple launchers (Bash, Ray, Spark—WIP)

class dagster_slurm.AllocationSnapshot(slurm_job_id, role, state, reason=None, time_left_seconds=None, time_limit_seconds=None, requested_nodes=None, requested_gpus_per_node=None, tres=None, nodes=(), ray_nodes=None, draining=None, intent=None, drain_deadline=None, submitted_at=None)​

Bases: object

One allocation of a session at the time of a snapshot.

None means the value could not be observed: callers should not read it as zero, false or pending.

  • Parameters:
    • slurm_job_id (int)
    • role (Literal['head', 'successor', 'worker'])
    • state (str | None)
    • reason (str | None)
    • time_left_seconds (float | None)
    • time_limit_seconds (int | None)
    • requested_nodes (int | None)
    • requested_gpus_per_node (int | None)
    • tres (Mapping[str, str] | None)
    • nodes (tuple[str, ...])
    • ray_nodes (Mapping[str, str] | None)
    • draining (bool | None)
    • intent (str | None)
    • drain_deadline (float | None)
    • submitted_at (float | None)

property committed : bool | None​

Whether the allocation holds or waits for resources (pending or running).

drain_deadline : float | None = None​

Login-node epoch seconds after which reconcile_workers cancels it.

draining : bool | None = None​

property ended : bool​

property granted_gpus : int | None​

GPUs the scheduler granted to the running job, across its nodes.

intent : str | None = None​

Intent key of a worker allocation submitted by reconcile_workers.

nodes : tuple[str, ...] = ()​

ray_nodes : Mapping[str, str] | None = None​

Ray node ID per Slurm node that joined the current head (workers only).

property ready : bool | None​

Whether every node of a worker allocation runs Ray under the current head.

A running job is not necessarily a ready Ray worker. None when the state or the Ray membership is unknown, and for heads and successors.

reason : str | None = None​

Scheduler reason, such as Resources or Priority for a pending job.

requested_gpus_per_node : int | None = None​

requested_nodes : int | None = None​

role : Literal['head', 'successor', 'worker']​

slurm_job_id : int​

state : str | None​

Scheduler state, ENDED without an accounting record, None if unknown.

submitted_at : float | None = None​

Login-node epoch seconds when reconcile_workers submitted it.

time_left_seconds : float | None = None​

Remaining walltime of a running job, as the scheduler reports it.

time_limit_seconds : int | None = None​

tres : Mapping[str, str] | None = None​

the grant once the job runs.

  • Type: Trackable resources from squeue

class dagster_slurm.BashLauncher(**data)​

Bases: ComputeLauncher

Executes Python scripts via bash.

Uses the self-contained pixi environment extracted at runtime. Sources the activation script provided by pixi-pack.

  • Parameters: data (Any)

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Generate bash execution plan.

  • Parameters:
    • payload_path (str) – Path to Python script on remote host
    • python_executable (str) – Python from extracted environment
    • working_dir (str) – Working directory
    • pipes_context (dict[str, str]) – Dagster Pipes environment variables
    • extra_env (dict[str, str] | None) – Additional environment variables
    • allocation_context (dict[str, Any] | None) – Slurm allocation info (for session mode)
    • activation_script (str | None) – Path to activation script (provided by pixi-pack)
  • Return type: ExecutionPlan
  • Returns: ExecutionPlan with shell script

class dagster_slurm.ComputeLauncher(**data)​

Bases: ConfigurableResource

Base class for compute launchers.

  • Parameters: data (Any)

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Prepare execution plan.

  • Parameters:
    • payload_path (str) – Path to Python script on remote
    • python_executable (str) – Python interpreter path
    • working_dir (str) – Working directory
    • pipes_context (dict[str, str]) – Dagster Pipes environment
    • extra_env (dict[str, str] | None) – Additional environment variables
    • allocation_context (dict[str, Any] | None) – Slurm allocation info (for session mode)
    • activation_script (str | None) – Environment activation script
  • Return type: ExecutionPlan
  • Returns: ExecutionPlan with script and metadata

class dagster_slurm.ComputeResource(**data)​

Bases: ConfigurableResource

Unified compute resource - adapts to deployment.

This is the main facade that assets depend on. Hides complexity of local vs Slurm vs session execution.

Usage: : ```python @asset def my_asset(context: AssetExecutionContext, compute: ComputeResource): return compute.run( context=context, payload_path="script.py", launcher=RayLauncher(num_gpus_per_node=2) )


Configuration Examples:

Local mode (dev):

```python
compute = ComputeResource(mode="local")

Slurm per-asset mode (staging):

slurm = SlurmResource.from_env()
compute = ComputeResource(mode="slurm", slurm=slurm)

Slurm run-scoped Ray allocation (opt-in):

slurm = SlurmResource.from_env()
compute_ray = ComputeResource(
    mode="slurm",
    slurm=slurm,
    default_launcher=RayLauncher(num_gpus_per_node=1),
    allocation_scope=SlurmAllocationScope.RUN,
    run_allocation=SlurmRunAllocationConfig(num_nodes=2, gpus_per_node=1),
)

Slurm session mode with cluster reuse (prod):

slurm = SlurmResource.from_env()
session = SlurmSessionResource(slurm=slurm, num_nodes=10)
compute = ComputeResource(
    mode="slurm-session",
    slurm=slurm,
    session=session,
    enable_cluster_reuse=True,
    cluster_reuse_tolerance=0.2,
)

Heterogeneous job mode (optimal resource allocation):

compute = ComputeResource(
    mode="slurm-hetjob",
    slurm=slurm,
)
  • Parameters: data (Any)

allocation_scope : SlurmAllocationScope​

auto_detect_platform : bool​

cache_inject_globs : list[str] | None​

cleanup_deferred_run_dir(context)​

Delete the remote run_dir for a step whose run() calls used defer_cleanup=True.

run(..., defer_cleanup=True) (see SlurmPipesClient.run/SlurmPipesClient.cleanup_deferred_run_dir) exists for a caller that intentionally invokes run() repeatedly for one Dagster step against the same session/allocation (for example a per-wave loop) – that pattern reuses the same deterministic run_dir on every call, so per-call async cleanup can race the next call’s own upload into that directory. Callers that opt into defer_cleanup must call this once they are truly done calling run() for that step/partition, most naturally from the same scope that decided to reuse one session/allocation across calls (e.g. right before that scope releases its allocation lease).

A no-op outside Slurm execution modes (nothing was ever deferred).

  • Return type: None

cleanup_on_failure : bool​

cluster_reuse_tolerance : float​

debug_mode : bool​

default_environment_name : str | None​

default_extra_files : list[str] | None​

default_launcher : Annotated[ComputeLauncher | PartialResource]​

default_skip_payload_upload : bool​

enable_cluster_reuse : bool​

get_pipes_client(context, launcher=None)​

Get appropriate Pipes client for this mode.

  • Parameters:
    • context (InitResourceContext) – Dagster resource context
    • launcher (ComputeLauncher | None) – Override launcher (uses default if None)
  • Returns: LocalPipesClient or SlurmPipesClient

get_run_allocation_session(context)​

Get or initialize the run session for caller-managed walltime relays.

  • Parameters: context (InitResourceContext | AssetExecutionContext)
  • Return type: SlurmSessionResource

mode : ExecutionMode​

pack_on_remote : bool​

pack_platform : str | None​

pre_deployed_env_path : str | None​

project_setup_cmd : list[str] | None​

project_setup_env : dict[str, str]​

project_setup_input_globs : list[str] | None​

ray_head_allocation : SlurmRunAllocationConfig | None​

register_cluster(cluster_address, framework, cpus, gpus, memory_gb)​

Register a newly created cluster for future reuse.

  • Parameters:
    • cluster_address (str) – Address of the cluster (e.g., “10.0.0.1:6379”)
    • framework (str) – “ray” or “spark”
    • cpus (int) – Total CPUs in cluster
    • gpus (int) – Total GPUs in cluster
    • memory_gb (int) – Total memory in GB

remote_pack_timeout : int​

run(context, payload_path, launcher=None, extra_slurm_opts=None, resource_requirements=None, force_env_push=None, skip_payload_upload=None, remote_payload_path=None, environment_name=None, config=None, extra_files=None, metrics_collector=None, poll_timeout=3600, slurm_metrics=None, **kwargs)​

Execute asset with optional resource overrides.

  • Parameters:
    • context – Dagster execution context
    • payload_path (str) – Path to Python script
    • launcher (ComputeLauncher | None) – Override launcher for this asset
    • extra_slurm_opts (dict[str, Any] | None) – Override Slurm options (non-session mode)
      • nodes: int
      • cpus_per_task: int
      • mem: str (e.g., “32G”)
      • gpus_per_node: int
      • time_limit: str (e.g., “02:00:00”)
      • signal_before_timeout: str (e.g., “TERM@120”)
    • resource_requirements (dict[str, Any] | None) – Resource requirements for cluster reuse (session mode)
      • cpus: int
      • gpus: int
      • memory_gb: int
      • framework: str (“ray” or “spark”)
    • force_env_push (bool | None) – Force repacking and uploading the environment for this run. If not provided, falls back to config.force_env_push or asset metadata.
    • skip_payload_upload (bool | None) – If True, do not upload the payload script; expects the remote payload to already exist. Falls back to config.skip_payload_upload or asset metadata key ‘skip_slurm_payload_upload’.
    • remote_payload_path (str | None) – Remote path to an existing payload when skipping upload. Falls back to config.remote_payload_path or asset metadata.
    • environment_name (str | None) – Named environment produced by project_setup_cmd. Falls back to config.environment_name, slurm_environment_name asset metadata, then default_environment_name.
    • config (SlurmRunConfig | None) – Optional SlurmRunConfig for run-time configuration via launchpad. Values from config are used as defaults, but explicit parameters take precedence.
    • extra_files (list[str] | None) – List of local file paths to upload alongside the payload. Merged with default_extra_files, asset metadata, and config.extra_files.
    • metrics_collector (Callable[[SlurmMetricsContext], Mapping[str, Any]] | None) – Optional callback that receives a SlurmMetricsContext after a standalone Slurm job or shared-allocation step finishes and returns custom Dagster metadata key-value pairs. Callback errors and invalid metadata are logged without failing the workload.
    • poll_timeout (int) – Maximum time in seconds to wait for the Slurm job to complete. Defaults to 3600 (1 hour).
    • slurm_metrics (Collection[SlurmMetric | str] | None) – Built-in Slurm metadata fields to attach. None enables all fields; an empty collection disables optional built-in fields.
    • **kwargs – Passed to client.run()
  • Yields: Dagster events
  • Return type: PipesClientCompletedInvocation

Examples​

# Simple execution with default resources
yield from compute.run(context, "script.py")
# Using SlurmRunConfig for launchpad-configurable options
@dg.asset
def my_asset(
    context: dg.AssetExecutionContext,
    compute: ComputeResource,
    config: SlurmRunConfig,
):
    return compute.run(
        context=context,
        payload_path="script.py",
        config=config,
    ).get_results()
# Override launcher for this asset
ray_launcher = RayLauncher(num_gpus_per_node=4)
yield from compute.run(context, "script.py", launcher=ray_launcher)
# Non-session mode: override Slurm resources
yield from compute.run(
    context,
    "script.py",
    extra_slurm_opts={"nodes": 1, "cpus_per_task": 16, "mem": "64G", "gpus_per_node": 2}
)
# Session mode: specify resource requirements for cluster reuse
yield from compute.run(
    context,
    "script.py",
    launcher=RayLauncher(num_gpus_per_node=2),
    resource_requirements={"cpus": 32, "gpus": 2, "memory_gb": 128, "framework": "ray"}
)

run_allocation : SlurmRunAllocationConfig​

run_hetjob(context, assets, launchers=None)​

Run multiple assets as a heterogeneous Slurm job.

Submit all assets together with their specific resource requirements. Only waits in queue ONCE, but each asset gets the resources it needs.

  • Parameters:
    • context – Dagster execution context
    • assets (list[tuple[str, str, dict[str, Any]]]) – List of (asset_key, payload_path, resource_requirements). Each resource_requirements dict may contain nodes (int, default 1), cpus_per_task (int, default 2), mem (str, default "4G"), gpus_per_node (int, default 0), and time_limit (str, default "01:00:00").
    • launchers (dict[str, ComputeLauncher] | None) – Optional dict mapping asset_key to ComputeLauncher
  • Yields: Dagster events

Example​

compute.run_hetjob(
    context,
    assets=[
        ("prep", "prep.py", {"nodes": 1, "cpus_per_task": 8, "mem": "32G"}),
        ("train", "train.py", {"nodes": 4, "cpus_per_task": 32, "mem": "128G", "gpus_per_node": 2}),
        ("infer", "infer.py", {"nodes": 8, "cpus_per_task": 16, "mem": "64G", "gpus_per_node": 1}),
    ],
    launchers={
        "train": RayLauncher(num_gpus_per_node=2),
        "infer": RayLauncher(num_gpus_per_node=1),
    }
)

session : SlurmSessionResource | None​

slurm : SlurmResource | None​

teardown(context)​

Teardown method called by Dagster at end of run. Ensures session resources and clusters are cleaned up.

  • Parameters: context (InitResourceContext)

teardown_after_execution(context)​

Release run-scoped allocations through Dagster’s resource lifecycle.

  • Parameters: context (InitResourceContext)
  • Return type: None

validate_configuration()​

Validate configuration - runs during Pydantic validation.

class dagster_slurm.LocalPipesClient(launcher, base_dir=None, require_pixi=True)​

Bases: PipesClient

Pipes client for local execution (dev mode). No SSH, no Slurm - just runs scripts locally via subprocess.

  • Parameters:

cleanup()​

Explicitly clean up resources.

run(, context, payload_path, python_executable=None, extra_env=None, extras=None, extra_slurm_opts=None, extra_files=None)​

Execute payload locally.

  • Parameters:
    • context (AssetExecutionContext) – Dagster execution context
    • payload_path (str) – Path to Python script to execute
    • python_executable (str | None) – Python interpreter (defaults to current)
    • extra_env (dict[str, str] | None) – Additional environment variables
    • extras (dict[str, Any] | None) – Extra data to pass via Pipes
    • extra_slurm_opts (dict[str, Any] | None)
    • extra_files (list[str] | None)
  • Yields: Dagster events (materializations, logs, etc.)
  • Return type: PipesClientCompletedInvocation

class dagster_slurm.RayLauncher(**data)​

Bases: ComputeLauncher

Ray distributed computing launcher.

Features:

  • Robust cluster startup with sentinel-based shutdown
  • Graceful cleanup on SIGTERM/SIGINT
  • Worker registration monitoring
  • Automatic head node detection
  • IPv4/IPv6 normalization

Modes:

  • Local: Single-node Ray
  • Cluster: Multi-node Ray cluster across Slurm allocation (via allocation_context)
  • Connect: Connect to existing cluster (via ray_address)
  • Parameters: data (Any)

dashboard_host : str​

dashboard_port : int​

grace_period : int​

head_startup_timeout : int​

network_interface : str | None​

node_ip_address_command : str | None​

num_gpus_per_node : int​

object_store_memory_gb : int | None​

port_config : RayPortConfig​

port_strategy : Literal['fixed', 'hash_jobid', 'random']​

pre_start_commands : list[str]​

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Generate Ray execution plan.

  • Parameters:
    • payload_path (str)
    • python_executable (str)
    • working_dir (str)
    • pipes_context (dict[str, str])
    • extra_env (dict[str, str] | None)
    • allocation_context (dict[str, Any] | None)
    • activation_script (str | None)
  • Return type: ExecutionPlan

ray_address : str | None​

ray_port : int​

ray_start_args : list[str]​

redis_password : str | None​

use_head_ip : bool​

validate_fixed_ports()​

worker_cpu_bind : str | None​

worker_startup_delay : int​

class dagster_slurm.RayPortConfig(**config_dict)​

Bases: Config

Port pool and fixed-port settings for a Ray cluster.

block_size : int​

dashboard_agent_grpc_port : int​

dashboard_agent_listen_port : int​

lock_dir : str​

max_worker_port : int​

metrics_export_port : int​

min_worker_port : int​

node_manager_port : int​

object_manager_port : int​

range_end : int​

range_start : int​

ray_client_server_port : int​

redis_shard_port : int​

runtime_env_agent_port : int​

validate_port_ranges()​

class dagster_slurm.ReconcileResult(observed_at, live, submitted=(), released=(), cancelled=(), recovered=(), held_back=None)​

Bases: object

What one reconcile_workers call observed and did.

  • Parameters:
    • observed_at (float)
    • live (tuple[int, ...])
    • submitted (tuple[int, ...])
    • released (tuple[int, ...])
    • cancelled (tuple[int, ...])
    • recovered (tuple[int, ...])
    • held_back (str | None)

cancelled : tuple[int, ...] = ()​

Allocations cancelled by this call, including drains past their deadline.

held_back : str | None = None​

Why the call did not submit, although the intent wanted more allocations.

live : tuple[int, ...]​

Allocations that count toward the intent after this call.

observed_at : float​

Login-node epoch seconds of the revalidating scheduler query.

recovered : tuple[int, ...] = ()​

Submissions of an interrupted earlier call, found by their token.

released : tuple[int, ...] = ()​

drained when running, else cancelled.

  • Type: Allocations released by this call

submitted : tuple[int, ...] = ()​

class dagster_slurm.SSHConnectionResource(**data)​

Bases: ConfigurableResource

SSH connection settings.

This resource configures a connection to a remote host via SSH. It supports key or inherited agent authentication, pseudo-terminal allocation (-t), and connections through a proxy jump host.

Set key_path for an explicit SSH key, or leave both authentication fields unset to use ssh-agent and ~/.ssh/config. Explicit passwords are rejected when establishing the required background ControlMaster.

key_path and password are mutually exclusive.

Note that every option this resource emits overrides ~/.ssh/config, because OpenSSH prefers command-line options. That keeps a run reproducible across machines; see host_key_checking='inherit' and batch_mode=None to hand individual settings back to the operator’s own configuration.

Examples​

# Key-based auth
ssh = SSHConnectionResource(
    host="cluster.example.com",
    user="username",
    key_path="~/.ssh/id_rsa",
)

# With a proxy jump host
jump_box = SSHConnectionResource(
    host="jump.example.com", user="jumpuser", key_path="~/.ssh/jump_key"
)
ssh_via_jump = SSHConnectionResource(
    host="private-cluster",
    user="user_on_cluster",
    key_path="~/.ssh/cluster_key",
    jump_host=jump_box
)

# With a post-login command (e.g., for VSC)
vsc_ssh = SSHConnectionResource(
    host="vmos.vsc.ac.at",
    user="dagster01",
    key_path="~/.ssh/vsc_key",
    force_tty=True,
    post_login_command="vsc5"
)
# From environment variables
ssh = SSHConnectionResource.from_env()
  • Parameters: data (Any)

auth_error_hint(stderr)​

Return an actionable hint for a non-interactive authentication failure.

BatchMode refuses every prompt, so a passphrase-protected key with no agent fails with a bare “Permission denied” instead of asking.

  • Parameters: stderr (str)
  • Return type: str

batch_mode : bool | None​

control_socket_path()​

Return the shared ControlMaster socket path for this connection.

  • Return type: str | None

property defers_to_ssh_config : bool​

Whether this connection adds nothing OpenSSH cannot work out itself.

Shared jump connections still use an explicit attach-only ProxyCommand. Unspecified authentication and host-key options remain inherited.

extra_opts : list[str]​

force_tty : bool​

classmethod from_env(prefix='SLURM_SSH', _is_jump=False)​

Create from environment variables.

This method reads connection details from environment variables. The variable names are constructed using the provided prefix.

With the default prefix, the following variables are used:

  • SLURM_SSH_HOST - SSH hostname (required)
  • SLURM_SSH_PORT - SSH port (optional, default: 22)
  • SLURM_SSH_USER - SSH username (required)
  • SLURM_SSH_KEY - Path to SSH key (optional)
  • SLURM_SSH_PASSWORD - SSH password (optional)
  • SLURM_SSH_FORCE_TTY - Set to ‘true’ or ‘1’ to enable tty allocation (optional)
  • SLURM_SSH_POST_LOGIN_COMMAND - Post-login command string (optional)
  • SLURM_SSH_OPTS_EXTRA - Additional SSH options (optional)
  • SLURM_SSH_HOST_KEY_CHECKING - off, accept-new, or strict (optional)
  • SLURM_SSH_KNOWN_HOSTS_FILE - Known-hosts file path (optional)

For proxy jumps, use the _JUMP suffix for jump host variables (e.g., SLURM_SSH_JUMP_HOST, SLURM_SSH_JUMP_USER, etc.).

  • Parameters:
    • prefix (str) – Environment variable prefix (default: “SLURM_SSH”)
    • _is_jump (bool)
  • Return type: SSHConnectionResource
  • Returns: SSHConnectionResource instance

get_auth_opts()​

Build authentication options for this connection’s auth mode.

  • Return type: list[str]

get_batch_mode_opts()​

Build the BatchMode option, or nothing when it is inherited.

  • Return type: list[str]

get_common_ssh_opts(, server_alive_interval=None, server_alive_count_max=None)​

Build shared SSH/SCP options with caller overrides first.

  • Parameters:
    • server_alive_interval (int | None)
    • server_alive_count_max (int | None)
  • Return type: list[str]

get_host_key_opts()​

Build host-key verification options.

Anything emitted here overrides ~/.ssh/config, because OpenSSH always prefers command-line options. Only the settings this resource actually specifies are emitted, so host_key_checking='inherit' leaves the operator’s own configuration untouched.

  • Return type: list[str]

get_key_auth_opts(, batch_mode=None)​

Build key-based SSH authentication options.

  • Parameters: batch_mode (bool | None) – Force BatchMode for this command. When omitted the resource’s batch_mode setting applies, and None there emits no BatchMode option so ~/.ssh/config decides.
  • Return type: list[str]

get_multiplexing_opts(, control_path=None)​

Build ControlMaster options pointing at the shared control socket.

  • Parameters: control_path (str | None) – Override the socket path (used by the connection pool).
  • Return type: list[str]

get_password_auth_opts(, prompts=3)​

Build password-based SSH authentication options.

  • Parameters: prompts (int)
  • Return type: list[str]

get_proxy_command_opts()​

Build a ProxyCommand whose SSH settings apply to the jump host.

  • Return type: list[str]

get_remote_target()​

Get the remote target string for SCP commands.

  • Return type: str

get_scp_base_command()​

Build an attach-only SCP command; establish SSHConnectionPool first.

  • Return type: list[str]

get_ssh_base_command()​

Build an attach-only SSH command; establish SSHConnectionPool first.

  • Return type: list[str]

host : str​

host_key_checking : Literal['off', 'accept-new', 'strict', 'inherit']​

host_key_error_hint(stderr)​

Return an actionable hint for an OpenSSH host-key failure.

  • Parameters: stderr (str)
  • Return type: str

jump_host : SSHConnectionResource | None​

key_path : str | None​

known_hosts_file : str | None​

property multiplexing_unsupported_reason : str | None​

Explain why ControlMaster is unavailable, if it is.

password : str | None​

port : int​

post_login_command : str | None​

property requires_tty : bool​

Return True when the resource explicitly requires a TTY.

property supports_multiplexing : bool​

Whether SSH ControlMaster can be used for this connection.

This library’s background master startup does not handle explicit passwords for the target or jump host.

user : str​

property uses_inherited_auth : bool​

Returns True when authentication is left entirely to OpenSSH.

Neither a key nor a password was configured, so ~/.ssh/config, ssh-agent and the default identity files decide.

property uses_key_auth : bool​

Returns True if using key-based authentication.

property uses_password_auth : bool​

Returns True if using password-based authentication.

class dagster_slurm.SbatchTestOnlyEstimator(, max_candidates=8, cache_seconds=60.0)​

Bases: object

Ask sbatch --test-only about every candidate in one SSH round trip.

Each candidate is rendered with the options add_worker_allocation would submit, so inherited account, QoS, reservation and memory settings apply. Every candidate costs one scheduler RPC: max_candidates bounds a call, and answers are reused for cache_seconds.

  • Parameters:
    • max_candidates (int)
    • cache_seconds (float)

estimate(session, candidates)​

class dagster_slurm.SessionSnapshot(observed_at, head_job_id, allocations, scheduler_available)​

Bases: object

Non-blocking observation of a session’s allocations.

Taken in two SSH round trips and one scheduler query. It reserves nothing: act on it promptly, and let reconcile_workers revalidate it.

  • Parameters:
    • observed_at (float)
    • head_job_id (int)
    • allocations (tuple[AllocationSnapshot, ...])
    • scheduler_available (bool)

allocations : tuple[AllocationSnapshot, ...]​

property certain : bool​

Whether the scheduler answered and every state is known.

property head : AllocationSnapshot​

head_job_id : int​

observed_at : float​

Login-node epoch seconds when the scheduler was queried.

scheduler_available : bool​

workers(intent=None)​

Worker allocations, optionally only those of one intent.

class dagster_slurm.SlurmAllocation(slurm_job_id, nodes, working_dir, config, session_dir=None)​

Bases: object

Represents a running Slurm allocation.

  • Parameters:
    • slurm_job_id (int)
    • nodes (list[str])
    • working_dir (str)
    • config (SlurmSessionResource)
    • session_dir (str | None)

cancel(ssh_pool)​

Cancel the allocation.

  • Parameters: ssh_pool (SSHConnectionPool)

property end_time : str | None​

Current scheduler EndTime in the cluster’s timezone (None if unknown).

Read on every access: backfill and TimeLimit updates can change it.

ensure_ray_cluster(, ssh_pool, launcher, activation_script, startup_timeout)​

Start one persistent Ray cluster inside the allocation and return its address.

  • Parameters:
    • ssh_pool (SSHConnectionPool)
    • launcher (Any)
    • activation_script (str)
    • startup_timeout (int)
  • Return type: str

execute(execution_plan, asset_key, run_dir, ssh_pool, step_update_callback=None, poll_callback=None, timeout=None)​

Execute plan in this allocation via srun.

get_failed_nodes()​

Get list of failed nodes.

  • Return type: list[str]

has_active_payloads(ssh_pool)​

  • Parameters: ssh_pool (SSHConnectionPool)
  • Return type: bool

is_healthy(ssh_pool)​

Check if allocation and nodes are healthy.

  • Parameters: ssh_pool (SSHConnectionPool)
  • Return type: bool

property ray_dashboard_url : str | None​

Dashboard URL reported by the persistent Ray head.

property ray_node_ids : dict[str, str]​

Ray node ID per Slurm node that joined as a worker allocation.

request_drain()​

Ask the allocation shell to drain payload steps, keeping Ray alive.

  • Return type: None

stop_ray_head(ssh_pool)​

Stop the persistent Ray head in this allocation and forget its address.

The head is marked as exited, so a session that still points at it elects again.

  • Parameters: ssh_pool (SSHConnectionPool)
  • Return type: None

wait_for_step(result, , ssh_pool, step_update_callback=None, poll_callback=None, timeout=None)​

Wait for a launched allocation step, including after supervisor restart.

exception dagster_slurm.SlurmAllocationEnded(result)​

Bases: RuntimeError

The allocation ended while a step ran, before or as it reported its exit.

class dagster_slurm.SlurmAllocationScope(*values)​

Bases: str, Enum

Controls how Slurm allocations are scoped for mode="slurm".

ASSET = 'asset'​

RUN = 'run'​

class dagster_slurm.SlurmMetric(*values)​

Bases: str, Enum

Optional built-in Slurm metadata fields.

ALLOCATED_GPUS = 'slurm_allocated_gpus'​

CPU_EFFICIENCY = 'cpu_efficiency_pct'​

ELAPSED_SECONDS = 'elapsed_seconds'​

EXIT_CODE = 'slurm_exit_code'​

GPU_ACCOUNTING_AVAILABLE = 'slurm_gpu_accounting_available'​

GPU_MEMORY_MAX = 'slurm_gpu_memory_max_mb'​

GPU_UTILIZATION_AVG = 'slurm_gpu_utilization_avg_pct'​

GPU_UTILIZATION_MAX = 'slurm_gpu_utilization_max_pct'​

GPU_UTILIZATION_MAX_NODE = 'slurm_gpu_utilization_max_node'​

MAX_MEMORY = 'max_memory_mb'​

NODE_HOURS = 'node_hours'​

REQUESTED_GPUS = 'slurm_requested_gpus'​

STATE = 'slurm_state'​

TRES_ACCOUNTING = 'slurm_tres_accounting'​

class dagster_slurm.SlurmMetricsContext(job_id, ssh_pool, slurm_resource, session_resource, default_metrics, step_id=None)​

Bases: object

Context passed to a custom post-run metrics collector.

  • Parameters:
    • job_id (int)
    • ssh_pool (SSHConnectionPool)
    • slurm_resource (SlurmResource)
    • session_resource (SlurmSessionResource | None)
    • default_metrics (SlurmJobMetrics)
    • step_id (str | None)

default_metrics : SlurmJobMetrics​

job_id : int​

session_resource : SlurmSessionResource | None​

slurm_resource : SlurmResource​

ssh_pool : SSHConnectionPool​

step_id : str | None = None​

class dagster_slurm.SlurmPipesClient(slurm_resource, launcher, session_resource=None, cleanup_on_failure=True, debug_mode=False, auto_detect_platform=True, pack_platform=None, pre_deployed_env_path=None, cache_inject_globs=None, pack_on_remote=False, remote_pack_timeout=600, project_setup_cmd=None, project_setup_env=None, project_setup_input_globs=None, run_allocation_scope=False)​

Bases: PipesClient

Pipes client for Slurm execution with real-time log streaming and cancellation support.

Features:

  • Real-time stdout/stderr streaming to Dagster logs
  • Packaging environment with pixi pack
  • Auto-reconnect message reading
  • Metrics collection
  • Graceful cancellation with Slurm job termination

Works in two modes:

  1. Standalone: Each asset = separate sbatch job
  2. Session: Multiple assets share a Slurm allocation (operator fusion)
  • Parameters:
    • slurm_resource (SlurmResource)
    • launcher (ComputeLauncher)
    • session_resource (Optional[SlurmSessionResource])
    • cleanup_on_failure (bool)
    • debug_mode (bool)
    • auto_detect_platform (bool)
    • pack_platform (str | None)
    • pre_deployed_env_path (str | None)
    • cache_inject_globs (list[str] | None)
    • pack_on_remote (bool)
    • remote_pack_timeout (int)
    • project_setup_cmd (list[str] | None)
    • project_setup_env (dict[str, str] | None)
    • project_setup_input_globs (list[str] | None)
    • run_allocation_scope (bool)

cleanup_deferred_run_dir(, context)​

Delete one step’s run_dir after a run of defer_cleanup=True calls.

Call this once, after the caller is certain it will not invoke run(..., defer_cleanup=True) again for the same Dagster step/partition/mapping (see _get_remote_run_dir – run_dir is derived purely from remote_base, the run id, and that identity, so it is safe to recompute here without any state carried over from the deferred calls). A natural place to call this is the same scope that opted into reusing one session/allocation across repeated run() calls (for example, right before that scope releases its allocation lease), once nothing else will write into this directory.

A no-op in debug mode, matching every other cleanup path on this client (debug mode always preserves remote directories for inspection).

  • Parameters: context (AssetExecutionContext)
  • Return type: None

run(, context, payload_path, extra_env=None, extras=None, use_session=False, extra_slurm_opts=None, force_env_push=None, skip_payload_upload=None, remote_payload_path=None, pack_cmd_override=None, pre_deployed_env_path_override=None, environment_name=None, extra_files=None, slurm_metrics=None, metrics_collector=None, poll_timeout=3600, defer_cleanup=False, **kwargs)​

Execute payload on Slurm cluster with real-time log streaming.

  • Parameters:
    • context (AssetExecutionContext) – Dagster execution context
    • payload_path (str) – Local path to Python script
    • launcher – Ignored (launcher is set at client construction time)
    • extra_env (dict[str, str] | None) – Additional environment variables
    • extras (dict[str, Any] | None) – Extra data to pass via Pipes
    • use_session (bool) – If True and session_resource provided, use shared allocation
    • extra_slurm_opts (dict[str, Any] | None) – Override Slurm options (non-session mode)
    • force_env_push (bool | None) – If True, always repack and upload the environment even when a cached copy exists for the current lockfile/pack command.
    • skip_payload_upload (bool | None) – If True, do not upload the payload script (assumes it already exists remotely).
    • remote_payload_path (str | None) – Optional pre-existing remote payload path to use when skipping upload.
    • environment_name (str | None) – Named environment prepared by project_setup_cmd.
    • slurm_metrics (Collection[SlurmMetric | str] | None) – Built-in Slurm metadata fields to attach. None enables all fields; an empty collection disables optional built-in fields.
    • metrics_collector (Callable[[SlurmMetricsContext], Mapping[str, Any]] | None) – Optional callback invoked after a standalone Slurm job or shared-allocation step reaches a terminal state. It receives a SlurmMetricsContext and returns additional Dagster metadata key-value pairs.
    • poll_timeout (int) – Maximum time in seconds to wait for the Slurm job to complete. Defaults to 3600 (1 hour).
    • defer_cleanup (bool) – If True, skip the async run_dir cleanup this call would otherwise trigger (on success, on a reattach outcome, and on failure) and leave it for the caller to remove later via cleanup_deferred_run_dir(). run_dir is deterministic per Dagster step/partition/mapping (see _get_remote_run_dir), not per call, so a caller that invokes run() repeatedly for one step against the same allocation/session (e.g. a per-wave loop that intentionally reuses one SlurmPipesClient-backed ComputeResource across many run() calls) would otherwise have each call’s fire-and-forget nohup rm -rf <run_dir> & race the next call’s mkdir -p/upload into that same, still-being-deleted directory – a real, unsynchronized delete-vs-recreate race, not merely wasted re-staging work. Defaults to False so every existing caller (one run() call per directory) is unaffected.
    • **kwargs – Additional arguments (ignored, for forward compatibility)
    • pack_cmd_override (list[str] | None)
    • pre_deployed_env_path_override (str | None)
    • extra_files (list[str] | None)
  • Yields: Dagster events
  • Return type: PipesClientCompletedInvocation

class dagster_slurm.SlurmQueueConfig(**data)​

Bases: ConfigurableResource

Default Slurm job submission parameters. These can be overridden per-asset via metadata or function arguments.

  • Parameters: data (Any)

account : str | None​

cpus : int​

gpus_per_node : int​

mem : str | None​

mem_per_cpu : str | None​

num_nodes : int​

partition : str​

qos : str | None​

reservation : str | None​

signal_before_timeout : str | None​

time_limit : str​

class dagster_slurm.SlurmResource(**data)​

Bases: ConfigurableResource

Complete Slurm cluster configuration. Combines SSH connection, queue defaults, and cluster-specific paths.

  • Parameters: data (Any)

classmethod from_env()​

Create from environment variables.

classmethod from_env_slurm(ssh)​

Create a SlurmResource by populating most fields from environment variables, but requires an explicit, pre-configured SSHConnectionResource to be provided.

next_status_poll_interval(current)​

Return the next status polling interval after an unchanged state.

  • Parameters: current (float)
  • Return type: float

queue : Annotated[SlurmQueueConfig | PartialResource]​

remote_base : str | None​

set_auth_provider(provider)​

ssh : Annotated[SSHConnectionResource | PartialResource]​

status_poll_backoff_factor : float​

status_poll_interval_seconds : float​

status_poll_max_interval_seconds : float​

class dagster_slurm.SlurmRunAllocationConfig(**config_dict)​

Bases: Config

Configuration for a run-owned Slurm allocation.

account : str | None​

cleanup_policy : Literal['after_run']​

constraint : str | None​

cpus_per_task : int | None​

exclude : str | None​

extra_sbatch_directives : list[str]​

gpus_per_node : int | None​

mem : str | None​

mem_per_cpu : str | None​

nodelist : str | None​

num_nodes : int | None​

partition : str | None​

qos : str | None​

reservation : str | None​

signal_before_timeout : str | None​

time_limit : str | None​

time_min : str | None​

class dagster_slurm.SlurmRunConfig(**config_dict)​

Bases: Config

Per-run configuration for Slurm execution.

Use this to configure environment caching and payload upload behavior at job submission time via the Dagster launchpad.

Example usage in an asset:

@dg.asset
def my_asset(
    context: dg.AssetExecutionContext,
    compute: ComputeResource,
    config: SlurmRunConfig,
):
    return compute.run(
        context=context,
        payload_path="script.py",
        config=config,
    ).get_results()

Then in the Dagster launchpad, you can override:

  • force_env_push: True to force re-upload the environment
  • skip_payload_upload: True to skip uploading the payload script

environment_name : str | None​

extra_files : list[str] | None​

force_env_push : bool​

remote_payload_path : str | None​

skip_payload_upload : bool​

class dagster_slurm.SlurmSessionResource(**data)​

Bases: ConfigurableResource

Slurm session resource for operator fusion.

This is a proper Dagster resource that manages the lifecycle of a Slurm allocation across multiple assets in a run.

Usage in definitions.py:

session = SlurmSessionResource(
    slurm=slurm,
    num_nodes=4,
    time_limit="04:00:00",
)
  • Parameters: data (Any)

account : str | None​

add_worker_allocation(config=None)​

Queue an allocation whose nodes join the run’s Ray cluster as workers.

Returns after submission. Once the job runs and the session’s Ray head is ready, each node starts a Ray worker with the run’s launcher, environment and port strategy. The nodes follow the session’s head across successor promotions and head failovers.

property allocation : SlurmAllocation​

Current allocation, refreshed from the session’s published identity.

assign_worker_intent(slurm_job_id, intent)​

Move a worker allocation into an intent, or out of every intent.

For an application that changes an allocation’s role: a worker added with add_worker_allocation can join an intent, so that reconcile_workers counts and releases it, and a worker that must stay can leave its intent, so that scale-in never releases it. An allocation that reconcile_workers is releasing keeps its intent until it ends. An allocation without a submission time never expires through pending_timeout.

  • Parameters:
    • slurm_job_id (int)
    • intent (str | None)
  • Return type: None

constraint : str | None​

cpus_per_task : int | None​

enable_health_checks : bool​

enable_session : bool​

estimate_worker_starts(candidates, , estimator=None)​

Estimate when each candidate worker allocation could start.

The default SbatchTestOnlyEstimator asks sbatch --test-only about every candidate in one SSH round trip and caches the answers briefly. Estimates are advisory and reserve nothing.

exclude : str | None​

execute_in_session(execution_plan, asset_key, run_dir, step_update_callback=None, poll_callback=None, timeout=None, allocation=None)​

Execute workload in the shared allocation. Thread-safe for parallel asset execution.

extra_sbatch_directives : list[str]​

gpus_per_node : int | None​

inspect()​

Observe the session’s allocations without waiting for any of them.

Reports requested and granted resources, scheduler state and pending reason, actual time left, Ray membership under the current head and drain status. Values that could not be observed are None.

property logger : Any​

max_concurrent_jobs : int​

mem : str | None​

mem_per_cpu : str | None​

nodelist : str | None​

num_nodes : int​

partition : str | None​

promote_successor(, launcher=None, activation_script=None, startup_timeout=120)​

Promote a RUNNING successor after predecessor payloads have drained.

Call allocation.request_drain() and wait for payload invocations to finish first. Ray is started before publication, using the predecessor’s recorded launcher/environment unless explicitly supplied here.

  • Parameters:
    • launcher (Any)
    • activation_script (str | None)
    • startup_timeout (int)
  • Return type: SlurmAllocation

qos : str | None​

ray_head_only : bool​

reconcile_workers(intent, , observed=None, max_observation_age=300.0, release=(), max_submissions=None)​

Submit or release worker allocations until intent.count remain.

Runs once per call under the session lock, so step processes that express the same intent concurrently never submit it twice. Pending allocations count toward the intent. Each submission is recorded with a token before it reaches the scheduler, so an interrupted call is recovered by the next one. Allocations released to scale in get a drain deadline that is persisted before the drain starts; later calls retry the release without extending the deadline and cancel the job once it passes.

release names allocations of the intent to release in this call, whatever their order; count then applies to the others. Job IDs that are not live allocations of the intent, or that are already released, are ignored, so a caller can repeat a release. At most max_submissions allocations are submitted; 0 only maintains the intent.

The scheduler is queried again before acting. When it does not answer, or observed is older than max_observation_age seconds or was taken under another head, nothing is submitted, but known releases and cleanup continue.

remove_worker_allocation(allocation, , drain_timeout=None, wait=True)​

Drain a worker allocation’s Ray nodes, then release the job.

Ray stops placing work on the nodes and lets running tasks and actors finish for up to drain_timeout seconds (default: the same window a walltime signal gets). Each node leaves once idle, and the job ends when all have left. Work still running at the deadline is lost and retried by Ray according to its retry settings. A pending job is cancelled. Returns the Slurm state; with wait=False the drain continues in the background.

  • Parameters:
  • Return type: str

replace_head(config=None, , timeout=3600, launcher=None, activation_script=None, startup_timeout=120)​

Return a healthy Ray head, electing a new one if the current head ended.

Every step whose driver ran on a lost head can call this, from any process: the first caller elects, the others wait, and all get the same head. A healthy head is returned unchanged; to move one, call request_drain() on it first. The new head is a running published successor, else the running worker allocation with the most time left, else a new allocation shaped like the first head (config overrides that shape). A worker allocation that failed twice to host the head is passed over, and with port_strategy="fixed" none hosts it. Worker allocations re-join the new head. Payloads that ran on the old head must be launched again.

reservation : str | None​

setup_for_execution(context)​

Called by Dagster when resource is initialized for a run. This is the proper Dagster resource lifecycle hook.

  • Parameters: context (InitResourceContext | AssetExecutionContext)
  • Return type: None

signal_before_timeout : str | None​

slurm : SlurmResource​

submit_successor(config=None)​

Publish a queued successor immediately, inheriting unspecified fields.

Only one successor may be outstanding. All allocations keep the session job name so terminal-run cleanup also cancels pending successors.

teardown_after_execution(context)​

Called by Dagster when resource is torn down after run completion. This is the proper Dagster resource lifecycle hook.

  • Parameters: context (InitResourceContext)
  • Return type: None

time_limit : str​

time_min : str | None​

wait_for_worker_allocation(allocation, , timeout=600, poll_interval=5)​

Wait until every node of a worker allocation has joined the current head.

Returns the Ray node ID per Slurm node. Raises if the job ends first.

  • Parameters:
  • Return type: dict[str, str]

worker_allocation : SlurmWorkerAllocationConfig | None​

property worker_allocations : list[SlurmAllocation]​

Worker allocations of this session that have not ended.

Read from the session metadata, so a restarted supervisor finds the workers an earlier process added and can drain or cancel them.

worker_sbatch_options(configs)​

The sbatch options add_worker_allocation would submit per config.

Unset fields inherit as they would on submission. Useful to probe a candidate with the scheduler before submitting it.

exception dagster_slurm.SlurmStepDrained(result, , invocation)​

Bases: RuntimeError

A drained step, with its finalized Pipes results available to the caller.

class dagster_slurm.SlurmStepExecutionResult(job_id, stdout_path, stderr_path, step_id=None, step_id_path=None, status_path=None, drain_marker_path=None, drain_signal=None, exit_code=None, allocation_state=None)​

Bases: object

Result for one step executed inside a shared Slurm allocation.

  • Parameters:
    • job_id (int)
    • stdout_path (str)
    • stderr_path (str)
    • step_id (str | None)
    • step_id_path (str | None)
    • status_path (str | None)
    • drain_marker_path (str | None)
    • drain_signal (str | None)
    • exit_code (int | None)
    • allocation_state (str | None)

allocation_state : str | None = None​

drain_marker_path : str | None = None​

drain_signal : str | None = None​

property drained : bool​

Whether a drain signal fired during this invocation.

exit_code : int | None = None​

job_id : int​

status_path : str | None = None​

stderr_path : str​

stdout_path : str​

step_id : str | None = None​

step_id_path : str | None = None​

class dagster_slurm.SlurmWorkerAllocationConfig(**config_dict)​

Bases: SlurmRunAllocationConfig

An extra allocation whose nodes join a run’s Ray cluster as workers.

Unset Slurm fields inherit the session’s worker template (its first worker allocation) or else its current allocation, except nodelist and extra_sbatch_directives, which apply to one submission only.

ray_resources : dict[str, float]​

ray_start_args : list[str]​

rejoin_timeout : int​

class dagster_slurm.SparkLauncher(**data)​

Bases: ComputeLauncher

Apache Spark launcher.

Modes:

  • Local: Single-node Spark (no allocation_context)
  • Cluster: Spark cluster across Slurm allocation (via allocation_context)
  • Standalone: Connect to existing Spark cluster (via master_url)
  • Parameters: data (Any)

driver_memory : str​

executor_cores : int​

executor_memory : str​

master_url : str | None​

num_executors : int | None​

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Generate Spark execution plan.

  • Parameters:
    • payload_path (str)
    • python_executable (str)
    • working_dir (str)
    • pipes_context (dict[str, str])
    • extra_env (dict[str, str] | None)
    • allocation_context (dict[str, Any] | None)
    • activation_script (str | None)
  • Return type: ExecutionPlan

spark_home : str​

class dagster_slurm.StartEstimate(config, observed_at, time_limit_seconds=None, start_in_seconds=None, start_time=None, nodes=None, partition=None, error=None)​

Bases: object

The scheduler’s estimate for one candidate worker allocation.

An estimate is advisory: it reserves nothing, and other jobs can take the resources before a submission arrives.

  • Parameters:
    • config (SlurmWorkerAllocationConfig)
    • observed_at (float)
    • time_limit_seconds (int | None)
    • start_in_seconds (float | None)
    • start_time (str | None)
    • nodes (str | None)
    • partition (str | None)
    • error (str | None)

config : SlurmWorkerAllocationConfig​

error : str | None = None​

Why the scheduler gave no estimate, for example an invalid request.

nodes : str | None = None​

observed_at : float​

partition : str | None = None​

start_in_seconds : float | None = None​

Seconds until the estimated start, by the login node’s clock.

start_time : str | None = None​

Estimated start as the scheduler printed it, in the cluster’s timezone.

starts_within(seconds)​

  • Parameters: seconds (float)
  • Return type: bool

time_limit_seconds : int | None = None​

Walltime the candidate would request, including an inherited one.

class dagster_slurm.StartEstimator(*args, **kwargs)​

Bases: Protocol

Estimates start times for candidate worker allocations.

Replace the default with a site-specific provider, for example one that reads a reservation calendar or applies site knowledge to the answers.

estimate(session, candidates)​

class dagster_slurm.WorkerIntent(key, count, config=None, drain_timeout=None, pending_timeout=None)​

Bases: object

Desired worker allocations under one caller-chosen key.

Several step processes can express the same intent; reconcile_workers serializes them so the same intent is never submitted twice.

  • Parameters:

config : SlurmWorkerAllocationConfig | None = None​

Shape of new allocations; unset fields inherit as in add_worker_allocation.

count : int​

Allocations wanted, counting pending and running ones.

drain_timeout : int | None = None​

Seconds a released running allocation may drain before it is cancelled.

key : str​

pending_timeout : float | None = None​

Cancel an allocation of this intent still pending after this many seconds.

dagster_slurm.build_slurm_orphan_reconcile_sensor(slurm_resource, , name='slurm_orphan_reconcile_sensor', jobs=None, target=None, stale_after_seconds=120.0, limit=50, minimum_interval_seconds=30, default_status=DefaultSensorStatus.STOPPED)​

Build a sensor that retries orphaned Slurm-backed Dagster runs.

  • Parameters:
    • slurm_resource (SlurmResource)
    • name (str)
    • jobs (Sequence[Any] | None)
    • target (Any | None)
    • stale_after_seconds (float)
    • limit (int)
    • minimum_interval_seconds (int)
    • default_status (DefaultSensorStatus)
  • Return type: SensorDefinition

dagster_slurm.reconcile_orphaned_slurm_runs(instance, slurm_resource, , stale_after_seconds=120.0, limit=50, now=None)​

Mark dead-supervisor runs failed and return reattach retry requests.

The returned RunRequest objects carry the original Slurm job id and run directory. SlurmPipesClient consumes those tags on the retry and replays the remote messages.jsonl file instead of submitting another job.

  • Parameters:
    • instance (DagsterInstance)
    • slurm_resource (SlurmResource)
    • stale_after_seconds (float)
    • limit (int)
    • now (float | None)
  • Return type: list[RunRequest]

dagster_slurm.run_with_ray_reserve_topup(primary, reserve, minimum_resources, , stable_polls=2, poll_interval_seconds=1.0, timeout_seconds=None, resource_provider=None, provider_retry_attempts=3, provider_retry_backoff_seconds=1.0)​

Run primary work immediately and top it up after capacity stabilizes.

primary and the capacity watcher start concurrently. Once the watcher confirms persistent idle resources, reserve receives the qualifying snapshot and runs alongside any primary work still in progress. If the capacity wait times out, the primary result is returned with no reserve result. The callbacks can each create and materialize an independent Ray actor pool; callers remain responsible for merging their returned data.

  • Return type: tuple[TypeVar(PrimaryResultT), Optional[TypeVar(ReserveResultT)]]
  • Returns: A pair containing the primary result and the optional reserve result.
  • Parameters:
    • primary (Callable[[], TypeVar(PrimaryResultT)])
    • reserve (Callable[[Mapping[str, float]], TypeVar(ReserveResultT)])
    • minimum_resources (Mapping[str, float])
    • stable_polls (int)
    • poll_interval_seconds (float)
    • timeout_seconds (float | None)
    • resource_provider (Callable[[], Mapping[str, float]] | None)
    • provider_retry_attempts (int)
    • provider_retry_backoff_seconds (float)

dagster_slurm.wait_for_stable_ray_resources(minimum_resources, , stable_polls=2, poll_interval_seconds=1.0, timeout_seconds=None, resource_provider=None, provider_retry_attempts=3, provider_retry_backoff_seconds=1.0)​

Wait until Ray resources stay available across consecutive polls.

This observes capacity; it does not reserve it. Callers should submit the work that claims the returned capacity immediately. A common use is to launch a second Ray actor pool after co-tenant workloads release resources, because an actor pool that is already running cannot be resized.

  • Parameters:
    • minimum_resources (Mapping[str, float]) – Resource quantities that must all be available, such as {"GPU": 1, "CPU": 4}.
    • stable_polls (int) – Number of consecutive qualifying snapshots required.
    • poll_interval_seconds (float) – Delay between resource snapshots.
    • timeout_seconds (float | None) – Maximum wait time, or no limit when omitted.
    • resource_provider (Callable[[], Mapping[str, float]] | None) – Optional provider compatible with ray.available_resources. Primarily useful for custom Ray integrations and deterministic tests.
    • provider_retry_attempts (int) – Total attempts allowed for each resource snapshot when the provider raises.
    • provider_retry_backoff_seconds (float) – Initial retry delay. Subsequent delays use exponential backoff.
  • Return type: dict[str, float]
  • Returns: The final qualifying resource snapshot.
  • Raises:
    • ImportError – If Ray is not installed and no provider is supplied.
    • RuntimeError – If the resource provider exhausts its retry attempts.
    • TimeoutError – If capacity does not stabilize before the timeout.
    • ValueError – If polling or resource requirements are invalid.