diff --git a/rlix/scheduler/scheduler.py b/rlix/scheduler/scheduler.py index 99602c4..00c2eb9 100644 --- a/rlix/scheduler/scheduler.py +++ b/rlix/scheduler/scheduler.py @@ -585,6 +585,32 @@ async def clear_progress(self, *, pipeline_id: str) -> None: def _clear_rollout_intent_locked(self, pipeline_id: str) -> None: self._state.rollout_open_pipelines.pop(pipeline_id, None) + def _warn_gen_request_without_demand_locked( + self, *, cluster_id: str, pipeline_id: str, step_target_estimate: Optional[int] + ) -> None: + """Warn when a GENERATION request carries no usable demand signal. + + The gap-ratio planner skips pipelines whose registry entry has no + positive ``step_target_estimate`` AND no stored progress (see the + bootstrap-estimate branch in ``planner.plan_generation_gap_ratio``). + Such a request is never signaled and the caller blocks indefinitely. + Warn-only: progress may still arrive later on some backends, so + rejecting outright could break legitimate request-then-report flows. + """ + if step_target_estimate is not None and int(step_target_estimate) > 0: + return + _, step_target = self._pipeline_progress_totals_locked(pipeline_id=pipeline_id) + if step_target > 0.0: + return + logger.warning( + "request_gpus(GENERATION) for cluster_id=%r has no step_target_estimate " + "(got %r) and no stored progress; the gap-ratio planner will skip this " + "pipeline and the request may block indefinitely until progress or a " + "positive estimate arrives", + cluster_id, + step_target_estimate, + ) + async def request_gpus( self, *, @@ -636,6 +662,9 @@ async def request_gpus( self._state.pending_bucket(priority).append(pending) if priority == Priority.GENERATION: self._state.rollout_open_pipelines[pipeline_id] = step_target_estimate + self._warn_gen_request_without_demand_locked( + cluster_id=cluster_id, pipeline_id=pipeline_id, step_target_estimate=step_target_estimate + ) # Queue Tracing: Track enqueue AFTER append (depth is correct) self._tracer.trace_queue_enqueue( cluster_id, priority, lora_name, bucket_depth=len(self._state.pending_bucket(priority)) @@ -737,6 +766,11 @@ async def notify_release_then_request_gpus( if request_priority == Priority.GENERATION: pipeline_id_gen, _ = parse_cluster_id(request_cluster_id) self._state.rollout_open_pipelines[pipeline_id_gen] = request_step_target_estimate + self._warn_gen_request_without_demand_locked( + cluster_id=request_cluster_id, + pipeline_id=pipeline_id_gen, + step_target_estimate=request_step_target_estimate, + ) # Queue Tracing: Track enqueue AFTER append (depth is correct) self._tracer.trace_queue_enqueue( request_cluster_id, diff --git a/tests/test_scheduler_gen_request_warning.py b/tests/test_scheduler_gen_request_warning.py new file mode 100644 index 0000000..d05b3a5 --- /dev/null +++ b/tests/test_scheduler_gen_request_warning.py @@ -0,0 +1,113 @@ +"""PR ingress-validation test for the scheduler. + +A GENERATION request with neither a positive ``step_target_estimate`` nor any +stored progress is skipped by the gap-ratio planner (bootstrap-estimate +branch) and blocks its caller indefinitely. The scheduler now warns at +enqueue time; these tests pin the warn / no-warn matrix. +""" + +from __future__ import annotations + +import importlib +import logging +import sys +import types +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +RLIX_ROOT = REPO_ROOT / "rlix" + + +def _install_import_stubs(monkeypatch: pytest.MonkeyPatch) -> None: + for module_name in list(sys.modules): + if module_name == "ray" or module_name.startswith("rlix"): + monkeypatch.delitem(sys.modules, module_name, raising=False) + + ray_stub = types.ModuleType("ray") + + def _remote(*args, **kwargs): + def _decorate(obj): + return obj + + return _decorate + + ray_stub.remote = _remote + ray_stub.get_actor = lambda *args, **kwargs: None + ray_stub.get = lambda value: value + monkeypatch.setitem(sys.modules, "ray", ray_stub) + + package_roots = { + "rlix": RLIX_ROOT, + "rlix.protocol": RLIX_ROOT / "protocol", + "rlix.scheduler": RLIX_ROOT / "scheduler", + "rlix.utils": RLIX_ROOT / "utils", + } + for module_name, module_path in package_roots.items(): + package_module = types.ModuleType(module_name) + package_module.__path__ = [str(module_path)] # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, module_name, package_module) + + +def _make_scheduler(monkeypatch: pytest.MonkeyPatch): + _install_import_stubs(monkeypatch) + scheduler_module = importlib.import_module("rlix.scheduler.scheduler") + protocol_types = importlib.import_module("rlix.protocol.types") + return scheduler_module, protocol_types, scheduler_module.SchedulerImpl() + + +_PIPELINE_ID = "miles_test" +_CLUSTER_ID = f"{_PIPELINE_ID}_actor_infer" + + +def test_warns_when_no_estimate_and_no_progress( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + _, _, scheduler = _make_scheduler(monkeypatch) + with caplog.at_level(logging.WARNING): + scheduler._warn_gen_request_without_demand_locked( + cluster_id=_CLUSTER_ID, pipeline_id=_PIPELINE_ID, step_target_estimate=None + ) + assert "may block indefinitely" in caplog.text + + +def test_warns_on_non_positive_estimate( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + _, _, scheduler = _make_scheduler(monkeypatch) + with caplog.at_level(logging.WARNING): + scheduler._warn_gen_request_without_demand_locked( + cluster_id=_CLUSTER_ID, pipeline_id=_PIPELINE_ID, step_target_estimate=0 + ) + assert "may block indefinitely" in caplog.text + + +def test_silent_with_positive_estimate( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + _, _, scheduler = _make_scheduler(monkeypatch) + with caplog.at_level(logging.WARNING): + scheduler._warn_gen_request_without_demand_locked( + cluster_id=_CLUSTER_ID, pipeline_id=_PIPELINE_ID, step_target_estimate=8 + ) + assert caplog.text == "" + + +def test_silent_when_progress_already_stored( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + scheduler_module, protocol_types, scheduler = _make_scheduler(monkeypatch) + report = protocol_types.ProgressReport( + pipeline_id=_PIPELINE_ID, + step_target_trajectories=8, + metrics={"completed": 2}, + ) + scheduler._state.latest_progress_by_pipeline[_PIPELINE_ID] = { + "train": {"__full_finetune__": report} + } + with caplog.at_level(logging.WARNING): + scheduler._warn_gen_request_without_demand_locked( + cluster_id=_CLUSTER_ID, pipeline_id=_PIPELINE_ID, step_target_estimate=None + ) + assert caplog.text == ""