[Serve] Make sure Ray installs uvloop to be used as event-loop implem…

…entation for asyncio (#39336)

Currently, `HTTPProxy` is configured with `loop="auto"` which will prefer uvloop iff it's present in the runtime env, which from my recent profiling (see flame-graph attached) indicates that it's not.

This change makes it explicit requirement to use "uvloop" in production, w/ an optional fallback to asyncio when debugging (`DEBUG_MODE=1`).

Running Serve micro benchmarks (`microbenchmark.py`) following results were obtained:


Benchmark | asyncio (RPS) | uvloop (RPS) | increase
-- | -- | -- | --
num_client:1/replica:1/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:False | 419.26 | 477.14 | 13.81%
num_client:1/replica:1/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:True | 240.25 | 253.7 | 5.60%
num_client:1/replica:1/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:False | 419.2 | 479.02 | 14.27%
num_client:1/replica:1/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:True | 218.37 | 264.84 | 21.28%
num_client:1/replica:1/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:False | 370.01 | 457.36 | 23.61%
num_client:1/replica:1/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:True | 233.16 | 257.02 | 10.23%
num_client:1/replica:8/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:False | 365.49 | 399.59 | 9.33%
num_client:1/replica:8/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:True | 219.5 | 241.55 | 10.05%
num_client:1/replica:8/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:False | 359.04 | 400.67 | 11.59%
num_client:1/replica:8/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:True | 220.66 | 238.57 | 8.12%
num_client:1/replica:8/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:False | 354.05 | 373.93 | 5.62%
num_client:1/replica:8/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:True | 210.43 | 234.68 | 11.52%
num_client:8/replica:1/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:False | 106.82 | 57.88 | -45.82%
num_client:8/replica:1/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:True | 470.21 | 544.04 | 15.70%
num_client:8/replica:1/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:False | 723.65 | 869.66 | 20.18%
num_client:8/replica:1/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:True | 551.9 | 632.91 | 14.68%
num_client:8/replica:1/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:False | 737.23 | 809.28 | 9.77%
num_client:8/replica:1/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:True | 540.38 | 617.39 | 14.25%
num_client:8/replica:8/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:False | 660.34 | 720.91 | 9.17%
num_client:8/replica:8/batch_size:1/concurrent_queries:1/data_size:small/intermediate_handle:True | 523.58 | 590.37 | 12.76%
num_client:8/replica:8/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:False | 690.91 | 792.34 | 14.68%
num_client:8/replica:8/batch_size:1/concurrent_queries:10000/data_size:small/intermediate_handle:True | 525.47 | 608.2 | 15.74%
num_client:8/replica:8/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:False | 687.46 | 804.69 | 17.05%
num_client:8/replica:8/batch_size:10000/concurrent_queries:10000/data_size:small/intermediate_handle:True | 516.37 | 599.71 | 16.14%

python/ray/_private/async_compat.py

@@ -19,6 +19,14 @@ def get_new_event_loop():
         return asyncio.new_event_loop()
 
 
+def try_install_uvloop():
+    """Installs uvloop as event-loop implementation for asyncio (if available)"""
+    if uvloop:
+        uvloop.install()
+    else:
+        pass
+
+
 def is_async_func(func):
     """Return True if the function is an async or async generator method."""
     return inspect.iscoroutinefunction(func) or inspect.isasyncgenfunction(func)

python/ray/_private/workers/default_worker.py

@@ -9,11 +9,11 @@
 import ray._private.ray_constants as ray_constants
 import ray._private.utils
 import ray.actor
+from ray._private.async_compat import try_install_uvloop
 from ray._private.parameter import RayParams
 from ray._private.ray_logging import configure_log_file, get_worker_log_file_name
 from ray._private.runtime_env.setup_hook import load_and_execute_setup_hook
 
-
 parser = argparse.ArgumentParser(
     description=("Parse addresses for the worker to connect to.")
 )
@@ -194,6 +194,10 @@
     else:
         raise ValueError("Unknown worker type: " + args.worker_type)
 
+    # Try installing uvloop as default event-loop implementation
+    # for asyncio
+    try_install_uvloop()
+
     raylet_ip_address = args.raylet_ip_address
     if raylet_ip_address is None:
         raylet_ip_address = args.node_ip_address

python/ray/serve/_private/http_proxy.py

@@ -8,7 +8,17 @@
 import socket
 import time
 import grpc
-from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Generator,
+    List,
+    Optional,
+    Tuple,
+    Type,
+    Union,
+)
 import uuid
 
 import uvicorn
@@ -112,6 +122,9 @@
     or float(os.environ.get("SERVE_REQUEST_PROCESSING_TIMEOUT_S", 0))
     or None
 )
+# Controls whether Ray Serve is operating in debug-mode switching off some
+# of the performance optimizations to make troubleshooting easier
+RAY_SERVE_DEBUG_MODE = bool(os.environ.get("RAY_SERVE_DEBUG_MODE", 0))
 
 if os.environ.get("SERVE_REQUEST_PROCESSING_TIMEOUT_S") is not None:
     logger.warning(
@@ -1591,13 +1604,14 @@ async def run_http_server(self):
                 "Please make sure your http-host and http-port are specified correctly."
             )
 
-        # Note(simon): we have to use lower level uvicorn Config and Server
+        # NOTE: We have to use lower level uvicorn Config and Server
         # class because we want to run the server as a coroutine. The only
         # alternative is to call uvicorn.run which is blocking.
         config = uvicorn.Config(
             self.wrapped_http_proxy,
             host=self.host,
             port=self.port,
+            loop=_determine_target_loop(),
             root_path=self.root_path,
             lifespan="off",
             access_log=False,
@@ -1715,3 +1729,19 @@ async def _uvicorn_keep_alive(self) -> Optional[int]:
         """
         if self._uvicorn_server:
             return self._uvicorn_server.config.timeout_keep_alive
+
+
+def _determine_target_loop():
+    """We determine target loop based on whether RAY_SERVE_DEBUG_MODE is enabled:
+
+    - RAY_SERVE_DEBUG_MODE=0 (default): we use "uvloop" (Cython) providing
+                              high-performance, native implementation of the event-loop
+
+    - RAY_SERVE_DEBUG_MODE=1: we fall back to "asyncio" (pure Python) event-loop
+                              implementation that is considerably slower than "uvloop",
+                              but provides for easy access to the source implementation
+    """
+    if RAY_SERVE_DEBUG_MODE:
+        return "asyncio"
+    else:
+        return "uvloop"

python/setup.py

@@ -275,7 +275,7 @@ def get_packages(self):
             else "grpcio",
         ],
         "serve": [
-            "uvicorn",
+            "uvicorn[standard]",
             "requests",
             "starlette",
             "fastapi",