code: def test_actor_scheduling_not_block_with_placement_group(
      ray_start_cluster,
  ):
      """Tests the scheduling of lots of actors will not be blocked
      when using placement groups.
  
      For more detailed information please refer to:
      https://github.com/ray-project/ray/issues/15801.
      """
  
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1)
      ray.init(address=cluster.address)
  
      @ray.remote(num_cpus=1)
      class A:
          def ready(self):
              pass
  
      actor_num = 1000
      pgs = [
          ray.util.placement_group([{"CPU": 1}])
          for _ in range(actor_num)
      ]
      actors = [
          A.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              )
          ).remote()
          for pg in pgs
      ]
      refs = [actor.ready.remote() for actor in actors]
  
      expected_created_num = 1
  
      def is_actor_created_number_correct():
          ready, not_ready = ray.wait(
              refs, num_returns=len(refs), timeout=1
          )
          return len(ready) == expected_created_num
  
      def is_pg_created_number_correct():
          created_pgs = [
              pg
              for _, pg in ray.util.placement_group_table().items()
              if pg["state"] == "CREATED"
          ]
          return len(created_pgs) == expected_created_num
  
      wait_for_condition(is_pg_created_number_correct, timeout=3)
      wait_for_condition(
          is_actor_created_number_correct,
          timeout=30,
          retry_interval_ms=0,
      )
  
      # NOTE: we don't need to test all the actors create successfully.
      for _ in range(20):
          expected_created_num += 1
          cluster.add_node(num_cpus=1)
  
          wait_for_condition(is_pg_created_number_correct, timeout=10)
          # Make sure the node add event will cause a waiting actor
          # to create successfully in time.
          wait_for_condition(
              is_actor_created_number_correct,
              timeout=30,
              retry_interval_ms=0,
          )
  
imports: import sys
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  import ray.experimental.internal_kv
  from ray._private.ray_constants import DEBUG_AUTOSCALING_ERROR, DEBUG_AUTOSCALING_STATUS
  from ray.autoscaler._private.constants import AUTOSCALER_UPDATE_INTERVAL_S
  from ray._private.test_utils import convert_actor_state, generate_system_config_map, is_placement_group_removed, kill_actor_and_wait_for_failure, reset_autoscaler_v2_enabled_cache, run_string_as_driver, wait_for_condition
  from ray.autoscaler._private.commands import debug_status
  from ray.exceptions import RaySystemError
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 11 for calling a global function with example usage: test_actor_scheduling_not_block_with_placement_group(ray_start_cluster) and returns: unknown, unknown