Skip to content

[Fix][Core] Raise exception if ray.init() is called while is_client_mode_enabled is True #67407

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
justinyeh1995 wants to merge 4 commits into
base: master
Choose a base branch
from
Open

[Fix][Core] Raise exception if ray.init() is called while is_client_mode_enabled is True #67407

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3d4e773
raise exception if ray.init() is called while is_client_mode_enabled …
justinyeh1995 Oct 25, 2026
77ae2c7
fix ray_start_client_server cases
justinyeh1995 Oct 26, 2026
ab50fdf
only raise after a failed ray:// connect
justinyeh1995 Oct 28, 2026
2a1fd51
update impl according to the updated issue
justinyeh1995 Oct 28, 2026
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
6 changes: 5 additions & 1 deletion python/ray/_private/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.")
Comment on lines +1761 to +1762

@gemini-code-assist gemini-code-assist Bot Oct 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Issue: Misleading Error and Blocked Fallback due to Global State Leak\n\nThe check introduced here:\npython\nif client_mode_should_convert() and num_connected_contexts() == 0:\n raise RuntimeError(\"Ray Client is already initialized in this process.\")\n\nraises a misleading RuntimeError and blocks users from starting a local Ray cluster (via ray.init()) if a previous Ray Client connection attempt failed.\n\n#### Why this happens:\n1. When ray.init(\"ray://...\") is called, it enables client mode globally by calling _explicitly_enable_client_mode().\n2. If the connection attempt fails (e.g., throwing a ConnectionError), the exception is propagated, but _explicitly_disable_client_mode() is never called because RayAPIStub.connect() does not clean up on failure.\n3. This leaves is_client_mode_enabled as True globally, even though no client connection was ever established (num_connected_contexts() == 0).\n4. Consequently, any subsequent call to ray.init() (local) triggers this new check and raises RuntimeError: Ray Client is already initialized in this process., which is incorrect (it was never successfully initialized) and prevents the user from falling back to a local cluster.\n\n#### Recommended Solution:\nInstead of adding this check to ray.init(), we should fix the root cause by ensuring that is_client_mode_enabled is cleaned up/disabled if the connection attempt fails.\n\nWe can achieve this by wrapping the connection logic in RayAPIStub.connect() (located in python/ray/util/client/__init__.py) in a try...except block:\n\npython\n def connect(self, *args, **kw_args):\n self.get_context()._inside_client_test = self._inside_client_test\n try:\n conn = self.get_context().connect(*args, **kw_args)\n global _lock, _all_contexts\n with _lock:\n _all_contexts.add(self._cxt.handler)\n return conn\n except Exception:\n global _lock, _all_contexts\n with _lock:\n if len(_all_contexts) == 0:\n _explicitly_disable_client_mode()\n raise\n\n\nIf we fix the leak there, client_mode_should_convert() will correctly return False after a failed connection, and this check in worker.py becomes unnecessary and can be removed.

@justinyeh1995 justinyeh1995 Oct 30, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as #67407 (comment). Fixing the leak of global client mode state is not in-scope for this pr. It is also an existing issue. The goal is to raise with a clearer message in this pr.


# Try to increase the file descriptor limit, which is too low by
# default for Ray: https://github.com/ray-project/ray/issues/11239
try:
Expand Down
22 changes: 22 additions & 0 deletions python/ray/tests/test_client_multi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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
Expand Down
29 changes: 28 additions & 1 deletion python/ray/tests/test_ray_init.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Comment on lines +381 to +386

@gemini-code-assist gemini-code-assist Bot Oct 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

This test asserts that calling ray.init() after a failed client connection raises a RuntimeError stating that Ray Client is already initialized. However, if a client connection fails, Ray Client is not successfully initialized, and the global is_client_mode_enabled flag should be reset to False.\n\nBy expecting a RuntimeError here, we are codifying a buggy behavior where a failed connection attempt permanently locks the process into client mode, preventing any subsequent fallback to a local Ray cluster or retries.\n\nInstead of asserting this error, we should fix the root cause in the client builder (where connect() is called) to ensure that is_client_mode_enabled is reset to False if the connection attempt fails. Once that is fixed, a subsequent ray.init() should be able to succeed.

@justinyeh1995 justinyeh1995 Oct 25, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Marking is_client_mode_enabled to False would be out of scope for #67142.

Comment on lines +381 to +386

@gemini-code-assist gemini-code-assist Bot Oct 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

This test asserts that calling ray.init() (local) after a failed client connection raises a RuntimeError. However, if a client connection failed, the user should be allowed to fall back to starting a local Ray cluster. Asserting that it raises RuntimeError enforces a broken/undesirable behavior caused by the global state leak of is_client_mode_enabled.\n\nOnce the leak is fixed in RayAPIStub.connect(), this test should instead assert that ray.init() successfully starts a local cluster.



def test_shutdown_wait_for_processes(shutdown_only):
ray.init()
node = ray._private.worker._global_node
Expand Down
Loading