Ignore blocking ray.wait if timeout is zero (#6891)

python/ray/tests/py3_test.py

@@ -254,10 +254,16 @@ def test_asyncio_actor_async_get(ray_start_regular_shared):
     def remote_task():
         return 1
 
+    plasma_object = ray.put(2)
+
     @ray.remote
     class AsyncGetter:
         async def get(self):
             return await remote_task.remote()
 
+        async def plasma_get(self):
+            return await plasma_object
+
     getter = AsyncGetter.options(is_asyncio=True).remote()
     assert ray.get(getter.get.remote()) == 1
+    assert ray.get(getter.plasma_get.remote()) == 2

python/ray/worker.py

@@ -1573,9 +1573,9 @@ def wait(object_ids, num_returns=1, timeout=None):
     """
     worker = global_worker
 
-    if hasattr(
-            worker,
-            "core_worker") and worker.core_worker.current_actor_is_asyncio():
+    if hasattr(worker,
+               "core_worker") and worker.core_worker.current_actor_is_asyncio(
+               ) and timeout != 0:
         raise RayError("Using blocking ray.wait inside async method. "
                        "This blocks the event loop. Please use `await` "
                        "on object id with asyncio.wait. ")