Ray pipeline

python/ray/experimental/internal_kv.py

@@ -22,6 +22,12 @@ def _internal_kv_get(key):
     return worker.redis_client.hget(key, "value")
 
 
+def _internal_kv_exists(key):
+    worker = ray.worker.get_global_worker()
+    if worker.mode == ray.worker.LOCAL_MODE:
+        return key in _local
+    return worker.redis_client.hexists(key,"value")
+
 def _internal_kv_put(key, value, overwrite=False):
     """Globally associates a value with a given binary key.
 

python/ray/experimental/serve/__init__.py

@@ -2,11 +2,11 @@
 if sys.version_info < (3, 0):
     raise ImportError("serve is Python 3 only.")
 
-from ray.experimental.serve.api import (init, create_backend, create_endpoint,
-                                        link, split, rollback, get_handle,
-                                        global_state)  # noqa: E402
+from ray.experimental.serve.api import (init,provision_pipeline, create_backend, create_endpoint_pipeline,create_no_http_service,
+                                        add_service_dependencies,link_service, get_handle,
+                                        global_state,get_service_dependencies,add_service)  # noqa: E402
 
 __all__ = [
-    "init", "create_backend", "create_endpoint", "link", "split", "rollback",
-    "get_handle", "global_state"
+    "init","provision_pipeline", "create_backend", "create_endpoint_pipeline","create_no_http_service","add_service_dependencies", "link_service",
+    "get_handle" ,"global_state","get_service_dependencies","add_service"
 ]

python/ray/experimental/serve/api.py

@@ -37,7 +37,7 @@ def init(blocking=False, object_store_memory=int(1e8)):
         global_state.wait_until_http_ready()
 
 
-def create_endpoint(endpoint_name, route_expression, blocking=True):
+def create_endpoint_pipeline(pipeline_name, route_expression, blocking=True):
     """Create a service endpoint given route_expression.
 
     Args:
@@ -48,14 +48,18 @@ def create_endpoint(endpoint_name, route_expression, blocking=True):
         blocking (bool): If true, the function will wait for service to be
             registered before returning
     """
+    assert pipeline_name in global_state.provisioned_services
+
     future = global_state.kv_store_actor_handle.register_service.remote(
-        route_expression, endpoint_name)
+        route_expression, pipeline_name)
     if blocking:
         ray.get(future)
-    global_state.registered_endpoints.add(endpoint_name)
+    global_state.registered_endpoints.add(pipeline_name)
 
+def create_no_http_service(service_name):
+    global_state.registered_services.add(service_name)
 
-def create_backend(func_or_class, backend_tag, *actor_init_args):
+def create_backend(func_or_class, backend_tag, num_gpu,*actor_init_args):
     """Create a backend using func_or_class and assign backend_tag.
 
     Args:
@@ -71,7 +75,7 @@ def create_backend(func_or_class, backend_tag, *actor_init_args):
     elif inspect.isclass(func_or_class):
         # Python inheritance order is right-to-left. We put RayServeMixin
         # on the left to make sure its methods are not overriden.
-        @ray.remote
+        @ray.remote(num_gpus=num_gpu)
         class CustomActor(RayServeMixin, func_or_class):
             pass
 
@@ -89,8 +93,45 @@ class CustomActor(RayServeMixin, func_or_class):
 
     global_state.registered_backends.add(backend_tag)
 
+# def add_service_to_pipeline(pipeline_name,service_name,blocking=True):
+#     assert service_name in global_state.registered_services
+#     # assert pipeline_name in global_state.registered_endpoints
+
+#     future = global_state.kv_store_actor_handle_pipeline.add_node.remote(pipeline_name,service_name)
+#     if blocking:
+#         ray.get(future)
+
+def add_service_dependencies(pipeline_name,service_name_1,service_name_2,blocking=True):
+    assert service_name_1 in global_state.registered_services
+    assert service_name_2 in global_state.registered_services
+    assert pipeline_name not in global_state.provisioned_services
+
+    future = global_state.kv_store_actor_handle_pipeline.add_edge.remote(pipeline_name,service_name_1,service_name_2)
+    if blocking:
+        ray.get(future)
+
+def provision_pipeline(pipeline_name,blocking=True) :
+    assert pipeline_name not in global_state.provisioned_services
+    future = global_state.kv_store_actor_handle_pipeline.provision.remote(pipeline_name)
+    if blocking : 
+        ray.get(future)
+
+    global_state.provisioned_services.add(pipeline_name)
+
 
-def link(endpoint_name, backend_tag):
+def get_service_dependencies(pipeline_name):
+    assert pipeline_name in global_state.provisioned_services
+    future = global_state.kv_store_actor_handle_pipeline.get_dependency.remote(pipeline_name)
+    return ray.get(future)
+def add_service(pipeline_name,service_name,blocking = True):
+    assert pipeline_name not in global_state.provisioned_services
+    assert service_name in global_state.registered_services
+    future = global_state.kv_store_actor_handle_pipeline.add_node.remote(pipeline_name,service_name)
+    if blocking : 
+        ray.get(future)
+
+
+def link_service(service_name, backend_tag):
     """Associate a service endpoint with backend tag.
 
     Example:
@@ -102,75 +143,76 @@ def link(endpoint_name, backend_tag):
 
     >>> serve.split("service-name", {"backend:v1": 1.0})
     """
