Repository: ray-project/ray
File: python/ray/serve/_private/replica.py (Python)

Code before:
                # properly in user code.
                asyncio.set_event_loop(self._user_code_event_loop)
                self._configure_user_code_threadpool()
                # Start monitoring before run_forever so the task is scheduled.
                self._user_code_loop_monitor.start(self._user_code_event_loop)
                self._user_code_event_loop.run_forever()

            self._user_code_event_loop_thread = threading.Thread(
                daemon=True,
                target=_run_user_code_event_loop,
            )
            self._user_code_event_loop_thread.start()
        else:
            self._user_code_event_loop = asyncio.get_running_loop()
            self._user_code_loop_monitor = None
            self._configure_user_code_threadpool()

    @property
    def event_loop(self) -> asyncio.AbstractEventLoop:
        return self._user_code_event_loop

    def _get_user_code_threadpool_max_workers(self) -> Optional[int]:
        num_cpus = self._ray_actor_options.get("num_cpus")
        if num_cpus is None:
            return None
        return max(1, int(math.ceil(num_cpus)))

    def _configure_user_code_threadpool(self) -> None:
        max_workers = self._get_user_code_threadpool_max_workers()
        if max_workers is None:
            return
        self._user_code_threadpool = concurrent.futures.ThreadPoolExecutor(
            max_workers=max_workers
        )
        self._user_code_event_loop.set_default_executor(self._user_code_threadpool)

    def _run_user_code(f: Callable) -> Callable:
        """Decorator to run a coroutine method on the user code event loop.

        The method will be modified to be a sync function that returns a
        `asyncio.Future` if user code is running in a separate event loop.
        Otherwise, it will return the coroutine directly.
        """
        assert inspect.iscoroutinefunction(
            f
        ), "_run_user_code can only be used on coroutine functions."

        @functools.wraps(f)
        def wrapper(self, *args, **kwargs) -> Any:
            coro = f(self, *args, **kwargs)
            if self._run_user_code_in_separate_thread:

Review comment (suggestion):
i would try and maintain somewhat of the stock behavior `min(32, int(math.ceil(num_cpus)) + 4)`

<img width="816" height="163" alt="Image" src="https://github.com/user-attachments/assets/1c66a50d-2aa6-4d19-96ac-7bbd1e690545" />

https://docs.python.org/3/library/concurrent.futures.html

Code after:
        soon as the actor has started running. We use this mechanism to
        detect when a replica has been allocated a worker slot.
        At this time, the replica can transition from PENDING_ALLOCATION
        to PENDING_INITIALIZATION startup state.

        Returns:
            The PID, actor ID, node ID, node IP, and log filepath id of the replica.
        """

        return (
            os.getpid(),
            ray.get_runtime_context().get_actor_id(),
            ray.get_runtime_context().get_worker_id(),
            ray.get_runtime_context().get_node_id(),
            ray.util.get_node_ip_address(),
            ray.util.get_node_instance_id(),
            get_component_logger_file_path(),
        )

    def list_outbound_deployments(self) -> Optional[List[DeploymentID]]:
        return self._replica_impl.list_outbound_deployments()

    async def initialize_and_get_metadata(
        self, deployment_config: DeploymentConfig = None, rank: ReplicaRank = None
    ) -> ReplicaMetadata:
        """Handles initializing the replica.

        Returns: 5-tuple containing
            1. DeploymentConfig of the replica
            2. DeploymentVersion of the replica
            3. Initialization duration in seconds
            4. Port
            5. FastAPI `docs_path`, if relevant (i.e. this is an ingress deployment integrated with FastAPI).
        """
        # Unused `_after` argument is for scheduling: passing an ObjectRef
        # allows delaying this call until after the `_after` call has returned.
        await self._replica_impl.initialize(deployment_config, rank)
        return self._replica_impl.get_metadata()

    async def check_health(self):
        await self._replica_impl.check_health()

    async def record_routing_stats(self) -> Dict[str, Any]:
        return await self._replica_impl.record_routing_stats()

    async def reconfigure(
        self, deployment_config, rank: ReplicaRank, route_prefix: Optional[str] = None
    ) -> ReplicaMetadata:
        await self._replica_impl.reconfigure(deployment_config, rank, route_prefix)
        return self._replica_impl.get_metadata()