Module refinery.lib.scripts.analysis.cfg
Per-body control-flow graphs, derived from an AST and independent of the language it came from.
Each function — and the script itself — gets one ControlFlowGraph: a graph whose nodes wrap the
statements and loop-head expressions the body evaluates, and the steps a language evaluates one
after the other inside a statement, connected by the order in which control may pass between them.
Sequential flow, the branches of a conditional, loop back-edges, the non-local jumps a
break/continue/return performs, and exceptional edges from any point inside a guarded block
to the handler that would catch a throw.
The graph is keyed to AST node identity (node_of) and is a disposable, per-body view — the tree
stays the spine. It is conservative by construction: where modelling control flow precisely would
be intricate (the order of evaluation inside an expression, the exact point a statement throws) the
graph adds edges rather than omits them, so an analysis reading it sees at least every path the
program can take. Nested function bodies are not descended into; each
has its own graph.
What a language supplies is the dispatch, not the shapes. CfgBuilder holds the frontier
threading, the jump-target stack, the handler stack and one method per control-flow shape —
branch_on, loop_head_tested, loop_over, loop_tail_tested, loop_counted, dispatch,
guarded, labelled, jump_out, jump_back, terminate. A subclass implements statement,
recognises its own node types, pulls the parts out of them, and calls the shape that matches. No
method here reads a field a language declares, which is what lets two languages whose for loops
share nothing but their meaning share this code.
The shapes are parameterised where languages genuinely differ rather than being duplicated: what an
arm of a dispatch reaches when control runs off its end is an ArmFlow, and each of its members
is some language's answer to the same construct.
Expand source code Browse git
"""
Per-body control-flow graphs, derived from an AST and independent of the language it came from.
Each function — and the script itself — gets one `ControlFlowGraph`: a graph whose nodes wrap the
statements and loop-head expressions the body evaluates, and the steps a language evaluates one
after the other inside a statement, connected by the order in which control may pass between them.
Sequential flow, the branches of a conditional, loop back-edges, the non-local jumps a
`break`/`continue`/`return` performs, and *exceptional* edges from any point inside a guarded block
to the handler that would catch a throw.
The graph is keyed to AST node identity (`node_of`) and is a disposable, per-body view — the tree
stays the spine. It is *conservative by construction*: where modelling control flow precisely would
be intricate (the order of evaluation inside an expression, the exact point a statement throws) the
graph adds edges rather than omits them, so an analysis reading it sees at least every path the
program can take. Nested function bodies are not descended into; each
has its own graph.
**What a language supplies is the dispatch, not the shapes.** `CfgBuilder` holds the frontier
threading, the jump-target stack, the handler stack and one method per control-flow *shape* —
`branch_on`, `loop_head_tested`, `loop_over`, `loop_tail_tested`, `loop_counted`, `dispatch`,
`guarded`, `labelled`, `jump_out`, `jump_back`, `terminate`. A subclass implements `statement`,
recognises its own node types, pulls the parts out of them, and calls the shape that matches. No
method here reads a field a language declares, which is what lets two languages whose `for` loops
share nothing but their meaning share this code.
The shapes are parameterised where languages genuinely differ rather than being duplicated: what an
arm of a `dispatch` reaches when control runs off its end is an `ArmFlow`, and each of its members
is some language's answer to the same construct.
"""
from __future__ import annotations
import enum
from bisect import bisect_right
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import AbstractSet as Set
from typing import Callable, Generic, Hashable, Iterable, Iterator, Mapping, Sequence, TypeVar
from refinery.lib.scripts import Node
from refinery.lib.scripts.analysis.components import Condensation
_Label = TypeVar('_Label', bound=Hashable)
class CfgEdge(enum.Flag):
"""
What an edge says about the run that takes it, beyond the two nodes it joins.
`NORMAL` — control simply passed on, the source having done whatever it does.
`ERROR_CARRYING` — the source threw and the error travels here: the edge into the handler that
is offered it, and onward from a handler set that may decline. A definition the source makes is
not guaranteed to have happened along one of these.
`RESUMPTION_HUB`, `RESUMPTION_FORWARD` — the two projections of a handler that *resumes* the
block it guards rather than ending it. The engine carries on at the statement after the one that
threw, and no graph knows which statement that was, so the shape is carried twice: the hub is
the over-approximation, joining every resumption point to every statement of the block, and the
forward edges are the precise half, joining each statement to the one it would resume at.
The two are *not* interchangeable, and reading one where the other is meant is how this goes
wrong. Every run the forward edges draw the hub draws too; the hub draws more, and what it adds
is the claim that a resumption point reaches the statements written *above* it — which is the
claim the forward half exists to drop. An analysis asking whether *any* resumption path exists
reads the hub, since dropping a path there would let it call a live statement unreachable; one
asking what a point reaches *going forward* reads the forward edges and skips the hub, since
keeping the backward reach there would poison a whole trap-guarded script from one leak — see
`reachable_forward_from_any`.
The distinction the resumption kinds force is that `ERROR_CARRYING` conflates two bits which
coincide nowhere else: *taken only when the source threw*, and *the error object travels here*.
A resumption edge is the first that is the former and not the latter — the handler swallowed the
error — so a consumer reasoning about whether the source completed must read `RAISE_TAKEN`, and
only one reasoning about where an error went may read `ERROR_CARRYING` alone.
Which edges of the hub carry the kind follows from that and is not symmetric. The edges *out*
of a hub do: their target is reached only because something threw. The edge *into* one does
not, and is plain — its source is the handler body running to its end, which is a statement
that completed, and a store it makes did happen. Reading the way in as raise-taken says no read
below a `trap` can ever observe a store the handler made.
A pair of nodes carries the kinds of every edge drawn between it, `NORMAL` among them, so a
pair joined both ways answers for both. The last statement of a `try` body falls through into
the `finally` that guards it and carries its error there too; with no bit of its own, the plain
edge would read as error-carrying, and every walk over plain control flow would lose the path.
A pair joined by plain edges alone is stored as nothing at all, which reads as `NORMAL`.
"""
NORMAL = enum.auto()
ERROR_CARRYING = enum.auto()
RESUMPTION_HUB = enum.auto()
RESUMPTION_FORWARD = enum.auto()
#: The kinds an edge is taken along only on runs where its source threw rather than completing. A
#: store the source makes may not have happened at the other end of one, whichever of the three it
#: is, which is the one question every flow-sensitive consumer asks of the kind.
RAISE_TAKEN = CfgEdge.ERROR_CARRYING | CfgEdge.RESUMPTION_HUB | CfgEdge.RESUMPTION_FORWARD
#: The raw bit masks the two hot edge-kind predicates test. `enum.Flag.__and__` rebuilds a member on
#: every call — its `_get_value`/`__call__`/`__new__` cost dominates a graph flood, which asks these
#: of millions of edges — so the predicates read the stored member's `.value` and mask it as a plain
#: integer. The result is identical; only the construction of a throwaway member is skipped.
_NORMAL_MASK = CfgEdge.NORMAL.value
_ERROR_CARRYING_MASK = CfgEdge.ERROR_CARRYING.value
_RAISE_TAKEN_MASK = RAISE_TAKEN.value
class Projection(enum.Enum):
"""
Which reading of a resuming handler a walk over a control-flow graph is asked for — the two
halves `CfgEdge` describes, named so that a walk states which one it means.
`MAY` — every path the graph draws, the resumption hub among them. A resumption point is taken
to reach every statement of its block, the ones written above it included. This is the reading a
caller asking whether something *can* happen has to have: dropping a path there would let it
call a live statement unreachable, or a store that is read dead.
`FORWARD` — the paths a run takes going forward. The hub is declined, so a resumption reaches
only the statement control would resume at and what follows it. This is the reading a flood
needs: under `MAY` one leak anywhere in a trap-guarded block poisons the whole of it, the
statements above the leak included.
Neither is the safe one. `MAY` claims runs that cannot happen and `FORWARD` drops the fact that a
handler body carries on somewhere the forward edges do not name, so which of them fails closed
depends on what is asked. That is why every walk here names one rather than taking a default.
The hub is declined as a *node*, and in both directions. Filtering the edges instead would let a
walk cross a hub the moment some other edge joined the same pair, and would strand a backward
walk inside a hub it can never leave.
**The forward reading is sound by excision.** Take any run: the segment where it is inside a
resuming handler body is exactly the part the forward edges do not draw, and cutting that segment
out leaves a path over the guarded statements that the forward edges do draw — the statement that
threw, then the one after it. So every ordering the forward projection reports holds of the real
run, over a subset of the nodes the run visited. What it does *not* report is where a statement
of the handler body itself leads, which is why a source inside one is walked under `MAY` instead
(`CfgNode.is_hub_bound`).
"""
MAY = enum.auto()
FORWARD = enum.auto()
def successors(self, node: CfgNode) -> list[CfgNode]:
"""
The nodes control may pass to from *node* under this reading.
"""
if self is Projection.MAY:
return node.successors
if node.is_resumption_hub:
return []
return [target for target in node.successors if not target.is_resumption_hub]
def predecessors(self, node: CfgNode) -> list[CfgNode]:
"""
The nodes control may arrive at *node* from under this reading.
"""
if self is Projection.MAY:
return node.predecessors
if node.is_resumption_hub:
return []
return [source for source in node.predecessors if not source.is_resumption_hub]
@property
def declines_the_hub(self) -> bool:
"""
Whether a walk under this reading refuses to stand at a resumption hub, which is the whole of
what the two readings disagree about and the form a walk over millions of edges can afford to
ask it in. `successors` and `predecessors` say the same thing by building a list, which is
what a caller wanting an adjacency it can iterate twice wants and what a single depth-first
sweep must not pay per node.
"""
return self is Projection.FORWARD
class ArmFlow(enum.Enum):
"""
What an arm of a multi-way branch may reach when control runs off its end.
The three members are three languages' answers to the same construct, and reading one as another
either invents paths that cannot be taken or drops paths that can — the second of which is the
direction that lets an analysis call live code unreachable.
`EXCLUSIVE` — at most one arm ever runs, so an arm's exits leave the construct.
`SEQUENTIAL` — an arm that runs off its end enters the *next* arm's body unconditionally, which
is C-style fallthrough and what JavaScript's `switch` does.
`CUMULATIVE` — every arm is tested in turn and every matching one runs, so an arm's exits may
reach *any* later arm, not only the next. PowerShell's `switch` is this: it does not fall
through, it keeps matching, and a `break` is what stops it.
"""
EXCLUSIVE = enum.auto() # noqa
SEQUENTIAL = enum.auto() # noqa
CUMULATIVE = enum.auto() # noqa
@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
def flood(
sources: Iterable[CfgNode], *, forward: bool, projection: Projection,
) -> set[int]:
"""
The ids of the nodes reachable from *any* of *sources*, each source included: over successor
edges when *forward* and over predecessor edges otherwise, and over the projection that declines
the resumption hub when *projected*.
The one walk every reachability question over these graphs is asked through, so that a caller
choosing a direction or a projection chooses it by naming an argument rather than by writing a
fourth copy of the sweep that quietly disagrees with the other three about what an edge means.
Neither dimension has a default: `forward` decides whether the answer is *what this reaches* or
*what reaches this*, and `projection` decides how a resumption is read. A caller that did not
think about either asked the wrong question.
**The hub-bound fallback is a property of a source, so it is applied here.** A forward walk
under `Projection.FORWARD` seeded at a node the graph calls `CfgNode.is_hub_bound` is walked
with the hub followed instead, because the forward edges carry nothing for such a source and the
precise walk would stop dead at the resumption point — the fail-open direction, and the one a
flood cannot afford. It is applied to the *sources* and never to a node the walk merely met,
which would follow the hub out of any handler body the sweep happened to pass through. The two
sweeps share one `seen`, so the second never re-walks what the first already reached; the
first walks on through every other source it meets, and the second starts only from the sources
the first did not reach, because what the precise walk reaches from a node is a part of what the
hub-followed walk reaches from it.
Walked as one depth-first sweep seeded with every source rather than a union of per-source
walks: the union grows the same answer at the cost of one full walk per source, where a single
sweep visits each node once however many sources reach it. The result is a fresh set the caller
owns, so it may intersect or discard it in place.
A graph none of whose handlers resume draws no hub, so the two readings walk it identically and
the screen is dropped before the sweep begins. That is nearly every script, and this is the one
walk every layer above reaches the graphs through — asking a projection per node instead costs
the whole run more than the resumption edges do.
"""
coarse, precise, declines = _split_sources(
distinct(sources), forward=forward, projection=projection)
seen: set[int] = {id(node) for node in coarse}
if coarse:
_sweep(coarse, seen, forward=True, declines=False)
pending = [node for node in precise if id(node) not in seen]
seen.update(id(node) for node in pending)
_sweep(pending, seen, forward=forward, declines=declines)
return seen
def walk_reading(source: CfgNode, *, forward: bool, projection: Projection) -> Projection:
"""
The reading `flood` walks *source* alone under, which is `_split_sources`' rule for one source,
named as `ControlFlowGraph.reading` names it. Two walks of one graph under one reading walk one
relation, which is what a caller combining their answers over the condensation of that reading
(`ControlFlowGraph.condensation`) must know: the components reached from that of *source* hold
exactly the nodes `flood` returns from it. That holds for *source* and not for every node
sharing its component, since another node of it may be walked under the other reading.
"""
coarse, _, _ = _split_sources([source], forward=forward, projection=projection)
return Projection.MAY if coarse else source.graph.reading(projection)
def _split_sources(
sources: list[CfgNode], *, forward: bool, projection: Projection,
) -> tuple[list[CfgNode], list[CfgNode], bool]:
"""
The rule `flood` documents, applied to *sources*: those walked with the hub followed, because
the forward reading carries nothing for them; those walked under *projection*; and whether that
walk declines the hub at all, which it need not where no source's graph draws one. Every walk
that floods from sources reads it here, so no two of them can disagree about which source takes
which walk.
"""
declines = projection.declines_the_hub and any(node.graph.carries_resumption for node in sources)
if declines and forward:
coarse = [node for node in sources if node.is_hub_bound]
if coarse:
return coarse, [node for node in sources if not node.is_hub_bound], True
return [], sources, declines
def _sweep(
stack: list[CfgNode], seen: set[int], *, forward: bool, declines: bool,
) -> None:
"""
One depth-first sweep of `flood`, growing *seen* in place and consuming *stack*.
*declines* is the `Projection.declines_the_hub` screen, hoisted out of the loop because it is
the same for every edge of one sweep. It is asked of the node the walk stands at as well as of
each neighbour, which is what makes this the same projection `Projection.successors` and
`Projection.predecessors` build a list for: a hub is declined as a *node*, so a walk seeded at
one leaves it by no edge at all.
"""
while stack:
node = stack.pop()
if declines and node.is_resumption_hub:
continue
for neighbour in (node.successors if forward else node.predecessors):
if id(neighbour) in seen:
continue
if declines and neighbour.is_resumption_hub:
continue
seen.add(id(neighbour))
stack.append(neighbour)
def reachable_from_any(sources: Iterable[CfgNode]) -> frozenset[int]:
"""
The ids of the nodes forward-reachable from *any* of *sources*, each source included — every
path the graph draws, whatever an edge means.
A flood over a graph that may carry resumption edges wants `reachable_forward_from_any` instead:
this walk follows the resumption hub, which claims a statement is reached from every later one
of its block, so one leak anywhere in a trap-guarded script poisons the whole of it.
"""
return frozenset(flood(sources, forward=True, projection=Projection.MAY))
def normal_reach(
sources: Iterable[CfgNode], *, barrier: Iterable[CfgNode] = (),
) -> frozenset[int]:
"""
The ids of the nodes forward-reachable from *any* of *sources* over `CfgEdge.NORMAL` edges only,
each source included, never expanding *through* a node of *barrier*.
This is the plain-control-flow reach — where a run carries on when nothing throws — as opposed to
`flood`/`Projection`, which read a resuming handler and follow the error edges beside the plain
ones. The two questions the resuming-trap step-over asks both want this reading and neither wants
the error edges: the region a resumed trap *skips* is what runs on the fall-through from the soft
error's local step-over up to the reconvergence point (seed *sources* at the step-over, put the
reconvergence nodes in *barrier*), and the continuation a skipped write is read on is what runs
forward from the reconvergence point (seed *sources* there, no barrier). Following an error edge
would route either walk back through a handler into the very block it stepped out of, reading a
read that the continuation never performs.
*barrier* is seeded into the visited set, so a walk never enters one and the returned set excludes
it: the cut is `reach(sources) \\ through(barrier)`, not the unsound difference
`reach(sources) - reach(barrier)`, which would drop a node the barrier merely *also* reaches by a
back-edge around it.
"""
barrier_ids = {id(node) for node in barrier}
seen: set[int] = set(barrier_ids)
stack: list[CfgNode] = []
for source in sources:
if id(source) not in seen:
seen.add(id(source))
stack.append(source)
while stack:
for target in plain_successors(stack.pop()):
if id(target) in seen:
continue
seen.add(id(target))
stack.append(target)
return frozenset(seen - barrier_ids)
def plain_successors(node: CfgNode) -> list[CfgNode]:
"""
The successors *node* reaches over a `CfgEdge.NORMAL` edge: where control carries on from it
when nothing throws and nothing resumes. Every walk over plain control flow reads its edges
here, so that no two of them can disagree about which edge is plain.
"""
graph = node.graph
return distinct(target for target in node.successors if graph.completes_into(node, target))
def plain_predecessors(node: CfgNode) -> list[CfgNode]:
"""
The predecessors that reach *node* over a `CfgEdge.NORMAL` edge, the reverse of
`plain_successors`.
"""
graph = node.graph
return distinct(source for source in node.predecessors if graph.completes_into(source, node))
class LabelledReach(Generic[_Label]):
"""
Which of several labelled sets of sources reach each node, for every label at once: the answer
`flood_labels` gives where a caller would otherwise walk the same graph once per label.
A label reaches a node exactly when the walk of the same name, seeded with that label's sources
alone, would include the node; each source reaches itself. A label that was never named reaches
nothing. The labels are kept once per strongly connected component of what the sources reach,
so the cost is one bit per label per component, and a caller that labels only what it will ask
about pays for nothing else.
"""
def __init__(self, bits: dict[_Label, int], walks: list[tuple[Condensation[CfgNode], list[int]]]):
self._bits = bits
self._walks = walks
def reaches(self, label: _Label, node: CfgNode) -> bool:
"""
Whether *label* reaches *node*.
"""
bit = self._bits.get(label)
if bit is None:
return False
for condensation, masks in self._walks:
number = condensation.number_of(node)
if number is not None and masks[number] & bit:
return True
return False
def _label_walk(
seeds: dict[int, tuple[CfgNode, int]],
adjacent: Callable[[CfgNode], list[CfgNode]],
) -> tuple[Condensation[CfgNode], list[int]]:
"""
Push the labels *seeds* carry along *adjacent*: the part of the graph the seeds reach is
condensed, and every component's labels are its own seeds' joined with those of every component
that reaches it, in one pass whatever the number of labels.
"""
condensation = Condensation([node for node, _ in seeds.values()], adjacent)
def own(number: int) -> int:
mask = 0
for member in condensation.components[number]:
seeded = seeds.get(id(member))
if seeded is not None:
mask |= seeded[1]
return mask
return condensation, condensation.accumulate(own, int.__or__, downstream=False)
def _seed_labels(
sources: Mapping[_Label, Iterable[CfgNode]],
) -> tuple[dict[_Label, int], dict[int, tuple[CfgNode, int]]]:
bits: dict[_Label, int] = {}
seeds: dict[int, tuple[CfgNode, int]] = {}
for label, nodes in sources.items():
bit = bits.setdefault(label, 1 << len(bits))
for node in nodes:
known = seeds.get(id(node))
seeds[id(node)] = (node, bit if known is None else known[1] | bit)
return bits, seeds
def flood_labels(
sources: Mapping[_Label, Iterable[CfgNode]], *, forward: bool, projection: Projection,
) -> LabelledReach[_Label]:
"""
`flood` for every label of *sources* in one walk: a label reaches a node exactly when `flood`
seeded with that label's sources would return the node's id. Which source takes which walk is
`flood`'s own rule (`_split_sources`), and the hub is screened by `Projection`, so neither is
spelled a second time here.
"""
bits, seeds = _seed_labels(sources)
coarse, precise, declines = _split_sources(
[node for node, _ in seeds.values()], forward=forward, projection=projection)
walks: list[tuple[Condensation[CfgNode], list[int]]] = []
for nodes, reading in ((coarse, Projection.MAY), (precise, projection if declines else Projection.MAY)):
if nodes:
adjacent = reading.successors if forward else reading.predecessors
walks.append(_label_walk({id(node): seeds[id(node)] for node in nodes}, adjacent))
return LabelledReach(bits, walks)
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
def reachable_forward_from_any(
graph: ControlFlowGraph,
sources: Iterable[CfgNode],
) -> frozenset[int]:
"""
`reachable_from_any` restricted to the paths a run takes *going forward* from each source: the
resumption hub is not followed, so a source does not reach the statements standing before it in
its own block merely because some later statement of that block may throw and resume.
The hub claims every resumption point reaches every statement of the guarded block, earlier ones
included. That is the reading a *may* query wants and the ruin of a flood: one leak anywhere in
a trap-guarded script poisons the whole of it. The `CfgEdge.RESUMPTION_FORWARD` edges carry
every run that goes forward — each statement joined to the one control would resume at — so
declining the hub drops no path that goes forward, only the backward reach the hub adds.
A source the graph reports `is_hub_bound` for is flooded with the hub followed instead, because
the forward edges carry nothing for such a source and the precise walk would stop at the
resumption point: an opener written inside a `trap` body would then vouch for the very
statements the handler resumes into. Its flood is the over-approximate one, which is the
fail-closed direction. `flood` performs that fallback, so every door onto the forward reading
inherits it rather than only this one.
Every source must be a node of *graph*, which is what this adds over naming the projection at
`flood` directly. Both facts the walk reads — `is_hub_bound` and `edge_kind` — are keyed by
node identity in that graph alone, so a node lifted from a nested body's graph reads as neither
hub-bound nor kinded and takes the precise walk over what the graph calls plain flow, which is
the fail-open direction.
"""
sources = list(sources)
require_sources_in(graph, sources)
return frozenset(flood(sources, forward=True, projection=Projection.FORWARD))
def require_sources_in(graph: ControlFlowGraph, sources: Iterable[CfgNode]) -> None:
"""
Refuse a source of a forward flood over *graph* that is a node of another body. The two facts
the walk reads, `CfgNode.is_hub_bound` and the edge kinds, are recorded per graph, so such a
source reads as neither a hub nor hub-bound and takes the precise walk over what this graph
calls plain flow, which is the fail-open direction.
"""
for source in sources:
if source.graph is not graph:
raise ValueError(
'a source of a forward flood belongs to another body: the two facts this walk '
'reads are recorded per graph, so such a source reads as neither a hub nor '
'hub-bound and takes the precise walk over what this graph calls plain flow'
)
def reachable_forward_labels(
graph: ControlFlowGraph,
sources: Mapping[_Label, Iterable[CfgNode]],
) -> LabelledReach[_Label]:
"""
`reachable_forward_from_any` for every label of *sources* in one walk, with the same refusal of
a source that belongs to another body.
"""
listed = {label: list(nodes) for label, nodes in sources.items()}
for nodes in listed.values():
require_sources_in(graph, nodes)
return flood_labels(listed, forward=True, projection=Projection.FORWARD)
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
@dataclass
class _Target:
"""
A jump destination active while a breakable or continuable construct is being built. `breaks`
collects the nodes that leave it early, wired to whatever follows once that is known;
`continue_to` is the node a back-jump reaches, or `None` for a construct only a break can leave.
`continues` collects the back-jumps of a loop whose `continue_to` is not yet known — a counted
loop with neither a test nor an update jumps back to its body's own entry, and that entry only
exists once the body has been built. Wiring them to the exit instead would drop the back-edge
the loop is made of.
"""
labels: frozenset[str]
breaks: list[CfgNode]
continue_to: CfgNode | None
is_continuable: bool
is_breakable: bool
continues: list[CfgNode] = field(default_factory=list)
finalizers: int = 0
class _Jump(enum.Enum):
"""
How a jump arrives at the construct it names: past its end, or at its next iteration.
"""
OUT = 'out'
BACK = 'back'
@dataclass(eq=False)
class _Finalizer:
"""
A finalizer open while its guarded block is built, and the jumps that leave the block through
it. Each runs the body from `entry` and goes on from its end: to the next finalizer out where it
leaves that one as well, which `passes_outward` records, and else to the place it jumps to, one
entry of `landings` per place.
"""
entry: CfgNode
passes_outward: bool = False
landings: dict[tuple[_Jump, int], _Target | None] = field(default_factory=dict)
@dataclass(eq=False)
class Resumption:
"""
One statement block being built whose handler set *resumes* it rather than ending it, and the two
synthetic nodes that carry the two projections of `CfgEdge` for it.
`hub` is the over-approximate half: every point the block may resume from joins it, and it joins
every node the block owns. `slot` is the precise half for the statement being built right now —
the point control resumes at when that statement throws, which is the statement after it — and is
created on the first node that needs one, because a statement building no node of its own resumes
at the same place the one before it does.
A frame owns only the nodes built while it is the *innermost* one. A nested resumable block opens
a frame of its own and takes the nodes below it; what joins the two is one edge from the outer hub
to the inner, and the tail of the inner block's forward chain arriving in the frontier the outer
block threads on. Both are edges per nesting *pair*, where naming each node from each enclosing
level is edges per node per level, and a script nested sixty-four deep costs twelve thousand
edges rather than nine hundred.
"""
hub: CfgNode
slot: CfgNode | None = None
def _draw(source: CfgNode, target: CfgNode, kind: CfgEdge) -> None:
"""
An edge carrying *kind*, in the graph *source* belongs to. Every edge is drawn here, so that the
adjacency lists and the kind map are never written apart.
A pair joined by plain edges alone has no entry in the kind map, which is what lets most of a
graph cost nothing there. A pair drawn with another kind first is stored as that kind, beside
`CfgEdge.NORMAL` where a plain edge joined it already, and a pair drawn again with a kind it
already carries is left as it is. `enum.Flag.__or__` builds a member through the class
constructor and costs an order of magnitude more than the dict write around it, and every edge
of a script wrapped in resuming traps passes here.
"""
kinds = source.graph._edge_kinds
if kinds or kind is not CfgEdge.NORMAL:
key = (id(source), id(target))
carried = kinds.get(key)
if carried is None:
if kind is not CfgEdge.NORMAL:
kinds[key] = CfgEdge.NORMAL | kind if _joined(source, target) else kind
elif carried.value & kind.value != kind.value:
kinds[key] = carried | kind
source.successors.append(target)
target.predecessors.append(source)
def _joined(source: CfgNode, target: CfgNode) -> bool:
"""
Whether an edge joins *source* to *target* already, read off the shorter of the two adjacency
lists that would hold it: a resumption hub reaches every node its block owns and the exit is
reached from every node that may leave the body, and neither list is walked for one edge.
"""
if len(source.successors) <= len(target.predecessors):
return any(node is target for node in source.successors)
return any(node is source for node in target.predecessors)
def distinct(nodes: Iterable[CfgNode]) -> list[CfgNode]:
"""
*nodes* with every repetition dropped, compared by identity and in first-seen order.
A frontier carried from one arm of a construct into the next must not accumulate duplicates. An
arm that creates no node of its own hands its incoming frontier straight back, so concatenating
the two doubles the list on every such arm: a `switch` with thirty empty clause bodies would
build 2**30 edges, every one of them a repetition of an edge already there.
"""
seen: set[int] = set()
result: list[CfgNode] = []
for node in nodes:
if id(node) in seen:
continue
seen.add(id(node))
result.append(node)
return result
class CfgBuilder:
"""
Single-pass construction of one `ControlFlowGraph` by structural recursion over a body, threading
a *frontier* — the set of nodes from which normal control currently falls through — into each
statement and out the other side.
A language subclasses this and implements `statement` and `body_blocks`. Everything else is
shared, and the shape methods below take the parts of a construct rather than the construct, so
that no code here has to know what a language calls the pieces of its `for` loop.
"""
def __init__(self, owner: Node):
self.cfg = ControlFlowGraph(owner)
self._handlers: list[CfgNode] = []
self._targets: list[_Target] = []
self._finalizers: list[_Finalizer] = []
self._resumptions: list[Resumption] = []
self._pending_labels: frozenset[str] = frozenset()
def _open_target(self, target: _Target) -> None:
"""
Make *target* the innermost jump destination, remembering how many finalizers stand open
outside it: a jump to it runs every finalizer opened since.
"""
target.finalizers = len(self._finalizers)
self._targets.append(target)
def _leave(self, node: CfgNode, jump: _Jump, target: _Target | None) -> None:
"""
Send the jump *node* to *target* the way *jump* arrives there, or out of the body where
*target* is `None`. A jump that leaves finalizers on its way enters the innermost of them
and goes on from its end, one finalizer after the other outward, and to *target* from the
end of the outermost; `guarded` draws those edges once the finalizer bodies are built. No
edge joins the jump to *target* directly: control arrives there only after every finalizer
it leaves has run to its end.
"""
crossed = self._finalizers if target is None else self._finalizers[target.finalizers:]
if not crossed:
self._arrive(node, jump, target)
return
self.add_edge(node, crossed[-1].entry)
for finalizer in crossed[1:]:
finalizer.passes_outward = True
crossed[0].landings[jump, id(target)] = target
def _arrive(self, node: CfgNode, jump: _Jump, target: _Target | None) -> None:
"""
Join *node* to where a jump arrives at *target* the way *jump* says, or to the body's exit
where *target* is `None`.
"""
if target is None:
self.add_edge(node, self.cfg.exit)
elif jump is _Jump.OUT:
target.breaks.append(node)
elif target.continue_to is not None:
self.add_edge(node, target.continue_to)
else:
target.continues.append(node)
def _leave_finalizer(self, finalizer: _Finalizer, ends: list[CfgNode]) -> None:
"""
Join *ends*, the points from which the body of *finalizer* runs off its end, to where the
runs that entered it go on besides what follows its guarded statement. Each goes on from a
point of its own, built here and so inside the guarded statement: a reader asking where a
run goes once that statement completed meets none of them past it. An error that entered the
finalizer goes on outward. A jump that left the block through it goes on to the next
finalizer out where it leaves that one too, and else to the place it jumps to, one point per
place. A finalizer that jumps or throws itself has no *ends*, and the run it was entered
with goes nowhere from it.
"""
entry = finalizer.entry
if any(self.cfg.is_exceptional(source, entry) for source in entry.predecessors):
self.exceptional_edge(self._point_after(ends), self.escape())
if finalizer.passes_outward:
self.add_edge(self._point_after(ends), self._finalizers[-1].entry)
for (jump, _), target in finalizer.landings.items():
self._arrive(self._point_after(ends), jump, target)
def _point_after(self, ends: list[CfgNode]) -> CfgNode:
node = self.synthetic_node()
self.link(ends, node)
return node
def build(self) -> ControlFlowGraph:
frontier = [self.cfg.entry]
for statements in self.body_blocks(self.cfg.owner):
frontier = self.block(statements, frontier)
self.link(frontier, self.cfg.exit)
return self.cfg
def body_blocks(self, owner: Node) -> Sequence[Sequence[Node]]:
"""
The statement blocks *owner* runs, in the order it runs them. A language whose body may be a
single unbraced statement resolves that here, and one that splits a body across several
named blocks reports them apart rather than joined, because a construct scoped to a block
cannot be modelled from a body that has forgotten where its blocks ended.
"""
raise NotImplementedError
def block(self, statements: Sequence[Node], frontier: list[CfgNode]) -> list[CfgNode]:
"""
One statement block. The statements in order, which is all a block is to a language whose
blocks carry nothing of their own; a language whose blocks scope a handler installs it here
and calls this for every block it builds.
"""
return self.sequence(statements, frontier)
def statement(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]:
"""
Add *statement* to the graph and report the frontier that follows it. A language recognises
its own node types here and calls the shape method that matches; anything it does not
recognise goes to `opaque`, which is the conservative answer.
"""
raise NotImplementedError
def node(self, element: Node) -> CfgNode:
"""
A graph node for *element*, joined to the innermost active handler if one is open.
The handler edge is added here rather than at each site that creates a node, because *any*
statement inside a guarded block may throw and the whole point of the edge is that it does
not depend on which statement it is; `offers` is where a language says otherwise.
"""
node = self.detached_node(element)
if self._handlers:
target = self._handlers[-1]
if self.offers(element, target):
self.exceptional_edge(node, target)
return node
def offers(self, element: Node, target: CfgNode) -> bool:
"""
Whether the node for *element* gets an error edge to *target*, the innermost entry of the
handler stack. An edge into a handler stands for a position, so that a question about where
an error raised there would go is answered whatever stands there, and it is always drawn. A
language whose stack holds something other than a handler may refuse an edge into that.
"""
return True
def synthetic_node(self) -> CfgNode:
"""
A graph node standing for no element: a join or fan-out point the graph needs and the
program does not name. It locates into no query and generates no dataflow fact, which is
what lets one stand in for a set of edges between real nodes without an analysis reading it
as a statement.
"""
node = CfgNode(self.cfg, None)
self.cfg.nodes.append(node)
return node
def detached_node(self, element: Node) -> CfgNode:
"""
A graph node for *element* that the enclosing handler does not guard.
A handler's own entry is one: it stands for the point at which a throw is *offered* to a
clause, which is not a point inside the region that clause guards. The edge outward from
there says the clause did not take the throw, and whether any run says that is a property of
the whole handler set, which `guarded` reads once.
This is the one place an element gets a node, which is why the resumption edges are drawn
here rather than by the caller that knows a block resumes. A block's own statements, the
arms of a construct nested in it, and the entries of a handler set written inside it are
built by three different methods that share nothing but this one, and a level that had to
name its nodes afterwards would have to name the ones its nested levels built as well.
"""
node = CfgNode(self.cfg, element)
self.cfg.nodes.append(node)
self.cfg._node_of[id(element)] = node
if self._resumptions:
self.join_resumption(node)
return node
@staticmethod
def add_edge(source: CfgNode, target: CfgNode) -> None:
"""
A plain edge: `kind_edge` with `CfgEdge.NORMAL`.
"""
_draw(source, target, CfgEdge.NORMAL)
def kind_edge(self, source: CfgNode, target: CfgNode, kind: CfgEdge) -> None:
"""
An edge carrying *kind*.
"""
_draw(source, target, kind)
def exceptional_edge(self, source: CfgNode, target: CfgNode) -> None:
self.kind_edge(source, target, CfgEdge.ERROR_CARRYING)
def resumption_hub(self) -> CfgNode:
"""
A fresh synthetic node standing for the over-approximate half of a resuming handler: every
point the block may resume from joins it, and it joins every point the block may resume at.
Recorded as a hub so that `CfgNode.is_resumption_hub` answers for it, which is what a
forward-only walk declines to enter.
"""
node = self.synthetic_node()
node.is_resumption_hub = True
self.cfg._resuming = True
return node
def resumption_hub_edge(self, source: CfgNode, target: CfgNode) -> None:
"""
An edge *out* of a resumption hub — see `CfgEdge`. The way in is drawn plain, because the
handler body reaching it completed.
"""
self.cfg._resuming = True
self.kind_edge(source, target, CfgEdge.RESUMPTION_HUB)
def resumption_forward_edge(self, source: CfgNode, target: CfgNode) -> None:
"""
An edge of the precise, forward-only half of a resuming handler — see `CfgEdge`.
"""
self.cfg._resuming = True
self.kind_edge(source, target, CfgEdge.RESUMPTION_FORWARD)
def mark_hub_bound(self, nodes: Iterable[CfgNode]) -> None:
"""
Record that the forward-only projection of resumption does not reach *nodes* — see
`CfgNode.is_hub_bound`.
A hub among *nodes* is passed over. A caller naming a handler body names whatever the
level it walked appended, and a resumable block written inside that body contributes its
own hub — a node the forward reading declines to stand at, not one it fails to reach.
Marking it both would leave one node routed to the over-approximate walk as a *source* and
refused as a *neighbour* in the same query, and `ControlFlowGraph.hub_bound` would report a
fan-out as a statement.
"""
self.cfg._resuming = True
for node in nodes:
if node.is_resumption_hub:
continue
node.is_hub_bound = True
self.cfg._hub_bound.add(id(node))
@contextmanager
def resumption(self, resumes: Iterable[CfgNode]) -> Iterator[None]:
"""
Build the enclosed block as one whose handler set resumes it from *resumes*, taking
ownership of every node built inside it — see `Resumption`.
The hub is wired on the way in, before the block is walked, because `detached_node` draws
the edge out of it as each node appears. Where a resumable block is opened inside one, the
outer hub reaches the inner rather than reaching the inner block's nodes one by one: a hub
reaches exactly the nodes its block owns and everything those nodes lead to, and an inner
hub is one of them.
Ownership ends on every way out, which is why this is a block and not a pair of calls. A
build abandoned part-way through a guarded statement would otherwise leave the frame open,
and every node built afterwards would draw both halves of resumption against a handler set
that does not guard it.
"""
hub = self.resumption_hub()
self.resumption_hub_edge(hub, self.cfg.exit)
if self._resumptions:
self.resumption_hub_edge(self._resumptions[-1].hub, hub)
for resume in resumes:
self.add_edge(resume, hub)
self._resumptions.append(Resumption(hub))
try:
yield
finally:
self._resumptions.pop()
def resume_past(self, frontier: list[CfgNode]) -> list[CfgNode]:
"""
*frontier* extended by the point control resumes at when the statement just built throws,
with the innermost open block left ready for the statement after it.
`join_resumption` claims the slot for the first node the statement builds and joins every
later one to it, so there is one here exactly when the statement built a node of its own. A
statement that built none claims nothing — a `trap` declaration is one, and so is an empty
nested block — and the slot of the statement before it stays in the frontier and carries
on to the next, which is where control really resumes.
Claiming and clearing the slot are one step here so that no caller holds the frame or has to
remember to reset it between statements. `distinct` is what keeps a statement that hands its
incoming frontier straight back from drawing the same edge once per statement that passed it
on.
"""
frame = self._resumptions[-1]
slot, frame.slot = frame.slot, None
if slot is None:
return frontier
return distinct([*frontier, slot])
def join_resumption(self, node: CfgNode) -> None:
"""
Draw both projections of resumption for *node* against the innermost open block: the hub
reaches it, and it resumes at the slot standing for the statement being built.
One edge each. A node reached from the hub of the level above it is reached from this one
too, through the edge between the two hubs, and a node resuming at this level's slot carries
on to the level above through the frontier the inner block hands back — so neither half loses
a path by naming only the innermost level, and both stop costing an edge per level.
"""
frame = self._resumptions[-1]
if frame.slot is None:
frame.slot = self.synthetic_node()
self.resumption_forward_edge(node, frame.slot)
self.resumption_hub_edge(frame.hub, node)
def close_handler_set(
self,
entries: Sequence[CfgNode],
*,
escapes: bool,
outward: CfgNode | None = None,
) -> None:
"""
Finish a chain of handler entries: draw the edge from the last of them to what guards the
construct when some run may leave all of them.
Called once the set's bodies are built and the set itself is off the handler stack, so that
`unwinding` names what guards the *construct* rather than the set being closed.
"""
if escapes and entries:
self.exceptional_edge(entries[-1], self.unwinding() if outward is None else outward)
def unwinding(self) -> CfgNode:
"""
Where a throw goes when nothing open here takes it: the innermost active handler, or the
body's exit, which is how the graph says the throw leaves the body.
"""
return self._handlers[-1] if self._handlers else self.cfg.exit
def escape(self) -> CfgNode:
"""
Where an error goes once it gets past every handler of the construct being closed. That is
`unwinding` unless the language hands such an error to something that is not a handler.
"""
return self.unwinding()
def handler_body_guard(self) -> CfgNode | None:
"""
What guards the handler bodies of a guarded block that has no finalizer, or `None` where
nothing beyond what guards the construct does. A language that hands an error leaving a
handler body to something that is not a handler names it here.
"""
return None
def link(self, frontier: Iterable[CfgNode], target: CfgNode) -> None:
for node in frontier:
self.add_edge(node, target)
def sequence(self, statements: Sequence[Node], frontier: list[CfgNode]) -> list[CfgNode]:
for statement in statements:
frontier = self.member(statement, frontier)
self.cfg._completions[id(statement)] = (statement, list(frontier))
return frontier
def member(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]:
"""
Build *statement* as a member of a statement sequence and record the nodes it built, which
`ControlFlowGraph.builder_of` answers from. A statement whose own node already exists was
started by the construct around it — a declaration its block hoists, or a finalizer whose
entry the guarded construct draws first — so the nodes built here are not all of its nodes,
and none are recorded.
"""
if self.cfg.node_of(statement) is not None:
return self.statement(statement, frontier)
start = len(self.cfg.nodes)
frontier = self.statement(statement, frontier)
self.cfg._spans[id(statement)] = (statement, range(start, len(self.cfg.nodes)))
return frontier
def opaque(self, element: Node, frontier: list[CfgNode]) -> list[CfgNode]:
"""
An element whose internal control flow this does not model: one node, entered from the
frontier and falling through. The default for anything a language does not recognise.
"""
node = self.node(element)
self.link(frontier, node)
return [node]
def evaluated(self, element: Node, frontier: list[CfgNode]) -> list[CfgNode]:
"""
The nodes evaluating *element*, a part of a construct evaluated at a point of its own, such
as the initializer of a counted loop: one node, unless a language evaluates the part in
steps it gives nodes of their own.
"""
return self.opaque(element, frontier)
def _body(self, body: Node | None, frontier: list[CfgNode]) -> list[CfgNode]:
return self.statement(body, list(frontier)) if body is not None else list(frontier)
def _branch(self, body: Node | None, head: CfgNode) -> list[CfgNode]:
return self._body(body, [head])
def _capture_body(
self, body: Node | None, frontier: list[CfgNode],
) -> tuple[CfgNode | None, list[CfgNode]]:
"""
Build *body* and return its entry node — the node control reaches first — alongside its exit
frontier. Used where a back-edge must target the body's own entry, which the plain frontier
threading does not expose.
The entry is the first successor the incoming *frontier* gains while *body* is built, not the
first node created. A body that opens with a guarded block builds its handler or finalizer
node before any guarded statement, so creation order would return that handler — a node with
no edge back into the body — and the loop's back-edge would be wired to it, hiding the real
body head from a backward reachability walk. The frontier instead links to the first guarded
statement, which is the node control actually enters.
"""
before = [(node, len(node.successors)) for node in frontier]
exits = self._body(body, frontier)
for node, count in before:
if len(node.successors) > count:
return node.successors[count], exits
return None, exits
def park_label(self, *labels: str | None) -> None:
"""
Park *labels* for the construct about to be built, for `take_labels` to consume.
`labelled` calls this for a language whose label is a statement wrapping the construct. A
language whose label is a field *of* the construct calls it directly, because there is no
wrapping statement to recognise — and a construct built without it carries no label, so
every jump naming it misses and leaves the body instead. A construct may carry several
labels, as `a: b: while (…)` does, and a jump naming any of them targets it.
"""
self._pending_labels = frozenset(label for label in labels if label is not None)
def take_labels(self) -> frozenset[str]:
"""
The labels parked for the construct about to be built, consumed once.
"""
labels = self._pending_labels
self._pending_labels = frozenset()
return labels
def branch_on(
self,
element: Node,
arms: Sequence[Node | None],
frontier: list[CfgNode],
*,
exhaustive: bool = False,
) -> list[CfgNode]:
"""
A conditional: one head node the frontier enters, and one arm per branch. When the arms do
not cover every case — an `if` with no `else` — the head itself is an exit, because control
may pass the whole construct without entering any arm. `exhaustive` says the arms do cover
it, which is what an `if`/`else` pair reports.
"""
head = self.node(element)
self.link(frontier, head)
exits: list[CfgNode] = []
for arm in arms:
exits += self._branch(arm, head)
if not exhaustive:
exits.append(head)
return exits
def loop_head_tested(
self, element: Node, body: Node | None, frontier: list[CfgNode],
) -> list[CfgNode]:
"""
A loop whose condition is evaluated before the body, so the head is both the entry and an
exit: `while`, and every `foreach` whose iteration may run zero times.
"""
head = self.node(element)
self.link(frontier, head)
target = _Target(self.take_labels(), [], head, is_continuable=True, is_breakable=True)
self._open_target(target)
body_exits = self._branch(body, head)
self._targets.pop()
self.link(body_exits, head)
return [head] + target.breaks
def loop_over(
self,
element: Node,
subject: Node | None,
target: Node | None,
body: Node | None,
frontier: list[CfgNode],
) -> list[CfgNode]:
"""
A loop over the values a subject yields. The subject is evaluated once, at a node of its
own before the loop. The head, which asks for the next value, is the entry of every
iteration and an exit. The target receives the value at a node of its own on the way from
the head into the body, so a run leaving the loop has not passed it. A language whose loop
variable and subject are both read at the head builds the loop `loop_head_tested` instead.
"""
if subject is not None:
start = self.node(subject)
self.link(frontier, start)
frontier = [start]
head = self.node(element)
self.link(frontier, head)
loop = _Target(self.take_labels(), [], head, is_continuable=True, is_breakable=True)
self._open_target(loop)
entry = head
if target is not None:
entry = self.node(target)
self.link([head], entry)
body_exits = self._branch(body, entry)
self._targets.pop()
self.link(body_exits, head)
return [head] + loop.breaks
def loop_tail_tested(
self, element: Node, body: Node | None, frontier: list[CfgNode],
) -> list[CfgNode]:
"""
A loop whose condition is evaluated after the body, so the body always runs once and the
back-edge targets the body's own entry rather than the test.
"""
test = self.node(element)
target = _Target(self.take_labels(), [], test, is_continuable=True, is_breakable=True)
self._open_target(target)
entry, body_exits = self._capture_body(body, frontier)
self._targets.pop()
self.link(body_exits, test)
self.add_edge(test, entry if entry is not None else test)
return [test] + target.breaks
def loop_counted(
self,
initializer: Node | None,
test: Node | None,
update: Node | None,
body: Node | None,
frontier: list[CfgNode],
) -> list[CfgNode]:
"""
A loop with a separate initializer, test and update, each evaluated at its own point and so
each given its own node. Any of the three may be absent; a loop with no test has no exit
other than the jumps out of it, and its back-edge targets the body entry.
"""
labels = self.take_labels()
if initializer is not None:
frontier = self.evaluated(initializer, frontier)
head = self.node(test) if test is not None else None
if head is not None:
self.link(frontier, head)
body_frontier: list[CfgNode] = [head]
else:
body_frontier = list(frontier)
step = self.node(update) if update is not None else None
target = _Target(labels, [], step or head, is_continuable=True, is_breakable=True)
self._open_target(target)
entry, body_exits = self._capture_body(body, body_frontier)
self._targets.pop()
latch = body_exits
if step is not None:
self.link(body_exits, step)
latch = [step]
back_to = head if head is not None else entry
if back_to is not None:
self.link(latch, back_to)
self.link(target.continues, back_to if back_to is not None else self.cfg.exit)
exits = list(target.breaks)
if head is not None:
exits.append(head)
return exits
def branch_chain(
self,
clauses: Sequence[tuple[Node, Node | None]],
otherwise: Node | None,
frontier: list[CfgNode],
) -> list[CfgNode]:
"""
A chain of guarded arms, each tested only when every earlier test failed: `if`/`elseif`/`else`
where the whole chain is one node rather than a nest of two-armed conditionals.
Each test gets its own node, because the tests run at different points and an analysis that
collapsed them could not order two of them. A test's node flows into its own arm and on to
the next test; the last test flows into `otherwise` when there is one, and out of the
construct when there is not.
"""
exits: list[CfgNode] = []
current = frontier
for test, body in clauses:
head = self.node(test)
self.link(current, head)
exits += self._branch(body, head)
current = [head]
if otherwise is not None:
return distinct(exits + self._body(otherwise, current))
return distinct(exits + current)
def dispatch(
self,
element: Node,
arms: Sequence[Sequence[Node]],
frontier: list[CfgNode],
*,
arm_flow: ArmFlow,
exhaustive: bool,
iterated: bool = False,
) -> list[CfgNode]:
"""
A multi-way branch: one head the frontier enters and one arm per clause, each arm a statement
sequence the head may jump into. `arm_flow` says what an arm reaches when it runs off its
end; see `ArmFlow`, which is where the languages differ.
`exhaustive` says some arm always runs — a default clause — so the head is not itself an
exit.
`iterated` says the construct runs its arms once per element of an input, so it is a loop:
a back-jump inside an arm re-enters the head rather than resolving to an enclosing loop, and
an arm that simply runs off its end re-enters it too, for the next element. Without that
second edge the arms lie on no cycle and an analysis reads a store inside one as happening
once, which is what lets a self-referential assignment be folded to its first value.
PowerShell's `switch` is this; JavaScript's is not, and reading one as the other both invents
a back-edge to a loop the jump never reaches and drops the one it does.
A jump *out* takes no back-edge: it leaves the construct rather than advancing it, which is
the whole difference between the two spellings.
"""
head = self.node(element)
self.link(frontier, head)
target = _Target(
self.take_labels(),
[],
head if iterated else None,
is_continuable=iterated,
is_breakable=True,
)
self._open_target(target)
carried: list[CfgNode] = []
completed: list[CfgNode] = []
for arm in arms:
reached = self.sequence(list(arm), distinct([head, *carried]))
if arm_flow is ArmFlow.SEQUENTIAL:
carried = reached
elif arm_flow is ArmFlow.CUMULATIVE:
carried = distinct([*carried, *reached])
else:
completed += reached
self._targets.pop()
completed = distinct(completed + carried)
if iterated:
self.link(completed, head)
exits = completed + target.breaks
if not exhaustive:
exits.append(head)
return distinct(exits)
def guarded(
self,
block: Node | None,
handlers: Sequence[tuple[Node, Node | None]],
finalizer: Node | None,
finalizer_body: Sequence[Node],
frontier: list[CfgNode],
*,
escapes: bool,
) -> list[CfgNode]:
"""
A guarded block with any number of handlers and an optional finalizer.
The handler nodes are created *before* the guarded block is built and the first is pushed on
the handler stack, so that `node` joins every statement created inside the block to it. That
ordering is the whole mechanism: it is why no statement inside the block has to know it is
guarded.
Several handlers are chained from the first, because which one runs depends on the type of
the exception and none of them is guaranteed — a language with one handler passes a
one-element sequence and the chain degenerates. The chain edge is *exceptional*: control
takes it exactly when the earlier handler did not match, so nothing that handler's node
stands for has run and its kill must not apply along it.
`escapes` says the set of handlers may fail to take a throw the block makes, which is what a
language whose clauses carry a type filter reports and what one whose single clause takes
everything denies. It decides one edge — from the last handler to whatever guards this
construct, or to the body exit when nothing does, which is how the graph says a throw leaves
the body with no handler having taken it. Denying the edge is the difference between a
clause that swallows and one that passes the throw on, so it is asked of every language
rather than defaulted: the safe answer invents a path the program cannot take, and the
other drops one it can.
The finalizer is entered from the block's normal exits and from every handler's, and runs on
every other way out of the construct as well: an error no handler takes enters it, and so
does a jump leaving the block or a handler body, and each goes on from its end, which
`_leave_finalizer` draws. It also guards every handler body: a handler that throws, returns
or jumps out of the construct runs the finalizer on its way, so where the block and every
handler end that way, the finalizer is still reached.
"""
handlers = list(handlers)
entries = [self.detached_node(handler) for handler, _ in handlers]
finalizer_entry = self.detached_node(finalizer) if finalizer is not None else None
record = _Finalizer(finalizer_entry) if finalizer_entry is not None else None
guard = entries[0] if entries else finalizer_entry
if record is not None:
self._finalizers.append(record)
if guard is not None:
self._handlers.append(guard)
block_exits = self.statement(block, frontier) if block is not None else list(frontier)
if guard is not None:
self._handlers.pop()
normal_exits = list(block_exits)
clause_guard = finalizer_entry
if clause_guard is None and entries:
clause_guard = self.handler_body_guard()
if clause_guard is not None:
self._handlers.append(clause_guard)
for index, ((_, body), entry) in enumerate(zip(handlers, entries)):
if index:
self.exceptional_edge(entries[index - 1], entry)
normal_exits += (
self.statement(body, [entry]) if body is not None else [entry])
if clause_guard is not None:
self._handlers.pop()
if record is not None:
self._finalizers.pop()
outward = self.escape() if finalizer_entry is None else finalizer_entry
self.close_handler_set(entries, escapes=escapes, outward=outward)
if record is not None and finalizer_entry is not None:
self.link(normal_exits, finalizer_entry)
final_exits = self.sequence(list(finalizer_body), [finalizer_entry])
self._leave_finalizer(record, final_exits)
return final_exits
return normal_exits
def labelled(
self,
labels: Sequence[str],
body: Node | None,
frontier: list[CfgNode],
*,
binds_to_body: bool,
) -> list[CfgNode]:
"""
A statement *body* with every label written in front of it, *labels*. When they name a
construct a jump can target directly — `binds_to_body` — they are parked for that construct
to consume through `take_labels`; otherwise they name this statement itself and only a break
can leave it.
"""
if binds_to_body:
self.park_label(*labels)
return self.statement(body, frontier) if body is not None else list(frontier)
target = _Target(frozenset(labels), [], None, is_continuable=False, is_breakable=False)
self._open_target(target)
exits = self.statement(body, frontier) if body is not None else list(frontier)
self._targets.pop()
return exits + target.breaks
def jump_out(
self, element: Node, label: str | None, frontier: list[CfgNode],
) -> list[CfgNode]:
"""
A jump that leaves the construct it names, or the innermost breakable one when unlabelled. A
jump naming nothing this body holds leaves the body, which is the conservative reading.
"""
node = self.node(element)
self.link(frontier, node)
self._leave(node, _Jump.OUT, self._break_target(label))
return []
def jump_back(
self, element: Node, label: str | None, frontier: list[CfgNode],
) -> list[CfgNode]:
"""
A jump to the next iteration of the loop it names, or of the innermost loop when unlabelled.
"""
node = self.node(element)
self.link(frontier, node)
self._leave(node, _Jump.BACK, self._continue_target(label))
return []
def terminate(
self, element: Node, frontier: list[CfgNode], *, exceptional: bool,
) -> list[CfgNode]:
"""
A statement after which control does not continue in this body: a return, or a throw.
A throw is `exceptional`, so it reaches the innermost open handler rather than the exit, and
only reaches the exit when no handler is open. A return leaves the body through every
finalizer open around it, as a jump out of the body does.
"""
node = self.node(element)
self.link(frontier, node)
if exceptional:
self.exceptional_edge(node, self.unwinding())
else:
self._leave(node, _Jump.OUT, None)
return []
def has_continue_target(self, label: str | None) -> bool:
"""
Whether a back-jump naming *label* — or an unlabelled one — resolves to a construct
currently being built. A language in which `continue` means something other than a back-jump
when no such construct is open asks this to tell the two spellings apart.
"""
return self._continue_target(label) is not None
def has_break_target(self, label: str | None) -> bool:
"""
Whether a jump out naming *label* — or an unlabelled one — resolves to a construct currently
being built, the counterpart of `has_continue_target` for the language in which `break`
means something other than leaving a construct when no construct is open to leave.
"""
return self._break_target(label) is not None
def _break_target(self, label: str | None) -> _Target | None:
for target in reversed(self._targets):
if label is None:
if target.is_breakable:
return target
elif label in target.labels:
return target
return None
def _continue_target(self, label: str | None) -> _Target | None:
for target in reversed(self._targets):
if not target.is_continuable:
continue
if label is None or label in target.labels:
return target
return None
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)
def build_control_flow(
root: Node,
builder: Callable[[Node], CfgBuilder],
owns_a_graph: Callable[[Node], bool],
) -> dict[int, ControlFlowGraph]:
"""
Build one control-flow graph for the script and one for every node *owns_a_graph* accepts,
keyed by the owner node's identity: a function, and in a language that draws one for it, the
initializer of a class field. The graphs are independent: a nested function appears in its
parent's graph only as the statement that defines it, never as descended-into control flow.
"""
graphs: dict[int, ControlFlowGraph] = {id(root): builder(root).build()}
for node in root.walk():
if owns_a_graph(node):
graphs[id(node)] = builder(node).build()
return graphsGlobal variables
var RAISE_TAKEN-
The kinds an edge is taken along only on runs where its source threw rather than completing. A store the source makes may not have happened at the other end of one, whichever of the three it is, which is the one question every flow-sensitive consumer asks of the kind.
Functions
def flood(sources, *, forward, projection)-
The ids of the nodes reachable from any of sources, each source included: over successor edges when forward and over predecessor edges otherwise, and over the projection that declines the resumption hub when projected.
The one walk every reachability question over these graphs is asked through, so that a caller choosing a direction or a projection chooses it by naming an argument rather than by writing a fourth copy of the sweep that quietly disagrees with the other three about what an edge means. Neither dimension has a default:
forwarddecides whether the answer is what this reaches or what reaches this, andprojectiondecides how a resumption is read. A caller that did not think about either asked the wrong question.The hub-bound fallback is a property of a source, so it is applied here. A forward walk under
Projection.FORWARDseeded at a node the graph callsCfgNode.is_hub_boundis walked with the hub followed instead, because the forward edges carry nothing for such a source and the precise walk would stop dead at the resumption point — the fail-open direction, and the one a flood cannot afford. It is applied to the sources and never to a node the walk merely met, which would follow the hub out of any handler body the sweep happened to pass through. The two sweeps share oneseen, so the second never re-walks what the first already reached; the first walks on through every other source it meets, and the second starts only from the sources the first did not reach, because what the precise walk reaches from a node is a part of what the hub-followed walk reaches from it.Walked as one depth-first sweep seeded with every source rather than a union of per-source walks: the union grows the same answer at the cost of one full walk per source, where a single sweep visits each node once however many sources reach it. The result is a fresh set the caller owns, so it may intersect or discard it in place.
A graph none of whose handlers resume draws no hub, so the two readings walk it identically and the screen is dropped before the sweep begins. That is nearly every script, and this is the one walk every layer above reaches the graphs through — asking a projection per node instead costs the whole run more than the resumption edges do.
Expand source code Browse git
def flood( sources: Iterable[CfgNode], *, forward: bool, projection: Projection, ) -> set[int]: """ The ids of the nodes reachable from *any* of *sources*, each source included: over successor edges when *forward* and over predecessor edges otherwise, and over the projection that declines the resumption hub when *projected*. The one walk every reachability question over these graphs is asked through, so that a caller choosing a direction or a projection chooses it by naming an argument rather than by writing a fourth copy of the sweep that quietly disagrees with the other three about what an edge means. Neither dimension has a default: `forward` decides whether the answer is *what this reaches* or *what reaches this*, and `projection` decides how a resumption is read. A caller that did not think about either asked the wrong question. **The hub-bound fallback is a property of a source, so it is applied here.** A forward walk under `Projection.FORWARD` seeded at a node the graph calls `CfgNode.is_hub_bound` is walked with the hub followed instead, because the forward edges carry nothing for such a source and the precise walk would stop dead at the resumption point — the fail-open direction, and the one a flood cannot afford. It is applied to the *sources* and never to a node the walk merely met, which would follow the hub out of any handler body the sweep happened to pass through. The two sweeps share one `seen`, so the second never re-walks what the first already reached; the first walks on through every other source it meets, and the second starts only from the sources the first did not reach, because what the precise walk reaches from a node is a part of what the hub-followed walk reaches from it. Walked as one depth-first sweep seeded with every source rather than a union of per-source walks: the union grows the same answer at the cost of one full walk per source, where a single sweep visits each node once however many sources reach it. The result is a fresh set the caller owns, so it may intersect or discard it in place. A graph none of whose handlers resume draws no hub, so the two readings walk it identically and the screen is dropped before the sweep begins. That is nearly every script, and this is the one walk every layer above reaches the graphs through — asking a projection per node instead costs the whole run more than the resumption edges do. """ coarse, precise, declines = _split_sources( distinct(sources), forward=forward, projection=projection) seen: set[int] = {id(node) for node in coarse} if coarse: _sweep(coarse, seen, forward=True, declines=False) pending = [node for node in precise if id(node) not in seen] seen.update(id(node) for node in pending) _sweep(pending, seen, forward=forward, declines=declines) return seen def walk_reading(source, *, forward, projection)-
The reading
flood()walks source alone under, which is_split_sources' rule for one source, named asControlFlowGraph.reading()names it. Two walks of one graph under one reading walk one relation, which is what a caller combining their answers over the condensation of that reading (ControlFlowGraph.condensation()) must know: the components reached from that of source hold exactly the nodesflood()returns from it. That holds for source and not for every node sharing its component, since another node of it may be walked under the other reading.Expand source code Browse git
def walk_reading(source: CfgNode, *, forward: bool, projection: Projection) -> Projection: """ The reading `flood` walks *source* alone under, which is `_split_sources`' rule for one source, named as `ControlFlowGraph.reading` names it. Two walks of one graph under one reading walk one relation, which is what a caller combining their answers over the condensation of that reading (`ControlFlowGraph.condensation`) must know: the components reached from that of *source* hold exactly the nodes `flood` returns from it. That holds for *source* and not for every node sharing its component, since another node of it may be walked under the other reading. """ coarse, _, _ = _split_sources([source], forward=forward, projection=projection) return Projection.MAY if coarse else source.graph.reading(projection) def reachable_from_any(sources)-
The ids of the nodes forward-reachable from any of sources, each source included — every path the graph draws, whatever an edge means.
A flood over a graph that may carry resumption edges wants
reachable_forward_from_any()instead: this walk follows the resumption hub, which claims a statement is reached from every later one of its block, so one leak anywhere in a trap-guarded script poisons the whole of it.Expand source code Browse git
def reachable_from_any(sources: Iterable[CfgNode]) -> frozenset[int]: """ The ids of the nodes forward-reachable from *any* of *sources*, each source included — every path the graph draws, whatever an edge means. A flood over a graph that may carry resumption edges wants `reachable_forward_from_any` instead: this walk follows the resumption hub, which claims a statement is reached from every later one of its block, so one leak anywhere in a trap-guarded script poisons the whole of it. """ return frozenset(flood(sources, forward=True, projection=Projection.MAY)) def normal_reach(sources, *, barrier=())-
The ids of the nodes forward-reachable from any of sources over
CfgEdge.NORMALedges only, each source included, never expanding through a node of barrier.This is the plain-control-flow reach — where a run carries on when nothing throws — as opposed to
flood()/Projection, which read a resuming handler and follow the error edges beside the plain ones. The two questions the resuming-trap step-over asks both want this reading and neither wants the error edges: the region a resumed trap skips is what runs on the fall-through from the soft error's local step-over up to the reconvergence point (seed sources at the step-over, put the reconvergence nodes in barrier), and the continuation a skipped write is read on is what runs forward from the reconvergence point (seed sources there, no barrier). Following an error edge would route either walk back through a handler into the very block it stepped out of, reading a read that the continuation never performs.barrier is seeded into the visited set, so a walk never enters one and the returned set excludes it: the cut is
reach(sources) \ through(barrier), not the unsound differencereach(sources) - reach(barrier), which would drop a node the barrier merely also reaches by a back-edge around it.Expand source code Browse git
def normal_reach( sources: Iterable[CfgNode], *, barrier: Iterable[CfgNode] = (), ) -> frozenset[int]: """ The ids of the nodes forward-reachable from *any* of *sources* over `CfgEdge.NORMAL` edges only, each source included, never expanding *through* a node of *barrier*. This is the plain-control-flow reach — where a run carries on when nothing throws — as opposed to `flood`/`Projection`, which read a resuming handler and follow the error edges beside the plain ones. The two questions the resuming-trap step-over asks both want this reading and neither wants the error edges: the region a resumed trap *skips* is what runs on the fall-through from the soft error's local step-over up to the reconvergence point (seed *sources* at the step-over, put the reconvergence nodes in *barrier*), and the continuation a skipped write is read on is what runs forward from the reconvergence point (seed *sources* there, no barrier). Following an error edge would route either walk back through a handler into the very block it stepped out of, reading a read that the continuation never performs. *barrier* is seeded into the visited set, so a walk never enters one and the returned set excludes it: the cut is `reach(sources) \\ through(barrier)`, not the unsound difference `reach(sources) - reach(barrier)`, which would drop a node the barrier merely *also* reaches by a back-edge around it. """ barrier_ids = {id(node) for node in barrier} seen: set[int] = set(barrier_ids) stack: list[CfgNode] = [] for source in sources: if id(source) not in seen: seen.add(id(source)) stack.append(source) while stack: for target in plain_successors(stack.pop()): if id(target) in seen: continue seen.add(id(target)) stack.append(target) return frozenset(seen - barrier_ids) def plain_successors(node)-
The successors node reaches over a
CfgEdge.NORMALedge: where control carries on from it when nothing throws and nothing resumes. Every walk over plain control flow reads its edges here, so that no two of them can disagree about which edge is plain.Expand source code Browse git
def plain_successors(node: CfgNode) -> list[CfgNode]: """ The successors *node* reaches over a `CfgEdge.NORMAL` edge: where control carries on from it when nothing throws and nothing resumes. Every walk over plain control flow reads its edges here, so that no two of them can disagree about which edge is plain. """ graph = node.graph return distinct(target for target in node.successors if graph.completes_into(node, target)) def plain_predecessors(node)-
The predecessors that reach node over a
CfgEdge.NORMALedge, the reverse ofplain_successors().Expand source code Browse git
def plain_predecessors(node: CfgNode) -> list[CfgNode]: """ The predecessors that reach *node* over a `CfgEdge.NORMAL` edge, the reverse of `plain_successors`. """ graph = node.graph return distinct(source for source in node.predecessors if graph.completes_into(source, node)) def flood_labels(sources, *, forward, projection)-
flood()for every label of sources in one walk: a label reaches a node exactly whenflood()seeded with that label's sources would return the node's id. Which source takes which walk isflood()'s own rule (_split_sources), and the hub is screened byProjection, so neither is spelled a second time here.Expand source code Browse git
def flood_labels( sources: Mapping[_Label, Iterable[CfgNode]], *, forward: bool, projection: Projection, ) -> LabelledReach[_Label]: """ `flood` for every label of *sources* in one walk: a label reaches a node exactly when `flood` seeded with that label's sources would return the node's id. Which source takes which walk is `flood`'s own rule (`_split_sources`), and the hub is screened by `Projection`, so neither is spelled a second time here. """ bits, seeds = _seed_labels(sources) coarse, precise, declines = _split_sources( [node for node, _ in seeds.values()], forward=forward, projection=projection) walks: list[tuple[Condensation[CfgNode], list[int]]] = [] for nodes, reading in ((coarse, Projection.MAY), (precise, projection if declines else Projection.MAY)): if nodes: adjacent = reading.successors if forward else reading.predecessors walks.append(_label_walk({id(node): seeds[id(node)] for node in nodes}, adjacent)) return LabelledReach(bits, walks) def reachable_forward_from_any(graph, sources)-
reachable_from_any()restricted to the paths a run takes going forward from each source: the resumption hub is not followed, so a source does not reach the statements standing before it in its own block merely because some later statement of that block may throw and resume.The hub claims every resumption point reaches every statement of the guarded block, earlier ones included. That is the reading a may query wants and the ruin of a flood: one leak anywhere in a trap-guarded script poisons the whole of it. The
CfgEdge.RESUMPTION_FORWARDedges carry every run that goes forward — each statement joined to the one control would resume at — so declining the hub drops no path that goes forward, only the backward reach the hub adds.A source the graph reports
is_hub_boundfor is flooded with the hub followed instead, because the forward edges carry nothing for such a source and the precise walk would stop at the resumption point: an opener written inside atrapbody would then vouch for the very statements the handler resumes into. Its flood is the over-approximate one, which is the fail-closed direction.flood()performs that fallback, so every door onto the forward reading inherits it rather than only this one.Every source must be a node of graph, which is what this adds over naming the projection at
flood()directly. Both facts the walk reads —is_hub_boundandedge_kind— are keyed by node identity in that graph alone, so a node lifted from a nested body's graph reads as neither hub-bound nor kinded and takes the precise walk over what the graph calls plain flow, which is the fail-open direction.Expand source code Browse git
def reachable_forward_from_any( graph: ControlFlowGraph, sources: Iterable[CfgNode], ) -> frozenset[int]: """ `reachable_from_any` restricted to the paths a run takes *going forward* from each source: the resumption hub is not followed, so a source does not reach the statements standing before it in its own block merely because some later statement of that block may throw and resume. The hub claims every resumption point reaches every statement of the guarded block, earlier ones included. That is the reading a *may* query wants and the ruin of a flood: one leak anywhere in a trap-guarded script poisons the whole of it. The `CfgEdge.RESUMPTION_FORWARD` edges carry every run that goes forward — each statement joined to the one control would resume at — so declining the hub drops no path that goes forward, only the backward reach the hub adds. A source the graph reports `is_hub_bound` for is flooded with the hub followed instead, because the forward edges carry nothing for such a source and the precise walk would stop at the resumption point: an opener written inside a `trap` body would then vouch for the very statements the handler resumes into. Its flood is the over-approximate one, which is the fail-closed direction. `flood` performs that fallback, so every door onto the forward reading inherits it rather than only this one. Every source must be a node of *graph*, which is what this adds over naming the projection at `flood` directly. Both facts the walk reads — `is_hub_bound` and `edge_kind` — are keyed by node identity in that graph alone, so a node lifted from a nested body's graph reads as neither hub-bound nor kinded and takes the precise walk over what the graph calls plain flow, which is the fail-open direction. """ sources = list(sources) require_sources_in(graph, sources) return frozenset(flood(sources, forward=True, projection=Projection.FORWARD)) def require_sources_in(graph, sources)-
Refuse a source of a forward flood over graph that is a node of another body. The two facts the walk reads,
CfgNode.is_hub_boundand the edge kinds, are recorded per graph, so such a source reads as neither a hub nor hub-bound and takes the precise walk over what this graph calls plain flow, which is the fail-open direction.Expand source code Browse git
def require_sources_in(graph: ControlFlowGraph, sources: Iterable[CfgNode]) -> None: """ Refuse a source of a forward flood over *graph* that is a node of another body. The two facts the walk reads, `CfgNode.is_hub_bound` and the edge kinds, are recorded per graph, so such a source reads as neither a hub nor hub-bound and takes the precise walk over what this graph calls plain flow, which is the fail-open direction. """ for source in sources: if source.graph is not graph: raise ValueError( 'a source of a forward flood belongs to another body: the two facts this walk ' 'reads are recorded per graph, so such a source reads as neither a hub nor ' 'hub-bound and takes the precise walk over what this graph calls plain flow' ) def reachable_forward_labels(graph, sources)-
reachable_forward_from_any()for every label of sources in one walk, with the same refusal of a source that belongs to another body.Expand source code Browse git
def reachable_forward_labels( graph: ControlFlowGraph, sources: Mapping[_Label, Iterable[CfgNode]], ) -> LabelledReach[_Label]: """ `reachable_forward_from_any` for every label of *sources* in one walk, with the same refusal of a source that belongs to another body. """ listed = {label: list(nodes) for label, nodes in sources.items()} for nodes in listed.values(): require_sources_in(graph, nodes) return flood_labels(listed, forward=True, projection=Projection.FORWARD) def distinct(nodes)-
nodes with every repetition dropped, compared by identity and in first-seen order.
A frontier carried from one arm of a construct into the next must not accumulate duplicates. An arm that creates no node of its own hands its incoming frontier straight back, so concatenating the two doubles the list on every such arm: a
switchwith thirty empty clause bodies would build 2**30 edges, every one of them a repetition of an edge already there.Expand source code Browse git
def distinct(nodes: Iterable[CfgNode]) -> list[CfgNode]: """ *nodes* with every repetition dropped, compared by identity and in first-seen order. A frontier carried from one arm of a construct into the next must not accumulate duplicates. An arm that creates no node of its own hands its incoming frontier straight back, so concatenating the two doubles the list on every such arm: a `switch` with thirty empty clause bodies would build 2**30 edges, every one of them a repetition of an edge already there. """ seen: set[int] = set() result: list[CfgNode] = [] for node in nodes: if id(node) in seen: continue seen.add(id(node)) result.append(node) return result def build_control_flow(root, builder, owns_a_graph)-
Build one control-flow graph for the script and one for every node owns_a_graph accepts, keyed by the owner node's identity: a function, and in a language that draws one for it, the initializer of a class field. The graphs are independent: a nested function appears in its parent's graph only as the statement that defines it, never as descended-into control flow.
Expand source code Browse git
def build_control_flow( root: Node, builder: Callable[[Node], CfgBuilder], owns_a_graph: Callable[[Node], bool], ) -> dict[int, ControlFlowGraph]: """ Build one control-flow graph for the script and one for every node *owns_a_graph* accepts, keyed by the owner node's identity: a function, and in a language that draws one for it, the initializer of a class field. The graphs are independent: a nested function appears in its parent's graph only as the statement that defines it, never as descended-into control flow. """ graphs: dict[int, ControlFlowGraph] = {id(root): builder(root).build()} for node in root.walk(): if owns_a_graph(node): graphs[id(node)] = builder(node).build() return graphs
Classes
class CfgEdge (*args, **kwds)-
What an edge says about the run that takes it, beyond the two nodes it joins.
NORMAL— control simply passed on, the source having done whatever it does.ERROR_CARRYING— the source threw and the error travels here: the edge into the handler that is offered it, and onward from a handler set that may decline. A definition the source makes is not guaranteed to have happened along one of these.RESUMPTION_HUB,RESUMPTION_FORWARD— the two projections of a handler that resumes the block it guards rather than ending it. The engine carries on at the statement after the one that threw, and no graph knows which statement that was, so the shape is carried twice: the hub is the over-approximation, joining every resumption point to every statement of the block, and the forward edges are the precise half, joining each statement to the one it would resume at.The two are not interchangeable, and reading one where the other is meant is how this goes wrong. Every run the forward edges draw the hub draws too; the hub draws more, and what it adds is the claim that a resumption point reaches the statements written above it — which is the claim the forward half exists to drop. An analysis asking whether any resumption path exists reads the hub, since dropping a path there would let it call a live statement unreachable; one asking what a point reaches going forward reads the forward edges and skips the hub, since keeping the backward reach there would poison a whole trap-guarded script from one leak — see
reachable_forward_from_any().The distinction the resumption kinds force is that
ERROR_CARRYINGconflates two bits which coincide nowhere else: taken only when the source threw, and the error object travels here. A resumption edge is the first that is the former and not the latter — the handler swallowed the error — so a consumer reasoning about whether the source completed must readRAISE_TAKEN, and only one reasoning about where an error went may readERROR_CARRYINGalone.Which edges of the hub carry the kind follows from that and is not symmetric. The edges out of a hub do: their target is reached only because something threw. The edge into one does not, and is plain — its source is the handler body running to its end, which is a statement that completed, and a store it makes did happen. Reading the way in as raise-taken says no read below a
trapcan ever observe a store the handler made.A pair of nodes carries the kinds of every edge drawn between it,
NORMALamong them, so a pair joined both ways answers for both. The last statement of atrybody falls through into thefinallythat guards it and carries its error there too; with no bit of its own, the plain edge would read as error-carrying, and every walk over plain control flow would lose the path. A pair joined by plain edges alone is stored as nothing at all, which reads asNORMAL.Expand source code Browse git
class CfgEdge(enum.Flag): """ What an edge says about the run that takes it, beyond the two nodes it joins. `NORMAL` — control simply passed on, the source having done whatever it does. `ERROR_CARRYING` — the source threw and the error travels here: the edge into the handler that is offered it, and onward from a handler set that may decline. A definition the source makes is not guaranteed to have happened along one of these. `RESUMPTION_HUB`, `RESUMPTION_FORWARD` — the two projections of a handler that *resumes* the block it guards rather than ending it. The engine carries on at the statement after the one that threw, and no graph knows which statement that was, so the shape is carried twice: the hub is the over-approximation, joining every resumption point to every statement of the block, and the forward edges are the precise half, joining each statement to the one it would resume at. The two are *not* interchangeable, and reading one where the other is meant is how this goes wrong. Every run the forward edges draw the hub draws too; the hub draws more, and what it adds is the claim that a resumption point reaches the statements written *above* it — which is the claim the forward half exists to drop. An analysis asking whether *any* resumption path exists reads the hub, since dropping a path there would let it call a live statement unreachable; one asking what a point reaches *going forward* reads the forward edges and skips the hub, since keeping the backward reach there would poison a whole trap-guarded script from one leak — see `reachable_forward_from_any`. The distinction the resumption kinds force is that `ERROR_CARRYING` conflates two bits which coincide nowhere else: *taken only when the source threw*, and *the error object travels here*. A resumption edge is the first that is the former and not the latter — the handler swallowed the error — so a consumer reasoning about whether the source completed must read `RAISE_TAKEN`, and only one reasoning about where an error went may read `ERROR_CARRYING` alone. Which edges of the hub carry the kind follows from that and is not symmetric. The edges *out* of a hub do: their target is reached only because something threw. The edge *into* one does not, and is plain — its source is the handler body running to its end, which is a statement that completed, and a store it makes did happen. Reading the way in as raise-taken says no read below a `trap` can ever observe a store the handler made. A pair of nodes carries the kinds of every edge drawn between it, `NORMAL` among them, so a pair joined both ways answers for both. The last statement of a `try` body falls through into the `finally` that guards it and carries its error there too; with no bit of its own, the plain edge would read as error-carrying, and every walk over plain control flow would lose the path. A pair joined by plain edges alone is stored as nothing at all, which reads as `NORMAL`. """ NORMAL = enum.auto() ERROR_CARRYING = enum.auto() RESUMPTION_HUB = enum.auto() RESUMPTION_FORWARD = enum.auto()Ancestors
- enum.Flag
- enum.Enum
Class variables
var NORMAL-
The type of the None singleton.
var ERROR_CARRYING-
The type of the None singleton.
var RESUMPTION_HUB-
The type of the None singleton.
var RESUMPTION_FORWARD-
The type of the None singleton.
class Projection (*args, **kwds)-
Which reading of a resuming handler a walk over a control-flow graph is asked for — the two halves
CfgEdgedescribes, named so that a walk states which one it means.MAY— every path the graph draws, the resumption hub among them. A resumption point is taken to reach every statement of its block, the ones written above it included. This is the reading a caller asking whether something can happen has to have: dropping a path there would let it call a live statement unreachable, or a store that is read dead.FORWARD— the paths a run takes going forward. The hub is declined, so a resumption reaches only the statement control would resume at and what follows it. This is the reading a flood needs: underMAYone leak anywhere in a trap-guarded block poisons the whole of it, the statements above the leak included.Neither is the safe one.
MAYclaims runs that cannot happen andFORWARDdrops the fact that a handler body carries on somewhere the forward edges do not name, so which of them fails closed depends on what is asked. That is why every walk here names one rather than taking a default.The hub is declined as a node, and in both directions. Filtering the edges instead would let a walk cross a hub the moment some other edge joined the same pair, and would strand a backward walk inside a hub it can never leave.
The forward reading is sound by excision. Take any run: the segment where it is inside a resuming handler body is exactly the part the forward edges do not draw, and cutting that segment out leaves a path over the guarded statements that the forward edges do draw — the statement that threw, then the one after it. So every ordering the forward projection reports holds of the real run, over a subset of the nodes the run visited. What it does not report is where a statement of the handler body itself leads, which is why a source inside one is walked under
MAYinstead (CfgNode.is_hub_bound).Expand source code Browse git
class Projection(enum.Enum): """ Which reading of a resuming handler a walk over a control-flow graph is asked for — the two halves `CfgEdge` describes, named so that a walk states which one it means. `MAY` — every path the graph draws, the resumption hub among them. A resumption point is taken to reach every statement of its block, the ones written above it included. This is the reading a caller asking whether something *can* happen has to have: dropping a path there would let it call a live statement unreachable, or a store that is read dead. `FORWARD` — the paths a run takes going forward. The hub is declined, so a resumption reaches only the statement control would resume at and what follows it. This is the reading a flood needs: under `MAY` one leak anywhere in a trap-guarded block poisons the whole of it, the statements above the leak included. Neither is the safe one. `MAY` claims runs that cannot happen and `FORWARD` drops the fact that a handler body carries on somewhere the forward edges do not name, so which of them fails closed depends on what is asked. That is why every walk here names one rather than taking a default. The hub is declined as a *node*, and in both directions. Filtering the edges instead would let a walk cross a hub the moment some other edge joined the same pair, and would strand a backward walk inside a hub it can never leave. **The forward reading is sound by excision.** Take any run: the segment where it is inside a resuming handler body is exactly the part the forward edges do not draw, and cutting that segment out leaves a path over the guarded statements that the forward edges do draw — the statement that threw, then the one after it. So every ordering the forward projection reports holds of the real run, over a subset of the nodes the run visited. What it does *not* report is where a statement of the handler body itself leads, which is why a source inside one is walked under `MAY` instead (`CfgNode.is_hub_bound`). """ MAY = enum.auto() FORWARD = enum.auto() def successors(self, node: CfgNode) -> list[CfgNode]: """ The nodes control may pass to from *node* under this reading. """ if self is Projection.MAY: return node.successors if node.is_resumption_hub: return [] return [target for target in node.successors if not target.is_resumption_hub] def predecessors(self, node: CfgNode) -> list[CfgNode]: """ The nodes control may arrive at *node* from under this reading. """ if self is Projection.MAY: return node.predecessors if node.is_resumption_hub: return [] return [source for source in node.predecessors if not source.is_resumption_hub] @property def declines_the_hub(self) -> bool: """ Whether a walk under this reading refuses to stand at a resumption hub, which is the whole of what the two readings disagree about and the form a walk over millions of edges can afford to ask it in. `successors` and `predecessors` say the same thing by building a list, which is what a caller wanting an adjacency it can iterate twice wants and what a single depth-first sweep must not pay per node. """ return self is Projection.FORWARDAncestors
- enum.Enum
Class variables
var MAY-
The type of the None singleton.
var FORWARD-
The type of the None singleton.
Instance variables
var declines_the_hub-
Whether a walk under this reading refuses to stand at a resumption hub, which is the whole of what the two readings disagree about and the form a walk over millions of edges can afford to ask it in.
successorsandpredecessorssay the same thing by building a list, which is what a caller wanting an adjacency it can iterate twice wants and what a single depth-first sweep must not pay per node.Expand source code Browse git
@property def declines_the_hub(self) -> bool: """ Whether a walk under this reading refuses to stand at a resumption hub, which is the whole of what the two readings disagree about and the form a walk over millions of edges can afford to ask it in. `successors` and `predecessors` say the same thing by building a list, which is what a caller wanting an adjacency it can iterate twice wants and what a single depth-first sweep must not pay per node. """ return self is Projection.FORWARD
Methods
def successors(self, node)-
The nodes control may pass to from node under this reading.
Expand source code Browse git
def successors(self, node: CfgNode) -> list[CfgNode]: """ The nodes control may pass to from *node* under this reading. """ if self is Projection.MAY: return node.successors if node.is_resumption_hub: return [] return [target for target in node.successors if not target.is_resumption_hub] def predecessors(self, node)-
The nodes control may arrive at node from under this reading.
Expand source code Browse git
def predecessors(self, node: CfgNode) -> list[CfgNode]: """ The nodes control may arrive at *node* from under this reading. """ if self is Projection.MAY: return node.predecessors if node.is_resumption_hub: return [] return [source for source in node.predecessors if not source.is_resumption_hub]
class ArmFlow (*args, **kwds)-
What an arm of a multi-way branch may reach when control runs off its end.
The three members are three languages' answers to the same construct, and reading one as another either invents paths that cannot be taken or drops paths that can — the second of which is the direction that lets an analysis call live code unreachable.
EXCLUSIVE— at most one arm ever runs, so an arm's exits leave the construct.SEQUENTIAL— an arm that runs off its end enters the next arm's body unconditionally, which is C-style fallthrough and what JavaScript'sswitchdoes.CUMULATIVE— every arm is tested in turn and every matching one runs, so an arm's exits may reach any later arm, not only the next. PowerShell'sswitchis this: it does not fall through, it keeps matching, and abreakis what stops it.Expand source code Browse git
class ArmFlow(enum.Enum): """ What an arm of a multi-way branch may reach when control runs off its end. The three members are three languages' answers to the same construct, and reading one as another either invents paths that cannot be taken or drops paths that can — the second of which is the direction that lets an analysis call live code unreachable. `EXCLUSIVE` — at most one arm ever runs, so an arm's exits leave the construct. `SEQUENTIAL` — an arm that runs off its end enters the *next* arm's body unconditionally, which is C-style fallthrough and what JavaScript's `switch` does. `CUMULATIVE` — every arm is tested in turn and every matching one runs, so an arm's exits may reach *any* later arm, not only the next. PowerShell's `switch` is this: it does not fall through, it keeps matching, and a `break` is what stops it. """ EXCLUSIVE = enum.auto() # noqa SEQUENTIAL = enum.auto() # noqa CUMULATIVE = enum.auto() # noqaAncestors
- enum.Enum
Class variables
var EXCLUSIVE-
The type of the None singleton.
var SEQUENTIAL-
The type of the None singleton.
var CUMULATIVE-
The type of the None singleton.
class CfgNode (graph, element, successors=<factory>, predecessors=<factory>, is_resumption_hub=False, is_hub_bound=False)-
One vertex of a control-flow graph.
elementis 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 — orNonefor the synthetic entry and exit.successorslists the nodes control may pass to next.graphis the body this vertex belongs to, which is what makes the questions below answerable of a node alone.is_resumption_hubmarks the synthetic fan-out standing for the over-approximate half of a resuming handler — seeCfgEdge. 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_boundmarks 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.
graphremains because a caller naming which body it means is a contract worth checking, and becauseControlFlowGraph.hub_boundanswers the same question for a caller holding ids.eq=Falsebecause every map in every layer above is keyed byid(node), and two structurally equal statements are two distinct points in the program.repr=Falsebecause a generated repr expandssuccessorsandpredecessors, 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, orpytest --showlocalswould 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 = FalseSubclasses
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 LabelledReach (bits, walks)-
Which of several labelled sets of sources reach each node, for every label at once: the answer
flood_labels()gives where a caller would otherwise walk the same graph once per label.A label reaches a node exactly when the walk of the same name, seeded with that label's sources alone, would include the node; each source reaches itself. A label that was never named reaches nothing. The labels are kept once per strongly connected component of what the sources reach, so the cost is one bit per label per component, and a caller that labels only what it will ask about pays for nothing else.
Expand source code Browse git
class LabelledReach(Generic[_Label]): """ Which of several labelled sets of sources reach each node, for every label at once: the answer `flood_labels` gives where a caller would otherwise walk the same graph once per label. A label reaches a node exactly when the walk of the same name, seeded with that label's sources alone, would include the node; each source reaches itself. A label that was never named reaches nothing. The labels are kept once per strongly connected component of what the sources reach, so the cost is one bit per label per component, and a caller that labels only what it will ask about pays for nothing else. """ def __init__(self, bits: dict[_Label, int], walks: list[tuple[Condensation[CfgNode], list[int]]]): self._bits = bits self._walks = walks def reaches(self, label: _Label, node: CfgNode) -> bool: """ Whether *label* reaches *node*. """ bit = self._bits.get(label) if bit is None: return False for condensation, masks in self._walks: number = condensation.number_of(node) if number is not None and masks[number] & bit: return True return FalseAncestors
- typing.Generic
Methods
def reaches(self, label, node)-
Whether label reaches node.
Expand source code Browse git
def reaches(self, label: _Label, node: CfgNode) -> bool: """ Whether *label* reaches *node*. """ bit = self._bits.get(label) if bit is None: return False for condensation, masks in self._walks: number = condensation.number_of(node) if number is not None and masks[number] & bit: return True return False
class ControlFlowGraph (owner)-
The control-flow graph of one function or script body.
entryandexitare synthetic; every other node wraps an AST element reachable throughnode_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 foundInstance variables
var hub_bound-
The ids of the nodes
CfgNode.is_hub_boundis set on, for a caller holding node ids rather than nodes. Written beside the field byCfgBuilder.mark_hub_bound(), which is the one place either is recorded.ReachabilityQueryis 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
CfgEdgediffer. A body with none — every script that writes notrap, and every one whose traps rethrow — draws no resumption edge and marks no node hub-bound, so a forward walk over it answers exactly whatreachable_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
Noneif 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()), orNonefor 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;
Nonewhere 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
Nonewhere 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 asCfgEdge.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
faultsasks — where an error goes — and it is narrower thanraise_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_intofor 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_takenanswers 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 atrywhose handler swallowed the throw — is in both. A caller that needs only the second set asksreached_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_reachanswers: the ids of the nodes reached from source by first leaving it along an edgeraise_takenanswers 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, andProjection.MAYwhere 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 inDominatorModel, 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_afterand_positionsare 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
nodeslists 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 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 = holderMethods
def node_of(self, element)-
The control-flow node standing for element in whichever graph owns it, or
Nonewhen 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
Nonewhen 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;
Nonesays 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
class Resumption (hub, slot=None)-
One statement block being built whose handler set resumes it rather than ending it, and the two synthetic nodes that carry the two projections of
CfgEdgefor it.hubis the over-approximate half: every point the block may resume from joins it, and it joins every node the block owns.slotis the precise half for the statement being built right now — the point control resumes at when that statement throws, which is the statement after it — and is created on the first node that needs one, because a statement building no node of its own resumes at the same place the one before it does.A frame owns only the nodes built while it is the innermost one. A nested resumable block opens a frame of its own and takes the nodes below it; what joins the two is one edge from the outer hub to the inner, and the tail of the inner block's forward chain arriving in the frontier the outer block threads on. Both are edges per nesting pair, where naming each node from each enclosing level is edges per node per level, and a script nested sixty-four deep costs twelve thousand edges rather than nine hundred.
Expand source code Browse git
@dataclass(eq=False) class Resumption: """ One statement block being built whose handler set *resumes* it rather than ending it, and the two synthetic nodes that carry the two projections of `CfgEdge` for it. `hub` is the over-approximate half: every point the block may resume from joins it, and it joins every node the block owns. `slot` is the precise half for the statement being built right now — the point control resumes at when that statement throws, which is the statement after it — and is created on the first node that needs one, because a statement building no node of its own resumes at the same place the one before it does. A frame owns only the nodes built while it is the *innermost* one. A nested resumable block opens a frame of its own and takes the nodes below it; what joins the two is one edge from the outer hub to the inner, and the tail of the inner block's forward chain arriving in the frontier the outer block threads on. Both are edges per nesting *pair*, where naming each node from each enclosing level is edges per node per level, and a script nested sixty-four deep costs twelve thousand edges rather than nine hundred. """ hub: CfgNode slot: CfgNode | None = NoneInstance variables
var hub-
The type of the None singleton.
var slot-
The type of the None singleton.
class CfgBuilder (owner)-
Single-pass construction of one
ControlFlowGraphby structural recursion over a body, threading a frontier — the set of nodes from which normal control currently falls through — into each statement and out the other side.A language subclasses this and implements
statementandbody_blocks. Everything else is shared, and the shape methods below take the parts of a construct rather than the construct, so that no code here has to know what a language calls the pieces of itsforloop.Expand source code Browse git
class CfgBuilder: """ Single-pass construction of one `ControlFlowGraph` by structural recursion over a body, threading a *frontier* — the set of nodes from which normal control currently falls through — into each statement and out the other side. A language subclasses this and implements `statement` and `body_blocks`. Everything else is shared, and the shape methods below take the parts of a construct rather than the construct, so that no code here has to know what a language calls the pieces of its `for` loop. """ def __init__(self, owner: Node): self.cfg = ControlFlowGraph(owner) self._handlers: list[CfgNode] = [] self._targets: list[_Target] = [] self._finalizers: list[_Finalizer] = [] self._resumptions: list[Resumption] = [] self._pending_labels: frozenset[str] = frozenset() def _open_target(self, target: _Target) -> None: """ Make *target* the innermost jump destination, remembering how many finalizers stand open outside it: a jump to it runs every finalizer opened since. """ target.finalizers = len(self._finalizers) self._targets.append(target) def _leave(self, node: CfgNode, jump: _Jump, target: _Target | None) -> None: """ Send the jump *node* to *target* the way *jump* arrives there, or out of the body where *target* is `None`. A jump that leaves finalizers on its way enters the innermost of them and goes on from its end, one finalizer after the other outward, and to *target* from the end of the outermost; `guarded` draws those edges once the finalizer bodies are built. No edge joins the jump to *target* directly: control arrives there only after every finalizer it leaves has run to its end. """ crossed = self._finalizers if target is None else self._finalizers[target.finalizers:] if not crossed: self._arrive(node, jump, target) return self.add_edge(node, crossed[-1].entry) for finalizer in crossed[1:]: finalizer.passes_outward = True crossed[0].landings[jump, id(target)] = target def _arrive(self, node: CfgNode, jump: _Jump, target: _Target | None) -> None: """ Join *node* to where a jump arrives at *target* the way *jump* says, or to the body's exit where *target* is `None`. """ if target is None: self.add_edge(node, self.cfg.exit) elif jump is _Jump.OUT: target.breaks.append(node) elif target.continue_to is not None: self.add_edge(node, target.continue_to) else: target.continues.append(node) def _leave_finalizer(self, finalizer: _Finalizer, ends: list[CfgNode]) -> None: """ Join *ends*, the points from which the body of *finalizer* runs off its end, to where the runs that entered it go on besides what follows its guarded statement. Each goes on from a point of its own, built here and so inside the guarded statement: a reader asking where a run goes once that statement completed meets none of them past it. An error that entered the finalizer goes on outward. A jump that left the block through it goes on to the next finalizer out where it leaves that one too, and else to the place it jumps to, one point per place. A finalizer that jumps or throws itself has no *ends*, and the run it was entered with goes nowhere from it. """ entry = finalizer.entry if any(self.cfg.is_exceptional(source, entry) for source in entry.predecessors): self.exceptional_edge(self._point_after(ends), self.escape()) if finalizer.passes_outward: self.add_edge(self._point_after(ends), self._finalizers[-1].entry) for (jump, _), target in finalizer.landings.items(): self._arrive(self._point_after(ends), jump, target) def _point_after(self, ends: list[CfgNode]) -> CfgNode: node = self.synthetic_node() self.link(ends, node) return node def build(self) -> ControlFlowGraph: frontier = [self.cfg.entry] for statements in self.body_blocks(self.cfg.owner): frontier = self.block(statements, frontier) self.link(frontier, self.cfg.exit) return self.cfg def body_blocks(self, owner: Node) -> Sequence[Sequence[Node]]: """ The statement blocks *owner* runs, in the order it runs them. A language whose body may be a single unbraced statement resolves that here, and one that splits a body across several named blocks reports them apart rather than joined, because a construct scoped to a block cannot be modelled from a body that has forgotten where its blocks ended. """ raise NotImplementedError def block(self, statements: Sequence[Node], frontier: list[CfgNode]) -> list[CfgNode]: """ One statement block. The statements in order, which is all a block is to a language whose blocks carry nothing of their own; a language whose blocks scope a handler installs it here and calls this for every block it builds. """ return self.sequence(statements, frontier) def statement(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ Add *statement* to the graph and report the frontier that follows it. A language recognises its own node types here and calls the shape method that matches; anything it does not recognise goes to `opaque`, which is the conservative answer. """ raise NotImplementedError def node(self, element: Node) -> CfgNode: """ A graph node for *element*, joined to the innermost active handler if one is open. The handler edge is added here rather than at each site that creates a node, because *any* statement inside a guarded block may throw and the whole point of the edge is that it does not depend on which statement it is; `offers` is where a language says otherwise. """ node = self.detached_node(element) if self._handlers: target = self._handlers[-1] if self.offers(element, target): self.exceptional_edge(node, target) return node def offers(self, element: Node, target: CfgNode) -> bool: """ Whether the node for *element* gets an error edge to *target*, the innermost entry of the handler stack. An edge into a handler stands for a position, so that a question about where an error raised there would go is answered whatever stands there, and it is always drawn. A language whose stack holds something other than a handler may refuse an edge into that. """ return True def synthetic_node(self) -> CfgNode: """ A graph node standing for no element: a join or fan-out point the graph needs and the program does not name. It locates into no query and generates no dataflow fact, which is what lets one stand in for a set of edges between real nodes without an analysis reading it as a statement. """ node = CfgNode(self.cfg, None) self.cfg.nodes.append(node) return node def detached_node(self, element: Node) -> CfgNode: """ A graph node for *element* that the enclosing handler does not guard. A handler's own entry is one: it stands for the point at which a throw is *offered* to a clause, which is not a point inside the region that clause guards. The edge outward from there says the clause did not take the throw, and whether any run says that is a property of the whole handler set, which `guarded` reads once. This is the one place an element gets a node, which is why the resumption edges are drawn here rather than by the caller that knows a block resumes. A block's own statements, the arms of a construct nested in it, and the entries of a handler set written inside it are built by three different methods that share nothing but this one, and a level that had to name its nodes afterwards would have to name the ones its nested levels built as well. """ node = CfgNode(self.cfg, element) self.cfg.nodes.append(node) self.cfg._node_of[id(element)] = node if self._resumptions: self.join_resumption(node) return node @staticmethod def add_edge(source: CfgNode, target: CfgNode) -> None: """ A plain edge: `kind_edge` with `CfgEdge.NORMAL`. """ _draw(source, target, CfgEdge.NORMAL) def kind_edge(self, source: CfgNode, target: CfgNode, kind: CfgEdge) -> None: """ An edge carrying *kind*. """ _draw(source, target, kind) def exceptional_edge(self, source: CfgNode, target: CfgNode) -> None: self.kind_edge(source, target, CfgEdge.ERROR_CARRYING) def resumption_hub(self) -> CfgNode: """ A fresh synthetic node standing for the over-approximate half of a resuming handler: every point the block may resume from joins it, and it joins every point the block may resume at. Recorded as a hub so that `CfgNode.is_resumption_hub` answers for it, which is what a forward-only walk declines to enter. """ node = self.synthetic_node() node.is_resumption_hub = True self.cfg._resuming = True return node def resumption_hub_edge(self, source: CfgNode, target: CfgNode) -> None: """ An edge *out* of a resumption hub — see `CfgEdge`. The way in is drawn plain, because the handler body reaching it completed. """ self.cfg._resuming = True self.kind_edge(source, target, CfgEdge.RESUMPTION_HUB) def resumption_forward_edge(self, source: CfgNode, target: CfgNode) -> None: """ An edge of the precise, forward-only half of a resuming handler — see `CfgEdge`. """ self.cfg._resuming = True self.kind_edge(source, target, CfgEdge.RESUMPTION_FORWARD) def mark_hub_bound(self, nodes: Iterable[CfgNode]) -> None: """ Record that the forward-only projection of resumption does not reach *nodes* — see `CfgNode.is_hub_bound`. A hub among *nodes* is passed over. A caller naming a handler body names whatever the level it walked appended, and a resumable block written inside that body contributes its own hub — a node the forward reading declines to stand at, not one it fails to reach. Marking it both would leave one node routed to the over-approximate walk as a *source* and refused as a *neighbour* in the same query, and `ControlFlowGraph.hub_bound` would report a fan-out as a statement. """ self.cfg._resuming = True for node in nodes: if node.is_resumption_hub: continue node.is_hub_bound = True self.cfg._hub_bound.add(id(node)) @contextmanager def resumption(self, resumes: Iterable[CfgNode]) -> Iterator[None]: """ Build the enclosed block as one whose handler set resumes it from *resumes*, taking ownership of every node built inside it — see `Resumption`. The hub is wired on the way in, before the block is walked, because `detached_node` draws the edge out of it as each node appears. Where a resumable block is opened inside one, the outer hub reaches the inner rather than reaching the inner block's nodes one by one: a hub reaches exactly the nodes its block owns and everything those nodes lead to, and an inner hub is one of them. Ownership ends on every way out, which is why this is a block and not a pair of calls. A build abandoned part-way through a guarded statement would otherwise leave the frame open, and every node built afterwards would draw both halves of resumption against a handler set that does not guard it. """ hub = self.resumption_hub() self.resumption_hub_edge(hub, self.cfg.exit) if self._resumptions: self.resumption_hub_edge(self._resumptions[-1].hub, hub) for resume in resumes: self.add_edge(resume, hub) self._resumptions.append(Resumption(hub)) try: yield finally: self._resumptions.pop() def resume_past(self, frontier: list[CfgNode]) -> list[CfgNode]: """ *frontier* extended by the point control resumes at when the statement just built throws, with the innermost open block left ready for the statement after it. `join_resumption` claims the slot for the first node the statement builds and joins every later one to it, so there is one here exactly when the statement built a node of its own. A statement that built none claims nothing — a `trap` declaration is one, and so is an empty nested block — and the slot of the statement before it stays in the frontier and carries on to the next, which is where control really resumes. Claiming and clearing the slot are one step here so that no caller holds the frame or has to remember to reset it between statements. `distinct` is what keeps a statement that hands its incoming frontier straight back from drawing the same edge once per statement that passed it on. """ frame = self._resumptions[-1] slot, frame.slot = frame.slot, None if slot is None: return frontier return distinct([*frontier, slot]) def join_resumption(self, node: CfgNode) -> None: """ Draw both projections of resumption for *node* against the innermost open block: the hub reaches it, and it resumes at the slot standing for the statement being built. One edge each. A node reached from the hub of the level above it is reached from this one too, through the edge between the two hubs, and a node resuming at this level's slot carries on to the level above through the frontier the inner block hands back — so neither half loses a path by naming only the innermost level, and both stop costing an edge per level. """ frame = self._resumptions[-1] if frame.slot is None: frame.slot = self.synthetic_node() self.resumption_forward_edge(node, frame.slot) self.resumption_hub_edge(frame.hub, node) def close_handler_set( self, entries: Sequence[CfgNode], *, escapes: bool, outward: CfgNode | None = None, ) -> None: """ Finish a chain of handler entries: draw the edge from the last of them to what guards the construct when some run may leave all of them. Called once the set's bodies are built and the set itself is off the handler stack, so that `unwinding` names what guards the *construct* rather than the set being closed. """ if escapes and entries: self.exceptional_edge(entries[-1], self.unwinding() if outward is None else outward) def unwinding(self) -> CfgNode: """ Where a throw goes when nothing open here takes it: the innermost active handler, or the body's exit, which is how the graph says the throw leaves the body. """ return self._handlers[-1] if self._handlers else self.cfg.exit def escape(self) -> CfgNode: """ Where an error goes once it gets past every handler of the construct being closed. That is `unwinding` unless the language hands such an error to something that is not a handler. """ return self.unwinding() def handler_body_guard(self) -> CfgNode | None: """ What guards the handler bodies of a guarded block that has no finalizer, or `None` where nothing beyond what guards the construct does. A language that hands an error leaving a handler body to something that is not a handler names it here. """ return None def link(self, frontier: Iterable[CfgNode], target: CfgNode) -> None: for node in frontier: self.add_edge(node, target) def sequence(self, statements: Sequence[Node], frontier: list[CfgNode]) -> list[CfgNode]: for statement in statements: frontier = self.member(statement, frontier) self.cfg._completions[id(statement)] = (statement, list(frontier)) return frontier def member(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ Build *statement* as a member of a statement sequence and record the nodes it built, which `ControlFlowGraph.builder_of` answers from. A statement whose own node already exists was started by the construct around it — a declaration its block hoists, or a finalizer whose entry the guarded construct draws first — so the nodes built here are not all of its nodes, and none are recorded. """ if self.cfg.node_of(statement) is not None: return self.statement(statement, frontier) start = len(self.cfg.nodes) frontier = self.statement(statement, frontier) self.cfg._spans[id(statement)] = (statement, range(start, len(self.cfg.nodes))) return frontier def opaque(self, element: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ An element whose internal control flow this does not model: one node, entered from the frontier and falling through. The default for anything a language does not recognise. """ node = self.node(element) self.link(frontier, node) return [node] def evaluated(self, element: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ The nodes evaluating *element*, a part of a construct evaluated at a point of its own, such as the initializer of a counted loop: one node, unless a language evaluates the part in steps it gives nodes of their own. """ return self.opaque(element, frontier) def _body(self, body: Node | None, frontier: list[CfgNode]) -> list[CfgNode]: return self.statement(body, list(frontier)) if body is not None else list(frontier) def _branch(self, body: Node | None, head: CfgNode) -> list[CfgNode]: return self._body(body, [head]) def _capture_body( self, body: Node | None, frontier: list[CfgNode], ) -> tuple[CfgNode | None, list[CfgNode]]: """ Build *body* and return its entry node — the node control reaches first — alongside its exit frontier. Used where a back-edge must target the body's own entry, which the plain frontier threading does not expose. The entry is the first successor the incoming *frontier* gains while *body* is built, not the first node created. A body that opens with a guarded block builds its handler or finalizer node before any guarded statement, so creation order would return that handler — a node with no edge back into the body — and the loop's back-edge would be wired to it, hiding the real body head from a backward reachability walk. The frontier instead links to the first guarded statement, which is the node control actually enters. """ before = [(node, len(node.successors)) for node in frontier] exits = self._body(body, frontier) for node, count in before: if len(node.successors) > count: return node.successors[count], exits return None, exits def park_label(self, *labels: str | None) -> None: """ Park *labels* for the construct about to be built, for `take_labels` to consume. `labelled` calls this for a language whose label is a statement wrapping the construct. A language whose label is a field *of* the construct calls it directly, because there is no wrapping statement to recognise — and a construct built without it carries no label, so every jump naming it misses and leaves the body instead. A construct may carry several labels, as `a: b: while (…)` does, and a jump naming any of them targets it. """ self._pending_labels = frozenset(label for label in labels if label is not None) def take_labels(self) -> frozenset[str]: """ The labels parked for the construct about to be built, consumed once. """ labels = self._pending_labels self._pending_labels = frozenset() return labels def branch_on( self, element: Node, arms: Sequence[Node | None], frontier: list[CfgNode], *, exhaustive: bool = False, ) -> list[CfgNode]: """ A conditional: one head node the frontier enters, and one arm per branch. When the arms do not cover every case — an `if` with no `else` — the head itself is an exit, because control may pass the whole construct without entering any arm. `exhaustive` says the arms do cover it, which is what an `if`/`else` pair reports. """ head = self.node(element) self.link(frontier, head) exits: list[CfgNode] = [] for arm in arms: exits += self._branch(arm, head) if not exhaustive: exits.append(head) return exits def loop_head_tested( self, element: Node, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop whose condition is evaluated before the body, so the head is both the entry and an exit: `while`, and every `foreach` whose iteration may run zero times. """ head = self.node(element) self.link(frontier, head) target = _Target(self.take_labels(), [], head, is_continuable=True, is_breakable=True) self._open_target(target) body_exits = self._branch(body, head) self._targets.pop() self.link(body_exits, head) return [head] + target.breaks def loop_over( self, element: Node, subject: Node | None, target: Node | None, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop over the values a subject yields. The subject is evaluated once, at a node of its own before the loop. The head, which asks for the next value, is the entry of every iteration and an exit. The target receives the value at a node of its own on the way from the head into the body, so a run leaving the loop has not passed it. A language whose loop variable and subject are both read at the head builds the loop `loop_head_tested` instead. """ if subject is not None: start = self.node(subject) self.link(frontier, start) frontier = [start] head = self.node(element) self.link(frontier, head) loop = _Target(self.take_labels(), [], head, is_continuable=True, is_breakable=True) self._open_target(loop) entry = head if target is not None: entry = self.node(target) self.link([head], entry) body_exits = self._branch(body, entry) self._targets.pop() self.link(body_exits, head) return [head] + loop.breaks def loop_tail_tested( self, element: Node, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop whose condition is evaluated after the body, so the body always runs once and the back-edge targets the body's own entry rather than the test. """ test = self.node(element) target = _Target(self.take_labels(), [], test, is_continuable=True, is_breakable=True) self._open_target(target) entry, body_exits = self._capture_body(body, frontier) self._targets.pop() self.link(body_exits, test) self.add_edge(test, entry if entry is not None else test) return [test] + target.breaks def loop_counted( self, initializer: Node | None, test: Node | None, update: Node | None, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop with a separate initializer, test and update, each evaluated at its own point and so each given its own node. Any of the three may be absent; a loop with no test has no exit other than the jumps out of it, and its back-edge targets the body entry. """ labels = self.take_labels() if initializer is not None: frontier = self.evaluated(initializer, frontier) head = self.node(test) if test is not None else None if head is not None: self.link(frontier, head) body_frontier: list[CfgNode] = [head] else: body_frontier = list(frontier) step = self.node(update) if update is not None else None target = _Target(labels, [], step or head, is_continuable=True, is_breakable=True) self._open_target(target) entry, body_exits = self._capture_body(body, body_frontier) self._targets.pop() latch = body_exits if step is not None: self.link(body_exits, step) latch = [step] back_to = head if head is not None else entry if back_to is not None: self.link(latch, back_to) self.link(target.continues, back_to if back_to is not None else self.cfg.exit) exits = list(target.breaks) if head is not None: exits.append(head) return exits def branch_chain( self, clauses: Sequence[tuple[Node, Node | None]], otherwise: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A chain of guarded arms, each tested only when every earlier test failed: `if`/`elseif`/`else` where the whole chain is one node rather than a nest of two-armed conditionals. Each test gets its own node, because the tests run at different points and an analysis that collapsed them could not order two of them. A test's node flows into its own arm and on to the next test; the last test flows into `otherwise` when there is one, and out of the construct when there is not. """ exits: list[CfgNode] = [] current = frontier for test, body in clauses: head = self.node(test) self.link(current, head) exits += self._branch(body, head) current = [head] if otherwise is not None: return distinct(exits + self._body(otherwise, current)) return distinct(exits + current) def dispatch( self, element: Node, arms: Sequence[Sequence[Node]], frontier: list[CfgNode], *, arm_flow: ArmFlow, exhaustive: bool, iterated: bool = False, ) -> list[CfgNode]: """ A multi-way branch: one head the frontier enters and one arm per clause, each arm a statement sequence the head may jump into. `arm_flow` says what an arm reaches when it runs off its end; see `ArmFlow`, which is where the languages differ. `exhaustive` says some arm always runs — a default clause — so the head is not itself an exit. `iterated` says the construct runs its arms once per element of an input, so it is a loop: a back-jump inside an arm re-enters the head rather than resolving to an enclosing loop, and an arm that simply runs off its end re-enters it too, for the next element. Without that second edge the arms lie on no cycle and an analysis reads a store inside one as happening once, which is what lets a self-referential assignment be folded to its first value. PowerShell's `switch` is this; JavaScript's is not, and reading one as the other both invents a back-edge to a loop the jump never reaches and drops the one it does. A jump *out* takes no back-edge: it leaves the construct rather than advancing it, which is the whole difference between the two spellings. """ head = self.node(element) self.link(frontier, head) target = _Target( self.take_labels(), [], head if iterated else None, is_continuable=iterated, is_breakable=True, ) self._open_target(target) carried: list[CfgNode] = [] completed: list[CfgNode] = [] for arm in arms: reached = self.sequence(list(arm), distinct([head, *carried])) if arm_flow is ArmFlow.SEQUENTIAL: carried = reached elif arm_flow is ArmFlow.CUMULATIVE: carried = distinct([*carried, *reached]) else: completed += reached self._targets.pop() completed = distinct(completed + carried) if iterated: self.link(completed, head) exits = completed + target.breaks if not exhaustive: exits.append(head) return distinct(exits) def guarded( self, block: Node | None, handlers: Sequence[tuple[Node, Node | None]], finalizer: Node | None, finalizer_body: Sequence[Node], frontier: list[CfgNode], *, escapes: bool, ) -> list[CfgNode]: """ A guarded block with any number of handlers and an optional finalizer. The handler nodes are created *before* the guarded block is built and the first is pushed on the handler stack, so that `node` joins every statement created inside the block to it. That ordering is the whole mechanism: it is why no statement inside the block has to know it is guarded. Several handlers are chained from the first, because which one runs depends on the type of the exception and none of them is guaranteed — a language with one handler passes a one-element sequence and the chain degenerates. The chain edge is *exceptional*: control takes it exactly when the earlier handler did not match, so nothing that handler's node stands for has run and its kill must not apply along it. `escapes` says the set of handlers may fail to take a throw the block makes, which is what a language whose clauses carry a type filter reports and what one whose single clause takes everything denies. It decides one edge — from the last handler to whatever guards this construct, or to the body exit when nothing does, which is how the graph says a throw leaves the body with no handler having taken it. Denying the edge is the difference between a clause that swallows and one that passes the throw on, so it is asked of every language rather than defaulted: the safe answer invents a path the program cannot take, and the other drops one it can. The finalizer is entered from the block's normal exits and from every handler's, and runs on every other way out of the construct as well: an error no handler takes enters it, and so does a jump leaving the block or a handler body, and each goes on from its end, which `_leave_finalizer` draws. It also guards every handler body: a handler that throws, returns or jumps out of the construct runs the finalizer on its way, so where the block and every handler end that way, the finalizer is still reached. """ handlers = list(handlers) entries = [self.detached_node(handler) for handler, _ in handlers] finalizer_entry = self.detached_node(finalizer) if finalizer is not None else None record = _Finalizer(finalizer_entry) if finalizer_entry is not None else None guard = entries[0] if entries else finalizer_entry if record is not None: self._finalizers.append(record) if guard is not None: self._handlers.append(guard) block_exits = self.statement(block, frontier) if block is not None else list(frontier) if guard is not None: self._handlers.pop() normal_exits = list(block_exits) clause_guard = finalizer_entry if clause_guard is None and entries: clause_guard = self.handler_body_guard() if clause_guard is not None: self._handlers.append(clause_guard) for index, ((_, body), entry) in enumerate(zip(handlers, entries)): if index: self.exceptional_edge(entries[index - 1], entry) normal_exits += ( self.statement(body, [entry]) if body is not None else [entry]) if clause_guard is not None: self._handlers.pop() if record is not None: self._finalizers.pop() outward = self.escape() if finalizer_entry is None else finalizer_entry self.close_handler_set(entries, escapes=escapes, outward=outward) if record is not None and finalizer_entry is not None: self.link(normal_exits, finalizer_entry) final_exits = self.sequence(list(finalizer_body), [finalizer_entry]) self._leave_finalizer(record, final_exits) return final_exits return normal_exits def labelled( self, labels: Sequence[str], body: Node | None, frontier: list[CfgNode], *, binds_to_body: bool, ) -> list[CfgNode]: """ A statement *body* with every label written in front of it, *labels*. When they name a construct a jump can target directly — `binds_to_body` — they are parked for that construct to consume through `take_labels`; otherwise they name this statement itself and only a break can leave it. """ if binds_to_body: self.park_label(*labels) return self.statement(body, frontier) if body is not None else list(frontier) target = _Target(frozenset(labels), [], None, is_continuable=False, is_breakable=False) self._open_target(target) exits = self.statement(body, frontier) if body is not None else list(frontier) self._targets.pop() return exits + target.breaks def jump_out( self, element: Node, label: str | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A jump that leaves the construct it names, or the innermost breakable one when unlabelled. A jump naming nothing this body holds leaves the body, which is the conservative reading. """ node = self.node(element) self.link(frontier, node) self._leave(node, _Jump.OUT, self._break_target(label)) return [] def jump_back( self, element: Node, label: str | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A jump to the next iteration of the loop it names, or of the innermost loop when unlabelled. """ node = self.node(element) self.link(frontier, node) self._leave(node, _Jump.BACK, self._continue_target(label)) return [] def terminate( self, element: Node, frontier: list[CfgNode], *, exceptional: bool, ) -> list[CfgNode]: """ A statement after which control does not continue in this body: a return, or a throw. A throw is `exceptional`, so it reaches the innermost open handler rather than the exit, and only reaches the exit when no handler is open. A return leaves the body through every finalizer open around it, as a jump out of the body does. """ node = self.node(element) self.link(frontier, node) if exceptional: self.exceptional_edge(node, self.unwinding()) else: self._leave(node, _Jump.OUT, None) return [] def has_continue_target(self, label: str | None) -> bool: """ Whether a back-jump naming *label* — or an unlabelled one — resolves to a construct currently being built. A language in which `continue` means something other than a back-jump when no such construct is open asks this to tell the two spellings apart. """ return self._continue_target(label) is not None def has_break_target(self, label: str | None) -> bool: """ Whether a jump out naming *label* — or an unlabelled one — resolves to a construct currently being built, the counterpart of `has_continue_target` for the language in which `break` means something other than leaving a construct when no construct is open to leave. """ return self._break_target(label) is not None def _break_target(self, label: str | None) -> _Target | None: for target in reversed(self._targets): if label is None: if target.is_breakable: return target elif label in target.labels: return target return None def _continue_target(self, label: str | None) -> _Target | None: for target in reversed(self._targets): if not target.is_continuable: continue if label is None or label in target.labels: return target return NoneSubclasses
- refinery.lib.scripts.js.analysis.cfg._Builder
- refinery.lib.scripts.ps1.analysis.cfg._Builder
Static methods
def add_edge(source, target)-
A plain edge:
kind_edgewithCfgEdge.NORMAL.Expand source code Browse git
@staticmethod def add_edge(source: CfgNode, target: CfgNode) -> None: """ A plain edge: `kind_edge` with `CfgEdge.NORMAL`. """ _draw(source, target, CfgEdge.NORMAL)
Methods
def build(self)-
Expand source code Browse git
def build(self) -> ControlFlowGraph: frontier = [self.cfg.entry] for statements in self.body_blocks(self.cfg.owner): frontier = self.block(statements, frontier) self.link(frontier, self.cfg.exit) return self.cfg def body_blocks(self, owner)-
The statement blocks owner runs, in the order it runs them. A language whose body may be a single unbraced statement resolves that here, and one that splits a body across several named blocks reports them apart rather than joined, because a construct scoped to a block cannot be modelled from a body that has forgotten where its blocks ended.
Expand source code Browse git
def body_blocks(self, owner: Node) -> Sequence[Sequence[Node]]: """ The statement blocks *owner* runs, in the order it runs them. A language whose body may be a single unbraced statement resolves that here, and one that splits a body across several named blocks reports them apart rather than joined, because a construct scoped to a block cannot be modelled from a body that has forgotten where its blocks ended. """ raise NotImplementedError def block(self, statements, frontier)-
One statement block. The statements in order, which is all a block is to a language whose blocks carry nothing of their own; a language whose blocks scope a handler installs it here and calls this for every block it builds.
Expand source code Browse git
def block(self, statements: Sequence[Node], frontier: list[CfgNode]) -> list[CfgNode]: """ One statement block. The statements in order, which is all a block is to a language whose blocks carry nothing of their own; a language whose blocks scope a handler installs it here and calls this for every block it builds. """ return self.sequence(statements, frontier) def statement(self, statement, frontier)-
Add statement to the graph and report the frontier that follows it. A language recognises its own node types here and calls the shape method that matches; anything it does not recognise goes to
opaque, which is the conservative answer.Expand source code Browse git
def statement(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ Add *statement* to the graph and report the frontier that follows it. A language recognises its own node types here and calls the shape method that matches; anything it does not recognise goes to `opaque`, which is the conservative answer. """ raise NotImplementedError def node(self, element)-
A graph node for element, joined to the innermost active handler if one is open.
The handler edge is added here rather than at each site that creates a node, because any statement inside a guarded block may throw and the whole point of the edge is that it does not depend on which statement it is;
offersis where a language says otherwise.Expand source code Browse git
def node(self, element: Node) -> CfgNode: """ A graph node for *element*, joined to the innermost active handler if one is open. The handler edge is added here rather than at each site that creates a node, because *any* statement inside a guarded block may throw and the whole point of the edge is that it does not depend on which statement it is; `offers` is where a language says otherwise. """ node = self.detached_node(element) if self._handlers: target = self._handlers[-1] if self.offers(element, target): self.exceptional_edge(node, target) return node def offers(self, element, target)-
Whether the node for element gets an error edge to target, the innermost entry of the handler stack. An edge into a handler stands for a position, so that a question about where an error raised there would go is answered whatever stands there, and it is always drawn. A language whose stack holds something other than a handler may refuse an edge into that.
Expand source code Browse git
def offers(self, element: Node, target: CfgNode) -> bool: """ Whether the node for *element* gets an error edge to *target*, the innermost entry of the handler stack. An edge into a handler stands for a position, so that a question about where an error raised there would go is answered whatever stands there, and it is always drawn. A language whose stack holds something other than a handler may refuse an edge into that. """ return True def synthetic_node(self)-
A graph node standing for no element: a join or fan-out point the graph needs and the program does not name. It locates into no query and generates no dataflow fact, which is what lets one stand in for a set of edges between real nodes without an analysis reading it as a statement.
Expand source code Browse git
def synthetic_node(self) -> CfgNode: """ A graph node standing for no element: a join or fan-out point the graph needs and the program does not name. It locates into no query and generates no dataflow fact, which is what lets one stand in for a set of edges between real nodes without an analysis reading it as a statement. """ node = CfgNode(self.cfg, None) self.cfg.nodes.append(node) return node def detached_node(self, element)-
A graph node for element that the enclosing handler does not guard.
A handler's own entry is one: it stands for the point at which a throw is offered to a clause, which is not a point inside the region that clause guards. The edge outward from there says the clause did not take the throw, and whether any run says that is a property of the whole handler set, which
guardedreads once.This is the one place an element gets a node, which is why the resumption edges are drawn here rather than by the caller that knows a block resumes. A block's own statements, the arms of a construct nested in it, and the entries of a handler set written inside it are built by three different methods that share nothing but this one, and a level that had to name its nodes afterwards would have to name the ones its nested levels built as well.
Expand source code Browse git
def detached_node(self, element: Node) -> CfgNode: """ A graph node for *element* that the enclosing handler does not guard. A handler's own entry is one: it stands for the point at which a throw is *offered* to a clause, which is not a point inside the region that clause guards. The edge outward from there says the clause did not take the throw, and whether any run says that is a property of the whole handler set, which `guarded` reads once. This is the one place an element gets a node, which is why the resumption edges are drawn here rather than by the caller that knows a block resumes. A block's own statements, the arms of a construct nested in it, and the entries of a handler set written inside it are built by three different methods that share nothing but this one, and a level that had to name its nodes afterwards would have to name the ones its nested levels built as well. """ node = CfgNode(self.cfg, element) self.cfg.nodes.append(node) self.cfg._node_of[id(element)] = node if self._resumptions: self.join_resumption(node) return node def kind_edge(self, source, target, kind)-
An edge carrying kind.
Expand source code Browse git
def kind_edge(self, source: CfgNode, target: CfgNode, kind: CfgEdge) -> None: """ An edge carrying *kind*. """ _draw(source, target, kind) def exceptional_edge(self, source, target)-
Expand source code Browse git
def exceptional_edge(self, source: CfgNode, target: CfgNode) -> None: self.kind_edge(source, target, CfgEdge.ERROR_CARRYING) def resumption_hub(self)-
A fresh synthetic node standing for the over-approximate half of a resuming handler: every point the block may resume from joins it, and it joins every point the block may resume at. Recorded as a hub so that
CfgNode.is_resumption_hubanswers for it, which is what a forward-only walk declines to enter.Expand source code Browse git
def resumption_hub(self) -> CfgNode: """ A fresh synthetic node standing for the over-approximate half of a resuming handler: every point the block may resume from joins it, and it joins every point the block may resume at. Recorded as a hub so that `CfgNode.is_resumption_hub` answers for it, which is what a forward-only walk declines to enter. """ node = self.synthetic_node() node.is_resumption_hub = True self.cfg._resuming = True return node def resumption_hub_edge(self, source, target)-
An edge out of a resumption hub — see
CfgEdge. The way in is drawn plain, because the handler body reaching it completed.Expand source code Browse git
def resumption_hub_edge(self, source: CfgNode, target: CfgNode) -> None: """ An edge *out* of a resumption hub — see `CfgEdge`. The way in is drawn plain, because the handler body reaching it completed. """ self.cfg._resuming = True self.kind_edge(source, target, CfgEdge.RESUMPTION_HUB) def resumption_forward_edge(self, source, target)-
An edge of the precise, forward-only half of a resuming handler — see
CfgEdge.Expand source code Browse git
def resumption_forward_edge(self, source: CfgNode, target: CfgNode) -> None: """ An edge of the precise, forward-only half of a resuming handler — see `CfgEdge`. """ self.cfg._resuming = True self.kind_edge(source, target, CfgEdge.RESUMPTION_FORWARD) def mark_hub_bound(self, nodes)-
Record that the forward-only projection of resumption does not reach nodes — see
CfgNode.is_hub_bound.A hub among nodes is passed over. A caller naming a handler body names whatever the level it walked appended, and a resumable block written inside that body contributes its own hub — a node the forward reading declines to stand at, not one it fails to reach. Marking it both would leave one node routed to the over-approximate walk as a source and refused as a neighbour in the same query, and
ControlFlowGraph.hub_boundwould report a fan-out as a statement.Expand source code Browse git
def mark_hub_bound(self, nodes: Iterable[CfgNode]) -> None: """ Record that the forward-only projection of resumption does not reach *nodes* — see `CfgNode.is_hub_bound`. A hub among *nodes* is passed over. A caller naming a handler body names whatever the level it walked appended, and a resumable block written inside that body contributes its own hub — a node the forward reading declines to stand at, not one it fails to reach. Marking it both would leave one node routed to the over-approximate walk as a *source* and refused as a *neighbour* in the same query, and `ControlFlowGraph.hub_bound` would report a fan-out as a statement. """ self.cfg._resuming = True for node in nodes: if node.is_resumption_hub: continue node.is_hub_bound = True self.cfg._hub_bound.add(id(node)) def resumption(self, resumes)-
Build the enclosed block as one whose handler set resumes it from resumes, taking ownership of every node built inside it — see
Resumption.The hub is wired on the way in, before the block is walked, because
detached_nodedraws the edge out of it as each node appears. Where a resumable block is opened inside one, the outer hub reaches the inner rather than reaching the inner block's nodes one by one: a hub reaches exactly the nodes its block owns and everything those nodes lead to, and an inner hub is one of them.Ownership ends on every way out, which is why this is a block and not a pair of calls. A build abandoned part-way through a guarded statement would otherwise leave the frame open, and every node built afterwards would draw both halves of resumption against a handler set that does not guard it.
Expand source code Browse git
@contextmanager def resumption(self, resumes: Iterable[CfgNode]) -> Iterator[None]: """ Build the enclosed block as one whose handler set resumes it from *resumes*, taking ownership of every node built inside it — see `Resumption`. The hub is wired on the way in, before the block is walked, because `detached_node` draws the edge out of it as each node appears. Where a resumable block is opened inside one, the outer hub reaches the inner rather than reaching the inner block's nodes one by one: a hub reaches exactly the nodes its block owns and everything those nodes lead to, and an inner hub is one of them. Ownership ends on every way out, which is why this is a block and not a pair of calls. A build abandoned part-way through a guarded statement would otherwise leave the frame open, and every node built afterwards would draw both halves of resumption against a handler set that does not guard it. """ hub = self.resumption_hub() self.resumption_hub_edge(hub, self.cfg.exit) if self._resumptions: self.resumption_hub_edge(self._resumptions[-1].hub, hub) for resume in resumes: self.add_edge(resume, hub) self._resumptions.append(Resumption(hub)) try: yield finally: self._resumptions.pop() def resume_past(self, frontier)-
frontier extended by the point control resumes at when the statement just built throws, with the innermost open block left ready for the statement after it.
join_resumptionclaims the slot for the first node the statement builds and joins every later one to it, so there is one here exactly when the statement built a node of its own. A statement that built none claims nothing — atrapdeclaration is one, and so is an empty nested block — and the slot of the statement before it stays in the frontier and carries on to the next, which is where control really resumes.Claiming and clearing the slot are one step here so that no caller holds the frame or has to remember to reset it between statements.
distinct()is what keeps a statement that hands its incoming frontier straight back from drawing the same edge once per statement that passed it on.Expand source code Browse git
def resume_past(self, frontier: list[CfgNode]) -> list[CfgNode]: """ *frontier* extended by the point control resumes at when the statement just built throws, with the innermost open block left ready for the statement after it. `join_resumption` claims the slot for the first node the statement builds and joins every later one to it, so there is one here exactly when the statement built a node of its own. A statement that built none claims nothing — a `trap` declaration is one, and so is an empty nested block — and the slot of the statement before it stays in the frontier and carries on to the next, which is where control really resumes. Claiming and clearing the slot are one step here so that no caller holds the frame or has to remember to reset it between statements. `distinct` is what keeps a statement that hands its incoming frontier straight back from drawing the same edge once per statement that passed it on. """ frame = self._resumptions[-1] slot, frame.slot = frame.slot, None if slot is None: return frontier return distinct([*frontier, slot]) def join_resumption(self, node)-
Draw both projections of resumption for node against the innermost open block: the hub reaches it, and it resumes at the slot standing for the statement being built.
One edge each. A node reached from the hub of the level above it is reached from this one too, through the edge between the two hubs, and a node resuming at this level's slot carries on to the level above through the frontier the inner block hands back — so neither half loses a path by naming only the innermost level, and both stop costing an edge per level.
Expand source code Browse git
def join_resumption(self, node: CfgNode) -> None: """ Draw both projections of resumption for *node* against the innermost open block: the hub reaches it, and it resumes at the slot standing for the statement being built. One edge each. A node reached from the hub of the level above it is reached from this one too, through the edge between the two hubs, and a node resuming at this level's slot carries on to the level above through the frontier the inner block hands back — so neither half loses a path by naming only the innermost level, and both stop costing an edge per level. """ frame = self._resumptions[-1] if frame.slot is None: frame.slot = self.synthetic_node() self.resumption_forward_edge(node, frame.slot) self.resumption_hub_edge(frame.hub, node) def close_handler_set(self, entries, *, escapes, outward=None)-
Finish a chain of handler entries: draw the edge from the last of them to what guards the construct when some run may leave all of them.
Called once the set's bodies are built and the set itself is off the handler stack, so that
unwindingnames what guards the construct rather than the set being closed.Expand source code Browse git
def close_handler_set( self, entries: Sequence[CfgNode], *, escapes: bool, outward: CfgNode | None = None, ) -> None: """ Finish a chain of handler entries: draw the edge from the last of them to what guards the construct when some run may leave all of them. Called once the set's bodies are built and the set itself is off the handler stack, so that `unwinding` names what guards the *construct* rather than the set being closed. """ if escapes and entries: self.exceptional_edge(entries[-1], self.unwinding() if outward is None else outward) def unwinding(self)-
Where a throw goes when nothing open here takes it: the innermost active handler, or the body's exit, which is how the graph says the throw leaves the body.
Expand source code Browse git
def unwinding(self) -> CfgNode: """ Where a throw goes when nothing open here takes it: the innermost active handler, or the body's exit, which is how the graph says the throw leaves the body. """ return self._handlers[-1] if self._handlers else self.cfg.exit def escape(self)-
Where an error goes once it gets past every handler of the construct being closed. That is
unwindingunless the language hands such an error to something that is not a handler.Expand source code Browse git
def escape(self) -> CfgNode: """ Where an error goes once it gets past every handler of the construct being closed. That is `unwinding` unless the language hands such an error to something that is not a handler. """ return self.unwinding() def handler_body_guard(self)-
What guards the handler bodies of a guarded block that has no finalizer, or
Nonewhere nothing beyond what guards the construct does. A language that hands an error leaving a handler body to something that is not a handler names it here.Expand source code Browse git
def handler_body_guard(self) -> CfgNode | None: """ What guards the handler bodies of a guarded block that has no finalizer, or `None` where nothing beyond what guards the construct does. A language that hands an error leaving a handler body to something that is not a handler names it here. """ return None def link(self, frontier, target)-
Expand source code Browse git
def link(self, frontier: Iterable[CfgNode], target: CfgNode) -> None: for node in frontier: self.add_edge(node, target) def sequence(self, statements, frontier)-
Expand source code Browse git
def sequence(self, statements: Sequence[Node], frontier: list[CfgNode]) -> list[CfgNode]: for statement in statements: frontier = self.member(statement, frontier) self.cfg._completions[id(statement)] = (statement, list(frontier)) return frontier def member(self, statement, frontier)-
Build statement as a member of a statement sequence and record the nodes it built, which
ControlFlowGraph.builder_of()answers from. A statement whose own node already exists was started by the construct around it — a declaration its block hoists, or a finalizer whose entry the guarded construct draws first — so the nodes built here are not all of its nodes, and none are recorded.Expand source code Browse git
def member(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ Build *statement* as a member of a statement sequence and record the nodes it built, which `ControlFlowGraph.builder_of` answers from. A statement whose own node already exists was started by the construct around it — a declaration its block hoists, or a finalizer whose entry the guarded construct draws first — so the nodes built here are not all of its nodes, and none are recorded. """ if self.cfg.node_of(statement) is not None: return self.statement(statement, frontier) start = len(self.cfg.nodes) frontier = self.statement(statement, frontier) self.cfg._spans[id(statement)] = (statement, range(start, len(self.cfg.nodes))) return frontier def opaque(self, element, frontier)-
An element whose internal control flow this does not model: one node, entered from the frontier and falling through. The default for anything a language does not recognise.
Expand source code Browse git
def opaque(self, element: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ An element whose internal control flow this does not model: one node, entered from the frontier and falling through. The default for anything a language does not recognise. """ node = self.node(element) self.link(frontier, node) return [node] def evaluated(self, element, frontier)-
The nodes evaluating element, a part of a construct evaluated at a point of its own, such as the initializer of a counted loop: one node, unless a language evaluates the part in steps it gives nodes of their own.
Expand source code Browse git
def evaluated(self, element: Node, frontier: list[CfgNode]) -> list[CfgNode]: """ The nodes evaluating *element*, a part of a construct evaluated at a point of its own, such as the initializer of a counted loop: one node, unless a language evaluates the part in steps it gives nodes of their own. """ return self.opaque(element, frontier) def park_label(self, *labels)-
Park labels for the construct about to be built, for
take_labelsto consume.labelledcalls this for a language whose label is a statement wrapping the construct. A language whose label is a field of the construct calls it directly, because there is no wrapping statement to recognise — and a construct built without it carries no label, so every jump naming it misses and leaves the body instead. A construct may carry several labels, asa: b: while (…)does, and a jump naming any of them targets it.Expand source code Browse git
def park_label(self, *labels: str | None) -> None: """ Park *labels* for the construct about to be built, for `take_labels` to consume. `labelled` calls this for a language whose label is a statement wrapping the construct. A language whose label is a field *of* the construct calls it directly, because there is no wrapping statement to recognise — and a construct built without it carries no label, so every jump naming it misses and leaves the body instead. A construct may carry several labels, as `a: b: while (…)` does, and a jump naming any of them targets it. """ self._pending_labels = frozenset(label for label in labels if label is not None) def take_labels(self)-
The labels parked for the construct about to be built, consumed once.
Expand source code Browse git
def take_labels(self) -> frozenset[str]: """ The labels parked for the construct about to be built, consumed once. """ labels = self._pending_labels self._pending_labels = frozenset() return labels def branch_on(self, element, arms, frontier, *, exhaustive=False)-
A conditional: one head node the frontier enters, and one arm per branch. When the arms do not cover every case — an
ifwith noelse— the head itself is an exit, because control may pass the whole construct without entering any arm.exhaustivesays the arms do cover it, which is what anif/elsepair reports.Expand source code Browse git
def branch_on( self, element: Node, arms: Sequence[Node | None], frontier: list[CfgNode], *, exhaustive: bool = False, ) -> list[CfgNode]: """ A conditional: one head node the frontier enters, and one arm per branch. When the arms do not cover every case — an `if` with no `else` — the head itself is an exit, because control may pass the whole construct without entering any arm. `exhaustive` says the arms do cover it, which is what an `if`/`else` pair reports. """ head = self.node(element) self.link(frontier, head) exits: list[CfgNode] = [] for arm in arms: exits += self._branch(arm, head) if not exhaustive: exits.append(head) return exits def loop_head_tested(self, element, body, frontier)-
A loop whose condition is evaluated before the body, so the head is both the entry and an exit:
while, and everyforeachwhose iteration may run zero times.Expand source code Browse git
def loop_head_tested( self, element: Node, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop whose condition is evaluated before the body, so the head is both the entry and an exit: `while`, and every `foreach` whose iteration may run zero times. """ head = self.node(element) self.link(frontier, head) target = _Target(self.take_labels(), [], head, is_continuable=True, is_breakable=True) self._open_target(target) body_exits = self._branch(body, head) self._targets.pop() self.link(body_exits, head) return [head] + target.breaks def loop_over(self, element, subject, target, body, frontier)-
A loop over the values a subject yields. The subject is evaluated once, at a node of its own before the loop. The head, which asks for the next value, is the entry of every iteration and an exit. The target receives the value at a node of its own on the way from the head into the body, so a run leaving the loop has not passed it. A language whose loop variable and subject are both read at the head builds the loop
loop_head_testedinstead.Expand source code Browse git
def loop_over( self, element: Node, subject: Node | None, target: Node | None, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop over the values a subject yields. The subject is evaluated once, at a node of its own before the loop. The head, which asks for the next value, is the entry of every iteration and an exit. The target receives the value at a node of its own on the way from the head into the body, so a run leaving the loop has not passed it. A language whose loop variable and subject are both read at the head builds the loop `loop_head_tested` instead. """ if subject is not None: start = self.node(subject) self.link(frontier, start) frontier = [start] head = self.node(element) self.link(frontier, head) loop = _Target(self.take_labels(), [], head, is_continuable=True, is_breakable=True) self._open_target(loop) entry = head if target is not None: entry = self.node(target) self.link([head], entry) body_exits = self._branch(body, entry) self._targets.pop() self.link(body_exits, head) return [head] + loop.breaks def loop_tail_tested(self, element, body, frontier)-
A loop whose condition is evaluated after the body, so the body always runs once and the back-edge targets the body's own entry rather than the test.
Expand source code Browse git
def loop_tail_tested( self, element: Node, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop whose condition is evaluated after the body, so the body always runs once and the back-edge targets the body's own entry rather than the test. """ test = self.node(element) target = _Target(self.take_labels(), [], test, is_continuable=True, is_breakable=True) self._open_target(target) entry, body_exits = self._capture_body(body, frontier) self._targets.pop() self.link(body_exits, test) self.add_edge(test, entry if entry is not None else test) return [test] + target.breaks def loop_counted(self, initializer, test, update, body, frontier)-
A loop with a separate initializer, test and update, each evaluated at its own point and so each given its own node. Any of the three may be absent; a loop with no test has no exit other than the jumps out of it, and its back-edge targets the body entry.
Expand source code Browse git
def loop_counted( self, initializer: Node | None, test: Node | None, update: Node | None, body: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A loop with a separate initializer, test and update, each evaluated at its own point and so each given its own node. Any of the three may be absent; a loop with no test has no exit other than the jumps out of it, and its back-edge targets the body entry. """ labels = self.take_labels() if initializer is not None: frontier = self.evaluated(initializer, frontier) head = self.node(test) if test is not None else None if head is not None: self.link(frontier, head) body_frontier: list[CfgNode] = [head] else: body_frontier = list(frontier) step = self.node(update) if update is not None else None target = _Target(labels, [], step or head, is_continuable=True, is_breakable=True) self._open_target(target) entry, body_exits = self._capture_body(body, body_frontier) self._targets.pop() latch = body_exits if step is not None: self.link(body_exits, step) latch = [step] back_to = head if head is not None else entry if back_to is not None: self.link(latch, back_to) self.link(target.continues, back_to if back_to is not None else self.cfg.exit) exits = list(target.breaks) if head is not None: exits.append(head) return exits def branch_chain(self, clauses, otherwise, frontier)-
A chain of guarded arms, each tested only when every earlier test failed:
if/elseif/elsewhere the whole chain is one node rather than a nest of two-armed conditionals.Each test gets its own node, because the tests run at different points and an analysis that collapsed them could not order two of them. A test's node flows into its own arm and on to the next test; the last test flows into
otherwisewhen there is one, and out of the construct when there is not.Expand source code Browse git
def branch_chain( self, clauses: Sequence[tuple[Node, Node | None]], otherwise: Node | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A chain of guarded arms, each tested only when every earlier test failed: `if`/`elseif`/`else` where the whole chain is one node rather than a nest of two-armed conditionals. Each test gets its own node, because the tests run at different points and an analysis that collapsed them could not order two of them. A test's node flows into its own arm and on to the next test; the last test flows into `otherwise` when there is one, and out of the construct when there is not. """ exits: list[CfgNode] = [] current = frontier for test, body in clauses: head = self.node(test) self.link(current, head) exits += self._branch(body, head) current = [head] if otherwise is not None: return distinct(exits + self._body(otherwise, current)) return distinct(exits + current) def dispatch(self, element, arms, frontier, *, arm_flow, exhaustive, iterated=False)-
A multi-way branch: one head the frontier enters and one arm per clause, each arm a statement sequence the head may jump into.
arm_flowsays what an arm reaches when it runs off its end; seeArmFlow, which is where the languages differ.exhaustivesays some arm always runs — a default clause — so the head is not itself an exit.iteratedsays the construct runs its arms once per element of an input, so it is a loop: a back-jump inside an arm re-enters the head rather than resolving to an enclosing loop, and an arm that simply runs off its end re-enters it too, for the next element. Without that second edge the arms lie on no cycle and an analysis reads a store inside one as happening once, which is what lets a self-referential assignment be folded to its first value. PowerShell'sswitchis this; JavaScript's is not, and reading one as the other both invents a back-edge to a loop the jump never reaches and drops the one it does.A jump out takes no back-edge: it leaves the construct rather than advancing it, which is the whole difference between the two spellings.
Expand source code Browse git
def dispatch( self, element: Node, arms: Sequence[Sequence[Node]], frontier: list[CfgNode], *, arm_flow: ArmFlow, exhaustive: bool, iterated: bool = False, ) -> list[CfgNode]: """ A multi-way branch: one head the frontier enters and one arm per clause, each arm a statement sequence the head may jump into. `arm_flow` says what an arm reaches when it runs off its end; see `ArmFlow`, which is where the languages differ. `exhaustive` says some arm always runs — a default clause — so the head is not itself an exit. `iterated` says the construct runs its arms once per element of an input, so it is a loop: a back-jump inside an arm re-enters the head rather than resolving to an enclosing loop, and an arm that simply runs off its end re-enters it too, for the next element. Without that second edge the arms lie on no cycle and an analysis reads a store inside one as happening once, which is what lets a self-referential assignment be folded to its first value. PowerShell's `switch` is this; JavaScript's is not, and reading one as the other both invents a back-edge to a loop the jump never reaches and drops the one it does. A jump *out* takes no back-edge: it leaves the construct rather than advancing it, which is the whole difference between the two spellings. """ head = self.node(element) self.link(frontier, head) target = _Target( self.take_labels(), [], head if iterated else None, is_continuable=iterated, is_breakable=True, ) self._open_target(target) carried: list[CfgNode] = [] completed: list[CfgNode] = [] for arm in arms: reached = self.sequence(list(arm), distinct([head, *carried])) if arm_flow is ArmFlow.SEQUENTIAL: carried = reached elif arm_flow is ArmFlow.CUMULATIVE: carried = distinct([*carried, *reached]) else: completed += reached self._targets.pop() completed = distinct(completed + carried) if iterated: self.link(completed, head) exits = completed + target.breaks if not exhaustive: exits.append(head) return distinct(exits) def guarded(self, block, handlers, finalizer, finalizer_body, frontier, *, escapes)-
A guarded block with any number of handlers and an optional finalizer.
The handler nodes are created before the guarded block is built and the first is pushed on the handler stack, so that
nodejoins every statement created inside the block to it. That ordering is the whole mechanism: it is why no statement inside the block has to know it is guarded.Several handlers are chained from the first, because which one runs depends on the type of the exception and none of them is guaranteed — a language with one handler passes a one-element sequence and the chain degenerates. The chain edge is exceptional: control takes it exactly when the earlier handler did not match, so nothing that handler's node stands for has run and its kill must not apply along it.
escapessays the set of handlers may fail to take a throw the block makes, which is what a language whose clauses carry a type filter reports and what one whose single clause takes everything denies. It decides one edge — from the last handler to whatever guards this construct, or to the body exit when nothing does, which is how the graph says a throw leaves the body with no handler having taken it. Denying the edge is the difference between a clause that swallows and one that passes the throw on, so it is asked of every language rather than defaulted: the safe answer invents a path the program cannot take, and the other drops one it can.The finalizer is entered from the block's normal exits and from every handler's, and runs on every other way out of the construct as well: an error no handler takes enters it, and so does a jump leaving the block or a handler body, and each goes on from its end, which
_leave_finalizerdraws. It also guards every handler body: a handler that throws, returns or jumps out of the construct runs the finalizer on its way, so where the block and every handler end that way, the finalizer is still reached.Expand source code Browse git
def guarded( self, block: Node | None, handlers: Sequence[tuple[Node, Node | None]], finalizer: Node | None, finalizer_body: Sequence[Node], frontier: list[CfgNode], *, escapes: bool, ) -> list[CfgNode]: """ A guarded block with any number of handlers and an optional finalizer. The handler nodes are created *before* the guarded block is built and the first is pushed on the handler stack, so that `node` joins every statement created inside the block to it. That ordering is the whole mechanism: it is why no statement inside the block has to know it is guarded. Several handlers are chained from the first, because which one runs depends on the type of the exception and none of them is guaranteed — a language with one handler passes a one-element sequence and the chain degenerates. The chain edge is *exceptional*: control takes it exactly when the earlier handler did not match, so nothing that handler's node stands for has run and its kill must not apply along it. `escapes` says the set of handlers may fail to take a throw the block makes, which is what a language whose clauses carry a type filter reports and what one whose single clause takes everything denies. It decides one edge — from the last handler to whatever guards this construct, or to the body exit when nothing does, which is how the graph says a throw leaves the body with no handler having taken it. Denying the edge is the difference between a clause that swallows and one that passes the throw on, so it is asked of every language rather than defaulted: the safe answer invents a path the program cannot take, and the other drops one it can. The finalizer is entered from the block's normal exits and from every handler's, and runs on every other way out of the construct as well: an error no handler takes enters it, and so does a jump leaving the block or a handler body, and each goes on from its end, which `_leave_finalizer` draws. It also guards every handler body: a handler that throws, returns or jumps out of the construct runs the finalizer on its way, so where the block and every handler end that way, the finalizer is still reached. """ handlers = list(handlers) entries = [self.detached_node(handler) for handler, _ in handlers] finalizer_entry = self.detached_node(finalizer) if finalizer is not None else None record = _Finalizer(finalizer_entry) if finalizer_entry is not None else None guard = entries[0] if entries else finalizer_entry if record is not None: self._finalizers.append(record) if guard is not None: self._handlers.append(guard) block_exits = self.statement(block, frontier) if block is not None else list(frontier) if guard is not None: self._handlers.pop() normal_exits = list(block_exits) clause_guard = finalizer_entry if clause_guard is None and entries: clause_guard = self.handler_body_guard() if clause_guard is not None: self._handlers.append(clause_guard) for index, ((_, body), entry) in enumerate(zip(handlers, entries)): if index: self.exceptional_edge(entries[index - 1], entry) normal_exits += ( self.statement(body, [entry]) if body is not None else [entry]) if clause_guard is not None: self._handlers.pop() if record is not None: self._finalizers.pop() outward = self.escape() if finalizer_entry is None else finalizer_entry self.close_handler_set(entries, escapes=escapes, outward=outward) if record is not None and finalizer_entry is not None: self.link(normal_exits, finalizer_entry) final_exits = self.sequence(list(finalizer_body), [finalizer_entry]) self._leave_finalizer(record, final_exits) return final_exits return normal_exits def labelled(self, labels, body, frontier, *, binds_to_body)-
A statement body with every label written in front of it, labels. When they name a construct a jump can target directly —
binds_to_body— they are parked for that construct to consume throughtake_labels; otherwise they name this statement itself and only a break can leave it.Expand source code Browse git
def labelled( self, labels: Sequence[str], body: Node | None, frontier: list[CfgNode], *, binds_to_body: bool, ) -> list[CfgNode]: """ A statement *body* with every label written in front of it, *labels*. When they name a construct a jump can target directly — `binds_to_body` — they are parked for that construct to consume through `take_labels`; otherwise they name this statement itself and only a break can leave it. """ if binds_to_body: self.park_label(*labels) return self.statement(body, frontier) if body is not None else list(frontier) target = _Target(frozenset(labels), [], None, is_continuable=False, is_breakable=False) self._open_target(target) exits = self.statement(body, frontier) if body is not None else list(frontier) self._targets.pop() return exits + target.breaks def jump_out(self, element, label, frontier)-
A jump that leaves the construct it names, or the innermost breakable one when unlabelled. A jump naming nothing this body holds leaves the body, which is the conservative reading.
Expand source code Browse git
def jump_out( self, element: Node, label: str | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A jump that leaves the construct it names, or the innermost breakable one when unlabelled. A jump naming nothing this body holds leaves the body, which is the conservative reading. """ node = self.node(element) self.link(frontier, node) self._leave(node, _Jump.OUT, self._break_target(label)) return [] def jump_back(self, element, label, frontier)-
A jump to the next iteration of the loop it names, or of the innermost loop when unlabelled.
Expand source code Browse git
def jump_back( self, element: Node, label: str | None, frontier: list[CfgNode], ) -> list[CfgNode]: """ A jump to the next iteration of the loop it names, or of the innermost loop when unlabelled. """ node = self.node(element) self.link(frontier, node) self._leave(node, _Jump.BACK, self._continue_target(label)) return [] def terminate(self, element, frontier, *, exceptional)-
A statement after which control does not continue in this body: a return, or a throw.
A throw is
exceptional, so it reaches the innermost open handler rather than the exit, and only reaches the exit when no handler is open. A return leaves the body through every finalizer open around it, as a jump out of the body does.Expand source code Browse git
def terminate( self, element: Node, frontier: list[CfgNode], *, exceptional: bool, ) -> list[CfgNode]: """ A statement after which control does not continue in this body: a return, or a throw. A throw is `exceptional`, so it reaches the innermost open handler rather than the exit, and only reaches the exit when no handler is open. A return leaves the body through every finalizer open around it, as a jump out of the body does. """ node = self.node(element) self.link(frontier, node) if exceptional: self.exceptional_edge(node, self.unwinding()) else: self._leave(node, _Jump.OUT, None) return [] def has_continue_target(self, label)-
Whether a back-jump naming label — or an unlabelled one — resolves to a construct currently being built. A language in which
continuemeans something other than a back-jump when no such construct is open asks this to tell the two spellings apart.Expand source code Browse git
def has_continue_target(self, label: str | None) -> bool: """ Whether a back-jump naming *label* — or an unlabelled one — resolves to a construct currently being built. A language in which `continue` means something other than a back-jump when no such construct is open asks this to tell the two spellings apart. """ return self._continue_target(label) is not None def has_break_target(self, label)-
Whether a jump out naming label — or an unlabelled one — resolves to a construct currently being built, the counterpart of
has_continue_targetfor the language in whichbreakmeans something other than leaving a construct when no construct is open to leave.Expand source code Browse git
def has_break_target(self, label: str | None) -> bool: """ Whether a jump out naming *label* — or an unlabelled one — resolves to a construct currently being built, the counterpart of `has_continue_target` for the language in which `break` means something other than leaving a construct when no construct is open to leave. """ return self._break_target(label) is not None
class ControlFlowModel (graphs, apart)-
The per-body control-flow graphs of one script, paired with the
ElementLocatorthat 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 theElementLocatorargument 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
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
Nonewhen 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
Nonewhen the graphs do not represent it directly. Delegates to the sharedElementLocator.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
Nonewhen it has no enclosing graph node. Delegates to the sharedElementLocator.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)