[Serve] add blocking to serve.run() (#43227)

Rename the existing serve.run to serve._run and added blocking option to the new serve.run

---------

Signed-off-by: Gene Su <[email protected]>

python/ray/serve/__init__.py

@@ -4,6 +4,7 @@
  from ray.serve.api import (
  Application,
  Deployment,
+ _run,
  delete,
  deployment,
  get_app_handle,
@@ -35,6 +36,7 @@
 ray._private.worker.blocking_get_inside_async_warned = True
 
 __all__ = [
+ "_run",
  "batch",
  "start",
  "HTTPOptions",

python/ray/serve/api.py

@@ -1,6 +1,7 @@
 import collections
 import inspect
 import logging
+import time
 from functools import wraps
 from typing import Any, Callable, Dict, List, Optional, Union
 
@@ -436,7 +437,7 @@ def list_deployments() -> Dict[str, Deployment]:
 
 
 @PublicAPI(stability="stable")
-def run(
+def _run(
  target: Application,
  _blocking: bool = True,
  name: str = SERVE_DEFAULT_APP_NAME,
@@ -445,28 +446,9 @@ def run(
 ) -> DeploymentHandle:
  """Run an application and return a handle to its ingress deployment.
 
- The application is returned by `Deployment.bind()`. Example:
-
- .. code-block:: python
-
- handle = serve.run(MyDeployment.bind())
- ray.get(handle.remote())
-
- Args:
- target:
- A Serve application returned by `Deployment.bind()`.
- name: Application name. If not provided, this will be the only
- application running on the cluster (it will delete all others).
- route_prefix: Route prefix for HTTP requests. If not provided, it will use
- route_prefix of the ingress deployment. If specified neither as an argument
- nor in the ingress deployment, the route prefix will default to '/'.
- logging_config: Application logging config. If provided, the config will
- be applied to all deployments which doesn't have logging config.
-
- Returns:
- DeploymentHandle: A handle that can be used to call the application.
+ This is only used internally with the _blocking not totally blocking the following
+ code indefinitely until Ctrl-C'd.
  """
-
  if len(name) == 0:
  raise RayServeException("Application name must a non-empty string.")
 
@@ -530,6 +512,56 @@ def run(
  return handle
 
 
+@PublicAPI(stability="stable")
+def run(
+ target: Application,
+ blocking: bool = False,
+ name: str = SERVE_DEFAULT_APP_NAME,
+ route_prefix: str = DEFAULT.VALUE,
+ logging_config: Optional[Union[Dict, LoggingConfig]] = None,
+) -> DeploymentHandle:
+ """Run an application and return a handle to its ingress deployment.
+
+ The application is returned by `Deployment.bind()`. Example:
+
+ .. code-block:: python
+
+ handle = serve.run(MyDeployment.bind())
+ ray.get(handle.remote())
+
+ Args:
+ target:
+ A Serve application returned by `Deployment.bind()`.
+ blocking: Whether this call should be blocking. If True, it
+ will loop and log status until Ctrl-C'd.
+ name: Application name. If not provided, this will be the only
+ application running on the cluster (it will delete all others).
+ route_prefix: Route prefix for HTTP requests. If not provided, it will use
+ route_prefix of the ingress deployment. If specified neither as an argument
+ nor in the ingress deployment, the route prefix will default to '/'.
+ logging_config: Application logging config. If provided, the config will
+ be applied to all deployments which doesn't have logging config.
+
+ Returns:
+ DeploymentHandle: A handle that can be used to call the application.
+ """
+ handle = _run(
+ target=target,
+ name=name,
+ route_prefix=route_prefix,
+ logging_config=logging_config,
+ )
+
+ if blocking:
+ try:
+ while True:
+ # Block, letting Ray print logs to the terminal.
+ time.sleep(10)
+ except KeyboardInterrupt:
+ logger.info("Got KeyboardInterrupt, release blocking...")
+ return handle
+
+
 @PublicAPI(stability="stable")
 def delete(name: str, _blocking: bool = True):
  """Delete an application by its name.

python/ray/serve/scripts.py

@@ -590,8 +590,14 @@ def run(
  if is_config:
  client.deploy_apps(config, _blocking=False)
  cli_logger.success("Submitted deploy config successfully.")
+ if blocking:
+ while True:
+ # Block, letting Ray print logs to the terminal.
+ time.sleep(10)
  else:
- serve.run(app, name=name, route_prefix=route_prefix)
+ # This should not block if reload is true so the watchfiles can be triggered
+ should_block = blocking and not reload
+ serve.run(app, blocking=should_block, name=name, route_prefix=route_prefix)
  cli_logger.success("Deployed app successfully.")
 
  if reload:
@@ -618,14 +624,11 @@ def run(
  app = _private_api.call_app_builder_with_args_if_necessary(
  import_attr(import_path, reload_module=True), args_dict
  )
- serve.run(app, name=name, route_prefix=route_prefix)
+ serve.run(
+ target=app, blocking=True, name=name, route_prefix=route_prefix
+ )
  cli_logger.success("Redeployed app successfully.")
 
- if blocking:
- while True:
- # Block, letting Ray print logs to the terminal.
- time.sleep(10)
-
  except KeyboardInterrupt:
  cli_logger.info("Got KeyboardInterrupt, shutting down...")
  serve.shutdown()

python/ray/serve/tests/test_api.py

@@ -848,7 +848,7 @@ class A:
  def __init__(self):
  1 / 0
 
- serve.run(A.bind(), _blocking=False)
+ serve._run(A.bind(), _blocking=False)
 
  def check_for_failed_deployment():
  default_app = serve.status().applications[SERVE_DEFAULT_APP_NAME]
@@ -880,7 +880,7 @@ def __init__(self):
  create_engine("mysql:https://some_wrong_url:3306").connect()
 
  ray_actor_options = {"runtime_env": {"pip": ["PyMySQL", "sqlalchemy==1.3.19"]}}
- serve.run(
+ serve._run(
  MyDeployment.options(ray_actor_options=ray_actor_options).bind(),
  _blocking=False,
  )

python/ray/serve/tests/test_autoscaling_policy.py

@@ -1081,7 +1081,7 @@ async def __init__(self):
  app = AutoscalingDeployment.bind()
 
  # Start the AutoscalingDeployment.
- serve.run(app, name=app_name, _blocking=False)
+ serve._run(app, name=app_name, _blocking=False)
 
  # Active replicas are replicas that are waiting or running.
  expected_num_active_replicas: int = min_replicas
@@ -1165,7 +1165,7 @@ def check_expected_statuses(
  max_replicas=max_replicas,
  )
  ).bind()
- serve.run(app, name=app_name, _blocking=False)
+ serve._run(app, name=app_name, _blocking=False)
  expected_num_active_replicas = min_replicas
 
  wait_for_condition(check_num_active_replicas, expected=expected_num_active_replicas)
@@ -1221,7 +1221,7 @@ def check_expected_statuses(
  max_replicas=max_replicas,
  )
  ).bind()
- serve.run(app, name=app_name, _blocking=False)
+ serve._run(app, name=app_name, _blocking=False)
  expected_num_active_replicas = min_replicas
 
  wait_for_condition(check_num_active_replicas, expected=expected_num_active_replicas)

python/ray/serve/tests/test_cluster.py

@@ -156,7 +156,7 @@ class E:
  async def __init__(self):
  await signal.wait.remote()
 
- serve.run(E.bind(), _blocking=False)
+ serve._run(E.bind(), _blocking=False)
 
  def get_replicas(replica_state):
  controller = client._controller

python/ray/serve/tests/test_controller_recovery.py

@@ -191,7 +191,7 @@ def make_nonblocking_calls(expected, expect_blocking=False, num_returns=1):
  # Redeploy new version. Since there is one replica blocking, only one new
  # replica should be started up.
  V2 = V1.options(func_or_class=V2, version="2")
- serve.run(V2.bind(), _blocking=False, name="app")
+ serve._run(V2.bind(), _blocking=False, name="app")
  with pytest.raises(TimeoutError):
  client._wait_for_application_running("app", timeout_s=0.1)
  responses3, blocking3 = make_nonblocking_calls({"1": 1}, expect_blocking=True)
@@ -231,7 +231,7 @@ async def __init__(self):
  def __call__(self, request):
  return f"1|{os.getpid()}"
 
- serve.run(V1.bind(), _blocking=False, name="app")
+ serve._run(V1.bind(), _blocking=False, name="app")
  ray.get(pending_init_indicator.remote())
 
  def get_actor_info(name: str):

python/ray/serve/tests/test_deploy.py

@@ -149,7 +149,7 @@ async def __call__(self, request):
  # Redeploy new version. This should not go through until the old version
  # replica completely stops.
  V2 = V1.options(func_or_class=V2, version="2")
- serve.run(V2.bind(), _blocking=False, name="app")
+ serve._run(V2.bind(), _blocking=False, name="app")
  with pytest.raises(TimeoutError):
  client._wait_for_application_running("app", timeout_s=0.1)
 
@@ -264,7 +264,7 @@ def make_nonblocking_calls(expected, expect_blocking=False):
  # Redeploy new version. Since there is one replica blocking, only one new
  # replica should be started up.
  V2 = V1.options(func_or_class=V2, version="2")
- serve.run(V2.bind(), _blocking=False, name="app")
+ serve._run(V2.bind(), _blocking=False, name="app")
  with pytest.raises(TimeoutError):
  client._wait_for_application_running("app", timeout_s=0.1)
  responses3, blocking3 = make_nonblocking_calls({"1": 1}, expect_blocking=True)
@@ -351,7 +351,9 @@ def make_nonblocking_calls(expected, expect_blocking=False):
 
  # Reconfigure should block one replica until the signal is sent. Check that
  # some requests are now blocking.
- serve.run(V1.options(user_config={"test": "2"}).bind(), name="app", _blocking=False)
+ serve._run(
+ V1.options(user_config={"test": "2"}).bind(), name="app", _blocking=False
+ )
  responses2, blocking2 = make_nonblocking_calls({"1": 1}, expect_blocking=True)
  assert list(responses2["1"])[0] in pids1
 

python/ray/serve/tests/test_deploy_2.py

@@ -256,7 +256,7 @@ class Model:
  def __call__(self):
  return "hello world"
 
- serve.run(Model.bind(), _blocking=False)
+ serve._run(Model.bind(), _blocking=False)
 
  def check_fail():
  app_status = serve.status().applications["default"]

python/ray/serve/tests/test_failure.py

@@ -233,7 +233,7 @@ def make_blocked_request():
  for _ in range(2):
  starting_actor = SignalActor.remote()
  finish_starting_actor = SignalActor.remote()
- serve.run(
+ serve._run(
  SlowStarter.bind(starting_actor, finish_starting_actor), _blocking=False
  )
 

python/ray/serve/tests/test_standalone.py

@@ -670,7 +670,7 @@ def test_updating_status_message(lower_slow_startup_threshold_and_reset):
  def f(*args):
  pass
 
- serve.run(f.bind(), _blocking=False)
+ serve._run(f.bind(), _blocking=False)
 
  def updating_message():
  deployment_status = (
@@ -699,7 +699,7 @@ def __init__(self):
  def __call__(self, request):
  pass
 
- serve.run(f.bind(), _blocking=False)
+ serve._run(f.bind(), _blocking=False)
 
  wait_for_condition(
  lambda: serve.status()

release/long_running_tests/workloads/serve_failure.py

@@ -140,7 +140,7 @@ def handler(self, *args):
 
  if blocking:
  ray.get(self.random_killer.spare.remote(new_name))
- serve.run(
+ serve._run(
  handler.bind(),
  name=new_name,
  route_prefix=f"/{new_name}",
@@ -149,7 +149,7 @@ def handler(self, *args):
  self.applications.append(new_name)
  ray.get(self.random_killer.stop_spare.remote(new_name))
  else:
- serve.run(
+ serve._run(
  handler.bind(),
  name=new_name,
  route_prefix=f"/{new_name}",