Skip to content

Sparse analysis

sparse_analysis

The sparse analysis module provides data flow analysis utilities for sparse lattices.

For more information on lattices, refer to this Wikipedia article.

AbstractLatticeValueInvT = TypeVar('AbstractLatticeValueInvT', bound=AbstractLatticeValue) module-attribute

PropagatingLatticeInvT = TypeVar('PropagatingLatticeInvT', bound=PropagatingLattice) module-attribute

AbstractLatticeValue

Bases: Protocol

Protocol for the mathematical lattice value types used within Lattice wrappers. A lattice is a mathematical structure with a partial ordering and two operations:

  • join (∨): computes the least upper bound (union of information)
  • meet (∧): computes the greatest lower bound (intersection of information)

Classes implementing this protocol should provide implementations for the meet and/or join methods. The class should also define the classmethod initial_value that takes no additional arguments and returns an initial lattice value.

This protocol represents the actual lattice element (the abstract value being tracked), separate from the propagation infrastructure. For example:

  • In constant propagation: the lattice value might be ⊥ (bottom) | Constant(n) | ⊤ (top)
  • In sign analysis: the lattice value might be Positive | Negative | Zero | Unknown
  • In range analysis: the lattice value might be Interval(min, max)
Source code in xdsl/analysis/sparse_analysis.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class AbstractLatticeValue(Protocol):
    """
    Protocol for the mathematical lattice value types used within Lattice wrappers.
    A lattice is a mathematical structure with a partial ordering and two operations:

    - join (∨): computes the least upper bound (union of information)
    - meet (∧): computes the greatest lower bound (intersection of information)

    Classes implementing this protocol should provide implementations for the `meet`
    and/or `join` methods. The class should also define the classmethod `initial_value`
    that takes no additional arguments and returns an initial lattice value.

    This protocol represents the actual lattice element (the abstract value being
    tracked), separate from the propagation infrastructure. For example:

    - In constant propagation: the lattice value might be `⊥ (bottom) | Constant(n) | ⊤ (top)`
    - In sign analysis: the lattice value might be `Positive | Negative | Zero | Unknown`
    - In range analysis: the lattice value might be `Interval(min, max)`
    """

    @classmethod
    def initial_value(cls) -> Self:
        """
        Returns an initial lattice value, typically the
        bottom (⊥) or uninitialized state of the lattice.
        """
        ...

    def meet(self, other: Self) -> Self:
        """
        Computes the greatest lower bound (intersection of information) of two lattice values.

        `a.meet(b)` (or `a ∧ b`) produces the most precise value that is less than or equal
        to both `a` and `b` in the lattice ordering. It represents the combination of two
        abstract values where we keep only information guaranteed to hold in both.

        In other words, `meet` refines information by taking their common part.

        Examples:

        - In constant propagation: `Constant(3) ∧ Constant(3) = Constant(3)`,
          but `Constant(3) ∧ Constant(4) = ⊥ (bottom)`
        - In sign analysis: `Positive ∧ NonZero = Positive`
        - In range analysis: `[0, 10] ∧ [5, 15] = [5, 10]`
        """
        ...

    def join(self, other: Self) -> Self:
        """
        Computes the least upper bound (union of information) of two lattice values.

        `a.join(b)` (or `a ∨ b`) produces the least precise value that is greater than or
        equal to both `a` and `b` in the lattice ordering. It represents the merging of
        two abstract values where we keep any information that could hold in either.

        In other words, `join` generalizes information by taking their union.

        Examples:

        - In constant propagation: `Constant(3) ∨ Constant(4) = ⊤ (top)`
        - In sign analysis: `Positive ∨ Negative = Unknown`
        - In range analysis: `[0, 10] ∨ [5, 15] = [0, 15]`
        """
        ...

initial_value() -> Self classmethod

Returns an initial lattice value, typically the bottom (⊥) or uninitialized state of the lattice.

Source code in xdsl/analysis/sparse_analysis.py 
46
47
48
49
50
51
52
@classmethod
def initial_value(cls) -> Self:
    """
    Returns an initial lattice value, typically the
    bottom (⊥) or uninitialized state of the lattice.
    """
    ...

meet(other: Self) -> Self

Computes the greatest lower bound (intersection of information) of two lattice values.

a.meet(b) (or a ∧ b) produces the most precise value that is less than or equal to both a and b in the lattice ordering. It represents the combination of two abstract values where we keep only information guaranteed to hold in both.

In other words, meet refines information by taking their common part.

Examples:

  • In constant propagation: Constant(3) ∧ Constant(3) = Constant(3), but Constant(3) ∧ Constant(4) = ⊥ (bottom)
  • In sign analysis: Positive ∧ NonZero = Positive
  • In range analysis: [0, 10] ∧ [5, 15] = [5, 10]
