[Serve] Set num_cpus to 0 in run_graph() task (ray-project#26177)

python/ray/serve/controller.py

@@ -547,7 +547,7 @@ def get_app_config(self) -> Dict:
             return config
 
 
-@ray.remote(max_calls=1)
+@ray.remote(num_cpus=0, max_calls=1)
 def run_graph(
     import_path: str, graph_env: dict, deployment_override_options: List[Dict]
 ):

python/ray/serve/tests/test_controller.py

@@ -2,6 +2,7 @@
 import time
 
 import ray
+
 from ray import serve
 from ray.serve.common import DeploymentInfo
 from ray.serve.generated.serve_pb2 import DeploymentRoute

python/ray/serve/tests/test_standalone.py

@@ -7,6 +7,7 @@
 import socket
 import subprocess
 import sys
+import time
 from tempfile import mkstemp
 
 import pydantic
@@ -34,6 +35,7 @@
 from ray.serve.generated.serve_pb2 import ActorNameList
 from ray.serve.http_util import set_socket_reuse_port
 from ray.serve.utils import block_until_http_ready, format_actor_name, get_all_node_ids
+from ray.serve.schema import ServeApplicationSchema
 
 # Explicitly importing it here because it is a ray core tests utility (
 # not in the tree)
@@ -785,5 +787,39 @@ def __call__(self, request):
         )
 
 
+@serve.deployment(ray_actor_options={"num_cpus": 0.1})
+class Waiter:
+    def __init__(self):
+        time.sleep(5)
+
+    def __call__(self, *args):
+        return "May I take your order?"
+
+
+WaiterNode = Waiter.bind()
+
+
+def test_run_graph_task_uses_zero_cpus():
+    """Check that the run_graph() task uses zero CPUs."""
+
+    ray.init(num_cpus=2)
+    client = serve.start(detached=True)
+
+    config = {"import_path": "ray.serve.tests.test_standalone.WaiterNode"}
+    config = ServeApplicationSchema.parse_obj(config)
+    client.deploy_app(config)
+
+    with pytest.raises(RuntimeError):
+        wait_for_condition(lambda: ray.available_resources()["CPU"] < 1.9, timeout=5)
+
+    wait_for_condition(
+        lambda: requests.get("http://localhost:8000/Waiter").text
+        == "May I take your order?"
+    )
+
+    serve.shutdown()
+    ray.shutdown()
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", "-s", __file__]))