code: def test_actor_concurrent(ray_start_regular_shared):
      @ray.remote
      class Batcher:
          def __init__(self):
              self.batch = []
              self.event = threading.Event()
  
          def add(self, x):
              self.batch.append(x)
              if len(self.batch) >= 3:
                  self.event.set()
              else:
                  self.event.wait()
              return sorted(self.batch)
  
      a = Batcher.options(max_concurrency=3).remote()
      x1 = a.add.remote(1)
      x2 = a.add.remote(2)
      x3 = a.add.remote(3)
      r1 = ray.get(x1)
      r2 = ray.get(x2)
      r3 = ray.get(x3)
      assert r1 == [1, 2, 3]
      assert r1 == r2 == r3
  
imports: import array
  import logging
  import os
  import random
  import subprocess
  import sys
  import tempfile
  import threading
  import time
  from unittest.mock import MagicMock, patch
  import pytest
  from ray._private.ray_constants import KV_NAMESPACE_FUNCTION_TABLE
  from ray._private.test_utils import client_test_enabled
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.exceptions import GetTimeoutError, RayTaskError
  from ray.tests.client_test_utils import create_remote_signal_actor
label_desc: Use image node_id 17 for calling a global function with example usage: test_actor_concurrent(ray_start_regular_shared) and returns: sorted