wehub-resource-sync
1395 lines
56 KiB
Python
1395 lines
56 KiB
Python
import copy
|
|
import logging
|
|
import uuid
|
|
import warnings
|
|
from abc import ABC, abstractmethod
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from functools import total_ordering
|
|
from typing import Any, Callable, Dict, List, Optional, Set, Tuple
|
|
|
|
import ray
|
|
from ray._raylet import node_labels_match_selector
|
|
from ray.serve._private.cluster_node_info_cache import ClusterNodeInfoCache
|
|
from ray.serve._private.common import (
|
|
GANG_PG_NAME_PREFIX,
|
|
CreatePlacementGroupRequest,
|
|
DeploymentID,
|
|
GangPlacementGroupRequest,
|
|
GangReservationResult,
|
|
ReplicaID,
|
|
)
|
|
from ray.serve._private.config import ReplicaConfig
|
|
from ray.serve._private.constants import (
|
|
RAY_SERVE_HIGH_PRIORITY_CUSTOM_RESOURCES,
|
|
RAY_SERVE_USE_COMPACT_SCHEDULING_STRATEGY,
|
|
RAY_SERVE_USE_PACK_SCHEDULING_STRATEGY,
|
|
SERVE_LOGGER_NAME,
|
|
)
|
|
from ray.util.placement_group import PlacementGroup
|
|
from ray.util.scheduling_strategies import (
|
|
LabelMatchExpressionsT,
|
|
NodeAffinitySchedulingStrategy,
|
|
NodeLabelSchedulingStrategy,
|
|
PlacementGroupSchedulingStrategy,
|
|
)
|
|
|
|
logger = logging.getLogger(SERVE_LOGGER_NAME)
|
|
|
|
|
|
class SpreadDeploymentSchedulingPolicy:
|
|
"""A scheduling policy that spreads replicas with best effort."""
|
|
|
|
pass
|
|
|
|
|
|
@total_ordering
|
|
class Resources(dict):
|
|
"""Base for per-node availability vs replica demand resource maps.
|
|
|
|
Do not instantiate directly; use ``AvailableNodeResources`` or
|
|
``RequestedResources``.
|
|
"""
|
|
|
|
# Custom resource priority from environment variable
|
|
CUSTOM_PRIORITY: List[str] = RAY_SERVE_HIGH_PRIORITY_CUSTOM_RESOURCES
|
|
EPSILON = 1e-9
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls is Resources:
|
|
raise TypeError(
|
|
"Resources cannot be instantiated directly; use "
|
|
"AvailableNodeResources or RequestedResources."
|
|
)
|
|
return super().__new__(cls)
|
|
|
|
def __eq__(self, other):
|
|
keys = set(self.keys()) | set(other.keys())
|
|
return all([self.get(k) == other.get(k) for k in keys])
|
|
|
|
def __add__(self, other):
|
|
keys = set(self.keys()) | set(other.keys())
|
|
|
|
kwargs = dict()
|
|
for key in keys:
|
|
if key.startswith(ray._raylet.IMPLICIT_RESOURCE_PREFIX):
|
|
kwargs[key] = min(1.0, self.get(key) + other.get(key))
|
|
else:
|
|
kwargs[key] = self.get(key) + other.get(key)
|
|
|
|
return type(self)(kwargs)
|
|
|
|
def __sub__(self, other):
|
|
keys = set(self.keys()) | set(other.keys())
|
|
kwargs = {key: self.get(key) - other.get(key) for key in keys}
|
|
return type(self)(kwargs)
|
|
|
|
def can_fit(self, other):
|
|
keys = set(self.keys()) | set(other.keys())
|
|
# We add a small epsilon to avoid floating point precision issues.
|
|
return all(self.get(k) + self.EPSILON >= other.get(k) for k in keys)
|
|
|
|
def __lt__(self, other):
|
|
"""Determines priority when sorting a list of SoftResources.
|
|
1. Custom resources defined in RAY_SERVE_HIGH_PRIORITY_CUSTOM_RESOURCES (sorted by priority)
|
|
2. GPU
|
|
3. CPU
|
|
4. memory
|
|
5. Other custom resources
|
|
This means a resource with a larger number of high-priority resources is always
|
|
sorted higher than one with fewer, regardless of other types.
|
|
"""
|
|
|
|
keys = set(self.keys()) | set(other.keys())
|
|
custom_keys = keys - {"GPU", "CPU", "memory"}
|
|
|
|
for key in self.CUSTOM_PRIORITY:
|
|
if self.get(key) < other.get(key):
|
|
return True
|
|
elif self.get(key) > other.get(key):
|
|
return False
|
|
|
|
if self.get("GPU") < other.get("GPU"):
|
|
return True
|
|
elif self.get("GPU") > other.get("GPU"):
|
|
return False
|
|
|
|
if self.get("CPU") < other.get("CPU"):
|
|
return True
|
|
elif self.get("CPU") > other.get("CPU"):
|
|
return False
|
|
|
|
if self.get("memory") < other.get("memory"):
|
|
return True
|
|
elif self.get("memory") > other.get("memory"):
|
|
return False
|
|
|
|
for key in custom_keys - set(self.CUSTOM_PRIORITY):
|
|
if self.get(key) < other.get(key):
|
|
return True
|
|
elif self.get(key) > other.get(key):
|
|
return False
|
|
|
|
return False
|
|
|
|
|
|
def _format_resources_for_scheduling_log(resources: Resources) -> str:
|
|
"""Compact resource summary for pack scheduling logs."""
|
|
priority_keys = list(Resources.CUSTOM_PRIORITY) + ["GPU", "CPU", "memory"]
|
|
seen = set()
|
|
parts = []
|
|
for key in priority_keys:
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
val = resources.get(key)
|
|
if val:
|
|
parts.append(f"{key}={val:g}" if isinstance(val, float) else f"{key}={val}")
|
|
for key in sorted(set(resources.keys()) - seen):
|
|
val = resources.get(key)
|
|
if val:
|
|
parts.append(f"{key}={val:g}" if isinstance(val, float) else f"{key}={val}")
|
|
return ", ".join(parts) if parts else "none"
|
|
|
|
|
|
class AvailableNodeResources(Resources):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def get(self, key: str):
|
|
val = super().get(key)
|
|
if val is not None:
|
|
return val
|
|
|
|
# Implicit resources by default have 1 total
|
|
# NOTE(zcin): Implicit resources are automatically and
|
|
# artificially injected into each node and are used to limit how
|
|
# many replicas of the same deployment can run on a single node.
|
|
# This is used to enforce `max_replicas_per_node`.
|
|
if key.startswith(ray._raylet.IMPLICIT_RESOURCE_PREFIX):
|
|
return 1
|
|
|
|
# Otherwise by default there is 0 of this resource
|
|
return 0
|
|
|
|
|
|
class RequestedResources(Resources):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def get(self, key: str):
|
|
# We DON'T inject implicit resources for required resources.
|
|
val = super().get(key)
|
|
if val is not None:
|
|
return val
|
|
|
|
return 0
|
|
|
|
|
|
class ReplicaSchedulingRequestStatus(str, Enum):
|
|
"""The status of a replica scheduling request."""
|
|
|
|
IN_PROGRESS = "IN_PROGRESS"
|
|
SUCCEEDED = "SUCCEEDED"
|
|
ACTOR_CREATION_FAILED = "ACTOR_CREATION_FAILED"
|
|
PLACEMENT_GROUP_CREATION_FAILED = "PLACEMENT_GROUP_CREATION_FAILED"
|
|
|
|
|
|
@dataclass
|
|
class ReplicaSchedulingRequest:
|
|
"""Request to schedule a single replica.
|
|
|
|
The scheduler is responsible for scheduling
|
|
based on the deployment scheduling policy.
|
|
"""
|
|
|
|
replica_id: ReplicaID
|
|
actor_def: ray.actor.ActorClass
|
|
actor_resources: Dict
|
|
actor_options: Dict
|
|
actor_init_args: Tuple
|
|
on_scheduled: Callable
|
|
status: ReplicaSchedulingRequestStatus = ReplicaSchedulingRequestStatus.IN_PROGRESS
|
|
# Placement group bundles and strategy *for this replica*.
|
|
# These are optional: by default replicas do not have a placement group.
|
|
placement_group_bundles: Optional[List[Dict[str, float]]] = None
|
|
placement_group_strategy: Optional[str] = None
|
|
placement_group_bundle_label_selector: Optional[List[Dict[str, str]]] = None
|
|
placement_group_fallback_strategy: Optional[List[Dict[str, Any]]] = None
|
|
max_replicas_per_node: Optional[int] = None
|
|
# Gang scheduling fields -- if set, replica should be scheduled on
|
|
# the reserved gang placement group at the specified bundle index.
|
|
gang_placement_group: Optional[PlacementGroup] = None
|
|
# Bundle index inside gang_placement_group where this replica actor is scheduled.
|
|
# Example: If each replica uses 2 bundles, ranks 0 and 1 use indices 0 and 2 respectively.
|
|
gang_pg_index: Optional[int] = None
|
|
|
|
@property
|
|
def requested_resources(self) -> RequestedResources:
|
|
"""The resources required to schedule this replica on a node.
|
|
|
|
STRICT_PACK placement group: sum of all bundles.
|
|
Other placement groups: bundle 0.
|
|
Otherwise: actor resources.
|
|
"""
|
|
|
|
if (
|
|
self.placement_group_bundles is not None
|
|
and self.placement_group_strategy == "STRICT_PACK"
|
|
):
|
|
return sum(
|
|
[RequestedResources(bundle) for bundle in self.placement_group_bundles],
|
|
RequestedResources(),
|
|
)
|
|
elif self.placement_group_bundles is not None:
|
|
return RequestedResources(self.placement_group_bundles[0])
|
|
else:
|
|
required = RequestedResources(self.actor_resources)
|
|
|
|
# Using implicit resource (resources that every node
|
|
# implicitly has and total is 1)
|
|
# to limit the number of replicas on a single node.
|
|
if (
|
|
self.max_replicas_per_node is not None
|
|
and self.max_replicas_per_node > 0
|
|
):
|
|
deployment_id = self.replica_id.deployment_id
|
|
implicit_resource = (
|
|
f"{ray._raylet.IMPLICIT_RESOURCE_PREFIX}"
|
|
f"{deployment_id.app_name}:{deployment_id.name}"
|
|
)
|
|
required[implicit_resource] = 1.0 / self.max_replicas_per_node
|
|
|
|
return required
|
|
|
|
|
|
@dataclass
|
|
class DeploymentDownscaleRequest:
|
|
"""Request to stop a certain number of replicas.
|
|
|
|
The scheduler is responsible for choosing the replicas to stop.
|
|
"""
|
|
|
|
deployment_id: DeploymentID
|
|
num_to_stop: int
|
|
|
|
# The scheduler uses these to select complete gangs to stop.
|
|
gang_id_by_replica: Optional[Dict[ReplicaID, str]] = None
|
|
replicas_by_gang_id: Optional[Dict[str, Set[ReplicaID]]] = None
|
|
gang_size: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class DeploymentSchedulingInfo:
|
|
deployment_id: DeploymentID
|
|
scheduling_policy: Any
|
|
actor_resources: Optional[RequestedResources] = None
|
|
label_selector: Optional[Dict[str, str]] = None
|
|
placement_group_bundles: Optional[List[RequestedResources]] = None
|
|
bundle_label_selector: Optional[List[Dict[str, str]]] = None
|
|
fallback_strategy: Optional[List[Dict[str, Any]]] = None
|
|
placement_group_strategy: Optional[str] = None
|
|
max_replicas_per_node: Optional[int] = None
|
|
|
|
@property
|
|
def required_resources(self) -> RequestedResources:
|
|
"""The resources required to schedule a replica of this deployment on a node.
|
|
|
|
STRICT_PACK placement group: sum of all bundles.
|
|
Other placement groups: bundle 0.
|
|
Otherwise: actor resources.
|
|
"""
|
|
|
|
if (
|
|
self.placement_group_bundles is not None
|
|
and self.placement_group_strategy == "STRICT_PACK"
|
|
):
|
|
return sum(self.placement_group_bundles, RequestedResources())
|
|
elif self.placement_group_bundles is not None:
|
|
return RequestedResources(self.placement_group_bundles[0])
|
|
else:
|
|
if self.actor_resources is None:
|
|
required = RequestedResources()
|
|
else:
|
|
required = RequestedResources(self.actor_resources)
|
|
|
|
# Using implicit resource (resources that every node
|
|
# implicitly has and total is 1)
|
|
# to limit the number of replicas on a single node.
|
|
if (
|
|
self.max_replicas_per_node is not None
|
|
and self.max_replicas_per_node > 0
|
|
):
|
|
implicit_resource = (
|
|
f"{ray._raylet.IMPLICIT_RESOURCE_PREFIX}"
|
|
f"{self.deployment_id.app_name}:{self.deployment_id.name}"
|
|
)
|
|
required[implicit_resource] = 1.0 / self.max_replicas_per_node
|
|
|
|
return required
|
|
|
|
def is_non_strict_pack_pg(self) -> bool:
|
|
return (
|
|
self.placement_group_bundles is not None
|
|
and self.placement_group_strategy != "STRICT_PACK"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class LaunchingReplicaInfo:
|
|
"""Describes a replica for which a schedule request has been sent to
|
|
core but has not been scheduled (placed on a node) yet.
|
|
|
|
Args:
|
|
target_node_id: The exact node that's been requested for this
|
|
replica. This is best effort and may not be fulfilled.
|
|
target_labels: The node labels that have been requested for this
|
|
replica. This is best effort and may not be fulfilled.
|
|
"""
|
|
|
|
target_node_id: Optional[str] = None
|
|
target_labels: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
def _flatten(
|
|
deployment_to_replicas: Dict[DeploymentID, Dict[ReplicaID, Any]],
|
|
) -> Dict[ReplicaID, Any]:
|
|
"""Flattens a dict of {deployment_id: {replica_id: val}} to {replica_id: val}."""
|
|
|
|
return {
|
|
replica_id: val
|
|
for replicas in deployment_to_replicas.values()
|
|
for replica_id, val in replicas.items()
|
|
}
|
|
|
|
|
|
class DeploymentScheduler(ABC):
|
|
"""A centralized scheduler for all Serve deployments.
|
|
|
|
It makes a batch of scheduling decisions in each update cycle.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
cluster_node_info_cache: ClusterNodeInfoCache,
|
|
head_node_id: str,
|
|
create_placement_group_fn: Callable,
|
|
):
|
|
# {deployment_id: scheduling_policy}
|
|
self._deployments: Dict[DeploymentID, DeploymentSchedulingInfo] = {}
|
|
# Replicas that are waiting to be scheduled.
|
|
# {deployment_id: {replica_id: deployment_upscale_request}}
|
|
self._pending_replicas: Dict[
|
|
DeploymentID, Dict[str, ReplicaSchedulingRequest]
|
|
] = defaultdict(dict)
|
|
# Replicas that are being scheduled.
|
|
# The underlying actors have been submitted.
|
|
# {deployment_id: {replica_id: target_node_id}}
|
|
self._launching_replicas: Dict[
|
|
DeploymentID, Dict[str, LaunchingReplicaInfo]
|
|
] = defaultdict(dict)
|
|
# Replicas that are recovering.
|
|
# We don't know where those replicas are running.
|
|
# {deployment_id: {replica_id}}
|
|
self._recovering_replicas = defaultdict(set)
|
|
# Replicas that are running.
|
|
# We know where those replicas are running.
|
|
# {deployment_id: {replica_id: running_node_id}}
|
|
self._running_replicas = defaultdict(dict)
|
|
# Dedupes pack scheduling order logs across control loops.
|
|
self._last_pack_schedule_order_log_key: Optional[tuple] = None
|
|
# Dedupes repeated pack placement failure logs for stuck replicas.
|
|
self._logged_pack_placement_failures: Set[ReplicaID] = set()
|
|
|
|
self._cluster_node_info_cache = cluster_node_info_cache
|
|
self._head_node_id = head_node_id
|
|
self._create_placement_group_fn = create_placement_group_fn
|
|
|
|
def on_deployment_created(
|
|
self,
|
|
deployment_id: DeploymentID,
|
|
scheduling_policy: SpreadDeploymentSchedulingPolicy,
|
|
) -> None:
|
|
"""Called whenever a new deployment is created."""
|
|
assert deployment_id not in self._pending_replicas
|
|
assert deployment_id not in self._launching_replicas
|
|
assert deployment_id not in self._recovering_replicas
|
|
assert deployment_id not in self._running_replicas
|
|
self._deployments[deployment_id] = DeploymentSchedulingInfo(
|
|
deployment_id=deployment_id, scheduling_policy=scheduling_policy
|
|
)
|
|
|
|
def on_deployment_deployed(
|
|
self,
|
|
deployment_id: DeploymentID,
|
|
replica_config: ReplicaConfig,
|
|
) -> None:
|
|
assert deployment_id in self._deployments
|
|
|
|
info = self._deployments[deployment_id]
|
|
info.actor_resources = RequestedResources(replica_config.resource_dict)
|
|
info.label_selector = replica_config.ray_actor_options.get("label_selector")
|
|
info.bundle_label_selector = (
|
|
replica_config.placement_group_bundle_label_selector
|
|
)
|
|
info.fallback_strategy = replica_config.ray_actor_options.get(
|
|
"fallback_strategy"
|
|
)
|
|
info.max_replicas_per_node = replica_config.max_replicas_per_node
|
|
if replica_config.placement_group_bundles:
|
|
info.placement_group_bundles = [
|
|
RequestedResources(bundle)
|
|
for bundle in replica_config.placement_group_bundles
|
|
]
|
|
if replica_config.placement_group_strategy:
|
|
info.placement_group_strategy = replica_config.placement_group_strategy
|
|
|
|
def on_deployment_deleted(self, deployment_id: DeploymentID) -> None:
|
|
"""Called whenever a deployment is deleted."""
|
|
assert not self._pending_replicas[deployment_id]
|
|
self._pending_replicas.pop(deployment_id, None)
|
|
|
|
assert not self._launching_replicas[deployment_id]
|
|
self._launching_replicas.pop(deployment_id, None)
|
|
|
|
assert not self._recovering_replicas[deployment_id]
|
|
self._recovering_replicas.pop(deployment_id, None)
|
|
|
|
assert not self._running_replicas[deployment_id]
|
|
self._running_replicas.pop(deployment_id, None)
|
|
|
|
del self._deployments[deployment_id]
|
|
|
|
def on_replica_stopping(self, replica_id: ReplicaID) -> None:
|
|
"""Called whenever a deployment replica is being stopped."""
|
|
deployment_id = replica_id.deployment_id
|
|
self._pending_replicas[deployment_id].pop(replica_id, None)
|
|
self._launching_replicas[deployment_id].pop(replica_id, None)
|
|
self._recovering_replicas[deployment_id].discard(replica_id)
|
|
self._running_replicas[deployment_id].pop(replica_id, None)
|
|
self._logged_pack_placement_failures.discard(replica_id)
|
|
|
|
def on_replica_running(self, replica_id: ReplicaID, node_id: str) -> None:
|
|
"""Called whenever a deployment replica is running with a known node id."""
|
|
deployment_id = replica_id.deployment_id
|
|
assert replica_id not in self._pending_replicas[deployment_id]
|
|
|
|
self._launching_replicas[deployment_id].pop(replica_id, None)
|
|
self._recovering_replicas[deployment_id].discard(replica_id)
|
|
|
|
self._running_replicas[deployment_id][replica_id] = node_id
|
|
|
|
def on_replica_recovering(self, replica_id: ReplicaID) -> None:
|
|
"""Called whenever a deployment replica is recovering."""
|
|
deployment_id = replica_id.deployment_id
|
|
assert replica_id not in self._pending_replicas[deployment_id]
|
|
assert replica_id not in self._launching_replicas[deployment_id]
|
|
assert replica_id not in self._running_replicas[deployment_id]
|
|
assert replica_id not in self._recovering_replicas[deployment_id]
|
|
|
|
self._recovering_replicas[deployment_id].add(replica_id)
|
|
|
|
def _on_replica_launching(
|
|
self,
|
|
replica_id: ReplicaID,
|
|
target_node_id: Optional[str] = None,
|
|
target_labels: Optional[Dict[str, Any]] = None,
|
|
):
|
|
deployment_id = replica_id.deployment_id
|
|
self._launching_replicas[deployment_id][replica_id] = LaunchingReplicaInfo(
|
|
target_node_id=target_node_id, target_labels=target_labels
|
|
)
|
|
|
|
def _get_node_to_running_replicas(
|
|
self, deployment_id: Optional[DeploymentID] = None
|
|
) -> Dict[str, Set[ReplicaID]]:
|
|
res = defaultdict(set)
|
|
if deployment_id:
|
|
for replica_id, node_id in self._running_replicas[deployment_id].items():
|
|
res[node_id].add(replica_id)
|
|
else:
|
|
for _, replicas in self._running_replicas.items():
|
|
for replica_id, node_id in replicas.items():
|
|
res[node_id].add(replica_id)
|
|
|
|
return res
|
|
|
|
def _get_available_resources_per_node(self) -> Dict[str, AvailableNodeResources]:
|
|
"""Gets current available resources per node.
|
|
|
|
This returns a conservative view of the available resources
|
|
currently in the cluster. It returns the minimum of:
|
|
|
|
1. The available resources per node fetched and cached from the
|
|
GCS every control loop.
|
|
2. The remaining resources left over on each node after
|
|
subtracting the resources taken up by running (already
|
|
scheduled by core) and launching (to-be-scheduled and soft
|
|
targeting that node) replicas.
|
|
|
|
Note that (1) may not be accurate because it uses cached info
|
|
and there is a delay from fetching data from GCS, and (2) may
|
|
not be accurate because there can be other actors (not replicas)
|
|
running in the cluster, and launching replicas may not end up on
|
|
the node we're targeting. So the information returned from this
|
|
method is only best effort.
|
|
"""
|
|
|
|
available_resources = (
|
|
self._cluster_node_info_cache.get_available_resources_per_node()
|
|
)
|
|
total_resources = self._cluster_node_info_cache.get_total_resources_per_node()
|
|
|
|
gcs_info = {
|
|
node_id: AvailableNodeResources(r)
|
|
for node_id, r in available_resources.items()
|
|
}
|
|
|
|
# Manually calculate available resources per node by subtracting
|
|
# launching and running replicas from total resources
|
|
total_minus_replicas = {
|
|
node_id: AvailableNodeResources(resources)
|
|
for node_id, resources in total_resources.items()
|
|
}
|
|
for deployment_id, replicas in self._launching_replicas.items():
|
|
deployment = self._deployments[deployment_id]
|
|
for info in replicas.values():
|
|
target_node_id = info.target_node_id
|
|
if not target_node_id or target_node_id not in total_minus_replicas:
|
|
continue
|
|
|
|
total_minus_replicas[target_node_id] -= deployment.required_resources
|
|
|
|
for deployment_id, replicas in self._running_replicas.items():
|
|
deployment = self._deployments[deployment_id]
|
|
for node_id in replicas.values():
|
|
if node_id not in total_minus_replicas:
|
|
continue
|
|
|
|
total_minus_replicas[node_id] -= deployment.required_resources
|
|
|
|
def custom_min(a: AvailableNodeResources, b: AvailableNodeResources):
|
|
keys = set(a.keys()) | set(b.keys())
|
|
res = AvailableNodeResources()
|
|
for key in keys:
|
|
res[key] = min(a.get(key), b.get(key))
|
|
return res
|
|
|
|
# Filter by active node ids (alive but not draining)
|
|
return {
|
|
node_id: custom_min(
|
|
gcs_info.get(node_id, AvailableNodeResources()),
|
|
total_minus_replicas.get(node_id, AvailableNodeResources()),
|
|
)
|
|
for node_id in self._cluster_node_info_cache.get_active_node_ids()
|
|
}
|
|
|
|
def _best_fit_node(
|
|
self,
|
|
required_resources: RequestedResources,
|
|
available_resources: Dict[str, AvailableNodeResources],
|
|
) -> Optional[str]:
|
|
"""Chooses a node using best fit strategy.
|
|
|
|
This strategy picks the node where, if the required resources
|
|
were to be scheduled on that node, it will leave the smallest
|
|
remaining space. This minimizes fragmentation of resources.
|
|
"""
|
|
|
|
min_remaining_space = None
|
|
chosen_node = None
|
|
|
|
for node_id in available_resources:
|
|
if not available_resources[node_id].can_fit(required_resources):
|
|
continue
|
|
|
|
# TODO(zcin): We can make this better by only considering
|
|
# custom resources that required_resources has.
|
|
remaining_space = available_resources[node_id] - required_resources
|
|
if min_remaining_space is None or remaining_space < min_remaining_space:
|
|
min_remaining_space = remaining_space
|
|
chosen_node = node_id
|
|
|
|
return chosen_node
|
|
|
|
@abstractmethod
|
|
def schedule(
|
|
self,
|
|
upscales: Dict[DeploymentID, List[ReplicaSchedulingRequest]],
|
|
downscales: Dict[DeploymentID, DeploymentDownscaleRequest],
|
|
) -> Dict[DeploymentID, Set[ReplicaID]]:
|
|
"""Called for each update cycle to do batch scheduling.
|
|
|
|
Args:
|
|
upscales: a dict of deployment name to a list of replicas to schedule.
|
|
downscales: a dict of deployment name to a downscale request.
|
|
|
|
Returns:
|
|
The name of replicas to stop for each deployment.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _schedule_replica(
|
|
self,
|
|
scheduling_request: ReplicaSchedulingRequest,
|
|
default_scheduling_strategy: str,
|
|
target_node_id: Optional[str] = None,
|
|
target_labels: Optional[LabelMatchExpressionsT] = None,
|
|
) -> bool:
|
|
"""Schedule a replica from a scheduling request.
|
|
|
|
The following special scheduling strategies will be used, in
|
|
order of highest to lowest priority.
|
|
1. If a replica requires gang scheduling, we will use a reserved
|
|
gang placement group.
|
|
2. If a replica requires placement groups, we will choose to use
|
|
a `PlacementGroupSchedulingStrategy`. This can also take a
|
|
target node into consideration (soft target), if provided.
|
|
However it cannot take into account target labels.
|
|
3. If a `target_node_id` is provided, we will choose to use a
|
|
`NodeAffinitySchedulingStrategy`.
|
|
4. If `target_labels` is provided, we will choose to use a
|
|
`NodeLabelSchedulingStrategy`.
|
|
|
|
Args:
|
|
scheduling_request: A request to schedule a replica.
|
|
default_scheduling_strategy: The scheduling strategy to fall
|
|
back to if no special scheduling strategy is necessary.
|
|
target_node_id: Attempt to schedule this replica onto this
|
|
target node.
|
|
target_labels: Attempt to schedule this replica onto nodes
|
|
with these target labels.
|
|
|
|
Returns:
|
|
True if the replica was successfully scheduled, False otherwise.
|
|
"""
|
|
|
|
replica_id = scheduling_request.replica_id
|
|
deployment_id = replica_id.deployment_id
|
|
placement_group = None
|
|
|
|
scheduling_strategy = default_scheduling_strategy
|
|
|
|
if scheduling_request.gang_placement_group is not None:
|
|
# Gang scheduling -- use the reserved gang placement group
|
|
placement_group = scheduling_request.gang_placement_group
|
|
scheduling_strategy = PlacementGroupSchedulingStrategy(
|
|
placement_group=placement_group,
|
|
placement_group_bundle_index=scheduling_request.gang_pg_index,
|
|
placement_group_capture_child_tasks=True,
|
|
)
|
|
# TODO (jeffreywang): Add support for target labels and node affinity
|
|
target_labels = None
|
|
target_node_id = None
|
|
elif scheduling_request.placement_group_bundles is not None:
|
|
placement_group_strategy = (
|
|
scheduling_request.placement_group_strategy
|
|
if scheduling_request.placement_group_strategy
|
|
else "PACK"
|
|
)
|
|
try:
|
|
pg = self._create_placement_group_fn(
|
|
CreatePlacementGroupRequest(
|
|
bundles=scheduling_request.placement_group_bundles,
|
|
strategy=placement_group_strategy,
|
|
target_node_id=target_node_id,
|
|
name=scheduling_request.actor_options["name"],
|
|
bundle_label_selector=scheduling_request.placement_group_bundle_label_selector,
|
|
)
|
|
)
|
|
except Exception:
|
|
# We add a defensive exception here, so the controller can
|
|
# make progress even if the placement group isn't created.
|
|
# See https://github.com/ray-project/ray/issues/43888.
|
|
logger.exception(
|
|
f"Failed to create a placement group for {replica_id}."
|
|
)
|
|
scheduling_request.status = (
|
|
ReplicaSchedulingRequestStatus.PLACEMENT_GROUP_CREATION_FAILED
|
|
)
|
|
return False
|
|
# Pin the actor as a subset of bundle 0. ReplicaConfig
|
|
# validates that actor resources fit in bundle 0, and
|
|
# required_resources assumes this pin.
|
|
scheduling_strategy = PlacementGroupSchedulingStrategy(
|
|
placement_group=pg,
|
|
placement_group_bundle_index=0,
|
|
placement_group_capture_child_tasks=True,
|
|
)
|
|
target_labels = None
|
|
elif target_node_id is not None:
|
|
scheduling_strategy = NodeAffinitySchedulingStrategy(
|
|
node_id=target_node_id, soft=True, _spill_on_unavailable=True
|
|
)
|
|
target_labels = None
|
|
elif target_labels is not None:
|
|
scheduling_strategy = NodeLabelSchedulingStrategy(
|
|
hard={}, soft=target_labels
|
|
)
|
|
target_node_id = None
|
|
|
|
actor_options = copy.deepcopy(scheduling_request.actor_options)
|
|
if (
|
|
scheduling_request.max_replicas_per_node is not None
|
|
and scheduling_request.max_replicas_per_node > 0
|
|
):
|
|
if "resources" not in actor_options:
|
|
actor_options["resources"] = {}
|
|
# Using implicit resource (resources that every node
|
|
# implicitly has and total is 1)
|
|
# to limit the number of replicas on a single node.
|
|
actor_options["resources"][
|
|
f"{ray._raylet.IMPLICIT_RESOURCE_PREFIX}"
|
|
f"{deployment_id.app_name}:{deployment_id.name}"
|
|
] = (1.0 / scheduling_request.max_replicas_per_node)
|
|
|
|
try:
|
|
actor_handle = scheduling_request.actor_def.options(
|
|
scheduling_strategy=scheduling_strategy,
|
|
**actor_options,
|
|
).remote(*scheduling_request.actor_init_args)
|
|
except Exception:
|
|
# We add a defensive exception here, so the controller can
|
|
# make progress even if the actor options are misconfigured.
|
|
logger.exception(f"Failed to create an actor for {replica_id}.")
|
|
scheduling_request.status = (
|
|
ReplicaSchedulingRequestStatus.ACTOR_CREATION_FAILED
|
|
)
|
|
return False
|
|
|
|
del self._pending_replicas[deployment_id][replica_id]
|
|
self._on_replica_launching(
|
|
replica_id, target_node_id=target_node_id, target_labels=target_labels
|
|
)
|
|
|
|
if isinstance(scheduling_strategy, PlacementGroupSchedulingStrategy):
|
|
placement_group = scheduling_strategy.placement_group
|
|
|
|
scheduling_request.status = ReplicaSchedulingRequestStatus.SUCCEEDED
|
|
scheduling_request.on_scheduled(actor_handle, placement_group=placement_group)
|
|
return True
|
|
|
|
@abstractmethod
|
|
def get_node_to_compact(
|
|
self, allow_new_compaction: bool
|
|
) -> Optional[Tuple[str, float]]:
|
|
"""Returns a node ID to be compacted and a compaction deadlne."""
|
|
raise NotImplementedError
|
|
|
|
def schedule_gang_placement_groups(
|
|
self,
|
|
gang_requests: Dict[DeploymentID, GangPlacementGroupRequest],
|
|
) -> Dict[DeploymentID, GangReservationResult]:
|
|
"""Reserve gang placement groups for gang scheduling.
|
|
|
|
Creates gang placement groups before replicas are created, allowing
|
|
the scheduler to verify resource feasibility upfront.
|
|
|
|
Args:
|
|
gang_requests: A dictionary of deployment ID to gang placement group request.
|
|
|
|
Returns:
|
|
A dictionary of deployment ID to gang reservation result.
|
|
"""
|
|
return {
|
|
deployment_id: self._prepare_gangs_for_deployment(deployment_id, request)
|
|
for deployment_id, request in gang_requests.items()
|
|
}
|
|
|
|
def _prepare_gangs_for_deployment(
|
|
self,
|
|
deployment_id: DeploymentID,
|
|
request: GangPlacementGroupRequest,
|
|
) -> GangReservationResult:
|
|
"""Create gang placement groups for a single deployment.
|
|
|
|
Example:
|
|
- Case 1: Per-replica bundles are defined
|
|
gang_size=2, replica_placement_group_bundles=[{"GPU":1,"CPU":1},{"CPU":1}]
|
|
|
|
Requested gang placement group:
|
|
[{"GPU":1,"CPU":1}, {"CPU":1}, {"GPU":1,"CPU":1}, {"CPU":1}]
|
|
^^^^^^^ replica 0 ^^^^^^^^^^ ^^^^^^^^ replica 1 ^^^^^^^^^
|
|
Replica 0 actor → bundle index 0, replica 1 actor → bundle index 2.
|
|
Remaining bundles (1, 3) are used by child tasks/actors.
|
|
|
|
- Case 2: Per-replica bundles are not defined
|
|
gang_size=2, replica_resource_dict={"CPU":2,"GPU":1}
|
|
|
|
Requested gang placement group:
|
|
[{"CPU":2,"GPU":1}, {"CPU":2,"GPU":1}]
|
|
^^^ replica 0 ^^^ ^^^ replica 1 ^^^
|
|
Replica 0 actor → bundle index 0, replica 1 actor → bundle index 1.
|
|
|
|
Args:
|
|
deployment_id: The deployment to create gangs for.
|
|
request: Contains gang config and number of replicas to add.
|
|
|
|
Returns:
|
|
GangReservationResult with all created gang PGs.
|
|
"""
|
|
gang_size = request.gang_size
|
|
|
|
if request.num_replicas_to_add % gang_size != 0:
|
|
logger.error(
|
|
f"num_replicas_to_add {request.num_replicas_to_add} "
|
|
f"is not divisible by gang_size {gang_size}."
|
|
)
|
|
return GangReservationResult(
|
|
success=False,
|
|
error_message=(
|
|
f"num_replicas_to_add {request.num_replicas_to_add} "
|
|
f"is not divisible by gang_size {gang_size}."
|
|
),
|
|
)
|
|
num_gangs = request.num_replicas_to_add // gang_size
|
|
|
|
per_replica_bundles = request.replica_placement_group_bundles
|
|
has_pg_bundles = (
|
|
per_replica_bundles is not None and len(per_replica_bundles) > 0
|
|
)
|
|
|
|
# Flatten per-replica bundles to form a placement group to atomically reserve resources
|
|
# required for each gang
|
|
gang_pgs: List[PlacementGroup] = []
|
|
gang_ids: List[str] = []
|
|
gang_pg_names: List[str] = []
|
|
for gang_index in range(num_gangs):
|
|
if has_pg_bundles:
|
|
bundles = [
|
|
bundle.copy()
|
|
for _ in range(gang_size)
|
|
for bundle in per_replica_bundles
|
|
]
|
|
label_selector = (
|
|
[
|
|
selector.copy()
|
|
for _ in range(gang_size)
|
|
for selector in request.replica_pg_bundle_label_selector
|
|
]
|
|
if request.replica_pg_bundle_label_selector is not None
|
|
else None
|
|
)
|
|
fallback_strategy = (
|
|
[
|
|
strategy.copy()
|
|
for _ in range(gang_size)
|
|
for strategy in request.replica_pg_fallback_strategy
|
|
]
|
|
if request.replica_pg_fallback_strategy is not None
|
|
else None
|
|
)
|
|
else:
|
|
bundles = [
|
|
request.replica_resource_dict.copy() for _ in range(gang_size)
|
|
]
|
|
label_selector = None
|
|
fallback_strategy = None
|
|
|
|
gang_id = uuid.uuid4().hex[:8]
|
|
pg_name = (
|
|
f"{GANG_PG_NAME_PREFIX}{deployment_id.app_name}"
|
|
f"_{deployment_id.name}"
|
|
f"_{gang_index}_{gang_id}"
|
|
)
|
|
|
|
try:
|
|
pg = self._create_placement_group_fn(
|
|
CreatePlacementGroupRequest(
|
|
bundles=bundles,
|
|
strategy=request.gang_placement_strategy,
|
|
target_node_id=None,
|
|
name=pg_name,
|
|
bundle_label_selector=label_selector,
|
|
fallback_strategy=fallback_strategy,
|
|
)
|
|
)
|
|
gang_pgs.append(pg)
|
|
gang_ids.append(gang_id)
|
|
gang_pg_names.append(pg_name)
|
|
except Exception:
|
|
# Follow the same pattern as single-replica PG creation failure:
|
|
# log and skip this gang so the controller can make progress with
|
|
# the other gangs. The missing replicas will be retried on the next
|
|
# reconciliation loop.
|
|
logger.exception(
|
|
f"Failed to create gang placement group "
|
|
f"{gang_index} for {deployment_id}."
|
|
)
|
|
continue
|
|
|
|
if not gang_pgs:
|
|
return GangReservationResult(
|
|
success=False,
|
|
error_message=(
|
|
f"Failed to create any gang placement groups for {deployment_id}."
|
|
),
|
|
)
|
|
|
|
logger.info(
|
|
f"Created {len(gang_pgs)} of {num_gangs} gang PG(s) for "
|
|
f"{deployment_id}. Actors will wait for resource allocation."
|
|
)
|
|
return GangReservationResult(
|
|
success=True,
|
|
gang_pgs=gang_pgs,
|
|
gang_ids=gang_ids,
|
|
gang_pg_names=gang_pg_names,
|
|
)
|
|
|
|
|
|
class DefaultDeploymentScheduler(DeploymentScheduler):
|
|
def schedule(
|
|
self,
|
|
upscales: Dict[DeploymentID, List[ReplicaSchedulingRequest]],
|
|
downscales: Dict[DeploymentID, DeploymentDownscaleRequest],
|
|
) -> Dict[DeploymentID, Set[ReplicaID]]:
|
|
"""Called for each update cycle to do batch scheduling.
|
|
|
|
Args:
|
|
upscales: a dict of deployment name to a list of replicas to schedule.
|
|
downscales: a dict of deployment name to a downscale request.
|
|
|
|
Returns:
|
|
The IDs of replicas to stop for each deployment.
|
|
"""
|
|
# Update pending replicas from upscales.
|
|
for upscale in upscales.values():
|
|
for scheduling_request in upscale:
|
|
replica_id = scheduling_request.replica_id
|
|
deployment_id = replica_id.deployment_id
|
|
self._pending_replicas[deployment_id][replica_id] = scheduling_request
|
|
|
|
# Check for deprecated environment variable usage
|
|
if RAY_SERVE_USE_COMPACT_SCHEDULING_STRATEGY:
|
|
warnings.warn(
|
|
"The environment variable 'RAY_SERVE_USE_COMPACT_SCHEDULING_STRATEGY' "
|
|
"is deprecated and will be removed in a v2.55.0 release. "
|
|
"Please use 'RAY_SERVE_USE_PACK_SCHEDULING_STRATEGY' instead.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
|
|
# Determine scheduling strategy
|
|
non_strict_pack_pgs_exist = any(
|
|
d.is_non_strict_pack_pg() for d in self._deployments.values()
|
|
)
|
|
use_pack_strategy = (
|
|
RAY_SERVE_USE_PACK_SCHEDULING_STRATEGY and not non_strict_pack_pgs_exist
|
|
)
|
|
|
|
if use_pack_strategy:
|
|
# This branch is only reached if each deployment either:
|
|
# 1. Use STRICT_PACK placement group strategy, or
|
|
# 2. Do not use placement groups at all.
|
|
# This ensures Serve's best-fit node selection is respected by Ray Core
|
|
# (since _soft_target_node_id only works with STRICT_PACK).
|
|
self._schedule_with_pack_strategy()
|
|
else:
|
|
self._schedule_with_spread_strategy()
|
|
|
|
# Handle downscales
|
|
deployment_to_replicas_to_stop = {}
|
|
for downscale in downscales.values():
|
|
deployment_to_replicas_to_stop[
|
|
downscale.deployment_id
|
|
] = self._get_replicas_to_stop(
|
|
downscale.deployment_id,
|
|
downscale.num_to_stop,
|
|
gang_id_by_replica=downscale.gang_id_by_replica,
|
|
replicas_by_gang_id=downscale.replicas_by_gang_id,
|
|
gang_size=downscale.gang_size,
|
|
)
|
|
|
|
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.requested_resources,
|
|
reverse=True,
|
|
)
|
|
|
|
if all_scheduling_requests:
|
|
order_log_key = tuple(r.replica_id for r in all_scheduling_requests)
|
|
if order_log_key != self._last_pack_schedule_order_log_key:
|
|
self._last_pack_schedule_order_log_key = order_log_key
|
|
if Resources.CUSTOM_PRIORITY:
|
|
priority_desc = (
|
|
f"{Resources.CUSTOM_PRIORITY} "
|
|
f"(then GPU, CPU, memory, other custom)"
|
|
)
|
|
else:
|
|
priority_desc = (
|
|
"GPU, CPU, memory, other custom "
|
|
"(RAY_SERVE_HIGH_PRIORITY_CUSTOM_RESOURCES unset)"
|
|
)
|
|
order_desc = ", ".join(
|
|
f"{r.replica_id.deployment_id.name}"
|
|
f"[{_format_resources_for_scheduling_log(r.requested_resources)}]"
|
|
for r in all_scheduling_requests
|
|
)
|
|
logger.info(
|
|
f"Pack scheduling {len(all_scheduling_requests)} pending "
|
|
f"replica(s). Resource priority: {priority_desc}. "
|
|
f"Schedule order (first scheduled first): {order_desc}."
|
|
)
|
|
|
|
# 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.requested_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
|
|
)
|
|
|
|
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
|
|
|
|
for scheduling_request in list(pending_replicas.values()):
|
|
self._schedule_replica(
|
|
scheduling_request=scheduling_request,
|
|
default_scheduling_strategy="SPREAD",
|
|
)
|
|
|
|
def _pack_schedule_replica(
|
|
self,
|
|
scheduling_request: ReplicaSchedulingRequest,
|
|
all_node_labels: Dict[str, Dict[str, str]],
|
|
available_resources_per_node: Dict[str, AvailableNodeResources],
|
|
node_to_assigned_replicas: Dict[str, Set[ReplicaID]],
|
|
) -> Optional[str]:
|
|
"""Attempts to schedule a single request on the best available node.
|
|
|
|
Args:
|
|
scheduling_request: The replica scheduling request.
|
|
all_node_labels: Labels for all active nodes.
|
|
available_resources_per_node: Pre-computed available resources
|
|
per node.
|
|
node_to_assigned_replicas: Mapping of node IDs to replica IDs
|
|
(both running and newly scheduled in this batch).
|
|
|
|
Returns:
|
|
The target node ID if scheduling succeeded, None otherwise.
|
|
"""
|
|
|
|
placement_candidates = self._build_pack_placement_candidates(scheduling_request)
|
|
|
|
target_node = None
|
|
for required_resources, required_labels in placement_candidates:
|
|
target_node = self._find_best_fit_node_for_pack(
|
|
required_resources,
|
|
available_resources_per_node,
|
|
node_to_assigned_replicas,
|
|
required_labels_list=required_labels,
|
|
node_labels=all_node_labels,
|
|
)
|
|
if target_node:
|
|
break
|
|
|
|
replica_id = scheduling_request.replica_id
|
|
|
|
if target_node is None:
|
|
if replica_id not in self._logged_pack_placement_failures:
|
|
self._logged_pack_placement_failures.add(replica_id)
|
|
logger.info(
|
|
f"Pack scheduling could not place {replica_id} "
|
|
f"({_format_resources_for_scheduling_log(scheduling_request.requested_resources)}): "
|
|
f"no node with sufficient resources. "
|
|
f"Falling back to default scheduling."
|
|
)
|
|
|
|
succeeded = self._schedule_replica(
|
|
scheduling_request,
|
|
default_scheduling_strategy="DEFAULT",
|
|
target_node_id=target_node,
|
|
)
|
|
|
|
if succeeded and target_node is not None:
|
|
self._logged_pack_placement_failures.discard(replica_id)
|
|
logger.info(
|
|
f"Pack scheduled {replica_id} "
|
|
f"({_format_resources_for_scheduling_log(scheduling_request.requested_resources)}) "
|
|
f"onto node {target_node}."
|
|
)
|
|
elif not succeeded and target_node is not None:
|
|
if replica_id not in self._logged_pack_placement_failures:
|
|
self._logged_pack_placement_failures.add(replica_id)
|
|
logger.info(
|
|
f"Pack scheduling failed to launch {replica_id} "
|
|
f"({_format_resources_for_scheduling_log(scheduling_request.requested_resources)}) "
|
|
f"on node {target_node}."
|
|
)
|
|
|
|
return target_node if succeeded else None
|
|
|
|
def _build_pack_placement_candidates(
|
|
self, scheduling_request: ReplicaSchedulingRequest
|
|
) -> List[Tuple[RequestedResources, List[Dict[str, str]]]]:
|
|
"""Returns a list of (resources, labels) tuples to attempt for scheduling."""
|
|
|
|
# Collect a list of required resources and labels to try to schedule to
|
|
# support replica compaction when fallback strategies are provided.
|
|
placement_candidates = []
|
|
primary_labels = []
|
|
primary_bundles = scheduling_request.placement_group_bundles
|
|
|
|
if primary_bundles:
|
|
# PG: Use PG bundle_label_selector
|
|
if scheduling_request.placement_group_bundle_label_selector:
|
|
pg_strategy = scheduling_request.placement_group_strategy or None
|
|
if pg_strategy == "STRICT_PACK":
|
|
# All bundle_label_selectors must be satisfied on same node.
|
|
primary_labels = (
|
|
scheduling_request.placement_group_bundle_label_selector
|
|
)
|
|
else:
|
|
# TODO(ryanaoleary@): Support PACK strategy with bundle label selectors.
|
|
raise NotImplementedError(
|
|
"Placement Group strategy 'PACK' with bundle_label_selector "
|
|
"is not yet supported in the Serve scheduler."
|
|
)
|
|
else:
|
|
# Actor: Use Actor label selector
|
|
if "label_selector" in scheduling_request.actor_options:
|
|
primary_labels = [
|
|
scheduling_request.actor_options["label_selector"] or {}
|
|
]
|
|
|
|
# If PG is defined on scheduling request, then `requested_resources` represents the sum across all bundles.
|
|
placement_candidates.append(
|
|
(scheduling_request.requested_resources, primary_labels)
|
|
)
|
|
|
|
if scheduling_request.placement_group_fallback_strategy:
|
|
# TODO(ryanaoleary@): Add support for placement group fallback_strategy when it's added to options.
|
|
raise NotImplementedError(
|
|
"Placement Group fallback strategies are not yet supported in the Serve scheduler."
|
|
)
|
|
|
|
elif scheduling_request.actor_options.get("fallback_strategy"):
|
|
# Fallback strategy provided for Ray Actor.
|
|
for fallback in scheduling_request.actor_options["fallback_strategy"]:
|
|
fallback_labels = [fallback.get("label_selector", {}) or {}]
|
|
placement_candidates.append(
|
|
(scheduling_request.requested_resources, fallback_labels)
|
|
)
|
|
|
|
return placement_candidates
|
|
|
|
def _get_replicas_to_stop(
|
|
self,
|
|
deployment_id: DeploymentID,
|
|
max_num_to_stop: int,
|
|
gang_id_by_replica: Optional[Dict[ReplicaID, str]] = None,
|
|
replicas_by_gang_id: Optional[Dict[str, Set[ReplicaID]]] = None,
|
|
gang_size: Optional[int] = None,
|
|
) -> Set[ReplicaID]:
|
|
"""Select which replicas to stop for a downscale request in the following priority:
|
|
1. Prioritize replicas that are not in the RUNNING state.
|
|
2. Prioritize replicas not on the head node because we can't relinquish the head node.
|
|
3. Prioritize replicas on fallback nodes that don't match the label or bundle label selector.
|
|
4. Prioritize replicas on nodes with fewest total replicas so we can relinquish them.
|
|
5. Prioritize newer replicas over older replicas.
|
|
Note that this algorithm doesn't consider other non-serve actors on the same node.
|
|
See more at https://github.com/ray-project/ray/issues/20599.
|
|
|
|
For gang deployments, the same priority order is applied, but entire
|
|
gangs are selected atomically instead of individual replicas.
|
|
"""
|
|
replicas_priority: List[ReplicaID] = []
|
|
|
|
# Replicas not in running state don't have node id.
|
|
# We will prioritize those first.
|
|
replicas_priority.extend(
|
|
set().union(
|
|
self._pending_replicas[deployment_id].keys(),
|
|
self._launching_replicas[deployment_id].keys(),
|
|
self._recovering_replicas[deployment_id],
|
|
)
|
|
)
|
|
labels_to_check: List[Dict[str, str]] = []
|
|
if label_selector := self._deployments[deployment_id].label_selector:
|
|
labels_to_check.append(label_selector)
|
|
elif bundle_label_selector := self._deployments[
|
|
deployment_id
|
|
].bundle_label_selector:
|
|
labels_to_check.extend(bundle_label_selector)
|
|
node_to_running_replicas_of_all_deployments = (
|
|
self._get_node_to_running_replicas()
|
|
)
|
|
|
|
# _running_replicas preserves insertion order (oldest → newest).
|
|
# Reverse once so we have newest → oldest, then bucket by node.
|
|
ordered_running_replicas = list(self._running_replicas[deployment_id].items())
|
|
ordered_running_replicas.reverse()
|
|
ordered_running_replicas_of_target_deployment: Dict[
|
|
str, List[ReplicaID]
|
|
] = defaultdict(list)
|
|
for replica_id, replica_node_id in ordered_running_replicas:
|
|
ordered_running_replicas_of_target_deployment[replica_node_id].append(
|
|
replica_id
|
|
)
|
|
|
|
# Prioritize based on following priority:
|
|
# 1. Prioritize replicas not on the head node because we can't relinquish the head node.
|
|
# 2. Prioritize replicas on fallback nodes that don't match the label or bundle label selector.
|
|
# 3. Prioritize replicas on nodes with fewer total replicas so we can relinquish them.
|
|
def scale_down_priority(
|
|
node_and_replicas: Tuple[str, Set[ReplicaID]],
|
|
) -> Tuple[int, int, int]:
|
|
node_id, all_replicas = node_and_replicas
|
|
node_labels = self._cluster_node_info_cache.get_node_labels(node_id)
|
|
match_labels = not labels_to_check or any(
|
|
node_labels_match_selector(node_labels, labels)
|
|
for labels in labels_to_check
|
|
)
|
|
is_head_node = node_id == self._head_node_id
|
|
return int(is_head_node), int(match_labels), len(all_replicas)
|
|
|
|
for node_id, _ in sorted(
|
|
node_to_running_replicas_of_all_deployments.items(), key=scale_down_priority
|
|
):
|
|
if node_id not in ordered_running_replicas_of_target_deployment:
|
|
continue
|
|
|
|
# Newest-first list for this node.
|
|
for replica_id in ordered_running_replicas_of_target_deployment[node_id]:
|
|
replicas_priority.append(replica_id)
|
|
|
|
if gang_id_by_replica is not None:
|
|
# Gang scheduling is enabled: select entire gangs to stop atomically
|
|
replicas_to_stop: Set[ReplicaID] = set()
|
|
selected_gangs: Set[str] = set()
|
|
for replica_id in replicas_priority:
|
|
gang_id = gang_id_by_replica.get(replica_id)
|
|
if gang_id is None or gang_id in selected_gangs:
|
|
continue
|
|
if len(replicas_to_stop) + gang_size > max_num_to_stop:
|
|
break
|
|
selected_gangs.add(gang_id)
|
|
replicas_to_stop.update(replicas_by_gang_id[gang_id])
|
|
else:
|
|
# Single-replica scheduling: select individual replicas to stop
|
|
replicas_to_stop: Set[ReplicaID] = set()
|
|
for replica_id in replicas_priority:
|
|
replicas_to_stop.add(replica_id)
|
|
if len(replicas_to_stop) == max_num_to_stop:
|
|
break
|
|
|
|
return replicas_to_stop
|
|
|
|
def _filter_nodes_by_label_selector(
|
|
self,
|
|
available_nodes: Dict[str, AvailableNodeResources],
|
|
required_labels: Dict[str, str],
|
|
node_labels: Dict[str, Dict[str, str]],
|
|
) -> Dict[str, AvailableNodeResources]:
|
|
"""Filters available nodes based on label selector constraints."""
|
|
return {
|
|
node_id: resources
|
|
for node_id, resources in available_nodes.items()
|
|
if node_labels_match_selector(node_labels.get(node_id, {}), required_labels)
|
|
}
|
|
|
|
def _find_best_fit_node_for_pack(
|
|
self,
|
|
required_resources: RequestedResources,
|
|
available_resources_per_node: Dict[str, AvailableNodeResources],
|
|
node_to_assigned_replicas: Dict[str, Set[ReplicaID]],
|
|
required_labels_list: Optional[List[Dict[str, str]]] = None,
|
|
node_labels: Optional[Dict[str, Dict[str, str]]] = None,
|
|
) -> Optional[str]:
|
|
"""Chooses best available node to schedule the required resources.
|
|
|
|
If there are available nodes, returns the node ID of the best
|
|
available node, minimizing fragmentation. Prefers non-idle nodes
|
|
over idle nodes.
|
|
|
|
Args:
|
|
required_resources: Requested resources needed for this replica.
|
|
available_resources_per_node: Available resources per node.
|
|
node_to_assigned_replicas: Mapping of node IDs to replica IDs
|
|
(both running and newly scheduled in this batch).
|
|
required_labels_list: Label selectors to filter nodes.
|
|
node_labels: Labels for each node.
|
|
|
|
Returns:
|
|
The target node ID if scheduling succeeded, None otherwise.
|
|
"""
|
|
|
|
# Filter feasible nodes by provided label selectors if provided.
|
|
if required_labels_list and node_labels:
|
|
for required_labels in required_labels_list:
|
|
available_resources_per_node = self._filter_nodes_by_label_selector(
|
|
available_resources_per_node, required_labels, node_labels
|
|
)
|
|
if not available_resources_per_node:
|
|
return None
|
|
|
|
non_idle_nodes = {
|
|
node_id: res
|
|
for node_id, res in available_resources_per_node.items()
|
|
if len(node_to_assigned_replicas.get(node_id, set())) > 0
|
|
}
|
|
idle_nodes = {
|
|
node_id: res
|
|
for node_id, res in available_resources_per_node.items()
|
|
if len(node_to_assigned_replicas.get(node_id, set())) == 0
|
|
}
|
|
|
|
# 1. Prefer non-idle nodes
|
|
chosen_node = self._best_fit_node(required_resources, non_idle_nodes)
|
|
if chosen_node:
|
|
return chosen_node
|
|
|
|
# 2. Consider idle nodes last
|
|
chosen_node = self._best_fit_node(required_resources, idle_nodes)
|
|
if chosen_node:
|
|
return chosen_node
|
|
|
|
def get_node_to_compact(
|
|
self, allow_new_compaction: bool
|
|
) -> Optional[Tuple[str, float]]:
|
|
return None
|