<commit_before>import time
from typing import Dict
from typing import List
from typing import Optional
from typing import Union

import ray
from ray._raylet import ObjectRef
from ray._raylet import PlacementGroupID
from ray._private.utils import hex_to_binary
from ray.util.annotations import PublicAPI, DeveloperAPI
from ray.ray_constants import to_memory_units
from ray._private.client_mode_hook import client_mode_should_convert
from ray._private.client_mode_hook import client_mode_wrap

bundle_reservation_check = None


# We need to import this method to use for ready API.
# But ray.remote is only available in runtime, and
# if we define this method inside ready method, this function is
# exported whenever ready is called, which can impact performance,
# https://github.com/ray-project/ray/issues/6240.
def _export_bundle_reservation_check_method_if_needed():
    global bundle_reservation_check
    if bundle_reservation_check:
        return

    @ray.remote(num_cpus=0, max_calls=0)
    def bundle_reservation_check_func(placement_group):
        return placement_group

    bundle_reservation_check = bundle_reservation_check_func


@PublicAPI
class PlacementGroup:
    """A handle to a placement group."""

    @staticmethod
    def empty() -> "PlacementGroup":
        return PlacementGroup(PlacementGroupID.nil())

    def __init__(self,
                 id: PlacementGroupID,
                 bundle_cache: Optional[List[Dict]] = None):
        self.id = id
        self.bundle_cache = bundle_cache

    def ready(self) -> ObjectRef:
        """Returns an ObjectRef to check ready status.

        This API runs a small dummy task to wait for placement group creation.
        It is compatible to ray.get and ray.wait.

        Example:

        >>> pg = placement_group([{"CPU": 1}])
            ray.get(pg.ready())

        >>> pg = placement_group([{"CPU": 1}])
            ray.wait([pg.ready()], timeout=0)
        """
        self._fill_bundle_cache_if_needed()

        _export_bundle_reservation_check_method_if_needed()

        assert len(self.bundle_cache) != 0, (
            "ready() cannot be called on placement group object with a "
            "bundle length == 0, current bundle length: "
            f"{len(self.bundle_cache)}")

        return bundle_reservation_check.options(
            placement_group=self, resources={
                "bundle": 0.001
            }).remote(self)

    def wait(self, timeout_seconds: Union[float, int]) -> bool:
        """Wait for the placement group to be ready within the specified time.
        Args:
             timeout_seconds(float|int): Timeout in seconds.
        Return:
             True if the placement group is created. False otherwise.
        """
        return _call_placement_group_ready(self.id, timeout_seconds)

    @property
    def bundle_specs(self) -> List[Dict]:
        """List[Dict]: Return bundles belonging to this placement group."""
        self._fill_bundle_cache_if_needed()
        return self.bundle_cache

    @property
    def bundle_count(self) -> int:
        self._fill_bundle_cache_if_needed()
        return len(self.bundle_cache)

    def to_dict(self) -> dict:
        """Convert this placement group into a dict for purposes of json
        serialization.

        Used when passing a placement group as an option to a Ray client remote
        function. See set_task_options in util/client/common.py.

        Return:
            Dictionary with json-serializable keys representing the placemnent
            group.
        """
        # Placement group id is converted to a hex /string/ to make it
        # serializable.
        return {"id": self.id.hex(), "bundle_cache": self.bundle_cache}

    @staticmethod
    def from_dict(pg_dict: dict) -> "PlacementGroup":
        """Instantiate and return a PlacementGroup from its json-serializable
        dict representation.

        Used by Ray Client on server-side to deserialize placement group
        option. See decode_options in util/client/server/server.py.

        Args:
            serializable_form(dict): Dictionary representing a placement group.
        Return:
            A placement group made from the data in the input dict.
        """
        # Validate serialized dict
        assert isinstance(pg_dict, dict)
        assert pg_dict.keys() == {"id", "bundle_cache"}
        # The value associated to key "id" is a hex string.
        assert isinstance(pg_dict["id"], str)
        if pg_dict["bundle_cache"] is not None:
            assert isinstance(pg_dict["bundle_cache"], list)

        # Deserialize and return a Placement Group.
        id_bytes = bytes.fromhex(pg_dict["id"])
        pg_id = PlacementGroupID(id_bytes)
        bundle_cache = pg_dict["bundle_cache"]
        return PlacementGroup(pg_id, bundle_cache)

    def _fill_bundle_cache_if_needed(self) -> None:
        if not self.bundle_cache:
            self.bundle_cache = _get_bundle_cache(self.id)


@client_mode_wrap
def _call_placement_group_ready(pg_id: PlacementGroupID,
                                timeout_seconds: int) -> bool:
    worker = ray.worker.global_worker
    worker.check_connected()

    return worker.core_worker.wait_placement_group_ready(
        pg_id, timeout_seconds)


@client_mode_wrap
def _get_bundle_cache(pg_id: PlacementGroupID) -> List[Dict]:
    # Since creating placement group is async, it is
    # possible table is not ready yet. To avoid the
    # problem, we should keep trying with timeout.
    TIMEOUT_SECOND = 30
    WAIT_INTERVAL = 0.05
    timeout_cnt = 0
    worker = ray.worker.global_worker
    worker.check_connected()

    while timeout_cnt < int(TIMEOUT_SECOND / WAIT_INTERVAL):
        pg_info = ray.state.state.placement_group_table(pg_id)
        if pg_info:
            return list(pg_info["bundles"].values())
        time.sleep(WAIT_INTERVAL)
        timeout_cnt += 1

    raise RuntimeError(
        "Couldn't get the bundle information of placement group id "
        f"{id} in {TIMEOUT_SECOND} seconds. It is likely "
        "because GCS server is too busy.")
