code: def run_trial(total_stage, num_pg_per_stage):
      creating_e2e_s = []
      removing_e2e_s = []
      # Create and remove placement groups.
      for i in range(total_stage):
          # Create pgs.
          pgs = []
          start = perf_counter()
          for _ in range(num_pg_per_stage):
              pgs.append(
                  placement_group(
                      bundles=[{"custom": 0.025} for _ in range(4)],
                      strategy="PACK",
                  )
              )
          logger.info(f"Created {num_pg_per_stage} pgs.")
          ray.get([pg.ready() for pg in pgs])
          end = perf_counter()
          total_creating_time = end - start
          logger.info(
              f"Creating {num_pg_per_stage} took "
              f"{total_creating_time} seconds at stage {i}"
          )
          creating_e2e_s.append(total_creating_time * 1000.0)
  
          # Remove pgs
          start = perf_counter()
          for _, pg in enumerate(pgs):
              remove_placement_group(pg)
          end = perf_counter()
          total_removal_time = end - start
          logger.info(
              f"removed {num_pg_per_stage} pgs took "
              f"{total_removal_time} seconds at stage {i}"
          )
          removing_e2e_s.append(total_removal_time * 1000.0)
          # time.sleep(1)
  
      # Calculate the scheduling latency (excluding queueing time).
      latencies = []
      for entry in ray.util.placement_group_table().values():
          latency = entry["stats"]["scheduling_latency_ms"]
          latencies.append(latency)
      latencies = sorted(latencies)
      removing_e2e_s = sorted(removing_e2e_s)
      creating_e2e_s = sorted(creating_e2e_s)
  
      def get_scheduling_perf(latencies):
          """Return P10, 50, 95, 99 latency"""
          p10 = latencies[int(len(latencies) * 0.1)]
          p50 = latencies[int(len(latencies) * 0.5)]
          p95 = latencies[int(len(latencies) * 0.95)]
          p99 = latencies[int(len(latencies) * 0.99)]
          return {
              "p10_ms": p10,
              "p50_ms": p50,
              "p95_ms": p95,
              "p99_ms": p99,
          }
  
      scheduling_perf = get_scheduling_perf(latencies)
      removing_perf = get_scheduling_perf(removing_e2e_s)
      creation_perf = get_scheduling_perf(creating_e2e_s)
  
      wait_for_condition(
          lambda: (
              ray.cluster_resources()["custom"]
              == ray.available_resources()["custom"]
          ),
          timeout=30,
      )
      wait_for_condition(
          lambda: (
              ray.cluster_resources()["pending"]
              == ray.available_resources()["pending"]
          ),
          timeout=30,
      )
  
      return scheduling_perf, removing_perf, creation_perf
  
file_path: ray/release/nightly_tests/placement_group_tests/long_running_performance_test.py
imports: import argparse
  import os
  import json
  import logging
  from time import perf_counter
  import ray
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray._private.test_utils import wait_for_condition
label_desc: Use image node_id 1 for calling a global function with example usage: run_trial(total_stage, num_pg_per_stage) and returns: scheduling_perf,  removing_perf,  creation_perf, dict