code: def test_spill_objects_on_object_transfer(
      object_spilling_config, ray_start_cluster_enabled
  ):
      object_spilling_config, _ = object_spilling_config
      # This test checks that objects get spilled to make room for transferred
      # objects.
      cluster = ray_start_cluster_enabled
      object_size = int(1e7)
      num_objects = 10
      num_tasks = 10
      # Head node can fit all of the objects at once.
      cluster.add_node(
          num_cpus=0,
          object_store_memory=2 * num_tasks * num_objects * object_size,
          _system_config={
              "max_io_workers": 1,
              "automatic_object_spilling_enabled": True,
              "object_store_full_delay_ms": 100,
              "object_spilling_config": object_spilling_config,
              "min_spilling_size": 0,
          },
      )
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      # Worker node can fit 1 tasks at a time.
      cluster.add_node(
          num_cpus=1,
          object_store_memory=1.5 * num_objects * object_size,
      )
      cluster.wait_for_nodes()
  
      @ray.remote
      def foo(*args):
          return
  
      @ray.remote
      def allocate(*args):
          return np.zeros(object_size, dtype=np.uint8)
  
      # Allocate some objects that must be spilled to make room for foo's
      # arguments.
      allocated = [allocate.remote() for _ in range(num_objects)]
      ray.get(allocated)
      print("done allocating")
  
      args = []
      for _ in range(num_tasks):
          task_args = [
              ray.put(np.zeros(object_size, dtype=np.uint8))
              for _ in range(num_objects)
          ]
          args.append(task_args)
  
      # Check that tasks scheduled to the worker node have enough room after
      # spilling.
      tasks = [foo.remote(*task_args) for task_args in args]
      ray.get(tasks)
      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 8 for calling a global function with example usage: test_spill_objects_on_object_transfer(object_spilling_config, ray_start_cluster_enabled) and returns: np