Fix linter

Signed-off-by: jeffreywang <jeffreywang@anyscale.com>

Author: jeffreywang-anyscale

python/ray/serve/_private/config.py

@@ -417,7 +417,9 @@ def from_proto(cls, proto: DeploymentConfigProto):
                 GangPlacementStrategyProto.Name(gang_config["gang_placement_strategy"])
             )
             gang_config["runtime_failure_policy"] = GangRuntimeFailurePolicy(
-                GangRuntimeFailurePolicyProto.Name(gang_config["runtime_failure_policy"])
+                GangRuntimeFailurePolicyProto.Name(
+                    gang_config["runtime_failure_policy"]
+                )
             )
             data["gang_scheduling_config"] = GangSchedulingConfig(**gang_config)
         else:

python/ray/serve/_private/deployment_scheduler.py

@@ -1023,10 +1023,7 @@ def _prepare_gangs_for_deployment(
 
         for gang_index in range(num_gangs_needed):
             # Build bundles - each bundle is for one replica in the gang
-            bundles = [
-                request.replica_resource_dict.copy()
-                for _ in range(gang_size)
-            ]
+            bundles = [request.replica_resource_dict.copy() for _ in range(gang_size)]
 
             pg_name = (
                 f"gang_{deployment_id.app_name}_{deployment_id.name}"

python/ray/serve/_private/deployment_state.py

@@ -37,7 +37,6 @@
     RunningReplicaInfo,
 )
 from ray.serve._private.config import DeploymentConfig
-from ray.serve.config import GangRuntimeFailurePolicy
 from ray.serve._private.constants import (
     DEFAULT_LATENCY_BUCKET_MS,
     MAX_PER_REPLICA_RETRY_COUNT,
@@ -72,6 +71,7 @@
     msgpack_serialize,
 )
 from ray.serve._private.version import DeploymentVersion
+from ray.serve.config import GangRuntimeFailurePolicy
 from ray.serve.generated.serve_pb2 import DeploymentLanguage
 from ray.serve.schema import (
     DeploymentDetails,
@@ -2486,7 +2486,11 @@ def get_num_running_replicas(self, version: DeploymentVersion = None) -> int:
         return self._replicas.count(states=[ReplicaState.RUNNING], version=version)
 
     def get_gang_config(self):
-        return self._target_state.info.deployment_config.gang_scheduling_config if self._target_state is not None else None
+        return (
+            self._target_state.info.deployment_config.gang_scheduling_config
+            if self._target_state is not None
+            else None
+        )
 
     def get_num_replicas_to_add(self) -> int:
         """Calculate the number of replicas to be added to reach the target state."""
@@ -2502,7 +2506,11 @@ def get_num_replicas_to_add(self) -> int:
         return max(0, delta)
 
     def get_replica_resource_dict(self) -> Dict[str, float]:
-        return self._target_state.info.replica_config.resource_dict.copy() if self._target_state is not None else {}
+        return (
+            self._target_state.info.replica_config.resource_dict.copy()
+            if self._target_state is not None
+            else {}
+        )
 
     def get_active_node_ids(self) -> Set[str]:
         """Get the node ids of all running replicas in this deployment.
@@ -2957,7 +2965,9 @@ def _check_and_stop_outdated_version_replicas(self) -> bool:
 
     def scale_deployment_replicas(
         self,
-        gang_placement_groups: Optional[Dict[DeploymentID, GangPreparationResult]] = None,
+        gang_placement_groups: Optional[
+            Dict[DeploymentID, GangPreparationResult]
+        ] = None,
     ) -> Tuple[List[ReplicaSchedulingRequest], DeploymentDownscaleRequest]:
         """Scale the given deployment to the number of replicas.
 
@@ -3020,7 +3030,9 @@ def scale_deployment_replicas(
                             assign_rank_callback=self._rank_manager.assign_rank,
                         )
                         upscale.append(scheduling_request)
-                        self._replicas.add(ReplicaState.STARTING, new_deployment_replica)
+                        self._replicas.add(
+                            ReplicaState.STARTING, new_deployment_replica
+                        )
 
         elif delta_replicas < 0:
             to_remove = -delta_replicas
@@ -3100,9 +3112,7 @@ def _add_replicas_with_gang_scheduling(
                     gang_id=gang_id,
                     rank=bundle_index,
                     world_size=gang_size,
-                    member_replica_ids=[
-                        r.unique_id for r in member_replica_ids
-                    ],
+                    member_replica_ids=[r.unique_id for r in member_replica_ids],
                 )
 
                 new_deployment_replica = DeploymentReplica(
@@ -3425,13 +3435,11 @@ def check_and_update_replicas(self):
                 )
                 self._stop_replica(replica, graceful_stop=False)
                 if replica.version == self._target_state.version:
-                    self._curr_status_info = (
-                        self._curr_status_info.handle_transition(
-                            trigger=DeploymentStatusInternalTrigger.HEALTH_CHECK_FAILED,
-                            message="A replica's health check failed. This "
-                            "deployment will be UNHEALTHY until the replica "
-                            "recovers or a new deploy happens.",
-                        )
+                    self._curr_status_info = self._curr_status_info.handle_transition(
+                        trigger=DeploymentStatusInternalTrigger.HEALTH_CHECK_FAILED,
+                        message="A replica's health check failed. This "
+                        "deployment will be UNHEALTHY until the replica "
+                        "recovers or a new deploy happens.",
                     )
             else:
                 self._replicas.add(replica.actor_details.state, replica)
@@ -4382,9 +4390,7 @@ def _prepare_gang_placement_groups(
             # Skip if deployment has replicas still stopping. Their resources
             # haven't been released yet, so PG creation would likely fail or
             # block waiting for resources. We'll retry next reconciliation loop.
-            if deployment_state._replicas.count(
-                states=[ReplicaState.STOPPING]
-            ) > 0:
+            if deployment_state._replicas.count(states=[ReplicaState.STOPPING]) > 0:
                 continue
 
             gang_requests[deployment_id] = GangPlacementGroupRequest(
@@ -4402,7 +4408,6 @@ def _prepare_gang_placement_groups(
 
         return self._deployment_scheduler.schedule_gang_placement_groups(gang_requests)
 
-
     def record_request_routing_info(self, info: RequestRoutingInfo) -> None:
         """
         Record request routing information for a replica.

python/ray/serve/_private/replica.py

@@ -48,6 +48,7 @@
 from ray.serve._private.common import (
     RUNNING_REQUESTS_KEY,
     DeploymentID,
+    GangContext,
     ReplicaID,
     ReplicaMetricReport,
     ReplicaQueueLengthInfo,

python/ray/serve/config.py

@@ -800,7 +800,10 @@ class GangRuntimeFailurePolicy(str, Enum):
     """Tear down and restart entire gang atomically when any replica fails."""
 
     RESTART_REPLICA = "RESTART_REPLICA"
-    """Tear down and restart individual replica when it fails. Other replicas in the gang will continue running."""
+    """
+    Tear down and restart individual replica when it fails.
+    Other replicas in the gang will continue running.
+    """
 
 
 @PublicAPI(stability="alpha")
@@ -838,7 +841,5 @@ class GangSchedulingConfig(BaseModel):
     @validator("runtime_failure_policy", always=True)
     def _validate_runtime_failure_policy(cls, v):
         if v == GangRuntimeFailurePolicy.RESTART_REPLICA:
-            raise NotImplementedError(
-                "RESTART_REPLICA policy is not yet implemented."
-            )
+            raise NotImplementedError("RESTART_REPLICA policy is not yet implemented.")
         return v

python/ray/serve/tests/test_deployment_scheduler.py

@@ -5,7 +5,6 @@
 
 import ray
 from ray import serve
-from ray.serve.config import GangSchedulingConfig, GangPlacementStrategy
 from ray._common.test_utils import wait_for_condition
 from ray._raylet import GcsClient
 from ray.serve._private import default_impl
@@ -17,6 +16,7 @@
 )
 from ray.serve._private.test_utils import check_apps_running, get_node_id
 from ray.serve._private.utils import get_head_node_id
+from ray.serve.config import GangPlacementStrategy, GangSchedulingConfig
 from ray.tests.conftest import *  # noqa
 
 
@@ -815,7 +815,9 @@ def test_pack_strategy(self, ray_start_cluster):
 
         @serve.deployment
         def PackDeployment():
-            return os.environ.get("RAY_NODE_ID", ray.get_runtime_context().get_node_id())
+            return os.environ.get(
+                "RAY_NODE_ID", ray.get_runtime_context().get_node_id()
+            )
 
         # 1 gang with PACK strategy - all replicas should be on same node
         app = PackDeployment.options(
@@ -860,7 +862,10 @@ def test_gang_scheduling_spread_strategy(self, ray_start_cluster):
         @serve.deployment
         def SpreadDeployment():
             import os
-            return os.environ.get("RAY_NODE_ID", ray.get_runtime_context().get_node_id())
+
+            return os.environ.get(
+                "RAY_NODE_ID", ray.get_runtime_context().get_node_id()
+            )
 
         # 1 gang with SPREAD strategy - replicas should be on different nodes
         app = SpreadDeployment.options(
@@ -891,7 +896,6 @@ def SpreadDeployment():
         serve.delete("gang_spread_app")
         serve.shutdown()
 
-
     def test_gang_context_populated(self, ray_start_cluster):
         """Verifies GangContext is correctly populated in ReplicaContext."""
         cluster = ray_start_cluster
@@ -968,9 +972,8 @@ def __call__(self):
 
         # Across gangs: member_replica_ids should be different
         gang_members_list = list(gangs.values())
-        assert (
-            sorted(gang_members_list[0][0]["member_replica_ids"])
-            != sorted(gang_members_list[1][0]["member_replica_ids"])
+        assert sorted(gang_members_list[0][0]["member_replica_ids"]) != sorted(
+            gang_members_list[1][0]["member_replica_ids"]
         )
 
         serve.delete("gang_context_app")

python/ray/serve/tests/test_healthcheck.py

@@ -361,7 +361,9 @@ def set_should_fail(self):
     # Wait for deployment to become UNHEALTHY.
     def check_unhealthy():
         app_status = serve.status().applications[SERVE_DEFAULT_APP_NAME]
-        assert app_status.deployments["GangPatient"].status == DeploymentStatus.UNHEALTHY
+        assert (
+            app_status.deployments["GangPatient"].status == DeploymentStatus.UNHEALTHY
+        )
         return True
 
     wait_for_condition(check_unhealthy, timeout=30)

python/ray/serve/tests/unit/test_config.py

@@ -765,15 +765,13 @@ def test_gang_scheduling_config_validation(self):
         config = GangSchedulingConfig(gang_size=4)
         assert config.gang_size == 4
 
-
     def test_gang_scheduling_config_defaults(self):
         """Test GangSchedulingConfig default values."""
         config = GangSchedulingConfig(gang_size=4)
 
         assert config.gang_placement_strategy == GangPlacementStrategy.PACK
         assert config.runtime_failure_policy == GangRuntimeFailurePolicy.RESTART_GANG
 
-
     def test_gang_scheduling_config_custom_values(self):
         """Test GangSchedulingConfig with custom values."""
         config = GangSchedulingConfig(
@@ -790,7 +788,6 @@ def test_gang_placement_strategy_options(self):
             config = GangSchedulingConfig(gang_size=4, gang_placement_strategy=strategy)
             assert config.gang_placement_strategy == strategy
 
-
     def test_gang_runtime_failure_policy_options(self):
         """Test all GangRuntimeFailurePolicy options are valid."""
         # RESTART_GANG should work.
@@ -807,14 +804,13 @@ def test_gang_runtime_failure_policy_options(self):
                 runtime_failure_policy=GangRuntimeFailurePolicy.RESTART_REPLICA,
             )
 
-
     def test_gang_scheduling_config_via_decorator_error(self):
         """Test that gang_scheduling_config validation errors are raised."""
-        with pytest.raises(ValueError, match="num_replicas.*must be a multiple of gang_size"):
+        with pytest.raises(
+            ValueError, match="num_replicas.*must be a multiple of gang_size"
+        ):
 
-            @serve.deployment(
-                gang_scheduling_config=GangSchedulingConfig(gang_size=4)
-            )
+            @serve.deployment(gang_scheduling_config=GangSchedulingConfig(gang_size=4))
             def f():
                 return "test"
 
@@ -849,9 +845,9 @@ def test_gang_scheduling_config_proto_roundtrip(self):
 
     def test_gang_scheduling_config_via_decorator(self):
         """Test that gang_scheduling_config can be passed via @serve.deployment decorator."""
+
         @serve.deployment(
-            num_replicas=8,
-            gang_scheduling_config=GangSchedulingConfig(gang_size=4)
+            num_replicas=8, gang_scheduling_config=GangSchedulingConfig(gang_size=4)
         )
         def f():
             return "test"