diff --git a/doc/source/serve/advanced-guides/gang-scheduling.md b/doc/source/serve/advanced-guides/gang-scheduling.md index 985884f85765..22f9d720506f 100644 --- a/doc/source/serve/advanced-guides/gang-scheduling.md +++ b/doc/source/serve/advanced-guides/gang-scheduling.md @@ -28,7 +28,7 @@ Configure gang scheduling by passing a `GangSchedulingConfig` to the `@serve.dep :language: python ``` -This creates 2 gangs of 4 replicas each resulting in a total of 8 replicas. Partial gang isn't allowed, and therefore `num_replicas` must be a multiple of `gang_size`. +This creates 2 gangs of 4 replicas each resulting in a total of 8 replicas. Partial gang isn't allowed, and therefore `num_replicas` must be a multiple of `gang_size`. With autoscaling, the autoscaling bounds (`min_replicas`, `max_replicas`, and `initial_replicas`) must also be multiples of `gang_size`; see [Autoscaling](#autoscaling). ### How resources are reserved within a gang diff --git a/python/ray/serve/_private/config.py b/python/ray/serve/_private/config.py index 9a22315ddebf..93bb8f1fa9b5 100644 --- a/python/ray/serve/_private/config.py +++ b/python/ray/serve/_private/config.py @@ -671,6 +671,46 @@ def handle_num_replicas_auto( return max_ongoing_requests, autoscaling_config +_GANG_ALIGNED_AUTOSCALING_BOUNDS = ( + "min_replicas", + "max_replicas", + "initial_replicas", +) + + +def validate_gang_scheduling_bounds( + gang_size: int, + autoscaling_config: Optional[Union[Dict, AutoscalingConfig]], +) -> None: + """Validate that explicitly set autoscaling bounds are multiples of gang_size. + + Bounds left at their defaults are not checked: gang scheduling rounds the + autoscaler's decisions up to whole gangs, so the defaults are safe. + """ + if autoscaling_config is None: + return + + if isinstance(autoscaling_config, AutoscalingConfig): + bounds = { + name: getattr(autoscaling_config, name) + for name in autoscaling_config.model_fields_set + if name in _GANG_ALIGNED_AUTOSCALING_BOUNDS + } + else: + bounds = { + name: autoscaling_config[name] + for name in _GANG_ALIGNED_AUTOSCALING_BOUNDS + if name in autoscaling_config + } + + for field_name, value in bounds.items(): + if value is not None and value % gang_size != 0: + raise ValueError( + f"autoscaling_config.{field_name} ({value}) must be a multiple of " + f"gang_size ({gang_size})." + ) + + class ReplicaConfig: """Internal datastructure wrapping config options for a deployment's replicas. diff --git a/python/ray/serve/api.py b/python/ray/serve/api.py index 48190a7d0693..2fdc00706190 100644 --- a/python/ray/serve/api.py +++ b/python/ray/serve/api.py @@ -17,6 +17,7 @@ DeploymentConfig, ReplicaConfig, handle_num_replicas_auto, + validate_gang_scheduling_bounds, ) from ray.serve._private.constants import ( RAY_SERVE_FORCE_LOCAL_TESTING_MODE, @@ -747,6 +748,19 @@ class MyDeployment: "gang_scheduling_config is provided. Use " "gang_scheduling_config.gang_placement_strategy instead." ) + # Validate the user-provided bounds before handle_num_replicas_auto() + # merges in defaults, so only explicitly set bounds are checked. + if gang_scheduling_config not in [ + DEFAULT.VALUE, + None, + ] and autoscaling_config not in [DEFAULT.VALUE, None]: + if isinstance(gang_scheduling_config, dict): + gang_size = gang_scheduling_config.get("gang_size") + else: + gang_size = gang_scheduling_config.gang_size + if gang_size is not None: + validate_gang_scheduling_bounds(gang_size, autoscaling_config) + if num_replicas == "auto": num_replicas = None max_ongoing_requests, autoscaling_config = handle_num_replicas_auto( diff --git a/python/ray/serve/deployment.py b/python/ray/serve/deployment.py index ba874a6bd2b4..1df3ce19ede2 100644 --- a/python/ray/serve/deployment.py +++ b/python/ray/serve/deployment.py @@ -6,6 +6,7 @@ ReplicaConfig, RequestRouterConfig, handle_num_replicas_auto, + validate_gang_scheduling_bounds, ) from ray.serve._private.usage import ServeUsageTag from ray.serve._private.utils import DEFAULT, Default @@ -337,6 +338,24 @@ def options( if max_ongoing_requests is None: raise ValueError("`max_ongoing_requests` must be non-null, got None.") + # Validate the user-provided bounds before handle_num_replicas_auto() + # merges in defaults, so only explicitly set bounds are checked. + gang_config_for_validation = ( + gang_scheduling_config + if gang_scheduling_config is not DEFAULT.VALUE + else self._deployment_config.gang_scheduling_config + ) + if gang_config_for_validation is not None and autoscaling_config not in [ + DEFAULT.VALUE, + None, + ]: + if isinstance(gang_config_for_validation, dict): + gang_size = gang_config_for_validation.get("gang_size") + else: + gang_size = gang_config_for_validation.gang_size + if gang_size is not None: + validate_gang_scheduling_bounds(gang_size, autoscaling_config) + if num_replicas == "auto": max_ongoing_requests, autoscaling_config = handle_num_replicas_auto( max_ongoing_requests, autoscaling_config diff --git a/python/ray/serve/tests/unit/test_config.py b/python/ray/serve/tests/unit/test_config.py index 11952599a75c..19b383d4a583 100644 --- a/python/ray/serve/tests/unit/test_config.py +++ b/python/ray/serve/tests/unit/test_config.py @@ -1140,6 +1140,73 @@ def f(): assert f2._deployment_config.autoscaling_config.min_replicas == 4 assert f2._deployment_config.autoscaling_config.max_replicas == 8 + @pytest.mark.parametrize( + "bounds", + [ + {"min_replicas": 3}, + {"max_replicas": 6}, + {"initial_replicas": 3}, + {"min_replicas": 0, "max_replicas": 6}, + ], + ) + def test_gang_scheduling_config_rejects_unaligned_autoscaling_bounds(self, bounds): + with pytest.raises( + ValueError, match="autoscaling_config.*must be a multiple of gang_size" + ): + + @serve.deployment( + num_replicas="auto", + gang_scheduling_config=GangSchedulingConfig(gang_size=4), + autoscaling_config=bounds, + ) + def f(): + return "test" + + def test_gang_scheduling_config_rejects_unaligned_autoscaling_config_instance( + self, + ): + with pytest.raises( + ValueError, match="autoscaling_config.*must be a multiple of gang_size" + ): + + @serve.deployment( + num_replicas="auto", + gang_scheduling_config=GangSchedulingConfig(gang_size=2), + autoscaling_config=AutoscalingConfig(min_replicas=3, max_replicas=8), + ) + def f(): + return "test" + + def test_gang_scheduling_config_allows_default_autoscaling_bounds(self): + """Unset bounds keep their defaults and don't trip the gang check.""" + + @serve.deployment( + num_replicas="auto", + gang_scheduling_config=GangSchedulingConfig(gang_size=4), + autoscaling_config={"target_ongoing_requests": 2.0}, + ) + def f(): + return "test" + + assert f._deployment_config.autoscaling_config.min_replicas == 1 + assert f._deployment_config.autoscaling_config.max_replicas == 100 + + def test_gang_scheduling_config_rejects_unaligned_bounds_via_options(self): + @serve.deployment( + num_replicas=4, + gang_scheduling_config=GangSchedulingConfig(gang_size=4), + ) + def f(): + return "test" + + with pytest.raises( + ValueError, match="autoscaling_config.*must be a multiple of gang_size" + ): + f.options( + num_replicas="auto", + autoscaling_config={"min_replicas": 3, "max_replicas": 8}, + ) + def test_gang_scheduling_config_proto_roundtrip(self): """Test roundtrip serialization of GangSchedulingConfig through protobuf."""