from __future__ import annotations
import itertools
import logging
from typing import Dict, List, Literal, Tuple, Callable, Sequence, Union
import sympy
from prodigy.pgcl import (BoolType, Program, VarExpr,
Instr)
from prodigy.analysis.config import ForwardAnalysisConfig
from prodigy.analysis.instructionhandler.program_info import ProgramInfo
from prodigy.analysis.solver.solver_type import SolverType
from prodigy.distribution.distribution import Distribution, State
from prodigy.util.color import Style
from prodigy.util.logger import log_setup
logger = log_setup(str(__name__).rsplit(".", maxsplit=1)[-1], logging.DEBUG)
[docs]
def generate_equivalence_test_distribution(
program: Program,
config: ForwardAnalysisConfig) -> Tuple[Distribution, Dict[str, str]]:
"""
Generates a second-order PGF, dependent on the given variables in a program. This SOP can be used to check
equivalences of two programs.
.. returns: The SOP expression.
"""
logger.debug("Generating test distribution.")
dist = config.factory.one()
so_vars: Dict[str, str] = {} # second order variables
for variable in program.variables:
new_var = dist.get_fresh_variable()
if isinstance(program.variables[variable], BoolType):
dist *= config.factory.from_expr(f"(1+{new_var} * {variable})",
VarExpr(var=new_var),
VarExpr(var=variable))
else:
dist *= config.factory.from_expr(f"1/(1-{new_var}*{variable})",
VarExpr(var=new_var),
VarExpr(var=variable))
so_vars[new_var] = variable
logger.debug("Test distribution is %s with second order variables %s",
dist, so_vars)
return dist.set_variables(*program.variables.keys(),
*so_vars.keys()).set_parameters(), so_vars
[docs]
def check_equivalence(
program: Program,
other_program: Program,
config: ForwardAnalysisConfig,
analyzer: Callable[
[Union[Instr, Sequence[Instr]], ProgramInfo, Distribution, Distribution, ForwardAnalysisConfig],
tuple[Distribution, Distribution]
]
) -> Tuple[Literal[True], List[Dict[str, str]]] | Tuple[
Literal[False], State] | Tuple[None, Distribution]:
"""
This method uses the fact that we can sometimes determine program equivalence,
by checking the equality of two parametrized infinite-state Distributions.
If they are equivalent, also returns a list of constraints under which this holds
(which may be empty if they are always equal). If not, also returns a counterexample.
If the solution cannot be determined (returns `None`), also returns the difference of
the second order distributions generated by the invariant and the once unrolled while
loop of the program. If this difference can be made equal to 0, the programs are
equivalent.
:param config: The configuration.
:param program: The first program
:param other_program: The second program
:params analyzer: The analyzer.
:returns: Whether the invariant and the program are equivalent.
"""
logger.debug("Checking equivalence.")
# If both programs do not have the same variables, they are not equal
if set(program.variables.keys()).difference(set(other_program.variables.keys())) != set():
logger.info("The two programs already have different variable sets, prog: %s, other: %s",
program.variables.keys(), other_program.variables.keys())
return False, State() # TODO how should the state look like?
# Now we have to generate an infinite state parametrized distribution for every program variable.
test_dist, new_vars = generate_equivalence_test_distribution(program, config)
# Compute the resulting distributions for both programs
logger.debug("Compute the other programs posterior...")
other_program_posterior, other_program_error = analyzer(
other_program.instructions,
ProgramInfo(other_program, so_vars=frozenset(new_vars.keys())),
test_dist,
config.factory.from_expr("0", *(other_program.variables | new_vars.keys())),
config
)
logger.debug("other programs result:\n%s", other_program_posterior)
logger.debug("Compute the posterior of the program...")
if config.show_intermediate_steps:
print(f"\n{Style.YELLOW} Compute the posterior of the program. {Style.RESET}")
program_posterior, program_error = analyzer(
program.instructions,
ProgramInfo(program, so_vars=frozenset(new_vars.keys())),
test_dist,
config.factory.one(*(program.variables | new_vars.keys())) * 0,
config
)
logger.debug("program result:\n%s", program_posterior)
diff = program_posterior - other_program_posterior
# Compare them and check whether they are equal.
params = program.parameters.keys() | other_program.parameters.keys()
solver = SolverType.make(config.solver_type)
dist_is_solution, dist_candidates = solver.solve(program_posterior.set_parameters(*params),
other_program_posterior.set_parameters(*params))
err_is_solution, err_candidates = solver.solve(program_error.set_parameters(*params),
other_program_error.set_parameters(*params))
# If there is no solution tell the user why.
if (dist_is_solution and err_is_solution) is False:
logger.debug("Equivalence refuted.")
state, _ = diff.get_state()
return False, State({var: state[sym] for sym, var in new_vars.items()})
# If there is a solution, individually, see whether they match.
if (dist_is_solution and err_is_solution) is True:
if not dist_candidates:
logger.debug("Posterior equivalence validated.")
return True, err_candidates
if not err_candidates:
logger.debug("Posterior error validated.")
return True, dist_candidates
logger.debug("Matching individual solutions\n%s\n%s", dist_candidates, err_candidates)
res_both = []
for dist_sol, err_sol in itertools.product(dist_candidates, err_candidates):
to_be_solved = []
for var, constr in dist_sol.items() | err_sol.items():
to_be_solved.append(sympy.S(f'({var}) - ({constr})'))
logger.debug("Solving the equation system: %s", to_be_solved)
res_both += sympy.solve(
to_be_solved, dict=True) if len(to_be_solved) > 1 else sympy.solve(
to_be_solved[0], dict=True)
logger.debug("Current partial solution space: %s", res_both)
if len(res_both) == 0:
return False, State() # TODO how to generate a counterexample here?
logger.debug("Equivalence validated.")
return True, res_both
logger.debug("Equivalence unknown.")
return None, diff