code: def test_object_transfer_dump(ray_start_cluster_enabled):
      cluster = ray_start_cluster_enabled
  
      num_nodes = 3
      for i in range(num_nodes):
          cluster.add_node(
              resources={str(i): 1}, object_store_memory=10**9
          )
      ray.init(address=cluster.address)
  
      @ray.remote
      def f(x):
          return
  
      # These objects will live on different nodes.
      object_refs = [
          f._remote(args=[1], resources={str(i): 1})
          for i in range(num_nodes)
      ]
  
      # Broadcast each object from each machine to each other machine.
      for object_ref in object_refs:
          ray.get(
              [
                  f._remote(args=[object_ref], resources={str(i): 1})
                  for i in range(num_nodes)
              ]
          )
  
      # The profiling information only flushes once every second.
      time.sleep(1.1)
  
      transfer_dump = ray._private.state.object_transfer_timeline()
      # Make sure the transfer dump can be serialized with JSON.
      json.loads(json.dumps(transfer_dump))
      assert len(transfer_dump) >= num_nodes**2
      assert (
          len(
              {
                  event["pid"]
                  for event in transfer_dump
                  if event["name"] == "transfer_receive"
              }
          )
          == num_nodes
      )
      assert (
          len(
              {
                  event["pid"]
                  for event in transfer_dump
                  if event["name"] == "transfer_send"
              }
          )
          == num_nodes
      )
  
imports: import json
  import logging
  import os
  import sys
  import time
  import numpy
  import pytest
  import ray._private.profiling
  import ray.cluster_utils
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import client_test_enabled, wait_for_condition
  from ray.exceptions import ObjectFreedError
label_desc: Use image node_id 7 for calling a global function with example usage: test_object_transfer_dump(ray_start_cluster_enabled) without return types