code: def test_ray_get_task_args_deadlock(ray_start_cluster):
      cluster = ray_start_cluster
      object_size = int(6e6)
      num_objects = 10
      # Head node can fit all of the objects at once.
      cluster.add_node(
          num_cpus=0, object_store_memory=4 * num_objects * object_size
      )
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      # Worker node can only fit 1 task at a time.
      cluster.add_node(
          num_cpus=1,
          object_store_memory=1.5 * num_objects * object_size,
      )
      cluster.wait_for_nodes()
  
      @ray.remote
      def foo(*args):
          return
  
      @ray.remote
      def test_deadlock(get_args, task_args):
          foo.remote(*task_args)
          ray.get(get_args)
  
      for i in range(5):
          start = time.time()
          get_args = [
              ray.put(np.zeros(object_size, dtype=np.uint8))
              for _ in range(num_objects)
          ]
          task_args = [
              ray.put(np.zeros(object_size, dtype=np.uint8))
              for _ in range(num_objects)
          ]
          ray.get(test_deadlock.remote(get_args, task_args))
          print(f"round {i} finished in {time.time() - start}")
  
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 12 for calling a global function with example usage: test_ray_get_task_args_deadlock(ray_start_cluster) without return types