Module `refinery.lib.scripts.js.analysis.cfg`

JavaScript's contribution to the shared control-flow substrate: which node types are which control-flow shape, and where the parts of each are stored.

Everything structural — the graph, the frontier threading, the jump-target and handler stacks, and the shapes themselves — lives in refinery.lib.scripts.analysis.cfg. What remains here is _Builder.statement, the recognition of Js* node types, and the accessors that pull a construct apart. Three of the shape parameters are answered from JavaScript semantics rather than from syntax: switch falls through from one case to the next, an unlabelled break may leave a switch as well as a loop, and a catch carries no type filter, so a throw the guarded block makes gets past it only where binding the caught value can itself throw — see _catch_may_rethrow.

Expand source code Browse git

"""
JavaScript's contribution to the shared control-flow substrate: which node types are which
control-flow shape, and where the parts of each are stored.

Everything structural — the graph, the frontier threading, the jump-target and handler stacks, and
the shapes themselves — lives in `refinery.lib.scripts.analysis.cfg`. What remains here is
`_Builder.statement`, the recognition of `Js*` node types, and the accessors that pull a construct
apart. Three of the shape parameters are answered from JavaScript semantics rather than from syntax:
`switch` falls through from one case to the next, an unlabelled `break` may leave a `switch` as well
as a loop, and a `catch` carries no type filter, so a throw the guarded block makes gets past it only
where binding the caught value can itself throw — see `_catch_may_rethrow`.
"""
from __future__ import annotations

from typing import Sequence

from refinery.lib.scripts import Node
from refinery.lib.scripts.analysis.cfg import (
    ArmFlow,
    CfgBuilder,
    CfgNode,
    ControlFlowGraph,
    ControlFlowModel,
    ElementLocator,
)
from refinery.lib.scripts.analysis.cfg import build_control_flow as _build_control_flow
from refinery.lib.scripts.js.analysis.evaluation import evaluated_later, owns_a_run
from refinery.lib.scripts.js.model import (
    LOOP_NODES,
    JsArrayPattern,
    JsBlockStatement,
    JsBreakStatement,
    JsCatchClause,
    JsContinueStatement,
    JsDoWhileStatement,
    JsExportNamedDeclaration,
    JsForInStatement,
    JsForOfStatement,
    JsForStatement,
    JsIfStatement,
    JsLabeledStatement,
    JsObjectPattern,
    JsReturnStatement,
    JsScript,
    JsSwitchCase,
    JsSwitchStatement,
    JsThrowStatement,
    JsTryStatement,
    JsVariableDeclaration,
    JsWhileStatement,
    JsWithStatement,
)

__all__ = [
    'CfgNode',
    'ControlFlowGraph',
    'ControlFlowModel',
    'ElementLocator',
    'build_cfg',
    'build_control_flow',
    'build_control_flow_model',
    'evaluated_part',
]


def _catch_may_rethrow(handler: JsCatchClause | None) -> bool:
    """
    Whether a throw offered to *handler* may still reach whatever guards the construct.

    A `catch` carries no type filter, so the clause always matches and the throw stops there — with
    one exception, which is why this is a question rather than a constant. Binding the caught value
    is a destructuring assignment when the parameter is a pattern, and destructuring evaluates:
    `catch ({ a })` over a thrown `null` throws again while binding, and that second throw leaves
    the construct. A plain identifier binds by name and an omitted parameter binds nothing, so
    neither can.
    """
    if handler is None:
        return False
    return isinstance(handler.param, (JsObjectPattern, JsArrayPattern))


def _label_name(statement: Node) -> str | None:
    label = getattr(statement, 'label', None)
    return label.name if label is not None else None


def _label_chain(statement: JsLabeledStatement) -> tuple[list[str], Node | None]:
    """
    The labels written in front of a statement, outermost first, and the statement they label. In
    `a: b: while (…)` both name the loop, and `continue a` continues it.
    """
    labels: list[str] = []
    body: Node | None = statement
    while isinstance(body, JsLabeledStatement):
        if (name := _label_name(body)) is not None:
            labels.append(name)
        body = body.body
    return labels, body