Source code in xdsl/analysis/sparse_analysis.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def meet(self, other: Self) -> Self:
    """
    Computes the greatest lower bound (intersection of information) of two lattice values.

    `a.meet(b)` (or `a ∧ b`) produces the most precise value that is less than or equal
    to both `a` and `b` in the lattice ordering. It represents the combination of two
    abstract values where we keep only information guaranteed to hold in both.

    In other words, `meet` refines information by taking their common part.

    Examples:

    - In constant propagation: `Constant(3) ∧ Constant(3) = Constant(3)`,
      but `Constant(3) ∧ Constant(4) = ⊥ (bottom)`
    - In sign analysis: `Positive ∧ NonZero = Positive`
    - In range analysis: `[0, 10] ∧ [5, 15] = [5, 10]`
    """
    ...

join(other: Self) -> Self

Computes the least upper bound (union of information) of two lattice values.

a.join(b) (or a ∨ b) produces the least precise value that is greater than or equal to both a and b in the lattice ordering. It represents the merging of two abstract values where we keep any information that could hold in either.

In other words, join generalizes information by taking their union.

Examples:

  • In constant propagation: Constant(3) ∨ Constant(4) = ⊤ (top)
  • In sign analysis: Positive ∨ Negative = Unknown
  • In range analysis: [0, 10] ∨ [5, 15] = [0, 15]
Source code in xdsl/analysis/sparse_analysis.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def join(self, other: Self) -> Self:
    """
    Computes the least upper bound (union of information) of two lattice values.

    `a.join(b)` (or `a ∨ b`) produces the least precise value that is greater than or
    equal to both `a` and `b` in the lattice ordering. It represents the merging of
    two abstract values where we keep any information that could hold in either.

    In other words, `join` generalizes information by taking their union.

    Examples:

    - In constant propagation: `Constant(3) ∨ Constant(4) = ⊤ (top)`
    - In sign analysis: `Positive ∨ Negative = Unknown`
    - In range analysis: `[0, 10] ∨ [5, 15] = [0, 15]`
    """
    ...

AbstractSparseLattice

Bases: Protocol

Protocol for sparse lattice elements used in data flow analysis.

See AbstractLatticeValue for more information about lattices and their operations.

In contrast to AbstractLatticeValue, the meet and join methods in this protocol are required to return a ChangeResult, signaling whether the lattice element has changed.

Source code in xdsl/analysis/sparse_analysis.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class AbstractSparseLattice(Protocol):
    """
    Protocol for sparse lattice elements used in data flow analysis.

    See [AbstractLatticeValue][xdsl.analysis.sparse_analysis.AbstractLatticeValue] for more
    information about lattices and their operations.

    In contrast to [AbstractLatticeValue][xdsl.analysis.sparse_analysis.AbstractLatticeValue],
    the `meet` and `join` methods in this protocol are required to return a ChangeResult,
    signaling whether the lattice element has changed.
    """

    def join(self, other: Self) -> ChangeResult:
        """
        Join two lattice elements. Returns `ChangeResult.CHANGE` if the lattice element has changed,
        otherwise `ChangeResult.NO_CHANGE`.

        For more information about the join operation, see
        [`AbstractLatticeValue.join`][xdsl.analysis.sparse_analysis.AbstractLatticeValue.join].
        """
        ...

    def meet(self, other: Self) -> ChangeResult:
        """
        Meet two lattice elements. Returns `ChangeResult.CHANGE` if the lattice element has changed,
        otherwise `ChangeResult.NO_CHANGE`.

        For more information about the meet operation, see
        [`AbstractLatticeValue.meet`][xdsl.analysis.sparse_analysis.AbstractLatticeValue.meet].
        """
        ...

join(other: Self) -> ChangeResult

Join two lattice elements. Returns ChangeResult.CHANGE if the lattice element has changed, otherwise ChangeResult.NO_CHANGE.

For more information about the join operation, see AbstractLatticeValue.join.

Source code in xdsl/analysis/sparse_analysis.py 
104
105
106
107
108
109
110
111
112
def join(self, other: Self) -> ChangeResult:
    """
    Join two lattice elements. Returns `ChangeResult.CHANGE` if the lattice element has changed,
    otherwise `ChangeResult.NO_CHANGE`.

    For more information about the join operation, see
    [`AbstractLatticeValue.join`][xdsl.analysis.sparse_analysis.AbstractLatticeValue.join].
    """
    ...

meet(other: Self) -> ChangeResult

Meet two lattice elements. Returns ChangeResult.CHANGE if the lattice element has changed, otherwise ChangeResult.NO_CHANGE.

For more information about the meet operation, see AbstractLatticeValue.meet.

Source code in xdsl/analysis/sparse_analysis.py 
114
115
116
117
118
119
120
121
122
def meet(self, other: Self) -> ChangeResult:
    """
    Meet two lattice elements. Returns `ChangeResult.CHANGE` if the lattice element has changed,
    otherwise `ChangeResult.NO_CHANGE`.

    For more information about the meet operation, see
    [`AbstractLatticeValue.meet`][xdsl.analysis.sparse_analysis.AbstractLatticeValue.meet].
    """
    ...

PropagatingLattice

Bases: AnalysisState[SSAValue], AbstractSparseLattice, ABC

