blob: a2c8225264f3027f19aebd23fbb6d5c7837e8ad5 [file]
############################################################################
# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
############################################################################
"""Basic heterogeneous job sbatch test.
Submit a 3-component hetjob via sbatch and verify that the batch script
sees the expected per-component environment (SLURM_HET_SIZE, the
SLURM_JOB_ID_HET_GROUP_<N> / SLURM_NTASKS_HET_GROUP_<N> /
SLURM_CPUS_PER_TASK_HET_GROUP_<N> / SLURM_MEM_PER_CPU_HET_GROUP_<N> vars,
and a per-component partition var for each group).
"""
import re
import pytest
import atf
# (cpus-per-task, mem-per-cpu, ntasks) per het group. The values differ
# between components so that broadcasting one component's value across
# every _HET_GROUP_<N> suffix would be caught.
COMPONENTS = [(4, 10, 1), (2, 2, 2), (1, 6, 3)]
# Echoed after the env dump. The env output order is not deterministic, so
# this marker is what tells us the whole dump has been flushed.
OUTPUT_MARKER = "HETJOB_ENV_DUMPED"
@pytest.fixture(scope="module", autouse=True)
def setup():
atf.require_nodes(3, [("CPUs", 4), ("RealMemory", 1024)])
# Het jobs are only started by backfill; don't wait for its full cycle
atf.require_config_parameter_includes("SchedulerParameters", ("bf_interval", 1))
atf.require_config_parameter("SchedulerType", "sched/backfill")
atf.require_slurm_running()
def test_hetjob_sbatch_env():
"""Verify hetjob env vars in the batch script."""
file_in = "hetjob.in"
file_out = "hetjob.out"
directives = "\n#SBATCH hetjob\n".join(
f"#SBATCH --cpus-per-task={cpus} --mem-per-cpu={mem} --ntasks={ntasks} -t1"
for cpus, mem, ntasks in COMPONENTS
)
atf.make_bash_script(file_in, f"{directives}\n\nenv\necho {OUTPUT_MARKER}\n")
leader_job_id = atf.submit_job_sbatch(f"-o {file_out} {file_in}", fatal=True)
jobs = atf.get_jobs(leader_job_id, fatal=True)
component_ids = sorted(
atf.range_to_list(jobs[leader_job_id]["HetJobIdSet"]),
key=lambda job_id: jobs[job_id]["HetJobOffset"],
)
atf.wait_for_job_state(leader_job_id, "DONE", fatal=True)
atf.assert_file_contents(file_out, OUTPUT_MARKER, contains=True)
output = atf.run_command_output(f"cat {file_out}", fatal=True)
assert (
re.search(r"^SLURM_HET_SIZE=3$", output, re.MULTILINE) is not None
), "Missing SLURM_HET_SIZE=3 in batch environment"
for offset, (cpus, mem, ntasks) in enumerate(COMPONENTS):
expected = {
"SLURM_JOB_ID": component_ids[offset],
"SLURM_CPUS_PER_TASK": cpus,
"SLURM_MEM_PER_CPU": mem,
"SLURM_NTASKS": ntasks,
}
for base, value in expected.items():
var = f"{base}_HET_GROUP_{offset}"
assert (
re.search(rf"^{var}={value}$", output, re.MULTILINE) is not None
), f"Missing {var}={value} in batch environment"
partition_offsets = sorted(
int(offset)
for offset in re.findall(
r"^SLURM_JOB_PARTITION_HET_GROUP_(\d+)=", output, re.MULTILINE
)
)
assert partition_offsets == [0, 1, 2], (
"Expected SLURM_JOB_PARTITION_HET_GROUP_ vars for offsets [0, 1, 2],"
f" got {partition_offsets}"
)