python/ray/serve/deployment_state.py
def _check_and_update_replicas(self) -> bool:
        """
        Check current state of all DeploymentReplica being tracked, and compare
        with state container from previous update() cycle to see if any state
        transition happened.

        Returns if any running replicas transitioned to another state.
        """

        running_replicas_changed = False
        for replica in self._replicas.pop(states=[ReplicaState.RUNNING]):
            if replica.check_health():
                self._replicas.add(ReplicaState.RUNNING, replica)
            else:
                running_replicas_changed = True
                logger.warning(
                    f"Replica {replica.replica_tag} of deployment "
                    f"{self._name} failed health check, stopping it."
                )
                replica.stop(graceful=False)
                self._replicas.add(ReplicaState.STOPPING, replica)
                # If this is a replica of the target version, the deployment
                # enters the "UNHEALTHY" status until the replica is
                # recovered or a new deploy happens.
                if replica.version == self._target_version:
                    self._curr_status_info: DeploymentStatusInfo = DeploymentStatusInfo(
                        DeploymentStatus.UNHEALTHY
                    )

        slow_start_replicas = []
        slow_start, starting_to_running = self._check_startup_replicas(
            ReplicaState.STARTING
        )
        slow_update, updating_to_running = self._check_startup_replicas(
            ReplicaState.UPDATING
        )
        slow_recover, recovering_to_running = self._check_startup_replicas(
            ReplicaState.RECOVERING, stop_on_slow=True
        )

        slow_start_replicas = slow_start + slow_update + slow_recover
        running_replicas_changed = (
            running_replicas_changed
            or starting_to_running
            or updating_to_running
            or recovering_to_running
        )

        if (
            len(slow_start_replicas)
            and time.time() - self._prev_startup_warning > SLOW_STARTUP_WARNING_PERIOD_S
        ):

            pending_allocation = []
            pending_initialization = []

            for replica, startup_status in slow_start_replicas:
                if startup_status == ReplicaStartupStatus.PENDING_ALLOCATION:
                    pending_allocation.append(replica)
                if startup_status == ReplicaStartupStatus.PENDING_INITIALIZATION:
                    pending_initialization.append(replica)

            if len(pending_allocation) > 0:
                required, available = slow_start_replicas[0][0].resource_requirements()
                logger.warning(
                    f"Deployment '{self._name}' has "
                    f"{len(pending_allocation)} replicas that have taken "
                    f"more than {SLOW_STARTUP_WARNING_S}s to be scheduled. "
                    f"This may be caused by waiting for the cluster to "
                    f"auto-scale, or waiting for a runtime environment "
                    f"to install. "
                    f"Resources required for each replica: {required}, "
                    f"resources available: {available}."
                )
                if _SCALING_LOG_ENABLED:
                    print_verbose_scaling_log()

            if len(pending_initialization) > 0:
                logger.warning(
                    f"Deployment '{self._name}' has "
                    f"{len(pending_initialization)} replicas that have taken "
                    f"more than {SLOW_STARTUP_WARNING_S}s to initialize. This "
                    f"may be caused by a slow __init__ or reconfigure method."
                )

            self._prev_startup_warning = time.time()

        for replica in self._replicas.pop(states=[ReplicaState.STOPPING]):
            stopped = replica.check_stopped()
            if not stopped:
                self._replicas.add(ReplicaState.STOPPING, replica)

        return running_replicas_changed

        Check current state of all DeploymentReplica being tracked, and compare
        with state container from previous update() cycle to see if any state
        transition happened.

        Returns if any running replicas transitioned to another state.
        
Check current state of all DeploymentReplica being tracked, and compare
with state container from previous update() cycle to see if any state
transition happened.

