Improve expression propagation by removing redundant Phi functions (#382

)

* Collapse redundant Phi in expressionpropagationmemory

* Move around code to expression propagation common

* Make sure replaced instructions are added after phi instructions

* Remove leftover code

* Use PhiFunctionCleaner to remove phi functions

* Add test

* black

decompiler/pipeline/commons/expressionpropagationcommons.py

@@ -1,15 +1,15 @@
 import logging
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from typing import DefaultDict, Dict, Iterator, Optional, Set
+from typing import DefaultDict, Iterator, Optional, Set
 
+from decompiler.pipeline.ssa.phi_cleaner import PhiFunctionCleaner
 from decompiler.pipeline.stage import PipelineStage
 from decompiler.structures.graphs.cfg import BasicBlock, ControlFlowGraph
 from decompiler.structures.maps import DefMap, UseMap
 from decompiler.structures.pointers import Pointers
 from decompiler.structures.pseudo import (
     Assignment,
-    Branch,
     Call,
     DataflowObject,
     Expression,
@@ -57,6 +57,8 @@ def perform(self, graph, iteration) -> bool:
         # block map is updated after substitution in EPM, in EP does nothing
         # use map is updated after substitution in EPM, in EP does nothing
         """
+        self._remove_redundant_phis(graph)
+
         is_changed = False
         self._cfg = graph
         self._initialize_maps(graph)
@@ -74,6 +76,10 @@ def perform(self, graph, iteration) -> bool:
                                 is_changed = old != str(instruction)
         return is_changed
 
+    def _remove_redundant_phis(self, graph: ControlFlowGraph):
+        phi_functions_of = {node: [i for i in node.instructions if isinstance(i, Phi)] for node in graph.nodes}
+        PhiFunctionCleaner(phi_functions_of).clean_up()
+
     @abstractmethod
     def _definition_can_be_propagated_into_target(self, definition: Assignment, target: Instruction) -> bool:
         """

decompiler/util/iteration_util.py

@@ -0,0 +1,8 @@
+from collections.abc import Iterable
+from itertools import groupby
+
+
+def all_equal(iterable: Iterable):
+    """See https://stackoverflow.com/a/3844832"""
+    g = groupby(iterable)
+    return next(g, True) and not next(g, False)

tests/pipeline/dataflowanalysis/test_expression_propagation.py

@@ -819,6 +819,39 @@ def test_contraction_copy():
     assert id(instr1.value) != id(instr2.value)
 
 
+def test_phi_simplifcation():
+    """
+    Test that redundant phi functions are removed, enabling better propagation.
+    """
+    var0 = Variable("var0", Integer.int32_t())
+    var1 = Variable("var1", Integer.int32_t())
+    var2 = Variable("var2", Integer.int32_t())
+    var3 = Variable("var3", Integer.int32_t())
+    arg0 = Variable("arg0", Integer.int32_t())
+
+    b0 = BasicBlock(
+        0,
+        [Assignment(var0, Constant(42, Integer.int32_t())), Branch(Condition(OperationType.less, [arg0, Constant(0, Integer.int32_t())]))],
+    )
+    b1 = BasicBlock(1, [Assignment(var1, var0)])
+    b2 = BasicBlock(2, [Assignment(var2, var0)])
+    b3 = BasicBlock(3, [Phi(var3, [var1, var2], {b1: var1, b2: var2}), ret_ins := Return([var3])])
+
+    cfg = ControlFlowGraph()
+    cfg.add_node(b0)
+    cfg.add_node(b1)
+    cfg.add_node(b2)
+    cfg.add_node(b3)
+    cfg.add_edge(TrueCase(b0, b1))
+    cfg.add_edge(FalseCase(b0, b2))
+    cfg.add_edge(UnconditionalEdge(b1, b3))
+    cfg.add_edge(UnconditionalEdge(b2, b3))
+
+    _run_expression_propagation(cfg)
+
+    assert ret_ins.values.operands == [Constant(42, Integer.int32_t())]
+
+
 def _generate_options(instr: int = 10, branch: int = 10, call: int = 10, assignment: int = 10) -> Options:
     options = Options()
     options.set("expression-propagation.maximum_instruction_complexity", instr)