[Asyncio] Allow Async_API to init when loop is running (#5323)

python/ray/experimental/async_api.py

@@ -1,8 +1,10 @@
 # Note: asyncio is only compatible with Python 3
 
 import asyncio
+
 import ray
 from ray.experimental.async_plasma import PlasmaProtocol, PlasmaEventHandler
+from ray.services import logger
 
 handler = None
 transport = None
@@ -19,17 +21,18 @@ async def _async_init():
         handler = PlasmaEventHandler(loop, worker)
         transport, protocol = await loop.create_connection(
             lambda: PlasmaProtocol(worker.plasma_client, handler), sock=rsock)
+        logger.debug("AsyncPlasma Connection Created!")
 
 
 def init():
     """
     Initialize synchronously.
     """
+    assert ray.is_initialized(), "Please call ray.init before async_api.init"
+
     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.")
+        asyncio.ensure_future(_async_init())
     else:
         asyncio.get_event_loop().run_until_complete(_async_init())