[Serve] Refactor ray.serve for compatibility

.travis.yml

@@ -135,7 +135,7 @@ matrix:
 
 install:
   - eval `python $TRAVIS_BUILD_DIR/ci/travis/determine_tests_to_run.py`
-  - if [ $RAY_CI_TUNE_AFFECTED != "1" ] && [ $RAY_CI_RLLIB_AFFECTED != "1" ] && [ $RAY_CI_PYTHON_AFFECTED != "1" ]; then exit; fi
+  - if [ $RAY_CI_TUNE_AFFECTED != "1" ] && [ $RAY_CI_RLLIB_AFFECTED != "1" ] && [ $RAY_CI_SERVE_AFFECTED != "1" ] && [ $RAY_CI_PYTHON_AFFECTED != "1" ]; then exit; fi
 
   - ./ci/suppress_output ./ci/travis/install-bazel.sh
   - ./ci/suppress_output ./ci/travis/install-dependencies.sh
@@ -173,6 +173,9 @@ script:
   - if [ $RAY_CI_RLLIB_AFFECTED == "1" ]; then ./ci/suppress_output python python/ray/rllib/tests/test_optimizers.py; fi
   - if [ $RAY_CI_RLLIB_AFFECTED == "1" ]; then ./ci/suppress_output python python/ray/rllib/tests/test_evaluators.py; fi
 
+  # ray serve tests
+  - if [ $RAY_CI_SERVE_AFFECTED == "1" ] && [ $RAY_CI_PY3 == "1" ]; then ./ci/suppress_output python -m pytest -v --durations=5 python/ray/experimental/serve/tests; fi
+
   # ray tests
   # Python3.5+ only. Otherwise we will get `SyntaxError` regardless of how we set the tester.
   - if [ $RAY_CI_PYTHON_AFFECTED == "1" ]; then python -c 'import sys;exit(sys.version_info>=(3,5))' || python -m pytest -v --durations=5 --timeout=300 python/ray/experimental/test/async_test.py; fi

ci/travis/determine_tests_to_run.py

@@ -5,6 +5,8 @@
 
 import os
 import subprocess
+import sys
+from pprint import pformat
 
 
 def list_changed_files(commit_range):
@@ -26,24 +28,37 @@ def list_changed_files(commit_range):
     return [s.strip() for s in out.decode().splitlines() if s is not None]
 
 
+def print_and_log(s):
+    print(s)
+    sys.stderr.write(s)
+    sys.stderr.write("\n")
+
+
 if __name__ == "__main__":
 
     RAY_CI_TUNE_AFFECTED = 0
     RAY_CI_RLLIB_AFFECTED = 0
+    RAY_CI_SERVE_AFFECTED = 0
     RAY_CI_JAVA_AFFECTED = 0
     RAY_CI_PYTHON_AFFECTED = 0
     RAY_CI_LINUX_WHEELS_AFFECTED = 0
     RAY_CI_MACOS_WHEELS_AFFECTED = 0
 
+    RAY_CI_PY3 = 1 if sys.version_info >= (3, 5) else 0
+
     if os.environ["TRAVIS_EVENT_TYPE"] == "pull_request":
 
         files = list_changed_files(os.environ["TRAVIS_COMMIT_RANGE"].replace(
             "...", ".."))
 
         skip_prefix_list = [
-            "doc/", "examples/", "dev/", "docker/", "kubernetes/", "site/"
+            "doc/", "examples/", "dev/", "docker/", "kubernetes/", "site/",
+            "ci/", ".travis.yml"
         ]
 
+        sys.stderr.write("Files Changed\n")
+        sys.stderr.write(pformat(files))
+
         for changed_file in files:
             if changed_file.startswith("python/ray/tune/"):
                 RAY_CI_TUNE_AFFECTED = 1
@@ -54,6 +69,10 @@ def list_changed_files(commit_range):
                 RAY_CI_RLLIB_AFFECTED = 1
                 RAY_CI_LINUX_WHEELS_AFFECTED = 1
                 RAY_CI_MACOS_WHEELS_AFFECTED = 1
