code: def test_fallback_memory(shutdown_only):
      """Test some fallback allocated objects"""
  
      expected_fallback = 6
      expected_in_memory = 5
      obj_size_mb = 20
  
      # So expected_in_memory objects could fit in object store
      delta_mb = 5
      info = ray.init(
          object_store_memory=expected_in_memory * obj_size_mb * MiB
          + delta_mb * MiB,
          _system_config=_SYSTEM_CONFIG,
      )
      obj_refs = [
          ray.put(np.zeros(obj_size_mb * MiB, dtype=np.uint8))
          for _ in range(expected_in_memory)
      ]
  
      # Getting and using the objects to prevent spilling
      in_use_objs = [ray.get(obj) for obj in obj_refs]
  
      # No fallback and spilling yet
      expected = {
          "MMAP_SHM": expected_in_memory * obj_size_mb * MiB,
          "MMAP_DISK": 0,
          "SPILLED": 0,
          "WORKER_HEAP": 0,
      }
  
      wait_for_condition(
          # 2KiB for metadata difference
          lambda: approx_eq_dict_in(
              objects_by_loc(info), expected, 3 * KiB
          ),
          timeout=20,
          retry_interval_ms=500,
      )
  
      # Fallback allocated and make them not spillable
      obj_refs_fallback = []
      in_use_objs_fallback = []
      for _ in range(expected_fallback):
          obj = ray.put(np.zeros(obj_size_mb * MiB, dtype=np.uint8))
          in_use_objs_fallback.append(ray.get(obj))
          obj_refs_fallback.append(obj)
  
          # NOTE(rickyx): I actually wasn't aware this reference would
          # keep the reference count? Removing this line would cause
          # a single object not deleted.
          del obj
  
      # Fallback allocated and still no spilling
      expected = {
          "MMAP_SHM": expected_in_memory * obj_size_mb * MiB,
          "MMAP_DISK": expected_fallback * obj_size_mb * MiB,
          "SPILLED": 0,
          "WORKER_HEAP": 0,
      }
  
      wait_for_condition(
          # 3KiB for metadata difference
          lambda: approx_eq_dict_in(
              objects_by_loc(info), expected, 3 * KiB
          ),
          timeout=20,
          retry_interval_ms=500,
      )
  
      # Free all of them
      del in_use_objs
      del obj_refs
      del in_use_objs_fallback
      del obj_refs_fallback
  
      expected = {
          "MMAP_SHM": 0,
          "MMAP_DISK": 0,
          "SPILLED": 0,
          "WORKER_HEAP": 0,
      }
  
      wait_for_condition(
          # 3KiB for metadata difference
          lambda: approx_eq_dict_in(
              objects_by_loc(info), expected, 3 * KiB
          ),
          timeout=20,
          retry_interval_ms=500,
      )
  
file_path: ray/python/ray/tests/test_object_store_metrics.py
imports: from collections import defaultdict
  import pytest
  from typing import Dict
  import numpy
  import sys
  import requests
  import ray
  from ray._private.test_utils import raw_metrics, wait_for_condition
  from ray._private.worker import RayContext
label_desc: Use image node_id 7 for calling a global function with example usage: test_fallback_memory(shutdown_only) without return types