Skip to content

cfg_builder

FusedCFGBuilder

Incrementally builds control-flow graph for Finch Fused IR.

Source code in src/finch/finch_fused/cfg_builder.py 
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
178
179
180
181
182
183
184
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
234
235
236
237
238
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
class FusedCFGBuilder:
    """Incrementally builds control-flow graph for Finch Fused IR."""

    def __init__(self) -> None:
        self.cfg: ControlFlowGraph = ControlFlowGraph()
        self.current_block: BasicBlock = self.cfg.entry_block

    def emit(self, stmt) -> None:
        """Add a statement to the current block."""
        self.current_block.add_statement(stmt)

    def build(self, node: FusedNode) -> ControlFlowGraph:
        return self(node)

    def __call__(
        self,
        node: FusedNode,
        break_block: BasicBlock | None = None,
        return_block: BasicBlock | None = None,
    ) -> ControlFlowGraph:
        match node:
            case Assign():
                self.emit(node)
            case NumberedStatement(stmt, _):
                match stmt:
                    case Return(_):
                        self.emit(node)
                        assert return_block
                        self.current_block.add_successor(return_block)
                        unreachable_block = self.cfg.new_block()
                        self.current_block = unreachable_block
                    case Break():
                        self.emit(node)
                        assert break_block
                        self.current_block.add_successor(break_block)
                        unreachable_block = self.cfg.new_block()
                        self.current_block = unreachable_block
                    case _:
                        self.emit(node)
            case Block(bodies):
                for body in bodies:
                    self(body, break_block, return_block)
            case If(cond, body, else_body):
                before_block = self.current_block
                self.emit(cond)

                # create blocks for if, else, and after
                if_block = self.cfg.new_block()
                else_block = self.cfg.new_block()
                after_block = self.cfg.new_block()

                # connect before block to if and else blocks
                before_block.add_successor(if_block)
                before_block.add_successor(else_block)

                # fill in the if block
                self.current_block = if_block
                self(body, break_block, return_block)
                self.current_block.add_successor(after_block)

                # fill in the else block
                self.current_block = else_block
                if else_body is not None:
                    self(else_body, break_block, return_block)
                self.current_block.add_successor(after_block)

                # continue building after the if-else
                self.current_block = after_block
            case While(cond, body):
                before_block = self.current_block
                self.emit(cond)

                # create blocks for the loop body and the code after the loop
                body_block = self.cfg.new_block()
                after_block = self.cfg.new_block()

                # connect before block to the loop body and the after block
                before_block.add_successor(body_block)
                before_block.add_successor(after_block)

                # fill in the loop body
                self.current_block = body_block
                self(body, after_block, return_block)
                self.emit(cond)

                # connect the end of loop body back to the beginning to form the loop
                self.current_block.add_successor(body_block)
                self.current_block.add_successor(after_block)
                self.current_block = after_block
            case For(_, iter, body):
                before_block = self.current_block
                body_block = self.cfg.new_block()
                after_block = self.cfg.new_block()

                # connect before block to the loop body and the after block
                before_block.add_successor(body_block)
                before_block.add_successor(after_block)

                # fill in the loop body
                self.current_block = body_block
                self.emit(iter)
                self(body, after_block, return_block)

                # connect the end of loop body back to the beginning to form the loop
                self.current_block.add_successor(body_block)
                self.current_block.add_successor(after_block)
                self.current_block = after_block

            case Function(func_name, _, body):
                # set block names to the function name
                self.cfg.block_name = str(func_name.val)

                # create entry/exit block for the function
                func_entry_block = self.cfg.new_block()
                func_exit_block = self.cfg.new_block()

                # connect CFG entry block to the function's entry block
                self.cfg.entry_block.add_successor(func_entry_block)

                # dive into the body of the function
                self.current_block = func_entry_block
                self(body, break_block, func_exit_block)

                # connect last block in the function to the
                # exit block of the function
                self.current_block.add_successor(func_exit_block)

                # connect function exit block to the exit block of the CFG
                func_exit_block.add_successor(self.cfg.exit_block)

            case node:
                raise NotImplementedError(node, "FusedCFGBuilder")

        return self.cfg

emit(stmt)

Add a statement to the current block.

Source code in src/finch/finch_fused/cfg_builder.py 
138
139
140
def emit(self, stmt) -> None:
    """Add a statement to the current block."""
    self.current_block.add_statement(stmt)

NumberedStatement dataclass

Bases: FusedTree, FusedStatement

Wrapper for AssemblyStatement that assigns a unique id to each statement for easier tracking in the CFG.

Attributes:

Name Type Description
stmt FusedStatement

The original FusedStatement being wrapped.
sid int

