code: def async_actor_helper():
      """Run a Ray async actor and check the spans produced."""
  
      @ray.remote
      class AsyncActor:
          # multiple invocation of this method can be running in
          # the event loop at the same time
          async def run_concurrent(self):
              await asyncio.sleep(2)  # concurrent workload here
  
      actor = AsyncActor.remote()
      ray.get([actor.run_concurrent.remote() for _ in range(4)])
  
      span_list = get_span_list()
      assert len(span_list) == 10
  
      # The spans could show up in a different order, so just check that
      # all spans are as expected
      span_names = get_span_dict(span_list)
      assert span_names == {
          "async_actor_helper.<locals>.AsyncActor.__init__ ray.remote": 1,
          "async_actor_helper.<locals>.AsyncActor.run_concurrent ray.remote": 4,
          "AsyncActor.__init__ ray.remote_worker": 1,
          "AsyncActor.run_concurrent ray.remote_worker": 4,
      }, span_names
  
imports: import asyncio
  import glob
  import json
  import os
  import pytest
  import shutil
  from unittest.mock import patch
  import ray
  from ray._private.test_utils import check_call_ray
  from ray.util.tracing.setup_local_tmp_tracing import spans_dir
label_desc: Use image node_id 9 for calling a global function with example usage: async_actor_helper() without return types