code: def test_negative_resource_availability(shutdown_only):
      """Test pg scheduling when resource availability is negative."""
      ray.init(num_cpus=1)
  
      signal1 = SignalActor.remote()
      signal2 = SignalActor.remote()
  
      @ray.remote(num_cpus=0)
      def child(signal1):
          ray.get(signal1.wait.remote())
  
      @ray.remote(num_cpus=1)
      def parent(signal1, signal2):
          # Release the CPU resource,
          # the resource will be acquired by Actor.
          ray.get(child.remote(signal1))
          # Re-acquire the CPU resource
          # the availability should be -1 afterwards.
          signal2.send.remote()
          while True:
              time.sleep(1)
  
      @ray.remote(num_cpus=1)
      class Actor:
          def ping(self):
              return "hello"
  
      parent.remote(signal1, signal2)
      actor = Actor.remote()
      ray.get(actor.ping.remote())
      signal1.send.remote()
      ray.get(signal2.wait.remote())
      # CPU resource availability should be negative now
      # and the pg should be pending.
      pg = placement_group([{"CPU": 1}])
      with pytest.raises(ray.exceptions.GetTimeoutError):
          ray.get(pg.ready(), timeout=2)
  
imports: import os
  import platform
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.experimental.internal_kv
  from ray._private.test_utils import make_global_state_accessor, wait_for_condition, get_metric_check_condition, MetricSamplePattern
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, PlacementGroupSchedulingStrategy
  from ray._private.test_utils import SignalActor
label_desc: Use image node_id 13 for calling a global function with example usage: test_negative_resource_availability(shutdown_only) and returns: str