code: def test_placement_group_perf(num_pgs, num_bundles, num_pending_pgs):
      # Run the placement group performance benchmark given arguments.
      assert ray.cluster_resources()["custom"] >= (
          RESOURCES_VALUE * num_pgs * num_bundles
      )
  
      def placement_group_create(num_pgs):
          pgs = [
              ray.util.placement_group(
                  bundles=[
                      {"custom": 0.001} for _ in range(num_bundles)
                  ],
                  strategy="SPREAD",
              )
              for _ in range(num_pgs)
          ]
          [pg.wait(timeout_seconds=30) for pg in pgs]
          for pg in pgs:
              ray.util.remove_placement_group(pg)
  
      print(
          f"Num pending pgs: {num_pending_pgs}, "
          f"Num pgs: {num_pgs}, "
          f"Num bundles {num_bundles}"
      )
  
      # Get the throughput.
      throughput = timeit(
          "placement group create per second",
          lambda: placement_group_create(num_pgs),
          num_pgs,
          warmup_time_sec=0,
      )
  
      # Get fine-grained scheduling stats.
      latencies = []
      e2e_latencies = []
      scheduling_attempts = []
      for entry in ray.util.placement_group_table().values():
          latency = entry["stats"]["scheduling_latency_ms"]
          e2e_latency = entry["stats"]["end_to_end_creation_latency_ms"]
          scheduling_attempt = entry["stats"]["scheduling_attempt"]
          latencies.append(latency)
          e2e_latencies.append(e2e_latency)
          scheduling_attempts.append(scheduling_attempt)
      latencies = sorted(latencies)
      e2e_latencies = sorted(e2e_latencies)
      scheduling_attempts = sorted(scheduling_attempts)
  
      # Pure scheduling latency without queuing time.
      print(
          "P50 scheduling latency ms: "
          f"{latencies[int(len(latencies) * 0.5)]}"
      )
      print(
          "P95 scheduling latency ms: "
          f"{latencies[int(len(latencies) * 0.95)]}"
      )
      print(
          "P99 scheduling latency ms: "
          f"{latencies[int(len(latencies) * 0.99)]}"
      )
  
      # Scheduling latency including queueing time.
      print(
          "P50 e2e scheduling latency ms: "
          f"{e2e_latencies[int(len(e2e_latencies) * 0.5)]}"
      )
      print(
          "P95 e2e scheduling latency ms: "
          f"{e2e_latencies[int(len(e2e_latencies) * 0.95)]}"
      )
      print(
          "P99 e2e scheduling latency ms: "
          f"{e2e_latencies[int(len(e2e_latencies) * 0.99)]}"
      )
  
      # Number of time scheduling was retried before succeeds.
      print(
          "P50 scheduling attempts: "
          f"{scheduling_attempts[int(len(scheduling_attempts) * 0.5)]}"
      )
      print(
          "P95 scheduling attempts: "
          f"{scheduling_attempts[int(len(scheduling_attempts) * 0.95)]}"
      )
      print(
          "P99 scheduling attempts: "
          f"{scheduling_attempts[int(len(scheduling_attempts) * 0.99)]}"
      )
  
      return {
          "pg_creation_per_second": throughput[0][1],
          "p50_scheduling_latency_ms": latencies[
              int(len(latencies) * 0.5)
          ],
          "p50_e2e_pg_creation_latency_ms": e2e_latencies[
              int(len(e2e_latencies) * 0.5)
          ],
      }
  
file_path: ray/release/nightly_tests/placement_group_tests/placement_group_performance_test.py
imports: import argparse
  import json
  import os
  import ray
  from ray._private.ray_microbenchmark_helpers import timeit
label_desc: Use image node_id 1 for calling a global function with example usage: test_placement_group_perf(num_pgs, num_bundles, num_pending_pgs) and returns: dict