#!/usr/bin/env python3 # Copyright (c) the Selfie Project authors. All rights reserved. # Please see the AUTHORS file for details. Use of this source code is # governed by a BSD license that can be found in the LICENSE file. # Selfie is a project of the Computational Systems Group at the # Department of Computer Sciences of the University of Salzburg # in Austria. For further information and code please refer to: # selfie.cs.uni-salzburg.at # Context-free Language Ordered Bitvector Decision Diagrams (CFLOBVDDs) # CFLOBVDDs generalize CFLOBDDs in two ways: # CFLOBVDDs are CFLOBDDs over multi-node bitvector decision diagrams (BVDDs) # rather than single-node BDDs. A single node of a BVDD maps a bitvector of # size n, rather than just one bit, to no more than 2^n different values, # rather than just to no more than two different values. Hence a tree of # BVDD nodes of depth f maps a bitvector of size n*2^f bits to no more # than 2^(n*2^f) different values. A CFLOBVDD of depth l>=f over BVDDs # (mapping n*2^f-bit bitvectors) maps a bitvector of size n*2^(l-f) bits # to no more than 2^(n*2^(l-f)) different values. # CFLOBVDDs are kept minimal with respect to all recursively pairwise # reorderings of BVDDs in CFLOBVDDs (out of the factorially many possible # reorderings). Reorderings are explored from CFLOBVDD root nodes down to # a configurable level called swap level. CFLOBVDDs are initialized with # single-node BVDDs at level 0 which may then grow to a configurable level # called fork level (downsampling). If the swap level is less than the # fork level, BVDDs may be split in half recursively down to swap level # to explore reorderings (upsampling). # ------------------------------------------------------------ import bvdd as BVDD from math import log2 from math import ceil import threading class BV_Grouping: NO_SWAP_CACHING = True swap_cache_lock = threading.Lock() swap_cache = {} swap_cache_hits = 0 NO_UPSAMPLE_CACHING = False upsample_cache_lock = threading.Lock() upsample_cache = {} upsample_cache_hits = 0 NO_DOWNSAMPLE_CACHING = False downsample_cache_lock = threading.Lock() downsample_cache = {} downsample_cache_hits = 0 NO_COMPRESSED_CACHING = False compressed_cache_lock = threading.Lock() compressed_cache = {} compressed_cache_hits = 0 NO_PAIR_CACHING = False pair_product_cache_lock = threading.Lock() pair_product_cache = {} pair_product_cache_hits = 0 NO_TRIPLE_CACHING = False triple_product_cache_lock = threading.Lock() triple_product_cache = {} triple_product_cache_hits = 0 NO_REDUCTION_CACHING = False reduction_cache_lock = threading.Lock() reduction_cache = {} reduction_cache_hits = 0 NO_CLASSES_CACHING = False NO_REPRESENTATIVES = False def __init__(self, level, number_of_exits): assert level >= 0 self.level = level self.swap_level = level self.fork_level = level self.number_of_exits = number_of_exits def __repr__(self): return f"{self.level} w/ {self.number_of_exits} exits" def is_consistent(self): assert self.number_of_exits > 0 return True def is_swap_cached(self): if self in BV_Grouping.swap_cache: BV_Grouping.swap_cache_hits += 1 return True else: return False def get_cached_swap(self): assert self.is_swap_cached() return BV_Grouping.swap_cache[self] def cache_swap(self, g): if BV_Grouping.NO_SWAP_CACHING: return g with BV_Grouping.swap_cache_lock: if self not in BV_Grouping.swap_cache: BV_Grouping.swap_cache[self] = g # swapping is involutory (self-inverse) but only modulo compression return BV_Grouping.swap_cache[self] def is_upsample_cached(self): if self in BV_Grouping.upsample_cache: BV_Grouping.upsample_cache_hits += 1 return True else: return False def get_cached_upsample(self): assert self.is_upsample_cached() return BV_Grouping.upsample_cache[self] def cache_upsample(self, g): if BV_Grouping.NO_UPSAMPLE_CACHING: return g with BV_Grouping.upsample_cache_lock: if self not in BV_Grouping.upsample_cache: BV_Grouping.upsample_cache[self] = g return BV_Grouping.upsample_cache[self] def is_downsample_cached(self): if self in BV_Grouping.downsample_cache: BV_Grouping.downsample_cache_hits += 1 return True else: return False def get_cached_downsample(self): assert self.is_downsample_cached() return BV_Grouping.downsample_cache[self] def cache_downsample(self, g): if BV_Grouping.NO_DOWNSAMPLE_CACHING: return g with BV_Grouping.downsample_cache_lock: if self not in BV_Grouping.downsample_cache: BV_Grouping.downsample_cache[self] = g return BV_Grouping.downsample_cache[self] def downsample(self): assert self.level <= self.fork_level if self.is_downsample_cached(): return self.get_cached_downsample() g = self.downsample_uncached() # downsampling is inverse of upsampling: assert self is g.upsample_uncached() return self.cache_downsample(g) def is_compressed_cached(self): if self in BV_Grouping.compressed_cache: BV_Grouping.compressed_cache_hits += 1 return True else: return False def get_cached_compressed(self): assert self.is_compressed_cached() return BV_Grouping.compressed_cache[self] def cache_compressed(self, g): if BV_Grouping.NO_COMPRESSED_CACHING: return g with BV_Grouping.compressed_cache_lock: if self not in BV_Grouping.compressed_cache: BV_Grouping.compressed_cache[self] = g return BV_Grouping.compressed_cache[self] def is_pair_product_cached(self, g2): if (self, g2) in BV_Grouping.pair_product_cache: BV_Grouping.pair_product_cache_hits += 1 return True else: return False def get_cached_pair_product(self, g2): assert self.is_pair_product_cached(g2) return BV_Grouping.pair_product_cache[(self, g2)] def cache_pair_product(self, g2, pair_product, pt_ans): if BV_Grouping.NO_PAIR_CACHING: return (pair_product, pt_ans) with BV_Grouping.pair_product_cache_lock: if (self, g2) not in BV_Grouping.pair_product_cache: BV_Grouping.pair_product_cache[(self, g2)] = (pair_product, pt_ans) return BV_Grouping.pair_product_cache[(self, g2)] def is_triple_product_cached(self, g2, g3): if (self, g2, g3) in BV_Grouping.triple_product_cache: BV_Grouping.triple_product_cache_hits += 1 return True else: return False def get_cached_triple_product(self, g2, g3): assert self.is_triple_product_cached(g2, g3) return BV_Grouping.triple_product_cache[(self, g2, g3)] def cache_triple_product(self, g2, g3, triple_product, tt_ans): if BV_Grouping.NO_TRIPLE_CACHING: return (triple_product, tt_ans) with BV_Grouping.triple_product_cache_lock: if (self, g2, g3) not in BV_Grouping.triple_product_cache: BV_Grouping.triple_product_cache[(self, g2, g3)] = (triple_product, tt_ans) return BV_Grouping.triple_product_cache[(self, g2, g3)] def reduction_hash(reduction_tuple): return hash(tuple(reduction_tuple.values())) def is_reduction_cached(self, reduction_tuple): if (self, BV_Grouping.reduction_hash(reduction_tuple)) in BV_Grouping.reduction_cache: BV_Grouping.reduction_cache_hits += 1 return True else: return False def get_cached_reduction(self, reduction_tuple): assert self.is_reduction_cached(reduction_tuple) return BV_Grouping.reduction_cache[(self, BV_Grouping.reduction_hash(reduction_tuple))] def cache_reduction(self, reduction_tuple, reduction): if BV_Grouping.NO_REDUCTION_CACHING: return reduction reduction_hash = BV_Grouping.reduction_hash(reduction_tuple) with BV_Grouping.reduction_cache_lock: if (self, reduction_hash) not in BV_Grouping.reduction_cache: BV_Grouping.reduction_cache[(self, reduction_hash)] = reduction return BV_Grouping.reduction_cache[(self, reduction_hash)] def is_no_distinction_proto(self): return (isinstance(self, BV_Dont_Care_Grouping) or isinstance(self, BV_No_Distinction_Proto)) def reduce(self, reduction_tuple): assert len(reduction_tuple) == self.number_of_exits if reduction_tuple == dict(enumerate(range(1, len(reduction_tuple) + 1), 1)): return 1, self else: reduction_length = len(set(reduction_tuple.values())) if reduction_length == 1: return 1, BV_No_Distinction_Proto.representative(self.level, self.swap_level, self.fork_level) else: return reduction_length, self class BV_Dont_Care_Grouping(BV_Grouping): representatives_lock = threading.Lock() representatives = None def __init__(self): super().__init__(0, 1) def __repr__(self): return "dontcare @ " + super().__repr__() def is_consistent(self): assert super().is_consistent() return True def number_of_connections(self): return 0 def number_of_distinct_inputs(self, exit_i): assert exit_i == 1 return 1 def get_paths(self, exit_i, index_i = 0): assert exit_i == 1 return [(index_i, 0)] def representative(): if BV_Grouping.NO_REPRESENTATIVES: return BV_Dont_Care_Grouping() if BV_Dont_Care_Grouping.representatives is None: with BV_Dont_Care_Grouping.representatives_lock: if BV_Dont_Care_Grouping.representatives is None: BV_Dont_Care_Grouping.representatives = BV_Dont_Care_Grouping() assert BV_Dont_Care_Grouping.representatives.is_consistent() return BV_Dont_Care_Grouping.representatives def downsample_uncached(self): return BV_No_Distinction_Proto.downsample_uncached(self) def pair_product(self, g2, inorder = True): assert isinstance(g2, BV_Grouping) if inorder: g1 = self else: g1 = g2 g2 = self if g1.is_pair_product_cached(g2): return g1.get_cached_pair_product(g2) if inorder: g = g2 pt_ans = dict([(k, (1, k)) for k in range(1, g2.number_of_exits + 1)]) else: g = g1 pt_ans = dict([(k, (k, 1)) for k in range(1, g1.number_of_exits + 1)]) return g1.cache_pair_product(g2, g, pt_ans) def triple_product(self, g2, g3, order = 1): assert isinstance(g2, BV_Grouping) or isinstance(g3, BV_Grouping) if order == 1: g1 = self elif order == 2: g1 = g2 g2 = self else: assert order == 3 g1 = g2 g2 = g3 g3 = self if g1.is_triple_product_cached(g2, g3): return g1.get_cached_triple_product(g2, g3) if g1.is_no_distinction_proto() and g2.is_no_distinction_proto(): g = g3 tt_ans = dict([(k, (1, 1, k)) for k in range(1, g3.number_of_exits + 1)]) elif g1.is_no_distinction_proto() and g3.is_no_distinction_proto(): g = g2 tt_ans = dict([(k, (1, k, 1)) for k in range(1, g2.number_of_exits + 1)]) elif g2.is_no_distinction_proto() and g3.is_no_distinction_proto(): g = g1 tt_ans = dict([(k, (k, 1, 1)) for k in range(1, g1.number_of_exits + 1)]) elif g1.is_no_distinction_proto(): g, pt = g2.pair_product(g3) tt_ans = dict([(jk, (1, pt[jk][0], pt[jk][1])) for jk in pt]) elif g2.is_no_distinction_proto(): g, pt = g1.pair_product(g3) tt_ans = dict([(jk, (pt[jk][0], 1, pt[jk][1])) for jk in pt]) else: assert g3.is_no_distinction_proto() g, pt = g1.pair_product(g2) tt_ans = dict([(jk, (pt[jk][0], pt[jk][1], 1)) for jk in pt]) return g1.cache_triple_product(g2, g3, g, tt_ans) def reduce(self, reduction_tuple): assert reduction_tuple == {1:1} return self def compress(self): return self class BV_Fork_Grouping(BV_Grouping): # generalizing CFLOBDD forking to BVDDs representatives_lock = threading.Lock() representatives = {} representatives_hits = 0 def __init__(self, level, swap_level, fork_level, number_of_exits, bvdd): super().__init__(level, number_of_exits) assert 0 <= swap_level assert 0 <= fork_level self.swap_level = swap_level self.fork_level = fork_level self.bvdd = bvdd def __repr__(self): indentation = " " * (CFLOBVDD.max_level - self.level + 1) return (indentation + "\n" + indentation + "fork @ " + super().__repr__() + ":\n" + indentation + f"{self.bvdd}") def __hash__(self): return hash((self.level, self.swap_level, self.fork_level, self.number_of_exits, self.bvdd)) def __eq__(self, g2): return self is g2 or (isinstance(g2, BV_Fork_Grouping) and self.level == g2.level and self.swap_level == g2.swap_level and self.fork_level == g2.fork_level and self.number_of_exits == g2.number_of_exits and self.bvdd == g2.bvdd) def is_consistent(self): assert super().is_consistent() assert self.bvdd.is_consistent() assert self.bvdd.number_of_exits() == self.number_of_exits return True def number_of_connections(self): # counting CFLOBVDD connections only return 0 def number_of_distinct_inputs(self, exit_i): return self.bvdd.number_of_distinct_inputs(exit_i) def get_input_values(inputs): return BVDD.BVDD.get_input_values(inputs) def get_paths(self, exit_i, index_i = 0): assert 1 <= exit_i <= self.number_of_exits return self.bvdd.get_paths(exit_i, index_i) def representative(self): if BV_Grouping.NO_REPRESENTATIVES: return self if self in BV_Fork_Grouping.representatives: BV_Fork_Grouping.representatives_hits += 1 else: with BV_Fork_Grouping.representatives_lock: if self not in BV_Fork_Grouping.representatives: assert self.is_consistent() BV_Fork_Grouping.representatives[self] = self return BV_Fork_Grouping.representatives[self] def projection_proto(level, swap_level, fork_level, input_value): assert level == 0 bvdd = BVDD.BVDD.projection_proto(0, input_value) return BV_Fork_Grouping(level, swap_level, fork_level, bvdd.number_of_exits(), bvdd).representative() def upsample_uncached(self): g_a_bvdd, g_a_rt_inv, g_b_rt_inv, g_b_bvdds, g_b_return_tuples = self.bvdd.upsample(self.level) assert g_a_bvdd.number_of_exits() == len(g_a_rt_inv) if self.level == 0: if g_a_bvdd.is_constant(): return BV_Dont_Care_Grouping.representative() else: return self # TODO: enable different input orderings g = BV_Internal_Grouping(self.level, self.swap_level, self.fork_level, True, len(g_b_rt_inv)) if g_a_bvdd.is_constant(): g.a_connection = BV_No_Distinction_Proto.representative(self.level - 1, self.swap_level, self.fork_level) else: g.a_connection = BV_Fork_Grouping(self.level - 1, self.swap_level, self.fork_level, g_a_bvdd.number_of_exits(), g_a_bvdd).representative() assert g.a_connection.number_of_exits == len(g_b_bvdds) g.number_of_b_connections = g.a_connection.number_of_exits g.b_connections = {} for g_b_i in g_b_bvdds: if g_b_bvdds[g_b_i].is_constant(): g_b = BV_No_Distinction_Proto.representative(self.level - 1, self.swap_level, self.fork_level) else: g_b = BV_Fork_Grouping(self.level - 1, self.swap_level, self.fork_level, g_b_bvdds[g_b_i].number_of_exits(), g_b_bvdds[g_b_i]).representative() g.b_connections[g_b_i] = g_b assert g_b.number_of_exits == len(g_b_return_tuples[g_b_i]) g.b_return_tuples = g_b_return_tuples if (g.a_connection.is_no_distinction_proto() and g.number_of_b_connections == 1 and g.b_connections[1].is_no_distinction_proto()): return BV_No_Distinction_Proto.representative(self.level, self.swap_level, self.fork_level) else: return g.representative() def upsample(self): if self.is_upsample_cached(): return self.get_cached_upsample() g = self.upsample_uncached() # upsampling is inverse of downsampling: assert self is g.downsample_uncached() return self.cache_upsample(g) def downsample(self): assert self.level <= self.fork_level return self def pair_product(self, g2): assert isinstance(g2, BV_Grouping) g1 = self if g2.is_no_distinction_proto(): return g2.pair_product(g1, False) else: g1_orig = g1 g2_orig = g2 if g1_orig.is_pair_product_cached(g2_orig): return g1_orig.get_cached_pair_product(g2_orig) if g1.level > g1.swap_level: if isinstance(g2, BV_Internal_Grouping): g, pt = g1.upsample().pair_product(g2) return g1_orig.cache_pair_product(g2_orig, g, pt) else: pass # TODO: align g1 and g2 when forks support different input orderings elif isinstance(g2, BV_Internal_Grouping): g2 = g2.downsample() assert isinstance(g2, BV_Fork_Grouping) # assert g1 and g2 are aligned bvdd, pt = g1.bvdd.pair_product(g2.bvdd) g = BV_Fork_Grouping(g1.level, g1.swap_level, g1.fork_level, len(pt), bvdd) return g1_orig.cache_pair_product(g2_orig, g.representative(), pt) def triple_product(self, g2, g3): assert isinstance(g2, BV_Grouping) and isinstance(g3, BV_Grouping) g1 = self if g2.is_no_distinction_proto(): return g2.triple_product(g1, g3, 2) elif g3.is_no_distinction_proto(): return g3.triple_product(g1, g2, 3) else: g1_orig = g1 g2_orig = g2 g3_orig = g3 if g1_orig.is_triple_product_cached(g2_orig, g3_orig): return g1_orig.get_cached_triple_product(g2_orig, g3_orig) if g1.level > g1.swap_level: if isinstance(g2, BV_Internal_Grouping): if isinstance(g3, BV_Fork_Grouping): g3 = g3.upsample() assert isinstance(g3, BV_Internal_Grouping) g, tt = g1.upsample().triple_product(g2, g3) return g1_orig.cache_triple_product(g2_orig, g3_orig, g, tt) elif isinstance(g3, BV_Internal_Grouping): assert isinstance(g2, BV_Fork_Grouping) g, tt = g1.upsample().triple_product(g2.upsample(), g3) return g1_orig.cache_triple_product(g2_orig, g3_orig, g, tt) else: pass # TODO: align g1, g2, g3 when forks support different input orderings else: if isinstance(g2, BV_Internal_Grouping): g2 = g2.downsample() if isinstance(g3, BV_Internal_Grouping): g3 = g3.downsample() assert isinstance(g2, BV_Fork_Grouping) and isinstance(g3, BV_Fork_Grouping) # assert g1, g2, g3 are aligned bvdd, tt = g1.bvdd.triple_product(g2.bvdd, g3.bvdd) g = BV_Fork_Grouping(g1.level, g1.swap_level, g1.fork_level, len(tt), bvdd) return g1_orig.cache_triple_product(g2_orig, g3_orig, g.representative(), tt) def reduce(self, reduction_tuple): reduction_length, reduction = super().reduce(reduction_tuple) if reduction_length == 1: return reduction else: if self.is_reduction_cached(reduction_tuple): return self.get_cached_reduction(reduction_tuple) bvdd = self.bvdd.reduce(reduction_tuple) if bvdd.is_constant(): g = No_Distinction_Proto.representative(self.level, self.swap_level, self.fork_level) else: g = BV_Fork_Grouping(self.level, self.swap_level, self.fork_level, reduction_length, bvdd).representative() assert g.number_of_exits == reduction_length return self.cache_reduction(reduction_tuple, g) def compress(self): # equivalent to pair_product of No_Distinction_Proto with self where # pair_product for No_Distinction_Proto actually traverses its connections # (except for final downsampling as BVDDs do not yet different input orderings) if self.level == self.swap_level: return self assert self.level > self.swap_level if self.is_compressed_cached(): return self.get_cached_compressed() g = self.upsample().compress() # TODO: enable when forks support different input orderings # if self.level <= self.fork_level: # g = g.downsample() return self.cache_compressed(g) class BV_Internal_Grouping(BV_Grouping): representatives_lock = threading.Lock() representatives = {} representatives_hits = 0 def __init__(self, level, swap_level, fork_level, a2b, number_of_exits = 0): assert level > 0 super().__init__(level, number_of_exits) assert 0 <= swap_level assert 0 <= fork_level self.swap_level = swap_level self.fork_level = fork_level self.a2b = a2b self.a_connection = None self.a_return_tuple = {} self.number_of_b_connections = 0 self.b_connections = {} self.b_return_tuples = {} def __repr__(self): indentation = " " * (CFLOBVDD.max_level - self.level + 1) return (indentation + "\n" + indentation + f"{type(self).__name__} @ " + super().__repr__() + ":\n" + indentation + f"a2b: {self.a2b}\n" + indentation + f"a_c: {self.a_connection}\n" + indentation + f"a_rt: {self.a_return_tuple}\n" + indentation + f"n_of_b: {self.number_of_b_connections}\n" + indentation + f"b_c: {self.b_connections}\n" + indentation + f"b_rt: {self.b_return_tuples}") def __hash__(self): return hash((self.level, self.number_of_exits, self.a2b, self.a_connection, tuple(self.a_return_tuple.values()), self.number_of_b_connections, tuple(self.b_connections.values()), tuple([(tuple(rt), tuple(rt.values())) for rt in self.b_return_tuples.values()]))) def __eq__(self, g2): return self is g2 or (isinstance(g2, BV_Internal_Grouping) and self.level == g2.level and self.number_of_exits == g2.number_of_exits and self.a2b == g2.a2b and self.a_connection == g2.a_connection and self.a_return_tuple == g2.a_return_tuple and self.number_of_b_connections == g2.number_of_b_connections and self.b_connections == g2.b_connections and self.b_return_tuples == g2.b_return_tuples) def is_consistent(self): assert super().is_consistent() g_a = self.a_connection assert isinstance(g_a, BV_Grouping) assert not self.a_return_tuple or len(self.a_return_tuple) == g_a.number_of_exits assert len(self.a_return_tuple) == len(set(self.a_return_tuple.values())) assert self.number_of_b_connections == g_a.number_of_exits assert len(self.b_connections) == self.number_of_b_connections assert len(self.b_return_tuples) == len(self.b_connections) for g_a_e_i in self.a_return_tuple: assert 1 <= g_a_e_i <= g_a.number_of_exits a_e_i = self.a_return_tuple[g_a_e_i] assert g_a_e_i == a_e_i assert 1 <= a_e_i <= self.number_of_b_connections n = 0 for g_b_i in self.b_connections: g_b = self.b_connections[g_b_i] assert g_b_i in self.b_return_tuples g_b_i_rt = self.b_return_tuples[g_b_i] assert len(g_b_i_rt) == len(set(g_b_i_rt.values())) max_g_b_i_rt_e_j_e_t = n for g_b_i_rt_e_j in g_b_i_rt: assert 1 <= g_b_i_rt_e_j <= g_b.number_of_exits g_b_i_rt_e_j_e_t = g_b_i_rt[g_b_i_rt_e_j] assert 1 <= g_b_i_rt_e_j_e_t <= self.number_of_exits if g_b_i_rt_e_j_e_t > max_g_b_i_rt_e_j_e_t: # TODO: canonicalize swapped groupings and then check consistency here max_g_b_i_rt_e_j_e_t = g_b_i_rt_e_j_e_t n = max(n, max_g_b_i_rt_e_j_e_t) return True def number_of_connections(self): g_a = self.a_connection count = g_a.number_of_connections() + g_a.number_of_exits for g_b_i in self.b_connections: g_b = self.b_connections[g_b_i] count += g_b.number_of_connections() return count def number_of_distinct_inputs(self, exit_i): assert 1 <= exit_i <= self.number_of_exits count = 0 g_a = self.a_connection for g_b_i in self.b_connections: g_b = self.b_connections[g_b_i] g_b_i_rt = self.b_return_tuples[g_b_i] for g_b_i_rt_e_j in g_b_i_rt: if exit_i == g_b_i_rt[g_b_i_rt_e_j]: count += (g_a.number_of_distinct_inputs(g_b_i) * g_b.number_of_distinct_inputs(g_b_i_rt_e_j)) return count def get_paths(self, exit_i, index_i = 0): assert 1 <= exit_i <= self.number_of_exits if self.a2b: a_index_i = index_i b_index_i = index_i + 2**(self.level - 1) else: a_index_i = index_i + 2**(self.level - 1) b_index_i = index_i inputs = [] for b_i in self.b_return_tuples: b_rt = self.b_return_tuples[b_i] for b_e_i in b_rt: if exit_i == b_rt[b_e_i]: inputs += [(self.a_connection.get_paths(b_i, a_index_i), self.b_connections[b_i].get_paths(b_e_i, b_index_i))] break return inputs def representative(self): if BV_Grouping.NO_REPRESENTATIVES: return self if self in BV_Internal_Grouping.representatives: BV_Internal_Grouping.representatives_hits += 1 else: with BV_Internal_Grouping.representatives_lock: if self not in BV_Internal_Grouping.representatives: assert self.is_consistent() BV_Internal_Grouping.representatives[self] = self return BV_Internal_Grouping.representatives[self] def projection_proto(level, swap_level, fork_level, input_i, input_value): # generalizing CFLOBDD projection to generational CFLOBVDD projection assert 0 <= input_i < 2**level if level == 0: return BV_Fork_Grouping.projection_proto(level, swap_level, fork_level, input_value) else: g = BV_Internal_Grouping(level, swap_level, fork_level, True) if input_i < 2**(level - 1): g.a_connection = BV_Internal_Grouping.projection_proto(level - 1, swap_level, fork_level, input_i, input_value) g.number_of_exits = g.a_connection.number_of_exits # g.a_return_tuple == {} representing g.a_return_tuple = dict([(e, e) # for e in range(1, g.number_of_exits + 1)]) g.number_of_b_connections = g.number_of_exits projection_proto = BV_No_Distinction_Proto.representative(level - 1, swap_level, fork_level) g.b_connections = dict([(c, projection_proto) for c in range(1, g.number_of_b_connections + 1)]) g.b_return_tuples = dict([(e, {1:e}) for e in range(1, g.number_of_exits + 1)]) else: g.a_connection = BV_No_Distinction_Proto.representative(level - 1, swap_level, fork_level) # g.a_return_tuple == {} representing g.a_return_tuple[1] = 1 g.number_of_b_connections = 1 projection_proto = BV_Internal_Grouping.projection_proto(level - 1, swap_level, fork_level, input_i - 2**(level - 1), input_value) g.number_of_exits = projection_proto.number_of_exits g.b_connections = {1:projection_proto} g.b_return_tuples = {1:dict([(e, e) for e in range(1, g.number_of_exits + 1)])} return g.representative() def swap_uncached(self): assert self.level > self.swap_level g = BV_Internal_Grouping(self.level, self.swap_level, self.fork_level, not self.a2b, self.number_of_exits) g.a_connection = BV_No_Distinction_Proto.representative(self.level - 1, self.swap_level, self.fork_level) g_b_return_tuples = {1:{}} # a_connection becomes product of all b_connections which # is already reduced because a_return_tuple is idempotent for g_b_i in self.b_connections: g_b = self.b_connections[g_b_i] g_b_i_rt = self.b_return_tuples[g_b_i] g.a_connection, pt = g.a_connection.pair_product(g_b) g_b_return_tuples = dict([(i, # pt[i][0] is the exit index of the already paired b_connections # pt[i][1] is the exit index of the next b_connection being paired # extend b_return_tuples for reduced versions of a_connection below g_b_return_tuples[pt[i][0]] | # with the original exit of the next b_connection being paired {len(g_b_return_tuples[pt[i][0]]) + 1:g_b_i_rt[pt[i][1]]}) for i in pt]) reduction_tuple_a = {} # b_connections become reduced versions of a_connection for g_b_i in g_b_return_tuples: g_b_i_rt = g_b_return_tuples[g_b_i] induced_return_tuple, induced_reduction_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(g_b_i_rt) h = self.a_connection.reduce(induced_reduction_tuple) position = g.insert_b_connection(h, induced_return_tuple) reduction_tuple_a[len(reduction_tuple_a) + 1] = position induced_return_tuple, induced_reduction_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(reduction_tuple_a) g.a_connection = g.a_connection.reduce(induced_reduction_tuple) assert all([i == induced_return_tuple[i] for i in induced_return_tuple]) return g.representative() def swap(self): assert self.level > self.swap_level if self.is_swap_cached(): return self.get_cached_swap() if BV_Grouping.NO_SWAP_CACHING: NO_PAIR_CACHING = BV_Grouping.NO_PAIR_CACHING NO_REDUCTION_CACHING = BV_Grouping.NO_REDUCTION_CACHING NO_CLASSES_CACHING = BV_Grouping.NO_CLASSES_CACHING NO_REPRESENTATIVES = BV_Grouping.NO_REPRESENTATIVES BV_Grouping.NO_PAIR_CACHING = True BV_Grouping.NO_REDUCTION_CACHING = True BV_Grouping.NO_CLASSES_CACHING = True BV_Grouping.NO_REPRESENTATIVES = True g = self.swap_uncached() if BV_Grouping.NO_SWAP_CACHING: BV_Grouping.NO_PAIR_CACHING = NO_PAIR_CACHING BV_Grouping.NO_REDUCTION_CACHING = NO_REDUCTION_CACHING BV_Grouping.NO_CLASSES_CACHING = NO_CLASSES_CACHING BV_Grouping.NO_REPRESENTATIVES = NO_REPRESENTATIVES # swapping is involutory: assert self is g.swap_uncached() return self.cache_swap(g) def downsample_uncached(self): assert self.level <= self.fork_level if self.a2b: g = self else: # forks do not yet support different input orderings g = self.swap() g_a_bvdd = g.a_connection.downsample().bvdd g_b_bvdds = {} for g_b_i in g.b_connections: g_b = g.b_connections[g_b_i] g_b_bvdds[g_b_i] = g_b.downsample().bvdd bvdd = g_a_bvdd.downsample(g.level, g_b_bvdds, g.b_return_tuples) assert bvdd.number_of_exits() == g.number_of_exits g = BV_Fork_Grouping(g.level, g.swap_level, g.fork_level, g.number_of_exits, bvdd) return g.representative() def pair_compressed(self, g2, g1_other, g2_other): assert isinstance(g2, BV_Internal_Grouping) g1 = self size = g1.number_of_b_connections + g2.number_of_b_connections size_other = g1_other.number_of_b_connections + g2_other.number_of_b_connections if size < size_other: return g1.representative(), g2.representative() elif size > size_other: return g1_other.representative(), g2_other.representative() elif g1.a2b: # prefer original input ordering return g1.representative(), g2.representative() else: return g1_other.representative(), g2_other.representative() def pair_align(self, g2): assert isinstance(g2, BV_Internal_Grouping) g1 = self g1_swapped = g1.swap() g2_swapped = g2.swap() if g1.a2b == g2.a2b: return g1.pair_compressed(g2, g1_swapped, g2_swapped) else: return g1_swapped.pair_compressed(g2, g1, g2_swapped) def pair_product(self, g2): assert isinstance(g2, BV_Grouping) g1 = self assert g1.level == g2.level if g2.is_no_distinction_proto(): return g2.pair_product(g1, False) else: g1_orig = g1 g2_orig = g2 if g1_orig.is_pair_product_cached(g2_orig): return g1_orig.get_cached_pair_product(g2_orig) if g1.level > g1.swap_level: if isinstance(g2, BV_Fork_Grouping): g2 = g2.upsample() # g1 and g2 may be unaligned g1, g2 = g1.pair_align(g2) elif isinstance(g2, BV_Fork_Grouping): g, pt_ans = g1.downsample().pair_product(g2) return g1_orig.cache_pair_product(g2_orig, g, pt_ans) assert isinstance(g2, BV_Internal_Grouping) assert g1.a2b == g2.a2b if g1.level <= g1.fork_level: # g1 and g2 are downsampled in original order (decompressing) g, pt_ans = g1.downsample().pair_product(g2.downsample()) return g1_orig.cache_pair_product(g2_orig, g, pt_ans) g = BV_Internal_Grouping(g1.level, g1.swap_level, g1.fork_level, g1.a2b) g_a, pt_a = g1.a_connection.pair_product(g2.a_connection) g.a_connection = g_a # g.a_return_tuple == {} representing g.a_return_tuple = dict([(i, i) for i in pt_a]) g.number_of_b_connections = len(pt_a) pt_ans = {} pt_ans_inv = {} for j in pt_a: g_b, pt_b = g1.b_connections[pt_a[j][0]].pair_product(g2.b_connections[pt_a[j][1]]) g.b_connections[j] = g_b g.b_return_tuples[j] = {} for i in pt_b: c1 = g1.b_return_tuples[pt_a[j][0]][pt_b[i][0]] c2 = g2.b_return_tuples[pt_a[j][1]][pt_b[i][1]] if (c1, c2) in pt_ans_inv: g.b_return_tuples[j][len(g.b_return_tuples[j]) + 1] = pt_ans_inv[(c1, c2)] else: g.number_of_exits += 1 g.b_return_tuples[j][len(g.b_return_tuples[j]) + 1] = g.number_of_exits pt_ans[len(pt_ans) + 1] = (c1, c2) pt_ans_inv[(c1, c2)] = len(pt_ans) return g1_orig.cache_pair_product(g2_orig, g.representative(), pt_ans) def triple_compressed(self, g2, g3, g1_other, g2_other, g3_other): assert isinstance(g2, BV_Internal_Grouping) and isinstance(g3, BV_Internal_Grouping) g1 = self size = (g1.number_of_b_connections + g2.number_of_b_connections + g3.number_of_b_connections) size_other = (g1_other.number_of_b_connections + g2_other.number_of_b_connections + g3_other.number_of_b_connections) if size < size_other: return g1.representative(), g2.representative(), g3.representative() elif size > size_other: return g1_other.representative(), g2_other.representative(), g3_other.representative() elif g1.a2b: # prefer original input ordering return g1.representative(), g2.representative(), g3.representative() else: return g1_other.representative(), g2_other.representative(), g3_other.representative() def triple_align(self, g2, g3): assert isinstance(g2, BV_Internal_Grouping) and isinstance(g3, BV_Internal_Grouping) g1 = self g1_swapped = g1.swap() g2_swapped = g2.swap() g3_swapped = g3.swap() if g1.a2b == g2.a2b: if g2.a2b == g3.a2b: return g1.triple_compressed(g2, g3, g1_swapped, g2_swapped, g3_swapped) else: return g1.triple_compressed(g2, g3_swapped, g1_swapped, g2_swapped, g3) elif g2.a2b == g3.a2b: return g1.triple_compressed(g2_swapped, g3_swapped, g1_swapped, g2, g3) else: return g1.triple_compressed(g2_swapped, g3, g1_swapped, g2, g3_swapped) def triple_product(self, g2, g3): assert isinstance(g2, BV_Grouping) and isinstance(g3, BV_Grouping) g1 = self assert g1.level == g2.level == g3.level if g2.is_no_distinction_proto(): return g2.triple_product(g1, g3, 2) elif g3.is_no_distinction_proto(): return g3.triple_product(g1, g2, 3) else: g1_orig = g1 g2_orig = g2 g3_orig = g3 if g1_orig.is_triple_product_cached(g2_orig, g3_orig): return g1_orig.get_cached_triple_product(g2_orig, g3_orig) if g1.level > g1.swap_level: if isinstance(g2, BV_Fork_Grouping): g2 = g2.upsample() if isinstance(g3, BV_Fork_Grouping): g3 = g3.upsample() # g1, g2, g3 may be unaligned g1, g2, g3 = g1.triple_align(g2, g3) elif isinstance(g2, BV_Fork_Grouping): if isinstance(g3, BV_Internal_Grouping): g3 = g3.downsample() assert isinstance(g3, BV_Fork_Grouping) g, tt_ans = g1.downsample().triple_product(g2, g3) return g1_orig.cache_triple_product(g2_orig, g3_orig, g, tt_ans) elif isinstance(g3, BV_Fork_Grouping): assert isinstance(g2, BV_Internal_Grouping) g, tt_ans = g1.downsample().triple_product(g2.downsample(), g3) return g1_orig.cache_triple_product(g2_orig, g3_orig, g, tt_ans) assert isinstance(g2, BV_Internal_Grouping) and isinstance(g3, BV_Internal_Grouping) assert g1.a2b == g2.a2b == g3.a2b if g1.level <= g1.fork_level: # g1, g2, g3 are downsampled in original order (decompressing) g, tt_ans = g1.downsample().triple_product(g2.downsample(), g3.downsample()) return g1_orig.cache_triple_product(g2_orig, g3_orig, g, tt_ans) g = BV_Internal_Grouping(g1.level, g1.swap_level, g1.fork_level, g1.a2b) g_a, tt_a = g1.a_connection.triple_product(g2.a_connection, g3.a_connection) g.a_connection = g_a # g.a_return_tuple == {} representing g.a_return_tuple = dict([(i, i) for i in tt_a]) g.number_of_b_connections = len(tt_a) tt_ans = {} tt_ans_inv = {} for j in tt_a: g_b, tt_b = g1.b_connections[tt_a[j][0]].triple_product(g2.b_connections[tt_a[j][1]], g3.b_connections[tt_a[j][2]]) g.b_connections[j] = g_b g.b_return_tuples[j] = {} for i in tt_b: c1 = g1.b_return_tuples[tt_a[j][0]][tt_b[i][0]] c2 = g2.b_return_tuples[tt_a[j][1]][tt_b[i][1]] c3 = g3.b_return_tuples[tt_a[j][2]][tt_b[i][2]] if (c1, c2, c3) in tt_ans_inv: g.b_return_tuples[j][len(g.b_return_tuples[j]) + 1] = tt_ans_inv[(c1, c2, c3)] else: g.number_of_exits += 1 g.b_return_tuples[j][len(g.b_return_tuples[j]) + 1] = g.number_of_exits tt_ans[len(tt_ans) + 1] = (c1, c2, c3) tt_ans_inv[(c1, c2, c3)] = len(tt_ans) return g1_orig.cache_triple_product(g2_orig, g3_orig, g.representative(), tt_ans) def insert_b_connection(self, h, return_tuple): assert isinstance(h, BV_Grouping) for i in self.b_connections: if self.b_connections[i] == h and self.b_return_tuples[i] == return_tuple: return i self.number_of_b_connections += 1 self.b_connections[self.number_of_b_connections] = h self.b_return_tuples[self.number_of_b_connections] = return_tuple return self.number_of_b_connections def reduce(self, reduction_tuple): reduction_length, reduction = super().reduce(reduction_tuple) if reduction_length == 1: return reduction else: g = self if g.is_reduction_cached(reduction_tuple): return g.get_cached_reduction(reduction_tuple) g_prime = BV_Internal_Grouping(g.level, g.swap_level, g.fork_level, g.a2b, reduction_length) reduction_tuple_a = {} for i in g.b_connections: induced_return_tuple, induced_reduction_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(dict(enumerate([reduction_tuple[v] for v in g.b_return_tuples[i].values()], 1))) h = g.b_connections[i].reduce(induced_reduction_tuple) position = g_prime.insert_b_connection(h, induced_return_tuple) reduction_tuple_a[len(reduction_tuple_a) + 1] = position induced_return_tuple, induced_reduction_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(reduction_tuple_a) g_prime.a_connection = g.a_connection.reduce(induced_reduction_tuple) assert all([i == induced_return_tuple[i] for i in induced_return_tuple]) # g_prime.a_return_tuple == {} representing g_prime.a_return_tuple = induced_return_tuple return g.cache_reduction(reduction_tuple, g_prime.representative()) def compress(self): # equivalent to pair_product of No_Distinction_Proto with self where # pair_product for No_Distinction_Proto actually traverses its connections # (except for final downsampling as BVDDs do not yet different input orderings) if self.level == self.swap_level: return self assert self.level > self.swap_level if self.is_compressed_cached(): return self.get_cached_compressed() # compress in levels top-down g_swapped = self.swap() g = (g_swapped if g_swapped.number_of_b_connections < self.number_of_b_connections else self if g_swapped.number_of_b_connections > self.number_of_b_connections else # prefer original input ordering g_swapped if g_swapped.a2b else self) compressed_a_connection = g.a_connection.compress() has_been_compressed = compressed_a_connection is not g.a_connection compressed_b_connections = {} for g_b_i in g.b_connections: g_b = g.b_connections[g_b_i] compressed_b_connections[g_b_i] = g_b.compress() has_been_compressed |= compressed_b_connections[g_b_i] is not g_b if has_been_compressed: compressed_g = BV_Internal_Grouping(g.level, g.swap_level, g.fork_level, g.a2b, g.number_of_exits) compressed_g.a_connection = compressed_a_connection compressed_g.number_of_b_connections = g.number_of_b_connections compressed_g.b_connections = compressed_b_connections compressed_g.b_return_tuples = g.b_return_tuples return self.cache_compressed(compressed_g.representative()) else: return self.cache_compressed(g.representative()) def swap_flow(self, outputs): assert self.a2b g = BV_Internal_Grouping(self.level, self.swap_level, self.fork_level, self.a2b, self.number_of_exits) if None not in outputs.values(): assert self.number_of_exits == 2 assert self.number_of_b_connections == 1 assert self.b_return_tuples == {1:{1:1,2:2}} g.a_connection = self.b_connections[1] none_exit_i = 2 g_b_rt = {1:1,2:3} g.number_of_exits += 1 outputs = {1:outputs[1],2:None,3:outputs[2]} else: none_exit_i = [i for i in outputs if outputs[i] is None][0] g.a_connection = BV_No_Distinction_Proto.representative(self.level - 1, self.swap_level, self.fork_level) g_b_rt = {1:none_exit_i} # a_connection becomes product of all b_connections for g_b_i in self.b_connections: g_b = self.b_connections[g_b_i] g_b_i_rt = self.b_return_tuples[g_b_i] g.a_connection, pt = g.a_connection.pair_product(g_b) new_g_b_rt = {} for i in pt: if g_b_rt[pt[i][0]] == none_exit_i or g_b_i == 1: new_g_b_rt[i] = g_b_i_rt[pt[i][1]] else: assert g_b_i_rt[pt[i][1]] == none_exit_i new_g_b_rt[i] = g_b_rt[pt[i][0]] g_b_rt = new_g_b_rt reduction_tuple_a = {} # b_connections become reduced versions of a_connection for g_b_i in g_b_rt: g_b_i_rt = dict([i, g_b_rt[i]] if i == g_b_i else [i, none_exit_i] for i in g_b_rt) induced_return_tuple, induced_reduction_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(g_b_i_rt) assert g.a_connection.number_of_exits == len(induced_reduction_tuple) h = g.a_connection.reduce(induced_reduction_tuple) position = g.insert_b_connection(h, induced_return_tuple) reduction_tuple_a[len(reduction_tuple_a) + 1] = position induced_return_tuple, induced_reduction_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(reduction_tuple_a) g.a_connection = g.a_connection.reduce(induced_reduction_tuple) assert all([i == induced_return_tuple[i] for i in induced_return_tuple]) return g.representative(), outputs class BV_No_Distinction_Proto(BV_Internal_Grouping): representatives_lock = threading.Lock() representatives = {} representatives_hits = 0 def __init__(self, level, swap_level, fork_level): assert level > 0 assert 0 <= swap_level assert 0 <= fork_level super().__init__(level, swap_level, fork_level, True, 1) def __repr__(self): return f"{type(self).__name__} @ {self.level}" def representative(level, swap_level, fork_level): if level == 0: return BV_Dont_Care_Grouping.representative() elif level in BV_No_Distinction_Proto.representatives: BV_No_Distinction_Proto.representatives_hits += 1 return BV_No_Distinction_Proto.representatives[level] else: g = BV_No_Distinction_Proto(level, swap_level, fork_level) g.a_connection = BV_No_Distinction_Proto.representative(level - 1, swap_level, fork_level) # g.a_return_tuple == {} representing g.a_return_tuple[1] = 1 g.number_of_b_connections = 1 g.b_connections[1] = g.a_connection g.b_return_tuples[1] = {1:1} if BV_Grouping.NO_REPRESENTATIVES: return g if level not in BV_No_Distinction_Proto.representatives: with BV_No_Distinction_Proto.representatives_lock: if level not in BV_No_Distinction_Proto.representatives: BV_No_Distinction_Proto.representatives[level] = g return BV_No_Distinction_Proto.representatives[level] def swap(self): assert self.level > self.swap_level return self def downsample_uncached(self): return BV_Fork_Grouping(self.level, self.swap_level, self.fork_level, 1, BVDD.BVDD.constant(1)).representative() def pair_product(self, g2, inorder = True): assert isinstance(g2, BV_Grouping) g1 = self assert g1.level == g2.level return BV_Dont_Care_Grouping.pair_product(g1, g2, inorder) def triple_product(self, g2, g3, order = 1): assert isinstance(g2, BV_Grouping) and isinstance(g3, BV_Grouping) g1 = self assert g1.level == g2.level == g3.level return BV_Dont_Care_Grouping.triple_product(g1, g2, g3, order) def reduce(self, reduction_tuple): assert reduction_tuple == {1:1} return self def compress(self): return self def swap_flow(self, outputs): return self, outputs class Collapsed_Classes: cache_lock = threading.Lock() cache = {} cache_hits = 0 def __init__(self, classes): self.classes = classes def __hash__(self): return hash((tuple(self.classes.values()), any(isinstance(self.classes[i], bool) for i in self.classes))) def __eq__(self, c2): return self is c2 or (isinstance(c2, Collapsed_Classes) and self.classes == c2.classes) def are_collapsed_classes_cached(equiv_classes): if Collapsed_Classes(equiv_classes) in Collapsed_Classes.cache: Collapsed_Classes.cache_hits += 1 return True else: return False def get_collapsed_classes(equiv_classes): return Collapsed_Classes.cache[Collapsed_Classes(equiv_classes)] def cache_collapsed_classes(equiv_classes, projected_classes, renumbered_classes): if BV_Grouping.NO_CLASSES_CACHING: return (projected_classes, renumbered_classes) collapsed_classes = Collapsed_Classes(equiv_classes) with Collapsed_Classes.cache_lock: if collapsed_classes not in Collapsed_Classes.cache: Collapsed_Classes.cache[collapsed_classes] = (projected_classes, renumbered_classes) return Collapsed_Classes.cache[collapsed_classes] class CFLOBVDD: REDUCE = True max_level_lock = threading.Lock() max_level = 0 representatives_lock = threading.Lock() representatives = {} representatives_hits = 0 def __init__(self, grouping, outputs): self.grouping = grouping self.outputs = outputs def __str__(self): return ("CFLOBVDD with\n" + f"g: {self.grouping}\n" + f"o: {self.outputs}\n") def __hash__(self): return hash((self.grouping, tuple(self.outputs.values()), any(isinstance(self.outputs[i], bool) for i in self.outputs))) def __eq__(self, n2): return self is n2 or (isinstance(n2, CFLOBVDD) and self.grouping == n2.grouping and self.outputs == n2.outputs) def print_profile(): print("CFLOBVDD cache profile:") print(f"forks: {BVDD.utilization(BV_Fork_Grouping.representatives_hits, len(BV_Fork_Grouping.representatives))}") print(f"internals: {BVDD.utilization(BV_Internal_Grouping.representatives_hits, len(BV_Internal_Grouping.representatives))}") print(f"protos: {BVDD.utilization(BV_No_Distinction_Proto.representatives_hits, len(BV_No_Distinction_Proto.representatives))}") print(f"swap: {BVDD.utilization(BV_Grouping.swap_cache_hits, len(BV_Grouping.swap_cache))}") print(f"upsample: {BVDD.utilization(BV_Grouping.upsample_cache_hits, len(BV_Grouping.upsample_cache))}") print(f"downsample: {BVDD.utilization(BV_Grouping.downsample_cache_hits, len(BV_Grouping.downsample_cache))}") print(f"compressed: {BVDD.utilization(BV_Grouping.compressed_cache_hits, len(BV_Grouping.compressed_cache))}") print(f"pair product: {BVDD.utilization(BV_Grouping.pair_product_cache_hits, len(BV_Grouping.pair_product_cache))}") print(f"triple product: {BVDD.utilization(BV_Grouping.triple_product_cache_hits, len(BV_Grouping.triple_product_cache))}") print(f"reduction: {BVDD.utilization(BV_Grouping.reduction_cache_hits, len(BV_Grouping.reduction_cache))}") print(f"classes: {BVDD.utilization(Collapsed_Classes.cache_hits, len(Collapsed_Classes.cache))}") print(f"CFLOBVDDs: {BVDD.utilization(CFLOBVDD.representatives_hits, len(CFLOBVDD.representatives))}") def number_of_connections(self): return self.grouping.number_of_connections() def number_of_outputs(self): return len(self.outputs) def get_distinct_outputs(self): return set(self.outputs.values()) def number_of_distinct_outputs(self): return len(self.get_distinct_outputs()) def number_of_distinct_inputs(self): return sum(self.grouping.number_of_distinct_inputs(exit_i) for exit_i in self.outputs) def number_of_solutions(self, value): return sum(self.grouping.number_of_distinct_inputs(exit_i) for exit_i in self.outputs if self.outputs[exit_i] == value) def get_printed_paths(paths, full_paths = True): printed_paths = [] for path in paths: if isinstance(path[0], int): assert isinstance(path[1], int) index_i = path[0] inputs = path[1] if inputs == 0: if full_paths: printed_paths += [f"[dontcare @ {index_i}]"] else: printed_paths += ["[" + f"input @ {index_i}: " + BV_Fork_Grouping.get_input_values(inputs) + "]"] else: a_paths = CFLOBVDD.get_printed_paths(path[0], full_paths) b_paths = CFLOBVDD.get_printed_paths(path[1], full_paths) if a_paths: if b_paths: printed_paths += ["(" + "&".join(a_paths + b_paths) + ")"] else: printed_paths += a_paths elif b_paths: printed_paths += b_paths if len(printed_paths) > 1: return ["[" + "|".join(printed_paths) + "]"] else: return printed_paths def get_printed_value_paths(self, value = None): value_paths = "" for exit_i in self.outputs: if value is None or self.outputs[exit_i] == value: if value_paths: value_paths += "\n" value_paths += (f"{self.outputs[exit_i]} <- " + CFLOBVDD.get_printed_paths(self.grouping.get_paths(exit_i))[0]) # without reductions value may appear more than once return value_paths def get_printed_inputs(self, value = None): return (f"CFLOBVDD:\n" + f"{2**self.grouping.level} input variables\n" + f"{self.number_of_connections()} connections\n" + f"{self.number_of_outputs()} output values\n" + f"{self.number_of_distinct_outputs()} disctinct output values (exits)\n" f"{self.number_of_distinct_inputs()} distinct inputs\n" + f"{self.number_of_solutions(value)} solutions\n" + f"{self.get_printed_value_paths(value)}") def is_consistent(self): assert self.grouping.is_consistent() assert self.number_of_outputs() == self.grouping.number_of_exits, f"{self.number_of_outputs()} == {self.grouping.number_of_exits}" assert self.number_of_outputs() >= self.number_of_distinct_outputs() assert not CFLOBVDD.REDUCE or self.number_of_outputs() == self.number_of_distinct_outputs() return True def representative(grouping, outputs): cflobvdd = CFLOBVDD(grouping, outputs) if BV_Grouping.NO_REPRESENTATIVES: return cflobvdd if cflobvdd in CFLOBVDD.representatives: CFLOBVDD.representatives_hits += 1 else: with CFLOBVDD.representatives_lock: if cflobvdd not in CFLOBVDD.representatives: assert cflobvdd.is_consistent() CFLOBVDD.representatives[cflobvdd] = cflobvdd return CFLOBVDD.representatives[cflobvdd] def is_always_false(self): # without reductions False may appear more than once return (False in self.outputs.values() and (self.number_of_distinct_outputs() == 1 or (self.number_of_distinct_outputs() == 2 and None in self.outputs.values()))) def is_always_true(self): # without reductions True may appear more than once return (True in self.outputs.values() and (self.number_of_distinct_outputs() == 1 or (self.number_of_distinct_outputs() == 2 and None in self.outputs.values()))) def constant_CFLOBVDD(level, swap_level, fork_level, output = 0): assert 0 <= swap_level <= level assert 0 <= fork_level <= level with CFLOBVDD.max_level_lock: CFLOBVDD.max_level = max(CFLOBVDD.max_level, level) return CFLOBVDD.representative( BV_No_Distinction_Proto.representative(level, swap_level, fork_level), {1:output}) def byte_constant(level, swap_level, fork_level, number_of_input_bytes, output): assert number_of_input_bytes > 0 level = max(level, swap_level, fork_level, ceil(log2(number_of_input_bytes))) return CFLOBVDD.constant_CFLOBVDD(level, swap_level, fork_level, output) def constant(output): return CFLOBVDD.byte_constant(CFLOBVDD.level, CFLOBVDD.swap_level, CFLOBVDD.fork_level, CFLOBVDD.number_of_inputs, output) def false(level, swap_level, fork_level): return CFLOBVDD.constant_CFLOBVDD(level, swap_level, fork_level, False) def true(level, swap_level, fork_level): return CFLOBVDD.constant_CFLOBVDD(level, swap_level, fork_level, True) def flip_value_tuple(self): # self must be reduced assert self.number_of_outputs() == 2 return CFLOBVDD.representative(self.grouping, {1:self.outputs[2], 2:self.outputs[1]}) def complement(self): if self == CFLOBVDD.false(self.grouping.level, self.grouping.swap_level, self.grouping.fork_level): return CFLOBVDD.true(self.grouping.level, self.grouping.swap_level, self.grouping.fork_level) elif self == CFLOBVDD.true(self.grouping.level, self.grouping.swap_level, self.grouping.fork_level): return CFLOBVDD.false(self.grouping.level, self.grouping.swap_level, self.grouping.fork_level) else: # self must be reduced return self.flip_value_tuple() def unary_apply_and_reduce(self, op, number_of_output_bits): return self.binary_apply_and_reduce( CFLOBVDD.constant_CFLOBVDD(self.grouping.level, self.grouping.swap_level, self.grouping.fork_level), lambda x, y: op(x) if x is not None else None, number_of_output_bits) def compute_unary(self, op, op_id, bits): return self.unary_apply_and_reduce(op, abs(bits)) def projection_CFLOBVDD(level, swap_level, fork_level, input_i, reorder = False, input_value = None): assert 0 <= swap_level <= level assert 0 <= fork_level <= level assert 0 <= input_i < 2**level with CFLOBVDD.max_level_lock: CFLOBVDD.max_level = max(CFLOBVDD.max_level, level) grouping = BV_Internal_Grouping.projection_proto(level, swap_level, fork_level, input_i, input_value) if reorder: grouping = grouping.compress() if input_value is not None: assert grouping.number_of_exits == 2 return CFLOBVDD.representative(grouping, {1:CFLOBVDD.has_input, 2:CFLOBVDD.no_input}) else: return CFLOBVDD.representative(grouping, dict([(output + 1, output) for output in range(grouping.number_of_exits)])) def byte_projection(level, swap_level, fork_level, number_of_input_bytes, byte_i, reorder = False, input_value = None): level = max(level, swap_level, fork_level, ceil(log2(number_of_input_bytes))) assert 0 <= byte_i < 2**level return CFLOBVDD.projection_CFLOBVDD(level, swap_level, fork_level, byte_i, reorder, input_value) def projection(index, input_value = None): return CFLOBVDD.byte_projection(CFLOBVDD.level, CFLOBVDD.swap_level, CFLOBVDD.fork_level, CFLOBVDD.number_of_inputs, index, True, input_value) def collapse_classes_leftmost(equiv_classes): # legacy code if Collapsed_Classes.are_collapsed_classes_cached(equiv_classes): return Collapsed_Classes.get_collapsed_classes(equiv_classes) # square-time iteration over equivalence classes projected_classes = dict(enumerate([equiv_classes[i] for i in equiv_classes if i == min([j for j in equiv_classes if equiv_classes[j] == equiv_classes[i]])], 1)) order_of_projected_classes = dict([(projected_classes[i], i) for i in projected_classes]) return Collapsed_Classes.cache_collapsed_classes(equiv_classes, projected_classes, dict(enumerate([order_of_projected_classes[v] for v in equiv_classes.values()], 1))) def linear_collapse_classes_leftmost(equiv_classes): if Collapsed_Classes.are_collapsed_classes_cached(equiv_classes): return Collapsed_Classes.get_collapsed_classes(equiv_classes) leftmost_equiv_class_index = {} projected_classes = {} renumbered_classes = {} # linear-time iteration over equivalence classes for index in equiv_classes: value = equiv_classes[index] if value not in leftmost_equiv_class_index: # remember leftmost equivalence class index leftmost_equiv_class_index[value] = len(leftmost_equiv_class_index) + 1 projected_classes[len(leftmost_equiv_class_index)] = value renumbered_classes[index] = leftmost_equiv_class_index[value] return Collapsed_Classes.cache_collapsed_classes(equiv_classes, projected_classes, renumbered_classes) def binary_apply_and_reduce(self, n2, op, number_of_output_bits): assert isinstance(n2, CFLOBVDD) n1 = self g, pt = n1.grouping.pair_product(n2.grouping) equiv_classes = dict([(i, op(n1.outputs[pt[i][0]], n2.outputs[pt[i][1]]) if n1.outputs[pt[i][0]] is not None and n2.outputs[pt[i][1]] is not None else None) for i in pt]) assert all(0 <= equiv_classes[i] < 2**number_of_output_bits for i in pt if equiv_classes[i] is not None) g = g.compress() if CFLOBVDD.REDUCE: induced_value_tuple, induced_return_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(equiv_classes) g = g.reduce(induced_return_tuple).compress() else: induced_value_tuple = equiv_classes return CFLOBVDD.representative(g, induced_value_tuple) def compute_binary(self, op, n2, op_id, bits): return self.binary_apply_and_reduce(n2, op, abs(bits)) def ternary_apply_and_reduce(self, n2, n3, op, number_of_output_bits): assert isinstance(n2, CFLOBVDD) and isinstance(n3, CFLOBVDD) n1 = self g, tt = n1.grouping.triple_product(n2.grouping, n3.grouping) equiv_classes = dict([(i, op(n1.outputs[tt[i][0]], n2.outputs[tt[i][1]], n3.outputs[tt[i][2]]) if n1.outputs[tt[i][0]] is not None and n2.outputs[tt[i][1]] is not None and n3.outputs[tt[i][2]] is not None else None) for i in tt]) assert all(0 <= equiv_classes[i] < 2**number_of_output_bits for i in tt if equiv_classes[i] is not None) g = g.compress() if CFLOBVDD.REDUCE: induced_value_tuple, induced_return_tuple = \ CFLOBVDD.linear_collapse_classes_leftmost(equiv_classes) g = g.reduce(induced_return_tuple).compress() else: induced_value_tuple = equiv_classes return CFLOBVDD.representative(g, induced_value_tuple) def compute_ternary(self, op, n2, n3, op_id, bits): return self.ternary_apply_and_reduce(n2, n3, op, abs(bits)) def compute_ite(self, n2, n3, op_id, bits): return self.compute_ternary(lambda x, y, z: y if x else z, n2, n3, op_id, bits) def swap_flow(self): grouping, outputs = self.grouping.swap_flow(self.outputs) return CFLOBVDD.representative(grouping, outputs) # for PDDs level = 0 swap_level = 0 fork_level = 0 number_of_inputs = 1 has_input = 0 no_input = 1 def number_of_partitioned_inputs(self): return self.number_of_solutions(CFLOBVDD.has_input) def partitioned_constant(): return CFLOBVDD.constant(CFLOBVDD.has_input) def partitioned_projection(index, input_value): return CFLOBVDD.projection(index, input_value) def op_union(output1, output2): return output1 if output2 == CFLOBVDD.no_input else output2 def union(self, n2): return self.compute_binary(lambda x, y: CFLOBVDD.op_union(x, y), n2, None, 1) def op_intersection(output1, output2, output3 = 0): return output1 if output1 == CFLOBVDD.no_input else output2 if output2 == CFLOBVDD.no_input else output3 def intersection(self, n2, n3 = None): if n3 is None: return self.compute_binary(lambda x, y: CFLOBVDD.op_intersection(x, y), n2, None, 1) else: return self.compute_ternary(lambda x, y, z: CFLOBVDD.op_intersection(x, y, z), n2, n3, None, 1) def is_not_empty(self): return not (self.number_of_distinct_outputs() == 1 and self.outputs[1] == CFLOBVDD.no_input) def is_not_full(self): return not (self.number_of_distinct_outputs() == 1 and self.outputs[1] == CFLOBVDD.has_input)