class _Builder(CfgBuilder):
    """
    The JavaScript dispatch over `refinery.lib.scripts.analysis.cfg.CfgBuilder`.
    """

    def body_blocks(self, owner: Node) -> Sequence[Sequence[Node]]:
        """
        The one block a JavaScript body is. A concise arrow function's body is an expression rather
        than a block, and it is reported as the single statement it evaluates, as the initializer of
        a field is.
        """
        if isinstance(owner, JsScript):
            return [list(owner.body)]
        body = getattr(owner, 'body', None)
        if isinstance(body, JsBlockStatement):
            return [list(body.body)]
        if isinstance(body, Node):
            return [[body]]
        return [[part for part in owner.children() if evaluated_later(owner, part)]]

    def statement(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]:
        if isinstance(statement, JsBlockStatement):
            return self.sequence(statement.body, frontier)
        if isinstance(statement, JsIfStatement):
            arms: list[Node | None] = [statement.consequent]
            if statement.alternate is not None:
                arms.append(statement.alternate)
            return self.branch_on(
                statement, arms, frontier, exhaustive=statement.alternate is not None)
        if isinstance(statement, JsWhileStatement):
            return self.loop_head_tested(statement, statement.body, frontier)
        if isinstance(statement, JsDoWhileStatement):
            return self.loop_tail_tested(statement, statement.body, frontier)
        if isinstance(statement, JsForStatement):
            return self.loop_counted(
                statement.init, statement.test, statement.update, statement.body, frontier)
        if isinstance(statement, (JsForInStatement, JsForOfStatement)):
            return self.loop_over(
                statement, statement.right, statement.left, statement.body, frontier)
        if isinstance(statement, JsSwitchStatement):
            return self._switch(statement, frontier)
        if isinstance(statement, JsTryStatement):
            handler = statement.handler
            finalizer = statement.finalizer
            return self.guarded(
                statement.block,
                [(handler, handler.body)] if handler is not None else (),
                finalizer,
                list(finalizer.body) if finalizer is not None else (),
                frontier,
                escapes=_catch_may_rethrow(handler),
            )
        if isinstance(statement, JsLabeledStatement):
            labels, body = _label_chain(statement)
            return self.labelled(
                labels,
                body,
                frontier,
                binds_to_body=isinstance(body, (*LOOP_NODES, JsSwitchStatement)),
            )
        if isinstance(statement, JsReturnStatement):
            return self.terminate(statement, frontier, exceptional=False)
        if isinstance(statement, JsThrowStatement):
            return self.terminate(statement, frontier, exceptional=True)
        if isinstance(statement, JsBreakStatement):
            return self.jump_out(statement, _label_name(statement), frontier)
        if isinstance(statement, JsContinueStatement):
            return self.jump_back(statement, _label_name(statement), frontier)
        if isinstance(statement, JsWithStatement):
            node = self.node(statement)
            self.link(frontier, node)
            return self.statement(statement.body, [node]) if statement.body else [node]
        if (declaration := _declaration_of(statement)) is not None:
            return self._in_steps(statement, declaration, frontier)
        return self.opaque(statement, frontier)

    def evaluated(self, element: Node, frontier: list[CfgNode]) -> list[CfgNode]:
        if isinstance(element, JsVariableDeclaration):
            return self._in_steps(element, element, frontier)
        return super().evaluated(element, frontier)

    def _in_steps(
        self,
        element: Node,
        declaration: JsVariableDeclaration,
        frontier: list[CfgNode],
    ) -> list[CfgNode]:
        """
        The nodes of *element*, which evaluates *declaration*: its declarators run one after the
        other, so each but the last is a node of its own, and the node of *element* evaluates the
        last one (`evaluated_part`). In `var s = '…', t = f(s);` the store of `s` completes before
        `t` is computed.
        """
        for declarator in declaration.declarations[:-1]:
            frontier = self.opaque(declarator, frontier)
        return self.opaque(element, frontier)

    def _switch(self, statement: JsSwitchStatement, frontier: list[CfgNode]) -> list[CfgNode]:
        cases = [case for case in statement.cases if isinstance(case, JsSwitchCase)]
        if len(cases) != len(statement.cases):
            return self.opaque(statement, frontier)
        arms: list[Sequence[Node]] = [list(case.body) for case in cases]
        exhaustive = any(case.test is None for case in cases)
        return self.dispatch(
            statement, arms, frontier, arm_flow=ArmFlow.SEQUENTIAL, exhaustive=exhaustive)


def _declaration_of(statement: Node) -> JsVariableDeclaration | None:
    """
    The declaration *statement* evaluates where it is one, standing alone or under `export`.
    """
    declared = statement.declaration if isinstance(statement, JsExportNamedDeclaration) else statement
    return declared if isinstance(declared, JsVariableDeclaration) else None


def evaluated_part(graph: ControlFlowGraph, element: Node) -> Node:
    """
    What the node of *element* evaluates itself: the last declarator of a declaration, standing
    alone, under `export` or in a `for` head, whose other declarators are nodes of their own
    (`_Builder._in_steps`), and *element* otherwise. Every reader of what a node evaluates asks
    here, so a node standing for part of a statement is never judged by the whole of it.
    """
    declaration = _declaration_of(element) or element
    if (
        isinstance(declaration, JsVariableDeclaration)
        and len(declaration.declarations) > 1
        and graph.node_of(declaration.declarations[0]) is not None
    ):
        return declaration.declarations[-1]
    return element


def build_cfg(owner: Node) -> ControlFlowGraph:
    """
    Build the control-flow graph of *owner*, a `refinery.lib.scripts.js.model.JsScript` or a
    function node, over its own body without descending into nested function bodies.
    """
    return _Builder(owner).build()


def build_control_flow(root: JsScript) -> dict[int, ControlFlowGraph]:
    """
    Build one control-flow graph for the script and one for each run owner, a function or a field
    with an initializer (`refinery.lib.scripts.js.analysis.evaluation.owns_a_run`), keyed by the
    owner node's identity.
    """
    return _build_control_flow(root, _Builder, owns_a_run)


def build_control_flow_model(root: JsScript) -> ControlFlowModel:
    """
    Build the `refinery.lib.scripts.analysis.cfg.ControlFlowModel` for a script root — the shared
    control-flow layer the `DominanceModel` and `LivenessModel` consume.
    """
    return ControlFlowModel(build_control_flow(root), evaluated_later)

Functions

def build_cfg(owner)

Build the control-flow graph of owner, a JsScript or a function node, over its own body without descending into nested function bodies.

Expand source code Browse git

def build_cfg(owner: Node) -> ControlFlowGraph:
    """
    Build the control-flow graph of *owner*, a `refinery.lib.scripts.js.model.JsScript` or a
    function node, over its own body without descending into nested function bodies.
    """
    return _Builder(owner).build()

def build_control_flow(root)

Build one control-flow graph for the script and one for each run owner, a function or a field with an initializer (owns_a_run()), keyed by the owner node's identity.

Expand source code Browse git

def build_control_flow(root: JsScript) -> dict[int, ControlFlowGraph]:
    """
    Build one control-flow graph for the script and one for each run owner, a function or a field
    with an initializer (`refinery.lib.scripts.js.analysis.evaluation.owns_a_run`), keyed by the
    owner node's identity.
    """
    return _build_control_flow(root, _Builder, owns_a_run)

def build_control_flow_model(root)

Build the ControlFlowModel for a script root — the shared control-flow layer the DominanceModel and LivenessModel consume.

Expand source code Browse git

def build_control_flow_model(root: JsScript) -> ControlFlowModel:
    """
    Build the `refinery.lib.scripts.analysis.cfg.ControlFlowModel` for a script root — the shared
    control-flow layer the `DominanceModel` and `LivenessModel` consume.
    """
    return ControlFlowModel(build_control_flow(root), evaluated_later)

def evaluated_part(graph, element)

What the node of element evaluates itself: the last declarator of a declaration, standing alone, under export or in a for head, whose other declarators are nodes of their own (_Builder._in_steps), and element otherwise. Every reader of what a node evaluates asks here, so a node standing for part of a statement is never judged by the whole of it.

Expand source code Browse git

