Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 14 additions & 5 deletions nemo_run/run/torchx_backend/schedulers/slurm.py
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,11 @@ def _cancel_existing(self, app_id: str) -> None:
self.tunnel.run(f"scancel {app_id}", hide=False)

def describe(self, app_id: str) -> Optional[DescribeAppResponse]:
job_dirs = _get_job_dirs()
try:
job_dirs = _get_job_dirs()
except OSError as e:
log.warning(f"Could not read job dirs file, returning UNKNOWN status for {app_id}: {e}")
return DescribeAppResponse(app_id=app_id, state=AppState.UNKNOWN)
if app_id in job_dirs:
_, tunnel_cfg, _ = job_dirs[app_id]
self._initialize_tunnel(tunnel_cfg)
Expand Down Expand Up @@ -426,7 +430,7 @@ def _save_job_dir(

def _get_job_dirs(retries: int = 5) -> dict[str, tuple[str, SSHTunnel | LocalTunnel, str]]:
last_exc: OSError | None = None
for _ in range(retries):
for attempt in range(retries):
try:
with open(SLURM_JOB_DIRS, "rt") as f:
lines = f.readlines()
Expand All @@ -435,12 +439,17 @@ def _get_job_dirs(retries: int = 5) -> dict[str, tuple[str, SSHTunnel | LocalTun
return {}
except OSError as e:
last_exc = e
time.sleep(1)
delay = min(2**attempt, 30)
log.warning(
f"OSError reading {SLURM_JOB_DIRS} (attempt {attempt + 1}/{retries}): {e}. "
f"Retrying in {delay}s..."
)
time.sleep(delay)
else:
if last_exc is not None:
raise last_exc
raise OSError(
f"Failed to read SLURM job dirs from {SLURM_JOB_DIRS} after {retries} retries"
raise RuntimeError(
f"Failed to read {SLURM_JOB_DIRS} after {retries} retries, but no OSError was captured."
)

out = {}
Expand Down
20 changes: 18 additions & 2 deletions test/run/torchx_backend/schedulers/test_slurm.py
Original file line number Diff line number Diff line change
Expand Up @@ -340,8 +340,8 @@ def test_get_job_dirs():


def test_get_job_dirs_retries_on_permission_error(tmp_path, mocker):
"""Transient PermissionError should be retried; success on 3rd attempt returns data."""
mocker.patch("time.sleep") # don't actually sleep in tests
"""Transient PermissionError should be retried with exponential backoff; success on 3rd attempt returns data."""
mock_sleep = mocker.patch("time.sleep")
mock_open = mocker.mock_open(read_data="")
mock_open.side_effect = [
PermissionError("[Errno 1] Operation not permitted"),
Expand All @@ -353,6 +353,8 @@ def test_get_job_dirs_retries_on_permission_error(tmp_path, mocker):
result = _get_job_dirs(retries=5)
assert result == {}
assert mock_open.call_count == 3
# Exponential backoff: attempt 0 -> sleep(1), attempt 1 -> sleep(2)
assert mock_sleep.call_args_list == [mock.call(1), mock.call(2)]


def test_get_job_dirs_raises_after_exhausting_retries(mocker):
Expand All @@ -364,6 +366,20 @@ def test_get_job_dirs_raises_after_exhausting_retries(mocker):
_get_job_dirs(retries=3)


def test_describe_returns_unknown_on_persistent_permission_error(slurm_scheduler, mocker):
"""describe() should return UNKNOWN state when _get_job_dirs() raises OSError, not propagate."""
from torchx.specs import AppState

mocker.patch(
"nemo_run.run.torchx_backend.schedulers.slurm._get_job_dirs",
side_effect=PermissionError("[Errno 1] Operation not permitted"),
)

result = slurm_scheduler.describe("12345")
assert result is not None
assert result.state == AppState.UNKNOWN


def test_schedule_with_dependencies(slurm_scheduler, slurm_executor):
mock_request = mock.MagicMock()
mock_request.cmd = ["sbatch", "--requeue", "--parsable"]
Expand Down
Loading