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

Code before:
        return deployment_to_replicas_to_stop

    def _schedule_with_pack_strategy(self):
        """Tries to schedule pending replicas using PACK strategy."""
        # Flatten dict of deployment replicas into all replicas,
        # then sort by decreasing resource size
        all_scheduling_requests = sorted(
            _flatten(self._pending_replicas).values(),
            key=lambda r: r.required_resources,
            reverse=True,
        )

        # Fetch node labels for active nodes.
        active_nodes = self._cluster_node_info_cache.get_active_node_ids()
        all_node_labels = {
            node_id: self._cluster_node_info_cache.get_node_labels(node_id)
            for node_id in active_nodes
        }

        # Compute available resources once upfront and update incrementally
        # as replicas are scheduled.
        # Complexity is O(launching + running + nodes + requests * nodes).
        available_resources_per_node = self._get_available_resources_per_node()

        # Running replicas don't change during scheduling, so compute once.
        node_to_running_replicas = self._get_node_to_running_replicas()

        for scheduling_request in all_scheduling_requests:
            target_node = self._pack_schedule_replica(
                scheduling_request,
                all_node_labels,
                available_resources_per_node,
                node_to_running_replicas,
            )
            # Incrementally update available resources for the target node.
            # This is slightly conservative compared to recomputing from
            # scratch (since we subtract from the min of GCS and calculated
            # resources rather than only from the calculated side), but
            # _get_available_resources_per_node is already best-effort.
            if target_node and target_node in available_resources_per_node:
                available_resources_per_node[target_node] = (
                    available_resources_per_node[target_node]
                    - scheduling_request.required_resources
                )

    def _schedule_with_spread_strategy(self):
        """Tries to schedule pending replicas using the SPREAD strategy."""
        for pending_replicas in self._pending_replicas.values():
            if not pending_replicas:
                continue


Review comment (suggestion):
Should `node_to_running_replicas` be updated to include the `target_node` when we return from `_pack_schedule_replica` below? If it's not in the `node_to_running_replicas` mapping prior to the below for loop, it's not included in the non-idle nodes when we check `if len(node_to_running_replicas.get(node_id, set())) > 0`  on line 988. If target node remains in the idle nodes list even after we schedule the first request, the PACK scheduler relies only on best fit when it should also be preferring non-idle nodes.

Code after:
        return deployment_to_replicas_to_stop

    def _schedule_with_pack_strategy(self):
        """Tries to schedule pending replicas using PACK strategy."""
        # Flatten dict of deployment replicas into all replicas,
        # then sort by decreasing resource size
        all_scheduling_requests = sorted(
            _flatten(self._pending_replicas).values(),
            key=lambda r: r.required_resources,
            reverse=True,
        )

        # Fetch node labels for active nodes.
        active_nodes = self._cluster_node_info_cache.get_active_node_ids()
        all_node_labels = {
            node_id: self._cluster_node_info_cache.get_node_labels(node_id)
            for node_id in active_nodes
        }

        # Compute available resources once upfront and update incrementally
        # as replicas are scheduled.
        # Complexity is O(launching + running + nodes + requests * nodes).
        available_resources_per_node = self._get_available_resources_per_node()

        # Seed with running replicas; updated incrementally as replicas are
        # scheduled so that newly occupied nodes are treated as non-idle.
        node_to_assigned_replicas = self._get_node_to_running_replicas()

        for scheduling_request in all_scheduling_requests:
            target_node = self._pack_schedule_replica(
                scheduling_request,
                all_node_labels,
                available_resources_per_node,
                node_to_assigned_replicas,
            )
            # Incrementally update available resources for the target node.
            # This is slightly conservative compared to recomputing from
            # scratch (since we subtract from the min of GCS and calculated
            # resources rather than only from the calculated side), but
            # _get_available_resources_per_node is already best-effort.
            if target_node and target_node in available_resources_per_node:
                available_resources_per_node[target_node] = (
                    available_resources_per_node[target_node]
                    - scheduling_request.required_resources
                )

            # Mark the node as non-idle so subsequent replicas prefer it.
            if target_node and target_node not in node_to_assigned_replicas:
                node_to_assigned_replicas[target_node] = set()
                node_to_assigned_replicas[target_node].add(
                    scheduling_request.replica_id