<commit_msg>remove the legacy retry mechanism (#18589)<commit_after>from typing import Dict
from typing import List
from typing import Optional
from typing import Union

import ray
from ray._raylet import ObjectRef
from ray._raylet import PlacementGroupID
from ray._private.utils import hex_to_binary
from ray.util.annotations import PublicAPI, DeveloperAPI
from ray.ray_constants import to_memory_units
from ray._private.client_mode_hook import client_mode_should_convert
from ray._private.client_mode_hook import client_mode_wrap

bundle_reservation_check = None


# We need to import this method to use for ready API.
# But ray.remote is only available in runtime, and
# if we define this method inside ready method, this function is
# exported whenever ready is called, which can impact performance,
# https://github.com/ray-project/ray/issues/6240.
def _export_bundle_reservation_check_method_if_needed():
    global bundle_reservation_check
    if bundle_reservation_check:
        return

    @ray.remote(num_cpus=0, max_calls=0)
    def bundle_reservation_check_func(placement_group):
        return placement_group

    bundle_reservation_check = bundle_reservation_check_func


@PublicAPI
class PlacementGroup:
    """A handle to a placement group."""

    @staticmethod
    def empty() -> "PlacementGroup":
        return PlacementGroup(PlacementGroupID.nil())

    def __init__(self,
                 id: PlacementGroupID,
                 bundle_cache: Optional[List[Dict]] = None):
        self.id = id
        self.bundle_cache = bundle_cache

    def ready(self) -> ObjectRef:
        """Returns an ObjectRef to check ready status.

        This API runs a small dummy task to wait for placement group creation.
        It is compatible to ray.get and ray.wait.

        Example:

        >>> pg = placement_group([{"CPU": 1}])
            ray.get(pg.ready())

        >>> pg = placement_group([{"CPU": 1}])
            ray.wait([pg.ready()], timeout=0)
        """
        self._fill_bundle_cache_if_needed()

        _export_bundle_reservation_check_method_if_needed()

        assert len(self.bundle_cache) != 0, (
            "ready() cannot be called on placement group object with a "
            "bundle length == 0, current bundle length: "
            f"{len(self.bundle_cache)}")

        return bundle_reservation_check.options(
            placement_group=self, resources={
                "bundle": 0.001
            }).remote(self)

    def wait(self, timeout_seconds: Union[float, int]) -> bool:
        """Wait for the placement group to be ready within the specified time.
        Args:
             timeout_seconds(float|int): Timeout in seconds.
        Return:
             True if the placement group is created. False otherwise.
        """
        return _call_placement_group_ready(self.id, timeout_seconds)

    @property
    def bundle_specs(self) -> List[Dict]:
        """List[Dict]: Return bundles belonging to this placement group."""
        self._fill_bundle_cache_if_needed()
        return self.bundle_cache

    @property
    def bundle_count(self) -> int:
        self._fill_bundle_cache_if_needed()
        return len(self.bundle_cache)

    def to_dict(self) -> dict:
        """Convert this placement group into a dict for purposes of json
        serialization.

        Used when passing a placement group as an option to a Ray client remote
        function. See set_task_options in util/client/common.py.

        Return:
            Dictionary with json-serializable keys representing the placemnent
            group.
        """
        # Placement group id is converted to a hex /string/ to make it
        # serializable.
        return {"id": self.id.hex(), "bundle_cache": self.bundle_cache}

    @staticmethod
    def from_dict(pg_dict: dict) -> "PlacementGroup":
        """Instantiate and return a PlacementGroup from its json-serializable
        dict representation.

        Used by Ray Client on server-side to deserialize placement group
        option. See decode_options in util/client/server/server.py.

        Args:
            serializable_form(dict): Dictionary representing a placement group.
        Return:
            A placement group made from the data in the input dict.
        """
        # Validate serialized dict
        assert isinstance(pg_dict, dict)
        assert pg_dict.keys() == {"id", "bundle_cache"}
        # The value associated to key "id" is a hex string.
        assert isinstance(pg_dict["id"], str)
        if pg_dict["bundle_cache"] is not None:
            assert isinstance(pg_dict["bundle_cache"], list)

        # Deserialize and return a Placement Group.
        id_bytes = bytes.fromhex(pg_dict["id"])
        pg_id = PlacementGroupID(id_bytes)
        bundle_cache = pg_dict["bundle_cache"]
        return PlacementGroup(pg_id, bundle_cache)

    def _fill_bundle_cache_if_needed(self) -> None:
        if not self.bundle_cache:
            self.bundle_cache = _get_bundle_cache(self.id)


@client_mode_wrap
def _call_placement_group_ready(pg_id: PlacementGroupID,
                                timeout_seconds: int) -> bool:
    worker = ray.worker.global_worker
    worker.check_connected()

    return worker.core_worker.wait_placement_group_ready(
        pg_id, timeout_seconds)


@client_mode_wrap
def _get_bundle_cache(pg_id: PlacementGroupID) -> List[Dict]:
    worker = ray.worker.global_worker
    worker.check_connected()

    return list(
        ray.state.state.placement_group_table(pg_id)["bundles"].values())

{'fraction_non_alphanumeric': 0.08070477009024495, 'fraction_numerical': 0.0036097980232058444, 'mean_word_length': 2.7926988265971318, 'pattern_counts': {'":': 14, '<': 4, '<?xml version=': 0, '>': 35, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1667704', 'n_tokens_mistral': 3258, 'n_tokens_neox': 3156, 'n_words': 965}