asyncio_ray.py

import asyncio
import ray

ray.init()

@ray.remote
class AsyncWorker:
    async def do_work(self):
        print("Started")
        await asyncio.sleep(0.5) # simulate network I/O
        print("Ended")
        return "Done!"

worker = AsyncWorker.remote()

async def main():
    ray_future = worker.do_work.remote()
    result = await ray_future
    assert result == "Done!"

    many_task_futures = [worker.do_work.remote() for _ in range(20)]
    result = await asyncio.gather(*many_task_futures)

asyncio.get_event_loop().run_until_complete(main())


## Should Output
# 2020-02-11 12:00:52,577	INFO resource_spec.py:212 -- Starting Ray with 24.02 GiB memory available for workers and up to 12.03 GiB for objects. You can adjust these settings with ray.init(memory=<bytes>, object_store_memory=<bytes>).
# 2020-02-11 12:00:52,978	INFO services.py:1082 -- View the Ray dashboard at localhost:8265

# (pid=83974) Started
# … 0.5 seconds later
# (pid=83974) Ended

# (pid=83974) Started # repeats 20 times
# … 0.5 seconds later
# (pid=83974) Ended # repeats 20 times
# ...