code: def test_asyncio_actor_high_concurrency(ray_start_regular_shared):
      # This tests actor can handle concurrency above recursionlimit.
  
      @ray.remote
      class AsyncConcurrencyBatcher:
          def __init__(self, batch_size):
              self.batch = []
              self.event = asyncio.Event()
              self.batch_size = batch_size
  
          async def add(self, x):
              self.batch.append(x)
              if len(self.batch) >= self.batch_size:
                  self.event.set()
              else:
                  await self.event.wait()
              return sorted(self.batch)
  
      batch_size = sys.getrecursionlimit() * 4
      actor = AsyncConcurrencyBatcher.options(
          max_concurrency=batch_size * 2
      ).remote(batch_size)
      result = ray.get([actor.add.remote(i) for i in range(batch_size)])
      assert result[0] == list(range(batch_size))
      assert result[-1] == list(range(batch_size))
  
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 4 for calling a global function with example usage: test_asyncio_actor_high_concurrency(ray_start_regular_shared) and returns: sorted