code: def test_actor_max_concurrency(ray_start_regular_shared):
      """
      Test that an actor of max_concurrency=N should only run
      N tasks at most concurrently.
      """
      CONCURRENCY = 3
  
      @ray.remote
      class ConcurentActor:
          def __init__(self):
              self.threads = set()
  
          def call(self):
              # Record the current thread that runs this function.
              self.threads.add(threading.current_thread())
  
          def get_num_threads(self):
              return len(self.threads)
  
      @ray.remote
      def call(actor):
          for _ in range(CONCURRENCY * 100):
              ray.get(actor.call.remote())
          return
  
      actor = ConcurentActor.options(
          max_concurrency=CONCURRENCY
      ).remote()
      # Start many concurrent tasks that will call the actor many times.
      ray.get([call.remote(actor) for _ in range(CONCURRENCY * 10)])
  
      # Check that the number of threads shouldn't be greater than CONCURRENCY.
      assert ray.get(actor.get_num_threads.remote()) <= CONCURRENCY
  
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 18 for calling a global function with example usage: test_actor_max_concurrency(ray_start_regular_shared) and returns: len