Repository: ray-project/ray
File: python/ray/dashboard/optional_utils.py (Python)

Code before:

def init_ray_and_catch_exceptions() -> Callable:
    """Decorator to be used on methods that require being connected to Ray."""

    def decorator_factory(f: Callable) -> Callable:
        @functools.wraps(f)
        async def decorator(
            self: Union[DashboardAgentModule, DashboardHeadModule], *args, **kwargs
        ):
            try:
                if not ray.is_initialized():
                    try:
                        address = self.gcs_address
                        logger.info(f"Connecting to ray with address={address}")
                        # Set the gcs rpc timeout to shorter
                        os.environ["RAY_gcs_server_request_timeout_seconds"] = str(
                            dashboard_consts.GCS_RPC_TIMEOUT_SECONDS
                        )
                        # Init ray without logging to driver
                        # to avoid infinite logging issue.

                        # The dashboard agent hosts a gRPC server on the
                        # same event loop, and ray.init() needs to health
                        # check that server. Running it on the event loop
                        # thread would self deadlock.
                        await asyncio.to_thread(
                            ray.init,
                            address=address,
                            log_to_driver=False,
                            configure_logging=False,
                            namespace=RAY_INTERNAL_DASHBOARD_NAMESPACE,
                            _skip_env_hook=True,
                        )
                    except Exception as e:
                        ray.shutdown()
                        raise e from None
                return await f(self, *args, **kwargs)
            except Exception as e:
                logger.exception(f"Unexpected error in handler: {e}")
                return Response(
                    text=traceback.format_exc(),
                    status=aiohttp.web.HTTPInternalServerError.status_code,
                )

        return decorator

    return decorator_factory


Review comment (bug):
Found that the dashboard agent hosts both its grpc server   and the JobAgent in the same event loop. When ray job submit arrives, JobAgent calls `ray.init()`(so that to create JobSupervisor Actor) which blocks the event loop waiting for the HealthCheck to succeed, but the gRPC server can't respond because the event loop is blocked. This is a self deadlock.

Fixed by running ray.init() in a thread pool and releasing the GIL during `CoreWorkerProcess::Initialize()` so the event loop thread can process the HealthCheck.

Code after:

def init_ray_and_catch_exceptions() -> Callable:
    """Decorator to be used on methods that require being connected to Ray."""

    def decorator_factory(f: Callable) -> Callable:
        @functools.wraps(f)
        async def decorator(
            self: Union[DashboardAgentModule, DashboardHeadModule], *args, **kwargs
        ):
            try:
                if not ray.is_initialized():
                    try:
                        address = self.gcs_address
                        logger.info(f"Connecting to ray with address={address}")
                        # Set the gcs rpc timeout to shorter
                        os.environ["RAY_gcs_server_request_timeout_seconds"] = str(
                            dashboard_consts.GCS_RPC_TIMEOUT_SECONDS
                        )
                        # Init ray without logging to driver
                        # to avoid infinite logging issue.
                        ray.init(
                            address=address,
                            log_to_driver=False,
                            configure_logging=False,
                            namespace=RAY_INTERNAL_DASHBOARD_NAMESPACE,
                            _skip_env_hook=True,
                        )
                    except Exception as e:
                        ray.shutdown()
                        raise e from None
                return await f(self, *args, **kwargs)
            except Exception as e:
                logger.exception(f"Unexpected error in handler: {e}")
                return Response(
                    text=traceback.format_exc(),
                    status=aiohttp.web.HTTPInternalServerError.status_code,
                )

        return decorator

    return decorator_factory