Base class for sparse lattice elements attached to SSA values.

This class implements the infrastructure for propagating lattice changes through the data flow analysis framework. When a lattice element changes (e.g., a value becomes a known constant), this class ensures that:

  1. All operations that use this SSA value are re-analyzed
  2. Subscribed analyses are notified of the change
  3. The solver's work queue is updated appropriately

The propagation follows use-def chains: when a lattice attached to an SSA value changes, all operations that use that value are marked for re-visiting by any analyses that have subscribed to this lattice.

Subclasses must implement the lattice operations (join/meet) and can override on_update() to customize propagation behavior beyond simple use-def chains.

For the concept of lattices in data flow analysis, see PropagatingLattice.

Use this as a base class when you need custom propagation logic (e.g., tracking equivalence classes, pointer aliases, or context-sensitive information). For simple cases, use the Lattice wrapper instead.

Source code in xdsl/analysis/sparse_analysis.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class PropagatingLattice(AnalysisState[SSAValue], AbstractSparseLattice, ABC):
    """
    Base class for sparse lattice elements attached to SSA values.

    This class implements the infrastructure for propagating lattice changes through
    the data flow analysis framework. When a lattice element changes (e.g., a value
    becomes a known constant), this class ensures that:

    1. All operations that use this SSA value are re-analyzed
    2. Subscribed analyses are notified of the change
    3. The solver's work queue is updated appropriately

    The propagation follows use-def chains: when a lattice attached to an SSA value
    changes, all operations that use that value are marked for re-visiting by any
    analyses that have subscribed to this lattice.

    Subclasses must implement the lattice operations (join/meet) and can override
    on_update() to customize propagation behavior beyond simple use-def chains.

    For the concept of lattices in data flow analysis, see
    [`PropagatingLattice`][xdsl.analysis.sparse_analysis.PropagatingLattice].

    Use this as a base class when you need custom propagation logic (e.g., tracking
    equivalence classes, pointer aliases, or context-sensitive information). For
    simple cases, use the Lattice wrapper instead.
    """

    use_def_subscribers: set[DataFlowAnalysis]

    def __init__(self, anchor: SSAValue):
        super().__init__(anchor)
        self.use_def_subscribers = set()

    def on_update(self, solver: DataFlowSolver) -> None:
        """
        When a sparse lattice changes, we propagate the change to explicit
        dependents and also to all users of the underlying SSA value for
        subscribed analyses.
        """
        super().on_update(solver)

        for use in self.anchor.uses:
            user_op = use.operation
            user_point = ProgramPoint.before(user_op)
            for analysis in self.use_def_subscribers:
                solver.enqueue((user_point, analysis))

    def use_def_subscribe(self, analysis: DataFlowAnalysis) -> None:
        """
        Subscribe an analysis to be re-invoked on all users of this value
        whenever this lattice state changes.
        """
        self.use_def_subscribers.add(analysis)

use_def_subscribers: set[DataFlowAnalysis] = set() instance-attribute

__init__(anchor: SSAValue)

Source code in xdsl/analysis/sparse_analysis.py 
154
155
156
def __init__(self, anchor: SSAValue):
    super().__init__(anchor)
    self.use_def_subscribers = set()

on_update(solver: DataFlowSolver) -> None

When a sparse lattice changes, we propagate the change to explicit dependents and also to all users of the underlying SSA value for subscribed analyses.

Source code in xdsl/analysis/sparse_analysis.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
def on_update(self, solver: DataFlowSolver) -> None:
    """
    When a sparse lattice changes, we propagate the change to explicit
    dependents and also to all users of the underlying SSA value for
    subscribed analyses.
    """
    super().on_update(solver)

    for use in self.anchor.uses:
        user_op = use.operation
        user_point = ProgramPoint.before(user_op)
        for analysis in self.use_def_subscribers:
            solver.enqueue((user_point, analysis))

use_def_subscribe(analysis: DataFlowAnalysis) -> None

Subscribe an analysis to be re-invoked on all users of this value whenever this lattice state changes.

Source code in xdsl/analysis/sparse_analysis.py 
172
173
174
175
176
177
def use_def_subscribe(self, analysis: DataFlowAnalysis) -> None:
    """
    Subscribe an analysis to be re-invoked on all users of this value
    whenever this lattice state changes.
    """
    self.use_def_subscribers.add(analysis)

Lattice

Bases: PropagatingLattice, Generic[AbstractLatticeValueInvT]

Generic wrapper that combines a lattice value with sparse propagation infrastructure.

See AbstractLatticeValue for more information around the meet and join operations that need to be implemented.

If you need meet/join functionality that does not match with what Lattice provides, consider using PropagatingLattice directly.

