diff --git a/python/ray/_private/worker.py b/python/ray/_private/worker.py index d339a9ad29cf..940ff043597c 100644 --- a/python/ray/_private/worker.py +++ b/python/ray/_private/worker.py @@ -66,7 +66,7 @@ ensure_token_if_auth_enabled, maybe_enable_token_auth_if_token_available, ) -from ray._private.client_mode_hook import client_mode_hook +from ray._private.client_mode_hook import client_mode_hook, client_mode_should_convert from ray._private.function_manager import FunctionActorManager from ray._private.inspect_util import is_cython from ray._private.ray_logging import ( @@ -111,6 +111,7 @@ from ray.experimental.tqdm_ray import RAY_TQDM_MAGIC from ray.runtime_env.runtime_env import _merge_runtime_env from ray.util.annotations import Deprecated, PublicAPI +from ray.util.client import num_connected_contexts from ray.util.debug import log_once from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy from ray.util.tracing.tracing_helper import _import_from_string @@ -1757,6 +1758,9 @@ def sigterm_handler(signum, frame): unknown = ", ".join(kwargs) raise RuntimeError(f"Unknown keyword argument(s): {unknown}") + if client_mode_should_convert() and num_connected_contexts() == 0: + raise RuntimeError("Ray Client is already initialized in this process.") + # Try to increase the file descriptor limit, which is too low by # default for Ray: https://github.com/ray-project/ray/issues/11239 try: diff --git a/python/ray/tests/test_client_multi.py b/python/ray/tests/test_client_multi.py index 69e8b7254691..1b07c66271c0 100644 --- a/python/ray/tests/test_client_multi.py +++ b/python/ray/tests/test_client_multi.py @@ -3,6 +3,7 @@ import pytest import ray +from ray._common.network_utils import find_free_port def test_multi_raise_exception_on_non_client(shutdown_only): @@ -201,6 +202,27 @@ def get(idx): assert ret == [0, 1] +@pytest.mark.skipif( + sys.platform == "win32", reason="PSUtil does not work the same on windows." +) +@pytest.mark.parametrize( + "call_ray_start", + ["ray start --head --ray-client-server-port 25308 --port 0"], + indirect=True, +) +def test_multi_client_connect_after_client_connect_failed(call_ray_start): + @ray.remote + def hello(): + return "world" + + with pytest.raises(ConnectionError): + ray.init(f"ray://localhost:{find_free_port()}") + + cli = ray.init("ray://localhost:25308", allow_multiple=True) + with cli: + assert ray.get(hello.remote()) == "world" + + if __name__ == "__main__": # The following should be removed after # https://github.com/ray-project/ray/issues/20355 diff --git a/python/ray/tests/test_ray_init.py b/python/ray/tests/test_ray_init.py index 0024408c7253..c92d79027002 100644 --- a/python/ray/tests/test_ray_init.py +++ b/python/ray/tests/test_ray_init.py @@ -12,8 +12,9 @@ import pytest import ray -from ray._common.network_utils import build_address, parse_address +from ray._common.network_utils import build_address, find_free_port, parse_address from ray._private import ray_constants +from ray._private.client_mode_hook import disable_client_hook from ray._private.test_utils import persistent_gcs_test_enabled from ray.client_builder import ClientContext from ray.cluster_utils import Cluster @@ -359,6 +360,32 @@ def test_ray_init_with_runtime_env_as_object( assert "gcs://" in parsed_runtime_env["py_modules"][0] +@pytest.mark.parametrize( + "call_ray_start", + ["ray start --head --ray-client-server-port 25307 --port 0"], + indirect=True, +) +def test_ray_init_after_client_connect_succeeded(call_ray_start): + ray.init("ray://localhost:25307") + + try: + # A local ray.init() while the Ray Client connection is open must not raise. + ray.init() + finally: + # While client mode is on, ray.shutdown() is routed to the Ray Client. + # Bypass the client hook so ray.shutdown() disconnects the local driver instead. + with disable_client_hook(): + ray.shutdown() + + +def test_ray_init_after_client_connect_failed(shutdown_only): + with pytest.raises(ConnectionError): + ray.init(f"ray://localhost:{find_free_port()}") + + with pytest.raises(RuntimeError, match="Ray Client is already initialized"): + ray.init() + + def test_shutdown_wait_for_processes(shutdown_only): ray.init() node = ray._private.worker._global_node