code: def test_async_actor_cg_have_bounded_num_of_threads(shutdown_only):
      ray.init()
  
      @ray.remote(concurrency_groups={"io": 2, "compute": 4})
      class A:
          async def get_my_threads(self):
              return my_threads()
  
          @ray.method(concurrency_group="io")
          async def io_add(self, i, j):
              return i + j
  
          @ray.method(concurrency_group="compute")
          async def compute_add(self, i, j):
              return i + j
  
          async def default_add(self, i, j):
              return i + j
  
      # Spawns a lot of workers, each making 1 call to A.
      @ray.remote
      def fibonacci_cg(a, i):
          if i < 2:
              return 1
          f1 = fibonacci_cg.remote(a, i - 1)
          f2 = fibonacci_cg.remote(a, i - 2)
          assert ray.get(a.io_add.remote(1, 2)) == 3
          assert ray.get(a.compute_add.remote(4, 5)) == 9
          return ray.get(a.default_add.remote(f1, f2))
  
      a = A.options(max_concurrency=2).remote()
  
      prev_threads = ray.get(a.get_my_threads.remote())
  
      assert ray.get(fibonacci_cg.remote(a, 1)) == 1
      now_threads = ray.get(a.get_my_threads.remote())
      assert_threads_are_bounded(prev_threads, now_threads)
  
      # Creates a lot of workers sending to actor
      assert ray.get(fibonacci_cg.remote(a, 10)) == 89
      now_threads = ray.get(a.get_my_threads.remote())
      assert_threads_are_bounded(prev_threads, now_threads)
  
file_path: ray/python/ray/tests/test_actor_bounded_threads.py
imports: import sys
  import os
  import ray
  import logging
  from typing import Dict
  from collections import Counter
  import pytest
label_desc: Use image node_id 6 for calling a global function with example usage: test_async_actor_cg_have_bounded_num_of_threads(shutdown_only) and returns: ray, my_threads, unknown, unknown, unknown, int