Source code in xdsl/analysis/sparse_analysis.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class Lattice(PropagatingLattice, Generic[AbstractLatticeValueInvT]):
    """
    Generic wrapper that combines a lattice value with sparse propagation infrastructure.

    See [`AbstractLatticeValue`][xdsl.analysis.sparse_analysis.AbstractLatticeValue] for more information around
    the meet and join operations that need to be implemented.

    If you need meet/join functionality that does not match with what `Lattice` provides,
    consider using [`PropagatingLattice`][xdsl.analysis.sparse_analysis.PropagatingLattice] directly.
    """

    value_cls: type[AbstractLatticeValueInvT]
    _value: AbstractLatticeValueInvT

    def __init__(
        self,
        anchor: SSAValue,
        value: AbstractLatticeValueInvT | None = None,
    ):
        super().__init__(anchor)
        if value is not None:
            self._value = value
        else:
            self._value = self.value_cls.initial_value()

    @property
    def value(self) -> AbstractLatticeValueInvT:
        return self._value

    def meet(self, other: Self) -> ChangeResult:
        new_value = self.value.meet(other.value)

        if new_value == self.value:
            return ChangeResult.NO_CHANGE

        self._value = new_value
        return ChangeResult.CHANGE

    def join(self, other: Self) -> ChangeResult:
        new_value = self.value.join(other.value)

        if new_value == self.value:
            return ChangeResult.NO_CHANGE

        self._value = new_value
        return ChangeResult.CHANGE

    def __str__(self) -> str:
        return str(self.value)

value_cls: type[AbstractLatticeValueInvT] instance-attribute

value: AbstractLatticeValueInvT property

__init__(anchor: SSAValue, value: AbstractLatticeValueInvT | None = None)

Source code in xdsl/analysis/sparse_analysis.py 
199
200
201
202
203
204
205
206
207
208
def __init__(
    self,
    anchor: SSAValue,
    value: AbstractLatticeValueInvT | None = None,
):
    super().__init__(anchor)
    if value is not None:
        self._value = value
    else:
        self._value = self.value_cls.initial_value()

meet(other: Self) -> ChangeResult

Source code in xdsl/analysis/sparse_analysis.py 
214
215
216
217
218
219
220
221
def meet(self, other: Self) -> ChangeResult:
    new_value = self.value.meet(other.value)

    if new_value == self.value:
        return ChangeResult.NO_CHANGE

    self._value = new_value
    return ChangeResult.CHANGE

join(other: Self) -> ChangeResult

Source code in xdsl/analysis/sparse_analysis.py 
223
224
225
226
227
228
229
230
def join(self, other: Self) -> ChangeResult:
    new_value = self.value.join(other.value)

    if new_value == self.value:
        return ChangeResult.NO_CHANGE

    self._value = new_value
    return ChangeResult.CHANGE

__str__() -> str

Source code in xdsl/analysis/sparse_analysis.py 
232
233
def __str__(self) -> str:
    return str(self.value)

SparseForwardDataFlowAnalysis

Bases: DataFlowAnalysis, ABC, Generic[PropagatingLatticeInvT]

Base class for sparse forward data-flow analyses. It propagates lattices attached to SSA values along the direction of data flow.

