code: def test_basic(ray_start_regular_shared):
      @ray.remote(concurrency_groups={"io": 2, "compute": 4})
      class AsyncActor:
          def __init__(self):
              self.eventloop_f1 = None
              self.eventloop_f2 = None
              self.eventloop_f3 = None
              self.eventloop_f4 = None
              self.default_eventloop = get_or_create_event_loop()
  
          @ray.method(concurrency_group="io")
          async def f1(self):
              self.eventloop_f1 = get_or_create_event_loop()
              return threading.current_thread().ident
  
          @ray.method(concurrency_group="io")
          def f2(self):
              self.eventloop_f2 = get_or_create_event_loop()
              return threading.current_thread().ident
  
          @ray.method(concurrency_group="compute")
          def f3(self):
              self.eventloop_f3 = get_or_create_event_loop()
              return threading.current_thread().ident
  
          @ray.method(concurrency_group="compute")
          def f4(self):
              self.eventloop_f4 = get_or_create_event_loop()
              return threading.current_thread().ident
  
          def f5(self):
              # If this method is executed in default eventloop.
              assert (
                  get_or_create_event_loop() == self.default_eventloop
              )
              return threading.current_thread().ident
  
          @ray.method(concurrency_group="io")
          def do_assert(self):
              if self.eventloop_f1 != self.eventloop_f2:
                  return False
              if self.eventloop_f3 != self.eventloop_f4:
                  return False
              if self.eventloop_f1 == self.eventloop_f3:
                  return False
              if self.eventloop_f1 == self.eventloop_f4:
                  return False
              return True
  
      ###############################################
      a = AsyncActor.remote()
      f1_thread_id = ray.get(
          a.f1.remote()
      )  # executed in the "io" group.
      f2_thread_id = ray.get(
          a.f2.remote()
      )  # executed in the "io" group.
      f3_thread_id = ray.get(
          a.f3.remote()
      )  # executed in the "compute" group.
      f4_thread_id = ray.get(
          a.f4.remote()
      )  # executed in the "compute" group.
  
      assert f1_thread_id == f2_thread_id
      assert f3_thread_id == f4_thread_id
      assert f1_thread_id != f3_thread_id
  
      assert ray.get(a.do_assert.remote())
  
      assert ray.get(a.f5.remote())  # executed in the default group.
  
      # It also has the ability to specify it at runtime.
      # This task will be invoked in the `compute` thread pool.
      result = ray.get(
          a.f2.options(concurrency_group="compute").remote()
      )
      assert result == f3_thread_id
  
file_path: ray/python/ray/tests/test_concurrency_group.py
imports: import asyncio
  import sys
  import threading
  import pytest
  import ray
  import time
  from ray._private.utils import get_or_create_event_loop
label_desc: Use image node_id 1 for calling a global function with example usage: test_basic(ray_start_regular_shared) and returns: threading, threading, threading, threading, threading, True, False, False, False, False