A unique integer identifier for the statement.
Source code in src/finch/finch_fused/cfg_builder.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass(eq=True, frozen=True)
class NumberedStatement(FusedTree, FusedStatement):
    """
    Wrapper for AssemblyStatement that assigns a unique id to each statement
    for easier tracking in the CFG.

    Attributes:
        stmt: The original FusedStatement being wrapped.
        sid: A unique integer identifier for the statement.
    """

    stmt: FusedStatement
    sid: int

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

    @property
    def children(self):
        return (self.stmt, self.sid)

fused_build_cfg(node)

Build control-flow graph for a FinchAssembly node. Args: node: Root FinchAssembly node to build CFG for. Returns: ControlFlowGraph: The constructed control-flow graph.

Source code in src/finch/finch_fused/cfg_builder.py 
267
268
269
270
271
272
273
274
275
276
def fused_build_cfg(node: FusedNode) -> ControlFlowGraph:
    """
    Build control-flow graph for a FinchAssembly node.
    Args:
        node: Root FinchAssembly node to build CFG for.
    Returns:
        ControlFlowGraph: The constructed control-flow graph.
    """
    ctx = FusedCFGBuilder()
    return ctx.build(node)

fused_desugar(node)

Lower surface syntax to a core AST shape before CFG construction.

  • make function parameters explicit in the function body so they get statement ids and can be referenced in the CFG (e.g. desugar def f(x): return x + 1 to def f(x): x_1 = x; return x_1 + 1 so that the parameter x gets a statement id and can be referenced in the CFG)
Source code in src/finch/finch_fused/cfg_builder.py 
 91
 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
123
124
125
126
127
128
def fused_desugar(node: FusedNode) -> FusedNode:
    """
    Lower surface syntax to a core AST shape before CFG construction.

    - make function parameters explicit in the function body so they get statement ids
         and can be referenced in the CFG (e.g. desugar `def f(x): return x + 1` to
        `def f(x): x_1 = x; return x_1 + 1` so that the parameter `x` gets a statement
        id and can be referenced in the CFG)
    """

    def go(node: FusedNode) -> FusedNode:
        """Recursively desugar the AST."""
        match node:
            case Module(funcs):
                return Module(cast(tuple[Function, ...], tuple(go(f) for f in funcs)))
            case Function(name, args, body):
                body_2 = cast(Block, go(body))

                # Make argument definitions explicit so they get statement ids.
                func_prologue = tuple((Assign(arg, arg)) for arg in args)
                return Function(name, args, Block((*func_prologue, *body_2.body)))
            case Block(bodies):
                new_bodies: list[FusedStatement] = []
                for b in bodies:
                    b2 = go(b)
                    new_bodies.append(cast(FusedStatement, b2))
                return Block(tuple(new_bodies))
            case If(cond, body, else_body):
                new_else = cast(Block, go(else_body)) if else_body is not None else None
                return If(cond, cast(Block, go(body)), new_else)
            case While(cond, body):
                return While(cond, cast(Block, go(body)))
            case For(target, iter, body):
                return For(target, iter, cast(Block, go(body)))
            case node:
                return node

    return go(node)

number_statements(node, sid=0)

Recursively traverse the AST and wrap each statement in a NumberedStatement with a unique id.

Parameters:

Name Type Description Default
node FusedNode

The root node of the AST to number.

 required
sid int

The starting statement id for numbering. Defaults to 0.

 0

Returns:

Type Description
FusedNode

A tuple containing the new AST with numbered statements and the next
int

available statement id.
Source code in src/finch/finch_fused/cfg_builder.py 
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
def number_statements(node: FusedNode, sid: int = 0) -> tuple[FusedNode, int]:
    """
    Recursively traverse the AST and wrap each statement in a NumberedStatement
      with a unique id.

    Args:
        node: The root node of the AST to number.
        sid: The starting statement id for numbering. Defaults to 0.

    Returns:
        A tuple containing the new AST with numbered statements and the next
        available statement id.
    """

    def go(node: FusedNode) -> FusedNode:
        match node:
            case Module(funcs):
                return Module(cast(tuple[Function, ...], tuple(go(f) for f in funcs)))
            case Function(name, args, body):
                return Function(name, args, cast(Block, go(body)))
            case Block(bodies):
                new_bodies: list[FusedStatement] = []
                for b in bodies:
                    b2 = go(b)
                    if isinstance(b2, Block):
                        new_bodies.extend(b2.body)
                    else:
                        new_bodies.append(cast(FusedStatement, b2))
                return Block(tuple(new_bodies))
            case If(cond, body, else_body):
                new_else = cast(Block, go(else_body)) if else_body is not None else None
                return If(cond, cast(Block, go(body)), new_else)
            case While(cond, body):
                return While(cond, cast(Block, go(body)))
            case For(target, iter, body):
                return For(target, iter, cast(Block, go(body)))
            case node:
                if isinstance(node, (Assign, Return, Break)):
                    nonlocal sid
                    s = NumberedStatement(node, sid)
                    sid += 1
                    return s
                return node

    return go(node), sid