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

Code before:
    if custom_resources is None:
        custom_resources = {}
    for name, amount in custom_resources.items():
        if amount is None:
            continue
        if amount > 0:
            required_resources[name] = float(amount)

    # If there are no positive resource requirements, the task can run on a
    # zero-resource head node (e.g., num_cpus=0 opt-out), so nothing to check.
    if not required_resources:
        return

    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

    head_resources = head_node.get("Resources", {})
    insufficient = {
        name: (req, head_resources.get(name, 0.0))
        for name, req in required_resources.items()
        if head_resources.get(name, 0.0) < req
    }

    if not insufficient:
        return

    details = "; ".join(
        f"{name} required {req:g} but head has {avail:g}"
        for name, (req, avail) in insufficient.items()
    )

    raise ValueError(
        f"{op_name} must run on the head node (e.g., for local:// paths), "
        f"but the head node doesn't have enough resources: {details}. "
        "Add resources to the head node, switch to a shared filesystem instead "
        "of local://, or set the resource requests on this operation to 0 "
        "(for example, num_cpus=0) so it can run without head resources."
    )



Review comment (suggestion):
For my understanding, do u have a script of when this can occur? (`head_node is None` , BUT `next(...)` doesn't throw a `StopIteration` exception?)

Code after:
        # 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:
        lines = list(filter(bool, reversed(head.splitlines())))
    # Should always have at least one non-empty line in the docstring.
    assert len(lines) > 0
    indent = " " * (len(lines[0]) - len(lines[0].lstrip()))
    # Handle directive.
    message = message.strip("\n")
    if directive is not None:
        base = f"{indent}.. {directive}::\n"
        message = message.replace("\n", "\n" + indent + " " * 4)
        message = base + indent + " " * 4 + message
    else:
        message = indent + message.replace("\n", "\n" + indent)
    # Add two blank lines before/after message, if necessary.
    if insert_after ^ (pattern == "\n\n"):
        # Only two blank lines before message if:
        # 1. Inserting message after pattern and pattern is not two blank lines.
        # 2. Inserting message before pattern and pattern is two blank lines.
        message = "\n\n" + message