code: def main():
      """Run a long running placement group creation/removal tests.
  
      This test runs 20 trials first and measure the P50 performance.
  
      After that it runs trials for a long time and make sure the
      P50 creation/scheduling/removal performance is not regressed
      after the long running job.
      """
      args, _ = parse_script_args()
      NUM_PG_AT_EACH_STAGE = args.num_pgs_stage
      NUM_PENDING_PG = args.num_pending_pgs
      TOTAL_STAGE = args.num_stages
  
      if args.local:
          ray.init(resources={"custom": 100, "pending": 1})
      else:
          ray.init(address="auto")
  
      assert (
          ray.cluster_resources()["custom"] >= NUM_PG_AT_EACH_STAGE * 4
      )
      assert ray.cluster_resources()["pending"] >= 1
  
      # Create pending placement groups.
      pending_pgs = []
      for _ in range(NUM_PENDING_PG):
          # Right now, we don't have infeasible pgs,
          # so this will simulate the pending pgs.
          pending_pgs.append(
              placement_group([{"pending": 1}], strategy="PACK")
          )
  
      (scheduling_perf, removing_perf, creation_perf) = run_trial(
          20, NUM_PG_AT_EACH_STAGE
      )
      (
          scheduling_perf_final,
          removing_perf_final,
          creation_perf_final,
      ) = run_trial(TOTAL_STAGE, NUM_PG_AT_EACH_STAGE)
  
      print(f"Scheduling performance 20 trials: {scheduling_perf}")
      print(
          f"Scheduling performance {TOTAL_STAGE} trials: {scheduling_perf_final}"
      )
      print(f"Removal performance 20 trials: {removing_perf}")
      print(
          f"Removal performance {TOTAL_STAGE} trials: {removing_perf_final}"
      )
      print(f"Creation performance 20 trials: {creation_perf}")
      print(
          f"Creation performance {TOTAL_STAGE} trials: {creation_perf_final}"
      )
  
      assert (
          scheduling_perf["p50_ms"] * 100
          > scheduling_perf_final["p50_ms"]
      )
      assert (
          removing_perf["p50_ms"] * 100 > removing_perf_final["p50_ms"]
      )
      assert (
          creation_perf["p50_ms"] * 100 > creation_perf_final["p50_ms"]
      )
  
      if "TEST_OUTPUT_JSON" in os.environ:
          out_file = open(os.environ["TEST_OUTPUT_JSON"], "w")
          results = {}
          json.dump(results, out_file)
  
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 3 for calling a global function with example usage: main() without return types