code: def test_locality_aware_leasing_cached_objects(ray_start_cluster):
      # This test ensures that a task will run where its task dependencies are
      # located, even when those objects aren't primary copies.
      cluster = ray_start_cluster
  
      # Disable worker caching so worker leases are not reused, and disable
      # inlining of return objects so return objects are always put into Plasma.
      cluster.add_node(
          num_cpus=1,
          _system_config={
              "worker_lease_timeout_milliseconds": 0,
              "max_direct_call_object_size": 0,
          },
      )
      # Use a custom resource for pinning tasks to a node.
      cluster.add_node(num_cpus=1, resources={"pin_worker1": 1})
      worker2 = cluster.add_node(
          num_cpus=1, resources={"pin_worker2": 1}
      )
      ray.init(address=cluster.address)
  
      @ray.remote
      def f():
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote
      def g(x):
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote
      def h(x, y):
          return ray._private.worker.global_worker.node.unique_id
  
      # f_obj1 pinned on worker1.
      f_obj1 = f.options(resources={"pin_worker1": 1}).remote()
      # f_obj2 pinned on worker2.
      f_obj2 = f.options(resources={"pin_worker2": 1}).remote()
      # f_obj1 cached copy pulled to worker 2 in order to execute g() task.
      ray.get(g.options(resources={"pin_worker2": 1}).remote(f_obj1))
      # Confirm that h is scheduled onto worker 2, since it should have the
      # primary copy of f_obj12 and a cached copy of f_obj1.
      assert ray.get(h.remote(f_obj1, f_obj2)) == worker2.unique_id
  
imports: import collections
  import logging
  import platform
  import subprocess
  import sys
  import time
  import unittest
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  import ray.util.accelerators
  from ray._private.internal_api import memory_summary
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray._private.test_utils import Semaphore, SignalActor, object_memory_usage, get_metric_check_condition, wait_for_condition, MetricSamplePattern
label_desc: Use image node_id 10 for calling a global function with example usage: test_locality_aware_leasing_cached_objects(ray_start_cluster) and returns: ray, ray, ray