Repository: ray-project/ray
File: python/ray/data/_internal/util.py (Python)

Code before:
    head node via ``NodeAffinitySchedulingStrategy``. If the head node was
    intentionally started with zero logical resources (a common practice to
    avoid OOMs), those tasks become unschedulable. Detect this upfront and
    raise a clear error with remediation steps.
    """

    # Ray defaults to reserving 1 CPU per task when num_cpus isn't provided.
    num_cpus = ray_remote_args.get("num_cpus", required_num_cpus)
    num_gpus = ray_remote_args.get("num_gpus", required_num_gpus)
    memory = ray_remote_args.get("memory", required_memory)

    # Resource keys follow the Resources map of ray.nodes() (e.g., CPU, GPU, memory).
    required_resources: Dict[str, float] = {}
    required_resources["CPU"] = float(num_cpus)
    required_resources["GPU"] = float(num_gpus)
    required_resources["memory"] = float(memory)

    # Include any additional custom resources requested.
    custom_resources = ray_remote_args.get("resources", {})
    for name, amount in custom_resources.items():
        if amount is None:
            continue
        try:
            amount = float(amount)
        except (TypeError, ValueError) as err:
            raise ValueError(f"Invalid resource amount for '{name}': {amount}") from err
        required_resources[name] = amount

    head_node = next(
        (
            node
            for node in ray.nodes()
            if node.get("Alive")
            and "node:__internal_head__" in node.get("Resources", {})
        ),
        None,
    )
    if not head_node:
        # The head node metadata is unavailable (e.g., during shutdown). Fall back
        # to the default behavior and let Ray surface its own error.
        return

    # Build a map of required vs available resources on the head node.
    head_resources: Dict[str, float] = head_node.get("Resources", {})
    # Map: resource name -> (required, available).
    insufficient: Dict[str, Tuple[float, float]] = {}
    for name, req in required_resources.items():
        avail = head_resources.get(name, 0.0)
        if avail < req:
            insufficient[name] = (req, avail)


Review comment (refactor):
No need to changes anything since I think this might conflict with an earlier review comment, but I felt surprised that we're validating custom resources types in this abstraction. I guess I would've assumed a valid `ray_remote_args` is more of a pre-condition and we validate `ray_remote_args` earlier, if at all

Code after:
    message: str,
    pattern: str,
    insert_after: bool = True,
    directive: Optional[str] = None,
    skip_matches: int = 0,
) -> str:
    if "\n" in message:
        raise ValueError(
            "message shouldn't contain any newlines, since this function will insert "
            f"its own linebreaks when text wrapping: {message}"
        )

    doc = obj.__doc__.strip()
    if not doc:
        doc = ""

    if pattern == "" and insert_after:
        # Empty pattern + insert_after means that we want to append the message to the
        # end of the docstring.
        head = doc
        tail = ""
    else:
        tail = doc
        i = tail.find(pattern)
        skip_matches_left = skip_matches
        while i != -1:
            if insert_after:
                # Set offset to the first character after the pattern.
                offset = i + len(pattern)
            else:
                # Set offset to the first character in the matched line.
                offset = tail[:i].rfind("\n") + 1
            head = tail[:offset]
            tail = tail[offset:]
            skip_matches_left -= 1
            if skip_matches_left <= 0:
                break
            elif not insert_after:
                # Move past the found pattern, since we're skipping it.
                tail = tail[i - offset + len(pattern) :]
            i = tail.find(pattern)
        else:
            raise ValueError(
                f"Pattern {pattern} not found after {skip_matches} skips in docstring "
                f"{doc}"
            )
    # Get indentation of the to-be-inserted text.
    after_lines = list(filter(bool, tail.splitlines()))
    if len(after_lines) > 0:
        lines = after_lines
    else: