code: def test_task_unlimited():
      try:
          address = _init_ray()
          x1 = ray.put(np.zeros(400 * MB, dtype=np.uint8))
          refs = [x1]
          # x1 is spilled.
          x2 = ray.put(np.zeros(400 * MB, dtype=np.uint8))
          x2p = ray.get(x2)
          sentinel = ray.put(np.zeros(100 * MB, dtype=np.uint8))
          check_spilled_mb(address, spilled=400)
  
          @ray.remote
          def consume(refs):
              # triggers fallback allocation, spilling of the sentinel
              ray.get(refs[0])
              check_spilled_mb(
                  address, spilled=500, restored=400, fallback=400
              )
              # triggers fallback allocation.
              return ray.put(np.zeros(400 * MB, dtype=np.uint8))
  
          # round 1
          _ = ray.get(ray.get(consume.remote(refs)))
          check_spilled_mb(
              address, spilled=500, restored=400, fallback=400
          )
  
          del x2p
          del sentinel
      finally:
          ray.shutdown()
  
imports: import numpy
  import json
  import random
  import os
  import shutil
  import platform
  import pytest
  import psutil
  import ray
  from ray._private.test_utils import check_spilled_mb, fetch_prometheus, wait_for_condition
label_desc: Use image node_id 8 for calling a global function with example usage: test_task_unlimited() and returns: ray