Experimental asyncio support

.travis.yml

@@ -146,6 +146,9 @@ script:
   - python -m pytest -v python/ray/rllib/test/test_optimizers.py
   - python -m pytest -v python/ray/rllib/test/test_evaluators.py
 
+  # Python3.5+ only. Otherwise we will get `SyntaxError` regardless of how we set the tester.
+  - python -c 'import sys;exit(sys.version_info>=(3,5))' || python -m pytest -v python/ray/experimental/test/async_test.py
+
   - python -m pytest -v python/ray/test/test_global_state.py
   - python -m pytest -v python/ray/test/test_queue.py
   - python -m pytest -v python/ray/test/test_ray_init.py

cmake/Modules/ArrowExternalProject.cmake

@@ -15,10 +15,10 @@
 #  - PLASMA_SHARED_LIB
 
 set(arrow_URL https://github.com/apache/arrow.git)
-# The PR for this commit is https://github.com/apache/arrow/pull/3061. We
+# The PR for this commit is https://github.com/apache/arrow/pull/3093. We
 # include the link here to make it easier to find the right commit because
 # Arrow often rewrites git history and invalidates certain commits.
-set(arrow_TAG a667fca3b71772886bb2595986266d2039823dcc)
+set(arrow_TAG 187b98ed338d4995317dae9efd19870c532192cb)
 
 set(ARROW_INSTALL_PREFIX ${CMAKE_CURRENT_BINARY_DIR}/external/arrow-install)
 set(ARROW_HOME ${ARROW_INSTALL_PREFIX})

doc/source/async_api.rst

@@ -0,0 +1,87 @@
+Async API (Experimental)
+========================
+
+Since Python 3.5, it is possible to write concurrent code using the ``async/await`` `syntax <https://docs.python.org/3/library/asyncio.html>`__.
+
+This document describes Ray's support for asyncio, which enables integration with popular async frameworks (e.g., aiohttp, aioredis, etc.) for high performance web and prediction serving.
+
+Starting Ray
+------------
+
+You must initialize Ray first.
+
+Please refer to `Starting Ray`_ for instructions.
+
+.. _`Starting Ray`: http://ray.readthedocs.io/en/latest/tutorial.html#starting-ray
+
+
+Converting Ray objects into asyncio futures
+-------------------------------------------
+
+Ray object IDs can be converted into asyncio futures with ``ray.experimental.async_api``.
+
+.. code-block:: python
+
+  import asyncio
+  import time
+  import ray
+  from ray.experimental import async_api
+
+  @ray.remote
+  def f():
+      time.sleep(1)
+      return {'key1': ['value']}
+
+  ray.init()
+  future = async_api.as_future(f.remote())
+  asyncio.get_event_loop().run_until_complete(future)  # {'key1': ['value']}
+
+
+.. autofunction:: ray.experimental.async_api.as_future
+
+
+Example Usage
+-------------
+
++----------------------------------------+-----------------------------------------------------+
+| **Basic Python**                       | **Distributed with Ray**                            |
++----------------------------------------+-----------------------------------------------------+
+| .. code-block:: python                 | .. code-block:: python                              |
+|                                        |                                                     |
+|   # Execute f serially.                |   # Execute f in parallel.                          |
+|                                        |                                                     |
+|                                        |                                                     |
+|   def f():                             |   @ray.remote                                       |
+|     time.sleep(1)                      |   def f():                                          |
+|     return 1                           |       time.sleep(1)                                 |
+|                                        |       return 1                                      |
+|                                        |                                                     |
+|                                        |   ray.init()                                        |
+|   results = [f() for i in range(4)]    |   results = ray.get([f.remote() for i in range(4)]) |
++----------------------------------------+-----------------------------------------------------+
+| **Async Python**                       | **Async Ray**                                       |
++----------------------------------------+-----------------------------------------------------+
+| .. code-block:: python                 | .. code-block:: python                              |
+|                                        |                                                     |
+|   # Execute f asynchronously.          |   # Execute f asynchronously with Ray/asyncio.      |
+|                                        |                                                     |
+|                                        |   from ray.experimental import async_api            |
+|                                        |                                                     |
+|                                        |   @ray.remote                                       |
+|   async def f():                       |   def f():                                          |
+|       await asyncio.sleep(1)           |       time.sleep(1)                                 |
+|       return 1                         |       return 1                                      |
+|                                        |                                                     |
+|                                        |   ray.init()                                        |
+|   loop = asyncio.get_event_loop()      |   loop = asyncio.get_event_loop()                   |
+|   tasks = [f() for i in range(4)]      |   tasks = [async_api.as_future(f.remote())          |
+|                                        |            for i in range(4)]                       |
+|   results = loop.run_until_complete(   |   results = loop.run_until_complete(                |
+|       asyncio.gather(tasks))           |       asyncio.gather(tasks))                        |
++----------------------------------------+-----------------------------------------------------+
+
+
+Known Issues
+------------
+
+Async API support is experimental, and we are working to improve its performance. Please `let us know <https://github.com/ray-project/ray/issues>`__ any issues you encounter.

doc/source/index.rst

@@ -65,6 +65,7 @@ Ray comes with libraries that accelerate deep learning and reinforcement learnin
    actors.rst
    using-ray-with-gpus.rst
    webui.rst
+   async_api.rst
 
 .. toctree::
    :maxdepth: 1

python/ray/experimental/async_api.py

@@ -0,0 +1,62 @@
+# Note: asyncio is only compatible with Python 3
+
+import asyncio
+import ray
+from ray.experimental.async_plasma import PlasmaProtocol, PlasmaEventHandler
+
+handler = None
+transport = None
+protocol = None
+
+
+async def _async_init():
+    global handler, transport, protocol
+    if handler is None:
+        worker = ray.worker.global_worker
+        loop = asyncio.get_event_loop()
+        worker.plasma_client.subscribe()
+        rsock = worker.plasma_client.get_notification_socket()
+        handler = PlasmaEventHandler(loop, worker)
+        transport, protocol = await loop.create_connection(
+            lambda: PlasmaProtocol(worker.plasma_client, handler), sock=rsock)
+
+
+def init():
+    """
+    Initialize synchronously.
+    """
+    loop = asyncio.get_event_loop()
+    if loop.is_running():
+        raise Exception("You must initialize the Ray async API by calling "
+                        "async_api.init() or async_api.as_future(obj) before "
+                        "the event loop starts.")
+    else:
+        asyncio.get_event_loop().run_until_complete(_async_init())
+
+
+def as_future(object_id):
+    """Turn an object_id into a Future object.
+
+    Args:
+        object_id: A Ray object_id.
+
+    Returns:
+        PlasmaObjectFuture: A future object that waits the object_id.
+    """
+    if handler is None:
+        init()
+    return handler.as_future(object_id)
+
+
+def shutdown():
+    """Manually shutdown the async API.
+
+    Cancels all related tasks and all the socket transportation.
+    """
+    global handler, transport, protocol
+    if handler is not None:
+        handler.close()
+        transport.close()
+        handler = None
+        transport = None
+        protocol = None