code: def test_fetch_local(ray_start_cluster_head):
      cluster = ray_start_cluster_head
      cluster.add_node(num_cpus=2, object_store_memory=75 * 1024 * 1024)
      signal_actor = SignalActor.remote()
  
      @ray.remote
      def put():
          ray.wait([signal_actor.wait.remote()])
          return bytearray(40 * 1024 * 1024)  # 40 MB data
  
      local_ref = ray.put(bytearray(40 * 1024 * 1024))
      remote_ref = put.remote()
      # Data is not ready in any node
      (ready_ref, remaining_ref) = ray.wait(
          [remote_ref], timeout=2, fetch_local=False
      )
      assert (0, 1) == (len(ready_ref), len(remaining_ref))
      ray.wait([signal_actor.send.remote()])
  
      # Data is ready in some node, but not local node.
      (ready_ref, remaining_ref) = ray.wait(
          [remote_ref], fetch_local=False
      )
      assert (1, 0) == (len(ready_ref), len(remaining_ref))
      (ready_ref, remaining_ref) = ray.wait(
          [remote_ref], timeout=2, fetch_local=True
      )
      assert (0, 1) == (len(ready_ref), len(remaining_ref))
      del local_ref
      (ready_ref, remaining_ref) = ray.wait(
          [remote_ref], fetch_local=True
      )
      assert (1, 0) == (len(ready_ref), len(remaining_ref))
  
imports: import logging
  import os
  import pickle
  import random
  import sys
  import time
  import pytest
  import ray
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, client_test_enabled, get_error_message, run_string_as_driver
label_desc: Use image node_id 17 for calling a global function with example usage: test_fetch_local(ray_start_cluster_head) and returns: bytearray