+            elif changed_file.startswith("python/ray/experimental/serve"):
+                RAY_CI_SERVE_AFFECTED = 1
+                RAY_CI_LINUX_WHEELS_AFFECTED = 1
+                RAY_CI_MACOS_WHEELS_AFFECTED = 1
             elif changed_file.startswith("python/"):
                 RAY_CI_TUNE_AFFECTED = 1
                 RAY_CI_RLLIB_AFFECTED = 1
@@ -89,11 +108,18 @@ def list_changed_files(commit_range):
         RAY_CI_LINUX_WHEELS_AFFECTED = 1
         RAY_CI_MACOS_WHEELS_AFFECTED = 1
 
-    print("export RAY_CI_TUNE_AFFECTED={}".format(RAY_CI_TUNE_AFFECTED))
-    print("export RAY_CI_RLLIB_AFFECTED={}".format(RAY_CI_RLLIB_AFFECTED))
-    print("export RAY_CI_JAVA_AFFECTED={}".format(RAY_CI_JAVA_AFFECTED))
-    print("export RAY_CI_PYTHON_AFFECTED={}".format(RAY_CI_PYTHON_AFFECTED))
-    print("export RAY_CI_LINUX_WHEELS_AFFECTED={}"
-          .format(RAY_CI_LINUX_WHEELS_AFFECTED))
-    print("export RAY_CI_MACOS_WHEELS_AFFECTED={}"
-          .format(RAY_CI_MACOS_WHEELS_AFFECTED))
+    print_and_log("export RAY_CI_PY3={}".format(RAY_CI_PY3))
+    print_and_log(
+        "export RAY_CI_TUNE_AFFECTED={}".format(RAY_CI_TUNE_AFFECTED))
+    print_and_log(
+        "export RAY_CI_RLLIB_AFFECTED={}".format(RAY_CI_RLLIB_AFFECTED))
+    print_and_log(
+        "export RAY_CI_SERVE_AFFECTED={}".format(RAY_CI_SERVE_AFFECTED))
+    print_and_log(
+        "export RAY_CI_JAVA_AFFECTED={}".format(RAY_CI_JAVA_AFFECTED))
+    print_and_log(
+        "export RAY_CI_PYTHON_AFFECTED={}".format(RAY_CI_PYTHON_AFFECTED))
+    print_and_log("export RAY_CI_LINUX_WHEELS_AFFECTED={}".format(
+        RAY_CI_LINUX_WHEELS_AFFECTED))
+    print_and_log("export RAY_CI_MACOS_WHEELS_AFFECTED={}".format(
+        RAY_CI_MACOS_WHEELS_AFFECTED))

ci/travis/install-dependencies.sh

@@ -34,7 +34,8 @@ elif [[ "$PYTHON" == "3.5" ]] && [[ "$platform" == "linux" ]]; then
   bash miniconda.sh -b -p $HOME/miniconda
   export PATH="$HOME/miniconda/bin:$PATH"
   pip install -q scipy tensorflow cython==0.29.0 gym opencv-python-headless pyyaml pandas==0.24.2 requests \
-    feather-format lxml openpyxl xlrd py-spy setproctitle pytest-timeout flaky networkx tabulate psutil aiohttp
+    feather-format lxml openpyxl xlrd py-spy setproctitle pytest-timeout flaky networkx tabulate psutil aiohttp \
+    flask
 elif [[ "$PYTHON" == "2.7" ]] && [[ "$platform" == "macosx" ]]; then
   # Install miniconda.
   wget https://repo.continuum.io/miniconda/Miniconda2-4.5.4-MacOSX-x86_64.sh -O miniconda.sh -nv
@@ -48,7 +49,8 @@ elif [[ "$PYTHON" == "3.5" ]] && [[ "$platform" == "macosx" ]]; then
   bash miniconda.sh -b -p $HOME/miniconda
   export PATH="$HOME/miniconda/bin:$PATH"
   pip install -q cython==0.29.0 tensorflow gym opencv-python-headless pyyaml pandas==0.24.2 requests \
