code: def test_parent_task_correct_concurrent_async_actor(shutdown_only):
      """Make sure when there are concurrent async tasks
      the parent -> children task ids are properly mapped.
      """
      sig = SignalActor.remote()
  
      @ray.remote
      def child(sig):
          ray.get(sig.wait.remote())
  
      @ray.remote
      class AsyncActor:
          async def f(self, sig):
              refs = [child.remote(sig) for _ in range(2)]
              core_worker = (
                  ray._private.worker.global_worker.core_worker
              )
              expected = {ref.task_id().hex() for ref in refs}
              task_id = ray.get_runtime_context().task_id
              children_task_ids = (
                  core_worker.get_pending_children_task_ids(task_id)
              )
              actual = {task_id.hex() for task_id in children_task_ids}
              await sig.wait.remote()
              ray.get(refs)
              return actual, expected
  
      a = AsyncActor.remote()
      # Run 3 concurrent tasks.
      refs = [a.f.remote(sig) for _ in range(20)]
      # 3 concurrent task will finish.
      ray.get(sig.send.remote())
      # Verify children task mapping is correct.
      result = ray.get(refs)
      for actual, expected in result:
          assert actual, expected
  
imports: import datetime
  import os
  import random
  import sys
  import tempfile
  import numpy
  import pytest
  import ray
  from ray import cloudpickle
  from ray._private import ray_constants
  from ray._private.test_utils import client_test_enabled, wait_for_condition, wait_for_pid_to_exit
  from ray.actor import ActorClassInheritanceException
  from ray.tests.client_test_utils import create_remote_signal_actor
  from ray._private.test_utils import SignalActor
  import setproctitle
label_desc: Use image node_id 48 for calling a global function with example usage: test_parent_task_correct_concurrent_async_actor(shutdown_only) and returns: actual,  expected