def evaluated_part(graph: ControlFlowGraph, element: Node) -> Node:
    """
    What the node of *element* evaluates itself: the last declarator of a declaration, standing
    alone, under `export` or in a `for` head, whose other declarators are nodes of their own
    (`_Builder._in_steps`), and *element* otherwise. Every reader of what a node evaluates asks
    here, so a node standing for part of a statement is never judged by the whole of it.
    """
    declaration = _declaration_of(element) or element
    if (
        isinstance(declaration, JsVariableDeclaration)
        and len(declaration.declarations) > 1
        and graph.node_of(declaration.declarations[0]) is not None
    ):
        return declaration.declarations[-1]
    return element

Classes

class CfgNode (graph, element, successors=<factory>, predecessors=<factory>, is_resumption_hub=False, is_hub_bound=False)

One vertex of a control-flow graph. element is the AST node it stands for — a statement, a loop-head expression whose reads and writes occur at this point, or a step a language gives a node of its own inside a statement — or None for the synthetic entry and exit. successors lists the nodes control may pass to next. graph is the body this vertex belongs to, which is what makes the questions below answerable of a node alone.

is_resumption_hub marks the synthetic fan-out standing for the over-approximate half of a resuming handler — see CfgEdge. A walk that wants only the paths going forward declines to enter one, and declines to leave one, which makes it a projection of the graph rather than a filter on its edges. It is a fact about the node and not about the edges into it, because an edge kind is keyed by the pair of nodes it joins: a plain edge drawn beside a hub edge makes the pair carry both, and a walk declining pairs that carry the hub kind would decline the plain path too. No pair of edges can disagree about what the node is.

is_hub_bound marks a node the precise, forward-only projection does not reach: it sits inside a handler body that resumes the block around it. That block's forward edges join each guarded statement to the one it resumes at; a statement of the handler itself is on neither end of one, and everything it may reach afterwards hangs off the hub. A forward edge it does carry belongs to a block further out — one whose own resuming set guards the statement this handler is written inside — and stands for that block's resumption, not this one, so it says nothing about where this handler carries on. A forward-only walk seeded here would stop dead where the real run carries on, which for a flood is the unsound direction; reachable_forward_from_any reads this and falls back to the hub for such a source.

Both are fields rather than questions asked of graph, and that is a matter of what a walk can afford as much as of where the fact lives. Every layer that had to ask them of a graph it was separately handed could be handed the wrong one — a node lifted from a nested body reads as neither against the graph around it, and the walk that asked takes the wrong branch without anything failing — and one depth-first sweep over a large script asks tens of millions of times. graph remains because a caller naming which body it means is a contract worth checking, and because ControlFlowGraph.hub_bound answers the same question for a caller holding ids.

eq=False because every map in every layer above is keyed by id(node), and two structurally equal statements are two distinct points in the program.

repr=False because a generated repr expands successors and predecessors, and the guard against recursion only covers the node currently being formatted, so a graph re-expands at every join: a chain of five hundred nodes — a script of five hundred statements — exhausts the interpreter's stack, and a handful of branches produces megabytes. A debugger, a failing assertion, or pytest --showlocals would print one.

Expand source code Browse git

@dataclass(eq=False, repr=False)
class CfgNode:
    """
    One vertex of a control-flow graph. `element` is the AST node it stands for — a statement, a
    loop-head expression whose reads and writes occur at this point, or a step a language gives a
    node of its own inside a statement — or `None` for the synthetic entry and exit. `successors`
    lists the nodes control may pass to next. `graph` is the body this vertex belongs to, which is
    what makes the questions below answerable of a node alone.

    `is_resumption_hub` marks the synthetic fan-out standing for the over-approximate half of a
    resuming handler — see `CfgEdge`. A walk that wants only the paths going forward declines to
    enter one, and declines to leave one, which makes it a *projection* of the graph rather than a
    filter on its edges. It is a fact about the node and not about the edges into it, because an
    edge kind is keyed by the pair of nodes it joins: a plain edge drawn beside a hub edge makes the
    pair carry both, and a walk declining pairs that carry the hub kind would decline the plain path
    too. No pair of edges can disagree about what the node is.

    `is_hub_bound` marks a node the precise, forward-only projection does not reach: it sits inside a
    handler body that resumes the block around it. That block's forward edges join each *guarded*
    statement to the one it resumes at; a statement of the handler itself is on neither end of one,
    and everything it may reach afterwards hangs off the hub. A forward edge it does carry belongs to
    a block further out — one whose own resuming set guards the statement this handler is written
    inside — and stands for that block's resumption, not this one, so it says nothing about where
    this handler carries on. A forward-only walk seeded here would stop dead where the real run
    carries on, which for a flood is the unsound direction; `reachable_forward_from_any` reads this
    and falls back to the hub for such a source.

    Both are fields rather than questions asked of *graph*, and that is a matter of what a walk can
    afford as much as of where the fact lives. Every layer that had to ask them of a graph it was
    separately handed could be handed the wrong one — a node lifted from a nested body reads as
    neither against the graph around it, and the walk that asked takes the wrong branch without
    anything failing — and one depth-first sweep over a large script asks tens of millions of times.
    `graph` remains because a caller naming which body it means is a contract worth checking, and
    because `ControlFlowGraph.hub_bound` answers the same question for a caller holding ids.

    `eq=False` because every map in every layer above is keyed by `id(node)`, and two structurally
    equal statements are two distinct points in the program.

    `repr=False` because a generated repr expands `successors` and `predecessors`, and the guard
    against recursion only covers the node currently being formatted, so a graph re-expands at every
    join: a chain of five hundred nodes — a script of five hundred statements — exhausts the
    interpreter's stack, and a handful of branches produces megabytes. A debugger, a failing
    assertion, or `pytest --showlocals` would print one.
    """
    graph: ControlFlowGraph
    element: Node | None
    successors: list[CfgNode] = field(default_factory=list)
    predecessors: list[CfgNode] = field(default_factory=list)
    is_resumption_hub: bool = False
    is_hub_bound: bool = False

Subclasses

Ps1ReportNode

Instance variables

var graph: The type of the None singleton.
var element: The type of the None singleton.
var successors: The type of the None singleton.
var predecessors: The type of the None singleton.
var is_resumption_hub: The type of the None singleton.
var is_hub_bound: The type of the None singleton.

