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)