Source code in xdsl/analysis/sparse_analysis.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
class SparseForwardDataFlowAnalysis(
    DataFlowAnalysis, ABC, Generic[PropagatingLatticeInvT]
):
    """
    Base class for sparse forward data-flow analyses. It propagates lattices
    attached to SSA values along the direction of data flow.
    """

    def __init__(
        self, solver: DataFlowSolver, lattice_type: type[PropagatingLatticeInvT]
    ):
        super().__init__(solver)
        self.lattice_type = lattice_type

    def initialize(self, op: Operation) -> None:
        # Set entry state for all arguments of the top-level regions.
        for region in op.regions:
            if region.first_block is not None:
                for arg in region.first_block.args:
                    self.set_to_entry_state(self.get_lattice_element(arg))

        # Iteratively visit all ops to build initial dependencies.
        stack = [op]

        while stack:
            current_op = stack.pop()

            self.visit(ProgramPoint.before(current_op))

            for region in current_op.regions:
                for block in region.blocks:
                    block_start_point = ProgramPoint.at_start_of_block(block)
                    executable = self.get_or_create_state(block_start_point, Executable)
                    executable.block_content_subscribers.add(self)
                    self.visit(block_start_point)

                    # Add nested ops to stack in reverse order to maintain traversal order
                    stack.extend(reversed(block.ops))

    def visit(self, point: ProgramPoint) -> None:
        if point.op is not None:
            self.visit_operation(point.op)
        elif point.block is not None:
            # This case handles end-of-block points, which for forward analysis
            # means we are visiting the block itself to handle its arguments.
            self.visit_block(point.block)

    def visit_operation(self, op: Operation) -> None:
        """Transfer function for an operation's results."""
        if not op.results:
            return

        # If the parent block is not executable, do nothing.
        if (
            op.parent is not None
            and not self.get_or_create_state(
                ProgramPoint.at_start_of_block(op.parent), Executable
            ).live
        ):
            return

        # Get operand and result lattices
        point = ProgramPoint.before(op)
        operands = [self.get_lattice_element_for(point, o) for o in op.operands]
        results = [self.get_lattice_element(r) for r in op.results]

        # Subscribe to operand changes for future updates.
        for o in op.operands:
            self.get_lattice_element(o).use_def_subscribe(self)

        # Check if operation requires special interface support
        if op.regions:
            raise NotImplementedError(
                f"Operation {op.name} has regions. Full support requires "
                "RegionBranchOpInterface to properly handle control flow "
                "between regions."
            )

        # TODO: handle call operations when CallOpInterface is implemented

        self.visit_operation_impl(op, operands, results)

    def visit_block(self, block: Block) -> None:
        """Transfer function for a block's arguments."""
        if not block.args:
            return

        point = ProgramPoint.at_start_of_block(block)
        if not self.get_or_create_state(point, Executable).live:
            return

        # For non-entry blocks, join values from predecessors.
        if block.parent is None or block.parent.first_block is not block:
            for pred_block in block.predecessors():
                edge = CFGEdge(pred_block, block)
                executable = self.get_state(edge, Executable)
                if not executable or not executable.live:
                    continue

                terminator = pred_block.last_op
                if terminator is None:
                    continue

                # Requires BranchOpInterface to correctly map terminator operands
                # to successor block arguments
                raise NotImplementedError(
                    f"Mapping values across control flow edges requires "
                    "BranchOpInterface. Cannot determine which operands of "
                    f"terminator {terminator.name} correspond to arguments of "
                    f"successor block."
                )
        # else:  # For entry blocks of regions
        # TODO: Requires CallableOpInterface and RegionBranchOpInterface

    def join(self, lhs: PropagatingLatticeInvT, rhs: PropagatingLatticeInvT) -> None:
        """Joins the rhs lattice into the lhs and propagates if changed."""
        self.propagate_if_changed(lhs, lhs.join(rhs))

    def get_lattice_element(self, value: SSAValue) -> PropagatingLatticeInvT:
        return self.get_or_create_state(value, self.lattice_type)

    def get_lattice_element_for(
        self, point: ProgramPoint, value: SSAValue
    ) -> PropagatingLatticeInvT:
        lattice = self.get_lattice_element(value)
        self.add_dependency(lattice, point)
        return lattice

    def set_all_to_entry_state(self, lattices: list[PropagatingLatticeInvT]) -> None:
        for lattice in lattices:
            self.set_to_entry_state(lattice)

    @abstractmethod
    def visit_operation_impl(
        self,
        op: Operation,
        operands: list[PropagatingLatticeInvT],
        results: list[PropagatingLatticeInvT],
    ) -> None:
        """The user-defined transfer function for a generic operation."""
        ...

    @abstractmethod
    def set_to_entry_state(self, lattice: PropagatingLatticeInvT) -> None:
        """Sets a lattice to its most pessimistic (entry) state."""
        ...

lattice_type = lattice_type instance-attribute

__init__(solver: DataFlowSolver, lattice_type: type[PropagatingLatticeInvT])

Source code in xdsl/analysis/sparse_analysis.py 
247
248
249
250
251
def __init__(
    self, solver: DataFlowSolver, lattice_type: type[PropagatingLatticeInvT]
):
    super().__init__(solver)
    self.lattice_type = lattice_type

initialize(op: Operation) -> None

Source code in xdsl/analysis/sparse_analysis.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def initialize(self, op: Operation) -> None:
    # Set entry state for all arguments of the top-level regions.
    for region in op.regions:
        if region.first_block is not None:
            for arg in region.first_block.args:
                self.set_to_entry_state(self.get_lattice_element(arg))

    # Iteratively visit all ops to build initial dependencies.
    stack = [op]

    while stack:
        current_op = stack.pop()

        self.visit(ProgramPoint.before(current_op))

        for region in current_op.regions:
            for block in region.blocks:
                block_start_point = ProgramPoint.at_start_of_block(block)
                executable = self.get_or_create_state(block_start_point, Executable)
                executable.block_content_subscribers.add(self)
                self.visit(block_start_point)

                # Add nested ops to stack in reverse order to maintain traversal order
                stack.extend(reversed(block.ops))

visit(point: ProgramPoint) -> None

Source code in xdsl/analysis/sparse_analysis.py 
278
279
280
281
282
283
284
def visit(self, point: ProgramPoint) -> None:
    if point.op is not None:
        self.visit_operation(point.op)
    elif point.block is not None:
        # This case handles end-of-block points, which for forward analysis
        # means we are visiting the block itself to handle its arguments.
        self.visit_block(point.block)

visit_operation(op: Operation) -> None

Transfer function for an operation's results.