class ControlFlowGraph (owner)

The control-flow graph of one function or script body. entry and exit are synthetic; every other node wraps an AST element reachable through node_of.

Expand source code Browse git

class ControlFlowGraph:
    """
    The control-flow graph of one function or script body. `entry` and `exit` are synthetic; every
    other node wraps an AST element reachable through `node_of`.
    """

    def __init__(self, owner: Node):
        self.owner = owner
        self._node_of: dict[int, CfgNode] = {}
        self._edge_kinds: dict[tuple[int, int], CfgEdge] = {}
        self._hub_bound: set[int] = set()
        self._resuming = False
        self._reached_after: dict[tuple[int, bool], frozenset[int]] = {}
        self._completions: dict[int, tuple[Node, list[CfgNode]]] = {}
        self._spans: dict[int, tuple[Node, range]] = {}
        self._positions: dict[int, int] | None = None
        self._condensations: dict[Projection, Condensation[CfgNode]] = {}
        self.entry = CfgNode(self, None)
        self.exit = CfgNode(self, None)
        self.nodes: list[CfgNode] = [self.entry, self.exit]

    def node_of(self, element: Node) -> CfgNode | None:
        """
        The graph node standing for *element*, or `None` if *element* is not part of this body, or is
        a node the graph does not represent on its own such as a plain expression inside a statement.
        """
        return self._node_of.get(id(element))

    def completion_of(self, statement: Node) -> list[CfgNode] | None:
        """
        The nodes from which normal control falls through past *statement*, a member of a statement
        sequence this graph was built from (`CfgBuilder.sequence`), or `None` for any other element.
        An empty list is a statement control never runs off the end of.
        """
        if (found := self._completions.get(id(statement))) is None or found[0] is not statement:
            return None
        return found[1]

    def built_for(self, statement: Node) -> list[CfgNode] | None:
        """
        The nodes built for *statement*, a member of a statement sequence this graph was built
        from: its own node, those of every construct nested in it, and the synthetic joins and
        resumption slots its build drew; `None` where it is any other element.
        """
        if (found := self._spans.get(id(statement))) is None or found[0] is not statement:
            return None
        span = found[1]
        return self.nodes[span.start:span.stop]

    def builder_of(self, statements: Iterable[Node]) -> Callable[[CfgNode], bool] | None:
        """
        A test for whether a node of this graph was built for one of *statements*, each a member of
        a statement sequence this graph was built from — its own node, those of every construct
        nested in it, and the synthetic joins and resumption slots its build drew — or `None` where
        one of them is any other element.

        Read off the order the builder appends nodes in: building a statement is one recursive call,
        so everything built for it is appended between that call's start and its end, and two such
        stretches are nested or apart, as the statements are. A test is one search among the
        outermost stretches however deep the statements nest. A walk that measures what a run does
        inside some statements asks this of every node it meets to see where the run has left them.
        """
        spans: list[range] = []
        for statement in statements:
            if (found := self._spans.get(id(statement))) is None or found[0] is not statement:
                return None
            spans.append(found[1])
        starts: list[int] = []
        stops: list[int] = []
        for span in sorted(spans, key=lambda span: (span.start, -span.stop)):
            if stops and span.start < stops[-1]:
                continue
            starts.append(span.start)
            stops.append(span.stop)
        positions = self._positions
        if positions is None:
            positions = self._positions = {id(node): index for index, node in enumerate(self.nodes)}

        def holds(node: CfgNode) -> bool:
            position = positions.get(id(node), -1)
            index = bisect_right(starts, position) - 1
            return index >= 0 and position < stops[index]

        return holds

    def edge_kind(self, source: CfgNode, target: CfgNode) -> CfgEdge:
        """
        What the edges from *source* to *target* say about the runs that take them — see `CfgEdge`.
        A pair joined more than once carries the kind of every edge between it, and an unrelated
        pair of nodes reads as `CfgEdge.NORMAL`.
        """
        return self._edge_kinds.get((id(source), id(target)), CfgEdge.NORMAL)

    def completes_into(self, source: CfgNode, target: CfgNode) -> bool:
        """
        Whether control passes from *source* to *target* on a run where *source* completed: a plain
        edge joins them, whatever other edge joins them as well.
        """
        kind = self._edge_kinds.get((id(source), id(target)))
        return kind is None or bool(kind.value & _NORMAL_MASK)

    def is_exceptional(self, source: CfgNode, target: CfgNode) -> bool:
        """
        Whether the error travels along the edge from *source* to *target*: the edge into a handler
        offered the throw, or outward from a set that declined it. This is the question `faults`
        asks — where an error goes — and it is *narrower* than `raise_taken`, because a handler that
        resumes swallows the error and carries on along an edge no error travels.
        """
        return bool(self.edge_kind(source, target).value & _ERROR_CARRYING_MASK)

    def raise_taken(self, source: CfgNode, target: CfgNode) -> bool:
        """
        Whether an edge from *source* to *target* is taken only when *source* throws rather than
        completing normally. A definition *source* makes is not guaranteed to have happened along
        such an edge, so a flow-sensitive analysis must not treat it as a kill there. A pair joined
        by such an edge and by a plain one as well answers for the first, and `completes_into` for
        the second; a consumer reading only this one reads the pair as the run that did not finish,
        which keeps every fact a kill would drop.

        This is the question about *completion*, and it is the one nearly every consumer means. A
        resumption edge answers it although no error travels along it: the statement that resumed
        the block is precisely the one that did not finish.
        """
        return bool(self.edge_kind(source, target).value & _RAISE_TAKEN_MASK)

    def exit_reach(self, source: CfgNode) -> tuple[frozenset[int], frozenset[int]]:
        """
        The ids of the nodes reached from *source* by first leaving it normally, and the ids of
        those reached by first leaving it along an edge `raise_taken` answers for — on a run where
        *source* threw before it completed. A store *source* makes has happened at every node of the
        first set and may not have happened at a node of the second. Once control has left *source*
        normally the store is done, so only the first edge out decides which set a node belongs to,
        and a node control rejoins both ways — the statement after a `try` whose handler swallowed
        the throw — is in both. A caller that needs only the second set asks
        `reached_after_a_throw`.
        """
        return self._reached_after_cached(source, raising=False), self.reached_after_a_throw(source)

    def reached_after_a_throw(self, source: CfgNode) -> frozenset[int]:
        """
        The second set `exit_reach` answers: the ids of the nodes reached from *source* by first
        leaving it along an edge `raise_taken` answers for. Each set is computed on first use and
        kept, so a caller asking only this one never pays for the other, which is most of the graph.
        """
        return self._reached_after_cached(source, raising=True)

    def _reached_after_cached(self, source: CfgNode, *, raising: bool) -> frozenset[int]:
        key = (id(source), raising)
        found = self._reached_after.get(key)
        if found is None:
            found = self._reached_after[key] = self._reached_after_uncached(source, raising=raising)
        return found

    def _reached_after_uncached(self, source: CfgNode, *, raising: bool) -> frozenset[int]:
        seen: set[int] = set()
        stack: list[CfgNode] = []
        leaves = self.raise_taken if raising else self.completes_into
        for successor in source.successors:
            if not leaves(source, successor):
                continue
            if id(successor) not in seen:
                seen.add(id(successor))
                stack.append(successor)
        while stack:
            node = stack.pop()
            for successor in node.successors:
                if id(successor) not in seen:
                    seen.add(id(successor))
                    stack.append(successor)
        return frozenset(seen)

    @property
    def hub_bound(self) -> Set[int]:
        """
        The ids of the nodes `CfgNode.is_hub_bound` is set on, for a caller holding node *ids*
        rather than nodes. Written beside the field by `CfgBuilder.mark_hub_bound`, which is the one
        place either is recorded. `refinery.lib.scripts.analysis.reaching.ReachabilityQuery` is one: its
        candidate sets are ids because the layers above memoize them that way, and turning them back
        into nodes to ask each one is what those caches exist to avoid.
        """
        return self._hub_bound

    @property
    def carries_resumption(self) -> bool:
        """
        Whether any handler of this body resumes the block it guards, which is the only shape over
        which the two projections of `CfgEdge` differ. A body with none — every script that writes
        no `trap`, and every one whose traps rethrow — draws no resumption edge and marks no node
        hub-bound, so a forward walk over it answers exactly what `reachable_from_any` does and may
        be answered by the cheaper one.
        """
        return self._resuming

    def reading(self, projection: Projection) -> Projection:
        """
        The reading a walk over this graph under *projection* takes: *projection* itself where the
        graph `carries_resumption`, and `Projection.MAY` where it does not, since a graph that draws
        no hub is walked alike under both. Everything kept per reading of a graph is keyed by this,
        here and in `refinery.lib.scripts.analysis.dominance.DominatorModel`, so that a condensation
        and a dominance tree asked for under one projection describe one relation.
        """
        return projection if self.carries_resumption else Projection.MAY

    def condensation(self, projection: Projection) -> Condensation[CfgNode]:
        """
        This graph condensed into its strongly connected components under *projection*, computed
        on first use and kept per `reading`, as `_reached_after` and `_positions` are kept: a graph
        does not change once it is built.

        The walk that numbers the components starts at the entry, so every component the entry
        reaches is numbered below every component it does not, whatever order `nodes` lists them
        in: a walk bounded by the number of a node the entry reaches stays within what it reaches.
        """
        reading = self.reading(projection)
        found = self._condensations.get(reading)
        if found is None:
            found = Condensation((self.entry, *self.nodes), reading.successors)
            self._condensations[reading] = found
        return found

