code: def test_placement_group_parallel_submission(
      ray_start_cluster, scheduling_strategy
  ):
      @ray.remote(resources={"custom_resource": 1})
      def task(input):
          return input
  
      @ray.remote(num_cpus=0)
      def manage_tasks(input):
          pg = ray.util.placement_group(
              [{"custom_resource": 1, "CPU": 1}],
              strategy=scheduling_strategy,
          )
          ray.get(pg.ready())
          pg_strategy = ray.util.scheduling_strategies.PlacementGroupSchedulingStrategy(
              placement_group=pg
          )
  
          obj_ref = task.options(
              scheduling_strategy=pg_strategy
          ).remote(input)
          ray.get(obj_ref)
  
          ray.util.remove_placement_group(pg)
          return "OK"
  
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1, resources={"custom_resource": 20})
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      # Test all tasks will not hang
      ray.get([manage_tasks.remote(i) for i in range(20)], timeout=50)
  
file_path: ray/python/ray/tests/test_placement_group_5.py
imports: import sys
  import pytest
  import ray
  from ray.tests.test_placement_group import are_pairwise_unique
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 5 for calling a global function with example usage: test_placement_group_parallel_submission(ray_start_cluster, scheduling_strategy) and returns: input, str