code: def test_async_actor_concurrent(shutdown_only):
      """Verify the async actor generator tasks are concurrent."""
  
      @ray.remote
      class A:
          async def gen(self):
              for i in range(5):
                  await asyncio.sleep(1)
                  yield i
  
      a = A.remote()
  
      async def co():
          async for ref in a.gen.remote():
              print(await ref)
  
      async def main():
          await asyncio.gather(co(), co(), co())
  
      s = time.time()
      asyncio.run(main())
      assert 4.5 < time.time() - s < 6.5
  
file_path: ray/python/ray/tests/test_streaming_generator_2.py
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import gc
  import ray
  from ray.experimental.state.api import list_actors
label_desc: Use image node_id 7 for calling a global function with example usage: test_async_actor_concurrent(shutdown_only) without return types