Instance variables

var hub_bound

The ids of the nodes CfgNode.is_hub_bound is set on, for a caller holding node ids rather than nodes. Written beside the field by CfgBuilder.mark_hub_bound, which is the one place either is recorded. ReachabilityQuery is one: its candidate sets are ids because the layers above memoize them that way, and turning them back into nodes to ask each one is what those caches exist to avoid.

Expand source code Browse git

@property
def hub_bound(self) -> Set[int]:
    """
    The ids of the nodes `CfgNode.is_hub_bound` is set on, for a caller holding node *ids*
    rather than nodes. Written beside the field by `CfgBuilder.mark_hub_bound`, which is the one
    place either is recorded. `refinery.lib.scripts.analysis.reaching.ReachabilityQuery` is one: its
    candidate sets are ids because the layers above memoize them that way, and turning them back
    into nodes to ask each one is what those caches exist to avoid.
    """
    return self._hub_bound

var carries_resumption

Whether any handler of this body resumes the block it guards, which is the only shape over which the two projections of CfgEdge differ. A body with none — every script that writes no trap, and every one whose traps rethrow — draws no resumption edge and marks no node hub-bound, so a forward walk over it answers exactly what reachable_from_any does and may be answered by the cheaper one.

Expand source code Browse git

@property
def carries_resumption(self) -> bool:
    """
    Whether any handler of this body resumes the block it guards, which is the only shape over
    which the two projections of `CfgEdge` differ. A body with none — every script that writes
    no `trap`, and every one whose traps rethrow — draws no resumption edge and marks no node
    hub-bound, so a forward walk over it answers exactly what `reachable_from_any` does and may
    be answered by the cheaper one.
    """
    return self._resuming

Methods

def node_of(self, element)

The graph node standing for element, or None if element is not part of this body, or is a node the graph does not represent on its own such as a plain expression inside a statement.

Expand source code Browse git

def node_of(self, element: Node) -> CfgNode | None:
    """
    The graph node standing for *element*, or `None` if *element* is not part of this body, or is
    a node the graph does not represent on its own such as a plain expression inside a statement.
    """
    return self._node_of.get(id(element))

def completion_of(self, statement)

The nodes from which normal control falls through past statement, a member of a statement sequence this graph was built from (CfgBuilder.sequence), or None for any other element. An empty list is a statement control never runs off the end of.

Expand source code Browse git

def completion_of(self, statement: Node) -> list[CfgNode] | None:
    """
    The nodes from which normal control falls through past *statement*, a member of a statement
    sequence this graph was built from (`CfgBuilder.sequence`), or `None` for any other element.
    An empty list is a statement control never runs off the end of.
    """
    if (found := self._completions.get(id(statement))) is None or found[0] is not statement:
        return None
    return found[1]

