code: def test_task_unlimited_multiget_args():
      try:
          address = _init_ray()
          # Too many refs to fit into memory.
          refs = []
          for _ in range(10):
              refs.append(ray.put(np.zeros(200 * MB, dtype=np.uint8)))
          x2 = ray.put(np.zeros(600 * MB, dtype=np.uint8))
          x2p = ray.get(x2)
          check_spilled_mb(address, spilled=2000)
  
          @ray.remote
          def consume(refs):
              # Should work without thrashing.
              ray.get(refs)
              return os.getpid()
  
          ray.get([consume.remote(refs) for _ in range(1000)])
          check_spilled_mb(
              address, spilled=2000, restored=2000, fallback=2000
          )
          del x2p
      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 9 for calling a global function with example usage: test_task_unlimited_multiget_args() and returns: os