code: def test_actor_multiple_gpus_from_multiple_tasks(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=4 * num_gpus_per_raylet,
              num_gpus=num_gpus_per_raylet,
              _system_config={"health_check_failure_threshold": 100}
              if i == 0
              else {},
          )
      ray.init(address=cluster.address)
  
      @ray.remote
      def create_actors(i, n):
          @ray.remote(num_gpus=1)
          class Actor:
              def __init__(self, i, j):
                  self.gpu_ids = ray.get_gpu_ids()
  
              def get_location_and_ids(self):
                  return (
                      (
                          ray._private.worker.global_worker.node.unique_id
                      ),
                      tuple(self.gpu_ids),
                  )
  
              def sleep(self):
                  time.sleep(100)
  
          # Create n actors.
          actors = []
          for j in range(n):
              actors.append(Actor.remote(i, j))
  
          locations = ray.get(
              [actor.get_location_and_ids.remote() for actor in actors]
          )
  
          # Put each actor to sleep for a long time to prevent them from getting
          # terminated.
          for actor in actors:
              actor.sleep.remote()
  
          return locations
  
      all_locations = ray.get(
          [
              create_actors.remote(i, num_gpus_per_raylet)
              for i in range(num_nodes)
          ]
      )
  
      # Make sure that no two actors are assigned to the same GPU.
      node_names = {
          location
          for locations in all_locations
          for location, gpu_id in locations
      }
      assert len(node_names) == num_nodes
  
      # Keep track of which GPU IDs are being used for each location.
      gpus_in_use = {node_name: [] for node_name in node_names}
      for locations in all_locations:
          for location, gpu_ids in locations:
              gpus_in_use[location].extend(gpu_ids)
      for node_name in node_names:
          assert len(set(gpus_in_use[node_name])) == num_gpus_per_raylet
  
      @ray.remote(num_gpus=1)
      class Actor:
          def __init__(self):
              self.gpu_ids = ray.get_gpu_ids()
  
          def get_location_and_ids(self):
              return (
                  ray._private.worker.global_worker.node.unique_id,
                  tuple(self.gpu_ids),
              )
  
      # All the GPUs should be used up now.
      a = Actor.remote()
      ready_ids, _ = ray.wait(
          [a.get_location_and_ids.remote()], timeout=0.01
      )
      assert ready_ids == []
  
imports: import collections
  import os
  import sys
  import time
  import pytest
  import ray
  import ray.cluster_utils
label_desc: Use image node_id 7 for calling a global function with example usage: test_actor_multiple_gpus_from_multiple_tasks(ray_start_cluster) and returns: locations, ray,  tuple, ray,  tuple