def built_for(self, statement)

The nodes built for statement, a member of a statement sequence this graph was built from: its own node, those of every construct nested in it, and the synthetic joins and resumption slots its build drew; None where it is any other element.

Expand source code Browse git

def built_for(self, statement: Node) -> list[CfgNode] | None:
    """
    The nodes built for *statement*, a member of a statement sequence this graph was built
    from: its own node, those of every construct nested in it, and the synthetic joins and
    resumption slots its build drew; `None` where it is any other element.
    """
    if (found := self._spans.get(id(statement))) is None or found[0] is not statement:
        return None
    span = found[1]
    return self.nodes[span.start:span.stop]

def builder_of(self, statements)

A test for whether a node of this graph was built for one of statements, each a member of a statement sequence this graph was built from — its own node, those of every construct nested in it, and the synthetic joins and resumption slots its build drew — or None where one of them is any other element.

Read off the order the builder appends nodes in: building a statement is one recursive call, so everything built for it is appended between that call's start and its end, and two such stretches are nested or apart, as the statements are. A test is one search among the outermost stretches however deep the statements nest. A walk that measures what a run does inside some statements asks this of every node it meets to see where the run has left them.

Expand source code Browse git

def builder_of(self, statements: Iterable[Node]) -> Callable[[CfgNode], bool] | None:
    """
    A test for whether a node of this graph was built for one of *statements*, each a member of
    a statement sequence this graph was built from — its own node, those of every construct
    nested in it, and the synthetic joins and resumption slots its build drew — or `None` where
    one of them is any other element.

    Read off the order the builder appends nodes in: building a statement is one recursive call,
    so everything built for it is appended between that call's start and its end, and two such
    stretches are nested or apart, as the statements are. A test is one search among the
    outermost stretches however deep the statements nest. A walk that measures what a run does
    inside some statements asks this of every node it meets to see where the run has left them.
    """
    spans: list[range] = []
    for statement in statements:
        if (found := self._spans.get(id(statement))) is None or found[0] is not statement:
            return None
        spans.append(found[1])
    starts: list[int] = []
    stops: list[int] = []
    for span in sorted(spans, key=lambda span: (span.start, -span.stop)):
        if stops and span.start < stops[-1]:
            continue
        starts.append(span.start)
        stops.append(span.stop)
    positions = self._positions
    if positions is None:
        positions = self._positions = {id(node): index for index, node in enumerate(self.nodes)}

    def holds(node: CfgNode) -> bool:
        position = positions.get(id(node), -1)
        index = bisect_right(starts, position) - 1
        return index >= 0 and position < stops[index]

    return holds

def edge_kind(self, source, target)

What the edges from source to target say about the runs that take them — see CfgEdge. A pair joined more than once carries the kind of every edge between it, and an unrelated pair of nodes reads as CfgEdge.NORMAL.

Expand source code Browse git

def edge_kind(self, source: CfgNode, target: CfgNode) -> CfgEdge:
    """
    What the edges from *source* to *target* say about the runs that take them — see `CfgEdge`.
    A pair joined more than once carries the kind of every edge between it, and an unrelated
    pair of nodes reads as `CfgEdge.NORMAL`.
    """
    return self._edge_kinds.get((id(source), id(target)), CfgEdge.NORMAL)

def completes_into(self, source, target)

Whether control passes from source to target on a run where source completed: a plain edge joins them, whatever other edge joins them as well.

Expand source code Browse git

def completes_into(self, source: CfgNode, target: CfgNode) -> bool:
    """
    Whether control passes from *source* to *target* on a run where *source* completed: a plain
    edge joins them, whatever other edge joins them as well.
    """
    kind = self._edge_kinds.get((id(source), id(target)))
    return kind is None or bool(kind.value & _NORMAL_MASK)

def is_exceptional(self, source, target)

Whether the error travels along the edge from source to target: the edge into a handler offered the throw, or outward from a set that declined it. This is the question faults asks — where an error goes — and it is narrower than raise_taken, because a handler that resumes swallows the error and carries on along an edge no error travels.

Expand source code Browse git

def is_exceptional(self, source: CfgNode, target: CfgNode) -> bool:
    """
    Whether the error travels along the edge from *source* to *target*: the edge into a handler
    offered the throw, or outward from a set that declined it. This is the question `faults`
    asks — where an error goes — and it is *narrower* than `raise_taken`, because a handler that
    resumes swallows the error and carries on along an edge no error travels.
    """
    return bool(self.edge_kind(source, target).value & _ERROR_CARRYING_MASK)

def raise_taken(self, source, target)

Whether an edge from source to target is taken only when source throws rather than completing normally. A definition source makes is not guaranteed to have happened along such an edge, so a flow-sensitive analysis must not treat it as a kill there. A pair joined by such an edge and by a plain one as well answers for the first, and completes_into for the second; a consumer reading only this one reads the pair as the run that did not finish, which keeps every fact a kill would drop.

This is the question about completion, and it is the one nearly every consumer means. A resumption edge answers it although no error travels along it: the statement that resumed the block is precisely the one that did not finish.

Expand source code Browse git

def raise_taken(self, source: CfgNode, target: CfgNode) -> bool:
    """
    Whether an edge from *source* to *target* is taken only when *source* throws rather than
    completing normally. A definition *source* makes is not guaranteed to have happened along
    such an edge, so a flow-sensitive analysis must not treat it as a kill there. A pair joined
    by such an edge and by a plain one as well answers for the first, and `completes_into` for
    the second; a consumer reading only this one reads the pair as the run that did not finish,
    which keeps every fact a kill would drop.

    This is the question about *completion*, and it is the one nearly every consumer means. A
    resumption edge answers it although no error travels along it: the statement that resumed
    the block is precisely the one that did not finish.
    """
    return bool(self.edge_kind(source, target).value & _RAISE_TAKEN_MASK)

def exit_reach(self, source)

