code: def test(
      num_tasks,
      num_actors,
      num_objects,
      num_actors_for_objects,
      log_file_size_byte,
      smoke_test,
  ):
      ray.init(address="auto", log_to_driver=False)
  
      if smoke_test:
          num_tasks = "1,100"
          num_actors = "1,10"
          num_objects = "1,100"
          num_actors_for_objects = 1
          log_file_size_byte = f"64,{16*MiB}"
          global STATE_LIST_LIMIT
          STATE_LIST_LIMIT = STATE_LIST_LIMIT // 1000
  
      # Parse the input
      (
          num_tasks_arr,
          num_actors_arr,
          num_objects_arr,
          log_file_size_arr,
      ) = _parse_input(
          num_tasks, num_actors, num_objects, log_file_size_byte
      )
  
      test_utils.wait_for_condition(no_resource_leaks)
      monitor_actor = test_utils.monitor_memory_usage()
      start_time = time.perf_counter()
      # Run some long-running tasks
      for n in num_tasks_arr:
          logger.info(f"Running with many tasks={n}")
          test_many_tasks(num_tasks=n)
          logger.info(f"test_many_tasks({n}) PASS")
  
      # Run many actors
      for n in num_actors_arr:
          logger.info(f"Running with many actors={n}")
          test_many_actors(num_actors=n)
          logger.info(f"test_many_actors({n}) PASS")
  
      # Create many objects
      for n in num_objects_arr:
          logger.info(f"Running with many objects={n}")
          test_many_objects(
              num_objects=n, num_actors=num_actors_for_objects
          )
          logger.info(f"test_many_objects({n}) PASS")
  
      # Create large logs
      for n in log_file_size_arr:
          logger.info(f"Running with large file={n} bytes")
          test_large_log_file(log_file_size_byte=n)
          logger.info(f"test_large_log_file({n} bytes) PASS")
  
      print("\n\nPASS")
      end_time = time.perf_counter()
  
      # Collect mem usage
      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}")
      del monitor_actor
  
      state_perf_result = aggregate_perf_results()
      results = {
          "time": end_time - start_time,
          "success": "1",
          "_peak_memory": round(used_gb, 2),
          "_peak_process_memory": usage,
      }
  
      if not smoke_test:
          results["perf_metrics"] = [
              {
                  "perf_metric_name": "avg_state_api_latency_sec",
                  "perf_metric_value": state_perf_result[
                      "avg_state_api_latency_sec"
                  ],
                  "perf_metric_type": "LATENCY",
              },
              {
                  "perf_metric_name": "avg_state_api_get_log_latency_sec",
                  "perf_metric_value": state_perf_result[
                      "avg_get_log_latency_sec"
                  ],
                  "perf_metric_type": "LATENCY",
              },
              {
                  "perf_metric_name": "avg_state_api_list_tasks_10000_latency_sec",
                  "perf_metric_value": state_perf_result[
                      "avg_list_tasks_10000_latency_sec"
                  ],
                  "perf_metric_type": "LATENCY",
              },
              {
                  "perf_metric_name": "avg_state_api_list_actors_5000_latency_sec",
                  "perf_metric_value": state_perf_result[
                      "avg_list_actors_5000_latency_sec"
                  ],
                  "perf_metric_type": "LATENCY",
              },
              {
                  "perf_metric_name": "avg_state_api_list_objects_50000_latency_sec",
                  "perf_metric_value": state_perf_result[
                      "avg_list_objects_50000_latency_sec"
                  ],
                  "perf_metric_type": "LATENCY",
              },
          ]
  
      if "TEST_OUTPUT_JSON" in os.environ:
          out_file = open(os.environ["TEST_OUTPUT_JSON"], "w")
          json.dump(results, out_file)
  
      results.update(state_perf_result)
      print(json.dumps(results, indent=2))
  
imports: import click
  import json
  import ray
  from ray._private.ray_constants import LOG_PREFIX_ACTOR_NAME, LOG_PREFIX_JOB_ID
  from ray._private.state_api_test_utils import STATE_LIST_LIMIT, StateAPIMetric, aggregate_perf_results, invoke_state_api, GLOBAL_STATE_STATS
  import ray._private.test_utils
  import tqdm
  import time
  import os
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray.util.state import get_log, list_actors, list_objects, list_tasks
  import logging
label_desc: Use image node_id 8 for calling a global function with example usage: test(num_tasks, num_actors, num_objects, num_actors_for_objects, log_file_size_byte, smoke_test) without return types