Dataflow analysis

examples/analog/test.analog

@@ -91,3 +91,9 @@ a = true
 if (a) {
     b = 5
 }
+
+
+if (a) {}
+
+
+while (a) {}

mkdocs.yaml

@@ -150,7 +150,6 @@ markdown_extensions:
 
 extra_javascript:
   - javascripts/mathjax.js
-  - https://polyfill.io/v3/polyfill.min.js?features=es6
   - https://cdn.jsdelivr.net/npm/mathjax@3/es5/tex-mml-chtml.js
 
 extra_css:

pyproject.toml

@@ -89,7 +89,7 @@ select  = ["E4", "E7", "E9", "F", "I"]
 fixable = ["ALL"]
 
 [tool.uv.sources]
-oqd-compiler-infrastructure = { git = "https://github.com/openquantumdesign/oqd-compiler-infrastructure" }
+oqd-compiler-infrastructure = { git = "https://github.com/openquantumdesign/oqd-compiler-infrastructure", branch = "fix_maplattice" }
 
 [dependency-groups]
 dev = ["jupyter>=1.1.1", "pre-commit>=4.1.0", "ruff>=0.15.9"]

src/oqd_core/analysis/analog/__init__.py

@@ -1,12 +1,4 @@
 from .cfg import AnalogCFGBuilder
-from .symbol_table import (
-    AnalogSymbolError,
-    AnalogSymbolTable,
-    AnalogSymbolTableBuilder,
-    RegisterEnv,
-    SymbolBinding,
-    target_dim,
-)
 from .type_checker import AnalogTypeChecker
 from .types import AnalogTypeError
 
@@ -15,10 +7,4 @@
     "AnalogCFGBuilder",
     "AnalogTypeChecker",
     "AnalogTypeError",
-    "AnalogSymbolError",
-    "AnalogSymbolTable",
-    "AnalogSymbolTableBuilder",
-    "SymbolBinding",
-    "RegisterEnv",
-    "target_dim",
 ]

src/oqd_core/analysis/analog/cfg.py

