class_docstr: Resource manager using placement groups as the resource backend.
  
  This manager will use placement groups to fulfill resource requests. Requesting
  a resource will schedule the placement group. Acquiring a resource will
  return a ``PlacementGroupAcquiredResources`` that can be used to schedule
  Ray tasks and actors on the placement group. Freeing an acquired resource
  will destroy the associated placement group.
  
  Ray core does not emit events when resources are available. Instead, the
  scheduling state has to be periodically updated.
  
  Per default, placement group scheduling state is refreshed every time when
  resource state is inquired, but not more often than once every ``update_interval_s``
  seconds. Alternatively, staging futures can be retrieved (and awaited) with
  ``get_resource_futures()`` and state update can be force with ``update_state()``.
  
  Args:
      update_interval_s: Minimum interval in seconds between updating scheduling
          state of placement groups.
class_docstr_tok: ["Resource","manager","using","placement","groups","as","the","resource","backend",".","This","manager","will","use","placement","groups","to","fulfill","resource","requests",".","Requesting","a","resource","will","schedule","the","placement","group",".","Acquiring","a","resource","will","return","a","``","PlacementGroupAcquiredResources","``","that","can","be","used","to","schedule","Ray","tasks","and","actors","on","the","placement","group",".","Freeing","an","acquired","resource","will","destroy","the","associated","placement","group",".","Ray","core","does","not","emit","events","when","resources","are","available",".","Instead",",","the","scheduling","state","has","to","be","periodically","updated",".","Per","default",",","placement","group","scheduling","state","is","refreshed","every","time","when","resource","state","is","inquired",",","but","not","more","often","than","once","every","``","update_interval_s","``","seconds",".","Alternatively",",","staging","futures","can","be","retrieved","(","and","awaited",")","with","``","get_resource_futures","(",")","``","and","state","update","can","be","force","with","``","update_state","(",")","``",".","Args",":","update_interval_s",":","Minimum","interval","in","seconds","between","updating","scheduling","state","of","placement","groups","."]
code: def __init__(self, update_interval_s: float = 0.1):
      # Internally, the placement group lifecycle is like this:
      # - Resources are requested with ``request_resources()``
      # - A placement group is scheduled ("staged")
      # - A ``PlacementGroup.ready()`` future is scheduled ("staging future")
      # - We update the scheduling state when we need to
      #   (e.g. when ``has_resources_ready()`` is called)
      # - When staging futures resolve, a placement group is moved from "staging"
      #   to "ready"
      # - When a resource request is canceled, we remove a placement group from
      #   "staging". If there are not staged placement groups
      #   (because they are already "ready"), we remove one from "ready" instead.
      # - When a resource is acquired, the pg is removed from "ready" and moved
      #   to "acquired"
      # - When a resource is freed, the pg is removed from "acquired" and destroyed
  
      # Mapping of placement group to request
      self._pg_to_request: Dict[PlacementGroup, ResourceRequest] = {}
  
      # PGs that are staged but not "ready", yet (i.e. not CREATED)
      self._request_to_staged_pgs: Dict[
          ResourceRequest, Set[PlacementGroup]
      ] = defaultdict(set)
  
      # PGs that are CREATED and can be used by tasks and actors
      self._request_to_ready_pgs: Dict[
          ResourceRequest, Set[PlacementGroup]
      ] = defaultdict(set)
  
      # Staging futures used to update internal state.
      # We keep a double mapping here for better lookup efficiency.
      self._staging_future_to_pg: Dict[
          ray.ObjectRef, PlacementGroup
      ] = dict()
      self._pg_to_staging_future: Dict[
          PlacementGroup, ray.ObjectRef
      ] = dict()
  
      # Set of acquired PGs. We keep track of these here to make sure we
      # only free PGs that this manager managed.
      self._acquired_pgs: Set[PlacementGroup] = set()
  
      # Minimum time between updates of the internal state
      self.update_interval_s = update_interval_s
      self._last_update = time.monotonic() - self.update_interval_s - 1
  
imports: import time
  from collections import defaultdict
  from typing import Dict, List, Optional, Set
  from dataclasses import dataclass
  import ray
  from ray.air.execution.resources.request import ResourceRequest, AcquiredResources, RemoteRayEntity
  from ray.air.execution.resources.resource_manager import ResourceManager
  from ray.util.annotations import DeveloperAPI
  from ray.util.placement_group import PlacementGroup, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 1 to create a new PlacementGroupResourceManager object from inherited base classes: ResourceManager with example: obj = PlacementGroupResourceManager(update_interval_s)