Add ray.util package and move libraries from experimental (#7100)

python/ray/serve/BUILD

@@ -0,0 +1,28 @@
+# This is a dummy test dependency that causes the above tests to be
+# re-run if any of these files changes.
+py_library(
+    name = "serve_lib",
+    srcs = glob(["**/*.py"], exclude=["tests/*.py"]),
+)
+
+# This test aggregates all serve tests and run them in a single session
+# similar to `pytest .`
+# Serve tests need to run in a single session because starting and stopping
+# serve cluster take a large chunk of time. All serve tests use a shared
+# cluster.
+py_test(
+    name = "test_serve",
+    size = "medium",
+    srcs = glob(["tests/*.py"]),
+    tags = ["exclusive"],
+    deps = [":serve_lib"],
+)
+
+# Make sure the example showing in doc is tested
+py_test(
+    name = "echo_full",
+    size = "small",
+    srcs = glob(["examples/*.py"]),
+    tags = ["exclusive"],
+    deps = [":serve_lib"]
+)

python/ray/serve/__init__.py

@@ -0,0 +1,11 @@
+from ray.serve.backend_config import BackendConfig
+from ray.serve.policy import RoutePolicy
+from ray.serve.api import (
+    init, create_backend, create_endpoint, link, split, get_handle, stat,
+    set_backend_config, get_backend_config, accept_batch, route)  # noqa: E402
+
+__all__ = [
+    "init", "create_backend", "create_endpoint", "link", "split", "get_handle",
+    "stat", "set_backend_config", "get_backend_config", "BackendConfig",
+    "RoutePolicy", "accept_batch", "route"
+]

python/ray/serve/api.py

@@ -0,0 +1,474 @@
+import inspect
+from functools import wraps
+from tempfile import mkstemp
+
+from multiprocessing import cpu_count
+
+import numpy as np
+
+import ray
+from ray.serve.constants import (DEFAULT_HTTP_HOST, DEFAULT_HTTP_PORT,
+                                 SERVE_NURSERY_NAME)
+from ray.serve.global_state import (GlobalState, start_initial_state)
+from ray.serve.kv_store_service import SQLiteKVStore
+from ray.serve.task_runner import RayServeMixin, TaskRunnerActor
+from ray.serve.utils import (block_until_http_ready, get_random_letters,
+                             expand)
+from ray.serve.exceptions import RayServeException
+from ray.serve.backend_config import BackendConfig
+from ray.serve.policy import RoutePolicy
+from ray.serve.queues import Query
+global_state = None
+
+
+def _get_global_state():
+    """Used for internal purpose. Because just import serve.global_state
+    will always reference the original None object
+    """
+    return global_state
+
+
+def _ensure_connected(f):
+    @wraps(f)
+    def check(*args, **kwargs):
+        if _get_global_state() is None:
+            raise RayServeException("Please run serve.init to initialize or "
+                                    "connect to existing ray serve cluster.")
+        return f(*args, **kwargs)
+
+    return check
+
+
+def accept_batch(f):
+    """Annotation to mark a serving function that batch is accepted.
+
+    This annotation need to be used to mark a function expect all arguments
+    to be passed into a list.
+
+    Example:
+
+    >>> @serve.accept_batch
+        def serving_func(flask_request):
+            assert isinstance(flask_request, list)
+            ...
+
+    >>> class ServingActor:
+            @serve.accept_batch
+            def __call__(self, *, python_arg=None):
+                assert isinstance(python_arg, list)
+    """
+    f.serve_accept_batch = True
+    return f
+
+
+def init(kv_store_connector=None,
+         kv_store_path=None,
+         blocking=False,
+         start_server=True,
+         http_host=DEFAULT_HTTP_HOST,
+         http_port=DEFAULT_HTTP_PORT,
+         ray_init_kwargs={
+             "object_store_memory": int(1e8),
+             "num_cpus": max(cpu_count(), 8)
+         },
+         gc_window_seconds=3600,
+         queueing_policy=RoutePolicy.Random,
+         policy_kwargs={}):
+    """Initialize a serve cluster.
+
+    If serve cluster has already initialized, this function will just return.
+
+    Calling `ray.init` before `serve.init` is optional. When there is not a ray
+    cluster initialized, serve will call `ray.init` with `object_store_memory`
+    requirement.
+
+    Args:
+        kv_store_connector (callable): Function of (namespace) => TableObject.
+            We will use a SQLite connector that stores to /tmp by default.
+        kv_store_path (str, path): Path to the SQLite table.
+        blocking (bool): If true, the function will wait for the HTTP server to
+            be healthy, and other components to be ready before returns.
+        start_server (bool): If true, `serve.init` starts http server.
+            (Default: True)
+        http_host (str): Host for HTTP server. Default to "0.0.0.0".
+        http_port (int): Port for HTTP server. Default to 8000.
+        ray_init_kwargs (dict): Argument passed to ray.init, if there is no ray
+            connection. Default to {"object_store_memory": int(1e8)} for
+            performance stability reason
+        gc_window_seconds(int): How long will we keep the metric data in
+            memory. Data older than the gc_window will be deleted. The default
+            is 3600 seconds, which is 1 hour.
+        queueing_policy(RoutePolicy): Define the queueing policy for selecting
+            the backend for a service. (Default: RoutePolicy.Random)
+        policy_kwargs: Arguments required to instantiate a queueing policy
+    """
+    global global_state
+    # Noop if global_state is no longer None
+    if global_state is not None:
+        return
+
+    # Initialize ray if needed.
+    if not ray.is_initialized():
+        ray.init(**ray_init_kwargs)
+
+    # Try to get serve nursery if there exists
+    try:
+        ray.util.get_actor(SERVE_NURSERY_NAME)
+        global_state = GlobalState()
+        return
+    except ValueError:
+        pass
+
+    # Register serialization context once
+    ray.register_custom_serializer(Query, Query.ray_serialize,
+                                   Query.ray_deserialize)
+
+    if kv_store_path is None:
+        _, kv_store_path = mkstemp()
+
+    # Serve has not been initialized, perform init sequence
+    # Todo, move the db to session_dir
+    #    ray.worker._global_node.address_info["session_dir"]
+    def kv_store_connector(namespace):
+        return SQLiteKVStore(namespace, db_path=kv_store_path)
+
+    nursery = start_initial_state(kv_store_connector)
+
+    global_state = GlobalState(nursery)
+    if start_server:
+        global_state.init_or_get_http_server(host=http_host, port=http_port)
+    global_state.init_or_get_router(
+        queueing_policy=queueing_policy, policy_kwargs=policy_kwargs)
+    global_state.init_or_get_metric_monitor(
+        gc_window_seconds=gc_window_seconds)
+
+    if start_server and blocking:
+        block_until_http_ready("http://{}:{}".format(http_host, http_port))
+
+
+@_ensure_connected
+def create_endpoint(endpoint_name, route=None, blocking=True):
+    """Create a service endpoint given route_expression.
+
+    Args:
+        endpoint_name (str): A name to associate to the endpoint. It will be
+            used as key to set traffic policy.
+        route (str): A string begin with "/". HTTP server will use
+            the string to match the path.
+        blocking (bool): If true, the function will wait for service to be
+            registered before returning
+    """
+    global_state.route_table.register_service(route, endpoint_name)
+
+
+@_ensure_connected
+def set_backend_config(backend_tag, backend_config):
+    """Set a backend configuration for a backend tag
+
+    Args:
+        backend_tag(str): A registered backend.
+        backend_config(BackendConfig) : Desired backend configuration.
+    """
+    assert backend_tag in global_state.backend_table.list_backends(), (
+        "Backend {} is not registered.".format(backend_tag))
+    assert isinstance(backend_config,
+                      BackendConfig), ("backend_config must be"
+                                       " of instance BackendConfig")
+    backend_config_dict = dict(backend_config)
+
+    old_backend_config_dict = global_state.backend_table.get_info(backend_tag)
+    global_state.backend_table.register_info(backend_tag, backend_config_dict)
+
+    # inform the router about change in configuration
+    # particularly for setting max_batch_size
+    ray.get(global_state.init_or_get_router().set_backend_config.remote(
+        backend_tag, backend_config_dict))
+
+    # checking if replicas need to be restarted
+    # Replicas are restarted if there is any change in the backend config
+    # related to restart_configs
+    # TODO(alind) : have replica restarting policies selected by the user
+
+    need_to_restart_replicas = any(
+        old_backend_config_dict[k] != backend_config_dict[k]
+        for k in BackendConfig.restart_on_change_fields)
+    if need_to_restart_replicas:
+        # kill all the replicas for restarting with new configurations
+        scale(backend_tag, 0)
+
+    # scale the replicas with new configuration
+    scale(backend_tag, backend_config_dict["num_replicas"])
+
+
+@_ensure_connected
+def get_backend_config(backend_tag):
+    """get the backend configuration for a backend tag
+
+    Args:
+        backend_tag(str): A registered backend.
+    """
+    assert backend_tag in global_state.backend_table.list_backends(), (
+        "Backend {} is not registered.".format(backend_tag))
+    backend_config_dict = global_state.backend_table.get_info(backend_tag)
+    return BackendConfig(**backend_config_dict)
+
+
+@_ensure_connected
+def create_backend(func_or_class,
+                   backend_tag,
+                   *actor_init_args,
+                   backend_config=BackendConfig()):
+    """Create a backend using func_or_class and assign backend_tag.
+
+    Args:
+        func_or_class (callable, class): a function or a class implements
+            __call__ protocol.
+        backend_tag (str): a unique tag assign to this backend. It will be used
+            to associate services in traffic policy.
+        backend_config (BackendConfig): An object defining backend properties
+        for starting a backend.
+        *actor_init_args (optional): the argument to pass to the class
+            initialization method.
+    """
+    assert isinstance(backend_config,
+                      BackendConfig), ("backend_config must be"
+                                       " of instance BackendConfig")
+    backend_config_dict = dict(backend_config)
+
+    should_accept_batch = (True if backend_config.max_batch_size is not None
+                           else False)
+    batch_annotation_not_found = RayServeException(
+        "max_batch_size is set in config but the function or method does not "
+        "accept batching. Please use @serve.accept_batch to explicitly mark "
+        "the function or method as batchable and takes in list as arguments.")
+
+    arg_list = []
+    if inspect.isfunction(func_or_class):
+        if should_accept_batch and not hasattr(func_or_class,
+                                               "serve_accept_batch"):
+            raise batch_annotation_not_found
+
+        # arg list for a fn is function itself
+        arg_list = [func_or_class]
+        # ignore lint on lambda expression
+        creator = lambda kwrgs: TaskRunnerActor._remote(**kwrgs)  # noqa: E731
+    elif inspect.isclass(func_or_class):
+        if should_accept_batch and not hasattr(func_or_class.__call__,
+                                               "serve_accept_batch"):
+            raise batch_annotation_not_found
+
+        # Python inheritance order is right-to-left. We put RayServeMixin
+        # on the left to make sure its methods are not overriden.
+        @ray.remote
+        class CustomActor(RayServeMixin, func_or_class):
+            pass
+
+        arg_list = actor_init_args
+        # ignore lint on lambda expression
+        creator = lambda kwargs: CustomActor._remote(**kwargs)  # noqa: E731
+    else:
+        raise TypeError(
+            "Backend must be a function or class, it is {}.".format(
+                type(func_or_class)))
+
+    # save creator which starts replicas
+    global_state.backend_table.register_backend(backend_tag, creator)
+
+    # save information about configurations needed to start the replicas
+    global_state.backend_table.register_info(backend_tag, backend_config_dict)
+
+    # save the initial arguments needed by replicas
+    global_state.backend_table.save_init_args(backend_tag, arg_list)
+
+    # set the backend config inside the router
+    # particularly for max-batch-size
+    ray.get(global_state.init_or_get_router().set_backend_config.remote(
+        backend_tag, backend_config_dict))
+    scale(backend_tag, backend_config_dict["num_replicas"])
+
+
+def _start_replica(backend_tag):
+    assert backend_tag in global_state.backend_table.list_backends(), (
+        "Backend {} is not registered.".format(backend_tag))
+
+    replica_tag = "{}#{}".format(backend_tag, get_random_letters(length=6))
+
+    # get the info which starts the replicas
+    creator = global_state.backend_table.get_backend_creator(backend_tag)
+    backend_config_dict = global_state.backend_table.get_info(backend_tag)
+    backend_config = BackendConfig(**backend_config_dict)
+    init_args = global_state.backend_table.get_init_args(backend_tag)
+
+    # get actor creation kwargs
+    actor_kwargs = backend_config.get_actor_creation_args(init_args)
+
+    # Create the runner in the nursery
+    [runner_handle] = ray.get(
+        global_state.actor_nursery_handle.start_actor_with_creator.remote(
+            creator, actor_kwargs, replica_tag))
+
+    # Setup the worker
+    ray.get(
+        runner_handle._ray_serve_setup.remote(
+            backend_tag, global_state.init_or_get_router(), runner_handle))
+    runner_handle._ray_serve_fetch.remote()
+
+    # Register the worker in config tables as well as metric monitor
+    global_state.backend_table.add_replica(backend_tag, replica_tag)
+    global_state.init_or_get_metric_monitor().add_target.remote(runner_handle)
+
+
+def _remove_replica(backend_tag):
+    assert backend_tag in global_state.backend_table.list_backends(), (
+        "Backend {} is not registered.".format(backend_tag))
+    assert len(global_state.backend_table.list_replicas(backend_tag)) > 0, (
+        "Backend {} does not have enough replicas to be removed.".format(
+            backend_tag))
+
+    replica_tag = global_state.backend_table.remove_replica(backend_tag)
+    [replica_handle] = ray.get(
+        global_state.actor_nursery_handle.get_handle.remote(replica_tag))
+
+    # Remove the replica from metric monitor.
+    ray.get(global_state.init_or_get_metric_monitor().remove_target.remote(
+        replica_handle))
+
+    # Remove the replica from actor nursery.
+    ray.get(
+        global_state.actor_nursery_handle.remove_handle.remote(replica_tag))
+
+    # Remove the replica from router.
+    # This will also destory the actor handle.
+    ray.get(global_state.init_or_get_router()
+            .remove_and_destory_replica.remote(backend_tag, replica_handle))
+
+
+@_ensure_connected
+def scale(backend_tag, num_replicas):
+    """Set the number of replicas for backend_tag.
+
+    Args:
+        backend_tag (str): A registered backend.
+        num_replicas (int): Desired number of replicas
+    """
+    assert backend_tag in global_state.backend_table.list_backends(), (
+        "Backend {} is not registered.".format(backend_tag))
+    assert num_replicas >= 0, ("Number of replicas must be"
+                               " greater than or equal to 0.")
+
+    replicas = global_state.backend_table.list_replicas(backend_tag)
+    current_num_replicas = len(replicas)
+    delta_num_replicas = num_replicas - current_num_replicas
+
+    if delta_num_replicas > 0:
+        for _ in range(delta_num_replicas):
+            _start_replica(backend_tag)
+    elif delta_num_replicas < 0:
+        for _ in range(-delta_num_replicas):
+            _remove_replica(backend_tag)
+
+
+@_ensure_connected
+def link(endpoint_name, backend_tag):
+    """Associate a service endpoint with backend tag.
+
+    Example:
+
+    >>> serve.link("service-name", "backend:v1")
+
+    Note:
+    This is equivalent to
+
+    >>> serve.split("service-name", {"backend:v1": 1.0})
+    """
+    split(endpoint_name, {backend_tag: 1.0})
+
+
+@_ensure_connected
+def split(endpoint_name, traffic_policy_dictionary):
+    """Associate a service endpoint with traffic policy.
+
+    Example:
+
+    >>> serve.split("service-name", {
+        "backend:v1": 0.5,
+        "backend:v2": 0.5
+    })
+
+    Args:
+        endpoint_name (str): A registered service endpoint.
+        traffic_policy_dictionary (dict): a dictionary maps backend names
+            to their traffic weights. The weights must sum to 1.
+    """
+    assert endpoint_name in expand(
+        global_state.route_table.list_service(include_headless=True).values())
+
+    assert isinstance(traffic_policy_dictionary,
+                      dict), "Traffic policy must be dictionary"
+    prob = 0
+    for backend, weight in traffic_policy_dictionary.items():
+        prob += weight
+        assert (backend in global_state.backend_table.list_backends()
+                ), "backend {} is not registered".format(backend)
+    assert np.isclose(
+        prob, 1,
+        atol=0.02), "weights must sum to 1, currently it sums to {}".format(
+            prob)
+
+    global_state.policy_table.register_traffic_policy(
+        endpoint_name, traffic_policy_dictionary)
+    ray.get(global_state.init_or_get_router().set_traffic.remote(
+        endpoint_name, traffic_policy_dictionary))
+
+
+@_ensure_connected
+def get_handle(endpoint_name, relative_slo_ms=None, absolute_slo_ms=None):
+    """Retrieve RayServeHandle for service endpoint to invoke it from Python.
+
+    Args:
+        endpoint_name (str): A registered service endpoint.
+        relative_slo_ms(float): Specify relative deadline in milliseconds for
+            queries fired using this handle. (Default: None)
+        absolute_slo_ms(float): Specify absolute deadline in milliseconds for
+            queries fired using this handle. (Default: None)
+
+    Returns:
+        RayServeHandle
+    """
+    assert endpoint_name in expand(
+        global_state.route_table.list_service(include_headless=True).values())
+
+    # Delay import due to it's dependency on global_state
+    from ray.serve.handle import RayServeHandle
+
+    return RayServeHandle(global_state.init_or_get_router(), endpoint_name,
+                          relative_slo_ms, absolute_slo_ms)
+
+
+@_ensure_connected
+def stat(percentiles=[50, 90, 95],
+         agg_windows_seconds=[10, 60, 300, 600, 3600]):
+    """Retrieve metric statistics about ray serve system.
+
+    Args:
+        percentiles(List[int]): The percentiles for aggregation operations.
+            Default is 50th, 90th, 95th percentile.
+        agg_windows_seconds(List[int]): The aggregation windows in seconds.
+            The longest aggregation window must be shorter or equal to the
+            gc_window_seconds.
+    """
+    return ray.get(global_state.init_or_get_metric_monitor().collect.remote(
+        percentiles, agg_windows_seconds))
+
+
+class route:
+    def __init__(self, url_route):
+        self.route = url_route
+
+    def __call__(self, func_or_class):
+        name = func_or_class.__name__
+        backend_tag = "{}:v0".format(name)
+
+        create_backend(func_or_class, backend_tag)
+        create_endpoint(name, self.route)
+        link(name, backend_tag)

python/ray/serve/backend_config.py

@@ -0,0 +1,58 @@
+from copy import deepcopy
+
+
+class BackendConfig:
+    # configs not needed for actor creation when
+    # instantiating a replica
+    _serve_configs = ["_num_replicas", "max_batch_size"]
+
+    # configs which when changed leads to restarting
+    # the existing replicas.
+    restart_on_change_fields = ["resources", "num_cpus", "num_gpus"]
+
+    def __init__(self,
+                 num_replicas=1,
+                 resources=None,
+                 max_batch_size=None,
+                 num_cpus=None,
+                 num_gpus=None,
+                 memory=None,
+                 object_store_memory=None):
+        """
+        Class for defining backend configuration.
+        """
+
+        # serve configs
+        self.num_replicas = num_replicas
+        self.max_batch_size = max_batch_size
+
+        # ray actor configs
+        self.resources = resources
+        self.num_cpus = num_cpus
+        self.num_gpus = num_gpus
+        self.memory = memory
+        self.object_store_memory = object_store_memory
+
+    @property
+    def num_replicas(self):
+        return self._num_replicas
+
+    @num_replicas.setter
+    def num_replicas(self, val):
+        if not (val > 0):
+            raise Exception("num_replicas must be greater than zero")
+        self._num_replicas = val
+
+    def __iter__(self):
+        for k in self.__dict__.keys():
+            key, val = k, self.__dict__[k]
+            if key == "_num_replicas":
+                key = "num_replicas"
+            yield key, val
+
+    def get_actor_creation_args(self, init_args):
+        ret_d = deepcopy(self.__dict__)
+        for k in self._serve_configs:
+            ret_d.pop(k)
+        ret_d["args"] = init_args
+        return ret_d

python/ray/serve/constants.py

@@ -0,0 +1,26 @@
+#: The interval which http server refreshes its routing table
+HTTP_ROUTER_CHECKER_INTERVAL_S = 2
+
+#: Actor name used to register actor nursery
+SERVE_NURSERY_NAME = "SERVE_ACTOR_NURSERY"
+
+#: KVStore connector key in bootstrap config
+BOOTSTRAP_KV_STORE_CONN_KEY = "kv_store_connector"
+
+#: HTTP Address
+DEFAULT_HTTP_ADDRESS = "http://127.0.0.1:8000"
+
+#: HTTP Host
+DEFAULT_HTTP_HOST = "127.0.0.1"
+
+#: HTTP Port
+DEFAULT_HTTP_PORT = 8000
+
+#: Max concurrency
+ASYNC_CONCURRENCY = int(1e6)
+
+#: Default latency SLO
+DEFAULT_LATENCY_SLO_MS = 1e9
+
+#: Key for storing no http route services
+NO_ROUTE_KEY = "NO_ROUTE"

python/ray/serve/context.py

@@ -0,0 +1,34 @@
+from enum import IntEnum
+
+from ray.serve.exceptions import RayServeException
+
+
+class TaskContext(IntEnum):
+    """TaskContext constants for queue.enqueue method"""
+    Web = 1
+    Python = 2
+
+
+# Global variable will be modified in worker
+# web == True: currrently processing a request from web server
+# web == False: currently processing a request from python
+web = False
+
+# batching information in serve context
+# batch_size == None : the backend doesn't support batching
+# batch_size(int)    : the number of elements of input list
+batch_size = None
+
+_not_in_web_context_error = """
+Accessing the request object outside of the web context. Please use
+"serve.context.web" to determine when the function is called within
+a web context.
+"""
+
+
+class FakeFlaskRequest:
+    def __getattribute__(self, name):
+        raise RayServeException(_not_in_web_context_error)
+
+    def __setattr__(self, name, value):
+        raise RayServeException(_not_in_web_context_error)

python/ray/serve/examples/benchmark.py

@@ -0,0 +1,33 @@
+from ray import serve
+from ray.serve.constants import DEFAULT_HTTP_ADDRESS
+import requests
+import time
+import pandas as pd
+from tqdm import tqdm
+
+serve.init(blocking=True)
+
+
+@serve.route("/noop")
+def noop(_):
+    return ""
+
+
+url = "{}/noop".format(DEFAULT_HTTP_ADDRESS)
+while requests.get(url).status_code == 404:
+    time.sleep(1)
+    print("Waiting for noop route to showup.")
+
+latency = []
+for _ in tqdm(range(5200)):
+    start = time.perf_counter()
+    resp = requests.get(url)
+    end = time.perf_counter()
+    latency.append(end - start)
+
+# Remove initial samples
+latency = latency[200:]
+
+series = pd.Series(latency) * 1000
+print("Latency for single noop backend (ms)")
+print(series.describe(percentiles=[0.5, 0.9, 0.95, 0.99]))

python/ray/serve/examples/echo.py

@@ -0,0 +1,28 @@
+"""
+Example service that prints out http context.
+"""
+
+import time
+
+import requests
+
+from ray import serve
+from ray.serve.utils import pformat_color_json
+
+
+def echo(flask_request):
+    return "hello " + flask_request.args.get("name", "serve!")
+
+
+serve.init(blocking=True)
+
+serve.create_endpoint("my_endpoint", "/echo", blocking=True)
+serve.create_backend(echo, "echo:v1")
+serve.link("my_endpoint", "echo:v1")
+
+while True:
+    resp = requests.get("http://127.0.0.1:8000/echo").json()
+    print(pformat_color_json(resp))
+
+    print("...Sleeping for 2 seconds...")
+    time.sleep(2)

python/ray/serve/examples/echo_actor.py

@@ -0,0 +1,46 @@
+"""
+Example actor that adds an increment to a number. This number can
+come from either web (parsing Flask request) or python call.
+
+This actor can be called from HTTP as well as from Python.
+"""
+
+import time
+
+import requests
+
+import ray
+from ray import serve
+from ray.serve.utils import pformat_color_json
+
+
+class MagicCounter:
+    def __init__(self, increment):
+        self.increment = increment
+
+    def __call__(self, flask_request, base_number=None):
+        if serve.context.web:
+            base_number = int(flask_request.args.get("base_number", "0"))
+        return base_number + self.increment
+
+
+serve.init(blocking=True)
+serve.create_endpoint("magic_counter", "/counter", blocking=True)
+serve.create_backend(MagicCounter, "counter:v1", 42)  # increment=42
+serve.link("magic_counter", "counter:v1")
+
+print("Sending ten queries via HTTP")
+for i in range(10):
+    url = "http://127.0.0.1:8000/counter?base_number={}".format(i)
+    print("> Pinging {}".format(url))
+    resp = requests.get(url).json()
+    print(pformat_color_json(resp))
+
+    time.sleep(0.2)
+
+print("Sending ten queries via Python")
+handle = serve.get_handle("magic_counter")
+for i in range(10):
+    print("> Pinging handle.remote(base_number={})".format(i))
+    result = ray.get(handle.remote(base_number=i))
+    print("< Result {}".format(result))

python/ray/serve/examples/echo_actor_batch.py

@@ -0,0 +1,60 @@
+"""
+Example actor that adds an increment to a number. This number can
+come from either web (parsing Flask request) or python call.
+The queries incoming to this actor are batched.
+This actor can be called from HTTP as well as from Python.
+"""
+
+import time
+
+import requests
+
+import ray
+from ray import serve
+from ray.serve.utils import pformat_color_json
+from ray.serve import BackendConfig
+
+
+class MagicCounter:
+    def __init__(self, increment):
+        self.increment = increment
+
+    @serve.accept_batch
+    def __call__(self, flask_request_list, base_number=None):
+        # batch_size = serve.context.batch_size
+        if serve.context.web:
+            result = []
+            for flask_request in flask_request_list:
+                base_number = int(flask_request.args.get("base_number", "0"))
+                result.append(base_number)
+            return list(map(lambda x: x + self.increment, result))
+        else:
+            result = []
+            for b in base_number:
+                ans = b + self.increment
+                result.append(ans)
+            return result
+
+
+serve.init(blocking=True)
+serve.create_endpoint("magic_counter", "/counter", blocking=True)
+b_config = BackendConfig(max_batch_size=5)
+serve.create_backend(
+    MagicCounter, "counter:v1", 42, backend_config=b_config)  # increment=42
+serve.link("magic_counter", "counter:v1")
+
+print("Sending ten queries via HTTP")
+for i in range(10):
+    url = "http://127.0.0.1:8000/counter?base_number={}".format(i)
+    print("> Pinging {}".format(url))
+    resp = requests.get(url).json()
+    print(pformat_color_json(resp))
+
+    time.sleep(0.2)
+
+print("Sending ten queries via Python")
+handle = serve.get_handle("magic_counter")
+for i in range(10):
+    print("> Pinging handle.remote(base_number={})".format(i))
+    result = ray.get(handle.remote(base_number=i))
+    print("< Result {}".format(result))

python/ray/serve/examples/echo_batching.py

@@ -0,0 +1,56 @@
+"""
+This example has backend which has batching functionality enabled.
+"""
+
+import ray
+from ray import serve
+from ray.serve import BackendConfig
+
+
+class MagicCounter:
+    def __init__(self, increment):
+        self.increment = increment
+
+    @serve.accept_batch
+    def __call__(self, flask_request, base_number=None):
+        # __call__ fn should preserve the batch size
+        # base_number is a python list
+
+        if serve.context.batch_size is not None:
+            batch_size = serve.context.batch_size
+            result = []
+            for base_num in base_number:
+                ret_str = "Number: {} Batch size: {}".format(
+                    base_num, batch_size)
+                result.append(ret_str)
+            return result
+        return ""
+
+
+serve.init(blocking=True)
+serve.create_endpoint("magic_counter", "/counter", blocking=True)
+# specify max_batch_size in BackendConfig
+b_config = BackendConfig(max_batch_size=5)
+serve.create_backend(
+    MagicCounter, "counter:v1", 42, backend_config=b_config)  # increment=42
+print("Backend Config for backend: 'counter:v1'")
+print(b_config)
+serve.link("magic_counter", "counter:v1")
+
+handle = serve.get_handle("magic_counter")
+future_list = []
+
+# fire 30 requests
+for r in range(30):
+    print("> [REMOTE] Pinging handle.remote(base_number={})".format(r))
+    f = handle.remote(base_number=r)
+    future_list.append(f)
+
+# get results of queries as they complete
+left_futures = future_list
+while left_futures:
+    completed_futures, remaining_futures = ray.wait(left_futures, timeout=0.05)
+    if len(completed_futures) > 0:
+        result = ray.get(completed_futures[0])
+        print("< " + result)
+    left_futures = remaining_futures

python/ray/serve/examples/echo_error.py

@@ -0,0 +1,44 @@
+"""
+Example of error handling mechanism in ray serve.
+
+We are going to define a buggy function that raise some exception:
+>>> def echo(_):
+        raise Exception("oh no")
+
+The expected behavior is:
+- HTTP server should respond with "internal error" in the response JSON
+- ray.get(handle.remote()) should raise RayTaskError with traceback.
+
+This shows that error is hidden from HTTP side but always visible when calling
+from Python.
+"""
+
+import time
+
+import requests
+
+import ray
+from ray import serve
+from ray.serve.utils import pformat_color_json
+
+
+def echo(_):
+    raise Exception("Something went wrong...")
+
+
+serve.init(blocking=True)
+
+serve.create_endpoint("my_endpoint", "/echo", blocking=True)
+serve.create_backend(echo, "echo:v1")
+serve.link("my_endpoint", "echo:v1")
+
+for _ in range(2):
+    resp = requests.get("http://127.0.0.1:8000/echo").json()
+    print(pformat_color_json(resp))
+
+    print("...Sleeping for 2 seconds...")
+    time.sleep(2)
+
+handle = serve.get_handle("my_endpoint")
+print("Invoke from python will raise exception with traceback:")
+ray.get(handle.remote())

python/ray/serve/examples/echo_fixed_packing.py

@@ -0,0 +1,48 @@
+"""
+Example showing fixed packing policy. The outputs from
+v1 and v2 will be coming according to packing_num specified!
+This is a packed round robin example. First batch of packing_num
+(five in this example) queries would go to 'echo:v1' backend and
+then next batch of packing_num queries would go to 'echo:v2'
+backend.
+"""
+import time
+
+import requests
+
+from ray import serve
+from ray.serve.utils import pformat_color_json
+
+
+def echo_v1(_):
+    return "v1"
+
+
+def echo_v2(_):
+    return "v2"
+
+
+# specify the router policy as FixedPacking with packing num as 5
+serve.init(
+    blocking=True,
+    queueing_policy=serve.RoutePolicy.FixedPacking,
+    policy_kwargs={"packing_num": 5})
+
+# create a service
+serve.create_endpoint("my_endpoint", "/echo", blocking=True)
+
+# create first backend
+serve.create_backend(echo_v1, "echo:v1")
+
+# create second backend
+serve.create_backend(echo_v2, "echo:v2")
+
+# link and split the service to two backends
+serve.split("my_endpoint", {"echo:v1": 0.5, "echo:v2": 0.5})
+
+while True:
+    resp = requests.get("http://127.0.0.1:8000/echo").json()
+    print(pformat_color_json(resp))
+
+    print("...Sleeping for 2 seconds...")
+    time.sleep(2)

python/ray/serve/examples/echo_full.py

@@ -0,0 +1,69 @@
+"""
+Full example of ray.serve module
+"""
+
+import time
+
+import requests
+
+import ray
+import ray.serve as serve
+from ray.serve.utils import pformat_color_json
+
+# initialize ray serve system.
+# blocking=True will wait for HTTP server to be ready to serve request.
+serve.init(blocking=True)
+
+# an endpoint is associated with an http URL.
+serve.create_endpoint("my_endpoint", "/echo")
+
+
+# a backend can be a function or class.
+# it can be made to be invoked from web as well as python.
+def echo_v1(flask_request, response="hello from python!"):
+    if serve.context.web:
+        response = flask_request.url
+    return response
+
+
+serve.create_backend(echo_v1, "echo:v1")
+backend_config_v1 = serve.get_backend_config("echo:v1")
+
+# We can link an endpoint to a backend, the means all the traffic
+# goes to my_endpoint will now goes to echo:v1 backend.
+serve.link("my_endpoint", "echo:v1")
+
+print(requests.get("http://127.0.0.1:8000/echo", timeout=0.5).json())
+# The service will be reachable from http
+
+print(ray.get(serve.get_handle("my_endpoint").remote(response="hello")))
+
+# as well as within the ray system.
+
+
+# We can also add a new backend and split the traffic.
+def echo_v2(flask_request):
+    # magic, only from web.
+    return "something new"
+
+
+serve.create_backend(echo_v2, "echo:v2")
+backend_config_v2 = serve.get_backend_config("echo:v2")
+
+# The two backend will now split the traffic 50%-50%.
+serve.split("my_endpoint", {"echo:v1": 0.5, "echo:v2": 0.5})
+
+# Observe requests are now split between two backends.
+for _ in range(10):
+    print(requests.get("http://127.0.0.1:8000/echo").json())
+    time.sleep(0.5)
+
+# You can also change number of replicas
+# for each backend independently.
+backend_config_v1.num_replicas = 2
+serve.set_backend_config("echo:v1", backend_config_v1)
+backend_config_v2.num_replicas = 2
+serve.set_backend_config("echo:v2", backend_config_v2)
+
+# As well as retrieving relevant system metrics
+print(pformat_color_json(serve.stat()))

python/ray/serve/examples/echo_pipeline.py

@@ -0,0 +1,70 @@
+"""
+Ray serve pipeline example
+"""
+import ray
+import ray.serve as serve
+import time
+
+# initialize ray serve system.
+# blocking=True will wait for HTTP server to be ready to serve request.
+serve.init(blocking=True)
+
+
+# a backend can be a function or class.
+# it can be made to be invoked from web as well as python.
+@serve.route("/echo_v1")
+def echo_v1(_, response="hello from python!"):
+    return f"echo_v1({response})"
+
+
+@serve.route("/echo_v2")
+def echo_v2(_, relay=""):
+    return f"echo_v2({relay})"
+
+
+@serve.route("/echo_v3")
+def echo_v3(_, relay=""):
+    return f"echo_v3({relay})"
+
+
+@serve.route("/echo_v4")
+def echo_v4(_, relay1="", relay2=""):
+    return f"echo_v4({relay1} , {relay2})"
+
+
+"""
+The pipeline created is as follows -
+            "my_endpoint1"
+                  /\
+                 /  \
+                /    \
+               /      \
+              /        \
+             /          \
+  "my_endpoint2"     "my_endpoint3"
+            \            /
+             \          /
+              \        /
+               \      /
+                \    /
+                 \  /
+                  \/
+            "my_endpoint4"
+"""
+
+# get the handle of the endpoints
+handle1 = serve.get_handle("echo_v1")
+handle2 = serve.get_handle("echo_v2")
+handle3 = serve.get_handle("echo_v3")
+handle4 = serve.get_handle("echo_v4")
+
+start = time.time()
+print("Start firing to the pipeline: {} s".format(time.time()))
+handle1_oid = handle1.remote(response="hello")
+handle4_oid = handle4.remote(
+    relay1=handle2.remote(relay=handle1_oid),
+    relay2=handle3.remote(relay=handle1_oid))
+print("Firing ended now waiting for the result,"
+      "time taken: {} s".format(time.time() - start))
+result = ray.get(handle4_oid)
+print("Result: {}, time taken: {} s".format(result, time.time() - start))

python/ray/serve/examples/echo_round_robin.py

@@ -0,0 +1,41 @@
+"""
+Example showing round robin policy. The outputs from
+v1 and v2 will be (almost) interleaved as queries get processed.
+"""
+import time
+
+import requests
+
+from ray import serve
+from ray.serve.utils import pformat_color_json
+
+
+def echo_v1(_):
+    return "v1"
+
+
+def echo_v2(_):
+    return "v2"
+
+
+# specify the router policy as RoundRobin
+serve.init(blocking=True, queueing_policy=serve.RoutePolicy.RoundRobin)
+
+# create a service
+serve.create_endpoint("my_endpoint", "/echo", blocking=True)
+
+# create first backend
+serve.create_backend(echo_v1, "echo:v1")
+
+# create second backend
+serve.create_backend(echo_v2, "echo:v2")
+
+# link and split the service to two backends
+serve.split("my_endpoint", {"echo:v1": 0.5, "echo:v2": 0.5})
+
+while True:
+    resp = requests.get("http://127.0.0.1:8000/echo").json()
+    print(pformat_color_json(resp))
+
+    print("...Sleeping for 2 seconds...")
+    time.sleep(2)

python/ray/serve/examples/echo_slo_reverse.py

@@ -0,0 +1,76 @@
+"""
+SLO [reverse] example of ray.serve module
+"""
+
+import time
+
+import requests
+
+import ray
+import ray.serve as serve
+
+# initialize ray serve system.
+# blocking=True will wait for HTTP server to be ready to serve request.
+serve.init(blocking=True)
+
+# an endpoint is associated with an http URL.
+serve.create_endpoint("my_endpoint", "/echo")
+
+
+# a backend can be a function or class.
+# it can be made to be invoked from web as well as python.
+def echo_v1(flask_request, response="hello from python!"):
+    if serve.context.web:
+        response = flask_request.url
+    return response
+
+
+serve.create_backend(echo_v1, "echo:v1")
+serve.link("my_endpoint", "echo:v1")
+
+# wait for routing table to get populated
+time.sleep(2)
+
+# relative slo (10 ms deadline) can be specified via http
+slo_ms = 10.0
+# absolute slo (10 ms deadline) can be specified via http
+abs_slo_ms = 11.9
+print("> [HTTP] Pinging http://127.0.0.1:8000/"
+      "echo?relative_slo_ms={}".format(slo_ms))
+print(
+    requests.get("http://127.0.0.1:8000/"
+                 "echo?relative_slo_ms={}".format(slo_ms)).json())
+print("> [HTTP] Pinging http://127.0.0.1:8000/"
+      "echo?absolute_slo_ms={}".format(abs_slo_ms))
+print(
+    requests.get("http://127.0.0.1:8000/"
+                 "echo?absolute_slo_ms={}".format(abs_slo_ms)).json())
+
+# get the handle of the endpoint
+handle = serve.get_handle("my_endpoint")
+
+future_list = []
+
+# fire 10 requests with slo's in the (almost) reverse order of the order in
+# which remote procedure call is done
+for r in range(10):
+    slo_ms = 1000 - 100 * r
+    response = "hello from request: {} slo: {}".format(r, slo_ms)
+    print("> [REMOTE] Pinging handle.remote(response='{}',slo_ms={})".format(
+        response, slo_ms))
+
+    # overriding slo for each query.
+    # Generally slo is specified for a service handle but it can
+    # be overrided using options for query specific demands
+    f = handle.options(relative_slo_ms=slo_ms).remote(response=response)
+    future_list.append(f)
+
+# get results of queries as they complete
+# should be completed (almost) according to the order of their slo time
+left_futures = future_list
+while left_futures:
+    completed_futures, remaining_futures = ray.wait(left_futures, timeout=0.05)
+    if len(completed_futures) > 0:
+        result = ray.get(completed_futures[0])
+        print(result)
+    left_futures = remaining_futures

python/ray/serve/examples/echo_split.py

@@ -0,0 +1,41 @@
+"""
+Example of traffic splitting. We will first use echo:v1. Then v1 and v2
+will split the incoming traffic evenly.
+"""
+import time
+
+import requests
+
+from ray import serve
+from ray.serve.utils import pformat_color_json
+
+
+def echo_v1(_):
+    return "v1"
+
+
+def echo_v2(_):
+    return "v2"
+
+
+serve.init(blocking=True)
+
+serve.create_endpoint("my_endpoint", "/echo", blocking=True)
+serve.create_backend(echo_v1, "echo:v1")
+serve.link("my_endpoint", "echo:v1")
+
+for _ in range(3):
+    resp = requests.get("http://127.0.0.1:8000/echo").json()
+    print(pformat_color_json(resp))
+
+    print("...Sleeping for 2 seconds...")
+    time.sleep(2)
+
+serve.create_backend(echo_v2, "echo:v2")
+serve.split("my_endpoint", {"echo:v1": 0.5, "echo:v2": 0.5})
+while True:
+    resp = requests.get("http://127.0.0.1:8000/echo").json()
+    print(pformat_color_json(resp))
+
+    print("...Sleeping for 2 seconds...")
+    time.sleep(2)

python/ray/serve/exceptions.py

@@ -0,0 +1,2 @@
+class RayServeException(Exception):
+    pass

python/ray/serve/global_state.py

@@ -0,0 +1,170 @@
+import ray
+from ray.serve.constants import (BOOTSTRAP_KV_STORE_CONN_KEY,
+                                 DEFAULT_HTTP_HOST, DEFAULT_HTTP_PORT,
+                                 SERVE_NURSERY_NAME, ASYNC_CONCURRENCY)
+from ray.serve.kv_store_service import (BackendTable, RoutingTable,
+                                        TrafficPolicyTable)
+from ray.serve.metric import (MetricMonitor, start_metric_monitor_loop)
+
+from ray.serve.policy import RoutePolicy
+from ray.serve.server import HTTPActor
+
+
+def start_initial_state(kv_store_connector):
+    nursery_handle = ActorNursery.remote()
+    ray.util.register_actor(SERVE_NURSERY_NAME, nursery_handle)
+
+    ray.get(
+        nursery_handle.store_bootstrap_state.remote(
+            BOOTSTRAP_KV_STORE_CONN_KEY, kv_store_connector))
+    return nursery_handle
+
+
+@ray.remote
+class ActorNursery:
+    """Initialize and store all actor handles.
+
+    Note:
+        This actor is necessary because ray will destory actors when the
+        original actor handle goes out of scope (when driver exit). Therefore
+        we need to initialize and store actor handles in a seperate actor.
+    """
+
+    def __init__(self):
+        self.tag_to_actor_handles = dict()
+
+        self.bootstrap_state = dict()
+
+    def start_actor(self,
+                    actor_cls,
+                    tag,
+                    init_args=(),
+                    init_kwargs={},
+                    is_asyncio=False):
+        """Start an actor and add it to the nursery"""
+        # Avoid double initialization
+        if tag in self.tag_to_actor_handles.keys():
+            return [self.tag_to_actor_handles[tag]]
+
+        max_concurrency = ASYNC_CONCURRENCY if is_asyncio else None
+        handle = (actor_cls.options(max_concurrency=max_concurrency).remote(
+            *init_args, **init_kwargs))
+        self.tag_to_actor_handles[tag] = handle
+        return [handle]
+
+    def start_actor_with_creator(self, creator, kwargs, tag):
+        """
+        Args:
+            creator (Callable[Dict]): a closure that should return
+                a newly created actor handle when called with kwargs.
+                The kwargs input is passed to `ActorCls_remote` method.
+        """
+        handle = creator(kwargs)
+        self.tag_to_actor_handles[tag] = handle
+        return [handle]
+
+    def get_all_handles(self):
+        return self.tag_to_actor_handles
+
+    def get_handle(self, actor_tag):
+        return [self.tag_to_actor_handles[actor_tag]]
+
+    def remove_handle(self, actor_tag):
+        if actor_tag in self.tag_to_actor_handles.keys():
+            self.tag_to_actor_handles.pop(actor_tag)
+
+    def store_bootstrap_state(self, key, value):
+        self.bootstrap_state[key] = value
+
+    def get_bootstrap_state_dict(self):
+        return self.bootstrap_state
+
+
+class GlobalState:
+    """Encapsulate all global state in the serving system.
+
+    The information is fetch lazily from
+        1. A collection of namespaced key value stores
+        2. A actor supervisor service
+    """
+
+    def __init__(self, actor_nursery_handle=None):
+        # Get actor nursery handle
+        if actor_nursery_handle is None:
+            actor_nursery_handle = ray.util.get_actor(SERVE_NURSERY_NAME)
+        self.actor_nursery_handle = actor_nursery_handle
+
+        # Connect to all the table
+        bootstrap_config = ray.get(
+            self.actor_nursery_handle.get_bootstrap_state_dict.remote())
+        kv_store_connector = bootstrap_config[BOOTSTRAP_KV_STORE_CONN_KEY]
+        self.route_table = RoutingTable(kv_store_connector)
+        self.backend_table = BackendTable(kv_store_connector)
+        self.policy_table = TrafficPolicyTable(kv_store_connector)
+
+        self.refresh_actor_handle_cache()
+
+    def refresh_actor_handle_cache(self):
+        self.actor_handle_cache = ray.get(
+            self.actor_nursery_handle.get_all_handles.remote())
+
+    def init_or_get_http_server(self,
+                                host=DEFAULT_HTTP_HOST,
+                                port=DEFAULT_HTTP_PORT):
+        if "http_server" not in self.actor_handle_cache:
+            [handle] = ray.get(
+                self.actor_nursery_handle.start_actor.remote(
+                    HTTPActor, tag="http_server"))
+
+            handle.run.remote(host=host, port=port)
+            self.refresh_actor_handle_cache()
+        return self.actor_handle_cache["http_server"]
+
+    def _get_queueing_policy(self, default_policy):
+        return_policy = default_policy
+        # check if there is already a queue_actor running
+        # with policy as p.name for the case where
+        # serve nursery exists: ray.util.get_actor(SERVE_NURSERY_NAME)
+        for p in RoutePolicy:
+            queue_actor_tag = "queue_actor::" + p.name
+            if queue_actor_tag in self.actor_handle_cache:
+                return_policy = p
+                break
+        return return_policy
+
+    def init_or_get_router(self,
+                           queueing_policy=RoutePolicy.Random,
+                           policy_kwargs={}):
+        # get queueing policy
+        self.queueing_policy = self._get_queueing_policy(
+            default_policy=queueing_policy)
+        queue_actor_tag = "queue_actor::" + self.queueing_policy.name
+        if queue_actor_tag not in self.actor_handle_cache:
+            [handle] = ray.get(
+                self.actor_nursery_handle.start_actor.remote(
+                    self.queueing_policy.value,
+                    init_kwargs=policy_kwargs,
+                    tag=queue_actor_tag,
+                    is_asyncio=True))
+            # handle.register_self_handle.remote(handle)
+            self.refresh_actor_handle_cache()
+
+        return self.actor_handle_cache[queue_actor_tag]
+
+    def init_or_get_metric_monitor(self, gc_window_seconds=3600):
+        if "metric_monitor" not in self.actor_handle_cache:
+            [handle] = ray.get(
+                self.actor_nursery_handle.start_actor.remote(
+                    MetricMonitor,
+                    init_args=(gc_window_seconds, ),
+                    tag="metric_monitor"))
+
+            start_metric_monitor_loop.remote(handle)
+
+            if "queue_actor" in self.actor_handle_cache:
+                handle.add_target.remote(
+                    self.actor_handle_cache["queue_actor"])
+
+            self.refresh_actor_handle_cache()
+
+        return self.actor_handle_cache["metric_monitor"]

python/ray/serve/handle.py

@@ -0,0 +1,106 @@
+from ray import serve
+from ray.serve.context import TaskContext
+from ray.serve.exceptions import RayServeException
+from ray.serve.constants import DEFAULT_HTTP_ADDRESS
+from ray.serve.request_params import RequestMetadata
+
+
+class RayServeHandle:
+    """A handle to a service endpoint.
+
+    Invoking this endpoint with .remote is equivalent to pinging
+    an HTTP endpoint.
+
+    Example:
+       >>> handle = serve.get_handle("my_endpoint")
+       >>> handle
+       RayServeHandle(
+            Endpoint="my_endpoint",
+            URL="...",
+            Traffic=...
+       )
+       >>> handle.remote(my_request_content)
+       ObjectID(...)
+       >>> ray.get(handle.remote(...))
+       # result
+       >>> ray.get(handle.remote(let_it_crash_request))
+       # raises RayTaskError Exception
+    """
+
+    def __init__(self,
+                 router_handle,
+                 endpoint_name,
+                 relative_slo_ms=None,
+                 absolute_slo_ms=None):
+        self.router_handle = router_handle
+        self.endpoint_name = endpoint_name
+        assert (relative_slo_ms is None
+                or absolute_slo_ms is None), ("Can't specify both "
+                                              "relative and absolute "
+                                              "slo's together!")
+        self.relative_slo_ms = self._check_slo_ms(relative_slo_ms)
+        self.absolute_slo_ms = self._check_slo_ms(absolute_slo_ms)
+
+    def _check_slo_ms(self, slo_value):
+        if slo_value is not None:
+            try:
+                slo_value = float(slo_value)
+                if slo_value < 0:
+                    raise ValueError(
+                        "Request SLO must be positive, it is {}".format(
+                            slo_value))
+                return slo_value
+            except ValueError as e:
+                raise RayServeException(str(e))
+        return None
+
+    def remote(self, *args, **kwargs):
+        if len(args) != 0:
+            raise RayServeException(
+                "handle.remote must be invoked with keyword arguments.")
+
+        # create RequestMetadata instance
+        request_in_object = RequestMetadata(
+            self.endpoint_name, TaskContext.Python, self.relative_slo_ms,
+            self.absolute_slo_ms)
+        return self.router_handle.enqueue_request.remote(
+            request_in_object, **kwargs)
+
+    def options(self, relative_slo_ms=None, absolute_slo_ms=None):
+        # If both the slo's are None then then we use a high default
+        # value so other queries can be prioritize and put in front of these
+        # queries.
+        assert (relative_slo_ms is None
+                or absolute_slo_ms is None), ("Can't specify both "
+                                              "relative and absolute "
+                                              "slo's together!")
+        return RayServeHandle(self.router_handle, self.endpoint_name,
+                              relative_slo_ms, absolute_slo_ms)
+
+    def get_traffic_policy(self):
+        # TODO(simon): This method is implemented via checking global state
+        # because we are sure handle and global_state are in the same process.
+        # However, once global_state is deprecated, this method need to be
+        # updated accordingly.
+        history = serve.global_state.policy_action_history[self.endpoint_name]
+        if len(history):
+            return history[-1]
+        else:
+            return None
+
+    def get_http_endpoint(self):
+        return DEFAULT_HTTP_ADDRESS
+
+    def __repr__(self):
+        return """
+RayServeHandle(
+    Endpoint="{endpoint_name}",
+    URL="{http_endpoint}/{endpoint_name}",
+    Traffic={traffic_policy}
+)
+""".format(endpoint_name=self.endpoint_name,
+           http_endpoint=self.get_http_endpoint(),
+           traffic_policy=self.get_traffic_policy())
+
+    # TODO(simon): a convenience function that dumps equivalent requests
+    # code for a given call.

python/ray/serve/http_util.py

@@ -0,0 +1,69 @@
+import io
+
+import flask
+
+
+def build_flask_request(asgi_scope_dict, request_body):
+    """Build and return a flask request from ASGI payload
+
+    This function is indented to be used immediately before task invocation
+    happen.
+    """
+    wsgi_environ = build_wsgi_environ(asgi_scope_dict, request_body)
+    return flask.Request(wsgi_environ)
+
+
+def build_wsgi_environ(scope, body):
+    """
+    Builds a scope and request body into a WSGI environ object.
+
+    This code snippet is taken from https://github.com/django/asgiref/blob
+    /36c3e8dc70bf38fe2db87ac20b514f21aaf5ea9d/asgiref/wsgi.py#L52
+
+    WSGI specification can be found at
+    https://www.python.org/dev/peps/pep-0333/
+
+    This function helps translate ASGI scope and body into a flask request.
+    """
+    environ = {
+        "REQUEST_METHOD": scope["method"],
+        "SCRIPT_NAME": scope.get("root_path", ""),
+        "PATH_INFO": scope["path"],
+        "QUERY_STRING": scope["query_string"].decode("ascii"),
+        "SERVER_PROTOCOL": "HTTP/{}".format(scope["http_version"]),
+        "wsgi.version": (1, 0),
+        "wsgi.url_scheme": scope.get("scheme", "http"),
+        "wsgi.input": body,
+        "wsgi.errors": io.BytesIO(),
+        "wsgi.multithread": True,
+        "wsgi.multiprocess": True,
+        "wsgi.run_once": False,
+    }
+
+    # Get server name and port - required in WSGI, not in ASGI
+    environ["SERVER_NAME"] = scope["server"][0]
+    environ["SERVER_PORT"] = str(scope["server"][1])
+    environ["REMOTE_ADDR"] = scope["client"][0]
+
+    # Transforms headers into environ entries.
+    for name, value in scope.get("headers", []):
+        # name, values are both bytes, we need to decode them to string
+        name = name.decode("latin1")
+        value = value.decode("latin1")
+
+        # Handle name correction to conform to WSGI spec
+        # https://www.python.org/dev/peps/pep-0333/#environ-variables
+        if name == "content-length":
+            corrected_name = "CONTENT_LENGTH"
+        elif name == "content-type":
+            corrected_name = "CONTENT_TYPE"
+        else:
+            corrected_name = "HTTP_%s" % name.upper().replace("-", "_")
+
+        # If the header value repeated,
+        # we will just concatenate it to the field.
+        if corrected_name in environ:
+            value = environ[corrected_name] + "," + value
+
+        environ[corrected_name] = value
+    return environ

python/ray/serve/kv_store_service.py

@@ -0,0 +1,283 @@
+import json
+import sqlite3
+from abc import ABC
+from typing import Union
+
+from ray import cloudpickle as pickle
+import ray.experimental.internal_kv as ray_kv
+from ray.serve.utils import logger
+from ray.serve.constants import NO_ROUTE_KEY
+
+
+class NamespacedKVStore(ABC):
+    """Abstract base class for a namespaced key-value store.
+
+    The idea is that multiple key-value stores can be created while sharing
+    the same storage system. The keys of each instance are namespaced to avoid
+    object_id key collision.
+
+    Example:
+
+    >>> store_ns1 = NamespacedKVStore(namespace="ns1")
+    >>> store_ns2 = NamespacedKVStore(namespace="ns2")
+    # Two stores can share the same connection like Redis or SQL Table
+    >>> store_ns1.put("same-key", 1)
+    >>> store_ns1.get("same-key")
+    1
+    >>> store_ns2.put("same-key", 2)
+    >>> store_ns2.get("same-key", 2)
+    2
+    """
+
+    def __init__(self, namespace):
+        raise NotImplementedError()
+
+    def get(self, key):
+        """Retrieve the value for the given key.
+
+        Args:
+            key (str)
+        """
+        raise NotImplementedError()
+
+    def put(self, key, value):
+        """Serialize the value and store it under the given key.
+
+        Args:
+            key (str)
+            value (object): any serializable object. The serialization method
+                is determined by the subclass implementation.
+        """
+        raise NotImplementedError()
+
+    def as_dict(self):
+        """Return the entire namespace as a dictionary.
+
+        Returns:
+            data (dict): key value pairs in current namespace
+        """
+        raise NotImplementedError()
+
+
+class InMemoryKVStore(NamespacedKVStore):
+    """A reference implementation used for testing."""
+
+    def __init__(self, namespace):
+        self.data = dict()
+
+        # Namespace is ignored, because each namespace is backed by
+        # an in-memory Python dictionary.
+        self.namespace = namespace
+
+    def get(self, key):
+        return self.data[key]
+
+    def put(self, key, value):
+        self.data[key] = value
+
+    def as_dict(self):
+        return self.data.copy()
+
+
+class RayInternalKVStore(NamespacedKVStore):
+    """A NamespacedKVStore implementation using ray's `internal_kv`."""
+
+    def __init__(self, namespace):
+        assert ray_kv._internal_kv_initialized()
+        self.index_key = "RAY_SERVE_INDEX"
+        self.namespace = namespace
+        self._put(self.index_key, [])
+
+    def _format_key(self, key):
+        return "{ns}-{key}".format(ns=self.namespace, key=key)
+
+    def _remove_format_key(self, formatted_key):
+        return formatted_key.replace(self.namespace + "-", "", 1)
+
+    def _serialize(self, obj):
+        return json.dumps(obj)
+
+    def _deserialize(self, buffer):
+        return json.loads(buffer)
+
+    def _put(self, key, value):
+        ray_kv._internal_kv_put(
+            self._format_key(self._serialize(key)),
+            self._serialize(value),
+            overwrite=True,
+        )
+
+    def _get(self, key):
+        return self._deserialize(
+            ray_kv._internal_kv_get(self._format_key(self._serialize(key))))
+
+    def get(self, key):
+        return self._get(key)
+
+    def put(self, key, value):
+        assert isinstance(key, str), "Key must be a string."
+
+        self._put(key, value)
+
+        all_keys = set(self._get(self.index_key))
+        all_keys.add(key)
+        self._put(self.index_key, list(all_keys))
+
+    def as_dict(self):
+        data = {}
+        all_keys = self._get(self.index_key)
+        for key in all_keys:
+            data[self._remove_format_key(key)] = self._get(key)
+        return data
+
+
+class SQLiteKVStore(NamespacedKVStore):
+    def __init__(self, namespace, db_path):
+        self.namespace = namespace
+        self.conn = sqlite3.connect(db_path)
+
+        cursor = self.conn.cursor()
+        cursor.execute(
+            "CREATE TABLE IF NOT EXISTS {} (key TEXT UNIQUE, value TEXT)".
+            format(self.namespace))
+        self.conn.commit()
+
+    def put(self, key, value):
+        cursor = self.conn.cursor()
+        cursor.execute(
+            "INSERT OR REPLACE INTO {} (key, value) VALUES (?,?)".format(
+                self.namespace), (key, value))
+        self.conn.commit()
+
+    def get(self, key, default=None):
+        cursor = self.conn.cursor()
+        result = list(
+            cursor.execute(
+                "SELECT value FROM {} WHERE key = (?)".format(self.namespace),
+                (key, )))
+        if len(result) == 0:
+            return default
+        else:
+            # Due to UNIQUE constraint, there can only be one value.
+            value, *_ = result[0]
+            return value
+
+    def as_dict(self):
+        cursor = self.conn.cursor()
+        result = list(
+            cursor.execute("SELECT key, value FROM {}".format(self.namespace)))
+        return dict(result)
+
+
+# Tables
+class RoutingTable:
+    def __init__(self, kv_connector):
+        self.routing_table = kv_connector("routing_table")
+        self.request_count = 0
+
+    def register_service(self, route: Union[str, None], service: str):
+        """Create an entry in the routing table
+
+        Args:
+            route: http path name. Must begin with '/'.
+            service: service name. This is the name http actor will push
+                the request to.
+        """
+        logger.debug("[KV] Registering route {} to service {}.".format(
+            route, service))
+
+        # put no route services in default key
+        if route is None:
+            no_http_services = json.loads(
+                self.routing_table.get(NO_ROUTE_KEY, "[]"))
+            no_http_services.append(service)
+            self.routing_table.put(NO_ROUTE_KEY, json.dumps(no_http_services))
+        else:
+            self.routing_table.put(route, service)
+
+    def list_service(self, include_headless=False):
+        """Returns the routing table.
+        Args:
+            include_headless: If True, returns a no route services (headless)
+                services with normal services. (Default: False)
+        """
+        table = self.routing_table.as_dict()
+        if include_headless:
+            table[NO_ROUTE_KEY] = json.loads(table.get(NO_ROUTE_KEY, "[]"))
+        else:
+            table.pop(NO_ROUTE_KEY, None)
+        return table
+
+    def get_request_count(self):
+        """Return the number of requests that fetched the routing table.
+
+        This method is used for two purpose:
+
+        1. Make sure HTTP server has started and healthy. Incremented request
+           count means HTTP server is actively fetching routing table.
+
+        2. Make sure HTTP server does not have stale routing table. This number
+           should be incremented every HTTP_ROUTER_CHECKER_INTERVAL_S seconds.
+           Supervisor should check this number as indirect indicator of http
+           server's health.
+        """
+        return self.request_count
+
+
+class BackendTable:
+    def __init__(self, kv_connector):
+        self.backend_table = kv_connector("backend_creator")
+        self.replica_table = kv_connector("replica_table")
+        self.backend_info = kv_connector("backend_info")
+        self.backend_init_args = kv_connector("backend_init_args")
+
+    def register_backend(self, backend_tag: str, backend_creator):
+        backend_creator_serialized = pickle.dumps(backend_creator)
+        self.backend_table.put(backend_tag, backend_creator_serialized)
+
+    def save_init_args(self, backend_tag: str, arg_list):
+        serialized_arg_list = pickle.dumps(arg_list)
+        self.backend_init_args.put(backend_tag, serialized_arg_list)
+
+    def get_init_args(self, backend_tag):
+        return pickle.loads(self.backend_init_args.get(backend_tag))
+
+    def register_info(self, backend_tag: str, backend_info_d):
+        self.backend_info.put(backend_tag, json.dumps(backend_info_d))
+
+    def get_info(self, backend_tag):
+        return json.loads(self.backend_info.get(backend_tag, "{}"))
+
+    def get_backend_creator(self, backend_tag):
+        return pickle.loads(self.backend_table.get(backend_tag))
+
+    def list_backends(self):
+        return list(self.backend_table.as_dict().keys())
+
+    def list_replicas(self, backend_tag: str):
+        return json.loads(self.replica_table.get(backend_tag, "[]"))
+
+    def add_replica(self, backend_tag: str, new_replica_tag: str):
+        replica_tags = self.list_replicas(backend_tag)
+        replica_tags.append(new_replica_tag)
+        self.replica_table.put(backend_tag, json.dumps(replica_tags))
+
+    def remove_replica(self, backend_tag):
+        replica_tags = self.list_replicas(backend_tag)
+        removed_replica = replica_tags.pop()
+        self.replica_table.put(backend_tag, json.dumps(replica_tags))
+        return removed_replica
+
+
+class TrafficPolicyTable:
+    def __init__(self, kv_connector):
+        self.traffic_policy_table = kv_connector("traffic_policy")
+
+    def register_traffic_policy(self, service_name, policy_dict):
+        self.traffic_policy_table.put(service_name, json.dumps(policy_dict))
+
+    def list_traffic_policy(self):
+        return {
+            service: json.loads(policy)
+            for service, policy in self.traffic_policy_table.as_dict()
+        }

python/ray/serve/metric.py

@@ -0,0 +1,157 @@
+import time
+
+import numpy as np
+import pandas as pd
+
+import ray
+
+
+@ray.remote(num_cpus=0)
+class MetricMonitor:
+    def __init__(self, gc_window_seconds=3600):
+        """Metric monitor scrapes metrics from ray serve actors
+        and allow windowed query operations.
+
+        Args:
+            gc_window_seconds(int): How long will we keep the metric data in
+                memory. Data older than the gc_window will be deleted.
+        """
+        #: Mapping actor ID (hex) -> actor handle
+        self.actor_handles = dict()
+
+        self.data_entries = []
+
+        self.gc_window_seconds = gc_window_seconds
+        self.latest_gc_time = time.time()
+
+    def is_ready(self):
+        return True
+
+    def add_target(self, target_handle):
+        hex_id = target_handle._actor_id.hex()
+        self.actor_handles[hex_id] = target_handle
+
+    def remove_target(self, target_handle):
+        hex_id = target_handle._actor_id.hex()
+        self.actor_handles.pop(hex_id)
+
+    def scrape(self):
+        # If expected gc time has passed, we will perform metric value GC.
+        expected_gc_time = self.latest_gc_time + self.gc_window_seconds
+        if expected_gc_time < time.time():
+            self._perform_gc()
+            self.latest_gc_time = time.time()
+
+        curr_time = time.time()
+        result = [
+            handle._serve_metric.remote()
+            for handle in self.actor_handles.values()
+        ]
+        # TODO(simon): handle the possibility that an actor_handle is removed
+        for handle_result in ray.get(result):
+            for metric_name, metric_info in handle_result.items():
+                data_entry = {
+                    "retrieved_at": curr_time,
+                    "name": metric_name,
+                    "type": metric_info["type"],
+                }
+
+                if metric_info["type"] == "counter":
+                    data_entry["value"] = metric_info["value"]
+                    self.data_entries.append(data_entry)
+
+                elif metric_info["type"] == "list":
+                    for metric_value in metric_info["value"]:
+                        new_entry = data_entry.copy()
+                        new_entry["value"] = metric_value
+                        self.data_entries.append(new_entry)
+
+    def _perform_gc(self):
+        curr_time = time.time()
+        earliest_time_allowed = curr_time - self.gc_window_seconds
+
+        # If we don"t have any data at hand, no need to gc.
+        if len(self.data_entries) == 0:
+            return
+
+        df = pd.DataFrame(self.data_entries)
+        df = df[df["retrieved_at"] >= earliest_time_allowed]
+        self.data_entries = df.to_dict(orient="record")
+
+    def _get_dataframe(self):
+        return pd.DataFrame(self.data_entries)
+
+    def collect(self,
+                percentiles=[50, 90, 95],
+                agg_windows_seconds=[10, 60, 300, 600, 3600]):
+        """Collect and perform aggregation on all metrics.
+
+        Args:
+            percentiles(List[int]): The percentiles for aggregation operations.
+                Default is 50th, 90th, 95th percentile.
+            agg_windows_seconds(List[int]): The aggregation windows in seconds.
+                The longest aggregation window must be shorter or equal to the
+                gc_window_seconds.
+        """
+        result = {}
+        df = pd.DataFrame(self.data_entries)
+
+        if len(df) == 0:  # no metric to report
+            return {}
+
+        # Retrieve the {metric_name -> metric_type} mapping
+        metric_types = df[["name",
+                           "type"]].set_index("name").squeeze().to_dict()
+
+        for metric_name, metric_type in metric_types.items():
+            if metric_type == "counter":
+                result[metric_name] = df.loc[df["name"] == metric_name,
+                                             "value"].tolist()[-1]
+            if metric_type == "list":
+                result.update(
+                    self._aggregate(metric_name, percentiles,
+                                    agg_windows_seconds))
+        return result
+
+    def _aggregate(self, metric_name, percentiles, agg_windows_seconds):
+        """Perform aggregation over a metric.
+
+        Note:
+            This metric must have type `list`.
+        """
+        assert max(agg_windows_seconds) <= self.gc_window_seconds, (
+            "Aggregation window exceeds gc window. You should set a longer gc "
+            "window or shorter aggregation window.")
+
+        curr_time = time.time()
+        df = pd.DataFrame(self.data_entries)
+        filtered_df = df[df["name"] == metric_name]
+        if len(filtered_df) == 0:
+            return dict()
+
+        data_types = filtered_df["type"].unique().tolist()
+        assert data_types == [
+            "list"
+        ], ("Can't aggreagte over non-list type. {} has type {}".format(
+            metric_name, data_types))
+
+        aggregated_metric = {}
+        for window in agg_windows_seconds:
+            earliest_time = curr_time - window
+            windowed_df = filtered_df[
+                filtered_df["retrieved_at"] > earliest_time]
+            percentile_values = np.percentile(windowed_df["value"],
+                                              percentiles)
+            for percentile, value in zip(percentiles, percentile_values):
+                result_key = "{name}_{perc}th_perc_{window}_window".format(
+                    name=metric_name, perc=percentile, window=window)
+                aggregated_metric[result_key] = value
+
+        return aggregated_metric
+
+
+@ray.remote(num_cpus=0)
+def start_metric_monitor_loop(monitor_handle, duration_s=5):
+    while True:
+        ray.get(monitor_handle.scrape.remote())
+        time.sleep(duration_s)

python/ray/serve/policy.py

@@ -0,0 +1,188 @@
+from enum import Enum
+import itertools
+
+import numpy as np
+
+import ray
+from ray.serve.queues import (CentralizedQueues)
+from ray.serve.utils import logger
+
+
+class RandomPolicyQueue(CentralizedQueues):
+    """
+    A wrapper class for Random policy.This backend selection policy is
+    `Stateless` meaning the current decisions of selecting backend are
+    not dependent on previous decisions. Random policy (randomly) samples
+    backends based on backend weights for every query. This policy uses the
+    weights assigned to backends.
+    """
+
+    async def _flush_service_queues(self):
+        # perform traffic splitting for requests
+        for service, queue in self.service_queues.items():
+            # while there are incoming requests and there are backends
+            while queue.qsize() and len(self.traffic[service]):
+                backend_names = list(self.traffic[service].keys())
+                backend_weights = list(self.traffic[service].values())
+                # randomly choose a backend for every query
+                chosen_backend = np.random.choice(
+                    backend_names, replace=False, p=backend_weights).squeeze()
+                logger.debug("Matching service {} to backend {}".format(
+                    service, chosen_backend))
+
+                request = await queue.get()
+                self.buffer_queues[chosen_backend].add(request)
+
+
+@ray.remote
+class RandomPolicyQueueActor(RandomPolicyQueue):
+    pass
+
+
+class RoundRobinPolicyQueue(CentralizedQueues):
+    """
+    A wrapper class for RoundRobin policy. This backend selection policy
+    is `Stateful` meaning the current decisions of selecting backend are
+    dependent on previous decisions. RoundRobinPolicy assigns queries in
+    an interleaved manner to every backend serving for a service. Consider
+    backend A,B linked to a service. Now queries will be assigned to backends
+    in the following order - [ A, B, A, B ... ] . This policy doesn't use the
+    weights assigned to backends.
+    """
+
+    # Saves the information about last assigned
+    # backend for every service
+    round_robin_iterator_map = {}
+
+    async def set_traffic(self, service, traffic_dict):
+        logger.debug("Setting traffic for service %s to %s", service,
+                     traffic_dict)
+        self.traffic[service] = traffic_dict
+        backend_names = list(self.traffic[service].keys())
+        self.round_robin_iterator_map[service] = itertools.cycle(backend_names)
+        await self.flush()
+
+    async def _flush_service_queues(self):
+        # perform traffic splitting for requests
+        for service, queue in self.service_queues.items():
+            # if there are incoming requests and there are backends
+            if queue.qsize() and len(self.traffic[service]):
+                while queue.qsize():
+                    # choose the next backend available from persistent
+                    # information
+                    chosen_backend = next(
+                        self.round_robin_iterator_map[service])
+                    request = await queue.get()
+                    self.buffer_queues[chosen_backend].add(request)
+
+
+@ray.remote
+class RoundRobinPolicyQueueActor(RoundRobinPolicyQueue):
+    pass
+
+
+class PowerOfTwoPolicyQueue(CentralizedQueues):
+    """
+    A wrapper class for powerOfTwo policy. This backend selection policy is
+    `Stateless` meaning the current decisions of selecting backend are
+    dependent on previous decisions. PowerOfTwo policy (randomly) samples two
+    backends (say Backend A,B among A,B,C) based on the backend weights
+    specified and chooses the backend which is less loaded. This policy uses
+    the weights assigned to backends.
+    """
+
+    async def _flush_service_queues(self):
+        # perform traffic splitting for requests
+        for service, queue in self.service_queues.items():
+            # while there are incoming requests and there are backends
+            while queue.qsize() and len(self.traffic[service]):
+                backend_names = list(self.traffic[service].keys())
+                backend_weights = list(self.traffic[service].values())
+                if len(self.traffic[service]) >= 2:
+                    # randomly pick 2 backends
+                    backend1, backend2 = np.random.choice(
+                        backend_names, 2, replace=False, p=backend_weights)
+
+                    # see the length of buffer queues of the two backends
+                    # and pick the one which has less no. of queries
+                    # in the buffer
+                    if (len(self.buffer_queues[backend1]) <= len(
+                            self.buffer_queues[backend2])):
+                        chosen_backend = backend1
+                    else:
+                        chosen_backend = backend2
+                    logger.debug("[Power of two chocies] found two backends "
+                                 "{} and {}: choosing {}.".format(
+                                     backend1, backend2, chosen_backend))
+                else:
+                    chosen_backend = np.random.choice(
+                        backend_names, replace=False,
+                        p=backend_weights).squeeze()
+                request = await queue.get()
+                self.buffer_queues[chosen_backend].add(request)
+
+
+@ray.remote
+class PowerOfTwoPolicyQueueActor(PowerOfTwoPolicyQueue):
+    pass
+
+
+class FixedPackingPolicyQueue(CentralizedQueues):
+    """
+    A wrapper class for FixedPacking policy. This backend selection policy is
+    `Stateful` meaning the current decisions of selecting backend are dependent
+    on previous decisions. FixedPackingPolicy is k RoundRobin policy where
+    first packing_num queries are handled by 'backend-1' and next k queries are
+    handled by 'backend-2' and so on ... where 'backend-1' and 'backend-2' are
+    served by the same service. This policy doesn't use the weights assigned to
+    backends.
+
+    """
+
+    def __init__(self, packing_num=3):
+        # Saves the information about last assigned
+        # backend for every service
+        self.fixed_packing_iterator_map = {}
+        self.packing_num = packing_num
+        super().__init__()
+
+    async def set_traffic(self, service, traffic_dict):
+        logger.debug("Setting traffic for service %s to %s", service,
+                     traffic_dict)
+        self.traffic[service] = traffic_dict
+        backend_names = list(self.traffic[service].keys())
+        self.fixed_packing_iterator_map[service] = itertools.cycle(
+            itertools.chain.from_iterable(
+                itertools.repeat(x, self.packing_num) for x in backend_names))
+        await self.flush()
+
+    async def _flush_service_queues(self):
+        # perform traffic splitting for requests
+        for service, queue in self.service_queues.items():
+            # if there are incoming requests and there are backends
+            if queue.qsize() and len(self.traffic[service]):
+                while queue.qsize():
+                    # choose the next backend available from persistent
+                    # information
+                    chosen_backend = next(
+                        self.fixed_packing_iterator_map[service])
+                    request = await queue.get()
+                    self.buffer_queues[chosen_backend].add(request)
+
+
+@ray.remote
+class FixedPackingPolicyQueueActor(FixedPackingPolicyQueue):
+    pass
+
+
+class RoutePolicy(Enum):
+    """
+    A class for registering the backend selection policy.
+    Add a name and the corresponding class.
+    Serve will support the added policy and policy can be accessed
+    in `serve.init` method through name provided here.
+    """
+    Random = RandomPolicyQueueActor
+    RoundRobin = RoundRobinPolicyQueueActor
+    PowerOfTwo = PowerOfTwoPolicyQueueActor
+    FixedPacking = FixedPackingPolicyQueueActor

python/ray/serve/queues.py

@@ -0,0 +1,305 @@
+import asyncio
+import copy
+from collections import defaultdict
+from typing import DefaultDict, List
+import pickle
+
+# Note on choosing blist instead of stdlib heapq
+# 1. pop operation should be O(1) (amortized)
+#    (helpful even for batched pop)
+# 2. There should not be significant overhead in
+#    maintaining the sorted list.
+# 3. The blist implementation is fast and uses C extensions.
+import blist
+
+import ray
+from ray.serve.utils import logger
+
+
+class Query:
+    def __init__(self, request_args, request_kwargs, request_context,
+                 request_slo_ms):
+        self.request_args = request_args
+        self.request_kwargs = request_kwargs
+        self.request_context = request_context
+
+        self.async_future = asyncio.get_event_loop().create_future()
+
+        # Service level objective in milliseconds. This is expected to be the
+        # absolute time since unix epoch.
+        self.request_slo_ms = request_slo_ms
+
+    def ray_serialize(self):
+        # NOTE: this method is needed because Query need to be serialized and
+        # sent to the replica worker. However, after we send the query to
+        # replica worker the async_future is still needed to retrieve the final
+        # result. Therefore we need a way to pass the information to replica
+        # worker without removing async_future.
+        clone = copy.copy(self)
+        clone.async_future = None
+        # We can't use cloudpickle due to a recursion issue
+        return pickle.dumps(clone)
+
+    @staticmethod
+    def ray_deserialize(value):
+        return pickle.loads(value)
+
+    # adding comparator fn for maintaining an
+    # ascending order sorted list w.r.t request_slo_ms
+    def __lt__(self, other):
+        return self.request_slo_ms < other.request_slo_ms
+
+    def __repr__(self):
+        return "<Query args={} kwargs={}>".format(self.request_args,
+                                                  self.request_kwargs)
+
+
+def _make_future_unwrapper(client_futures: List[asyncio.Future],
+                           host_future: asyncio.Future):
+    """Distribute the result of host_future to each of client_future"""
+    for client_future in client_futures:
+        # Keep a reference to host future so the host future won't get
+        # garbage collected.
+        client_future.host_ref = host_future
+
+    def unwrap_future(_):
+        result = host_future.result()
+
+        if isinstance(result, list):
+            for client_future, result_item in zip(client_futures, result):
+                client_future.set_result(result_item)
+        else:  # Result is an exception.
+            for client_future in client_futures:
+                client_future.set_result(result)
+
+    return unwrap_future
+
+
+class CentralizedQueues:
+    """A router that routes request to available workers.
+
+    Router aceepts each request from the `enqueue_request` method and enqueues
+    it. It also accepts worker request to work (called work_intention in code)
+    from workers via the `dequeue_request` method. The traffic policy is used
+    to match requests with their corresponding workers.
+
+    Behavior:
+        >>> # psuedo-code
+        >>> queue = CentralizedQueues()
+        >>> queue.enqueue_request(
+            "service-name", request_args, request_kwargs, request_context)
+        # nothing happens, request is queued.
+        # returns result ObjectID, which will contains the final result
+        >>> queue.dequeue_request('backend-1', replica_handle)
+        # nothing happens, work intention is queued.
+        # return work ObjectID, which will contains the future request payload
+        >>> queue.link('service-name', 'backend-1')
+        # here the enqueue_requester is matched with replica, request
+        # data is put into work ObjectID, and the replica processes the request
+        # and store the result into result ObjectID
+
+    Traffic policy splits the traffic among different replicas
+    probabilistically:
+
+    1. When all backends are ready to receive traffic, we will randomly
+       choose a backend based on the weights assigned by the traffic policy
+       dictionary.
+
+    2. When more than 1 but not all backends are ready, we will normalize the
+       weights of the ready backends to 1 and choose a backend via sampling.
+
+    3. When there is only 1 backend ready, we will only use that backend.
+    """
+
+    def __init__(self):
+        # Note: Several queues are used in the router
+        # - When a request come in, it's placed inside its corresponding
+        #   service_queue.
+        # - The service_queue is dequed during flush operation, which moves
+        #   the queries to backend buffer_queue. Here we match a request
+        #   for a service to a backend given some policy.
+        # - The worker_queue is used to collect idle actor handle. These
+        #   handles are dequed during the second stage of flush operation,
+        #   which assign queries in buffer_queue to actor handle.
+
+        # -- Queues -- #
+
+        # service_name -> request queue
+        self.service_queues: DefaultDict[asyncio.Queue[Query]] = defaultdict(
+            asyncio.Queue)
+        # backend_name -> worker request queue
+        self.worker_queues: DefaultDict[asyncio.Queue[
+            ray.actor.ActorHandle]] = defaultdict(asyncio.Queue)
+        # backend_name -> worker payload queue
+        self.buffer_queues = defaultdict(blist.sortedlist)
+
+        # -- Metadata -- #
+
+        # service_name -> traffic_policy
+        self.traffic = defaultdict(dict)
+        # backend_name -> backend_config
+        self.backend_info = dict()
+
+        # -- Synchronization -- #
+
+        # This lock guarantee that only one flush operation can happen at a
+        # time. Without the lock, multiple flush operation can pop from the
+        # same buffer_queue and worker_queue and create deadlock. For example,
+        # an operation holding the only query and the other flush operation
+        # holding the only idle replica. Additionally, allowing only one flush
+        # operation at a time simplifies design overhead for custom queuing and
+        # batching polcies.
+        self.flush_lock = asyncio.Lock()
+
+    def is_ready(self):
+        return True
+
+    def _serve_metric(self):
+        return {
+            "backend_{}_queue_size".format(backend_name): {
+                "value": len(queue),
+                "type": "counter",
+            }
+            for backend_name, queue in self.buffer_queues.items()
+        }
+
+    async def enqueue_request(self, request_in_object, *request_args,
+                              **request_kwargs):
+        service = request_in_object.service
+        logger.debug("Received a request for service {}".format(service))
+
+        # check if the slo specified is directly the
+        # wall clock time
+        if request_in_object.absolute_slo_ms is not None:
+            request_slo_ms = request_in_object.absolute_slo_ms
+        else:
+            request_slo_ms = request_in_object.adjust_relative_slo_ms()
+        request_context = request_in_object.request_context
+        query = Query(request_args, request_kwargs, request_context,
+                      request_slo_ms)
+        await self.service_queues[service].put(query)
+        await self.flush()
+
+        # Note: a future change can be to directly return the ObjectID from
+        # replica task submission
+        result = await query.async_future
+        return result
+
+    async def dequeue_request(self, backend, replica_handle):
+        logger.debug(
+            "Received a dequeue request for backend {}".format(backend))
+        await self.worker_queues[backend].put(replica_handle)
+        await self.flush()
+
+    async def remove_and_destory_replica(self, backend, replica_handle):
+        # We need this lock because we modify worker_queue here.
+        async with self.flush_lock:
+            old_queue = self.worker_queues[backend]
+            new_queue = asyncio.Queue()
+            target_id = replica_handle._actor_id
+
+            while not old_queue.empty():
+                replica_handle = await old_queue.get()
+                if replica_handle._actor_id != target_id:
+                    await new_queue.put(replica_handle)
+
+            self.worker_queues[backend] = new_queue
+            # TODO: consider await this with timeout, or use ray_kill
+            replica_handle.__ray_terminate__.remote()
+
+    async def link(self, service, backend):
+        logger.debug("Link %s with %s", service, backend)
+        await self.set_traffic(service, {backend: 1.0})
+
+    async def set_traffic(self, service, traffic_dict):
+        logger.debug("Setting traffic for service %s to %s", service,
+                     traffic_dict)
+        self.traffic[service] = traffic_dict
+        await self.flush()
+
+    async def set_backend_config(self, backend, config_dict):
+        logger.debug("Setting backend config for "
+                     "backend {} to {}".format(backend, config_dict))
+        self.backend_info[backend] = config_dict
+
+    async def flush(self):
+        """In the default case, flush calls ._flush.
+
+        When this class is a Ray actor, .flush can be scheduled as a remote
+        method invocation.
+        """
+        async with self.flush_lock:
+            await self._flush_service_queues()
+            await self._flush_buffer_queues()
+
+    def _get_available_backends(self, service):
+        backends_in_policy = set(self.traffic[service].keys())
+        available_workers = {
+            backend
+            for backend, queues in self.worker_queues.items()
+            if queues.qsize() > 0
+        }
+        return list(backends_in_policy.intersection(available_workers))
+
+    async def _flush_service_queues(self):
+        """Selects the backend and puts the service queue query to the buffer
+        Expected Implementation:
+            The implementer is expected to access and manipulate
+            self.service_queues        : dict[str,Deque]
+            self.buffer_queues : dict[str,sortedlist]
+        For registering the implemented policies register at policy.py
+        Expected Behavior:
+            the Deque of all services in self.service_queues linked with
+            atleast one backend must be empty irrespective of whatever
+            backend policy is implemented.
+        """
+        raise NotImplementedError(
+            "This method should be implemented by child class.")
+
+    # flushes the buffer queue and assigns work to workers
+    async def _flush_buffer_queues(self):
+        for service in self.traffic.keys():
+            ready_backends = self._get_available_backends(service)
+            for backend in ready_backends:
+                # no work available
+                if len(self.buffer_queues[backend]) == 0:
+                    continue
+
+                buffer_queue = self.buffer_queues[backend]
+                worker_queue = self.worker_queues[backend]
+
+                logger.debug("Assigning queries for backend {} with buffer "
+                             "queue size {} and worker queue size {}".format(
+                                 backend, len(buffer_queue),
+                                 worker_queue.qsize()))
+
+                max_batch_size = None
+                if backend in self.backend_info:
+                    max_batch_size = self.backend_info[backend][
+                        "max_batch_size"]
+
+                await self._assign_query_to_worker(buffer_queue, worker_queue,
+                                                   max_batch_size)
+
+    async def _assign_query_to_worker(self,
+                                      buffer_queue,
+                                      worker_queue,
+                                      max_batch_size=None):
+
+        while len(buffer_queue) and worker_queue.qsize():
+            worker = await worker_queue.get()
+            if max_batch_size is None:  # No batching
+                request = buffer_queue.pop(0)
+                future = worker._ray_serve_call.remote(request).as_future()
+                # chaining satisfies request.async_future with future result.
+                asyncio.futures._chain_future(future, request.async_future)
+            else:
+                real_batch_size = min(len(buffer_queue), max_batch_size)
+                requests = [
+                    buffer_queue.pop(0) for _ in range(real_batch_size)
+                ]
+                future = worker._ray_serve_call.remote(requests).as_future()
+                future.add_done_callback(
+                    _make_future_unwrapper(
+                        client_futures=[req.async_future for req in requests],
+                        host_future=future))

python/ray/serve/request_params.py

@@ -0,0 +1,37 @@
+import time
+from ray.serve.constants import DEFAULT_LATENCY_SLO_MS
+
+
+class RequestMetadata:
+    """
+    Request Arguments required for enqueuing a request to the service
+    queue.
+    Args:
+        service(str): A registered service endpoint.
+        request_context(TaskContext): Context of a request.
+        request_slo_ms(float): Expected time for the query to get
+            completed.
+        is_wall_clock_time(bool): if True, router won't add wall clock
+            time to `request_slo_ms`.
+    """
+
+    def __init__(self,
+                 service,
+                 request_context,
+                 relative_slo_ms=None,
+                 absolute_slo_ms=None):
+
+        self.service = service
+        self.request_context = request_context
+        self.relative_slo_ms = relative_slo_ms
+        self.absolute_slo_ms = absolute_slo_ms
+
+    def adjust_relative_slo_ms(self) -> float:
+        """Normalize the input latency objective to absoluate timestamp.
+
+        """
+        slo_ms = self.relative_slo_ms
+        if slo_ms is None:
+            slo_ms = DEFAULT_LATENCY_SLO_MS
+        current_time_ms = time.time() * 1000
+        return current_time_ms + slo_ms

python/ray/serve/scripts.py

@@ -0,0 +1,50 @@
+import json
+
+import click
+
+import ray
+import ray.serve as serve
+
+
+@click.group("serve", help="Commands working with ray serve")
+def serve_cli():
+    pass
+
+
+@serve_cli.command(help="Initialize ray serve components")
+def init():
+    ray.init(address="auto")
+    serve.init(blocking=True)
+
+
+@serve_cli.command(help="Split traffic for a endpoint")
+@click.argument("endpoint", required=True, type=str)
+# TODO(simon): Make traffic dictionary more ergonomic. e.g.
+#     --traffic backend1=0.5 --traffic backend2=0.5
+@click.option(
+    "--traffic",
+    required=True,
+    type=str,
+    help="Traffic dictionary in JSON format")
+def split(endpoint, traffic):
+    ray.init(address="auto")
+    serve.init()
+
+    serve.split(endpoint, json.loads(traffic))
+
+
+@serve_cli.command(help="Scale the number of replicas for a backend")
+@click.argument("backend", required=True, type=str)
+@click.option(
+    "--num-replicas",
+    required=True,
+    type=int,
+    help="New number of replicas to set")
+def scale(backend_tag, num_replicas):
+    if num_replicas <= 0:
+        click.Abort(
+            "Cannot set number of replicas to be smaller or equal to 0.")
+    ray.init(address="auto")
+    serve.init()
+
+    serve.scale(backend_tag, num_replicas)

python/ray/serve/server.py

@@ -0,0 +1,196 @@
+import asyncio
+import json
+
+import uvicorn
+
+import ray
+from ray.experimental.async_api import _async_init
+from ray.serve.constants import HTTP_ROUTER_CHECKER_INTERVAL_S
+from ray.serve.context import TaskContext
+from ray.serve.utils import BytesEncoder
+from ray.serve.request_params import RequestMetadata
+
+from urllib.parse import parse_qs
+
+
+class JSONResponse:
+    """ASGI compliant response class.
+
+    It is expected to be called in async context and pass along
+    `scope, receive, send` as in ASGI spec.
+
+    >>> await JSONResponse({"k": "v"})(scope, receive, send)
+    """
+
+    def __init__(self, content=None, status_code=200):
+        """Construct a JSON HTTP Response.
+
+        Args:
+            content (optional): Any JSON serializable object.
+            status_code (int, optional): Default status code is 200.
+        """
+        self.body = self.render(content)
+        self.status_code = status_code
+        self.raw_headers = [[b"content-type", b"application/json"]]
+
+    def render(self, content):
+        if content is None:
+            return b""
+        if isinstance(content, bytes):
+            return content
+        return json.dumps(content, cls=BytesEncoder, indent=2).encode()
+
+    async def __call__(self, scope, receive, send):
+        await send({
+            "type": "http.response.start",
+            "status": self.status_code,
+            "headers": self.raw_headers,
+        })
+        await send({"type": "http.response.body", "body": self.body})
+
+
+class HTTPProxy:
+    """
+    This class should be instantiated and ran by ASGI server.
+
+    >>> import uvicorn
+    >>> uvicorn.run(HTTPProxy(kv_store_actor_handle, router_handle))
+    # blocks forever
+    """
+
+    def __init__(self):
+        assert ray.is_initialized()
+
+        # Delay import due to GlobalState depends on HTTP actor
+        from ray.serve.global_state import GlobalState
+        self.serve_global_state = GlobalState()
+        self.route_table_cache = dict()
+
+        self.route_checker_task = None
+        self.route_checker_should_shutdown = False
+
+    async def route_checker(self, interval):
+        while True:
+            if self.route_checker_should_shutdown:
+                return
+
+            self.route_table_cache = (
+                self.serve_global_state.route_table.list_service())
+
+            await asyncio.sleep(interval)
+
+    async def handle_lifespan_message(self, scope, receive, send):
+        assert scope["type"] == "lifespan"
+
+        message = await receive()
+        if message["type"] == "lifespan.startup":
+            await _async_init()
+            self.route_checker_task = asyncio.get_event_loop().create_task(
+                self.route_checker(interval=HTTP_ROUTER_CHECKER_INTERVAL_S))
+            await send({"type": "lifespan.startup.complete"})
+        elif message["type"] == "lifespan.shutdown":
+            self.route_checker_task.cancel()
+            self.route_checker_should_shutdown = True
+            await send({"type": "lifespan.shutdown.complete"})
+
+    async def receive_http_body(self, scope, receive, send):
+        body_buffer = []
+        more_body = True
+        while more_body:
+            message = await receive()
+            assert message["type"] == "http.request"
+
+            more_body = message["more_body"]
+            body_buffer.append(message["body"])
+
+        return b"".join(body_buffer)
+
+    def _check_slo_ms(self, request_slo_ms):
+        if request_slo_ms is not None:
+            if len(request_slo_ms) != 1:
+                raise ValueError(
+                    "Multiple SLO specified, please specific only one.")
+            request_slo_ms = request_slo_ms[0]
+            request_slo_ms = float(request_slo_ms)
+            if request_slo_ms < 0:
+                raise ValueError(
+                    "Request SLO must be positive, it is {}".format(
+                        request_slo_ms))
+            return request_slo_ms
+        return None
+
+    async def __call__(self, scope, receive, send):
+        # NOTE: This implements ASGI protocol specified in
+        #       https://asgi.readthedocs.io/en/latest/specs/index.html
+
+        if scope["type"] == "lifespan":
+            await self.handle_lifespan_message(scope, receive, send)
+            return
+
+        assert scope["type"] == "http"
+        current_path = scope["path"]
+        if current_path == "/":
+            await JSONResponse(self.route_table_cache)(scope, receive, send)
+            return
+
+        # TODO(simon): Use werkzeug route mapper to support variable path
+        if current_path not in self.route_table_cache:
+            error_message = ("Path {} not found. "
+                             "Please ping http://.../ for routing table"
+                             ).format(current_path)
+            await JSONResponse(
+                {
+                    "error": error_message
+                }, status_code=404)(scope, receive, send)
+            return
+
+        endpoint_name = self.route_table_cache[current_path]
+        http_body_bytes = await self.receive_http_body(scope, receive, send)
+
+        # get slo_ms before enqueuing the query
+        query_string = scope["query_string"].decode("ascii")
+        query_kwargs = parse_qs(query_string)
+        relative_slo_ms = query_kwargs.pop("relative_slo_ms", None)
+        absolute_slo_ms = query_kwargs.pop("absolute_slo_ms", None)
+        try:
+            relative_slo_ms = self._check_slo_ms(relative_slo_ms)
+            absolute_slo_ms = self._check_slo_ms(absolute_slo_ms)
+            if relative_slo_ms is not None and absolute_slo_ms is not None:
+                raise ValueError("Both relative and absolute slo's"
+                                 "cannot be specified.")
+        except ValueError as e:
+            await JSONResponse({"error": str(e)})(scope, receive, send)
+            return
+
+        # create objects necessary for enqueue
+        # enclosing http_body_bytes to list due to
+        # https://github.com/ray-project/ray/issues/6944
+        # TODO(alind):  remove list enclosing after issue is fixed
+        args = (scope, [http_body_bytes])
+        request_in_object = RequestMetadata(
+            endpoint_name,
+            TaskContext.Web,
+            relative_slo_ms=relative_slo_ms,
+            absolute_slo_ms=absolute_slo_ms)
+
+        actual_result = await (self.serve_global_state.init_or_get_router()
+                               .enqueue_request.remote(request_in_object,
+                                                       *args))
+        result = actual_result
+
+        if isinstance(result, ray.exceptions.RayTaskError):
+            await JSONResponse({
+                "error": "internal error, please use python API to debug"
+            })(scope, receive, send)
+        else:
+            await JSONResponse({"result": result})(scope, receive, send)
+
+
+@ray.remote
+class HTTPActor:
+    def __init__(self):
+        self.app = HTTPProxy()
+
+    def run(self, host="0.0.0.0", port=8000):
+        uvicorn.run(
+            self.app, host=host, port=port, lifespan="on", access_log=False)

python/ray/serve/task_runner.py

@@ -0,0 +1,215 @@
+import time
+import traceback
+
+import ray
+from ray.serve import context as serve_context
+from ray.serve.context import FakeFlaskRequest
+from collections import defaultdict
+from ray.serve.utils import parse_request_item
+from ray.serve.exceptions import RayServeException
+
+
+class TaskRunner:
+    """A simple class that runs a function.
+
+    The purpose of this class is to model what the most basic actor could be.
+    That is, a ray serve actor should implement the TaskRunner interface.
+    """
+
+    def __init__(self, func_to_run):
+        self.func = func_to_run
+
+    def __call__(self, *args, **kwargs):
+        return self.func(*args, **kwargs)
+
+
+def wrap_to_ray_error(exception):
+    """Utility method that catch and seal exceptions in execution"""
+    try:
+        # Raise and catch so we can access traceback.format_exc()
+        raise exception
+    except Exception as e:
+        traceback_str = ray.utils.format_error_message(traceback.format_exc())
+        return ray.exceptions.RayTaskError(str(e), traceback_str, e.__class__)
+
+
+class RayServeMixin:
+    """This mixin class adds the functionality to fetch from router queues.
+
+    Warning:
+        It assumes the main execution method is `__call__` of the user defined
+        class. This means that serve will call `your_instance.__call__` when
+        each request comes in. This behavior will be fixed in the future to
+        allow assigning artibrary methods.
+
+    Example:
+        >>> # Use ray.remote decorator and RayServeMixin
+        >>> # to make MyClass servable.
+        >>> @ray.remote
+            class RayServeActor(RayServeMixin, MyClass):
+                pass
+    """
+    _ray_serve_self_handle = None
+    _ray_serve_router_handle = None
+    _ray_serve_setup_completed = False
+    _ray_serve_dequeue_requester_name = None
+
+    # Work token can be unfullfilled from last iteration.
+    # This cache will be used to determine whether or not we should
+    # work on the same task as previous iteration or we are ready to
+    # move on.
+    _ray_serve_cached_work_token = None
+
+    _serve_metric_error_counter = 0
+    _serve_metric_latency_list = []
+
+    def _serve_metric(self):
+        # Make a copy of the latency list and clear current list
+        latency_lst = self._serve_metric_latency_list[:]
+        self._serve_metric_latency_list = []
+
+        my_name = self._ray_serve_dequeue_requester_name
+
+        return {
+            "{}_error_counter".format(my_name): {
+                "value": self._serve_metric_error_counter,
+                "type": "counter",
+            },
+            "{}_latency_s".format(my_name): {
+                "value": latency_lst,
+                "type": "list",
+            },
+        }
+
+    def _ray_serve_setup(self, my_name, router_handle, my_handle):
+        self._ray_serve_dequeue_requester_name = my_name
+        self._ray_serve_router_handle = router_handle
+        self._ray_serve_self_handle = my_handle
+        self._ray_serve_setup_completed = True
+
+    def _ray_serve_fetch(self):
+        assert self._ray_serve_setup_completed
+
+        self._ray_serve_router_handle.dequeue_request.remote(
+            self._ray_serve_dequeue_requester_name,
+            self._ray_serve_self_handle)
+
+    def invoke_single(self, request_item):
+        args, kwargs, is_web_context = parse_request_item(request_item)
+        serve_context.web = is_web_context
+        start_timestamp = time.time()
+
+        try:
+            result = self.__call__(*args, **kwargs)
+        except Exception as e:
+            result = wrap_to_ray_error(e)
+            self._serve_metric_error_counter += 1
+
+        self._serve_metric_latency_list.append(time.time() - start_timestamp)
+        return result
+
+    def invoke_batch(self, request_item_list):
+        # TODO(alind) : create no-http services. The enqueues
+        # from such services will always be TaskContext.Python.
+
+        # Assumption : all the requests in a bacth
+        # have same serve context.
+
+        # For batching kwargs are modified as follows -
+        # kwargs [Python Context] : key,val
+        # kwargs_list             : key, [val1,val2, ... , valn]
+        # or
+        # args[Web Context]       : val
+        # args_list               : [val1,val2, ...... , valn]
+        # where n (current batch size) <= max_batch_size of a backend
+
+        arg_list = []
+        kwargs_list = defaultdict(list)
+        context_flags = set()
+        batch_size = len(request_item_list)
+
+        for item in request_item_list:
+            args, kwargs, is_web_context = parse_request_item(item)
+            context_flags.add(is_web_context)
+
+            if is_web_context:
+                # Python context only have kwargs
+                flask_request = args[0]
+                arg_list.append(flask_request)
+            else:
+                # Web context only have one positional argument
+                for k, v in kwargs.items():
+                    kwargs_list[k].append(v)
+
+                # Set the flask request as a list to conform
+                # with batching semantics: when in batching
+                # mode, each argument it turned into list.
+                arg_list.append(FakeFlaskRequest())
+
+        try:
+            # check mixing of query context
+            # unified context needed
+            if len(context_flags) != 1:
+                raise RayServeException(
+                    "Batched queries contain mixed context. Please only send "
+                    "the same type of requests in batching mode.")
+
+            serve_context.web = context_flags.pop()
+            serve_context.batch_size = batch_size
+            # Flask requests are passed to __call__ as a list
+            arg_list = [arg_list]
+
+            start_timestamp = time.time()
+            result_list = self.__call__(*arg_list, **kwargs_list)
+
+            self._serve_metric_latency_list.append(time.time() -
+                                                   start_timestamp)
+            if (not isinstance(result_list,
+                               list)) or (len(result_list) != batch_size):
+                raise RayServeException("__call__ function "
+                                        "doesn't preserve batch-size. "
+                                        "Please return a list of result "
+                                        "with length equals to the batch "
+                                        "size.")
+            return result_list
+        except Exception as e:
+            wrapped_exception = wrap_to_ray_error(e)
+            self._serve_metric_error_counter += batch_size
+            return [wrapped_exception for _ in range(batch_size)]
+
+    def _ray_serve_call(self, request):
+        # check if work_item is a list or not
+        # if it is list: then batching supported
+        if not isinstance(request, list):
+            result = self.invoke_single(request)
+        else:
+            result = self.invoke_batch(request)
+
+        # re-assign to default values
+        serve_context.web = False
+        serve_context.batch_size = None
+
+        # Tell router that current actor is idle
+        self._ray_serve_fetch()
+
+        return result
+
+
+class TaskRunnerBackend(TaskRunner, RayServeMixin):
+    """A simple function serving backend
+
+    Note that this is not yet an actor. To make it an actor:
+
+    >>> @ray.remote
+        class TaskRunnerActor(TaskRunnerBackend):
+            pass
+
+    Note:
+    This class is not used in the actual ray serve system. It exists
+    for documentation purpose.
+    """
+
+
+@ray.remote
+class TaskRunnerActor(TaskRunnerBackend):
+    pass

python/ray/serve/tests/conftest.py

@@ -0,0 +1,28 @@
+import os
+import tempfile
+
+import pytest
+
+import ray
+from ray import serve
+
+
+@pytest.fixture(scope="session")
+def serve_instance():
+    _, new_db_path = tempfile.mkstemp(suffix=".test.db")
+    serve.init(
+        kv_store_path=new_db_path,
+        blocking=True,
+        ray_init_kwargs={"num_cpus": 36})
+    yield
+    os.remove(new_db_path)
+
+
+@pytest.fixture(scope="session")
+def ray_instance():
+    ray_already_initialized = ray.is_initialized()
+    if not ray_already_initialized:
+        ray.init(object_store_memory=int(1e8))
+    yield
+    if not ray_already_initialized:
+        ray.shutdown()

python/ray/serve/tests/test_api.py

@@ -0,0 +1,221 @@
+import time
+import pytest
+import requests
+
+from ray import serve
+from ray.serve import BackendConfig
+import ray
+from ray.serve.constants import NO_ROUTE_KEY
+
+
+def test_e2e(serve_instance):
+    serve.init()  # so we have access to global state
+    serve.create_endpoint("endpoint", "/api", blocking=True)
+    result = serve.api._get_global_state().route_table.list_service()
+    assert result["/api"] == "endpoint"
+
+    retry_count = 5
+    timeout_sleep = 0.5
+    while True:
+        try:
+            resp = requests.get("http://127.0.0.1:8000/", timeout=0.5).json()
+            assert resp == result
+            break
+        except Exception:
+            time.sleep(timeout_sleep)
+            timeout_sleep *= 2
+            retry_count -= 1
+            if retry_count == 0:
+                assert False, "Route table hasn't been updated after 3 tries."
+
+    def function(flask_request):
+        return "OK"
+
+    serve.create_backend(function, "echo:v1")
+    serve.link("endpoint", "echo:v1")
+
+    resp = requests.get("http://127.0.0.1:8000/api").json()["result"]
+    assert resp == "OK"
+
+
+def test_no_route(serve_instance):
+    serve.create_endpoint("noroute-endpoint", blocking=True)
+    global_state = serve.api._get_global_state()
+
+    result = global_state.route_table.list_service(include_headless=True)
+    assert result[NO_ROUTE_KEY] == ["noroute-endpoint"]
+
+    without_headless_result = global_state.route_table.list_service()
+    assert NO_ROUTE_KEY not in without_headless_result
+
+    def func(_, i=1):
+        return 1
+
+    serve.create_backend(func, "backend:1")
+    serve.link("noroute-endpoint", "backend:1")
+    service_handle = serve.get_handle("noroute-endpoint")
+    result = ray.get(service_handle.remote(i=1))
+    assert result == 1
+
+
+def test_scaling_replicas(serve_instance):
+    class Counter:
+        def __init__(self):
+            self.count = 0
+
+        def __call__(self, _):
+            self.count += 1
+            return self.count
+
+    serve.create_endpoint("counter", "/increment")
+
+    # Keep checking the routing table until /increment is populated
+    while "/increment" not in requests.get("http://127.0.0.1:8000/").json():
+        time.sleep(0.2)
+
+    b_config = BackendConfig(num_replicas=2)
+    serve.create_backend(Counter, "counter:v1", backend_config=b_config)
+    serve.link("counter", "counter:v1")
+
+    counter_result = []
+    for _ in range(10):
+        resp = requests.get("http://127.0.0.1:8000/increment").json()["result"]
+        counter_result.append(resp)
+
+    # If the load is shared among two replicas. The max result cannot be 10.
+    assert max(counter_result) < 10
+
+    b_config = serve.get_backend_config("counter:v1")
+    b_config.num_replicas = 1
+    serve.set_backend_config("counter:v1", b_config)
+
+    counter_result = []
+    for _ in range(10):
+        resp = requests.get("http://127.0.0.1:8000/increment").json()["result"]
+        counter_result.append(resp)
+    # Give some time for a replica to spin down. But majority of the request
+    # should be served by the only remaining replica.
+    assert max(counter_result) - min(counter_result) > 6
+
+
+def test_batching(serve_instance):
+    class BatchingExample:
+        def __init__(self):
+            self.count = 0
+
+        @serve.accept_batch
+        def __call__(self, flask_request, temp=None):
+            self.count += 1
+            batch_size = serve.context.batch_size
+            return [self.count] * batch_size
+
+    serve.create_endpoint("counter1", "/increment")
+
+    # Keep checking the routing table until /increment is populated
+    while "/increment" not in requests.get("http://127.0.0.1:8000/").json():
+        time.sleep(0.2)
+
+    # set the max batch size
+    b_config = BackendConfig(max_batch_size=5)
+    serve.create_backend(
+        BatchingExample, "counter:v11", backend_config=b_config)
+    serve.link("counter1", "counter:v11")
+
+    future_list = []
+    handle = serve.get_handle("counter1")
+    for _ in range(20):
+        f = handle.remote(temp=1)
+        future_list.append(f)
+
+    counter_result = ray.get(future_list)
+    # since count is only updated per batch of queries
+    # If there atleast one __call__ fn call with batch size greater than 1
+    # counter result will always be less than 20
+    assert max(counter_result) < 20
+
+
+def test_batching_exception(serve_instance):
+    class NoListReturned:
+        def __init__(self):
+            self.count = 0
+
+        @serve.accept_batch
+        def __call__(self, flask_request, temp=None):
+            batch_size = serve.context.batch_size
+            return batch_size
+
+    serve.create_endpoint("exception-test", "/noListReturned")
+    # set the max batch size
+    b_config = BackendConfig(max_batch_size=5)
+    serve.create_backend(
+        NoListReturned, "exception:v1", backend_config=b_config)
+    serve.link("exception-test", "exception:v1")
+
+    handle = serve.get_handle("exception-test")
+    with pytest.raises(ray.exceptions.RayTaskError):
+        assert ray.get(handle.remote(temp=1))
+
+
+def test_killing_replicas(serve_instance):
+    class Simple:
+        def __init__(self):
+            self.count = 0
+
+        def __call__(self, flask_request, temp=None):
+            return temp
+
+    serve.create_endpoint("simple", "/simple")
+    b_config = BackendConfig(num_replicas=3, num_cpus=2)
+    serve.create_backend(Simple, "simple:v1", backend_config=b_config)
+    global_state = serve.api._get_global_state()
+    old_replica_tag_list = global_state.backend_table.list_replicas(
+        "simple:v1")
+
+    bnew_config = serve.get_backend_config("simple:v1")
+    # change the config
+    bnew_config.num_cpus = 1
+    # set the config
+    serve.set_backend_config("simple:v1", bnew_config)
+    new_replica_tag_list = global_state.backend_table.list_replicas(
+        "simple:v1")
+    global_state.refresh_actor_handle_cache()
+    new_all_tag_list = list(global_state.actor_handle_cache.keys())
+
+    # the new_replica_tag_list must be subset of all_tag_list
+    assert set(new_replica_tag_list) <= set(new_all_tag_list)
+
+    # the old_replica_tag_list must not be subset of all_tag_list
+    assert not set(old_replica_tag_list) <= set(new_all_tag_list)
+
+
+def test_not_killing_replicas(serve_instance):
+    class BatchSimple:
+        def __init__(self):
+            self.count = 0
+
+        @serve.accept_batch
+        def __call__(self, flask_request, temp=None):
+            batch_size = serve.context.batch_size
+            return [1] * batch_size
+
+    serve.create_endpoint("bsimple", "/bsimple")
+    b_config = BackendConfig(num_replicas=3, max_batch_size=2)
+    serve.create_backend(BatchSimple, "bsimple:v1", backend_config=b_config)
+    global_state = serve.api._get_global_state()
+    old_replica_tag_list = global_state.backend_table.list_replicas(
+        "bsimple:v1")
+
+    bnew_config = serve.get_backend_config("bsimple:v1")
+    # change the config
+    bnew_config.max_batch_size = 5
+    # set the config
+    serve.set_backend_config("bsimple:v1", bnew_config)
+    new_replica_tag_list = global_state.backend_table.list_replicas(
+        "bsimple:v1")
+    global_state.refresh_actor_handle_cache()
+    new_all_tag_list = list(global_state.actor_handle_cache.keys())
+
+    # the old and new replica tag list should be identical
+    # and should be subset of all_tag_list
+    assert set(old_replica_tag_list) <= set(new_all_tag_list)
+    assert set(old_replica_tag_list) == set(new_replica_tag_list)

python/ray/serve/tests/test_metric.py

@@ -0,0 +1,74 @@
+import numpy as np
+import pytest
+
+import ray
+from ray.serve.metric import MetricMonitor
+
+
+@pytest.fixture(scope="session")
+def start_target_actor(ray_instance):
+    @ray.remote
+    class Target:
+        def __init__(self):
+            self.counter_value = 0
+
+        def _serve_metric(self):
+            self.counter_value += 1
+            return {
+                "latency_list": {
+                    "type": "list",
+                    # Generate 0 to 100 inclusive.
+                    # This means total of 101 items.
+                    "value": np.arange(101).tolist()
+                },
+                "counter": {
+                    "type": "counter",
+                    "value": self.counter_value
+                }
+            }
+
+        def get_counter_value(self):
+            return self.counter_value
+
+    yield Target.remote()
+
+
+def test_metric_gc(ray_instance, start_target_actor):
+    target_actor = start_target_actor
+    # this means when new scrapes are invoked, the
+    metric_monitor = MetricMonitor.remote(gc_window_seconds=0)
+    ray.get(metric_monitor.add_target.remote(target_actor))
+
+    ray.get(metric_monitor.scrape.remote())
+    df = ray.get(metric_monitor._get_dataframe.remote())
+    assert len(df) == 102
+
+    # Old metric sould be cleared. So only 1 counter + 101 list values left.
+    ray.get(metric_monitor.scrape.remote())
+    df = ray.get(metric_monitor._get_dataframe.remote())
+    assert len(df) == 102
+
+
+def test_metric_system(ray_instance, start_target_actor):
+    target_actor = start_target_actor
+
+    metric_monitor = MetricMonitor.remote()
+
+    ray.get(metric_monitor.add_target.remote(target_actor))
+
+    # Scrape once
+    ray.get(metric_monitor.scrape.remote())
+
+    percentiles = [50, 90, 95]
+    agg_windows_seconds = [60]
+    result = ray.get(
+        metric_monitor.collect.remote(percentiles, agg_windows_seconds))
+    real_counter_value = ray.get(target_actor.get_counter_value.remote())
+
+    expected_result = {
+        "counter": real_counter_value,
+        "latency_list_50th_perc_60_window": 50.0,
+        "latency_list_90th_perc_60_window": 90.0,
+        "latency_list_95th_perc_60_window": 95.0,
+    }
+    assert result == expected_result

python/ray/serve/tests/test_persistence.py

@@ -0,0 +1,33 @@
+import os
+import subprocess
+import tempfile
+
+import ray
+from ray import serve
+
+
+def test_new_driver(serve_instance):
+    script = """
+import ray
+ray.init(address="auto")
+
+from ray import serve
+serve.init()
+
+@serve.route("/driver")
+def driver(flask_request):
+    return "OK!"
+"""
+
+    with tempfile.NamedTemporaryFile(mode="w", delete=False) as f:
+        path = f.name
+        f.write(script)
+
+    proc = subprocess.Popen(["python", path])
+    return_code = proc.wait(timeout=10)
+    assert return_code == 0
+
+    handle = serve.get_handle("driver")
+    assert ray.get(handle.remote()) == "OK!"
+
+    os.remove(path)

python/ray/serve/tests/test_queue.py

@@ -0,0 +1,192 @@
+import asyncio
+
+import pytest
+import ray
+
+from ray.serve.policy import (
+    RandomPolicyQueue, RandomPolicyQueueActor, RoundRobinPolicyQueueActor,
+    PowerOfTwoPolicyQueueActor, FixedPackingPolicyQueueActor)
+from ray.serve.request_params import RequestMetadata
+
+pytestmark = pytest.mark.asyncio
+
+
+def make_task_runner_mock():
+    @ray.remote(num_cpus=0)
+    class TaskRunnerMock:
+        def __init__(self):
+            self.query = None
+            self.queries = []
+
+        async def _ray_serve_call(self, request_item):
+            self.query = request_item
+            self.queries.append(request_item)
+            return "DONE"
+
+        def get_recent_call(self):
+            return self.query
+
+        def get_all_calls(self):
+            return self.queries
+
+    return TaskRunnerMock.remote()
+
+
+@pytest.fixture(scope="session")
+def task_runner_mock_actor():
+    yield make_task_runner_mock()
+
+
+async def test_single_prod_cons_queue(serve_instance, task_runner_mock_actor):
+    q = RandomPolicyQueueActor.remote()
+    q.link.remote("svc", "backend")
+    q.dequeue_request.remote("backend", task_runner_mock_actor)
+
+    # Make sure we get the request result back
+    result = await q.enqueue_request.remote(RequestMetadata("svc", None), 1)
+    assert result == "DONE"
+
+    # Make sure it's the right request
+    got_work = await task_runner_mock_actor.get_recent_call.remote()
+    assert got_work.request_args[0] == 1
+    assert got_work.request_kwargs == {}
+
+
+async def test_slo(serve_instance, task_runner_mock_actor):
+    q = RandomPolicyQueueActor.remote()
+    await q.link.remote("svc", "backend")
+
+    all_request_sent = []
+    for i in range(10):
+        slo_ms = 1000 - 100 * i
+        all_request_sent.append(
+            q.enqueue_request.remote(
+                RequestMetadata("svc", None, relative_slo_ms=slo_ms), i))
+
+    for i in range(10):
+        await q.dequeue_request.remote("backend", task_runner_mock_actor)
+
+    await asyncio.gather(*all_request_sent)
+
+    i_should_be = 9
+    all_calls = await task_runner_mock_actor.get_all_calls.remote()
+    all_calls = all_calls[-10:]
+    for call in all_calls:
+        assert call.request_args[0] == i_should_be
+        i_should_be -= 1
+
+
+async def test_alter_backend(serve_instance, task_runner_mock_actor):
+    q = RandomPolicyQueueActor.remote()
+
+    await q.set_traffic.remote("svc", {"backend-1": 1})
+    await q.dequeue_request.remote("backend-1", task_runner_mock_actor)
+    await q.enqueue_request.remote(RequestMetadata("svc", None), 1)
+    got_work = await task_runner_mock_actor.get_recent_call.remote()
+    assert got_work.request_args[0] == 1
+
+    await q.set_traffic.remote("svc", {"backend-2": 1})
+    await q.dequeue_request.remote("backend-2", task_runner_mock_actor)
+    await q.enqueue_request.remote(RequestMetadata("svc", None), 2)
+    got_work = await task_runner_mock_actor.get_recent_call.remote()
+    assert got_work.request_args[0] == 2
+
+
+async def test_split_traffic_random(serve_instance, task_runner_mock_actor):
+    q = RandomPolicyQueueActor.remote()
+
+    await q.set_traffic.remote("svc", {"backend-1": 0.5, "backend-2": 0.5})
+    runner_1, runner_2 = [make_task_runner_mock() for _ in range(2)]
+    for _ in range(20):
+        await q.dequeue_request.remote("backend-1", runner_1)
+        await q.dequeue_request.remote("backend-2", runner_2)
+
+    # assume 50% split, the probability of all 20 requests goes to a
+    # single queue is 0.5^20 ~ 1-6
+    for _ in range(20):
+        await q.enqueue_request.remote(RequestMetadata("svc", None), 1)
+
+    got_work = [
+        await runner.get_recent_call.remote()
+        for runner in (runner_1, runner_2)
+    ]
+    assert [g.request_args[0] for g in got_work] == [1, 1]
+
+
+async def test_round_robin(serve_instance, task_runner_mock_actor):
+    q = RoundRobinPolicyQueueActor.remote()
+
+    await q.set_traffic.remote("svc", {"backend-1": 0.5, "backend-2": 0.5})
+    runner_1, runner_2 = [make_task_runner_mock() for _ in range(2)]
+
+    # NOTE: this is the only difference between the
+    # test_split_traffic_random and test_round_robin
+    for _ in range(10):
+        await q.dequeue_request.remote("backend-1", runner_1)
+        await q.dequeue_request.remote("backend-2", runner_2)
+
+    for _ in range(20):
+        await q.enqueue_request.remote(RequestMetadata("svc", None), 1)
+
+    got_work = [
+        await runner.get_recent_call.remote()
+        for runner in (runner_1, runner_2)
+    ]
+    assert [g.request_args[0] for g in got_work] == [1, 1]
+
+
+async def test_fixed_packing(serve_instance):
+    packing_num = 4
+    q = FixedPackingPolicyQueueActor.remote(packing_num=packing_num)
+    await q.set_traffic.remote("svc", {"backend-1": 0.5, "backend-2": 0.5})
+
+    runner_1, runner_2 = (make_task_runner_mock() for _ in range(2))
+    # both the backends will get equal number of queries
+    # as it is packed round robin
+    for _ in range(packing_num):
+        await q.dequeue_request.remote("backend-1", runner_1)
+        await q.dequeue_request.remote("backend-2", runner_2)
+
+    for backend, runner in zip(["1", "2"], [runner_1, runner_2]):
+        for _ in range(packing_num):
+            input_value = "should-go-to-backend-{}".format(backend)
+            await q.enqueue_request.remote(
+                RequestMetadata("svc", None), input_value)
+            all_calls = await runner.get_all_calls.remote()
+            for call in all_calls:
+                assert call.request_args[0] == input_value
+
+
+async def test_power_of_two_choices(serve_instance):
+    q = PowerOfTwoPolicyQueueActor.remote()
+    enqueue_futures = []
+
+    # First, fill the queue for backend-1 with 3 requests
+    await q.set_traffic.remote("svc", {"backend-1": 1.0})
+    for _ in range(3):
+        future = q.enqueue_request.remote(RequestMetadata("svc", None), "1")
+        enqueue_futures.append(future)
+
+    # Then, add a new backend, this backend should be filled next
+    await q.set_traffic.remote("svc", {"backend-1": 0.5, "backend-2": 0.5})
+    for _ in range(2):
+        future = q.enqueue_request.remote(RequestMetadata("svc", None), "2")
+        enqueue_futures.append(future)
+
+    runner_1, runner_2 = (make_task_runner_mock() for _ in range(2))
+    for _ in range(3):
+        await q.dequeue_request.remote("backend-1", runner_1)
+        await q.dequeue_request.remote("backend-2", runner_2)
+
+    await asyncio.gather(*enqueue_futures)
+
+    assert len(await runner_1.get_all_calls.remote()) == 3
+    assert len(await runner_2.get_all_calls.remote()) == 2
+
+
+async def test_queue_remove_replicas(serve_instance):
+    temp_actor = make_task_runner_mock()
+    q = RandomPolicyQueue()
+    await q.dequeue_request("backend", temp_actor)
+    await q.remove_and_destory_replica("backend", temp_actor)
+    assert q.worker_queues["backend"].qsize() == 0

python/ray/serve/tests/test_routing.py

@@ -0,0 +1,54 @@
+import os
+import tempfile
+
+from ray.serve.kv_store_service import (InMemoryKVStore, RayInternalKVStore,
+                                        SQLiteKVStore)
+
+
+def test_default_in_memory_kv():
+    kv = InMemoryKVStore("")
+    kv.put("1", 2)
+    assert kv.get("1") == 2
+    kv.put("1", 3)
+    assert kv.get("1") == 3
+    assert kv.as_dict() == {"1": 3}
+
+
+def test_ray_interal_kv(ray_instance):
+    kv = RayInternalKVStore("")
+    kv.put("1", 2)
+    assert kv.get("1") == 2
+    kv.put("1", 3)
+    assert kv.get("1") == 3
+    assert kv.as_dict() == {"1": 3}
+
+    kv = RayInternalKVStore("othernamespace")
+    kv.put("1", 2)
+    assert kv.get("1") == 2
+    kv.put("1", 3)
+    assert kv.get("1") == 3
+    assert kv.as_dict() == {"1": 3}
+
+
+def test_sqlite_kv():
+    _, path = tempfile.mkstemp()
+
+    # Test get
+    kv = SQLiteKVStore("routing_table", db_path=path)
+    kv.put("/api", "api-endpoint")
+    assert kv.get("/api") == "api-endpoint"
+    assert kv.get("not-exist") is None
+
+    # Test namespace
+    kv2 = SQLiteKVStore("other_table", db_path=path)
+    kv2.put("/api", "api-endpoint-two")
+    assert kv2.get("/api") == "api-endpoint-two"
+
+    # Test as dict
+    assert kv.as_dict() == {"/api": "api-endpoint"}
+
+    # Test override
+    kv.put("/api", "api-new")
+    assert kv.get("/api") == "api-new"
+
+    os.remove(path)

python/ray/serve/tests/test_serve.py

@@ -0,0 +1,15 @@
+import pytest
+from pathlib import Path
+import sys
+
+if __name__ == "__main__":
+    curr_dir = Path(__file__).parent
+    test_paths = curr_dir.rglob("test_*.py")
+    sorted_path = sorted(map(lambda path: str(path.absolute()), test_paths))
+    serve_tests_files = list(sorted_path)
+
+    print("Testing the following files")
+    for test_file in serve_tests_files:
+        print("->", test_file.split("/")[-1])
+
+    sys.exit(pytest.main(["-v", "-s"] + serve_tests_files))

python/ray/serve/tests/test_task_runner.py

@@ -0,0 +1,99 @@
+import pytest
+
+import ray
+import ray.serve.context as context
+from ray.serve.policy import RoundRobinPolicyQueueActor
+from ray.serve.task_runner import (RayServeMixin, TaskRunner, TaskRunnerActor,
+                                   wrap_to_ray_error)
+from ray.serve.request_params import RequestMetadata
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_runner_basic():
+    def echo(i):
+        return i
+
+    r = TaskRunner(echo)
+    assert r(1) == 1
+
+
+async def test_runner_wraps_error():
+    wrapped = wrap_to_ray_error(Exception())
+    assert isinstance(wrapped, ray.exceptions.RayTaskError)
+
+
+async def test_runner_actor(serve_instance):
+    q = RoundRobinPolicyQueueActor.remote()
+
+    def echo(flask_request, i=None):
+        return i
+
+    CONSUMER_NAME = "runner"
+    PRODUCER_NAME = "prod"
+
+    runner = TaskRunnerActor.remote(echo)
+    runner._ray_serve_setup.remote(CONSUMER_NAME, q, runner)
+    runner._ray_serve_fetch.remote()
+
+    q.link.remote(PRODUCER_NAME, CONSUMER_NAME)
+
+    for query in [333, 444, 555]:
+        query_param = RequestMetadata(PRODUCER_NAME,
+                                      context.TaskContext.Python)
+        result = await q.enqueue_request.remote(query_param, i=query)
+        assert result == query
+
+
+async def test_ray_serve_mixin(serve_instance):
+    q = RoundRobinPolicyQueueActor.remote()
+
+    CONSUMER_NAME = "runner-cls"
+    PRODUCER_NAME = "prod-cls"
+
+    class MyAdder:
+        def __init__(self, inc):
+            self.increment = inc
+
+        def __call__(self, flask_request, i=None):
+            return i + self.increment
+
+    @ray.remote
+    class CustomActor(MyAdder, RayServeMixin):
+        pass
+
+    runner = CustomActor.remote(3)
+
+    runner._ray_serve_setup.remote(CONSUMER_NAME, q, runner)
+    runner._ray_serve_fetch.remote()
+
+    q.link.remote(PRODUCER_NAME, CONSUMER_NAME)
+
+    for query in [333, 444, 555]:
+        query_param = RequestMetadata(PRODUCER_NAME,
+                                      context.TaskContext.Python)
+        result = await q.enqueue_request.remote(query_param, i=query)
+        assert result == query + 3
+
+
+async def test_task_runner_check_context(serve_instance):
+    q = RoundRobinPolicyQueueActor.remote()
+
+    def echo(flask_request, i=None):
+        # Accessing the flask_request without web context should throw.
+        return flask_request.args["i"]
+
+    CONSUMER_NAME = "runner"
+    PRODUCER_NAME = "producer"
+
+    runner = TaskRunnerActor.remote(echo)
+
+    runner._ray_serve_setup.remote(CONSUMER_NAME, q, runner)
+    runner._ray_serve_fetch.remote()
+
+    q.link.remote(PRODUCER_NAME, CONSUMER_NAME)
+    query_param = RequestMetadata(PRODUCER_NAME, context.TaskContext.Python)
+    result_oid = q.enqueue_request.remote(query_param, i=42)
+
+    with pytest.raises(ray.exceptions.RayTaskError):
+        await result_oid

python/ray/serve/tests/test_util.py

@@ -0,0 +1,9 @@
+import json
+
+from ray.serve.utils import BytesEncoder
+
+
+def test_bytes_encoder():
+    data_before = {"inp": {"nest": b"bytes"}}
+    data_after = {"inp": {"nest": "bytes"}}
+    assert json.loads(json.dumps(data_before, cls=BytesEncoder)) == data_after

python/ray/serve/utils.py

@@ -0,0 +1,112 @@
+import json
+import logging
+import random
+import string
+import time
+import io
+import os
+
+import requests
+from pygments import formatters, highlight, lexers
+from ray.serve.context import FakeFlaskRequest, TaskContext
+from ray.serve.http_util import build_flask_request
+import itertools
+
+
+def expand(l):
+    """
+    Implements a nested flattening of a list.
+    Example:
+    >>> serve.utils.expand([1,2,[3,4,5],6])
+    [1,2,3,4,5,6]
+    >>> serve.utils.expand(["a", ["b", "c"], "d", ["e", "f"]])
+    ["a", "b", "c", "d", "e", "f"]
+    """
+    return list(
+        itertools.chain.from_iterable(
+            [x if isinstance(x, list) else [x] for x in l]))
+
+
+def parse_request_item(request_item):
+    if request_item.request_context == TaskContext.Web:
+        is_web_context = True
+        asgi_scope, body_bytes = request_item.request_args
+
+        # http_body_bytes enclosed in list due to
+        # https://github.com/ray-project/ray/issues/6944
+        # TODO(alind):  remove list enclosing after issue is fixed
+        flask_request = build_flask_request(asgi_scope,
+                                            io.BytesIO(body_bytes[0]))
+        args = (flask_request, )
+        kwargs = {}
+    else:
+        is_web_context = False
+        args = (FakeFlaskRequest(), )
+        kwargs = request_item.request_kwargs
+
+    return args, kwargs, is_web_context
+
+
+def _get_logger():
+    logger = logging.getLogger("ray.serve")
+    # TODO(simon): Make logging level configurable.
+    if os.environ.get("SERVE_LOG_DEBUG"):
+        logger.setLevel(logging.DEBUG)
+    else:
+        logger.setLevel(logging.INFO)
+    return logger
+
+
+logger = _get_logger()
+
+
+class BytesEncoder(json.JSONEncoder):
+    """Allow bytes to be part of the JSON document.
+
+    BytesEncoder will walk the JSON tree and decode bytes with utf-8 codec.
+
+    Example:
+    >>> json.dumps({b'a': b'c'}, cls=BytesEncoder)
+    '{"a":"c"}'
+    """
+
+    def default(self, o):  # pylint: disable=E0202
+        if isinstance(o, bytes):
+            return o.decode("utf-8")
+        return super().default(o)
+
+
+def pformat_color_json(d):
+    """Use pygments to pretty format and colroize dictionary"""
+    formatted_json = json.dumps(d, sort_keys=True, indent=4)
+
+    colorful_json = highlight(formatted_json, lexers.JsonLexer(),
+                              formatters.TerminalFormatter())
+
+    return colorful_json
+
+
+def block_until_http_ready(http_endpoint, num_retries=5, backoff_time_s=1):
+    http_is_ready = False
+    retries = num_retries
+
+    while not http_is_ready:
+        try:
+            resp = requests.get(http_endpoint)
+            assert resp.status_code == 200
+            http_is_ready = True
+        except Exception:
+            pass
+
+        # Exponential backoff
+        time.sleep(backoff_time_s)
+        backoff_time_s *= 2
+
+        retries -= 1
+        if retries == 0:
+            raise Exception(
+                "HTTP server not ready after {} retries.".format(num_retries))
+
+
+def get_random_letters(length=6):
+    return "".join(random.choices(string.ascii_letters, k=length))