The ids of the nodes reached from source by first leaving it normally, and the ids of those reached by first leaving it along an edge raise_taken answers for — on a run where source threw before it completed. A store source makes has happened at every node of the first set and may not have happened at a node of the second. Once control has left source normally the store is done, so only the first edge out decides which set a node belongs to, and a node control rejoins both ways — the statement after a try whose handler swallowed the throw — is in both. A caller that needs only the second set asks reached_after_a_throw.

Expand source code Browse git

def exit_reach(self, source: CfgNode) -> tuple[frozenset[int], frozenset[int]]:
    """
    The ids of the nodes reached from *source* by first leaving it normally, and the ids of
    those reached by first leaving it along an edge `raise_taken` answers for — on a run where
    *source* threw before it completed. A store *source* makes has happened at every node of the
    first set and may not have happened at a node of the second. Once control has left *source*
    normally the store is done, so only the first edge out decides which set a node belongs to,
    and a node control rejoins both ways — the statement after a `try` whose handler swallowed
    the throw — is in both. A caller that needs only the second set asks
    `reached_after_a_throw`.
    """
    return self._reached_after_cached(source, raising=False), self.reached_after_a_throw(source)

def reached_after_a_throw(self, source)

The second set exit_reach answers: the ids of the nodes reached from source by first leaving it along an edge raise_taken answers for. Each set is computed on first use and kept, so a caller asking only this one never pays for the other, which is most of the graph.

Expand source code Browse git

def reached_after_a_throw(self, source: CfgNode) -> frozenset[int]:
    """
    The second set `exit_reach` answers: the ids of the nodes reached from *source* by first
    leaving it along an edge `raise_taken` answers for. Each set is computed on first use and
    kept, so a caller asking only this one never pays for the other, which is most of the graph.
    """
    return self._reached_after_cached(source, raising=True)

def reading(self, projection)

The reading a walk over this graph under projection takes: projection itself where the graph carries_resumption, and Projection.MAY where it does not, since a graph that draws no hub is walked alike under both. Everything kept per reading of a graph is keyed by this, here and in DominatorModel, so that a condensation and a dominance tree asked for under one projection describe one relation.

Expand source code Browse git

def reading(self, projection: Projection) -> Projection:
    """
    The reading a walk over this graph under *projection* takes: *projection* itself where the
    graph `carries_resumption`, and `Projection.MAY` where it does not, since a graph that draws
    no hub is walked alike under both. Everything kept per reading of a graph is keyed by this,
    here and in `refinery.lib.scripts.analysis.dominance.DominatorModel`, so that a condensation
    and a dominance tree asked for under one projection describe one relation.
    """
    return projection if self.carries_resumption else Projection.MAY

def condensation(self, projection)

This graph condensed into its strongly connected components under projection, computed on first use and kept per reading, as _reached_after and _positions are kept: a graph does not change once it is built.

The walk that numbers the components starts at the entry, so every component the entry reaches is numbered below every component it does not, whatever order nodes lists them in: a walk bounded by the number of a node the entry reaches stays within what it reaches.

Expand source code Browse git

def condensation(self, projection: Projection) -> Condensation[CfgNode]:
    """
    This graph condensed into its strongly connected components under *projection*, computed
    on first use and kept per `reading`, as `_reached_after` and `_positions` are kept: a graph
    does not change once it is built.

    The walk that numbers the components starts at the entry, so every component the entry
    reaches is numbered below every component it does not, whatever order `nodes` lists them
    in: a walk bounded by the number of a node the entry reaches stays within what it reaches.
    """
    reading = self.reading(projection)
    found = self._condensations.get(reading)
    if found is None:
        found = Condensation((self.entry, *self.nodes), reading.successors)
        self._condensations[reading] = found
    return found

class ControlFlowModel (graphs, apart)

The per-body control-flow graphs of one script, paired with the ElementLocator that maps any AST node to the graph node evaluating it. Built once over the script root — the graphs are purely syntactic and need no semantic model — and shared by every solver layered on it, which would otherwise each rebuild the whole set. apart is the ElementLocator argument of that name.

Expand source code Browse git

class ControlFlowModel:
    """
    The per-body control-flow graphs of one script, paired with the `ElementLocator` that maps any
    AST node to the graph node evaluating it. Built once over the script root — the graphs are purely
    syntactic and need no semantic model — and shared by every solver layered on it, which would
    otherwise each rebuild the whole set. *apart* is the `ElementLocator` argument of that name.
    """

    def __init__(self, graphs: dict[int, ControlFlowGraph], apart: Callable[[Node, Node], bool]):
        self.graphs = graphs
        self._locator = ElementLocator(graphs, apart)

    def graph_of(self, owner: Node) -> ControlFlowGraph | None:
        """
        The control-flow graph owned by *owner* — the script root or a node holding a part that runs
        at a time of its own — or `None` when it owns none.
        """
        return self.graphs.get(id(owner))

    def built_from(self, statement: Node) -> bool:
        """
        Whether *statement* is a member of a statement sequence one of the graphs was built from,
        rather than a statement made since, of which no graph holds anything.
        """
        return any(graph.built_for(statement) is not None for graph in self.graphs.values())

    def node_of(self, element: Node) -> CfgNode | None:
        """
        The control-flow node standing for *element*, or `None` when the graphs do not represent it
        directly. Delegates to the shared `ElementLocator`.
        """
        return self._locator.node_of(element)

    def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
        """
        The graph and node that evaluate *element*, climbing out of any enclosing expression, or
        `None` when it has no enclosing graph node. Delegates to the shared `ElementLocator`.
        """
        return self._locator.locate(element)

Subclasses

Ps1ControlFlowModel

Methods

def graph_of(self, owner)

The control-flow graph owned by owner — the script root or a node holding a part that runs at a time of its own — or None when it owns none.

Expand source code Browse git

def graph_of(self, owner: Node) -> ControlFlowGraph | None:
    """
    The control-flow graph owned by *owner* — the script root or a node holding a part that runs
    at a time of its own — or `None` when it owns none.
    """
    return self.graphs.get(id(owner))

def built_from(self, statement)

Whether statement is a member of a statement sequence one of the graphs was built from, rather than a statement made since, of which no graph holds anything.

Expand source code Browse git

