code: def test_actor_resource_demand(shutdown_only):
      ray.shutdown()
      cluster = ray.init(num_cpus=3)
      global_state_accessor = make_global_state_accessor(cluster)
  
      @ray.remote(num_cpus=2)
      class Actor:
          def foo(self):
              return "ok"
  
      a = Actor.remote()
      ray.get(a.foo.remote())
      time.sleep(1)
  
      message = global_state_accessor.get_all_resource_usage()
      resource_usages = gcs_utils.ResourceUsageBatchData.FromString(
          message
      )
  
      # The actor is scheduled so there should be no more demands left.
      assert (
          len(resource_usages.resource_load_by_shape.resource_demands)
          == 0
      )
  
      @ray.remote(num_cpus=80)
      class Actor2:
          pass
  
      actors = []
      actors.append(Actor2.remote())
      time.sleep(1)
  
      # This actor cannot be scheduled.
      message = global_state_accessor.get_all_resource_usage()
      resource_usages = gcs_utils.ResourceUsageBatchData.FromString(
          message
      )
      assert (
          len(resource_usages.resource_load_by_shape.resource_demands)
          == 1
      )
      assert resource_usages.resource_load_by_shape.resource_demands[
          0
      ].shape == {"CPU": 80.0}
      assert (
          resource_usages.resource_load_by_shape.resource_demands[
              0
          ].num_infeasible_requests_queued
          == 1
      )
  
      actors.append(Actor2.remote())
      time.sleep(1)
  
      # Two actors cannot be scheduled.
      message = global_state_accessor.get_all_resource_usage()
      resource_usages = gcs_utils.ResourceUsageBatchData.FromString(
          message
      )
      assert (
          len(resource_usages.resource_load_by_shape.resource_demands)
          == 1
      )
      assert (
          resource_usages.resource_load_by_shape.resource_demands[
              0
          ].num_infeasible_requests_queued
          == 2
      )
  
      global_state_accessor.disconnect()
  
imports: import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  from ray.util.state import list_actors
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, convert_actor_state, kill_actor_and_wait_for_failure, make_global_state_accessor, run_string_as_driver, wait_for_condition, wait_for_pid_to_exit
  from ray._private.ray_constants import gcs_actor_scheduling_enabled
  from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_put
label_desc: Use image node_id 28 for calling a global function with example usage: test_actor_resource_demand(shutdown_only) and returns: str