[Serve] Reuse existing validation functions for Ray Serve config & bu…

…g fix (ray-project#24265)

* set default cpus in ray_actor_options

* remove unnecessary tests

* update message

python/ray/serve/config.py

@@ -29,6 +29,8 @@
  AutoscalingConfig as AutoscalingConfigProto,
  ReplicaConfig as ReplicaConfigProto,
 )
+from ray._private import ray_option_utils
+from ray._private.utils import resources_from_ray_options
 
 
 class AutoscalingConfig(BaseModel):
@@ -262,84 +264,35 @@ def __init__(
  if ray_actor_options is None:
  self.ray_actor_options = {}
  else:
+ if not isinstance(ray_actor_options, dict):
+ raise TypeError("ray_actor_options must be a dictionary.")
+ allowed_ray_actor_options = {
+ # resource options
+ "accelerator_type",
+ "memory",
+ "num_cpus",
+ "num_gpus",
+ "object_store_memory",
+ "resources",
+ # other options
+ "runtime_env",
+ }
+ for option in ray_actor_options:
+ if option not in allowed_ray_actor_options:
+ raise ValueError(
+ f"Specifying '{option}' in ray_actor_options is not allowed. "
+ f"Allowed options: {allowed_ray_actor_options}"
+ )
+ ray_option_utils.validate_actor_options(ray_actor_options, in_options=True)
  self.ray_actor_options = ray_actor_options
 
- self.resource_dict = {}
- self._validate()
-
- def _validate(self):
- if not isinstance(self.ray_actor_options, dict):
- raise TypeError("ray_actor_options must be a dictionary.")
-
- disallowed_ray_actor_options = {
- "args",
- "kwargs",
- "max_concurrency",
- "max_restarts",
- "max_task_retries",
- "name",
- "namespace",
- "lifetime",
- "placement_group",
- "placement_group_bundle_index",
- "placement_group_capture_child_tasks",
- "max_pending_calls",
- "scheduling_strategy",
- }
-
- for option in disallowed_ray_actor_options:
- if option in self.ray_actor_options:
- raise ValueError(
- f"Specifying {option} in ray_actor_options is not allowed."
- )
-
- # TODO(suquark): reuse options validation of remote function/actor.
- # Ray defaults to zero CPUs for placement, we default to one here.
- if self.ray_actor_options.get("num_cpus") is None:
+ # The ray_actor_options dictionary is what ultimately gets passed into
+ # each replica actor's .options() call. The resource_dict is used only
+ # to inform the user about their resource usage.
+ self.ray_actor_options.setdefault("num_cpus", None)
+ if self.ray_actor_options["num_cpus"] is None:
  self.ray_actor_options["num_cpus"] = 1
- num_cpus = self.ray_actor_options["num_cpus"]
- if not isinstance(num_cpus, (int, float)):
- raise TypeError("num_cpus in ray_actor_options must be an int or a float.")
- elif num_cpus < 0:
- raise ValueError("num_cpus in ray_actor_options must be >= 0.")
- self.resource_dict["CPU"] = num_cpus
-
- if self.ray_actor_options.get("num_gpus") is None:
- self.ray_actor_options["num_gpus"] = 0
- num_gpus = self.ray_actor_options["num_gpus"]
- if not isinstance(num_gpus, (int, float)):
- raise TypeError("num_gpus in ray_actor_options must be an int or a float.")
- elif num_gpus < 0:
- raise ValueError("num_gpus in ray_actor_options must be >= 0.")
- self.resource_dict["GPU"] = num_gpus
-
- # Serve deployments use Ray's default for actor memory.
- self.ray_actor_options.setdefault("memory", None)
- memory = self.ray_actor_options["memory"]
- if memory is not None and not isinstance(memory, (int, float)):
- raise TypeError(
- "memory in ray_actor_options must be an int, a float, or None."
- )
- elif memory is not None and memory <= 0:
- raise ValueError("memory in ray_actor_options must be > 0.")
- self.resource_dict["memory"] = memory
-
- object_store_memory = self.ray_actor_options.get("object_store_memory")
- if not isinstance(object_store_memory, (int, float, type(None))):
- raise TypeError(
- "object_store_memory in ray_actor_options must be an int, float "
- "or None."
- )
- elif object_store_memory is not None and object_store_memory < 0:
- raise ValueError("object_store_memory in ray_actor_options must be >= 0.")
- self.resource_dict["object_store_memory"] = object_store_memory
-
- if self.ray_actor_options.get("resources") is None:
- self.ray_actor_options["resources"] = {}
- custom_resources = self.ray_actor_options["resources"]
- if not isinstance(custom_resources, dict):
- raise TypeError("resources in ray_actor_options must be a dictionary.")
- self.resource_dict.update(custom_resources)
+ self.resource_dict = resources_from_ray_options(self.ray_actor_options)
 
  @classmethod
  def from_proto(

python/ray/serve/pipeline/tests/test_generate.py

@@ -111,7 +111,7 @@ def test_single_class_with_invalid_deployment_options(serve_instance):
  deployments = extract_deployments_from_serve_dag(serve_root_dag)
  assert len(deployments) == 1
  with pytest.raises(
- ValueError, match="Specifying name in ray_actor_options is not allowed"
+ ValueError, match="Specifying 'name' in ray_actor_options is not allowed"
  ):
  deployments[0].deploy()
 

python/ray/serve/tests/test_config.py

@@ -162,8 +162,6 @@ def function(_):
  ReplicaConfig(Class, ray_actor_options={"resources": []})
 
  disallowed_ray_actor_options = {
- "args",
- "kwargs",
  "max_concurrency",
  "max_restarts",
  "max_task_retries",
@@ -175,33 +173,15 @@ def function(_):
  "placement_group_capture_child_tasks",
  "max_pending_calls",
  "scheduling_strategy",
+ "get_if_exists",
+ "_metadata",
  }
 
  for option in disallowed_ray_actor_options:
  with pytest.raises(ValueError):
  ReplicaConfig(Class, ray_actor_options={option: None})
 
 
-@pytest.mark.parametrize(
- "memory_omitted_options",
- [
- None,
- {},
- {"num_cpus": 1, "num_gpus": 3},
- {"num_cpus": 1, "num_gpus": 3, "memory": None},
- ],
-)
-def test_replica_config_default_memory_none(memory_omitted_options):
- """Checks that ReplicaConfig's default memory is None."""
-
- if memory_omitted_options is None:
- config = ReplicaConfig("fake.import_path")
- assert config.ray_actor_options["memory"] is None
-
- config = ReplicaConfig("fake.import_path", ray_actor_options=memory_omitted_options)
- assert config.ray_actor_options["memory"] is None
-
-
 def test_http_options():
  HTTPOptions()
  HTTPOptions(host="8.8.8.8", middlewares=[object()])