Source code in xdsl/analysis/sparse_analysis.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def visit_operation(self, op: Operation) -> None:
    """Transfer function for an operation's results."""
    if not op.results:
        return

    # If the parent block is not executable, do nothing.
    if (
        op.parent is not None
        and not self.get_or_create_state(
            ProgramPoint.at_start_of_block(op.parent), Executable
        ).live
    ):
        return

    # Get operand and result lattices
    point = ProgramPoint.before(op)
    operands = [self.get_lattice_element_for(point, o) for o in op.operands]
    results = [self.get_lattice_element(r) for r in op.results]

    # Subscribe to operand changes for future updates.
    for o in op.operands:
        self.get_lattice_element(o).use_def_subscribe(self)

    # Check if operation requires special interface support
    if op.regions:
        raise NotImplementedError(
            f"Operation {op.name} has regions. Full support requires "
            "RegionBranchOpInterface to properly handle control flow "
            "between regions."
        )

    # TODO: handle call operations when CallOpInterface is implemented

    self.visit_operation_impl(op, operands, results)

visit_block(block: Block) -> None

Transfer function for a block's arguments.

Source code in xdsl/analysis/sparse_analysis.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def visit_block(self, block: Block) -> None:
    """Transfer function for a block's arguments."""
    if not block.args:
        return

    point = ProgramPoint.at_start_of_block(block)
    if not self.get_or_create_state(point, Executable).live:
        return

    # For non-entry blocks, join values from predecessors.
    if block.parent is None or block.parent.first_block is not block:
        for pred_block in block.predecessors():
            edge = CFGEdge(pred_block, block)
            executable = self.get_state(edge, Executable)
            if not executable or not executable.live:
                continue

            terminator = pred_block.last_op
            if terminator is None:
                continue

            # Requires BranchOpInterface to correctly map terminator operands
            # to successor block arguments
            raise NotImplementedError(
                f"Mapping values across control flow edges requires "
                "BranchOpInterface. Cannot determine which operands of "
                f"terminator {terminator.name} correspond to arguments of "
                f"successor block."
            )

join(lhs: PropagatingLatticeInvT, rhs: PropagatingLatticeInvT) -> None

Joins the rhs lattice into the lhs and propagates if changed.

Source code in xdsl/analysis/sparse_analysis.py 
353
354
355
def join(self, lhs: PropagatingLatticeInvT, rhs: PropagatingLatticeInvT) -> None:
    """Joins the rhs lattice into the lhs and propagates if changed."""
    self.propagate_if_changed(lhs, lhs.join(rhs))

get_lattice_element(value: SSAValue) -> PropagatingLatticeInvT

Source code in xdsl/analysis/sparse_analysis.py 
357
358
def get_lattice_element(self, value: SSAValue) -> PropagatingLatticeInvT:
    return self.get_or_create_state(value, self.lattice_type)

get_lattice_element_for(point: ProgramPoint, value: SSAValue) -> PropagatingLatticeInvT

Source code in xdsl/analysis/sparse_analysis.py 
360
361
362
363
364
365
def get_lattice_element_for(
    self, point: ProgramPoint, value: SSAValue
) -> PropagatingLatticeInvT:
    lattice = self.get_lattice_element(value)
    self.add_dependency(lattice, point)
    return lattice

set_all_to_entry_state(lattices: list[PropagatingLatticeInvT]) -> None

Source code in xdsl/analysis/sparse_analysis.py 
367
368
369
def set_all_to_entry_state(self, lattices: list[PropagatingLatticeInvT]) -> None:
    for lattice in lattices:
        self.set_to_entry_state(lattice)

visit_operation_impl(op: Operation, operands: list[PropagatingLatticeInvT], results: list[PropagatingLatticeInvT]) -> None abstractmethod

The user-defined transfer function for a generic operation.

Source code in xdsl/analysis/sparse_analysis.py 
371
372
373
374
375
376
377
378
379
@abstractmethod
def visit_operation_impl(
    self,
    op: Operation,
    operands: list[PropagatingLatticeInvT],
    results: list[PropagatingLatticeInvT],
) -> None:
    """The user-defined transfer function for a generic operation."""
    ...

set_to_entry_state(lattice: PropagatingLatticeInvT) -> None abstractmethod

Sets a lattice to its most pessimistic (entry) state.

Source code in xdsl/analysis/sparse_analysis.py 
381
382
383
384
@abstractmethod
def set_to_entry_state(self, lattice: PropagatingLatticeInvT) -> None:
    """Sets a lattice to its most pessimistic (entry) state."""
    ...

SparseBackwardDataFlowAnalysis

Bases: DataFlowAnalysis, ABC, Generic[PropagatingLatticeInvT]

Base class for sparse backward data-flow analyses. It propagates lattices attached to SSA values against the direction of data flow: information placed on an operation's results is meet-combined into its operands.

