Skip to content
Open
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
76 changes: 76 additions & 0 deletions tests/plugin/test_platform_abstraction.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@
"""Unit tests for the platform abstraction layer."""

import os
import sys
from contextlib import contextmanager
from types import SimpleNamespace
from unittest import mock

import pytest
Expand Down Expand Up @@ -145,6 +147,80 @@ def test_invalid_platform_raises(self):
_create_platform("invalid_platform")


class TestCpuOnlyRayWorkerLogging:
"""Test warning levels for accelerator-less Ray workers."""

@staticmethod
def _mock_ray(assigned_resources):
context = mock.Mock()
context.get_task_id.return_value = "task-id"
context.get_assigned_resources.return_value = assigned_resources
return SimpleNamespace(
is_initialized=mock.Mock(return_value=True),
get_runtime_context=mock.Mock(return_value=context),
)

@staticmethod
def _exercise_fallback():
with (
mock.patch.object(PlatformRegistry, "registered_names", return_value=("nvidia",)),
mock.patch("verl.plugin.platform.platform_manager.PlatformCUDA.is_platform_available", return_value=False),
mock.patch("verl.plugin.platform.platform_manager.PlatformCUDA.is_available", return_value=False),
):
assert _detect_platform_name() == "nvidia"
assert _create_platform("nvidia") is not None

def test_cpu_only_ray_worker_does_not_warn(self):
with (
mock.patch.dict(sys.modules, {"ray": self._mock_ray({"CPU": 1.0})}),
mock.patch("verl.plugin.platform.platform_manager.logger") as mock_logger,
):
self._exercise_fallback()

mock_logger.warning.assert_not_called()
mock_logger.debug.assert_any_call(
"No supported accelerator detected. Registered: %s. Falling back to 'nvidia'.",
("nvidia",),
)
mock_logger.debug.assert_any_call(
"Platform '%s' (%s) is registered but not available. "
"This may be due to this ray actor being a CPU-only actor.",
"nvidia",
"PlatformCUDA",
)

def test_gpu_assigned_ray_worker_still_warns(self):
with (
mock.patch.dict(sys.modules, {"ray": self._mock_ray({"CPU": 1.0, "GPU": 1.0})}),
mock.patch("verl.plugin.platform.platform_manager.logger") as mock_logger,
):
self._exercise_fallback()

assert mock_logger.warning.call_count == 2

def test_ray_unavailable_still_warns(self):
with (
mock.patch.dict(sys.modules, {"ray": None}),
mock.patch("verl.plugin.platform.platform_manager.logger") as mock_logger,
):
self._exercise_fallback()

assert mock_logger.warning.call_count == 2

def test_ray_runtime_context_failure_still_warns(self):
ray = SimpleNamespace(
is_initialized=mock.Mock(return_value=True),
get_runtime_context=mock.Mock(side_effect=RuntimeError("context unavailable")),
)
with (
mock.patch.dict(sys.modules, {"ray": ray}),
mock.patch("verl.plugin.platform.platform_manager.logger") as mock_logger,
):
self._exercise_fallback()

assert mock_logger.warning.call_count == 2


class TestPlatformSingleton:
"""Test singleton and external injection."""

Expand Down
32 changes: 30 additions & 2 deletions verl/plugin/platform/platform_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ class PlatformMetaX(PlatformBase):
logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "INFO"))

_current_platform: PlatformBase | None = None
_RAY_ACCELERATOR_RESOURCE_NAMES = frozenset({"GPU", "NPU"})


class PlatformRegistry:
Expand Down Expand Up @@ -110,13 +111,39 @@ def _detect_platform_name() -> str:
logger.debug("Platform '%s' detection failed: %s", name, e)
continue

logger.warning(
log = logger.debug if _is_cpu_only_ray_worker() else logger.warning
log(
"No supported accelerator detected. Registered: %s. Falling back to 'nvidia'.",
names,
)
return "nvidia"


def _is_cpu_only_ray_worker() -> bool:
"""Return whether this is a Ray worker with no accelerator resources.

Fail closed so missing Ray installations, drivers, and runtime-context
failures retain the existing warning behavior.
"""
try:
import ray

if not ray.is_initialized():
return False

context = ray.get_runtime_context()
if context.get_task_id() is None:
return False

assigned_resources = context.get_assigned_resources()
return not any(
resource_name in _RAY_ACCELERATOR_RESOURCE_NAMES and amount > 0
for resource_name, amount in assigned_resources.items()
)
except Exception:
return False


def _create_platform(name: str) -> PlatformBase:
"""Instantiate the concrete platform for *name*."""
platform_cls = PlatformRegistry.get(name)
Expand All @@ -128,7 +155,8 @@ def _create_platform(name: str) -> PlatformBase:
)
platform = platform_cls()
if not platform.is_available():
logger.warning(
log = logger.debug if _is_cpu_only_ray_worker() else logger.warning
log(
"Platform '%s' (%s) is registered but not available. "
"This may be due to this ray actor being a CPU-only actor.",
name,
Expand Down