code: def test_asyncio_actor_concurrency(ray_start_regular_shared):
      @ray.remote
      class RecordOrder:
          def __init__(self):
              self.history = []
  
          async def do_work(self):
              self.history.append("STARTED")
              # Force a context switch
              await asyncio.sleep(0)
              self.history.append("ENDED")
  
          def get_history(self):
              return self.history
  
      num_calls = 10
  
      a = RecordOrder.options(max_concurrency=1).remote()
      ray.get([a.do_work.remote() for _ in range(num_calls)])
      history = ray.get(a.get_history.remote())
  
      # We only care about ordered start-end-start-end sequence because
      # coroutines may be executed out of enqueued order.
      answer = []
      for _ in range(num_calls):
          for status in ["STARTED", "ENDED"]:
              answer.append(status)
  
      assert history == answer
  
imports: import asyncio
  import os
  import sys
  import threading
  import time
  import pytest
  import ray
  from ray._private.client_mode_hook import client_mode_should_convert
  from ray._private.test_utils import SignalActor, kill_actor_and_wait_for_failure, wait_for_condition, wait_for_pid_to_exit
label_desc: Use image node_id 3 for calling a global function with example usage: test_asyncio_actor_concurrency(ray_start_regular_shared) and returns: self