-    feather-format lxml openpyxl xlrd py-spy setproctitle pytest-timeout flaky networkx tabulate psutil aiohttp
+    feather-format lxml openpyxl xlrd py-spy setproctitle pytest-timeout flaky networkx tabulate psutil aiohttp \
+    flask
 elif [[ "$LINT" == "1" ]]; then
   sudo apt-get update
   sudo apt-get install -y build-essential curl unzip

python/ray/experimental/serve/README.rst

@@ -4,18 +4,6 @@ Ray Serve Module
 ``ray.experimental.serve`` is a module for publishing your actors to
 interact with outside world.
 
-Use Case
---------
-
-Serve machine learning model
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-Scalable anayltics query
-~~~~~~~~~~~~~~~~~~~~~~~~
-
-Composible pipelines
-~~~~~~~~~~~~~~~~~~~~
-
 Architecture
 ------------
 

python/ray/experimental/serve/__init__.py

@@ -10,8 +10,8 @@
 
 import sys
 
-assert sys.version_info >= (3, ), (
-    "ray.experimental.serve is a python3 only library")
+assert sys.version_info >= (
+    3, 5), "ray.experimental.serve is a python3 only library"
 
 from ray.experimental.serve.router import (DeadlineAwareRouter,
                                            SingleQuery)  # noqa: E402

python/ray/experimental/serve/example_actors.py

@@ -0,0 +1,108 @@
+"""
+This file contains example ray servable actors. These actors
+are used for testing as well as demoing purpose.
+"""
+
+from __future__ import absolute_import, division, print_function
+
+import time
+
+import numpy as np
+
+import ray
+from ray.experimental.serve import RayServeMixin, batched_input
+
+
+@ray.remote
+class VectorizedAdder(RayServeMixin):
+    """Actor that adds scaler_increment to input batch.
+
+    result = np.array(input_batch) + scaler_increment
+    """
+
+    def __init__(self, scaler_increment):
+        self.inc = scaler_increment
+
+    @batched_input
+    def __call__(self, input_batch):
+        arr = np.array(input_batch)
+        arr += self.inc
+        return arr.tolist()
+
+
+@ray.remote
+class ScalerAdder(RayServeMixin):
+    """Actor that adds a scaler_increment to a single input."""
+
+    def __init__(self, scaler_increment):
+        self.inc = scaler_increment
+
+    def __call__(self, input_scaler):
+        return input_scaler + self.inc
+
+
+@ray.remote
+class VectorDouble(RayServeMixin):
+    """Actor that doubles the batched input."""
+
+    @batched_input
+    def __call__(self, batched_vectors):
+        matrix = np.array(batched_vectors)
+        matrix *= 2
+        return [v.tolist() for v in matrix]
+
+
+@ray.remote
+class Counter(RayServeMixin):
+    """Return the query id. Used for testing router."""
+
+    def __init__(self):
+        self.counter = 0
+
+    def __call__(self, batched_input):
+        self.counter += 1
+        return self.counter
+
+
+@ray.remote
+class CustomCounter(RayServeMixin):
+    """Return the query id. Used for testing `serve_method` signature."""
+
+    serve_method = "count"
+
+    @batched_input
+    def count(self, input_batch):
+        return [1 for _ in range(len(input_batch))]
+
+
+@ray.remote
+class SleepOnFirst(RayServeMixin):
+    """Sleep on the first request, return batch size.
+
+    Used for testing the DeadlineAwareRouter.
+    """
+
+    def __init__(self, sleep_time):
+        self.nap_time = sleep_time
+
+    @batched_input
+    def __call__(self, input_batch):
+        time.sleep(self.nap_time)
+        return [len(input_batch) for _ in range(len(input_batch))]
+
+
+@ray.remote
+class SleepCounter(RayServeMixin):
+    """Sleep on input argument seconds, return the query id.
+
+    Used to test the DeadlineAwareRouter.
+    """
+
+    def __init__(self):
+        self.counter = 0
+
+    def __call__(self, inp):
+        time.sleep(inp)
+
+        self.counter += 1
+        return self.counter