code: def main():
      """The test simulates the workload with many threaded actors.
  
      Test is doing 4 things for 1 hour.
  
      - It first creates actors as many as num_cpus with max_concurrency=10
      - Each actor computes pi and put the result to the queue.
      - Driver keeps getting result & metadata from the actor.
      - Every X seconds, it kills all actors and restarts them.
      """
      ray.init(address="auto")
      args, unknown = parse_script_args()
      num_cpus = ray.cluster_resources()["CPU"]
      num_nodes = sum(1 for n in ray.nodes() if n["Alive"])
      print(f"Total number of actors: {num_cpus}, nodes: {num_nodes}")
      monitor_actor = monitor_memory_usage()
  
      start = time.time()
      while time.time() - start < args.test_runtime:
          # Step 1: Create actors and start computation loop.
          print("Create actors.")
          actors = start_actors(num_cpus, num_nodes)
  
          # Step 2: Get the pi result from actors.
          compute_start = time.time()
          print("Start computation.")
          while time.time() - compute_start < args.kill_interval_s:
              # Get the metadata.
              ray.get([actor.get_metadata.remote() for actor in actors])
              # Get the result.
              pb = ProgressBar("Computing Pi", num_cpus)
              results = [actor.get_pi.remote() for actor in actors]
              pb.fetch_until_complete(results)
              pb.close()
  
          # Step 3: Kill actors.
          print("Kill all actors.")
          for actor in actors:
              ray.kill(actor)
  
      # Report the result.
      print("PASSED.")
      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())
  
      result = {"success": 0}
      with open(os.environ["TEST_OUTPUT_JSON"], "w") as f:
          f.write(json.dumps(result))
  
imports: import argparse
  import json
  import logging
  import numpy
  import os
  import threading
  import time
  from ray._private.test_utils import monitor_memory_usage
  from ray.data._internal.progress_bar import ProgressBar
  from collections import namedtuple
  from queue import Queue
  import ray
label_desc: Use image node_id 3 for calling a global function with example usage: main() without return types