code: def test_actor_stats_async_actor_generator(ray_start_regular):
      signal = SignalActor.remote()
  
      @ray.remote
      class AysncActor:
          async def func(self):
              await signal.wait.remote()
              yield ray.get_runtime_context()._get_actor_call_stats()
  
      actor = AysncActor.options(max_concurrency=3).remote()
      gens = [actor.func.remote() for _ in range(6)]
      time.sleep(1)
      signal.send.remote()
      results = []
      for gen in gens:
          for ref in gen:
              results.append(ray.get(ref))
      assert (
          max(
              result["AysncActor.func"]["running"] for result in results
          )
          == 3
      )
      assert (
          max(
              result["AysncActor.func"]["pending"] for result in results
          )
          == 3
      )
  
imports: import ray
  import os
  import signal
  import time
  import sys
  import pytest
  import warnings
  from ray._private.test_utils import SignalActor
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray.util.state import list_tasks
  from ray._private.test_utils import wait_for_condition
label_desc: Use image node_id 9 for calling a global function with example usage: test_actor_stats_async_actor_generator(ray_start_regular) without return types