code: def stage4():
      num_tasks = int(ray.cluster_resources()["CPU"])
      logger.info("Scheduling many tasks for spillback.")
  
      @ray.remote(num_cpus=1)
      def func(t):
          if t % 100 == 0:
              logger.info(f"[spillback test] {t}/{num_tasks}")
          start = time.perf_counter()
          time.sleep(1)
          end = time.perf_counter()
          return (
              start,
              end,
              ray._private.worker.global_worker.node.unique_id,
          )
  
      results = ray.get([func.remote(i) for i in range(num_tasks)])
  
      host_to_start_times = defaultdict(list)
      for start, end, host in results:
          host_to_start_times[host].append(start)
  
      spreads = []
      for host in host_to_start_times:
          last = max(host_to_start_times[host])
          first = min(host_to_start_times[host])
          spread = last - first
          spreads.append(spread)
          logger.info(
              f"Spread: {last - first}\tLast: {last}\tFirst: {first}"
          )
  
      avg_spread = np.mean(spreads)
      logger.info(f"Avg spread: {np.mean(spreads)}")
      return avg_spread
  
file_path: ray/release/nightly_tests/stress_tests/test_many_tasks.py
imports: import argparse
  import json
  import logging
  import os
  import time
  from collections import defaultdict
  import numpy
  import ray
label_desc: Use image node_id 6 for calling a global function with example usage: stage4() and returns: avg_spread, start,  end,  ray