code: def test_placement_group_load_report(ray_start_cluster):
      cluster = ray_start_cluster
      # Add a head node that doesn't have gpu resource.
      cluster.add_node(num_cpus=4)
  
      global_state_accessor = make_global_state_accessor(
          ray.init(address=cluster.address)
      )
  
      class PgLoadChecker:
          def nothing_is_ready(self):
              resource_usage = self._read_resource_usage()
              if not resource_usage:
                  return False
              if resource_usage.HasField("placement_group_load"):
                  pg_load = resource_usage.placement_group_load
                  return len(pg_load.placement_group_data) == 2
              return False
  
          def only_first_one_ready(self):
              resource_usage = self._read_resource_usage()
              if not resource_usage:
                  return False
              if resource_usage.HasField("placement_group_load"):
                  pg_load = resource_usage.placement_group_load
                  return len(pg_load.placement_group_data) == 1
              return False
  
          def two_infeasible_pg(self):
              resource_usage = self._read_resource_usage()
              if not resource_usage:
                  return False
              if resource_usage.HasField("placement_group_load"):
                  pg_load = resource_usage.placement_group_load
                  return len(pg_load.placement_group_data) == 2
              return False
  
          def _read_resource_usage(self):
              message = global_state_accessor.get_all_resource_usage()
              if message is None:
                  return False
  
              resource_usage = (
                  gcs_utils.ResourceUsageBatchData.FromString(message)
              )
              return resource_usage
  
      checker = PgLoadChecker()
  
      # Create 2 placement groups that are infeasible.
      pg_feasible = ray.util.placement_group([{"A": 1}])
      pg_infeasible = ray.util.placement_group([{"B": 1}])
      _, unready = ray.wait(
          [pg_feasible.ready(), pg_infeasible.ready()], timeout=0
      )
      assert len(unready) == 2
      wait_for_condition(checker.nothing_is_ready)
  
      # Add a node that makes pg feasible. Make sure load include this change.
      cluster.add_node(resources={"A": 1})
      ray.get(pg_feasible.ready())
      wait_for_condition(checker.only_first_one_ready)
      # Create one more infeasible pg and make sure load is properly updated.
      pg_infeasible_second = ray.util.placement_group([{"C": 1}])
      _, unready = ray.wait([pg_infeasible_second.ready()], timeout=0)
      assert len(unready) == 1
      wait_for_condition(checker.two_infeasible_pg)
      global_state_accessor.disconnect()
  
imports: import os
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray._private.ray_constants
  from ray._raylet import GcsClient
  from ray.core.generated import autoscaler_pb2
  from ray._private.test_utils import convert_actor_state, make_global_state_accessor, wait_for_condition
label_desc: Use image node_id 11 for calling a global function with example usage: test_placement_group_load_report(ray_start_cluster) and returns: False, False, False, resource_usage, False, unknown, False, unknown, False, unknown, False