@@ -15,6 +15,9 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterable
+from functools import reduce
+
 from oqd_compiler_infrastructure import CFG, CFGBlock, RewriteRule
 
 from oqd_core.interface.analog import (
@@ -27,9 +30,12 @@
 
 
 class AnalogCFGBuilder(RewriteRule):
-    def new_node(self, preds, stmt):
+    def new_node(self, preds, stmt, tags=None):
         node = CFGBlock(
-            register_id=self.index, stmts=[stmt] if stmt else [], preds=preds
+            register_id=self.index,
+            stmts=[stmt] if stmt else [],
+            preds=preds,
+            tags=tags if tags else {},
         )
         self.blocks[node.register_id] = node
         self.index += 1
@@ -41,6 +47,7 @@ def new_node(self, preds, stmt):
             label = explicit_labels.get(pred)
             if label is None:
                 label = self.fallthrough_labels.pop(pred, None)
+
             self.blocks[pred].add_succ(node.register_id, label=label)
 
         return node.register_id
@@ -73,34 +80,59 @@ def map_AnalogCircuit(self, model: AnalogCircuit) -> CFG:
         self.preds = []
         self.edge_labels = None
         self.fallthrough_labels = {}
-        node = self.new_node([], {})
+        node = self.new_node([], [])
         node = self.walk_block(model.statements, [node])
-        node = self.new_node(node, {})
+        node = self.new_node(node, [])
         return CFG(blocks=self.blocks)
 
     def map_IfElse(self, model: IfElse):
-        node = self.new_node(self.preds, model.condition)
-        then_branch = self.walk_block(model.then_branch, [node], entry_label="true")
-        if model.else_branch:
-            else_branch = self.walk_block(
-                model.else_branch, [node], entry_label="false"
-            )
-            return then_branch + else_branch
+        node = self.new_node(
+            self.preds,
+            model.condition,
+            tags={"__scf__": model.__class__.__qualname__},
+        )
 
-        self.fallthrough_labels[node] = "false"
-        return then_branch + [node]
+        then_branch = (
+            self.walk_block(model.then_branch, [node], entry_label="true")
+            if model.then_branch
+            else [node]
+        )
+        else_branch = (
+            self.walk_block(model.else_branch, [node], entry_label="false")
+            if model.else_branch
+            else [node]
+        )
+
+        fallthrough_labels = []
+        if model.then_branch == []:
+            fallthrough_labels.append("true")
+        if model.else_branch == []:
+            fallthrough_labels.append("false")
+
+        self.fallthrough_labels[node] = fallthrough_labels
+
+        return then_branch + else_branch
 
     def map_While(self, model: While):
-        node = self.new_node(self.preds, model.condition)
-        self.fallthrough_labels[node] = "false"
-        self.loop_stack.append(node)
-        body = self.walk_block(model.body, [node], entry_label="true")
-        self.loop_stack.pop()
+        node = self.new_node(
+            self.preds,
+            model.condition,
+            tags={"__scf__": model.__class__.__qualname__},
+        )
 
-        self.blocks[node].add_preds(body)
-        for s in body:
-            label = self.fallthrough_labels.pop(s, None)
-            self.blocks[s].add_succ(node, label=label)
+        if model.body:
+            self.loop_stack.append(node)
+            body = self.walk_block(model.body, [node], entry_label="true")
+            self.loop_stack.pop()
+
+            self.blocks[node].add_preds(body)
+            for loop_back in body:
+                label = self.fallthrough_labels.pop(loop_back, None)
+                self.blocks[loop_back].add_succ(node, label=label)
+        else:
+            self.blocks[node].add_succ(node, label="true")
+
+        self.fallthrough_labels[node] = "false"
 
         return self.blocks[node].exit_nodes + [node]
 
@@ -121,3 +153,119 @@ def map_Continue(self, model: Continue):
 
     def generic_map(self, model):
         return [self.new_node(self.preds, model)]
+
+
+class AnalogCFGtoAST(RewriteRule):
+    def __init__(self, pdom_result):
+        super().__init__()
+        self._pdom_result = pdom_result
+
+    @property
+    def pdom(self):
+        return self._pdom_result.out_states
+
+    @property
+    def dataflow_analysis(self):
+        return self._pdom_result.dataflow_analysis
+
+    @property
+    def lattice(self):
+        return self._pdom_result.dataflow_analysis.lattice
+
+    def _consume_ifelse(self, current_block, blocks):
+        ifelse_until = reduce(
+            self.lattice.meet,
+            [self.pdom[succ] for succ in current_block.succs],
+        )
+
+        then_block, then_succ = self._consume(
+            blocks,
+            start=current_block.edge_labels["true"],
+            until=ifelse_until,
+        )
+
+        else_block, else_succ = self._consume(
+            blocks,
+            start=current_block.edge_labels["false"],
+            until=ifelse_until,
+        )
+
+        return IfElse(
+            condition=current_block.stmts[0],
+            then_branch=then_block,
+            else_branch=else_block,
+        ), else_succ
+
+    def _get_while_dead_blocks(self, blocks, end):
+        return [
+            b
+            for b in blocks.keys()
+            if (len(blocks[b].preds) == 0 and end in self.pdom[b])
+        ]
+
+    def _consume_while(self, current_block, blocks):
+        while_until = reduce(
+            self.lattice.meet,
+            [self.pdom[succ] for succ in current_block.succs],
+        )
+
+        loop_block, loop_succ = self._consume(
+            blocks,
+            start=current_block.edge_labels["true"],
+            until=while_until,
+        )
+
+        while_dead_blocks = self._get_while_dead_blocks(
+            blocks, current_block.edge_labels["false"]
+        )
+
+        for b in sorted(while_dead_blocks):
+            loop_block.extend(
+                self._consume(blocks, start=b, until={current_block.register_id})[0]
+            )
+
+        return While(
+            condition=current_block.stmts[0], body=loop_block
+        ), current_block.edge_labels["false"]
+
+    def _consume(self, blocks, start=0, until=None):
+        succ = start
+        statements = []
+        while succ in blocks.keys():
+            if until and succ in until:
+                break
+
+            current_block = blocks.pop(succ)
+
+            if len(current_block.succs) == 0:
+                statements.extend(current_block.stmts)
+                break
+
+            match current_block.tags.get("__scf__", None):
+                case "IfElse":
+                    ifelse_statement, ifelse_succ = self._consume_ifelse(
+                        current_block, blocks
+                    )
+                    statements.append(ifelse_statement)
+                    succ = ifelse_succ
+
+                case "While":
+                    while_statement, while_succ = self._consume_while(
+                        current_block, blocks
+                    )
+                    statements.append(while_statement)
+                    succ = while_succ
+
+                case _:
+                    statements.extend(current_block.stmts)
+                    succ = list(current_block.succs)[0]
+
+        return statements, succ
+
+    def map_CFG(self, model):
+        circuit = AnalogCircuit()
+
+        statements, _ = self._consume(model.blocks)
+        circuit.statements.extend(statements)
+
+        return circuit