code: def test_hybrid_policy(ray_start_cluster):
      cluster = ray_start_cluster
  
      num_cpus = 10
      cluster.add_node(
          num_cpus=num_cpus,
          memory=num_cpus,
          _system_config={
              "scheduler_top_k_absolute": 1,
              "scheduler_top_k_fraction": 0,
          },
      )
      cluster.add_node(
          num_cpus=num_cpus,
          memory=num_cpus,
      )
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      # `block_task` ensures that scheduled tasks do not return until all are
      # running.
      block_task = Semaphore.remote(0)
      # `block_driver` ensures that the driver does not allow tasks to continue
      # until all are running.
      block_driver = Semaphore.remote(0)
  
      # Add the memory resource because the cpu will be released in the ray.get
      @ray.remote(num_cpus=1, memory=1)
      def get_node():
          ray.get(block_driver.release.remote())
          ray.get(block_task.acquire.remote())
          return ray._private.worker.global_worker.current_node_id
  
      # Below the hybrid threshold we pack on the local node first.
      refs = [get_node.remote() for _ in range(5)]
      ray.get([block_driver.acquire.remote() for _ in refs])
      ray.get([block_task.release.remote() for _ in refs])
      nodes = ray.get(refs)
      assert len(set(nodes)) == 1
  
      # We pack the second node to the hybrid threshold.
      refs = [get_node.remote() for _ in range(10)]
      ray.get([block_driver.acquire.remote() for _ in refs])
      ray.get([block_task.release.remote() for _ in refs])
      nodes = ray.get(refs)
      counter = collections.Counter(nodes)
      for node_id in counter:
          print(f"{node_id}: {counter[node_id]}")
          assert counter[node_id] == 5
  
      # Once all nodes are past the hybrid threshold we round robin.
      # TODO (Alex): Ideally we could schedule less than 20 nodes here, but the
      # policy is imperfect if a resource report interrupts the process.
      refs = [get_node.remote() for _ in range(20)]
      ray.get([block_driver.acquire.remote() for _ in refs])
      ray.get([block_task.release.remote() for _ in refs])
      nodes = ray.get(refs)
      counter = collections.Counter(nodes)
      for node_id in counter:
          print(f"{node_id}: {counter[node_id]}")
          assert counter[node_id] == 10, counter
  
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 3 for calling a global function with example usage: test_hybrid_policy(ray_start_cluster) and returns: ray