code: def test_parent_task_id_concurrent_actor(
      shutdown_only, actor_concurrency
  ):
      # Test tasks runs in user started thread from actors have a parent_task_id
      # as the actor's creation task.
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      def run_task_in_thread(name, i):
          def thd_task():
              @ray.remote
              def thd_task():
                  pass
  
              ray.get(thd_task.options(name=f"{name}_{i}").remote())
  
          thd = threading.Thread(target=thd_task)
          thd.start()
          thd.join()
  
      @ray.remote
      class AsyncActor:
          async def main_task(self, i):
              run_task_in_thread("async_thd_task", i)
  
      @ray.remote
      class ThreadedActor:
          def main_task(self, i):
              run_task_in_thread("threaded_thd_task", i)
  
      def verify(actor_method_name, actor_class_name):
          tasks = list_tasks()
          print(tasks)
          expect_parent_task_id = None
          actual_parent_task_id = None
          for task in tasks:
              if f"{actor_class_name}.__init__" in task["name"]:
                  expect_parent_task_id = task["task_id"]
  
          assert expect_parent_task_id is not None
          for task in tasks:
              if f"{actor_method_name}" in task["name"]:
                  actual_parent_task_id = task["parent_task_id"]
                  assert (
                      expect_parent_task_id == actual_parent_task_id
                  ), task
  
          return True
  
      async_actor = AsyncActor.options(
          max_concurrency=actor_concurrency
      ).remote()
      ray.get([async_actor.main_task.remote(i) for i in range(20)])
      wait_for_condition(
          verify,
          actor_class_name="AsyncActor",
          actor_method_name="async_thd_task",
      )
  
      thd_actor = ThreadedActor.options(
          max_concurrency=actor_concurrency
      ).remote()
      ray.get([thd_actor.main_task.remote(i) for i in range(20)])
      wait_for_condition(
          verify,
          actor_class_name="ThreadedActor",
          actor_method_name="threaded_thd_task",
      )
  
imports: from collections import defaultdict
  from typing import Dict
  import os
  import pytest
  import sys
  import threading
  import time
  from ray._private.state_api_test_utils import verify_failed_task
  from ray.exceptions import RuntimeEnvSetupError
  from ray.runtime_env import RuntimeEnv
  import ray
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import raw_metrics, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_tasks
  from ray._private.worker import RayContext
label_desc: Use image node_id 9 for calling a global function with example usage: test_parent_task_id_concurrent_actor(shutdown_only, actor_concurrency) and returns: True