code: def test_multiple_raylets(ray_start_cluster):
      # This test will define a bunch of tasks that can only be assigned to
      # specific raylets, and we will check that they are assigned
      # to the correct raylets.
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=11, num_gpus=0)
      cluster.add_node(num_cpus=5, num_gpus=5)
      cluster.add_node(num_cpus=10, num_gpus=1)
      ray.init(address=cluster.address)
      cluster.wait_for_nodes()
  
      # Define a bunch of remote functions that all return the socket name of
      # the plasma store. Since there is a one-to-one correspondence between
      # plasma stores and raylets (at least right now), this can be
      # used to identify which raylet the task was assigned to.
  
      # This must be run on the zeroth raylet.
      @ray.remote(num_cpus=11)
      def run_on_0():
          return (
              ray._private.worker.global_worker.node.plasma_store_socket_name
          )
  
      # This must be run on the first raylet.
      @ray.remote(num_gpus=2)
      def run_on_1():
          return (
              ray._private.worker.global_worker.node.plasma_store_socket_name
          )
  
      # This must be run on the second raylet.
      @ray.remote(num_cpus=6, num_gpus=1)
      def run_on_2():
          return (
              ray._private.worker.global_worker.node.plasma_store_socket_name
          )
  
      # This can be run anywhere.
      @ray.remote(num_cpus=0, num_gpus=0)
      def run_on_0_1_2():
          return (
              ray._private.worker.global_worker.node.plasma_store_socket_name
          )
  
      # This must be run on the first or second raylet.
      @ray.remote(num_gpus=1)
      def run_on_1_2():
          return (
              ray._private.worker.global_worker.node.plasma_store_socket_name
          )
  
      # This must be run on the zeroth or second raylet.
      @ray.remote(num_cpus=8)
      def run_on_0_2():
          return (
              ray._private.worker.global_worker.node.plasma_store_socket_name
          )
  
      def run_lots_of_tasks():
          names = []
          results = []
          for i in range(100):
              index = np.random.randint(6)
              if index == 0:
                  names.append("run_on_0")
                  results.append(run_on_0.remote())
              elif index == 1:
                  names.append("run_on_1")
                  results.append(run_on_1.remote())
              elif index == 2:
                  names.append("run_on_2")
                  results.append(run_on_2.remote())
              elif index == 3:
                  names.append("run_on_0_1_2")
                  results.append(run_on_0_1_2.remote())
              elif index == 4:
                  names.append("run_on_1_2")
                  results.append(run_on_1_2.remote())
              elif index == 5:
                  names.append("run_on_0_2")
                  results.append(run_on_0_2.remote())
          return names, results
  
      client_table = ray.nodes()
      store_names = []
      store_names += [
          client["ObjectStoreSocketName"]
          for client in client_table
          if client["Resources"].get("GPU", 0) == 0
      ]
      store_names += [
          client["ObjectStoreSocketName"]
          for client in client_table
          if client["Resources"].get("GPU", 0) == 5
      ]
      store_names += [
          client["ObjectStoreSocketName"]
          for client in client_table
          if client["Resources"].get("GPU", 0) == 1
      ]
      assert len(store_names) == 3
  
      def validate_names_and_results(names, results):
          for name, result in zip(names, ray.get(results)):
              if name == "run_on_0":
                  assert result in [store_names[0]]
              elif name == "run_on_1":
                  assert result in [store_names[1]]
              elif name == "run_on_2":
                  assert result in [store_names[2]]
              elif name == "run_on_0_1_2":
                  assert result in [
                      store_names[0],
                      store_names[1],
                      store_names[2],
                  ]
              elif name == "run_on_1_2":
                  assert result in [store_names[1], store_names[2]]
              elif name == "run_on_0_2":
                  assert result in [store_names[0], store_names[2]]
              else:
                  raise Exception("This should be unreachable.")
              assert set(ray.get(results)) == set(store_names)
  
      names, results = run_lots_of_tasks()
      validate_names_and_results(names, results)
  
      # Make sure the same thing works when this is nested inside of a task.
  
      @ray.remote
      def run_nested1():
          names, results = run_lots_of_tasks()
          return names, results
  
      @ray.remote
      def run_nested2():
          names, results = ray.get(run_nested1.remote())
          return names, results
  
      names, results = ray.get(run_nested2.remote())
      validate_names_and_results(names, results)
  
imports: import logging
  import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  from ray._private.test_utils import RayTestTimeoutException, wait_for_condition
  from ray.util.placement_group import placement_group
  from ray.util.accelerators import AWS_NEURON_CORE
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 6 for calling a global function with example usage: test_multiple_raylets(ray_start_cluster) and returns: ray, ray, ray, ray, ray, ray, names,  results, names,  results, names,  results