code: def test_delete_objects_multi_node(
      multi_node_object_spilling_config, ray_start_cluster
  ):
      # Limit our object store to 75 MiB of memory.
      (
          object_spilling_config,
          temp_folder,
      ) = multi_node_object_spilling_config
  
      cluster = ray_start_cluster
      # Head node.
      cluster.add_node(
          num_cpus=1,
          object_store_memory=75 * 1024 * 1024,
          _system_config={
              "max_io_workers": 2,
              "min_spilling_size": 20 * 1024 * 1024,
              "automatic_object_spilling_enabled": True,
              "object_store_full_delay_ms": 100,
              "object_spilling_config": object_spilling_config,
          },
      )
      ray.init(address=cluster.address)
      # Add 2 worker nodes.
      for _ in range(2):
          cluster.add_node(
              num_cpus=1, object_store_memory=75 * 1024 * 1024
          )
      cluster.wait_for_nodes()
  
      arr = np.random.rand(1024 * 1024)  # 8 MB data
  
      @ray.remote(num_cpus=1)
      class Actor:
          def __init__(self):
              self.replay_buffer = []
  
          def ping(self):
              return
  
          def create_objects(self):
              for _ in range(80):
                  ref = None
                  while ref is None:
                      ref = ray.put(arr)
                      self.replay_buffer.append(ref)
                  # Remove the replay buffer with 60% probability.
                  if random.randint(0, 9) < 6:
                      self.replay_buffer.pop()
  
              # Do random sampling.
              for _ in range(50):
                  ref = random.choice(self.replay_buffer)
                  sample = ray.get(ref, timeout=10)
                  assert np.array_equal(sample, arr)
  
      actors = [Actor.remote() for _ in range(3)]
      ray.get([actor.create_objects.remote() for actor in actors])
  
      def wait_until_actor_dead(actor):
          try:
              ray.get(actor.ping.remote())
          except ray.exceptions.RayActorError:
              return True
          return False
  
      # Kill actors to remove all references.
      for actor in actors:
          ray.kill(actor)
          wait_for_condition(lambda: wait_until_actor_dead(actor))
      # The multi node deletion should work.
      wait_for_condition(lambda: is_dir_empty(temp_folder))
      assert_no_thrashing(cluster.address)
  
imports: import os
  import platform
  import random
  import subprocess
  import sys
  import tempfile
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.tests.test_object_spilling import assert_no_thrashing, is_dir_empty
  from ray._private.external_storage import FileSystemStorage, ExternalStorageRayStorageImpl
label_desc: Use image node_id 5 for calling a global function with example usage: test_delete_objects_multi_node(multi_node_object_spilling_config, ray_start_cluster) and returns: False, True