code: def test_local_scheduling_first(ray_start_cluster):
      cluster = ray_start_cluster
      num_cpus = 8
      # Disable worker caching.
      cluster.add_node(
          num_cpus=num_cpus,
          _system_config={
              "worker_lease_timeout_milliseconds": 0,
          },
      )
      cluster.add_node(num_cpus=num_cpus)
      ray.init(address=cluster.address)
  
      @ray.remote(num_cpus=1)
      def f():
          time.sleep(0.01)
          return ray._private.worker.global_worker.node.unique_id
  
      def local():
          return (
              ray.get(f.remote())
              == ray._private.worker.global_worker.node.unique_id
          )
  
      # Wait for a worker to get started.
      wait_for_condition(local)
  
      # Check that we are scheduling locally while there are resources available.
      for i in range(20):
          assert local()
  
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 5 for calling a global function with example usage: test_local_scheduling_first(ray_start_cluster) and returns: ray, unknown