-    assert endpoint_name in global_state.registered_endpoints
+    assert service_name in global_state.registered_services
+    assert backend_tag in global_state.registered_backends
 
-    global_state.router_actor_handle.link.remote(endpoint_name, backend_tag)
-    global_state.policy_action_history[endpoint_name].append({backend_tag: 1})
+    global_state.router_actor_handle.link.remote(service_name, backend_tag)
+    global_state.policy_action_history[service_name].append({backend_tag: 1})
 
 
-def split(endpoint_name, traffic_policy_dictionary):
-    """Associate a service endpoint with traffic policy.
+# def split(endpoint_name, traffic_policy_dictionary):
+#     """Associate a service endpoint with traffic policy.
 
-    Example:
+#     Example:
 
-    >>> serve.split("service-name", {
-        "backend:v1": 0.5,
-        "backend:v2": 0.5
-    })
+#     >>> 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.
-    """
+#     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.
+#     """
 
-    # Perform dictionary checks
-    assert endpoint_name in global_state.registered_endpoints
+#     # Perform dictionary checks
+#     assert endpoint_name in global_state.registered_endpoints
 
-    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.registered_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)
+#     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.registered_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.router_actor_handle.set_traffic.remote(
-        endpoint_name, traffic_policy_dictionary)
-    global_state.policy_action_history[endpoint_name].append(
-        traffic_policy_dictionary)
+#     global_state.router_actor_handle.set_traffic.remote(
+#         endpoint_name, traffic_policy_dictionary)
+#     global_state.policy_action_history[endpoint_name].append(
+#         traffic_policy_dictionary)
 
 
-def rollback(endpoint_name):
-    """Rollback a traffic policy decision.
+# def rollback(endpoint_name):
+#     """Rollback a traffic policy decision.
 
-    Args:
-        endpoint_name (str): A registered service endpoint.
-    """
-    assert endpoint_name in global_state.registered_endpoints
-    action_queues = global_state.policy_action_history[endpoint_name]
-    cur_policy, prev_policy = action_queues[-1], action_queues[-2]
+#     Args:
+#         endpoint_name (str): A registered service endpoint.
+#     """
+#     assert endpoint_name in global_state.registered_endpoints
+#     action_queues = global_state.policy_action_history[endpoint_name]
+#     cur_policy, prev_policy = action_queues[-1], action_queues[-2]
 
-    logger.warning("""
-Current traffic policy is:
-{cur_policy}
+#     logger.warning("""
+# Current traffic policy is:
+# {cur_policy}
 
-Will rollback to:
-{prev_policy}
-""".format(
-        cur_policy=pformat_color_json(cur_policy),
-        prev_policy=pformat_color_json(prev_policy)))
+# Will rollback to:
+# {prev_policy}
+# """.format(
+#         cur_policy=pformat_color_json(cur_policy),
+#         prev_policy=pformat_color_json(prev_policy)))
 
-    action_queues.pop()
-    global_state.router_actor_handle.set_traffic.remote(
-        endpoint_name, prev_policy)
+#     action_queues.pop()
+#     global_state.router_actor_handle.set_traffic.remote(
+#         endpoint_name, prev_policy)
 
 
-def get_handle(endpoint_name):
+def get_handle(pipeline_name):
     """Retrieve RayServeHandle for service endpoint to invoke it from Python.
 
     Args:
@@ -179,9 +221,9 @@ def get_handle(endpoint_name):
     Returns:
         RayServeHandle
     """
-    assert endpoint_name in global_state.registered_endpoints
+    assert pipeline_name in global_state.provisioned_services
 
     # Delay import due to it's dependency on global_state
     from ray.experimental.serve.handle import RayServeHandle
 
-    return RayServeHandle(global_state.router_actor_handle, endpoint_name)
+    return RayServeHandle(global_state.kv_store_actor_handle_pipeline,global_state.router_actor_handle, pipeline_name)

python/ray/experimental/serve/constants.py

@@ -1,2 +1,2 @@
 #: The interval which http server refreshes its routing table
-HTTP_ROUTER_CHECKER_INTERVAL_S = 2
+HTTP_ROUTER_CHECKER_INTERVAL_S = 1

python/ray/experimental/serve/examples/echo.py → python/ray/experimental/serve/echo_pipeline.py

@@ -1,7 +1,3 @@
-"""
-Example service that prints out http context.
-"""
-
 import time
 
 import requests
@@ -10,19 +6,25 @@
 from ray.experimental.serve.utils import pformat_color_json
 
 
-def echo(context):
+def echo1(context):
+	context["query_string"] += 'FROM MODEL1 -> '
+    return context
+def echo2(context):
+	context += 'FROM MODEL2 -> '
     return context
-
 
 serve.init(blocking=True)
 
-serve.create_endpoint("my_endpoint", "/echo", blocking=True)
+serve.create_endpoint_pipeline("pipeline1", "/echo", blocking=True)
+
 serve.create_backend(echo, "echo:v1")
-serve.link("my_endpoint", "echo:v1")
+serve.create_no_http_service("serve1")
+serve.link_service("serve1", "echo:v1")
+serve.add_service_to_pipeline("pipeline1","serve1")
 
 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)
+    time.sleep(2)