code: def test_max_pinned_args_memory(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(
          num_cpus=0,
          object_store_memory=200 * 1024 * 1024,
          _system_config={
              "max_task_args_memory_fraction": 0.7,
          },
      )
      ray.init(address=cluster.address)
      cluster.add_node(
          num_cpus=3, object_store_memory=100 * 1024 * 1024
      )
  
      @ray.remote
      def f(arg):
          time.sleep(1)
          return np.zeros(30 * 1024 * 1024, dtype=np.uint8)
  
      # Each task arg takes about 30% of the remote node's memory. We should
      # execute at most 2 at a time to make sure we have room for at least 1 task
      # output.
      x = np.zeros(30 * 1024 * 1024, dtype=np.uint8)
      ray.get([f.remote(ray.put(x)) for _ in range(3)])
  
      @ray.remote
      def large_arg(arg):
          return
  
      # Executing a task whose args are greater than the memory threshold is
      # okay.
      ref = np.zeros(80 * 1024 * 1024, dtype=np.uint8)
      ray.get(large_arg.remote(ref))
  
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 11 for calling a global function with example usage: test_max_pinned_args_memory(ray_start_cluster) and returns: np