code: def test_run_tasks_concurrent(ctx: dict):
      num_tasks = 2
      num_actors = 2
  
      @ray.remote(num_cpus=WORKER_NODE_CPU)
      def f():
          while True:
              time.sleep(1)
  
      @ray.remote(num_cpus=WORKER_NODE_CPU)
      class Actor:
          def __init__(self):
              pass
  
      tasks = [f.remote() for _ in range(num_tasks)]
      actors = [Actor.remote() for _ in range(num_actors)]
  
      ctx["num_cpus"] += (num_tasks + num_actors) * WORKER_NODE_CPU
      ctx["num_nodes"] += num_tasks + num_actors
  
      logger.info(
          f"Waiting for {ctx['num_nodes']} nodes, {ctx['num_cpus']} cpus"
      )
      wait_for_condition(
          check_cluster,
          timeout=60 * 5,  # 5min
          retry_interval_ms=DEFAULT_RETRY_INTERVAL_MS,
          targets=[
              NodeCountCheck(ctx["num_nodes"]),
              TotalResourceCheck({"CPU": ctx["num_cpus"]}),
          ],
      )
  
      [ray.cancel(task) for task in tasks]
      [ray.kill(actor) for actor in actors]
  
      ctx["num_cpus"] -= (num_actors + num_tasks) * WORKER_NODE_CPU
      ctx["num_nodes"] -= num_actors + num_tasks
  
      logger.info(
          f"Waiting for cluster to scale down to {ctx['num_nodes']} nodes, "
          f"{ctx['num_cpus']} cpus"
      )
      wait_for_condition(
          check_cluster,
          timeout=60 + IDLE_TERMINATION_S,
          retry_interval_ms=DEFAULT_RETRY_INTERVAL_MS,
          targets=[
              NodeCountCheck(ctx["num_nodes"]),
              TotalResourceCheck({"CPU": ctx["num_cpus"]}),
          ],
      )
  
imports: import ray
  from ray._private.test_utils import wait_for_condition
  from ray.autoscaler.v2.tests.util import NodeCountCheck, TotalResourceCheck, check_cluster
  import time
  from logger import logger
label_desc: Use image node_id 2 for calling a global function with example usage: test_run_tasks_concurrent(ctx) without return types