def built_from(self, statement: Node) -> bool:
    """
    Whether *statement* is a member of a statement sequence one of the graphs was built from,
    rather than a statement made since, of which no graph holds anything.
    """
    return any(graph.built_for(statement) is not None for graph in self.graphs.values())

def node_of(self, element)

The control-flow node standing for element, or None when the graphs do not represent it directly. Delegates to the shared ElementLocator.

Expand source code Browse git

def node_of(self, element: Node) -> CfgNode | None:
    """
    The control-flow node standing for *element*, or `None` when the graphs do not represent it
    directly. Delegates to the shared `ElementLocator`.
    """
    return self._locator.node_of(element)

def locate(self, element)

The graph and node that evaluate element, climbing out of any enclosing expression, or None when it has no enclosing graph node. Delegates to the shared ElementLocator.

Expand source code Browse git

def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
    """
    The graph and node that evaluate *element*, climbing out of any enclosing expression, or
    `None` when it has no enclosing graph node. Delegates to the shared `ElementLocator`.
    """
    return self._locator.locate(element)

class ElementLocator (graphs, apart)

Locates an AST node among the per-body control-flow graphs of one script. Built once from the graph set, it maps an element to the graph and node that evaluate it — directly for an element a graph node stands for (node_of), or by climbing to the enclosing statement for one nested inside an expression (locate). Every flow-sensitive layer built on the graphs shares it, so the AST-to-graph mapping and its parent-climb live in one place.

apart is the language's answer to whether a part of a node is evaluated at a time of its own rather than where the node is written, as the body of a function is when it is called and the initializer of a class property each time an instance is constructed. The climb never takes such a step (locate).

Expand source code Browse git

class ElementLocator:
    """
    Locates an AST node among the per-body control-flow graphs of one script. Built once from the
    graph set, it maps an element to the graph and node that evaluate it — directly for an element a
    graph node stands for (`node_of`), or by climbing to the enclosing statement for one nested
    inside an expression (`locate`). Every flow-sensitive layer built on the graphs shares it, so the
    AST-to-graph mapping and its parent-climb live in one place.

    *apart* is the language's answer to whether a part of a node is evaluated at a time of its own
    rather than where the node is written, as the body of a function is when it is called and the
    initializer of a class property each time an instance is constructed. The climb never takes
    such a step (`locate`).
    """

    def __init__(self, graphs: dict[int, ControlFlowGraph], apart: Callable[[Node, Node], bool]):
        self._element_graph: dict[int, ControlFlowGraph] = {}
        self._apart = apart
        for graph in graphs.values():
            for node in graph.nodes:
                if node.element is not None:
                    self._element_graph[id(node.element)] = graph

    def node_of(self, element: Node) -> CfgNode | None:
        """
        The control-flow node standing for *element* in whichever graph owns it, or `None` when
        *element* is not itself a node the graphs represent.
        """
        graph = self._element_graph.get(id(element))
        return graph.node_of(element) if graph is not None else None

    def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
        """
        The graph and node that evaluate *element*, climbing out of any expression it is nested in to
        the enclosing statement or loop head, or `None` when it has no enclosing graph node.

        The climb stops where it would step from a part evaluated at a time of its own to the node
        holding it (*apart*), rather than continuing into the code around that node. Something in
        such a part that no node of its own graph stands for — the default of a parameter, an
        attribute on a body, a part no graph is drawn for at all — is evaluated when that part runs,
        and the statement around the node holding it is not that point. Answering with that
        statement orders the element against code the run may never be beside, which is the false
        claim the per-body split exists to avoid; `None` says the graphs do not place it, and a
        caller reads that as unknown. A node that holds such a part is itself written at a point in
        the code around it, so locating a function climbs out to the statement that mentions it.
        """
        cursor = element
        while True:
            graph = self._element_graph.get(id(cursor))
            if graph is not None and (node := graph.node_of(cursor)) is not None:
                return graph, node
            holder = cursor.parent
            if holder is None or self._apart(holder, cursor):
                return None
            cursor = holder

Methods

def node_of(self, element)

The control-flow node standing for element in whichever graph owns it, or None when element is not itself a node the graphs represent.

Expand source code Browse git

def node_of(self, element: Node) -> CfgNode | None:
    """
    The control-flow node standing for *element* in whichever graph owns it, or `None` when
    *element* is not itself a node the graphs represent.
    """
    graph = self._element_graph.get(id(element))
    return graph.node_of(element) if graph is not None else None

def locate(self, element)

The graph and node that evaluate element, climbing out of any expression it is nested in to the enclosing statement or loop head, or None when it has no enclosing graph node.

The climb stops where it would step from a part evaluated at a time of its own to the node holding it (apart), rather than continuing into the code around that node. Something in such a part that no node of its own graph stands for — the default of a parameter, an attribute on a body, a part no graph is drawn for at all — is evaluated when that part runs, and the statement around the node holding it is not that point. Answering with that statement orders the element against code the run may never be beside, which is the false claim the per-body split exists to avoid; None says the graphs do not place it, and a caller reads that as unknown. A node that holds such a part is itself written at a point in the code around it, so locating a function climbs out to the statement that mentions it.

Expand source code Browse git

def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
    """
    The graph and node that evaluate *element*, climbing out of any expression it is nested in to
    the enclosing statement or loop head, or `None` when it has no enclosing graph node.

    The climb stops where it would step from a part evaluated at a time of its own to the node
    holding it (*apart*), rather than continuing into the code around that node. Something in
    such a part that no node of its own graph stands for — the default of a parameter, an
    attribute on a body, a part no graph is drawn for at all — is evaluated when that part runs,
    and the statement around the node holding it is not that point. Answering with that
    statement orders the element against code the run may never be beside, which is the false
    claim the per-body split exists to avoid; `None` says the graphs do not place it, and a
    caller reads that as unknown. A node that holds such a part is itself written at a point in
    the code around it, so locating a function climbs out to the statement that mentions it.
    """
    cursor = element
    while True:
        graph = self._element_graph.get(id(cursor))
        if graph is not None and (node := graph.node_of(cursor)) is not None:
            return graph, node
        holder = cursor.parent
        if holder is None or self._apart(holder, cursor):
            return None
        cursor = holder