code: def test_worker_task_refs(ray_start_regular):
      address = ray_start_regular["address"]
  
      @ray.remote
      def f(y):
          from ray._private.internal_api import memory_summary
  
          x_id = ray.put("HI")
          info = memory_summary(address)
          del x_id
          return info
  
      x_id = f.remote(np.zeros(100000))
      info = ray.get(x_id)
      print(info)
      assert num_objects(info) == 4, info
      # Task argument plus task return ids.
      assert count(info, TASK_CALL_OBJ) == 2, info
      assert count(info, DRIVER_PID) == 2, info
      assert count(info, WORKER_PID) == 2, info
      assert count(info, LOCAL_REF) == 2, info
      assert count(info, PINNED_IN_MEMORY) == 1, info
      assert count(info, PUT_OBJ) == 1, info
      assert count(info, DESER_TASK_ARG) == 1, info
      assert count(info, UNKNOWN_SIZE) == 1, info
  
      print(ray_start_regular)
      info = memory_summary(address)
      print(info)
      assert num_objects(info) == 1, info
      assert count(info, DRIVER_PID) == 1, info
      assert count(info, TASK_CALL_OBJ) == 1, info
      assert count(info, UNKNOWN_SIZE) == 0, info
      assert count(info, x_id.hex()) == 1, info
  
      del x_id
      info = memory_summary(address)
      assert num_objects(info) == 0, info
  
file_path: ray/python/ray/tests/test_memstat.py
imports: import os
  import time
  import numpy
  import pytest
  import ray
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import Semaphore, wait_for_condition
  from ray.cluster_utils import Cluster, cluster_not_supported
label_desc: Use image node_id 5 for calling a global function with example usage: test_worker_task_refs(ray_start_regular) and returns: info