Source code in xdsl/analysis/sparse_analysis.py 
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
class SparseBackwardDataFlowAnalysis(
    DataFlowAnalysis, ABC, Generic[PropagatingLatticeInvT]
):
    """
    Base class for sparse backward data-flow analyses. It propagates lattices
    attached to SSA values *against* the direction of data flow: information
    placed on an operation's results is meet-combined into its operands.
    """

    def __init__(
        self, solver: DataFlowSolver, lattice_type: type[PropagatingLatticeInvT]
    ):
        super().__init__(solver)
        self.lattice_type = lattice_type

    def initialize(self, op: Operation) -> None:
        # Iteratively visit all ops to build initial dependencies.
        stack: list[Operation] = [op]

        while stack:
            current_op = stack.pop()

            self.visit(ProgramPoint.before(current_op))

            for region in current_op.regions:
                for block in region.blocks:
                    block_start_point = ProgramPoint.at_start_of_block(block)
                    executable = self.get_or_create_state(block_start_point, Executable)
                    executable.block_content_subscribers.add(self)
                    # Push children in forward order so they are popped in
                    # reverse — yielding the backward walk within the block.
                    stack.extend(block.ops)

    def visit(self, point: ProgramPoint) -> None:
        if point.op is not None:
            self.visit_operation(point.op)
        # Block-start points are no-ops for backward analysis: block arguments
        # receive their lattice information from terminator operands (handled
        # via BranchOpInterface / RegionBranchOpInterface), not from
        # the block itself.

    def visit_operation(self, op: Operation) -> None:
        """Transfer function reading result lattices, writing operand lattices."""
        # Nothing to propagate backward into if the op has no operands.
        if not op.operands:
            return

        # If the parent block is not executable, do nothing.
        if (
            op.parent is not None
            and not self.get_or_create_state(
                ProgramPoint.at_start_of_block(op.parent), Executable
            ).live
        ):
            return

        # Gather lattices. Result lattices register this op's program point as
        # a dependent, so that when an upstream backward analysis writes to a
        # result, this op is re-enqueued to push the info into its operands.
        point = ProgramPoint.before(op)
        operand_lattices = [self.get_lattice_element(o) for o in op.operands]
        result_lattices = [self.get_lattice_element_for(point, r) for r in op.results]

        if op.regions:
            raise NotImplementedError(
                f"Operation {op.name} has regions. Backward propagation across "
                "region boundaries requires RegionBranchOpInterface and "
                "RegionBranchTerminatorOpInterface."
            )

        if op.successors:
            raise NotImplementedError(
                f"Operation {op.name} has block successors. Mapping operands to "
                "successor block arguments for backward propagation requires "
                "BranchOpInterface."
            )

        # TODO: dispatch CallOpInterface once a SymbolTableCollection equivalent
        #       is available — backward analysis must connect call-site operands
        #       to callee entry-block arguments, and callee return-op operands
        #       back to call-site results.
        # TODO: dispatch return-like ops in callable regions once
        #       CallableOpInterface support lands.

        self.visit_operation_impl(op, operand_lattices, result_lattices)

    def meet(self, lhs: PropagatingLatticeInvT, rhs: PropagatingLatticeInvT) -> None:
        """Meets the rhs lattice into the lhs and propagates if changed."""
        self.propagate_if_changed(lhs, lhs.meet(rhs))

    def get_lattice_element(self, value: SSAValue) -> PropagatingLatticeInvT:
        return self.get_or_create_state(value, self.lattice_type)

    def get_lattice_element_for(
        self, point: ProgramPoint, value: SSAValue
    ) -> PropagatingLatticeInvT:
        lattice = self.get_lattice_element(value)
        self.add_dependency(lattice, point)
        return lattice

    def set_all_to_exit_states(self, lattices: list[PropagatingLatticeInvT]) -> None:
        for lattice in lattices:
            self.set_to_exit_state(lattice)

    @abstractmethod
    def visit_operation_impl(
        self,
        op: Operation,
        operand_lattices: list[PropagatingLatticeInvT],
        result_lattices: list[PropagatingLatticeInvT],
    ) -> None:
        """
        User-defined transfer function. Should read `result_lattices` and use
        `meet` to push information into `operand_lattices`.
        """

    @abstractmethod
    def set_to_exit_state(self, lattice: PropagatingLatticeInvT) -> None:
        """
        Sets a lattice to its most pessimistic (exit) state. Called for values
        whose backward propagation reaches an external boundary, e.g. operands
        of public function returns or arguments of public function entry
        blocks — for those, the analysis cannot assume any structure and must
        conservatively widen.
        """

lattice_type = lattice_type instance-attribute

__init__(solver: DataFlowSolver, lattice_type: type[PropagatingLatticeInvT])

Source code in xdsl/analysis/sparse_analysis.py 
396
397
398
399
400
def __init__(
    self, solver: DataFlowSolver, lattice_type: type[PropagatingLatticeInvT]
):
    super().__init__(solver)
    self.lattice_type = lattice_type

initialize(op: Operation) -> None

Source code in xdsl/analysis/sparse_analysis.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
def initialize(self, op: Operation) -> None:
    # Iteratively visit all ops to build initial dependencies.
    stack: list[Operation] = [op]

    while stack:
        current_op = stack.pop()

        self.visit(ProgramPoint.before(current_op))

        for region in current_op.regions:
            for block in region.blocks:
                block_start_point = ProgramPoint.at_start_of_block(block)
                executable = self.get_or_create_state(block_start_point, Executable)
                executable.block_content_subscribers.add(self)
                # Push children in forward order so they are popped in
                # reverse — yielding the backward walk within the block.
                stack.extend(block.ops)

