code: def test(num_tasks):
      addr = ray.init(address="auto")
  
      test_utils.wait_for_condition(no_resource_leaks)
      monitor_actor = test_utils.monitor_memory_usage()
      dashboard_test = DashboardTestAtScale(addr)
  
      def not_none(res):
          return res is not None
  
      api_caller = periodic_invoke_state_apis_with_actor(
          apis=[StateAPICallSpec(summarize_tasks, not_none)],
          call_interval_s=4,
          print_result=True,
      )
  
      start_time = time.time()
      used_cpus = test_max_running_tasks(num_tasks)
      end_time = time.time()
      ray.get(monitor_actor.stop_run.remote())
      used_gb, usage = ray.get(
          monitor_actor.get_peak_memory_info.remote()
      )
      print(f"Peak memory usage: {round(used_gb, 2)}GB")
      print(f"Peak memory usage per processes:\n {usage}")
      ray.get(api_caller.stop.remote())
  
      del api_caller
      del monitor_actor
      test_utils.wait_for_condition(no_resource_leaks)
  
      try:
          summarize_worker_startup_time()
      except Exception as e:
          print("Failed to summarize worker startup time.")
          print(e)
  
      rate = num_tasks / (end_time - start_time - sleep_time)
      print(
          f"Success! Started {num_tasks} tasks in {end_time - start_time}s. "
          f"({rate} tasks/s)"
      )
  
      results = {
          "tasks_per_second": rate,
          "num_tasks": num_tasks,
          "time": end_time - start_time,
          "used_cpus": used_cpus,
          "success": "1",
          "_peak_memory": round(used_gb, 2),
          "_peak_process_memory": usage,
          "perf_metrics": [
              {
                  "perf_metric_name": "tasks_per_second",
                  "perf_metric_value": rate,
                  "perf_metric_type": "THROUGHPUT",
              },
              {
                  "perf_metric_name": "used_cpus_by_deadline",
                  "perf_metric_value": used_cpus,
                  "perf_metric_type": "THROUGHPUT",
              },
          ],
      }
  
      dashboard_test.update_release_test_result(results)
      test_utils.safe_write_to_results_json(results)
  
file_path: ray/release/benchmarks/distributed/test_many_tasks.py
imports: import click
  import ray
  import ray._private.test_utils
  import time
  import tqdm
  from ray.util.state import summarize_tasks
  from dashboard_test import DashboardTestAtScale
  from ray._private.state_api_test_utils import StateAPICallSpec, periodic_invoke_state_apis_with_actor, summarize_worker_startup_time
label_desc: Use image node_id 3 for calling a global function with example usage: test(num_tasks) and returns: unknown