code: def test_pull_bundles_admission_control_dynamic(
      enable_mac_large_object_store, ray_start_cluster
  ):
      # This test is the same as test_pull_bundles_admission_control, except that
      # the object store's capacity starts off higher and is later consumed
      # dynamically by concurrent workers.
      cluster = ray_start_cluster
      object_size = int(6e6)
      num_objects = 20
      num_tasks = 20
      # Head node can fit all of the objects at once.
      cluster.add_node(
          num_cpus=0,
          object_store_memory=2 * num_tasks * num_objects * object_size,
      )
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      # Worker node can fit 2 tasks at a time.
      cluster.add_node(
          num_cpus=1,
          object_store_memory=2.5 * num_objects * object_size,
      )
      cluster.wait_for_nodes()
  
      @ray.remote
      def foo(i, *args):
          print("foo", i)
          return
  
      @ray.remote
      def allocate(i):
          print("allocate", i)
          return np.zeros(object_size, dtype=np.uint8)
  
      args = []
      for _ in range(num_tasks):
          task_args = [
              ray.put(np.zeros(object_size, dtype=np.uint8))
              for _ in range(num_objects)
          ]
          args.append(task_args)
  
      allocated = [allocate.remote(i) for i in range(num_objects)]
      ray.get(allocated)
  
      tasks = [
          foo.remote(i, *task_args) for i, task_args in enumerate(args)
      ]
      ray.get(tasks)
      del allocated
  
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 10 for calling a global function with example usage: test_pull_bundles_admission_control_dynamic(enable_mac_large_object_store, ray_start_cluster) and returns: np