code: def test_request_pending(ray_start_4_cpus):
      """Test that requesting too many resources leads to pending PGs.
  
      - Cluster of 4 CPUs
      - Request 3 PGs a 2 CPUs
      - Acquire 2 PGs
      - Assert no resources are available anymore
      - Return both PGs
      - Assert resources are available again
      - Cancel request
      - Assert no resources are available again
      """
      manager = PlacementGroupResourceManager(update_interval_s=0)
      assert not manager.has_resources_ready(REQUEST_2_CPU)
  
      manager.request_resources(REQUEST_2_CPU)
      manager.request_resources(REQUEST_2_CPU)
      manager.request_resources(REQUEST_2_CPU)
  
      # Wait until some are ready
      ray.wait(manager.get_resource_futures(), num_returns=2)
  
      assert manager.has_resources_ready(REQUEST_2_CPU)
      assert len(manager.get_resource_futures()) == 1
  
      pg_states = _count_pg_states()
      assert pg_states["CREATED"] == 2
      assert pg_states["PENDING"] == 1
      assert pg_states["REMOVED"] == 0
  
      acq1 = manager.acquire_resources(REQUEST_2_CPU)
      acq2 = manager.acquire_resources(REQUEST_2_CPU)
  
      assert not manager.has_resources_ready(REQUEST_2_CPU)
  
      manager.free_resources(acq1)
      manager.free_resources(acq2)
  
      # Third PG becomes ready
      ray.wait(manager.get_resource_futures(), num_returns=1)
      assert manager.has_resources_ready(REQUEST_2_CPU)
  
      pg_states = _count_pg_states()
      assert pg_states["CREATED"] == 1
      assert pg_states["PENDING"] == 0
      assert pg_states["REMOVED"] == 2
  
      manager.cancel_resource_request(REQUEST_2_CPU)
      assert not manager.has_resources_ready(REQUEST_2_CPU)
  
      pg_states = _count_pg_states()
      assert pg_states["CREATED"] == 0
      assert pg_states["PENDING"] == 0
      assert pg_states["REMOVED"] == 3
  
file_path: ray/python/ray/air/tests/execution/test_resource_manager_placement_group.py
imports: import time
  from collections import Counter
  import pytest
  import ray
  from ray.air.execution.resources.placement_group import PlacementGroupResourceManager
  from ray.air.execution.resources.request import ResourceRequest
label_desc: Use image node_id 5 for calling a global function with example usage: test_request_pending(ray_start_4_cpus) without return types