code: def main():
      """Test task/actor/placement group basic chaos test.
  
      Currently, it only tests node failures scenario.
      Node failures are implemented by an actor that keeps calling
      Raylet's KillRaylet RPC.
  
      Ideally, we should setup the infra to cause machine failures/
      network partitions/etc., but we don't do that for now.
  
      In the short term, we will only test gRPC network delay +
      node failures.
  
      Currently, the test runs 3 steps. Each steps records the
      peak memory usage to observe the memory usage while there
      are node failures.
  
      Step 1: Warm up the cluster. It is needed to pre-start workers
          if necessary.
  
      Step 2: Start the test without a failure.
  
      Step 3: Start the test with constant node failures.
      """
      args, unknown = parse_script_args()
      logging.info("Received arguments: {}".format(args))
      ray.init(address="auto")
      total_num_cpus = ray.cluster_resources()["CPU"]
      total_nodes = 0
      for n in ray.nodes():
          if n["Alive"]:
              total_nodes += 1
      monitor_actor = monitor_memory_usage()
  
      workload = None
      if args.workload == "tasks":
          workload = run_task_workload
      elif args.workload == "actors":
          workload = run_actor_workload
      elif args.workload == "pg":
          workload = run_placement_group_workload
      else:
          assert False
  
      # Step 1
      print("Warm up... Prestarting workers if necessary.")
      start = time.time()
      workload(total_num_cpus, args.smoke)
      print(f"Runtime when warm up: {time.time() - start}")
  
      # Step 2
      print("Running without failures")
      start = time.time()
      workload(total_num_cpus, args.smoke)
      print(
          f"Runtime when there are no failures: {time.time() - start}"
      )
      used_gb, usage = ray.get(
          monitor_actor.get_peak_memory_info.remote()
      )
      print("Memory usage without failures.")
      print(f"Peak memory usage: {round(used_gb, 2)}GB")
      print(f"Peak memory usage per processes:\n {usage}")
  
      # Step 3
      print("Running with failures")
      start = time.time()
      node_killer = ray.get_actor(
          "NodeKillerActor", namespace="release_test_namespace"
      )
      node_killer.run.remote()
      workload(total_num_cpus, args.smoke)
      print(
          f"Runtime when there are many failures: {time.time() - start}"
      )
      print(
          f"Total node failures: "
          f"{ray.get(node_killer.get_total_killed.remote())}"
      )
      node_killer.stop_run.remote()
      used_gb, usage = ray.get(
          monitor_actor.get_peak_memory_info.remote()
      )
      print("Memory usage with failures.")
      print(f"Peak memory usage: {round(used_gb, 2)}GB")
      print(f"Peak memory usage per processes:\n {usage}")
  
      # Report the result.
      ray.get(monitor_actor.stop_run.remote())
      print(
          "Total number of killed nodes: "
          f"{ray.get(node_killer.get_total_killed.remote())}"
      )
      with open(os.environ["TEST_OUTPUT_JSON"], "w") as f:
          f.write(
              json.dumps(
                  {
                      "success": 1,
                      "_peak_memory": round(used_gb, 2),
                      "_peak_process_memory": usage,
                  }
              )
          )
  
file_path: ray/release/nightly_tests/chaos_test/test_chaos_basic.py
imports: import argparse
  import json
  import logging
  import os
  import random
  import string
  import time
  import numpy
  import ray
  from ray._private.test_utils import monitor_memory_usage, wait_for_condition
  from ray.data._internal.progress_bar import ProgressBar
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 5 for calling a global function with example usage: main() without return types