diff --git a/areno/cli/diagnostics.py b/areno/cli/diagnostics.py index b85b5249..ad789eb4 100644 --- a/areno/cli/diagnostics.py +++ b/areno/cli/diagnostics.py @@ -32,6 +32,20 @@ "HF_HUB_CACHE", ) +# Resource demand estimates for multi-process train/serve runs. These are +# deliberately conservative upper bounds -- a false warning is cheap, a silent +# FD/shm exhaustion mid-run is not. Constants are module-level so they are easy +# to tune and surface in diagnostics output; they are not public API. +_BASE_FDS_PER_WORKER = 64 +_SHM_BASELINE_BYTES = 1 << 30 # 1 GiB per tensor-parallel group; NCCL + CUDA IPC. +_SHMMAX_PATH = "/proc/sys/kernel/shmmax" + +# Severity reused across the host-resource preflight; the diagnostics `check` +# command uses OK/WARN/FAIL, which maps cleanly to normal/warning/blocking. +RESOURCE_OK = "OK" +RESOURCE_WARN = "WARN" +RESOURCE_FAIL = "FAIL" + @click.command(name="env", context_settings={"help_option_names": ["-h", "--help"]}) @click.option("--json", "as_json", is_flag=True, help="Emit a machine-readable JSON support report.") @@ -482,3 +496,232 @@ def _print_env_report(report: dict[str, Any]) -> None: click.echo(" Environment variables:") for name, value in report["env"].items(): click.echo(f" {name}={value if value is not None else ''}") + + +def collect_host_limits() -> dict[str, Any]: + """Probe OS-level per-process resource limits without touching the engine. + + Returns a dict with `file_descriptors`, `processes`, and `shared_memory` + entries. Each entry carries `available` (bool), `soft`/`hard` (for rlimits), + `value` (the effective limit used for comparison), and `error` when a probe + could not run. A probe that is unavailable on the current platform degrades + to a warning rather than a failure -- the preflight must not block runs on + platforms (e.g. macOS/Windows) where a limit simply does not exist. + """ + + return { + "file_descriptors": _fd_limit(), + "processes": _nproc_limit(), + "shared_memory": _shmmax_limit(), + } + + +def _fd_limit() -> dict[str, Any]: + try: + import resource + + soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + except (AttributeError, ImportError, OSError, ValueError) as exc: + return _unavailable(f"{type(exc).__name__}: {exc}") + return _rlimit_entry(soft, hard) + + +def _nproc_limit() -> dict[str, Any]: + try: + import resource + + soft, hard = resource.getrlimit(resource.RLIMIT_NPROC) + except (ImportError, OSError, ValueError) as exc: + return _unavailable(f"{type(exc).__name__}: {exc}") + except AttributeError: + # RLIMIT_NPROC is not defined on every platform; degrade to a warning. + return _unavailable("RLIMIT_NPROC unavailable") + return _rlimit_entry(soft, hard) + + +def _rlimit_entry(soft: Any, hard: Any) -> dict[str, Any]: + """Build a probe entry from a (soft, hard) rlimit pair. + + `RLIM_INFINITY` means the limit is unbounded -- that satisfies any demand, + so the probe is `available` with `unbounded=True` rather than unavailable. + A finite soft limit is the effective value used for comparison. + """ + + import resource + + soft_finite = soft != resource.RLIM_INFINITY + hard_finite = hard != resource.RLIM_INFINITY + if not soft_finite and not hard_finite: + return { + "available": True, + "unbounded": True, + "soft": soft, + "hard": hard, + "value": None, + "error": None, + } + value = soft if soft_finite else hard + return { + "available": True, + "unbounded": False, + "soft": soft, + "hard": hard, + "value": int(value), + "error": None, + } + + +def _unavailable(error: str) -> dict[str, Any]: + return { + "available": False, + "unbounded": False, + "soft": None, + "hard": None, + "value": None, + "error": error, + } + + +def _shmmax_limit() -> dict[str, Any]: + try: + text = Path(_SHMMAX_PATH).read_text(encoding="utf-8").strip() + value = int(text) + except (OSError, ValueError) as exc: + # /proc/sys/kernel/shmmax is Linux-only; absence is expected elsewhere. + return _unavailable(f"{type(exc).__name__}: {exc}") + return {"available": True, "unbounded": False, "soft": value, "hard": value, "value": value, "error": None} + + +def estimate_resource_demand(world_size: int, tp_size: int) -> dict[str, int]: + """Return a documented upper-bound estimate of resource demand for one run. + + - file descriptors: per-worker base plus one socket per cross-rank peer for + the NCCL/tensor-parallel mesh. `RLIMIT_NOFILE` is a per-process limit, so + this is the demand on a single worker process, not the fleet total. + - processes: `world_size` worker ranks plus the driver process. `RLIMIT_NPROC` + is a per-user limit, so the fleet total is the right quantity here. + - shared memory: baseline per tensor-parallel group, scaled by `tp_size`. + + The estimate is intentionally conservative -- it triggers a warning rather + than undershooting a real NCCL/CUDA IPC requirement mid-run. + """ + + if world_size < 1: + raise ValueError("world_size must be >= 1") + if tp_size < 1: + raise ValueError("tp_size must be >= 1") + cross_rank_peers_per_worker = world_size - 1 + return { + "file_descriptors": _BASE_FDS_PER_WORKER + cross_rank_peers_per_worker, + "processes": world_size + 1, + "shared_memory": _SHM_BASELINE_BYTES * tp_size, + } + + +def preflight_host_resources( + world_size: int, + tp_size: int, + *, + policy: str = "warn", + limits: dict[str, Any] | None = None, +) -> list[CheckResult]: + """Compare observed host limits against documented per-run demand. + + Returns one `CheckResult` per resource dimension (file descriptors, + processes, shared memory). `policy` is one of skip/warn/block; it does not + change the returned severities -- the caller decides whether a FAIL under + `block` should abort. Probes that are unavailable degrade to WARN so the + preflight never blocks a run on a platform that simply lacks the limit. + + `limits` is injectable for deterministic tests; production callers omit it + so `collect_host_limits()` probes the real host. + """ + + if policy not in {"skip", "warn", "block"}: + raise ValueError(f"resource-check policy must be skip/warn/block, got {policy!r}") + if policy == "skip": + # `skip` disables the preflight entirely; do not probe the host. + return [] + if limits is None: + limits = collect_host_limits() + demand = estimate_resource_demand(world_size, tp_size) + return [ + _fd_result(limits["file_descriptors"], demand["file_descriptors"]), + _nproc_result(limits["processes"], demand["processes"]), + _shmmax_result(limits["shared_memory"], demand["shared_memory"]), + ] + + +def _fd_result(observed: dict[str, Any], required: int) -> CheckResult: + return _resource_result( + observed, + required, + name="file descriptors (RLIMIT_NOFILE)", + unit="", + adjust="raise the soft limit before launching workers, e.g. `ulimit -n 65536`", + ) + + +def _nproc_result(observed: dict[str, Any], required: int) -> CheckResult: + return _resource_result( + observed, + required, + name="processes (RLIMIT_NPROC)", + unit="", + adjust="raise the soft limit or run from a shell without a low nproc ulimit, e.g. `ulimit -u 32768`", + ) + + +def _shmmax_result(observed: dict[str, Any], required: int) -> CheckResult: + return _resource_result( + observed, + required, + name="shared memory (kernel.shmmax)", + unit=" bytes", + adjust=f"raise the system limit, e.g. `sudo sysctl -w kernel.shmmax={required}`", + ) + + +def _resource_result( + observed: dict[str, Any], + required: int, + *, + name: str, + unit: str, + adjust: str, +) -> CheckResult: + if not observed.get("available"): + error = observed.get("error") or "probe unavailable" + return CheckResult( + RESOURCE_WARN, + name, + f"probe unavailable ({error}); required{unit}={required}", + f"Limit not observable on this platform; {adjust} if the run fails with fd/shmem exhaustion.", + ) + if observed.get("unbounded"): + return CheckResult(RESOURCE_OK, name, f"observed=unbounded required{unit}={required}") + value = int(observed["value"]) + delta = value - required + detail = f"observed{unit}={value} required{unit}={required} delta{unit}={delta}" + if value >= required: + return CheckResult(RESOURCE_OK, name, detail) + return CheckResult(RESOURCE_FAIL, name, detail, adjust) + + +def format_resource_preflight(results: list[CheckResult]) -> str: + """Render resource preflight results as a concise human-readable block.""" + + lines = ["Host resource preflight:"] + for result in results: + lines.append(f" {result.status:<4} {result.name}") + if result.detail: + lines.append(f" {result.detail}") + if result.next_step and result.status in {RESOURCE_WARN, RESOURCE_FAIL}: + lines.append(f" -> {result.next_step}") + return "\n".join(lines) + + +def should_block_on_resources(results: list[CheckResult]) -> bool: + """True if any resource preflight result is a blocking failure.""" + + return any(result.status == RESOURCE_FAIL for result in results) diff --git a/areno/cli/serve.py b/areno/cli/serve.py index 7dc3f548..9af275db 100644 --- a/areno/cli/serve.py +++ b/areno/cli/serve.py @@ -30,6 +30,7 @@ from areno.api.openai_chat import build_chat_completion_response, messages_to_prompt_tokens from areno.api.tokenizer import apply_chat_template_with_options, configure_chat_template_enable_thinking from areno.api.tool_call_parser import ToolCallParser, get_tool_call_parser, infer_tool_call_parser_name +from areno.cli.diagnostics import RESOURCE_FAIL, format_resource_preflight, preflight_host_resources from areno.cli.model_refs import resolve_model_ref from areno.engine.data.tokenizer import load_processor, load_tokenizer @@ -523,6 +524,29 @@ def _resolve_serve_attn_backend( return "native", warning +def _preflight_host_resources(world_size: int, tp_size: int, *, policy: str) -> None: + """Run host resource preflight before engine/worker initialization. + + Mirrors the train CLI: `skip` is a no-op, `warn` (default) emits a stderr + diagnostics block only on a FAIL and never aborts (preserving existing + behavior); `block` raises before workers start. + """ + + if policy == "skip": + return + results = preflight_host_resources(world_size, tp_size, policy=policy) + failures = [r for r in results if r.status == RESOURCE_FAIL] + if failures: + click.echo(format_resource_preflight(results), err=True) + if policy == "block" and failures: + failed = [r.name for r in failures] + raise click.UsageError( + "host resource limits are below the estimated demand for this serve run " + f"(world_size={world_size}, tp_size={tp_size}); failing probes: {failed}. " + "Raise the limits, re-run with --resource-check warn to ignore, or see the next-step hints above." + ) + + async def _run_request_task(app: FastAPI, item: PendingRequest) -> None: """Run one HTTP request as an independent concurrent rollout call.""" @@ -1000,6 +1024,14 @@ def _normalize_stop(stop: str | list[str] | None) -> list[str]: is_flag=True, help="Pass enable_thinking=False to tokenizer chat templates when supported.", ) +@click.option( + "--resource-check", + type=click.Choice(["skip", "warn", "block"], case_sensitive=False), + default="warn", + show_default=True, + help="Preflight host fd/process/shm limits vs run demand before workers start. " + "warn (default) prints and never aborts; block aborts on a failed probe; skip disables.", +) @click.option("--lora-rank", type=int, default=None, help="Enable native LoRA with this rank.") @click.option("--lora-alpha", type=float, default=16.0, show_default=True, help="Native LoRA alpha.") @click.option("--lora-dropout", type=float, default=0.0, show_default=True, help="Native LoRA dropout (must be 0).") @@ -1027,6 +1059,7 @@ def serve_command( eager_decode: bool, attn_backend: Literal["flash", "native"], disable_thinking: bool, + resource_check: str, lora_rank: int | None, lora_alpha: float, lora_dropout: float, @@ -1036,6 +1069,8 @@ def serve_command( """Click entry point: build the app and hand it to uvicorn.""" import uvicorn + if default_backend_type() != MLX: + _preflight_host_resources(world_size, tp_size, policy=resource_check) if base_model_name_or_path is None: base_model_name_or_path = model_path model_path = resolve_model_ref(model_path, model_hub=model_hub) diff --git a/areno/cli/train.py b/areno/cli/train.py index 1de993a3..92616e8d 100644 --- a/areno/cli/train.py +++ b/areno/cli/train.py @@ -35,6 +35,11 @@ RolloutTrainerConfig, TrainerConfig, ) +from areno.cli.diagnostics import ( + RESOURCE_FAIL, + format_resource_preflight, + preflight_host_resources, +) from areno.cli.model_refs import resolve_model_refs_for_config if TYPE_CHECKING: @@ -65,6 +70,8 @@ def flash_attention_unsupported_model_reason(model_config): return resolve_reason(model_config) +RESOURCE_CHECK_CHOICES = ("skip", "warn", "block") + # Group `areno train --help` flags by user intent rather than as one flat wall. # Each entry is (section title, option param names in display order). Every # declared option must appear in exactly one group; the help renderer keeps any @@ -97,6 +104,7 @@ def flash_attention_unsupported_model_reason(model_config): "tp_size", "sequence_parallel", "train_devices", + "resource_check", ), ), ( @@ -333,6 +341,8 @@ def _trainer_config_from_options(**options) -> TrainerConfig: raise click.UsageError("--rollout-devices count must be divisible by --rollout-tp-size") if args.policy_sync_bucket_mb <= 0: raise click.UsageError("--policy-sync-bucket-mb must be positive") + if args.backend != "mlx": + _preflight_host_resources(args.world_size, args.tp_size, policy=_resource_check_policy(args)) if args.batch_size <= 0: raise click.UsageError("--batch-size must be positive") if algorithm.requires_rollout and args.n_samples <= 0: @@ -793,6 +803,37 @@ def _preflight_task_hooks(args, algorithm) -> None: ) +def _resource_check_policy(args) -> str: + """Resolve the host-resource preflight policy from CLI args, default 'warn'.""" + + return str(getattr(args, "resource_check", "warn") or "warn") + + +def _preflight_host_resources(world_size: int, tp_size: int, *, policy: str) -> None: + """Run host resource preflight before backend/model initialization. + + `skip` is a no-op. `warn` (default) emits a stderr diagnostics block only + when a probed limit is below demand (FAIL) and never aborts, so existing + stdout/behavior stays backward compatible; platforms where a probe is + simply unavailable (e.g. macOS shmmax) stay silent. `block` raises a + ``UsageError`` on any FAIL, before any worker is spawned. + """ + + if policy == "skip": + return + results = preflight_host_resources(world_size, tp_size, policy=policy) + failures = [r for r in results if r.status == RESOURCE_FAIL] + if failures: + click.echo(format_resource_preflight(results), err=True) + if policy == "block" and failures: + failed = [r.name for r in failures] + raise click.UsageError( + "host resource limits are below the estimated demand for this run " + f"(world_size={world_size}, tp_size={tp_size}); failing probes: {failed}. " + "Raise the limits, re-run with --resource-check warn to ignore, or see the next-step hints above." + ) + + def _validate_python_callable( path: Path, symbol_name: str, @@ -1815,6 +1856,14 @@ def _dataset_builder_for_suffix(suffix: str) -> str: @click.option("--value-loss-coef", type=float, default=0.5, show_default=True, help="PPO value loss coefficient.") @click.option("--gamma", type=float, default=1.0, show_default=True, help="PPO GAE discount.") @click.option("--lam", type=float, default=0.95, show_default=True, help="PPO GAE lambda.") +@click.option( + "--resource-check", + type=click.Choice(RESOURCE_CHECK_CHOICES, case_sensitive=False), + default="warn", + show_default=True, + help="Preflight host fd/process/shm limits vs run demand before workers start. " + "warn (default) prints and never aborts; block aborts on a failed probe; skip disables.", +) def train_command(**options) -> None: """Click entrypoint for training.""" diff --git a/docs/cli/diagnostics.rst b/docs/cli/diagnostics.rst index 2c52953b..70b8a7a4 100644 --- a/docs/cli/diagnostics.rst +++ b/docs/cli/diagnostics.rst @@ -78,3 +78,78 @@ Checks include: ``WARN`` items usually indicate degraded or incomplete setup. ``FAIL`` items mean AReno is not ready to run the CUDA training/inference engine. + +Host resource preflight +----------------------- + +Before ``areno train`` and ``areno serve`` spawn worker ranks, AReno reads +process-level limits and compares them with a documented demand estimate for +the requested ``world_size``/``tp_size``. The preflight never changes the host; +it only reports and, optionally, blocks. The check applies to CUDA runs; the +single-process MLX backend does not spawn these workers and skips it. + +The ``--resource-check`` option (on both ``train`` and ``serve``) selects +behavior: + +* ``warn`` (default) -- emit a stderr diagnostic only when a probed limit is + below demand, then continue. Existing runs are unaffected. +* ``block`` -- raise a ``UsageError`` and abort before any worker starts if a + probed limit is below demand. +* ``skip`` -- disable the preflight entirely. + +Probed limits: + +* file descriptors via ``RLIMIT_NOFILE`` +* process count via ``RLIMIT_NPROC`` +* shared-memory ceiling via ``/proc/sys/kernel/shmmax`` (Linux) + +Demand estimate (conservative upper bound, see +``areno.cli.diagnostics.estimate_resource_demand``): + +* file descriptors: ``64 + (world_size - 1)`` -- the per-process base plus one + socket per cross-rank peer for the NCCL/tensor-parallel mesh. Since + ``RLIMIT_NOFILE`` is a per-process limit, this is not multiplied by the + number of workers. +* processes: ``world_size + 1`` worker ranks plus the driver. +* shared memory: ``1 GiB * tp_size`` for NCCL/CUDA IPC per tensor-parallel + group. + +Each probe produces one of three severities, mirroring ``areno check``: + +* ``OK`` -- observed limit meets demand (an unbounded ``RLIM_INFINITY`` limit + counts as meeting demand). +* ``WARN`` -- the probe is unavailable on this platform (for example + ``/proc/sys/kernel/shmmax`` is Linux-only, so macOS/Windows report this). A + ``WARN`` never blocks, even under ``--resource-check block``. +* ``FAIL`` -- the observed limit is below demand; the result carries the exact + ``observed``/``required``/``delta`` values and a remediation hint such as + ``ulimit -n 65536`` or a concrete command such as + ``sudo sysctl -w kernel.shmmax=4294967296``. + +Because a probe may be unavailable per platform, a run is only blocked when a +limit was actually observed and is below demand -- the preflight degrades +cleanly on platforms without one of the probes. + +Minimal example (success path, limits sufficient): + +.. code-block:: bash + + areno train --ckpt Qwen/Qwen3-0.6B --dataset-path gsm8k:main \ + --reward-fn-path examples/math/math_verify_reward.py --algo gspo \ + --tp-size 4 --world-size 8 + +Under the default ``warn`` policy this prints nothing to stdout when limits are +sufficient; a failing probe prints to stderr, for example: + +.. code-block:: text + + Host resource preflight: + FAIL file descriptors (RLIMIT_NOFILE) + observed=64 required=71 delta=-7 + -> raise the soft limit before launching workers, e.g. `ulimit -n 65536` + +Boundary/invalid input: with ``--resource-check block`` and a below-demand +``RLIMIT_NOFILE``, the run aborts before worker initialization with a +``UsageError`` naming the failing probe and the exact observed/required values. +Re-run with ``--resource-check warn`` to proceed despite the warning, or raise +the limit first. diff --git a/docs/cli/inference.rst b/docs/cli/inference.rst index fa945482..54c39c67 100644 --- a/docs/cli/inference.rst +++ b/docs/cli/inference.rst @@ -79,6 +79,15 @@ Options: ``world-size`` must be divisible by ``tp-size``. +``--resource-check [skip|warn|block]`` + Preflight host file-descriptor, process-count, and shared-memory limits + against a documented demand estimate for ``world_size``/``tp_size`` before + any worker starts. ``warn`` (default) prints a stderr diagnostic only when a + probed limit is below demand and never aborts; ``block`` aborts the run on a + failed probe; ``skip`` disables the check. See :doc:`diagnostics` for the + probe contract, output fields, and limitations. MLX runs skip this + CUDA-worker check. + Examples -------- diff --git a/docs/cli/training.rst b/docs/cli/training.rst index 5034c984..61c423fa 100644 --- a/docs/cli/training.rst +++ b/docs/cli/training.rst @@ -112,6 +112,15 @@ Built-in algorithms: ``sft``, ``dpo``, ``gspo``, ``grpo``, ``ppo``. ``0`` through ``world-size - 1``. CUDA only; MLX rejects device lists. +``--resource-check [skip|warn|block]`` + Preflight host file-descriptor, process-count, and shared-memory limits + against a documented demand estimate for ``world_size``/``tp_size`` before + any worker starts. ``warn`` (default) prints a stderr diagnostic only when a + probed limit is below demand and never aborts; ``block`` aborts the run on a + failed probe; ``skip`` disables the check. See :doc:`diagnostics` for the + probe contract, output fields, and limitations. MLX runs skip this + CUDA-worker check. + Rollout ~~~~~~~ diff --git a/tests/test_resource_preflight_cpu.py b/tests/test_resource_preflight_cpu.py new file mode 100644 index 00000000..4a97c56e --- /dev/null +++ b/tests/test_resource_preflight_cpu.py @@ -0,0 +1,371 @@ +"""CPU tests for host-resource preflight (fd / process / shm limits). + +These tests inject deterministic limit values so the preflight logic is +exercised without touching the real host or any GPU/engine code. The probes +themselves use only the stdlib `resource` module and `/proc/sys/kernel/shmmax`, +so they import cleanly in any Python 3.10+ environment. +""" + +from __future__ import annotations + +import builtins + +import pytest +from click import UsageError +from click.testing import CliRunner + +from areno.api import CUDA, MLX +from areno.cli import diagnostics +from areno.cli import serve as serve_mod +from areno.cli import train as train_mod +from areno.cli.diagnostics import ( + RESOURCE_FAIL, + RESOURCE_OK, + RESOURCE_WARN, + estimate_resource_demand, + preflight_host_resources, +) + + +def _limits( + *, + fd_value: int | None = 4096, + fd_unbounded: bool = False, + nproc_value: int | None = 100, + shm_value: int | None = 1 << 40, + shm_available: bool = True, +) -> dict: + """Build a deterministic limits dict for injection.""" + + fd_available = fd_value is not None or fd_unbounded + return { + "file_descriptors": { + "available": fd_available, + "unbounded": fd_unbounded, + "soft": fd_value, + "hard": fd_value, + "value": fd_value, + "error": None if fd_available else "probe unavailable", + }, + "processes": { + "available": nproc_value is not None, + "unbounded": False, + "soft": nproc_value, + "hard": nproc_value, + "value": nproc_value, + "error": None if nproc_value is not None else "probe unavailable", + }, + "shared_memory": { + "available": shm_available and shm_value is not None, + "unbounded": False, + "soft": shm_value, + "hard": shm_value, + "value": shm_value, + "error": None if (shm_available and shm_value is not None) else "FileNotFoundError", + }, + } + + +def _names(results): + return [r.status for r in results] + + +# --------------------------------------------------------------------------- +# Demand estimate +# --------------------------------------------------------------------------- + + +def test_estimate_demand_formula_matches_documented_upper_bound(): + demand = estimate_resource_demand(world_size=8, tp_size=4) + # RLIMIT_NOFILE is per process: 64 fds base + 7 cross-rank peers = 71. + assert demand["file_descriptors"] == 64 + 7 + # world_size workers + 1 driver + assert demand["processes"] == 9 + # 1 GiB * tp_size + assert demand["shared_memory"] == (1 << 30) * 4 + + +def test_estimate_demand_rejects_nonpositive_scale(): + with pytest.raises(ValueError): + estimate_resource_demand(0, 1) + with pytest.raises(ValueError): + estimate_resource_demand(1, 0) + + +# --------------------------------------------------------------------------- +# Severities: success / boundary / failure / unbounded +# --------------------------------------------------------------------------- + + +def test_all_ok_when_limits_exceed_demand(): + results = preflight_host_resources(8, 4, policy="warn", limits=_limits()) + assert _names(results) == [RESOURCE_OK, RESOURCE_OK, RESOURCE_OK] + fd = results[0] + # Exact observed/required/delta values are emitted, not just a status. + assert "observed=4096" in fd.detail + demand = estimate_resource_demand(8, 4) + assert f"required={demand['file_descriptors']}" in fd.detail + assert f"delta={4096 - demand['file_descriptors']}" in fd.detail + + +def test_boundary_observed_equals_required_is_ok(): + demand = estimate_resource_demand(2, 1) + limits = _limits( + fd_value=demand["file_descriptors"], nproc_value=demand["processes"], shm_value=demand["shared_memory"] + ) + results = preflight_host_resources(2, 1, policy="warn", limits=limits) + assert _names(results) == [RESOURCE_OK, RESOURCE_OK, RESOURCE_OK] + # fd/processes use unit="" -> "delta=0"; shm uses unit=" bytes" -> "delta bytes=0". + assert "delta=0" in results[0].detail + assert "delta=0" in results[1].detail + assert "delta bytes=0" in results[2].detail + + +def test_low_file_descriptors_is_fail_with_delta_and_remediation(): + demand = estimate_resource_demand(8, 4) + limits = _limits(fd_value=32) # well below the per-process demand of 71 + results = preflight_host_resources(8, 4, policy="warn", limits=limits) + fd = results[0] + assert fd.status == RESOURCE_FAIL + assert f"observed=32 required={demand['file_descriptors']}" in fd.detail + assert f"delta={32 - demand['file_descriptors']}" in fd.detail + assert "ulimit -n" in fd.next_step + + +def test_low_shared_memory_fail_points_at_sysctl(): + limits = _limits(shm_value=1024) + results = preflight_host_resources(8, 4, policy="warn", limits=limits) + shm = results[2] + assert shm.status == RESOURCE_FAIL + assert "bytes" in shm.detail + assert "kernel.shmmax" in shm.next_step + # The remediation hint carries the concrete required value, not a placeholder. + demand = estimate_resource_demand(8, 4) + assert f"kernel.shmmax={demand['shared_memory']}" in shm.next_step + + +def test_unbounded_rlimit_is_ok_not_unavailable(): + limits = _limits(fd_value=None, fd_unbounded=True) + results = preflight_host_resources(8, 4, policy="warn", limits=limits) + fd = results[0] + assert fd.status == RESOURCE_OK + assert "observed=unbounded" in fd.detail + + +# --------------------------------------------------------------------------- +# Graceful degradation when a probe cannot run +# --------------------------------------------------------------------------- + + +def test_unavailable_probe_degrades_to_warn_not_fail(): + # shmmax is Linux-only; on macOS/Windows the probe is unavailable. + limits = _limits(shm_value=None, shm_available=False) + results = preflight_host_resources(8, 4, policy="block", limits=limits) + shm = results[2] + assert shm.status == RESOURCE_WARN + assert "probe unavailable" in shm.detail + # An unavailable probe must never be a blocking failure. + assert shm.status != RESOURCE_FAIL + + +def test_missing_resource_module_degrades_rlimit_probes(monkeypatch): + real_import = builtins.__import__ + + def import_without_resource(name, *args, **kwargs): + if name == "resource": + raise ImportError("resource module unavailable") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", import_without_resource) + + fd = diagnostics._fd_limit() + processes = diagnostics._nproc_limit() + assert fd["available"] is False + assert processes["available"] is False + assert "ImportError" in fd["error"] + assert "ImportError" in processes["error"] + + +def test_should_block_on_resources_only_counts_fail(): + limits = _limits(shm_value=None, shm_available=False) + results = preflight_host_resources(8, 4, policy="block", limits=limits) + # WARN from the unavailable probe does not block. + assert diagnostics.should_block_on_resources(results) is False + fail_limits = _limits(fd_value=1, nproc_value=1, shm_value=1) + fail_results = preflight_host_resources(8, 4, policy="block", limits=fail_limits) + assert diagnostics.should_block_on_resources(fail_results) is True + + +# --------------------------------------------------------------------------- +# Policy validation +# --------------------------------------------------------------------------- + + +def test_invalid_policy_raises(): + with pytest.raises(ValueError): + preflight_host_resources(2, 1, policy="enforce") + + +def test_skip_short_circuits_without_probing(monkeypatch): + # `skip` must not touch the host at all and returns no results. + monkeypatch.setattr(diagnostics, "collect_host_limits", lambda: pytest.fail("host probed under skip")) + assert preflight_host_resources(2, 1, policy="skip") == [] + + +def test_format_resource_preflight_renders_status_and_next_steps(): + limits = _limits(fd_value=1, nproc_value=1000, shm_value=1 << 40) + results = preflight_host_resources(2, 1, policy="warn", limits=limits) + text = diagnostics.format_resource_preflight(results) + assert "Host resource preflight:" in text + assert "FAIL file descriptors" in text + assert "OK processes" in text + assert "ulimit -n" in text # next-step only rendered for WARN/FAIL + + +# =========================================================================== +# CLI integration -- train and serve honor --resource-check +# +# These tests need torch/FastAPI on the path because `areno.cli.train` and +# `areno.cli.serve` import them at module load. They mock the preflight probe +# (so no real host limits are read) and the heavyweight downstream steps +# (`run`, `create_app`, uvicorn) so only the preflight wiring is exercised. +# =========================================================================== + + +def _train_options(**overrides): + """Valid gspo options that pass `_trainer_config_from_options` validation. + + Reuses the canonical defaults from the train-cli config test so the full + validation surface (save_interval, lr, clip eps, ...) is satisfied. + """ + + from tests.test_train_cli_config_cpu import _options + + overrides.setdefault("resource_check", "warn") + overrides.setdefault("world_size", 2) + overrides.setdefault("tp_size", 1) + return _options(**overrides) + + +@pytest.fixture +def patched_preflight(monkeypatch): + """Patch the CLI preflight wrapper in train/serve with a recorder. + + The fake replaces `_preflight_host_resources` wholesale (it owns echo/raise + behavior), so no real host limits are read. Under `block` it raises + UsageError -- matching the real wrapper -- so block wiring is exercised. + """ + + calls: list[tuple] = [] + + def _fake(world_size, tp_size, *, policy): + calls.append((world_size, tp_size, policy)) + if policy == "skip": + return + if policy == "block": + import click + + raise click.UsageError( + "host resource limits are below the estimated demand for this run " + f"(world_size={world_size}, tp_size={tp_size}); failing probes: ['file descriptors']" + ) + + # The CLI calls the module-level `_preflight_host_resources` wrapper. + monkeypatch.setattr(train_mod, "_preflight_host_resources", _fake) + monkeypatch.setattr(serve_mod, "_preflight_host_resources", _fake) + monkeypatch.setattr(serve_mod, "default_backend_type", lambda: CUDA) + return calls + + +def test_train_default_warn_does_not_abort_on_fail(monkeypatch, patched_preflight): + # Real `_trainer_config_from_options` runs the preflight; we only stub `run` + # so the test stops right after config resolution. + monkeypatch.setattr(train_mod, "run", lambda trainer_config: None) + cfg = train_mod._trainer_config_from_options(**_train_options()) + assert patched_preflight == [(2, 1, "warn")] + assert cfg is not None # preflight did not abort config construction + + +def test_train_block_aborts_before_run_on_fail(monkeypatch, patched_preflight): + reached = {"run": False} + monkeypatch.setattr(train_mod, "run", lambda trainer_config: reached.__setitem__("run", True)) + with pytest.raises(UsageError, match="host resource limits are below"): + train_mod._trainer_config_from_options(**_train_options(resource_check="block")) + assert patched_preflight == [(2, 1, "block")] + assert reached["run"] is False # aborted before run() + + +def test_train_skip_does_not_invoke_preflight(monkeypatch): + # Under skip, the wrapper must not call the underlying probe at all. + probe_calls = [] + monkeypatch.setattr(train_mod, "preflight_host_resources", lambda *a, **k: probe_calls.append((a, k)) or []) + monkeypatch.setattr(train_mod, "run", lambda trainer_config: None) + train_mod._trainer_config_from_options(**_train_options(resource_check="skip")) + assert probe_calls == [] + + +def test_train_mlx_does_not_invoke_worker_resource_preflight(monkeypatch): + probe_calls = [] + monkeypatch.setattr(train_mod, "_preflight_host_resources", lambda *a, **k: probe_calls.append((a, k))) + train_mod._trainer_config_from_options( + **_train_options(backend="mlx", world_size=1, tp_size=1, resource_check="block") + ) + assert probe_calls == [] + + +def test_serve_block_aborts_before_engine_init(monkeypatch, patched_preflight): + reached = {"resolve": False, "create_app": False} + monkeypatch.setattr(serve_mod, "resolve_model_ref", lambda *a, **k: reached.__setitem__("resolve", True) or "model") + monkeypatch.setattr(serve_mod, "create_app", lambda **k: reached.__setitem__("create_app", True) or "app") + import types + + monkeypatch.setattr(serve_mod, "uvicorn", types.SimpleNamespace(run=lambda *a, **k: None), raising=False) + import areno.cli.dashboard_registry as dashboard_registry + + monkeypatch.setattr(dashboard_registry, "register_dashboard_job", lambda **k: None) + + runner = CliRunner() + result = runner.invoke( + serve_mod.serve_command, + ["--model-path", "x", "--world-size", "2", "--tp-size", "1", "--resource-check", "block"], + ) + assert patched_preflight == [(2, 1, "block")] + assert reached["resolve"] is False # preflight aborted before model resolution + assert reached["create_app"] is False + assert result.exit_code != 0 + assert "host resource limits are below" in result.output + + +def test_serve_warn_proceeds_to_engine_init(monkeypatch, patched_preflight): + reached = {"resolve": False} + monkeypatch.setattr(serve_mod, "resolve_model_ref", lambda *a, **k: reached.__setitem__("resolve", True) or "model") + monkeypatch.setattr(serve_mod, "create_app", lambda **k: "app") + import types + + monkeypatch.setattr(serve_mod, "uvicorn", types.SimpleNamespace(run=lambda *a, **k: None), raising=False) + import areno.cli.dashboard_registry as dashboard_registry + + monkeypatch.setattr(dashboard_registry, "register_dashboard_job", lambda **k: None) + + runner = CliRunner() + runner.invoke( + serve_mod.serve_command, + ["--model-path", "x", "--world-size", "2", "--tp-size", "1"], + ) + assert patched_preflight == [(2, 1, "warn")] + assert reached["resolve"] is True # warn policy did not block engine init + + +def test_serve_mlx_does_not_invoke_worker_resource_preflight(monkeypatch): + probe_calls = [] + monkeypatch.setattr(serve_mod, "default_backend_type", lambda: MLX) + monkeypatch.setattr(serve_mod, "_preflight_host_resources", lambda *a, **k: probe_calls.append((a, k))) + monkeypatch.setattr(serve_mod, "resolve_model_ref", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("stop"))) + + result = CliRunner().invoke( + serve_mod.serve_command, + ["--model-path", "x", "--world-size", "1", "--tp-size", "1", "--resource-check", "block"], + ) + + assert isinstance(result.exception, RuntimeError) + assert probe_calls == []