visit(point: ProgramPoint) -> None

Source code in xdsl/analysis/sparse_analysis.py 
420
421
422
def visit(self, point: ProgramPoint) -> None:
    if point.op is not None:
        self.visit_operation(point.op)

visit_operation(op: Operation) -> None

Transfer function reading result lattices, writing operand lattices.

Source code in xdsl/analysis/sparse_analysis.py 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
def visit_operation(self, op: Operation) -> None:
    """Transfer function reading result lattices, writing operand lattices."""
    # Nothing to propagate backward into if the op has no operands.
    if not op.operands:
        return

    # If the parent block is not executable, do nothing.
    if (
        op.parent is not None
        and not self.get_or_create_state(
            ProgramPoint.at_start_of_block(op.parent), Executable
        ).live
    ):
        return

    # Gather lattices. Result lattices register this op's program point as
    # a dependent, so that when an upstream backward analysis writes to a
    # result, this op is re-enqueued to push the info into its operands.
    point = ProgramPoint.before(op)
    operand_lattices = [self.get_lattice_element(o) for o in op.operands]
    result_lattices = [self.get_lattice_element_for(point, r) for r in op.results]

    if op.regions:
        raise NotImplementedError(
            f"Operation {op.name} has regions. Backward propagation across "
            "region boundaries requires RegionBranchOpInterface and "
            "RegionBranchTerminatorOpInterface."
        )

    if op.successors:
        raise NotImplementedError(
            f"Operation {op.name} has block successors. Mapping operands to "
            "successor block arguments for backward propagation requires "
            "BranchOpInterface."
        )

    # TODO: dispatch CallOpInterface once a SymbolTableCollection equivalent
    #       is available — backward analysis must connect call-site operands
    #       to callee entry-block arguments, and callee return-op operands
    #       back to call-site results.
    # TODO: dispatch return-like ops in callable regions once
    #       CallableOpInterface support lands.

    self.visit_operation_impl(op, operand_lattices, result_lattices)

meet(lhs: PropagatingLatticeInvT, rhs: PropagatingLatticeInvT) -> None

Meets the rhs lattice into the lhs and propagates if changed.

Source code in xdsl/analysis/sparse_analysis.py 
473
474
475
def meet(self, lhs: PropagatingLatticeInvT, rhs: PropagatingLatticeInvT) -> None:
    """Meets the rhs lattice into the lhs and propagates if changed."""
    self.propagate_if_changed(lhs, lhs.meet(rhs))

get_lattice_element(value: SSAValue) -> PropagatingLatticeInvT

Source code in xdsl/analysis/sparse_analysis.py 
477
478
def get_lattice_element(self, value: SSAValue) -> PropagatingLatticeInvT:
    return self.get_or_create_state(value, self.lattice_type)

get_lattice_element_for(point: ProgramPoint, value: SSAValue) -> PropagatingLatticeInvT

Source code in xdsl/analysis/sparse_analysis.py 
480
481
482
483
484
485
def get_lattice_element_for(
    self, point: ProgramPoint, value: SSAValue
) -> PropagatingLatticeInvT:
    lattice = self.get_lattice_element(value)
    self.add_dependency(lattice, point)
    return lattice

set_all_to_exit_states(lattices: list[PropagatingLatticeInvT]) -> None

Source code in xdsl/analysis/sparse_analysis.py 
487
488
489
def set_all_to_exit_states(self, lattices: list[PropagatingLatticeInvT]) -> None:
    for lattice in lattices:
        self.set_to_exit_state(lattice)

visit_operation_impl(op: Operation, operand_lattices: list[PropagatingLatticeInvT], result_lattices: list[PropagatingLatticeInvT]) -> None abstractmethod

User-defined transfer function. Should read result_lattices and use meet to push information into operand_lattices.

Source code in xdsl/analysis/sparse_analysis.py 
491
492
493
494
495
496
497
498
499
500
501
@abstractmethod
def visit_operation_impl(
    self,
    op: Operation,
    operand_lattices: list[PropagatingLatticeInvT],
    result_lattices: list[PropagatingLatticeInvT],
) -> None:
    """
    User-defined transfer function. Should read `result_lattices` and use
    `meet` to push information into `operand_lattices`.
    """

set_to_exit_state(lattice: PropagatingLatticeInvT) -> None abstractmethod

Sets a lattice to its most pessimistic (exit) state. Called for values whose backward propagation reaches an external boundary, e.g. operands of public function returns or arguments of public function entry blocks — for those, the analysis cannot assume any structure and must conservatively widen.

Source code in xdsl/analysis/sparse_analysis.py 
503
504
505
506
507
508
509
510
511
@abstractmethod
def set_to_exit_state(self, lattice: PropagatingLatticeInvT) -> None:
    """
    Sets a lattice to its most pessimistic (exit) state. Called for values
    whose backward propagation reaches an external boundary, e.g. operands
    of public function returns or arguments of public function entry
    blocks — for those, the analysis cannot assume any structure and must
    conservatively widen.
    """