Repository: ray-project/ray
File: python/ray/serve/_private/utils.py (Python)

Code before:
    return max(0, timeout_s - time_since_start_s)


def get_all_live_placement_group_names() -> List[str]:
    """Fetch and parse the Ray placement group table for live placement group names.

    Placement groups are filtered based on their `scheduling_state`; any placement
    group not in the "REMOVED" state is considered live.
    """
    placement_group_table = ray.util.placement_group_table()

    live_pg_names = []
    for entry in placement_group_table.values():
        pg_name = entry.get("name", "")
        if (
            pg_name
            and entry.get("stats", {}).get("scheduling_state", "UNKNOWN") != "REMOVED"
        ):
            live_pg_names.append(pg_name)

    return live_pg_names


def get_active_placement_group_ids() -> Set[str]:
    """
    Retrieve the set of placement group IDs referenced by alive actors.

    Returns:
        The set of placement group IDs referenced by alive actors.
    """
    accessor = state._connect_and_get_accessor()
    raw_actors = accessor.get_actor_table(None, "ALIVE")

    occupied: Set[str] = set()
    for raw_actor_bytes in raw_actors:
        actor_data = gcs_pb2.ActorTableData.FromString(raw_actor_bytes)
        if actor_data.HasField("placement_group_id"):
            occupied.add(binary_to_hex(actor_data.placement_group_id))

    return occupied


def get_current_actor_id() -> str:
    """Gets the ID of the calling actor.

    If this is called in a driver, returns "DRIVER."

    If otherwise called outside of an actor, returns an empty string.

    This function hangs when GCS is down due to the `ray.get_runtime_context()`
    call.

Review comment (suggestion):
need to learn more about why we need this. We need to figure out a way around this, one it uses private API from serve and this is disallowed at project level. (b) it is making RPC call to GSC from what it looks like, so we cannot be calling this from every controller iteration loop.

Code after:

    time_since_start_s = curr_time_s - start_time_s
    return max(0, timeout_s - time_since_start_s)


def get_all_live_placement_group_names() -> List[str]:
    """Fetch and parse the Ray placement group table for live placement group names.

    Placement groups are filtered based on their `scheduling_state`; any placement
    group not in the "REMOVED" state is considered live.
    """
    placement_group_table = ray.util.placement_group_table()

    live_pg_names = []
    for entry in placement_group_table.values():
        pg_name = entry.get("name", "")
        if (
            pg_name
            and entry.get("stats", {}).get("scheduling_state", "UNKNOWN") != "REMOVED"
        ):
            live_pg_names.append(pg_name)

    return live_pg_names


def get_active_placement_group_ids() -> Set[str]:
    """
    Retrieve the set of placement group IDs referenced by alive Serve actors.

    Returns:
        The set of placement group IDs referenced by alive Serve actors.
    """
    # TODO (jeffreywang): Move the imports to the top of the file.
    # https://github.com/ray-project/ray/issues/61330
    from ray.util.state import list_actors
    from ray.util.state.common import RAY_MAX_LIMIT_FROM_API_SERVER

    actors = list_actors(
        filters=[
            ("ray_namespace", "=", SERVE_NAMESPACE),
            ("state", "=", "ALIVE"),
        ],
        limit=RAY_MAX_LIMIT_FROM_API_SERVER,
        detail=True,
        raise_on_missing_output=False,
    )

    return {
        actor.placement_group_id
        for actor in actors
        if actor.placement_group_id is not None