code: def test_actors_and_tasks_with_gpus(ray_start_cluster):
      cluster = ray_start_cluster
      num_nodes = 3
      num_gpus_per_raylet = 2
      for i in range(num_nodes):
          cluster.add_node(
              num_cpus=num_gpus_per_raylet, num_gpus=num_gpus_per_raylet
          )
      ray.init(address=cluster.address)
  
      def check_intervals_non_overlapping(list_of_intervals):
          for i in range(len(list_of_intervals)):
              for j in range(i):
                  first_interval = list_of_intervals[i]
                  second_interval = list_of_intervals[j]
                  # Check that list_of_intervals[i] and list_of_intervals[j]
                  # don't overlap.
                  assert first_interval[0] < first_interval[1]
                  assert second_interval[0] < second_interval[1]
                  intervals_nonoverlapping = (
                      first_interval[1] <= second_interval[0]
                      or second_interval[1] <= first_interval[0]
                  )
                  assert (
                      intervals_nonoverlapping
                  ), "Intervals {} and {} are overlapping.".format(
                      first_interval, second_interval
                  )
  
      @ray.remote(num_gpus=1)
      def f1():
          t1 = time.time()
          time.sleep(0.1)
          t2 = time.time()
          gpu_ids = ray.get_gpu_ids()
          assert len(gpu_ids) == 1
          assert gpu_ids[0] in range(num_gpus_per_raylet)
          return (
              ray._private.worker.global_worker.node.unique_id,
              tuple(gpu_ids),
              [t1, t2],
          )
  
      @ray.remote(num_gpus=2)
      def f2():
          t1 = time.time()
          time.sleep(0.1)
          t2 = time.time()
          gpu_ids = ray.get_gpu_ids()
          assert len(gpu_ids) == 2
          assert gpu_ids[0] in range(num_gpus_per_raylet)
          assert gpu_ids[1] in range(num_gpus_per_raylet)
          return (
              ray._private.worker.global_worker.node.unique_id,
              tuple(gpu_ids),
              [t1, t2],
          )
  
      @ray.remote(num_gpus=1)
      class Actor1:
          def __init__(self):
              self.gpu_ids = ray.get_gpu_ids()
              assert len(self.gpu_ids) == 1
              assert self.gpu_ids[0] in range(num_gpus_per_raylet)
  
          def get_location_and_ids(self):
              assert ray.get_gpu_ids() == self.gpu_ids
              return (
                  ray._private.worker.global_worker.node.unique_id,
                  tuple(self.gpu_ids),
              )
  
      def locations_to_intervals_for_many_tasks():
          # Launch a bunch of GPU tasks.
          locations_ids_and_intervals = ray.get(
              [
                  f1.remote()
                  for _ in range(5 * num_nodes * num_gpus_per_raylet)
              ]
              + [
                  f2.remote()
                  for _ in range(5 * num_nodes * num_gpus_per_raylet)
              ]
              + [
                  f1.remote()
                  for _ in range(5 * num_nodes * num_gpus_per_raylet)
              ]
          )
  
          locations_to_intervals = collections.defaultdict(lambda: [])
          for (
              location,
              gpu_ids,
              interval,
          ) in locations_ids_and_intervals:
              for gpu_id in gpu_ids:
                  locations_to_intervals[(location, gpu_id)].append(
                      interval
                  )
          return locations_to_intervals
  
      # Run a bunch of GPU tasks.
      locations_to_intervals = locations_to_intervals_for_many_tasks()
      # For each GPU, verify that the set of tasks that used this specific
      # GPU did not overlap in time.
      for locations in locations_to_intervals:
          check_intervals_non_overlapping(
              locations_to_intervals[locations]
          )
  
      # Create an actor that uses a GPU.
      a = Actor1.remote()
      actor_location = ray.get(a.get_location_and_ids.remote())
      actor_location = (actor_location[0], actor_location[1][0])
      # This check makes sure that actor_location is formatted the same way
      # that the keys of locations_to_intervals are formatted.
      assert actor_location in locations_to_intervals
  
      # Run a bunch of GPU tasks.
      locations_to_intervals = locations_to_intervals_for_many_tasks()
      # For each GPU, verify that the set of tasks that used this specific
      # GPU did not overlap in time.
      for locations in locations_to_intervals:
          check_intervals_non_overlapping(
              locations_to_intervals[locations]
          )
      # Make sure that the actor's GPU was not used.
      assert actor_location not in locations_to_intervals
  
      # Create more actors to fill up all the GPUs.
      more_actors = [
          Actor1.remote()
          for _ in range(num_nodes * num_gpus_per_raylet - 1)
      ]
      # Wait for the actors to finish being created.
      ray.get(
          [actor.get_location_and_ids.remote() for actor in more_actors]
      )
  
      # Now if we run some GPU tasks, they should not be scheduled.
      results = [f1.remote() for _ in range(30)]
      ready_ids, remaining_ids = ray.wait(results, timeout=1.0)
      assert len(ready_ids) == 0
  
imports: import collections
  import os
  import sys
  import time
  import pytest
  import ray
  import ray.cluster_utils
label_desc: Use image node_id 8 for calling a global function with example usage: test_actors_and_tasks_with_gpus(ray_start_cluster) and returns: ray,  tuple,  list, ray,  tuple,  list, locations_to_intervals, ray,  tuple