diff --git a/.sqlx/query-13e015c082a9a9bf2b9c6807716aba1d5d9ef4efdd7c33f86a751473267f7bd7.json b/.sqlx/query-84c60a7116c68da5863d814db5336cfc175f1d41b19f69905ddee407ebbcf2fc.json similarity index 70% rename from .sqlx/query-13e015c082a9a9bf2b9c6807716aba1d5d9ef4efdd7c33f86a751473267f7bd7.json rename to .sqlx/query-84c60a7116c68da5863d814db5336cfc175f1d41b19f69905ddee407ebbcf2fc.json index 9b86ff838..58fc7e093 100644 --- a/.sqlx/query-13e015c082a9a9bf2b9c6807716aba1d5d9ef4efdd7c33f86a751473267f7bd7.json +++ b/.sqlx/query-84c60a7116c68da5863d814db5336cfc175f1d41b19f69905ddee407ebbcf2fc.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "\n INSERT INTO slurm_scheduler\n (\n workflow_id\n ,name\n ,account\n ,gres\n ,mem\n ,nodes\n ,ntasks_per_node\n ,partition\n ,qos\n ,tmp\n ,walltime\n ,extra\n )\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)\n RETURNING rowid\n ", + "query": "\n INSERT INTO slurm_scheduler\n (\n workflow_id\n ,name\n ,account\n ,gres\n ,mem\n ,nodes\n ,ntasks_per_node\n ,partition\n ,qos\n ,tmp\n ,walltime\n ,extra\n ,serialize_allocations\n )\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)\n RETURNING rowid\n ", "describe": { "columns": [ { @@ -10,11 +10,11 @@ } ], "parameters": { - "Right": 12 + "Right": 13 }, "nullable": [ false ] }, - "hash": "13e015c082a9a9bf2b9c6807716aba1d5d9ef4efdd7c33f86a751473267f7bd7" + "hash": "84c60a7116c68da5863d814db5336cfc175f1d41b19f69905ddee407ebbcf2fc" } diff --git a/api/openapi.codegen.yaml b/api/openapi.codegen.yaml index 45abf110c..954ed0577 100644 --- a/api/openapi.codegen.yaml +++ b/api/openapi.codegen.yaml @@ -7736,6 +7736,18 @@ components: type: - string - 'null' + serialize_allocations: + type: + - boolean + - 'null' + description: |- + Run this scheduler's allocations strictly one at a time. + + When set, every allocation submitted for this scheduler shares one Slurm job + name and carries `--dependency=singleton`, so Slurm serializes them. Submit N + allocations up front and they chain: each runs until its walltime can no longer + fit a ready job, exits, and the next starts. Used for long sequential workflows + that outlive any single allocation. tmp: type: - string diff --git a/api/openapi.yaml b/api/openapi.yaml index 45abf110c..954ed0577 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -7736,6 +7736,18 @@ components: type: - string - 'null' + serialize_allocations: + type: + - boolean + - 'null' + description: |- + Run this scheduler's allocations strictly one at a time. + + When set, every allocation submitted for this scheduler shares one Slurm job + name and carries `--dependency=singleton`, so Slurm serializes them. Submit N + allocations up front and they chain: each runs until its walltime can no longer + fit a ready job, exits, and the next starts. Used for long sequential workflows + that outlive any single allocation. tmp: type: - string diff --git a/docs/src/SUMMARY.md b/docs/src/SUMMARY.md index 693364db9..544df2edd 100644 --- a/docs/src/SUMMARY.md +++ b/docs/src/SUMMARY.md @@ -78,6 +78,7 @@ - [Slurm Overview](./specialized/hpc/slurm-workflows.md) - [Submitting Slurm Workflows](./specialized/hpc/submit-slurm-workflow.md) - [Multi-Node Jobs](./specialized/hpc/multi-node-jobs.md) + - [Chained Allocations](./specialized/hpc/chained-allocations.md) - [Advanced Slurm Configuration](./specialized/hpc/slurm.md) - [Slurm Exit Codes](./specialized/hpc/slurm-exit-codes.md) - [Debugging Slurm Workflows](./specialized/hpc/debugging-slurm.md) diff --git a/docs/src/core/reference/workflow-spec.md b/docs/src/core/reference/workflow-spec.md index 80778c03d..67ad601e4 100644 --- a/docs/src/core/reference/workflow-spec.md +++ b/docs/src/core/reference/workflow-spec.md @@ -266,19 +266,20 @@ A single rule within a failure handler for handling specific exit codes. Defines a Slurm HPC job scheduler configuration. -| Name | Type | Default | Description | -| ----------------- | ------- | ------------ | -------------------------------------------- | -| `name` | string | none | Name of the scheduler (used for referencing) | -| `account` | string | _required_ | Slurm account | -| `partition` | string | none | Slurm partition name | -| `nodes` | integer | `1` | Number of nodes to allocate | -| `walltime` | string | `"01:00:00"` | Wall time limit | -| `mem` | string | none | Memory specification | -| `gres` | string | none | Generic resources (e.g., GPUs) | -| `qos` | string | none | Quality of service | -| `ntasks_per_node` | integer | none | Number of tasks per node | -| `tmp` | string | none | Temporary storage specification | -| `extra` | string | none | Additional Slurm parameters | +| Name | Type | Default | Description | +| ----------------------- | ------- | ------------ | ----------------------------------------------------------------------------------------------------------------------- | +| `name` | string | none | Name of the scheduler (used for referencing) | +| `account` | string | _required_ | Slurm account | +| `partition` | string | none | Slurm partition name | +| `nodes` | integer | `1` | Number of nodes to allocate | +| `walltime` | string | `"01:00:00"` | Wall time limit | +| `mem` | string | none | Memory specification | +| `gres` | string | none | Generic resources (e.g., GPUs) | +| `qos` | string | none | Quality of service | +| `ntasks_per_node` | integer | none | Number of tasks per node | +| `tmp` | string | none | Temporary storage specification | +| `extra` | string | none | Additional Slurm parameters | +| `serialize_allocations` | boolean | `false` | Run this scheduler's allocations one at a time. See [Chained Allocations](../../specialized/hpc/chained-allocations.md) | ## ExecutionConfig diff --git a/docs/src/specialized/hpc/chained-allocations.md b/docs/src/specialized/hpc/chained-allocations.md new file mode 100644 index 000000000..3bdae331a --- /dev/null +++ b/docs/src/specialized/hpc/chained-allocations.md @@ -0,0 +1,102 @@ +# Chained Allocations + +Some workflows need more wall time than any single Slurm allocation provides. A chain of 500 +sequential jobs at 2-4 hours each is weeks of serial work, but partitions typically cap allocations +at hours. You need one allocation to run as many jobs as fit, exit, and the next to pick up where it +left off. + +Set `serialize_allocations` on a Slurm scheduler and torc submits every allocation for that +scheduler under one shared Slurm job name with `--dependency=singleton`. Slurm then runs them +strictly one at a time. Submit them all up front and they chain themselves, with no long-running +process on the login node. + +## Configuring a Scheduler + +```yaml +slurm_schedulers: + - name: chain + account: my_account + walltime: "12:00:00" + nodes: 1 + serialize_allocations: true + +resource_requirements: + - name: serial + num_cpus: 104 + memory: "200g" + runtime: "PT4H" +``` + +A complete runnable spec is at `examples/yaml/chained_allocations.yaml`. + +Or on an existing scheduler: + +```bash +torc slurm create -n chain -a my_account -W 12:00:00 --serialize-allocations +torc slurm update --serialize-allocations true +``` + +Then submit the whole chain at once: + +```bash +torc slurm schedule-nodes -n 167 +``` + +All 167 allocations enter the queue immediately. Slurm starts one, holds the rest, and releases the +next each time the current one ends. + +## How Many Allocations to Submit + +Divide the total work by what one allocation can absorb. A worker claims jobs until its remaining +wall time can no longer fit the next one, so with a 12-hour walltime and a declared `runtime` of +`PT4H`, each allocation completes at least three jobs: + +``` +allocations = ceil(total_jobs / floor(walltime / runtime)) + = ceil(500 / floor(12 / 4)) = 167 +``` + +Round up. Over-submitting is cheap: once the workflow has no runnable jobs left, the finishing +worker cancels every allocation still queued for the workflow, so the surplus never starts. + +## Why the Chain Beats Scheduling on Shutdown + +A worker could submit its own replacement as it exits, but that pays Slurm's full +submit-and-schedule latency at every link — 167 times over — and makes each link's submission depend +on its predecessor exiting cleanly. A chained allocation is already sitting in the queue when its +predecessor ends, so it is typically released the moment the slot frees. On clusters that accrue age +priority for dependency-held jobs (`PriorityFlags=ACCRUE_ALWAYS` in the Slurm config), each link +also builds priority while its predecessor runs; by default Slurm starts the age clock only when the +dependency clears. + +## Wall Time and Job Runtime + +The server only hands a worker jobs whose declared `runtime` fits in the allocation's remaining wall +time. This is what makes the handoff clean: an allocation stops claiming once the next job no longer +fits, idles briefly, and exits, releasing the slot early rather than sitting idle until walltime. + +Declare `runtime` at or above the worst case you expect. If a job overruns its declared runtime it +is terminated when the walltime expires, and jobs downstream of it are left blocked with nothing to +unblock them — which ends the chain, since the remaining queued allocations have no runnable work. + +## Scope of the Chain + +The shared job name is derived from the workflow ID and the scheduler ID, so: + +- Two schedulers in one workflow chain independently. +- The same workflow submitted twice chains independently per scheduler. +- Allocations added later — by a second `schedule-nodes` call, or by a `schedule_nodes` action + firing from a compute node — join the existing chain rather than running alongside it. + +Slurm scopes `singleton` to a job name **per user**, so the name is prefixed with `torc-` to keep +the chain from serializing against your unrelated Slurm jobs. + +Because the chain depends on every allocation sharing one fixed name, +`torc slurm schedule-nodes --job-prefix` is rejected for a serialized scheduler: a per-invocation +prefix would change the name and fork the chain. + +## Interaction with `extra` + +`extra` is emitted after torc's own `#SBATCH` directives, so a `--dependency` set there overrides +the generated `--dependency=singleton` and breaks the chain. Use `extra` for unrelated flags +(`--reservation`, `--constraint`) when serializing allocations. diff --git a/examples/yaml/chained_allocations.yaml b/examples/yaml/chained_allocations.yaml new file mode 100644 index 000000000..f558965e9 --- /dev/null +++ b/examples/yaml/chained_allocations.yaml @@ -0,0 +1,58 @@ +name: "Chained Allocations" +description: "Sequential jobs spanning multiple Slurm allocations chained with serialize_allocations" + +# A sequence of jobs whose total runtime exceeds any single allocation's walltime. +# The scheduler below sets serialize_allocations, so every allocation is submitted +# under one shared Slurm job name with --dependency=singleton: Slurm runs them one +# at a time, and each picks up where its predecessor left off. +# +# See docs: specialized/hpc/chained-allocations.md + +jobs: + - name: "step1" + command: "echo 'Running step 1' && sleep 10" + resource_requirements: "serial" + + - name: "step2" + command: "echo 'Running step 2' && sleep 10" + depends_on: ["step1"] + resource_requirements: "serial" + + - name: "step3" + command: "echo 'Running step 3' && sleep 10" + depends_on: ["step2"] + resource_requirements: "serial" + + - name: "step4" + command: "echo 'Running step 4' && sleep 10" + depends_on: ["step3"] + resource_requirements: "serial" + +resource_requirements: + # Declare runtime at or above the worst case: the server only hands a worker a job + # whose declared runtime fits the allocation's remaining walltime, which is what + # makes each allocation exit cleanly when the next job no longer fits. + - name: "serial" + num_cpus: 4 + num_nodes: 1 + memory: "8g" + runtime: "PT2H" + +slurm_schedulers: + - name: "chain" + account: "demo_project" + nodes: 1 + walltime: "04:00:00" + serialize_allocations: true + +actions: + # Submit the whole chain up front. With a 4-hour walltime and 2-hour job runtime, + # each allocation completes floor(4 / 2) = 2 jobs, so 4 jobs need + # ceil(4 / 2) = 2 allocations. Over-submitting is safe: once no runnable jobs + # remain, the finishing worker cancels the surplus still sitting in the queue. + - trigger_type: "on_jobs_ready" + action_type: "schedule_nodes" + jobs: ["step1"] + scheduler: "chain" + scheduler_type: "slurm" + num_allocations: 2 diff --git a/julia_client/Torc/src/api/models/model_SlurmSchedulerModel.jl b/julia_client/Torc/src/api/models/model_SlurmSchedulerModel.jl index 228b08221..5d2140528 100644 --- a/julia_client/Torc/src/api/models/model_SlurmSchedulerModel.jl +++ b/julia_client/Torc/src/api/models/model_SlurmSchedulerModel.jl @@ -15,6 +15,7 @@ ntasks_per_node=nothing, partition=nothing, qos=nothing, + serialize_allocations=nothing, tmp=nothing, walltime=nothing, workflow_id=nothing, @@ -30,6 +31,7 @@ - ntasks_per_node::Int64 - partition::String - qos::String + - serialize_allocations::Bool : Run this scheduler's allocations strictly one at a time. When set, every allocation submitted for this scheduler shares one Slurm job name and carries `--dependency=singleton`, so Slurm serializes them. Submit N allocations up front and they chain: each runs until its walltime can no longer fit a ready job, exits, and the next starts. Used for long sequential workflows that outlive any single allocation. - tmp::String - walltime::String - workflow_id::Int64 @@ -45,18 +47,19 @@ Base.@kwdef mutable struct SlurmSchedulerModel <: OpenAPI.APIModel ntasks_per_node::Union{Nothing, Int64} = nothing partition::Union{Nothing, String} = nothing qos::Union{Nothing, String} = nothing + serialize_allocations::Union{Nothing, Bool} = nothing tmp::Union{Nothing, String} = nothing walltime::Union{Nothing, String} = nothing workflow_id::Union{Nothing, Int64} = nothing - function SlurmSchedulerModel(account, extra, gres, id, mem, name, nodes, ntasks_per_node, partition, qos, tmp, walltime, workflow_id, ) - o = new(account, extra, gres, id, mem, name, nodes, ntasks_per_node, partition, qos, tmp, walltime, workflow_id, ) + function SlurmSchedulerModel(account, extra, gres, id, mem, name, nodes, ntasks_per_node, partition, qos, serialize_allocations, tmp, walltime, workflow_id, ) + o = new(account, extra, gres, id, mem, name, nodes, ntasks_per_node, partition, qos, serialize_allocations, tmp, walltime, workflow_id, ) OpenAPI.validate_properties(o) return o end end # type SlurmSchedulerModel -const _property_types_SlurmSchedulerModel = Dict{Symbol,String}(Symbol("account")=>"String", Symbol("extra")=>"String", Symbol("gres")=>"String", Symbol("id")=>"Int64", Symbol("mem")=>"String", Symbol("name")=>"String", Symbol("nodes")=>"Int64", Symbol("ntasks_per_node")=>"Int64", Symbol("partition")=>"String", Symbol("qos")=>"String", Symbol("tmp")=>"String", Symbol("walltime")=>"String", Symbol("workflow_id")=>"Int64", ) +const _property_types_SlurmSchedulerModel = Dict{Symbol,String}(Symbol("account")=>"String", Symbol("extra")=>"String", Symbol("gres")=>"String", Symbol("id")=>"Int64", Symbol("mem")=>"String", Symbol("name")=>"String", Symbol("nodes")=>"Int64", Symbol("ntasks_per_node")=>"Int64", Symbol("partition")=>"String", Symbol("qos")=>"String", Symbol("serialize_allocations")=>"Bool", Symbol("tmp")=>"String", Symbol("walltime")=>"String", Symbol("workflow_id")=>"Int64", ) OpenAPI.property_type(::Type{ SlurmSchedulerModel }, name::Symbol) = Union{Nothing,eval(Base.Meta.parse(_property_types_SlurmSchedulerModel[name]))} function OpenAPI.check_required(o::SlurmSchedulerModel) @@ -78,6 +81,7 @@ function OpenAPI.validate_properties(o::SlurmSchedulerModel) OpenAPI.validate_property(SlurmSchedulerModel, Symbol("ntasks_per_node"), o.ntasks_per_node) OpenAPI.validate_property(SlurmSchedulerModel, Symbol("partition"), o.partition) OpenAPI.validate_property(SlurmSchedulerModel, Symbol("qos"), o.qos) + OpenAPI.validate_property(SlurmSchedulerModel, Symbol("serialize_allocations"), o.serialize_allocations) OpenAPI.validate_property(SlurmSchedulerModel, Symbol("tmp"), o.tmp) OpenAPI.validate_property(SlurmSchedulerModel, Symbol("walltime"), o.walltime) OpenAPI.validate_property(SlurmSchedulerModel, Symbol("workflow_id"), o.workflow_id) @@ -106,6 +110,7 @@ function OpenAPI.validate_property(::Type{ SlurmSchedulerModel }, name::Symbol, + if name === Symbol("workflow_id") OpenAPI.validate_param(name, "SlurmSchedulerModel", :format, val, "int64") end diff --git a/julia_client/julia_client/docs/SlurmSchedulerModel.md b/julia_client/julia_client/docs/SlurmSchedulerModel.md index d5da8c1c5..23f75ae8a 100644 --- a/julia_client/julia_client/docs/SlurmSchedulerModel.md +++ b/julia_client/julia_client/docs/SlurmSchedulerModel.md @@ -14,6 +14,7 @@ Name | Type | Description | Notes **ntasks_per_node** | **Int64** | | [optional] [default to nothing] **partition** | **String** | | [optional] [default to nothing] **qos** | **String** | | [optional] [default to nothing] +**serialize_allocations** | **Bool** | Run this scheduler's allocations strictly one at a time. When set, every allocation submitted for this scheduler shares one Slurm job name and carries `--dependency=singleton`, so Slurm serializes them. Submit N allocations up front and they chain: each runs until its walltime can no longer fit a ready job, exits, and the next starts. Used for long sequential workflows that outlive any single allocation. | [optional] [default to nothing] **tmp** | **String** | | [optional] [default to nothing] **walltime** | **String** | | [default to nothing] **workflow_id** | **Int64** | | [default to nothing] diff --git a/python_client/src/torc/openapi_client/models/slurm_scheduler_model.py b/python_client/src/torc/openapi_client/models/slurm_scheduler_model.py index 8911e2064..e7b634d4c 100644 --- a/python_client/src/torc/openapi_client/models/slurm_scheduler_model.py +++ b/python_client/src/torc/openapi_client/models/slurm_scheduler_model.py @@ -17,7 +17,7 @@ import re # noqa: F401 import json -from pydantic import BaseModel, ConfigDict, StrictInt, StrictStr +from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt, StrictStr from typing import Any, ClassVar, Dict, List, Optional from typing import Optional, Set from typing_extensions import Self @@ -36,10 +36,11 @@ class SlurmSchedulerModel(BaseModel): ntasks_per_node: Optional[StrictInt] = None partition: Optional[StrictStr] = None qos: Optional[StrictStr] = None + serialize_allocations: Optional[StrictBool] = Field(default=None, description="Run this scheduler's allocations strictly one at a time. When set, every allocation submitted for this scheduler shares one Slurm job name and carries `--dependency=singleton`, so Slurm serializes them. Submit N allocations up front and they chain: each runs until its walltime can no longer fit a ready job, exits, and the next starts. Used for long sequential workflows that outlive any single allocation.") tmp: Optional[StrictStr] = None walltime: StrictStr workflow_id: StrictInt - __properties: ClassVar[List[str]] = ["account", "extra", "gres", "id", "mem", "name", "nodes", "ntasks_per_node", "partition", "qos", "tmp", "walltime", "workflow_id"] + __properties: ClassVar[List[str]] = ["account", "extra", "gres", "id", "mem", "name", "nodes", "ntasks_per_node", "partition", "qos", "serialize_allocations", "tmp", "walltime", "workflow_id"] model_config = ConfigDict( populate_by_name=True, @@ -120,6 +121,11 @@ def to_dict(self) -> Dict[str, Any]: if self.qos is None and "qos" in self.model_fields_set: _dict['qos'] = None + # set to None if serialize_allocations (nullable) is None + # and model_fields_set contains the field + if self.serialize_allocations is None and "serialize_allocations" in self.model_fields_set: + _dict['serialize_allocations'] = None + # set to None if tmp (nullable) is None # and model_fields_set contains the field if self.tmp is None and "tmp" in self.model_fields_set: @@ -147,6 +153,7 @@ def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: "ntasks_per_node": obj.get("ntasks_per_node"), "partition": obj.get("partition"), "qos": obj.get("qos"), + "serialize_allocations": obj.get("serialize_allocations"), "tmp": obj.get("tmp"), "walltime": obj.get("walltime"), "workflow_id": obj.get("workflow_id") diff --git a/src/client/commands/slurm.rs b/src/client/commands/slurm.rs index ef7b63c63..160d99345 100644 --- a/src/client/commands/slurm.rs +++ b/src/client/commands/slurm.rs @@ -164,6 +164,8 @@ struct SlurmSchedulerTableRow { partition: String, #[tabled(rename = "QOS")] qos: String, + #[tabled(rename = "Serialized")] + serialize_allocations: bool, } /// Select a Slurm scheduler interactively from available schedulers for a workflow @@ -306,6 +308,14 @@ EXAMPLES: /// Add extra Slurm parameters, for example --extra='--reservation=my-reservation' #[arg(short, long)] extra: Option, + /// Run this scheduler's allocations one at a time instead of concurrently + /// + /// All allocations submitted for this scheduler share one Slurm job name and + /// carry --dependency=singleton, so Slurm chains them. Submit N allocations up + /// front and each starts as its predecessor finishes. Use for sequential work + /// that needs more wall time than a single allocation provides. + #[arg(long, default_value = "false")] + serialize_allocations: bool, }, /// Modify a Slurm config in the database #[command(hide = true)] @@ -342,6 +352,11 @@ EXAMPLES: /// Add extra Slurm parameters #[arg(short, long)] extra: Option, + /// Run this scheduler's allocations one at a time instead of concurrently + /// + /// See `torc slurm create --help` for what serialization does. + #[arg(long)] + serialize_allocations: Option, }, /// Show the current Slurm configs in the database #[command( @@ -412,6 +427,9 @@ EXAMPLES: #[arg(long, default_value = "false")] start_one_worker_per_node: bool, /// Job prefix for the Slurm job names + /// + /// Not allowed when the scheduler has serialize_allocations set: chained + /// allocations must all share one fixed Slurm job name. #[arg(short, long, default_value = "")] job_prefix: String, /// Keep submission scripts after job submission @@ -1083,6 +1101,7 @@ pub fn handle_slurm_commands(config: &Configuration, command: &SlurmCommands, fo tmp, walltime, extra, + serialize_allocations, } => { let user_name = get_env_user_name(); let wf_id = workflow_id.unwrap_or_else(|| { @@ -1106,6 +1125,7 @@ pub fn handle_slurm_commands(config: &Configuration, command: &SlurmCommands, fo tmp: tmp.clone(), walltime: walltime.clone(), extra: extra.clone(), + serialize_allocations: Some(*serialize_allocations), }; match apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) { @@ -1139,6 +1159,7 @@ pub fn handle_slurm_commands(config: &Configuration, command: &SlurmCommands, fo tmp, walltime, extra, + serialize_allocations, } => { let mut scheduler = match apis::slurm_schedulers_api::get_slurm_scheduler(config, *scheduler_id) { @@ -1191,6 +1212,10 @@ pub fn handle_slurm_commands(config: &Configuration, command: &SlurmCommands, fo scheduler.extra = Some(e.clone()); changed = true; } + if let Some(s) = serialize_allocations { + scheduler.serialize_allocations = Some(*s); + changed = true; + } if !changed { warn!("No changes requested"); @@ -1248,6 +1273,7 @@ pub fn handle_slurm_commands(config: &Configuration, command: &SlurmCommands, fo walltime: s.walltime.clone(), partition: s.partition.clone().unwrap_or_default(), qos: s.qos.clone().unwrap_or_default(), + serialize_allocations: s.serialize_allocations.unwrap_or(false), }) .collect(); @@ -1295,6 +1321,10 @@ pub fn handle_slurm_commands(config: &Configuration, command: &SlurmCommands, fo " Extra: {}", scheduler.extra.unwrap_or_else(|| "None".to_string()) ); + eprintln!( + " Serialize allocations: {}", + scheduler.serialize_allocations.unwrap_or(false) + ); } } Err(e) => { @@ -2122,6 +2152,25 @@ pub fn review_submit_pending_actions( Ok(overrides) } +/// Build the shared Slurm job name used by a scheduler with `serialize_allocations`. +/// +/// `--dependency=singleton` is scoped to (user, job name), so every allocation that +/// should chain must submit under exactly this name -- including allocations submitted +/// by separate processes (a later `schedule-nodes` call, or a `schedule_nodes` action +/// fired from a compute node). That rules out the per-allocation name, which embeds the +/// submitting PID and a counter. +/// +/// Keying on `(workflow_id, scheduler_id)` keeps the name stable across those processes +/// while scoping the chain to one scheduler: two schedulers in one workflow, or the same +/// scheduler in two workflows, get distinct names and run independently. The `torc-` +/// prefix keeps the singleton set from colliding with the user's unrelated Slurm jobs, +/// which would otherwise serialize against this chain. The user's `--job-prefix` is +/// deliberately absent -- a per-invocation prefix would fork the chain, so +/// `schedule_slurm_nodes` rejects it for serialized schedulers. +fn serialized_slurm_job_name(workflow_id: i64, scheduler_id: i64) -> String { + format!("torc-wf{}-sched{}", workflow_id, scheduler_id) +} + /// Result indicating success or failure #[allow(clippy::too_many_arguments)] pub fn schedule_slurm_nodes( @@ -2153,6 +2202,22 @@ pub fn schedule_slurm_nodes( } }; + // A serialized scheduler submits every allocation under one fixed job name (see + // serialized_slurm_job_name); a per-invocation prefix would change that name and + // fork the chain, since later submissions -- a schedule_nodes action fired from a + // compute node, or a top-up schedule-nodes call without the flag -- carry no + // prefix. Reject it rather than silently splitting the chain. + let serialize_allocations = scheduler.serialize_allocations.unwrap_or(false); + if serialize_allocations && !job_prefix.is_empty() { + return Err(format!( + "--job-prefix is not supported for scheduler_id={} because it has \ + serialize_allocations set: every allocation in the chain must share one \ + Slurm job name", + scheduler_config_id + ) + .into()); + } + // Fetch workflow to get slurm_defaults let workflow = match utils::send_with_retries( config, @@ -2222,6 +2287,15 @@ pub fn schedule_slurm_nodes( config_map.insert("extra".to_string(), extra.clone()); } + // Serialized schedulers chain their allocations: all of them share one Slurm job + // name and carry --dependency=singleton, which Slurm scopes to (user, job name). + // Inserted after slurm_defaults so a workflow-level `dependency` default cannot + // silently break the chain. A `--dependency` in `extra` still wins: `extra` is + // emitted last, and overriding it is the documented escape hatch. + if serialize_allocations { + config_map.insert("dependency".to_string(), "singleton".to_string()); + } + std::fs::create_dir_all(output)?; // Compute startup jitter window for thundering herd mitigation. @@ -2232,11 +2306,19 @@ pub fn schedule_slurm_nodes( .get("nodes") .and_then(|v| v.parse().ok()) .unwrap_or(1); - let total_runners = if start_one_worker_per_node { - num_hpc_jobs * nodes_per_alloc + // Serialized allocations run one at a time, so only the runners inside a + // single allocation ever start together; sizing the window to the whole + // chain would just add dead time to every link. + let allocs_at_once = if serialize_allocations { + 1 } else { num_hpc_jobs }; + let total_runners = if start_one_worker_per_node { + allocs_at_once * nodes_per_alloc + } else { + allocs_at_once + }; let delay = compute_startup_delay(total_runners.max(0) as u32); if delay > 0 { info!( @@ -2257,13 +2339,20 @@ pub fn schedule_slurm_nodes( std::process::id(), job_num ); + // The submission script keeps the unique name so concurrent submissions never + // overwrite each other's files, even when the Slurm job name is shared. let script_path = format!("{}/{}.sh", output, job_name); + let slurm_job_name = if serialize_allocations { + serialized_slurm_job_name(workflow_id, scheduler_config_id) + } else { + job_name.clone() + }; let tls_ca_cert = config.tls.ca_cert_path.as_ref().and_then(|p| p.to_str()); let tls_insecure = config.tls.insecure; if let Err(e) = slurm_interface.create_submission_script( - &job_name, + &slurm_job_name, &config.base_path, workflow_id, output, @@ -2325,7 +2414,7 @@ pub fn schedule_slurm_nodes( .expect("Created scheduled compute node should have an ID"); info!( "Submitted Slurm job name={} with ID={} (scheduled_compute_node_id={})", - job_name, slurm_job_id_int, scn_id + slurm_job_name, slurm_job_id_int, scn_id ); } Err(e) => { @@ -5375,6 +5464,9 @@ fn handle_regenerate( qos: planned.qos.clone(), tmp: None, extra: None, + // Auto-generated schedulers size themselves for parallel work; serializing + // is an explicit opt-in on a user-authored scheduler. + serialize_allocations: None, }; let created_scheduler = match utils::send_with_retries( @@ -5816,4 +5908,29 @@ mod tests { assert_eq!(parse_walltime_secs("1-00:00:00").unwrap(), 24 * 3600); assert_eq!(parse_walltime_secs("30:00").unwrap(), 30 * 60); } + + /// The chain only forms if every allocation submits under the same name, so the + /// name must not vary with anything process-local (PID, allocation counter). + #[test] + fn test_serialized_slurm_job_name_is_stable() { + assert_eq!( + serialized_slurm_job_name(42, 7), + serialized_slurm_job_name(42, 7) + ); + assert_eq!(serialized_slurm_job_name(42, 7), "torc-wf42-sched7"); + } + + /// Distinct schedulers must not share a singleton set, or unrelated allocations + /// would serialize against each other. + #[test] + fn test_serialized_slurm_job_name_scopes_to_workflow_and_scheduler() { + assert_ne!( + serialized_slurm_job_name(42, 7), + serialized_slurm_job_name(42, 8) + ); + assert_ne!( + serialized_slurm_job_name(42, 7), + serialized_slurm_job_name(43, 7) + ); + } } diff --git a/src/client/scheduler_plan.rs b/src/client/scheduler_plan.rs index 75ec12edd..46e69f68f 100644 --- a/src/client/scheduler_plan.rs +++ b/src/client/scheduler_plan.rs @@ -1038,6 +1038,9 @@ pub fn apply_plan_to_spec(plan: &SchedulerPlan, spec: &mut WorkflowSpec) { qos: ps.qos.clone(), tmp: None, extra: None, + // Planned schedulers are sized for parallel work; serializing is an + // explicit opt-in on a user-authored scheduler. + serialize_allocations: None, }) .collect(); diff --git a/src/client/workflow_spec.rs b/src/client/workflow_spec.rs index e99f1fbcb..7a8635953 100644 --- a/src/client/workflow_spec.rs +++ b/src/client/workflow_spec.rs @@ -565,6 +565,14 @@ pub struct SlurmSchedulerSpec { /// Extra parameters #[serde(skip_serializing_if = "Option::is_none")] pub extra: Option, + /// Run this scheduler's allocations strictly one at a time. + /// + /// Every allocation submitted for this scheduler shares one Slurm job name and + /// carries `--dependency=singleton`, so Slurm chains them instead of running them + /// concurrently. Submit N allocations up front and each starts as its predecessor + /// finishes -- useful when a workflow's sequential work outlives a single walltime. + #[serde(skip_serializing_if = "Option::is_none")] + pub serialize_allocations: Option, } impl SlurmSchedulerSpec { @@ -3909,6 +3917,7 @@ impl WorkflowSpec { tmp: scheduler_spec.tmp.clone(), walltime: scheduler_spec.walltime.clone(), extra: scheduler_spec.extra.clone(), + serialize_allocations: scheduler_spec.serialize_allocations, }; let created_scheduler = @@ -5045,6 +5054,14 @@ impl WorkflowSpec { ); } } + "serialize_allocations" => { + if let Some(v) = child.entries().first().and_then(|e| e.value().as_bool()) { + obj.insert( + "serialize_allocations".to_string(), + serde_json::Value::Bool(v), + ); + } + } _ => {} } } @@ -6118,6 +6135,12 @@ impl WorkflowSpec { if let Some(ref extra) = sched.extra { lines.push(format!(" extra {}", escape(extra))); } + if let Some(serialize) = sched.serialize_allocations { + lines.push(format!( + " serialize_allocations {}", + if serialize { "#true" } else { "#false" } + )); + } lines.push("}".to_string()); } diff --git a/src/models.rs b/src/models.rs index f0d15b922..b4e1713a5 100644 --- a/src/models.rs +++ b/src/models.rs @@ -525,6 +525,15 @@ pub struct SlurmSchedulerModel { pub walltime: String, #[serde(skip_serializing_if = "Option::is_none")] pub extra: Option, + /// Run this scheduler's allocations strictly one at a time. + /// + /// When set, every allocation submitted for this scheduler shares one Slurm job + /// name and carries `--dependency=singleton`, so Slurm serializes them. Submit N + /// allocations up front and they chain: each runs until its walltime can no longer + /// fit a ready job, exits, and the next starts. Used for long sequential workflows + /// that outlive any single allocation. + #[serde(skip_serializing_if = "Option::is_none")] + pub serialize_allocations: Option, } #[cfg_attr(feature = "openapi-codegen", derive(utoipa::ToSchema))] @@ -1945,6 +1954,7 @@ impl SlurmSchedulerModel { tmp: None, walltime, extra: None, + serialize_allocations: None, } } } diff --git a/src/openapi_spec.rs b/src/openapi_spec.rs index 4f857e1bd..e3d57053f 100644 --- a/src/openapi_spec.rs +++ b/src/openapi_spec.rs @@ -1804,6 +1804,7 @@ pub fn parity_report(source: &str) -> Result, Box("serialize_allocations") != 0), }; Ok(GetSlurmSchedulerResponse::SuccessfulResponse( @@ -918,7 +929,7 @@ where ); // Build base query - let base_query = "SELECT id, workflow_id, name, account, gres, mem, nodes, ntasks_per_node, partition, qos, tmp, walltime, extra FROM slurm_scheduler".to_string(); + let base_query = "SELECT id, workflow_id, name, account, gres, mem, nodes, ntasks_per_node, partition, qos, tmp, walltime, extra, serialize_allocations FROM slurm_scheduler".to_string(); // Build WHERE clause let where_clause = "workflow_id = ?".to_string(); @@ -976,6 +987,7 @@ where tmp: record.get("tmp"), walltime: record.get("walltime"), extra: record.get("extra"), + serialize_allocations: Some(record.get::("serialize_allocations") != 0), }); } @@ -1215,7 +1227,8 @@ where ,tmp = COALESCE($10, tmp) ,walltime = COALESCE($11, walltime) ,extra = COALESCE($12, extra) - WHERE id = $13 + ,serialize_allocations = COALESCE($13, serialize_allocations) + WHERE id = $14 "#, ) .bind(body.workflow_id) @@ -1230,6 +1243,7 @@ where .bind(body.tmp) .bind(body.walltime) .bind(body.extra) + .bind(body.serialize_allocations) .bind(id) .execute(self.context.pool.as_ref()) .await diff --git a/tests/test_auto_schedule.rs b/tests/test_auto_schedule.rs index 6bbe4c721..fa738785e 100644 --- a/tests/test_auto_schedule.rs +++ b/tests/test_auto_schedule.rs @@ -238,6 +238,7 @@ fn test_create_slurm_scheduler(start_server: &ServerProcess) { qos: None, tmp: None, extra: None, + serialize_allocations: None, }; let created = apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) @@ -245,6 +246,10 @@ fn test_create_slurm_scheduler(start_server: &ServerProcess) { assert!(created.id.is_some(), "Scheduler should have an ID"); assert_eq!(created.account, "test_account"); + // The create response must report the stored value: the column is NOT NULL, so + // an omitted serialize_allocations comes back as an explicit false, matching + // what get/list return. + assert_eq!(created.serialize_allocations, Some(false)); // Verify we can list the scheduler let response = apis::slurm_schedulers_api::list_slurm_schedulers( @@ -283,6 +288,7 @@ fn test_create_scheduled_compute_node(start_server: &ServerProcess) { qos: None, tmp: None, extra: None, + serialize_allocations: None, }; let created_scheduler = apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) diff --git a/tests/test_hpc.rs b/tests/test_hpc.rs index a67896399..839d32427 100644 --- a/tests/test_hpc.rs +++ b/tests/test_hpc.rs @@ -629,6 +629,7 @@ fn test_generate_schedulers_existing_schedulers_no_force() { qos: None, tmp: None, extra: None, + serialize_allocations: None, }]), ..Default::default() }; @@ -688,6 +689,7 @@ fn test_generate_schedulers_existing_schedulers_with_force() { qos: None, tmp: None, extra: None, + serialize_allocations: None, }]), ..Default::default() }; diff --git a/tests/test_orphaned_jobs.rs b/tests/test_orphaned_jobs.rs index de0b1f3a8..47a6ffba1 100644 --- a/tests/test_orphaned_jobs.rs +++ b/tests/test_orphaned_jobs.rs @@ -57,6 +57,7 @@ fn create_test_slurm_scheduler( tmp: Some("100G".to_string()), walltime: "04:00:00".to_string(), extra: None, + serialize_allocations: None, }; apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) .expect("Failed to create test Slurm scheduler") diff --git a/tests/test_recover.rs b/tests/test_recover.rs index 33455f708..cb5401d9b 100644 --- a/tests/test_recover.rs +++ b/tests/test_recover.rs @@ -380,6 +380,7 @@ fn workflow_has_schedulers_reflects_scheduler_presence(start_server: &ServerProc tmp: None, walltime: "01:00:00".to_string(), extra: None, + serialize_allocations: None, }; apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) .expect("create scheduler"); @@ -432,6 +433,7 @@ fn detect_recovery_execution_mode_distinguishes_local_remote_slurm(start_server: tmp: None, walltime: "01:00:00".to_string(), extra: None, + serialize_allocations: None, }; apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) .expect("create scheduler"); diff --git a/tests/test_scheduled_compute_nodes.rs b/tests/test_scheduled_compute_nodes.rs index 1211c5cea..98794dfcd 100644 --- a/tests/test_scheduled_compute_nodes.rs +++ b/tests/test_scheduled_compute_nodes.rs @@ -25,6 +25,7 @@ fn create_test_slurm_scheduler( tmp: Some("100G".to_string()), walltime: "04:00:00".to_string(), extra: None, + serialize_allocations: None, }; apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) .expect("Failed to create test Slurm scheduler") diff --git a/tests/test_slurm_commands.rs b/tests/test_slurm_commands.rs index ac42eb8ff..97873f0ff 100644 --- a/tests/test_slurm_commands.rs +++ b/tests/test_slurm_commands.rs @@ -487,6 +487,114 @@ fn test_create_submission_script_with_extra() { let _ = fs::remove_file(&script_path); } +/// A serialized scheduler chains its allocations by submitting them all under one +/// Slurm job name with `--dependency=singleton`. Both halves are required: singleton +/// is scoped to (user, job name), so the shared name is what makes the dependency +/// select the right set of jobs. +#[test] +fn test_create_submission_script_with_singleton_dependency() { + let interface = SlurmInterface::new().expect("Failed to create SlurmInterface"); + + let temp_dir = env::temp_dir(); + let script_path = temp_dir.join("test_submission_script_singleton.sh"); + + let mut config = std::collections::HashMap::new(); + config.insert("account".to_string(), "test_account".to_string()); + config.insert("walltime".to_string(), "12:00:00".to_string()); + config.insert("dependency".to_string(), "singleton".to_string()); + + let result = interface.create_submission_script( + "torc-wf42-sched7", + "http://localhost:8080/torc-service/v1", + 42, + "/tmp/output", + 10, + None, + &script_path, + &config, + false, + None, + None, + false, + 0, + None, + ); + + assert!( + result.is_ok(), + "Failed to create submission script: {:?}", + result.err() + ); + + let script_content = + fs::read_to_string(&script_path).expect("Failed to read submission script"); + + assert!( + script_content.contains("#SBATCH --dependency=singleton"), + "Should carry the singleton dependency, got:\n{}", + script_content + ); + assert!( + script_content.contains("#SBATCH --job-name=torc-wf42-sched7"), + "Should submit under the shared chain name, got:\n{}", + script_content + ); + + let _ = fs::remove_file(&script_path); +} + +/// The chain name is fixed per (workflow, scheduler); a per-invocation `--job-prefix` +/// would change it and fork the chain, so scheduling must reject the combination +/// instead of silently splitting the singleton set. +#[rstest] +fn test_schedule_nodes_rejects_job_prefix_for_serialized_scheduler(start_server: &ServerProcess) { + let config = &start_server.config; + + let workflow = create_test_workflow(config, "test_serialized_job_prefix_workflow"); + let workflow_id = workflow.id.unwrap(); + + let scheduler = models::SlurmSchedulerModel { + id: None, + workflow_id, + name: Some("chain".to_string()), + account: "test_account".to_string(), + gres: None, + mem: None, + nodes: 1, + ntasks_per_node: None, + partition: None, + qos: None, + tmp: None, + walltime: "01:00:00".to_string(), + extra: None, + serialize_allocations: Some(true), + }; + let scheduler = apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) + .expect("Failed to create Slurm scheduler"); + let scheduler_id = scheduler.id.unwrap(); + + let result = torc::client::commands::slurm::schedule_slurm_nodes( + config, + workflow_id, + scheduler_id, + 1, + false, + "run1_", + "torc_output", + 30, + None, + false, + None, + ); + + let err = result.expect_err("job_prefix must be rejected for a serialized scheduler"); + assert!( + err.to_string().contains("--job-prefix"), + "Error should explain the job_prefix rejection, got: {}", + err + ); +} + #[test] fn test_create_submission_script_without_srun() { let interface = SlurmInterface::new().expect("Failed to create SlurmInterface"); @@ -1924,6 +2032,7 @@ fn create_test_slurm_scheduler( tmp: Some("50G".to_string()), walltime: "01:00:00".to_string(), extra: None, + serialize_allocations: None, }; apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) .expect("Failed to create Slurm scheduler") diff --git a/tests/test_slurm_regenerate.rs b/tests/test_slurm_regenerate.rs index b2e25ed4b..a43d112bf 100644 --- a/tests/test_slurm_regenerate.rs +++ b/tests/test_slurm_regenerate.rs @@ -557,6 +557,7 @@ fn test_regenerate_uses_existing_account(start_server: &ServerProcess) { qos: None, tmp: None, extra: None, + serialize_allocations: None, }; apis::slurm_schedulers_api::create_slurm_scheduler(config, scheduler) .expect("Failed to create scheduler"); diff --git a/tests/test_workflow_spec.rs b/tests/test_workflow_spec.rs index ca7d8c27b..6f9ecdbaa 100644 --- a/tests/test_workflow_spec.rs +++ b/tests/test_workflow_spec.rs @@ -202,6 +202,7 @@ fn test_workflow_specification_complete_serialization() { tmp: Some("10G".to_string()), walltime: "01:00:00".to_string(), extra: None, + serialize_allocations: None, }, SlurmSchedulerSpec { name: Some("gpu".to_string()), @@ -215,6 +216,7 @@ fn test_workflow_specification_complete_serialization() { tmp: Some("50G".to_string()), walltime: "04:00:00".to_string(), extra: Some("--constraint=v100".to_string()), + serialize_allocations: None, }, ]; @@ -1077,6 +1079,7 @@ fn test_workflow_specification_with_all_resource_types() { tmp: Some("20G".to_string()), walltime: "02:00:00".to_string(), extra: Some("--test-flag".to_string()), + serialize_allocations: None, }]; let mut job = JobSpec::new( @@ -1256,6 +1259,7 @@ fn test_specification_structs_serialization() { tmp: Some("50G".to_string()), walltime: "04:00:00".to_string(), extra: Some("--test-flag".to_string()), + serialize_allocations: None, }; // Test serialization roundtrip @@ -1324,6 +1328,7 @@ fn test_workflow_specification_with_new_structs() { tmp: Some("10G".to_string()), walltime: "02:00:00".to_string(), extra: None, + serialize_allocations: None, }]; let mut job = JobSpec::new("process_data".to_string(), "python process.py".to_string()); @@ -4389,3 +4394,51 @@ fn test_subgraph_workflow_execution_plan_spec_vs_database() { eprintln!("✓ Execution plan from spec matches execution plan from database"); } + +/// `serialize_allocations` must survive a KDL round-trip. KDL v2 spells booleans +/// `#true`/`#false`; emitting a bare `true` produces a file that fails to parse, and +/// nothing else would catch that since the field is optional and silently absent. +#[test] +fn test_slurm_scheduler_serialize_allocations_kdl_roundtrip() { + let yaml = r#" + name: serialize_roundtrip_test + user: test_user + jobs: + - name: job1 + command: "echo hello" + slurm_schedulers: + - name: chain + account: my_account + walltime: "12:00:00" + nodes: 1 + serialize_allocations: true + - name: parallel + account: my_account + walltime: "04:00:00" + nodes: 1 + "#; + let spec: WorkflowSpec = serde_yaml::from_str(yaml).unwrap(); + + let kdl_str = spec.to_kdl_str(); + let roundtripped = + WorkflowSpec::from_spec_file_content(&kdl_str, "kdl").expect("Failed to parse KDL"); + + let schedulers = roundtripped.slurm_schedulers.expect("schedulers missing"); + assert_eq!(schedulers[0].serialize_allocations, Some(true)); + // An omitted value stays omitted rather than round-tripping to Some(false). + assert_eq!(schedulers[1].serialize_allocations, None); +} + +/// The shipped chained-allocations example (referenced from the docs) parses and +/// carries `serialize_allocations` on its scheduler. +#[test] +fn test_chained_allocations_example_parses() { + let path = + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("examples/yaml/chained_allocations.yaml"); + let spec = WorkflowSpec::from_spec_file(&path).expect("Failed to parse example"); + + let schedulers = spec.slurm_schedulers.expect("schedulers missing"); + assert_eq!(schedulers.len(), 1); + assert_eq!(schedulers[0].serialize_allocations, Some(true)); + assert_eq!(spec.jobs.len(), 4); +} diff --git a/torc-server/migrations/20260801000000_add_serialize_allocations.down.sql b/torc-server/migrations/20260801000000_add_serialize_allocations.down.sql new file mode 100644 index 000000000..49cb44bd8 --- /dev/null +++ b/torc-server/migrations/20260801000000_add_serialize_allocations.down.sql @@ -0,0 +1 @@ +ALTER TABLE slurm_scheduler DROP COLUMN serialize_allocations; diff --git a/torc-server/migrations/20260801000000_add_serialize_allocations.up.sql b/torc-server/migrations/20260801000000_add_serialize_allocations.up.sql new file mode 100644 index 000000000..c1967f25b --- /dev/null +++ b/torc-server/migrations/20260801000000_add_serialize_allocations.up.sql @@ -0,0 +1,5 @@ +-- Opt a Slurm scheduler into serialized allocations: every allocation submitted for +-- this scheduler shares one Slurm job name and carries --dependency=singleton, so +-- Slurm runs them strictly one at a time. Used to chain allocations through a long +-- sequential workflow without a long-running process on the login node. +ALTER TABLE slurm_scheduler ADD COLUMN serialize_allocations INTEGER NOT NULL DEFAULT 0;