code: def test_many_tasks(num_tasks: int):
      TASK_NAME_TEMPLATE = "pi4_sample_{num_tasks}"
      if num_tasks == 0:
          logger.info("Skipping test with no tasks")
          return
  
      # No running tasks
      invoke_state_api_n(
          lambda res: len(res) == 0,
          list_tasks,
          filters=[
              (
                  "name",
                  "=",
                  TASK_NAME_TEMPLATE.format(num_tasks=num_tasks),
              )
          ],
          key_suffix="0",
          limit=STATE_LIST_LIMIT,
          err_msg=(
              "Expect 0 running tasks for "
              f"{TASK_NAME_TEMPLATE.format(num_tasks=num_tasks)}"
          ),
      )
  
      # Task definition adopted from:
      # https://docs.ray.io/en/master/ray-core/examples/highly_parallel.html
      from random import random
  
      SAMPLES = 100
  
      @ray.remote
      def pi4_sample():
          in_count = 0
          for _ in range(SAMPLES):
              x, y = random(), random()
              if x * x + y * y <= 1:
                  in_count += 1
          return in_count
  
      results = []
      for _ in tqdm.trange(num_tasks, desc="Launching tasks"):
          results.append(
              pi4_sample.options(
                  name=TASK_NAME_TEMPLATE.format(num_tasks=num_tasks)
              ).remote()
          )
  
      invoke_state_api_n(
          lambda res: len(res) == num_tasks,
          list_tasks,
          filters=[
              (
                  "name",
                  "=",
                  TASK_NAME_TEMPLATE.format(num_tasks=num_tasks),
              )
          ],
          key_suffix=f"{num_tasks}",
          limit=STATE_LIST_LIMIT,
          err_msg=f"Expect {num_tasks} non finished tasks.",
      )
  
      ray.get(results)
  
      # Clean up
      # All compute tasks done other than the signal actor
      invoke_state_api_n(
          lambda res: len(res) == 0,
          list_tasks,
          filters=[
              (
                  "name",
                  "=",
                  TASK_NAME_TEMPLATE.format(num_tasks=num_tasks),
              ),
              ("state", "=", "RUNNING"),
          ],
          key_suffix="0",
          limit=STATE_LIST_LIMIT,
          err_msg="Expect 0 running tasks",
      )
  
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 2 for calling a global function with example usage: test_many_tasks(num_tasks) and returns: in_count