class_docstr: This autoscaler simulator consists of a few components.
  
  State is stored in 3 main data structures:
      * Resource management state is stored in self.ip_to_nodes
      * The scheduler's work queue is stored in self.work_queue
      * An event queue which acts as the simulation's "timeline" in
        self.event_queue
  
  
  The logic is organized into 3 functions (and their helpers):
      * self.run_autoscaler plays the role of `monitor.py` and translates
        resource management state for load_metrics to consume.
      * self.schedule is the only consumer of the work queue. It dispatches
        work to the appropriate schedulers, which mutate cluster state and
        produce events for the event queue.
      * self.process_event is the sole consumer of the event queue. It
        dispatches work to the appropriate event handlers.
  
  There are 3 main ways of interacting with the simulator:
      * simulator.submit: To submit tasks
      * simulator.step: To go to the next "event"
      * task/actor/placement group start/done callbacks
class_docstr_tok: ["This","autoscaler","simulator","consists","of","a","few","components",".","State","is","stored","in","3","main","data","structures",":","*","Resource","management","state","is","stored","in","self.ip_to_nodes","*","The","scheduler","'s","work","queue","is","stored","in","self.work_queue","*","An","event","queue","which","acts","as","the","simulation","'s","``","timeline","''","in","self.event_queue","The","logic","is","organized","into","3","functions","(","and","their","helpers",")",":","*","self.run_autoscaler","plays","the","role","of","`","monitor.py","`","and","translates","resource","management","state","for","load_metrics","to","consume",".","*","self.schedule","is","the","only","consumer","of","the","work","queue",".","It","dispatches","work","to","the","appropriate","schedulers",",","which","mutate","cluster","state","and","produce","events","for","the","event","queue",".","*","self.process_event","is","the","sole","consumer","of","the","event","queue",".","It","dispatches","work","to","the","appropriate","event","handlers",".","There","are","3","main","ways","of","interacting","with","the","simulator",":","*","simulator.submit",":","To","submit","tasks","*","simulator.step",":","To","go","to","the","next","``","event","''","*","task\/actor\/placement","group","start\/done","callbacks"]
code: def _schedule_placement_group(self, pg, nodes):
      # This scheduling algorithm is bad, but it is approximately as bad as
      # the real placement group scheduler.
      to_allocate = []
      if (
          pg.strategy == PlacementStrategy.STRICT_PACK
          or pg.strategy == PlacementStrategy.PACK
      ):
          combined = collections.defaultdict(float)
          for bundle in pg.bundles:
              for k, v in bundle.items():
                  combined[k] += v
          ip, node_to_run = self._get_node_to_run(combined, nodes)
          if node_to_run is None:
              return False
          to_allocate.append((combined, ip))
      elif (
          pg.strategy == PlacementStrategy.STRICT_SPREAD
          or pg.strategy == PlacementStrategy.SPREAD
      ):
          # TODO (Alex): More accurate handling of non-STRICT_PACK groups.
          remaining_nodes = nodes.copy()
          for bundle in pg.bundles:
              ip, node_to_run = self._get_node_to_run(
                  bundle, remaining_nodes
              )
              if node_to_run is None:
                  return False
              del remaining_nodes[ip]
              to_allocate.append((bundle, ip))
  
      for bundle, ip in to_allocate:
          node = self.ip_to_nodes[ip]
          node.allocate(bundle)
      pg.start_time = self.virtual_time
      end_time = self.virtual_time + pg.duration
      self.event_queue.put(
          Event(end_time, SIMULATOR_EVENT_PG_DONE, (pg, to_allocate))
      )
      if pg.start_callback:
          pg.start_callback()
      return True
  
imports: import collections
  import copy
  import logging
  import yaml
  import tempfile
  from typing import Dict, Callable, List
  import shutil
  from queue import PriorityQueue
  import unittest
  import pytest
  import ray
  from ray.tests.test_autoscaler import MockProvider, MockProcessRunner, MockGcsClient, mock_raylet_id, MockAutoscaler
  from ray.tests.test_resource_demand_scheduler import MULTI_WORKER_CLUSTER
  from ray.autoscaler._private.providers import _NODE_PROVIDERS, _clear_provider_cache
  from ray.autoscaler._private.load_metrics import LoadMetrics
  from ray.autoscaler._private.node_launcher import NodeLauncher
  from ray.autoscaler.tags import TAG_RAY_USER_NODE_TYPE, TAG_RAY_NODE_KIND, NODE_KIND_HEAD
  from ray.autoscaler._private.constants import AUTOSCALER_UPDATE_INTERVAL_S
  from ray.autoscaler._private.cli_logger import cli_logger
  from ray.core.generated.common_pb2 import Bundle, PlacementStrategy
  from ray._private.gcs_utils import PlacementGroupTableData
label_desc: Use image node_id 7 for calling the Simulator obj's underlying member method code with example usage: obj._schedule_placement_group(pg, nodes) and returns: True, False, False