code: def test_object_unpin(ray_start_cluster):
      nodes = []
      cluster = ray_start_cluster
      head_node = cluster.add_node(
          num_cpus=0,
          object_store_memory=100 * 1024 * 1024,
          _system_config={
              "subscriber_timeout_ms": 100,
              "health_check_initial_delay_ms": 0,
              "health_check_period_ms": 1000,
              "health_check_failure_threshold": 5,
          },
      )
      ray.init(address=cluster.address)
  
      # Add worker nodes.
      for i in range(2):
          nodes.append(
              cluster.add_node(
                  num_cpus=1,
                  resources={f"node_{i}": 1},
                  object_store_memory=100 * 1024 * 1024,
              )
          )
      cluster.wait_for_nodes()
  
      one_mb_array = np.ones(1 * 1024 * 1024, dtype=np.uint8)
      ten_mb_array = np.ones(10 * 1024 * 1024, dtype=np.uint8)
  
      @ray.remote
      class ObjectsHolder:
          def __init__(self):
              self.ten_mb_objs = []
              self.one_mb_objs = []
  
          def put_10_mb(self):
              self.ten_mb_objs.append(ray.put(ten_mb_array))
  
          def put_1_mb(self):
              self.one_mb_objs.append(ray.put(one_mb_array))
  
          def pop_10_mb(self):
              if len(self.ten_mb_objs) == 0:
                  return False
              self.ten_mb_objs.pop()
              return True
  
          def pop_1_mb(self):
              if len(self.one_mb_objs) == 0:
                  return False
              self.one_mb_objs.pop()
              return True
  
      # Head node contains 11MB of data.
      one_mb_arrays = []
      ten_mb_arrays = []
  
      one_mb_arrays.append(ray.put(one_mb_array))
      ten_mb_arrays.append(ray.put(ten_mb_array))
  
      def check_memory(mb):
          return f"Plasma memory usage {mb} MiB" in memory_summary(
              address=head_node.address, stats_only=True
          )
  
      def wait_until_node_dead(node):
          for n in ray.nodes():
              if (
                  n["ObjectStoreSocketName"]
                  == node.address_info["object_store_address"]
              ):
                  return not n["Alive"]
          return False
  
      wait_for_condition(lambda: check_memory(11))
  
      # Pop one mb array and see if it works.
      one_mb_arrays.pop()
      wait_for_condition(lambda: check_memory(10))
  
      # Pop 10 MB.
      ten_mb_arrays.pop()
      wait_for_condition(lambda: check_memory(0))
  
      # Put 11 MB for each actor.
      # actor 1: 1MB + 10MB
      # actor 2: 1MB + 10MB
      actor_on_node_1 = ObjectsHolder.options(
          resources={"node_0": 1}
      ).remote()
      actor_on_node_2 = ObjectsHolder.options(
          resources={"node_1": 1}
      ).remote()
      ray.get(actor_on_node_1.put_1_mb.remote())
      ray.get(actor_on_node_1.put_10_mb.remote())
      ray.get(actor_on_node_2.put_1_mb.remote())
      ray.get(actor_on_node_2.put_10_mb.remote())
      wait_for_condition(lambda: check_memory(22))
  
      # actor 1: 10MB
      # actor 2: 1MB
      ray.get(actor_on_node_1.pop_1_mb.remote())
      ray.get(actor_on_node_2.pop_10_mb.remote())
      wait_for_condition(lambda: check_memory(11))
  
      # The second node is dead, and actor 2 is dead.
      cluster.remove_node(nodes[1], allow_graceful=False)
      wait_for_condition(lambda: wait_until_node_dead(nodes[1]))
      wait_for_condition(lambda: check_memory(10))
  
      # The first actor is dead, so object should be GC'ed.
      ray.kill(actor_on_node_1)
      wait_for_condition(lambda: check_memory(0))
  
imports: import logging
  import os
  import copy
  import platform
  import random
  import signal
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import SignalActor, put_object, wait_for_condition, wait_for_num_actors
  import ray._private.gcs_utils
label_desc: Use image node_id 9 for calling a global function with example usage: test_object_unpin(ray_start_cluster) and returns: unknown, False, True, True, False, False, unknown