code: def test_many_small_transfers(ray_start_cluster_with_resource):
      cluster, num_nodes = ray_start_cluster_with_resource
  
      @ray.remote
      def f(*args):
          pass
  
      # This function creates 1000 objects on each machine and then transfers
      # each object to every other machine.
      def do_transfers():
          id_lists = []
          for i in range(num_nodes):
              id_lists.append(
                  [
                      f._remote(
                          args=[], kwargs={}, resources={str(i): 1}
                      )
                      for _ in range(1000)
                  ]
              )
          ids = []
          for i in range(num_nodes):
              for j in range(num_nodes):
                  if i == j:
                      continue
                  ids.append(
                      f._remote(
                          args=id_lists[j],
                          kwargs={},
                          resources={str(i): 1},
                      )
                  )
  
          # Wait for all of the transfers to finish.
          ray.get(ids)
  
      do_transfers()
      do_transfers()
      do_transfers()
      do_transfers()
  
imports: import multiprocessing
  import time
  import warnings
  from collections import defaultdict
  import numpy
  import pytest
  import ray
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 6 for calling a global function with example usage: test_many_small_transfers(ray_start_cluster_with_resource) without return types