Returns if any running replicas transitioned to another state.
Check current state of all DeploymentReplica being tracked and compare with state container from previous update () cycle to see if any state transition happened . Returns if any running replicas transitioned to another state .
def _check_and_update_replicas(self) -> bool:
        running_replicas_changed = False
        for replica in self._replicas.pop(states=[ReplicaState.RUNNING]):
            if replica.check_health():
                self._replicas.add(ReplicaState.RUNNING, replica)
            else:
                running_replicas_changed = True
                logger.warning(
                    f"Replica {replica.replica_tag} of deployment "
                    f"{self._name} failed health check, stopping it."
                )
                replica.stop(graceful=False)
                self._replicas.add(ReplicaState.STOPPING, replica)
                if replica.version == self._target_version:
                    self._curr_status_info: DeploymentStatusInfo = DeploymentStatusInfo(
                        DeploymentStatus.UNHEALTHY
                    )
        slow_start_replicas = []
        slow_start, starting_to_running = self._check_startup_replicas(
            ReplicaState.STARTING
        )
        slow_update, updating_to_running = self._check_startup_replicas(
            ReplicaState.UPDATING
        )
        slow_recover, recovering_to_running = self._check_startup_replicas(
            ReplicaState.RECOVERING, stop_on_slow=True
        )
        slow_start_replicas = slow_start + slow_update + slow_recover
        running_replicas_changed = (
            running_replicas_changed
            or starting_to_running
            or updating_to_running
            or recovering_to_running
        )
        if (
            len(slow_start_replicas)
            and time.time() - self._prev_startup_warning > SLOW_STARTUP_WARNING_PERIOD_S
        ):
            pending_allocation = []
            pending_initialization = []
            for replica, startup_status in slow_start_replicas:
                if startup_status == ReplicaStartupStatus.PENDING_ALLOCATION:
                    pending_allocation.append(replica)
                if startup_status == ReplicaStartupStatus.PENDING_INITIALIZATION:
                    pending_initialization.append(replica)
            if len(pending_allocation) > 0:
                required, available = slow_start_replicas[0][0].resource_requirements()
                logger.warning(
                    f"Deployment '{self._name}' has "
                    f"{len(pending_allocation)} replicas that have taken "
                    f"more than {SLOW_STARTUP_WARNING_S}s to be scheduled. "
                    f"This may be caused by waiting for the cluster to "
                    f"auto-scale, or waiting for a runtime environment "
                    f"to install. "
                    f"Resources required for each replica: {required}, "
                    f"resources available: {available}."
                )
                if _SCALING_LOG_ENABLED:
                    print_verbose_scaling_log()
            if len(pending_initialization) > 0:
                logger.warning(
                    f"Deployment '{self._name}' has "
                    f"{len(pending_initialization)} replicas that have taken "
                    f"more than {SLOW_STARTUP_WARNING_S}s to initialize. This "
                    f"may be caused by a slow __init__ or reconfigure method."
                )
            self._prev_startup_warning = time.time()
        for replica in self._replicas.pop(states=[ReplicaState.STOPPING]):
            stopped = replica.check_stopped()
            if not stopped:
                self._replicas.add(ReplicaState.STOPPING, replica)
        return running_replicas_changed
def _check_and_update_replicas ( self ) -> bool : running_replicas_changed = False for replica in self . _replicas . pop ( states = [ ReplicaState . RUNNING ] ) : if replica . check_health ( ) : self . _replicas . add ( ReplicaState . RUNNING , replica ) else : running_replicas_changed = True logger . warning ( f"Replica {replica.replica_tag} of deployment " f"{self._name} failed health check, stopping it." ) replica . stop ( graceful = False ) self . _replicas . add ( ReplicaState . STOPPING , replica ) if replica . version == self . _target_version : self . _curr_status_info : DeploymentStatusInfo = DeploymentStatusInfo ( DeploymentStatus . UNHEALTHY ) slow_start_replicas = [ ] slow_start , starting_to_running = self . _check_startup_replicas ( ReplicaState . STARTING ) slow_update , updating_to_running = self . _check_startup_replicas ( ReplicaState . UPDATING ) slow_recover , recovering_to_running = self . _check_startup_replicas ( ReplicaState . RECOVERING , stop_on_slow = True ) slow_start_replicas = slow_start + slow_update + slow_recover running_replicas_changed = ( running_replicas_changed or starting_to_running or updating_to_running or recovering_to_running ) if ( len ( slow_start_replicas ) and time . time ( ) - self . _prev_startup_warning > SLOW_STARTUP_WARNING_PERIOD_S ) : pending_allocation = [ ] pending_initialization = [ ] for replica , startup_status in slow_start_replicas : if startup_status == ReplicaStartupStatus . PENDING_ALLOCATION : pending_allocation . append ( replica ) if startup_status == ReplicaStartupStatus . PENDING_INITIALIZATION : pending_initialization . append ( replica ) if len ( pending_allocation ) > 0 : required , available = slow_start_replicas [ 0 ] [ 0 ] . resource_requirements ( ) logger . warning ( f"Deployment '{self._name}' has " f"{len(pending_allocation)} replicas that have taken " f"more than {SLOW_STARTUP_WARNING_S}s to be scheduled. " f"This may be caused by waiting for the cluster to " f"auto-scale, or waiting for a runtime environment " f"to install. " f"Resources required for each replica: {required}, " f"resources available: {available}." ) if _SCALING_LOG_ENABLED : print_verbose_scaling_log ( ) if len ( pending_initialization ) > 0 : logger . warning ( f"Deployment '{self._name}' has " f"{len(pending_initialization)} replicas that have taken " f"more than {SLOW_STARTUP_WARNING_S}s to initialize. This " f"may be caused by a slow __init__ or reconfigure method." ) self . _prev_startup_warning = time . time ( ) for replica in self . _replicas . pop ( states = [ ReplicaState . STOPPING ] ) : stopped = replica . check_stopped ( ) if not stopped : self . _replicas . add ( ReplicaState . STOPPING , replica ) return running_replicas_changed
Check current state of all DeploymentReplica being tracked, and compare
with state container from previous update() cycle to see if any state
transition happened.
Check current state of all DeploymentReplica being tracked and compare with state container from previous update () cycle to see if any state transition happened .
"""
        Check current state of all DeploymentReplica being tracked, and compare
        with state container from previous update() cycle to see if any state
        transition happened.

        Returns if any running replicas transitioned to another state.
        """ # If this is a replica of the target version, the deployment # enters the "UNHEALTHY" status until the replica is # recovered or a new deploy happens.
self