code: def test_coschedule_actors_and_tasks(serve_instance):
      """Test that actor/tasks are placed in the replica's placement group by default."""
  
      @ray.remote(num_cpus=1)
      class TestActor:
          def get_pg(self) -> PlacementGroup:
              return get_current_placement_group()
  
      @ray.remote
      def get_pg():
          return get_current_placement_group()
  
      @serve.deployment(
          # Bundles have space for one additional 1-CPU actor.
          placement_group_bundles=[{"CPU": 1}, {"CPU": 1}],
      )
      class Parent:
          def run_test(self):
              # First actor should be scheduled in the placement group without issue.
              a1 = TestActor.remote()
              assert (
                  ray.get(a1.get_pg.remote())
                  == get_current_placement_group()
              )
  
              # Second actor can't be placed because there are no more resources in the
              # placement group (the first actor is occupying the second bundle).
              a2 = TestActor.remote()
              ready, _ = ray.wait([a2.get_pg.remote()], timeout=0.1)
              assert len(ready) == 0
              ray.kill(a2)
  
              # Second actor can be successfully scheduled outside the placement group.
              a3 = TestActor.options(
                  scheduling_strategy=PlacementGroupSchedulingStrategy(
                      placement_group=None
                  )
              ).remote()
              assert ray.get(a3.get_pg.remote()) is None
  
              # A zero-CPU task can be scheduled in the placement group.
              assert (
                  ray.get(get_pg.options(num_cpus=0).remote())
                  == get_current_placement_group()
              )
  
              # A two-CPU task cannot fit in the placement group.
              with pytest.raises(ValueError):
                  get_pg.options(num_cpus=2).remote()
  
              # A two-CPU task can be scheduled outside the placement group.
              assert (
                  ray.get(
                      get_pg.options(
                          num_cpus=2,
                          scheduling_strategy=PlacementGroupSchedulingStrategy(
                              placement_group=None
                          ),
                      ).remote()
                  )
                  is None
              )
  
      h = serve.run(Parent.bind())
      h.run_test.remote().result()
  
file_path: ray/python/ray/serve/tests/test_replica_placement_group.py
imports: import os
  import sys
  import pytest
  import ray
  from ray import serve
  from ray._private.test_utils import wait_for_condition
  from ray.serve._private.utils import get_all_live_placement_group_names
  from ray.serve.context import _get_global_client
  from ray.util.placement_group import PlacementGroup, get_current_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 9 for calling a global function with example usage: test_coschedule_actors_and_tasks(serve_instance) and returns: get_current_placement_group, get_current_placement_group