from __future__ import annotations # mypy: disallow-untyped-defs, disallow-incomplete-defs, disallow-untyped-calls import collections import hashlib import os import pickle import typing import itertools from . import types from .utils import consume, keep_until, split, default_id_dict, default_fwd_dict from .ordered import OrderedSet, OrderedFrozenSet from .actions import Action, Replay, Reduce, FilterStates, Seq from .grammar import End, ErrorSymbol, InitNt, Nt from .rewrites import CanonicalGrammar from .lr0 import LR0Generator, Term from .aps import APS, Edge, Path # StateAndTransitions objects are indexed using a StateId which is an integer. StateId = int # Action or ordered sequence of action which have to be performed. DelayedAction = typing.Union[Action, typing.Tuple[Action, ...]] class StateAndTransitions: """This is one state of the parse table, which has transitions based on terminals (text), non-terminals (grammar rules) and epsilon (reduce). In this model epsilon transitions are used to represent code to be executed such as reduce actions and any others actions. """ __slots__ = ["index", "locations", "terminals", "nonterminals", "errors", "epsilon", "delayed_actions", "arguments", "backedges", "_hash", "stable_hash"] # Numerical index of this state. index: StateId # The stable_str of each LRItem we could be parsing in this state: the # places in grammar productions that tell what we've already parsed, # i.e. how we got to this state. locations: OrderedFrozenSet[str] # Ordered set of Actions which are pushed to the next state after a # conflict. delayed_actions: OrderedFrozenSet[DelayedAction] # Number of argument of an action state. # # Instead of having action states with a non-empty replay list of terms, we # have a non-empty list of argument which size is described by this # variable. arguments: int # Outgoing edges taken when shifting terminals. terminals: typing.Dict[str, StateId] # Outgoing edges taken when shifting nonterminals after reducing. nonterminals: typing.Dict[Nt, StateId] # Error symbol transitions. errors: typing.Dict[ErrorSymbol, StateId] # List of epsilon transitions with associated actions. epsilon: typing.List[typing.Tuple[Action, StateId]] # Set of edges that lead to this state. backedges: OrderedSet[Edge] # Cached hash code. This class implements __hash__ and __eq__ in order to # help detect equivalent states (which must be merged, for correctness). _hash: int # A hash code computed the same way as _hash, but used only for # human-readable output. The stability is useful for debugging, to match # states across multiple runs of the parser generator. stable_hash: str def __init__( self, index: StateId, locations: OrderedFrozenSet[str], delayed_actions: OrderedFrozenSet[DelayedAction] = OrderedFrozenSet(), arguments: int = 0 ) -> None: assert isinstance(locations, OrderedFrozenSet) assert isinstance(delayed_actions, OrderedFrozenSet) self.index = index self.terminals = {} self.nonterminals = {} self.errors = {} self.epsilon = [] self.locations = locations self.delayed_actions = delayed_actions self.arguments = arguments self.backedges = OrderedSet() # NOTE: The hash of a state depends on its location in the LR0 # parse-table, as well as the actions which have not yet been executed. def hashed_content() -> typing.Iterator[object]: for item in sorted(self.locations): yield item yield "\n" yield "delayed_actions" for action in self.delayed_actions: yield hash(action) yield "arguments" yield arguments self._hash = hash(tuple(hashed_content())) h = hashlib.md5() h.update("".join(map(str, hashed_content())).encode()) self.stable_hash = h.hexdigest()[:6] def is_inconsistent(self) -> bool: "Returns True if the state transitions are inconsistent." # TODO: We could easily allow having a state with non-terminal # transition and other epsilon transitions, as the non-terminal shift # transitions are a form of condition based on the fact that a # non-terminal, produced by a reduce action is consumed by the # automaton. if len(self.terminals) + len(self.nonterminals) + len(self.errors) > 0 and len(self.epsilon) > 0: return True elif len(self.epsilon) == 1: if any(k.is_inconsistent() for k, s in self.epsilon): return True elif len(self.epsilon) > 1: if any(k.is_inconsistent() for k, s in self.epsilon): return True # NOTE: We can accept multiple conditions as epsilon transitions # iff they are checking the same variable with non-overlapping # values. This implies that we can implement these conditions as a # deterministic switch statement in the code emitter. if any(not k.is_condition() for k, s in self.epsilon): return True iterator = iter(self.epsilon) first, _ = next(iterator) if any(not first.check_same_variable(k) for k, s in iterator): return True # "type: ignore" because mypy does not see that the preceding if-statement # means all k.condition() actions are FilterFlags. pairs = itertools.combinations((k for k, s in self.epsilon), 2) if any(not k1.check_different_values(k2) for k1, k2 in pairs): return True else: try: self.get_error_symbol() except ValueError: return True return False def shifted_edges(self) -> typing.Iterator[ typing.Tuple[typing.Union[str, Nt, ErrorSymbol], StateId] ]: k: Term s: StateId for k, s in self.terminals.items(): yield (k, s) for k, s in self.nonterminals.items(): yield (k, s) for k, s in self.errors.items(): yield (k, s) def edges(self) -> typing.Iterator[typing.Tuple[Term, StateId]]: k: Term s: StateId for k, s in self.terminals.items(): yield (k, s) for k, s in self.nonterminals.items(): yield (k, s) for k, s in self.errors.items(): yield (k, s) for k, s in self.epsilon: yield (k, s) def rewrite_state_indexes( self, state_map: typing.Dict[StateId, StateId] ) -> None: def apply_on_term(term: typing.Union[Term, None]) -> Term: assert term is not None if isinstance(term, Action): return term.rewrite_state_indexes(state_map) return term self.index = state_map[self.index] self.terminals = { k: state_map[s] for k, s in self.terminals.items() } self.nonterminals = { k: state_map[s] for k, s in self.nonterminals.items() } self.errors = { k: state_map[s] for k, s in self.errors.items() } self.epsilon = [ (k.rewrite_state_indexes(state_map), state_map[s]) for k, s in self.epsilon ] # We cannot have multiple identical actions jumping to different locations. assert len(self.epsilon) == len(set(k for k, _ in self.epsilon)) self.backedges = OrderedSet( Edge(state_map[edge.src], apply_on_term(edge.term)) for edge in self.backedges ) def get_error_symbol(self) -> typing.Optional[ErrorSymbol]: if len(self.errors) > 1: raise ValueError("More than one error symbol on the same state.") else: return next(iter(self.errors), None) def __contains__(self, term: object) -> bool: if isinstance(term, Action): for t, s in self.epsilon: if t == term: return True return False elif isinstance(term, Nt): return term in self.nonterminals elif isinstance(term, ErrorSymbol): return term in self.errors else: return term in self.terminals def __getitem__(self, term: Term) -> StateId: if isinstance(term, Action): for t, s in self.epsilon: if t == term: return s raise KeyError(term) elif isinstance(term, Nt): return self.nonterminals[term] if isinstance(term, ErrorSymbol): return self.errors[term] else: return self.terminals[term] def get(self, term: Term, default: object) -> object: try: return self.__getitem__(term) except KeyError: return default def stable_str(self, states: typing.List[StateAndTransitions]) -> str: conflict = "" if self.is_inconsistent(): conflict = " (inconsistent)" return "{}{}:\n{}".format(self.stable_hash, conflict, "\n".join([ "\t{} --> {}".format(k, states[s].stable_hash) for k, s in self.edges()])) def __str__(self) -> str: conflict = "" if self.is_inconsistent(): conflict = " (inconsistent)" return "{}{}:\n{}".format(self.index, conflict, "\n".join([ "\t{} --> {}".format(k, s) for k, s in self.edges()])) def __eq__(self, other: object) -> bool: return (isinstance(other, StateAndTransitions) and sorted(self.locations) == sorted(other.locations) and sorted(self.delayed_actions) == sorted(other.delayed_actions) and self.arguments == other.arguments) def __hash__(self) -> int: return self._hash DebugInfo = typing.Dict[StateId, int] class ParseTable: """The parser can be represented as a matrix of state transitions where on one side we have the current state, and on the other we have the expected terminal, non-terminal or epsilon transition. a b c A B C #1 #2 #3 +---+---+---+---+---+---+----+----+----+ s1 | | | | | | | | | | s2 | | | | | | | | | | s3 | | | | | | | | | | . | | | | | | | | | | . | | | | | | | | | | . | | | | | | | | | | s67 | | | | | | | | | | s68 | | | | | | | | | | s69 | | | | | | | | | | +---+---+---+---+---+---+----+----+----+ The terminals `a` are the token which are read from the input. The non-terminals `A` are the token which are pushed by the reduce actions of the epsilon transitions. The epsilon transitions `#1` are the actions which have to be executed as code by the parser. A parse table is inconsistent if there is any state which has an epsilon transitions and terminals/non-terminals transitions (shift-reduce conflict), or a state with more than one epsilon transitions (reduce-reduce conflict). This is equivalent to having a non deterministic state machine. """ __slots__ = [ "actions", "states", "state_cache", "named_goals", "terminals", "nonterminals", "debug_info", "exec_modes", "assume_inconsistent" ] # Map of actions identifier to the corresponding object. actions: typing.List[Action] # Map of state identifier to the corresponding object. states: typing.List[StateAndTransitions] # Hash table of state objects, ensuring we never have two equal states. state_cache: typing.Dict[StateAndTransitions, StateAndTransitions] # List of (Nt, states) tuples which are the entry point of the state # machine. named_goals: typing.List[typing.Tuple[Nt, StateId]] # Set of all terminals. terminals: OrderedFrozenSet[typing.Union[str, End]] # List of non-terminals. nonterminals: typing.List[Nt] # Carry the info to be used when generating debug_context. If False, # then no debug_context is ever produced. debug_info: typing.Union[bool, DebugInfo] # Execution modes are used by the code generator to decide which # function is executed when. This is a dictionary of OrderedSet, where # the keys are the various parsing modes, and the mapped set contains # the list of traits which have to be implemented, and consequently # which functions would be encoded. exec_modes: typing.Optional[typing.DefaultDict[str, OrderedSet[types.Type]]] # True if the parse table might be inconsistent. When this is False, we add # extra assertions when computing the reduce path. assume_inconsistent: bool def __init__( self, grammar: CanonicalGrammar, verbose: bool = False, progress: bool = False, debug: bool = False ) -> None: self.actions = [] self.states = [] self.state_cache = {} self.named_goals = [] self.terminals = grammar.grammar.terminals self.nonterminals = typing.cast( typing.List[Nt], list(grammar.grammar.nonterminals.keys())) # typing.cast() doesn't actually check at run time, so let's do that: assert all(isinstance(nt, Nt) for nt in self.nonterminals) self.debug_info = debug self.exec_modes = grammar.grammar.exec_modes self.assume_inconsistent = True self.create_lr0_table(grammar, verbose, progress) self.fix_inconsistent_table(verbose, progress) # TODO: Optimize chains of actions into sequences. # Optimize by removing unused states. self.remove_all_unreachable_state(verbose, progress) # TODO: Statically compute replayed terms. (maybe?) # Replace reduce actions by programmatic stack manipulation. self.lower_reduce_actions(verbose, progress) # Fold Replay followed by Unwind instruction. self.fold_replay_unwind(verbose, progress) # Fold paths which have the same ending. self.fold_identical_endings(verbose, progress) # Group state with similar non-terminal edges close-by, to improve the # generated Rust code by grouping matched state numbers. self.group_nonterminal_states(verbose, progress) # Split shift states from epsilon states. # self.group_epsilon_states(verbose, progress) def save(self, filename: os.PathLike) -> None: with open(filename, 'wb') as f: pickle.dump(self, f) @classmethod def load(cls, filename: os.PathLike) -> ParseTable: with open(filename, 'rb') as f: obj = pickle.load(f) if len(f.read()) != 0: raise ValueError("file has unexpected extra bytes at end") if not isinstance(obj, cls): raise TypeError("file contains wrong kind of object: expected {}, got {}" .format(cls.__name__, obj.__class__.__name__)) return obj def is_inconsistent(self) -> bool: "Returns True if the grammar contains any inconsistent state." for s in self.states: if s is not None and s.is_inconsistent(): return True return False def rewrite_state_indexes(self, state_map: typing.Dict[StateId, StateId]) -> None: for s in self.states: if s is not None: s.rewrite_state_indexes(state_map) self.named_goals = [ (nt, state_map[s]) for nt, s in self.named_goals ] # After a rewrite, multiple actions (conditions) might jump to the same # target, attempt to fold these conditions based on having the same # target. If we can merge them, then remove previous edges (updating # the backedges of successor states) and replace them by the newly # created edges. for s in self.states: if s is not None and len(s.epsilon) != 0: epsilon_by_dest = collections.defaultdict(list) for k, d in s.epsilon: epsilon_by_dest[d].append(k) for d, ks in epsilon_by_dest.items(): if len(ks) == 1: continue new_ks = ks[0].fold_by_destination(ks) if new_ks == ks: continue # This collection is required by `remove_edge`, but in this # particular case we know for sure that at least one edge # would be added back. Therefore no need to use the content # of the set. maybe_unreachable_set: OrderedSet[StateId] = OrderedSet() assert len(new_ks) > 0 for k in ks: self.remove_edge(s, k, maybe_unreachable_set) for k in new_ks: self.add_edge(s, k, d) self.assert_table_invariants() def rewrite_reordered_state_indexes(self) -> None: state_map = { s.index: i for i, s in enumerate(self.states) if s is not None } self.rewrite_state_indexes(state_map) def new_state( self, locations: OrderedFrozenSet[str], delayed_actions: OrderedFrozenSet[DelayedAction] = OrderedFrozenSet(), arguments: int = 0 ) -> typing.Tuple[bool, StateAndTransitions]: """Get or create state with an LR0 location and delayed actions. Returns a tuple where the first element is whether the element is newly created, and the second element is the State object.""" index = len(self.states) state = StateAndTransitions(index, locations, delayed_actions, arguments) try: return False, self.state_cache[state] except KeyError: self.state_cache[state] = state self.states.append(state) return True, state def get_state( self, locations: OrderedFrozenSet[str], delayed_actions: OrderedFrozenSet[DelayedAction] = OrderedFrozenSet(), arguments: int = 0 ) -> StateAndTransitions: """Like new_state(), but only returns the state without returning whether it is newly created or not.""" _, state = self.new_state(locations, delayed_actions, arguments) return state def remove_state(self, s: StateId, maybe_unreachable_set: OrderedSet[StateId]) -> None: state = self.states[s] self.clear_edges(state, maybe_unreachable_set) del self.state_cache[state] # "type: ignore" because the type annotation on `states` doesn't allow # entries to be `None`. self.states[s] = None # type: ignore def add_edge( self, src: StateAndTransitions, term: Term, dest: StateId ) -> None: assert term not in src assert dest < len(self.states) if isinstance(term, Action): src.epsilon.append((term, dest)) elif isinstance(term, Nt): src.nonterminals[term] = dest elif isinstance(term, ErrorSymbol): src.errors[term] = dest else: src.terminals[term] = dest self.states[dest].backedges.add(Edge(src.index, term)) def remove_backedge( self, src: StateAndTransitions, term: Term, dest: StateId, maybe_unreachable_set: OrderedSet[StateId] ) -> None: self.states[dest].backedges.remove(Edge(src.index, term)) maybe_unreachable_set.add(dest) def replace_edge( self, src: StateAndTransitions, term: Term, dest: StateId, maybe_unreachable_set: OrderedSet[StateId] ) -> None: assert isinstance(dest, int) and dest < len(self.states) edge_existed = term in src if edge_existed: old_dest = src[term] self.remove_backedge(src, term, old_dest, maybe_unreachable_set) if isinstance(term, Action): src.epsilon = [(t, d) for t, d in src.epsilon if t != term] src.epsilon.append((term, dest)) elif isinstance(term, Nt): src.nonterminals[term] = dest elif isinstance(term, ErrorSymbol): src.errors[term] = dest else: src.terminals[term] = dest self.states[dest].backedges.add(Edge(src.index, term)) self.assert_state_invariants(src) self.assert_state_invariants(dest) if edge_existed: self.assert_state_invariants(old_dest) def remove_edge( self, src: StateAndTransitions, term: Term, maybe_unreachable_set: OrderedSet[StateId] ) -> None: edge_existed = term in src if edge_existed: old_dest = src[term] self.remove_backedge(src, term, old_dest, maybe_unreachable_set) if isinstance(term, Action): src.epsilon = [(t, d) for t, d in src.epsilon if t != term] elif isinstance(term, Nt): del src.nonterminals[term] elif isinstance(term, ErrorSymbol): del src.errors[term] else: del src.terminals[term] self.assert_state_invariants(src) if edge_existed: self.assert_state_invariants(old_dest) def clear_edges( self, src: StateAndTransitions, maybe_unreachable_set: OrderedSet[StateId] ) -> None: """Remove all existing edges, in order to replace them by new one. This is used when resolving shift-reduce conflicts.""" assert isinstance(src, StateAndTransitions) old_dest = [] for term, dest in src.edges(): self.remove_backedge(src, term, dest, maybe_unreachable_set) old_dest.append(dest) src.terminals = {} src.nonterminals = {} src.errors = {} src.epsilon = [] self.assert_state_invariants(src) for dest in old_dest: self.assert_state_invariants(dest) def assert_table_invariants(self) -> None: for s in self.states: if s is not None: self.assert_state_invariants(s) def assert_state_invariants(self, src: typing.Union[StateId, StateAndTransitions]) -> None: if not self.debug_info: return if isinstance(src, int): src = self.states[src] assert isinstance(src, StateAndTransitions) try: for term, dest in src.edges(): assert Edge(src.index, term) in self.states[dest].backedges for e in src.backedges: assert e.term is not None assert self.states[e.src][e.term] == src.index if not self.assume_inconsistent: assert not src.is_inconsistent() except AssertionError as exc: print("assert_state_inveriants for {}\n".format(src)) for e in src.backedges: print("backedge {} from {}\n".format(e, self.states[e.src])) raise exc def remove_unreachable_states( self, maybe_unreachable_set: OrderedSet[StateId] ) -> None: # TODO: This function is incomplete in case of loops, some cycle might # remain isolated while not being reachable from the init states. We # should maintain a notion of depth per-state, such that we can # identify loops by noticing the all backedges have a larger depth than # the current state. init: OrderedSet[StateId] init = OrderedSet(goal for name, goal in self.named_goals) while maybe_unreachable_set: next_set: OrderedSet[StateId] = OrderedSet() for s in maybe_unreachable_set: # Check if the state is reachable, if not remove the state and # fill the next_set with all outgoing edges. if len(self.states[s].backedges) == 0 and s not in init: self.remove_state(s, next_set) maybe_unreachable_set = next_set def is_reachable_state(self, s: StateId) -> bool: """Check whether the current state is reachable or not.""" if self.states[s] is None: return False reachable_back: OrderedSet[StateId] = OrderedSet() todo = [s] while todo: s = todo.pop() reachable_back.add(s) for edge in self.states[s].backedges: if edge.src not in reachable_back: todo.append(edge.src) for _, s in self.named_goals: if s in reachable_back: return True return False def debug_dump(self) -> None: # Sort the grammar by state hash, such that it can be compared # before/after grammar modifications. temp = [s for s in self.states if s is not None] temp = sorted(temp, key=lambda s: s.stable_hash) for s in temp: print(s.stable_str(self.states)) def create_lr0_table( self, grammar: CanonicalGrammar, verbose: bool, progress: bool ) -> None: if verbose or progress: print("Create LR(0) parse table.") goals = grammar.grammar.goals() self.named_goals = [] # Temporary work queue. todo: typing.Deque[typing.Tuple[LR0Generator, StateAndTransitions]] todo = collections.deque() # Record the starting goals in the todo list. for nt in goals: init_nt = Nt(InitNt(nt), ()) it = LR0Generator.start(grammar, init_nt) s = self.get_state(it.stable_locations()) todo.append((it, s)) self.named_goals.append((nt, s.index)) # Iterate the grammar with sets of LR Items abstracted by the # LR0Generator, and create new states in the parse table as long as new # sets of LR Items are discovered. def visit_grammar() -> typing.Iterator[None]: while todo: yield # progress bar. # TODO: Compare stack / queue, for the traversal of the states. s_it, s = todo.popleft() if verbose: print("\nMapping state {} to LR0:\n{}".format(s.stable_hash, s_it)) for k, sk_it in s_it.transitions().items(): locations = sk_it.stable_locations() if not self.term_is_shifted(k): locations = OrderedFrozenSet() is_new, sk = self.new_state(locations) if is_new: todo.append((sk_it, sk)) # Add the edge from s to sk with k. self.add_edge(s, k, sk.index) consume(visit_grammar(), progress) if verbose: print("Create LR(0) Table Result:") self.debug_dump() def term_is_shifted(self, term: typing.Optional[Term]) -> bool: return not isinstance(term, Action) or term.follow_edge() def is_valid_path( self, path: typing.Sequence[Edge], state: typing.Optional[StateId] = None ) -> bool: """This function is used to check a list of edges and returns whether it corresponds to a valid path within the parse table. This is useful when merging sequences of edges from various locations.""" if not state and path != []: state = path[0].src while path: edge = path[0] path = path[1:] if state != edge.src: return False assert isinstance(state, StateId) term = edge.term if term is None and len(path) == 0: return True row = self.states[state] if term not in row: return False assert term is not None state = row[term] return True def term_is_stacked(self, term: typing.Optional[Term]) -> bool: # The `term` argument is annotated as Optional because `Edge.term` is a # common argument. If it's ever None in practice, the caller has a bug. assert term is not None return not isinstance(term, Action) def aps_visitor(self, aps: APS, visit: typing.Callable[[APS], bool]) -> None: """Visit all the states of the parse table, as-if we were running a Generalized LR parser. However, instead parsing content, we use this algorithm to generate both the content which remains to be parsed as well as the context which might lead us to be in the state which from which we started. This algorithm takes an APS (Abstract Parser State) and a callback, and consider all edges of the parse table, unless restricted by one of the previously encountered actions. These restrictions, such as replayed lookahead or the path which might be reduced are used for filtering out states which are not handled by this parse table. For each edge, this functions calls the visit functions to know whether to stop or continue. The visit function might capture APS given as argument to be used for other analysis. """ todo = [aps] while todo: aps = todo.pop() cont = visit(aps) if not cont: continue todo.extend(aps.shift_next(self)) def context_lanes(self, state: StateId) -> typing.Tuple[bool, typing.List[APS]]: """Compute lanes, such that each reduce action can have set of unique stack to reach the given state. The stacks are assumed to be loop-free by reducing edges at most once. In order to avoid attempting to eagerly solve everything using context information, we break this loop as soon as we have one token of lookahead in a case which does not have enough context. The return value is a tuple where the first element is a boolean which is True if we should fallback on solving this issue with more lookahead, and the second is the list of APS lanes which are providing enough context to disambiguate the inconsistency of the given state.""" def not_interesting(aps: APS) -> bool: reduce_list = [e for e in aps.history if self.term_is_shifted(e.term)] has_reduce_loop = len(reduce_list) != len(set(reduce_list)) return has_reduce_loop # The context is a dictionary which maps all stack suffixes from an APS # stack. It is mapped to a list of tuples, where the each tuple is the # index with the APS stack and the APS action used to follow this path. context: typing.DefaultDict[typing.Tuple[Edge, ...], typing.List[Edge]] context = collections.defaultdict(lambda: []) def has_enough_context(aps: APS) -> bool: try: assert aps.history[0] in context[tuple(aps.stack)] # Check the number of different actions which can reach this # location. If there is more than 1, then we do not have enough # context. return len(set(context[tuple(aps.stack)])) <= 1 except IndexError: return False collect = [APS.start(state)] enough_context = False while not enough_context: # print("collect.len = {}".format(len(collect))) # Fill the context dictionary with all the sub-stack which might be # encountered by other APS. recurse = [] context = collections.defaultdict(lambda: []) while collect: aps = collect.pop() recurse.append(aps) if aps.history == []: continue for i in range(len(aps.stack)): context[tuple(aps.stack[i:])].append(aps.history[0]) assert collect == [] # print("recurse.len = {}".format(len(recurse))) # Iterate over APS which do not yet have enough context information # to uniquely identify a single action. enough_context = True while recurse: aps = recurse.pop() if not_interesting(aps): # print("discard uninteresting context lane:") # print(aps.string("\tcontext")) continue if has_enough_context(aps): collect.append(aps) continue # If we have not enough context but some lookahead is # available, attempt to first solve this issue using more # lookahead before attempting to use context information. if len(aps.lookahead) >= 1: # print("discard context_lanes due to lookahead:") # for aps in itertools.chain(collect, recurse, [aps]): # print(aps.string("\tcontext")) return True, [] enough_context = False # print("extend starting at:\n{}".format(aps.string("\tcontext"))) collect.extend(aps.shift_next(self)) assert recurse == [] # print("context_lanes:") # for aps in collect: # print(aps.string("\tcontext")) return False, collect def lookahead_lanes(self, state: StateId) -> typing.List[APS]: """Compute lanes to collect all lookahead symbols available. After each reduce action, there is no need to consider the same non-terminal multiple times, we are only interested in lookahead token and not in the context information provided by reducing action.""" record = [] # After the first reduce action, we do not want to spend too much # resource visiting edges which would give us the same information. # Therefore, if we already reduce an action to a given state, then we # skip looking for lookahead that we already visited. # # Set of (first-reduce-edge, reducing-base, last-reduce-edge) seen_edge_after_reduce: typing.Set[typing.Tuple[Edge, StateId, typing.Optional[Term]]] seen_edge_after_reduce = set() def find_first_reduce( edges: Path ) -> typing.Tuple[int, typing.Optional[Edge]]: for i, edge in enumerate(edges): if not self.term_is_shifted(edge.term): return i, edge return 0, None def find_last_reduce( edges: Path ) -> typing.Tuple[int, typing.Optional[Edge]]: for i, edge in zip(reversed(range(len(edges))), reversed(edges)): if not self.term_is_shifted(edge.term): return i, edge return 0, None def visit(aps: APS) -> bool: # Note, this suppose that we are not considering flags when # computing, as flag might prevent some lookahead investigations. reduce_key = None first_index, first_reduce = find_first_reduce(aps.history) last_index, last_reduce = find_last_reduce(aps.history) if first_index != last_index and first_reduce and last_reduce: if not isinstance(aps.history[-1].term, Action): reduce_key = (first_reduce, aps.shift[0].src, last_reduce.term) has_seen_edge_after_reduce = reduce_key and reduce_key in seen_edge_after_reduce has_lookahead = len(aps.lookahead) >= 1 stop = has_seen_edge_after_reduce or has_lookahead # print("stop: {}, size lookahead: {}, seen_edge_after_reduce: {}".format( # stop, len(aps.lookahead), repr(reduce_key) # )) # print(aps.string("\tvisitor")) if stop: if has_lookahead: record.append(aps) if reduce_key: seen_edge_after_reduce.add(reduce_key) return not stop self.aps_visitor(APS.start(state), visit) return record def fix_with_context(self, s: StateId, aps_lanes: typing.List[APS]) -> None: raise ValueError("fix_with_context: Not Implemented") # # This strategy is about using context information. By using chains of # # reduce actions, we are able to increase the knowledge of the stack # # content. The stack content is the context which can be used to # # determine how to consider a reduction. The stack content is also # # called a lane, as defined in the Lane Table algorithm. # # # # To add context information to the current graph, we add flags # # manipulation actions. # # # # Consider each edge as having an implicit function which can map one # # flag value to another. The following implements a unification # # algorithm which is attempting to solve the question of what is the # # flag value, and where it should be changed. # # # # NOTE: (nbp) I would not be surprised if there is a more specialized # # algorithm, but I failed to find one so far, and this problem # # definitely looks like a unification problem. # Id = collections.namedtuple("Id", "edge") # Eq = collections.namedtuple("Eq", "flag_in edge flag_out") # Var = collections.namedtuple("Var", "n") # SubSt = collections.namedtuple("SubSt", "var by") # # # Unify expression, and return one substitution if both expressions # # can be unified. # def unify_expr(expr1, expr2, swapped=False): # if isinstance(expr1, Eq) and isinstance(expr2, Id): # if expr1.edge != expr2.edge: # # Different edges are ok, but produce no substituions. # return True # if isinstance(expr1.flag_in, Var): # return SubSt(expr1.flag_in, expr1.flag_out) # if isinstance(expr1.flag_out, Var): # return SubSt(expr1.flag_out, expr1.flag_in) # # We are unifying with a relation which consider the current # # function as an identity function. Having different values as # # input and output fails the unification rule. # return expr1.flag_out == expr1.flag_in # if isinstance(expr1, Eq) and isinstance(expr2, Eq): # if expr1.edge != expr2.edge: # # Different edges are ok, but produce no substituions. # return True # if expr1.flag_in is None and isinstance(expr2.flag_in, Var): # return SubSt(expr2.flag_in, None) # if expr1.flag_out is None and isinstance(expr2.flag_out, Var): # return SubSt(expr2.flag_out, None) # if expr1.flag_in == expr2.flag_in: # if isinstance(expr1.flag_out, Var): # return SubSt(expr1.flag_out, expr2.flag_out) # elif isinstance(expr2.flag_out, Var): # return SubSt(expr2.flag_out, expr1.flag_out) # # Reject solutions which are not deterministic. We do not # # want the same input flag to have multiple outputs. # return expr1.flag_out == expr2.flag_out # if expr1.flag_out == expr2.flag_out: # if isinstance(expr1.flag_in, Var): # return SubSt(expr1.flag_in, expr2.flag_in) # elif isinstance(expr2.flag_in, Var): # return SubSt(expr2.flag_in, expr1.flag_in) # return True # if not swapped: # return unify_expr(expr2, expr1, True) # return True # # # Apply substituion rule to an expression. # def subst_expr(subst, expr): # if expr == subst.var: # return True, subst.by # if isinstance(expr, Eq): # subst1, flag_in = subst_expr(subst, expr.flag_in) # subst2, flag_out = subst_expr(subst, expr.flag_out) # return subst1 or subst2, Eq(flag_in, expr.edge, flag_out) # return False, expr # # # Add an expression to an existing knowledge based which is relying on # # a set of free variables. # def unify_with(expr, knowledge, free_vars): # old_knowledge = knowledge # old_free_Vars = free_vars # while True: # subst = None # for rel in knowledge: # subst = unify_expr(rel, expr) # if subst is False: # raise Error("Failed to find a coherent solution") # if subst is True: # continue # break # else: # return knowledge + [expr], free_vars # free_vars = [fv for fv in free_vars if fv != subst.var] # # Substitue variables, and re-add rules which have substituted # # vars to check the changes to other rules, in case 2 rules are # # now in conflict or in case we can propagate more variable # # changes. # subst_rules = [subst_expr(subst, k) for k in knowledge] # knowledge = [rule for changed, rule in subst_rule if not changed] # for changed, rule in subst_rule: # if not changed: # continue # knowledge, free_vars = unify_with(rule, knowledge, free_vars) # # # Register boundary conditions as part of the knowledge based, i-e that # # reduce actions are expecting to see the flag value matching the # # reduced non-terminal, and that we have no flag value at the start of # # every lane head. # # # # TODO: Catch exceptions from the unify function in case we do not yet # # have enough context to disambiguate. # rules = [] # free_vars = [] # last_free = 0 # maybe_id_edges = set() # nts = set() # for aps in aps_lanes: # assert len(aps.stack) >= 1 # flag_in = None # for edge in aps.stack[-1]: # i = last_free # last_free += 1 # free_vars.append(Var(i)) # rule = Eq(flag_in, edge, Var(i)) # rules, free_vars = unify_with(rule, rules, free_vars) # flag_in = Var(i) # if flag_in is not None: # maybe_id_edges.add(Id(edge)) # edge = aps.stack[-1] # nt = edge.term.update_stack_with().nt # rule = Eq(nt, edge, None) # rules, free_vars = unify_with(rule, rules, free_vars) # nts.add(nt) # # # We want to produce a parse table where most of the node are ignoring # # the content of the flag which is being added. Thus we want to find a # # solution where most edges are the identical function. # def fill_with_id_functions(rules, free_vars, maybe_id_edges): # min_rules, min_vars = rules, free_vars # for num_id_edges in reversed(range(len(maybe_id_edges))): # for id_edges in itertools.combinations(edges, num_id_edges): # for edge in id_edges: # new_rules, new_free_vars = unify_with(rule, rules, free_vars) # if new_free_vars == []: # return new_rules, new_free_vars # if len(new_free_vars) < len(min_free_vars): # min_vars = new_free_vars # min_rules = new_rules # return rules, free_vars # # rules, free_vars = fill_with_id_functions(rules, free_vars, maybe_id_edges) # if free_vars != []: # raise Error("Hum … maybe we can try to iterate over the remaining free-variable.") # print("debug: Great we found a solution for a reduce-reduce conflict") # # # The set of rules describe the function that each edge is expected to # # support. If there is an Id(edge), then we know that we do not have to # # change the graph for the given edge. If the rule is Eq(A, edge, B), # # then we have to (filter A & pop) and push B, except if A or B is # # None. # # # # For each edge, collect the set of rules concerning the edge to # # determine which edges have to be transformed to add the filter&pop # # and push actions. # edge_rules = collections.defaultdict(lambda: []) # for rule in rules: # if isinstance(rule, Id): # edge_rules[rule.edge] = None # elif isinstance(rule, Eq): # if edge_rules[rule.edge] is not None: # edge_rules[rule.edge].append(rule) # # maybe_unreachable_set = set() # flag_name = self.get_flag_for(nts) # for edge, rules in edge_rules.items(): # # If the edge is an identity function, then skip doing any # # modifications on it. # if rules is None: # continue # # Otherwise, create a new state and transition for each mapping. # src = self.states[edge.src] # dest = src[edge.term] # dest_state = self.states[dest] # # TODO: Add some information to avoid having identical hashes as # # the destination. # actions = [] # for rule in OrderedFrozenSet(rules): # assert isinstance(rule, Eq) # seq = [] # if rule.flag_in is not None: # seq.append(FilterFlag(flag_name, True)) # if rule.flag_in != rule.flag_out: # seq.append(PopFlag(flag_name)) # if rule.flag_out is not None and rule.flag_in != rule.flag_out: # seq.append(PushFlag(flag_name, rule.flag_out)) # actions.append(Seq(seq)) # # Assert that we do not map flag_in more than once. # assert len(set(eq.flag_in for eq in rules)) < len(rules) # # Create the new state and add edges. # is_new, switch = self.new_state(dest.locations, OrderedFrozenSet(actions)) # assert is_new # for seq in actions: # self.add_edge(switch, seq, dest) # # # Replace the edge from src to dest, by an edge from src to the # # newly created switch state, which then decide which flag to set # # before going to the destination target. # self.replace_edge(src, edge.term, switch, maybe_unreachable_set) # # self.remove_unreachable_states(maybe_unreachable_set) # pass def fix_with_lookahead(self, s: StateId, aps_lanes: typing.List[APS]) -> None: # Find the list of terminals following each actions (even reduce # actions). assert all(len(aps.lookahead) >= 1 for aps in aps_lanes) if self.debug_info: for aps in aps_lanes: print(str(aps)) maybe_unreachable_set: OrderedSet[StateId] = OrderedSet() # For each shifted term, associate a set of state and actions which # would have to be executed. shift_map: typing.DefaultDict[ Term, typing.List[typing.Tuple[StateAndTransitions, typing.List[Edge]]] ] shift_map = collections.defaultdict(lambda: []) for aps in aps_lanes: actions = aps.history assert isinstance(actions[-1], Edge) src = actions[-1].src term = actions[-1].term assert term == aps.lookahead[0] assert isinstance(term, (str, End, ErrorSymbol, Nt)) # No need to consider any action beyind the first reduced action # since the reduced action is in charge of replaying the lookahead # terms. actions = list(keep_until(actions[:-1], lambda edge: not self.term_is_shifted(edge.term))) assert all(isinstance(edge.term, Action) for edge in actions) # Change the order of the shifted term, shift all actions by 1 with # the given lookahead term, in order to match the newly generated # state machine. # # Shifting actions with the list of shifted terms is used to record # the number of terms to be replayed, as well as verifying whether # Lookahead filter actions should accept or reject this lane. new_actions = [] accept = True for edge in actions: edge_term = edge.term assert isinstance(edge_term, Action) new_term = edge_term.shifted_action(term) if isinstance(new_term, bool): if new_term is False: accept = False break else: continue new_actions.append(Edge(edge.src, new_term)) if accept: target_id = self.states[src][term] target = self.states[target_id] shift_map[term].append((target, new_actions)) # Restore the new state machine based on a given state to use as a base # and the shift_map corresponding to edges. def restore_edges( state: StateAndTransitions, shift_map: typing.DefaultDict[ Term, typing.List[typing.Tuple[StateAndTransitions, typing.List[Edge]]] ], depth: str ) -> None: # print("{}starting with {}\n".format(depth, state)) edges = {} for term, actions_list in shift_map.items(): # print("{}term: {}, lists: {}\n".format(depth, repr(term), repr(actions_list))) # Collect all the states reachable after shifting the term. # Compute the unique name, based on the locations and actions # which are delayed. locations: OrderedSet[str] = OrderedSet() delayed: OrderedSet[DelayedAction] = OrderedSet() new_shift_map: typing.DefaultDict[ Term, typing.List[typing.Tuple[StateAndTransitions, typing.List[Edge]]] ] new_shift_map = collections.defaultdict(lambda: []) recurse = False if not self.term_is_shifted(term): # There is no more target after a reduce action. actions_list = [] for target, actions in actions_list: assert isinstance(target, StateAndTransitions) locations |= target.locations delayed |= target.delayed_actions if actions != []: # Pull edges, with delayed actions. edge = actions[0] assert isinstance(edge, Edge) for action in actions: action_term = action.term assert isinstance(action_term, Action) delayed.add(action_term) edge_term = edge.term assert edge_term is not None new_shift_map[edge_term].append((target, actions[1:])) recurse = True else: # Pull edges, as a copy of existing edges. for next_term, next_dest_id in target.edges(): next_dest = self.states[next_dest_id] new_shift_map[next_term].append((next_dest, [])) is_new, new_target = self.new_state( OrderedFrozenSet(locations), OrderedFrozenSet(delayed)) edges[term] = new_target.index if self.debug_info: print("{}is_new = {}, index = {}".format(depth, is_new, new_target.index)) print("{}Add: {} -- {} --> {}".format(depth, state.index, str(term), new_target.index)) print("{}continue: (is_new: {}) or (recurse: {})".format(depth, is_new, recurse)) if is_new or recurse: restore_edges(new_target, new_shift_map, depth + " ") self.clear_edges(state, maybe_unreachable_set) for term, target_id in edges.items(): self.add_edge(state, term, target_id) if self.debug_info: print("{}replaced by {}\n".format(depth, state)) state = self.states[s] restore_edges(state, shift_map, "") self.remove_unreachable_states(maybe_unreachable_set) def fix_inconsistent_state(self, s: StateId, verbose: bool) -> bool: # Fix inconsistent states works one state at a time. The goal is to # achieve the same method as the Lane Tracer, but instead of building a # table to then mutate the parse state, we mutate the parse state # directly. # # This strategy is simpler, and should be able to reproduce the same # graph mutations as seen with Lane Table algorithm. One of the problem # with the Lane Table algorithm is that it assume reduce operations, # and as such it does not apply simply to epsilon transitions which are # used as conditions on the parse table. # # By using push-flag and filter-flag actions, we are capable to # decompose the Lane Table transformation of the parse table into # multiple steps which are working one step at a time, and with less # table state duplication. state = self.states[s] if state is None or not state.is_inconsistent(): return False all_reduce = all(a.update_stack() for a, _ in state.epsilon) any_shift = (len(state.terminals) + len(state.nonterminals) + len(state.errors)) > 0 try_with_context = all_reduce and not any_shift try_with_lookahead = not try_with_context # if verbose: # print(aps_lanes_str(aps_lanes, "fix_inconsistent_state:", "\taps")) if try_with_context: if verbose: print("\tFix with context.") try_with_lookahead, aps_lanes = self.context_lanes(s) if not try_with_lookahead: assert aps_lanes != [] self.fix_with_context(s, aps_lanes) elif verbose: print("\tFallback on fixing with lookahead.") if try_with_lookahead: if verbose: print("\tFix with lookahead.") aps_lanes = self.lookahead_lanes(s) assert aps_lanes != [] self.fix_with_lookahead(s, aps_lanes) return True def fix_inconsistent_table(self, verbose: bool, progress: bool) -> None: """The parse table might be inconsistent. We fix the parse table by looking around the inconsistent states for more context. Either by looking at the potential stack state which might lead to the inconsistent state, or by increasing the lookahead.""" self.assume_inconsistent = True if verbose or progress: print("Fix parse table inconsistencies.") todo: typing.Deque[StateId] = collections.deque() for state in self.states: if state.is_inconsistent(): todo.append(state.index) if verbose and todo: print("\n".join([ "\nGrammar is inconsistent.", "\tNumber of States = {}", "\tNumber of inconsistencies found = {}"]).format( len(self.states), len(todo))) count = 0 def visit_table() -> typing.Iterator[None]: nonlocal count unreachable = [] while todo: while todo: yield # progress bar. # TODO: Compare stack / queue, for the traversal of the states. s = todo.popleft() if not self.is_reachable_state(s): # NOTE: We do not fix unreachable states, as we might # not be able to compute the reduce actions. However, # we should not clean edges not backedges as the state # might become reachable later on, since states are # shared if they have the same locations. unreachable.append(s) continue assert self.states[s].is_inconsistent() start_len = len(self.states) if verbose: count = count + 1 print("Fixing state {}\n".format(self.states[s].stable_str(self.states))) try: self.fix_inconsistent_state(s, verbose) except Exception as exc: self.debug_info = True raise ValueError( "Error while fixing conflict in state {}\n\n" "In the following grammar productions:\n{}" .format(self.states[s].stable_str(self.states), self.debug_context(s, "\n", "\t")) ) from exc new_inconsistent_states = [ s.index for s in self.states[start_len:] if s.is_inconsistent() ] if verbose: print("\tAdding {} states".format(len(self.states[start_len:]))) print("\tWith {} inconsistent states".format(len(new_inconsistent_states))) todo.extend(new_inconsistent_states) # Check whether none of the previously inconsistent and # unreahable state became reachable. If so add it back to the # todo list. still_unreachable = [] for s in unreachable: if self.is_reachable_state(s): todo.append(s) else: still_unreachable.append(s) unreachable = still_unreachable consume(visit_table(), progress) if verbose: print("\n".join([ "\nGrammar is now consistent.", "\tNumber of States = {}", "\tNumber of inconsistencies solved = {}"]).format( len(self.states), count)) assert not self.is_inconsistent() self.assume_inconsistent = False if verbose: print("Fix Inconsistent Table Result:") self.debug_dump() def remove_all_unreachable_state(self, verbose: bool, progress: bool) -> None: self.states = [s for s in self.states if s is not None] self.rewrite_reordered_state_indexes() def lower_reduce_actions(self, verbose: bool, progress: bool) -> None: # Remove Reduce actions and replace them by the programmatic # equivalent. # # This transformation preserves the stack manipulations of the parse # table. It only changes it from being implicitly executed by the LR # parser, to being explicitly executed with actions. # # This transformation converts the hard-to-predict load of the shift # table into a branch prediction which is potentially easier to # predict. # # A side-effect of this transformation is that it removes the need for # replaying non-terminals, thus the backends could safely ignore the # ability of the shift function from handling non-terminals. if verbose or progress: print("Lower Reduce actions.") maybe_unreachable_set: OrderedSet[StateId] = OrderedSet() def transform() -> typing.Iterator[None]: for s in self.states: term, _ = next(iter(s.epsilon), (None, None)) if self.term_is_shifted(term): continue assert len(s.epsilon) == 1 yield # progress bar. reduce_state = s if verbose: print("Inlining shift-operation for state {}".format(str(reduce_state))) # The reduced_aps should contain all reduced path of the single # Reduce action which is present on this state. However, as # state of the graph are shared, some reduced paths might follow # the same path and reach the same state. # # This code collect for each replayed path, the tops of the # stack on top of which these states are replayed. aps = APS.start(s.index) states_by_replay_term = collections.defaultdict(list) # print("Start:\n{}".format(aps.string(name="\titer_aps"))) # print(s.stable_str(self.states)) for reduced_aps in aps.shift_next(self): # As long as we have elements to replay, we should only # have a single path for each reduced path. If the next # state contains an action, then we stop here. iter_aps = reduced_aps next_is_action = self.states[iter_aps.state].epsilon != [] has_replay = iter_aps.replay != [] assert next_is_action is False and has_replay is True while (not next_is_action) and has_replay: # print("Step {}:\n{}".format(len(iter_aps.history), # iter_aps.string(name="\titer_aps"))) next_aps = list(iter_aps.shift_next(self)) if len(next_aps) == 0: # Note, this might happen as we are adding # lookahead tokens from any successor, we might not # always have a way to replay all tokens, in such # case an error should be produced, but in the mean # time, let's use the shift function as usual. break assert len(next_aps) == 1 iter_aps = next_aps[0] next_is_action = self.states[iter_aps.state].epsilon != [] has_replay = iter_aps.replay != [] # print("End at {}:\n{}".format(len(iter_aps.history), # iter_aps.string(name="\titer_aps"))) replay_list = [e.src for e in iter_aps.shift] assert len(replay_list) >= 2 replay_term = Replay(replay_list[1:]) states_by_replay_term[replay_term].append(replay_list[0]) # Create FilterStates actions. filter_by_replay_term = { replay_term: FilterStates(states) for replay_term, states in states_by_replay_term.items() } # Convert the Reduce action to an Unwind action. reduce_term, _ = next(iter(s.epsilon)) if isinstance(reduce_term, Reduce): unwind_term: Action = reduce_term.unwind else: assert isinstance(reduce_term, Seq) assert isinstance(reduce_term.actions[-1], Reduce) unwind_term = Seq(list(reduce_term.actions[:-1]) + [reduce_term.actions[-1].unwind]) # Remove the old Reduce edge if still present. # print("Before:\n{}".format(reduce_state.stable_str(self.states))) self.remove_edge(reduce_state, reduce_term, maybe_unreachable_set) # Add Unwind action. # print("After:\n") locations = reduce_state.locations delayed: OrderedFrozenSet[DelayedAction] = OrderedFrozenSet(filter_by_replay_term.items()) replay_size = 1 # Replay the unwound non-terminal is_new, filter_state = self.new_state(locations, delayed, replay_size) self.add_edge(reduce_state, unwind_term, filter_state.index) if not is_new: # The destination state already exists. Assert that all # outgoing edges are matching what we would have generated. if len(filter_by_replay_term) == 1: # There is only one predecessor, no need for a # FilterState condition. replay_term = next(iter(filter_by_replay_term)) assert replay_term in filter_state continue for replay_term, filter_term in filter_by_replay_term.items(): assert filter_term in filter_state replay_state = self.states[filter_state[filter_term]] assert replay_term in replay_state continue if len(filter_by_replay_term) == 1: replay_term = next(iter(filter_by_replay_term)) dest_idx = replay_term.replay_steps[-1] # Do not add the FilterStates action, as there is only one. # Add Replay actions from the filter_state to the destination. self.add_edge(filter_state, replay_term, dest_idx) else: for replay_term, filter_term in filter_by_replay_term.items(): dest_idx = replay_term.replay_steps[-1] dest = self.states[dest_idx] # Add FilterStates action from the filter_state to the replay_state. locations = dest.locations delayed = OrderedFrozenSet(itertools.chain(dest.delayed_actions, [replay_term])) is_new, replay_state = self.new_state(locations, delayed, replay_size) self.add_edge(filter_state, filter_term, replay_state.index) assert (not is_new) == (replay_term in replay_state) # Add Replay actions from the replay_state to the destination. if is_new: dest_idx = replay_term.replay_steps[-1] self.add_edge(replay_state, replay_term, dest_idx) # print(replay_state.stable_str(self.states)) assert not replay_state.is_inconsistent() # print(filter_state.stable_str(self.states)) # print(reduce_state.stable_str(self.states)) assert not reduce_state.is_inconsistent() assert not filter_state.is_inconsistent() consume(transform(), progress) def fold_replay_unwind(self, verbose: bool, progress: bool) -> None: """Convert Replay action falling into Unwind action to an Unwind action which replay less terms.""" if verbose or progress: print("Fold Replay followed by Unwind actions.") maybe_unreachable_set: OrderedSet[StateId] = OrderedSet() def try_transform(s: StateAndTransitions) -> bool: if len(s.epsilon) != 1: return False replay_term, replay_dest_idx = next(iter(s.epsilon)) if not isinstance(replay_term, Replay): return False replay_dest = self.states[replay_dest_idx] if len(replay_dest.epsilon) != 1: return False unwind_term, unwind_dest_idx = next(iter(replay_dest.epsilon)) if not unwind_term.update_stack(): return False stack_diff = unwind_term.update_stack_with() if not stack_diff.reduce_stack(): return False if stack_diff.pop + stack_diff.replay <= 0: return False # Remove replayed terms from the Unwind action. replayed = replay_term.replay_steps unshifted = min(stack_diff.replay + min(s.arguments, stack_diff.pop), len(replayed)) if unshifted < len(replayed): # We do not have all replayed terms as arguments, thus do not # consume arguments unshifted = min(stack_diff.replay, len(replayed)) if unshifted == 0: return False new_unwind_term = unwind_term.unshift_action(unshifted) new_replay = new_unwind_term.update_stack_with().replay # Replace the replay_term and unwind_term by terms which are # avoiding extra replay actions. self.remove_edge(s, replay_term, maybe_unreachable_set) if len(replayed) == unshifted: # The Unwind action replay more terms than what we originally # had. The replay term is replaced by an Unwind edge instead. assert s.arguments >= -new_replay self.add_edge(s, new_unwind_term, unwind_dest_idx) else: # The Unwind action replay and pop less terms than what we # originally had. Thus the replay action is shortened and a new # state is created to accomodate the new Unwind action. assert unshifted >= 1 new_replay_term = Replay(replayed[:-unshifted]) implicit_replay_term = Replay(replayed[-unshifted:]) locations = replay_dest.locations delayed: OrderedFrozenSet[DelayedAction] delayed = OrderedFrozenSet( itertools.chain(replay_dest.delayed_actions, [implicit_replay_term])) is_new, unwind_state = self.new_state(locations, delayed) assert (not is_new) == (new_unwind_term in unwind_state) # Add new Replay and new Unwind actions. self.add_edge(s, new_replay_term, unwind_state.index) if is_new: assert unwind_state.arguments >= -new_replay self.add_edge(unwind_state, new_unwind_term, unwind_dest_idx) assert not unwind_state.is_inconsistent() assert not s.is_inconsistent() return True def transform() -> typing.Iterator[None]: for s in self.states: if try_transform(s): yield # progress bar consume(transform(), progress) self.remove_unreachable_states(maybe_unreachable_set) def fold_identical_endings(self, verbose: bool, progress: bool) -> None: # If 2 states have the same outgoing edges, then we can merge the 2 # states into a single state, and rewrite all the backedges leading to # these states to be replaced by edges going to the reference state. if verbose or progress: print("Fold identical endings.") def rewrite_backedges(state_list: typing.List[StateAndTransitions], state_map: typing.Dict[StateId, StateId], backrefs: typing.Dict[StateId, typing.List[typing.Tuple[StateId, Action, StateId]]], maybe_unreachable: OrderedSet[StateId]) -> bool: all_backrefs = [] new_backrefs = set() for s in state_list: all_backrefs.extend(backrefs[s.index]) # All states have the same outgoing edges. Thus we replace all of # them by a single state. We do that by replacing edges of which # are targeting the state in the state_list by edges targetting the # ref state. ref = state_list.pop() tmp_state_map = default_fwd_dict(state_map) for s in state_list: tmp_state_map[s.index] = ref.index for ref_s, ref_t, _d in all_backrefs: new_backrefs.add((ref_s, ref_t.rewrite_state_indexes(tmp_state_map))) if len(all_backrefs) != len(new_backrefs): # Skip this rewrite if when rewritting we are going to cause # some aliasing to happen between actions which are going to # different states. return False replace_edges = [e for s in state_list for e in s.backedges] hit = False for edge in replace_edges: edge_term = edge.term assert edge_term is not None src = self.states[edge.src] old_dest = src[edge_term] # print("replace {} -- {} --> {}, by {} -- {} --> {}" # .format(src.index, edge_term, src[edge_term], src.index, edge_term, ref.index)) self.replace_edge(src, edge_term, ref.index, maybe_unreachable) state_map[old_dest] = ref.index hit = True return hit def rewrite_if_same_outedges(state_list: typing.List[StateAndTransitions]) -> bool: maybe_unreachable: OrderedSet[StateId] = OrderedSet() backrefs = collections.defaultdict(list) outedges = collections.defaultdict(list) for s in state_list: # Iterate first over actions, then over ordinary states. self.assert_state_invariants(s) outedges[tuple(s.edges())].append(s) if s.epsilon == []: continue for t, d in s.edges(): if not isinstance(t, Action): continue for r in t.state_refs(): backrefs[r].append((s.index, t, d)) hit = False state_map: typing.Dict[StateId, StateId] = default_id_dict() for same in outedges.values(): if len(same) > 1: hit = rewrite_backedges(same, state_map, backrefs, maybe_unreachable) or hit if hit: self.remove_unreachable_states(maybe_unreachable) self.rewrite_state_indexes(state_map) self.remove_all_unreachable_state(verbose, progress) return hit def visit_table() -> typing.Iterator[None]: hit = True while hit: yield # progress bar. hit = rewrite_if_same_outedges(self.states) consume(visit_table(), progress) def group_epsilon_states(self, verbose: bool, progress: bool) -> None: def all_action_inedges(s: StateAndTransitions) -> bool: return all(isinstance(e.term, Action) for e in s.backedges) shift_states, action_states = split(self.states, lambda s: len(s.epsilon) == 0) from_act_action_states, from_shf_action_states = split(action_states, all_action_inedges) self.states = [] self.states.extend(shift_states) self.states.extend(from_shf_action_states) self.states.extend(from_act_action_states) self.rewrite_reordered_state_indexes() def group_nonterminal_states(self, verbose: bool, progress: bool) -> None: # This function is used to reduce the range of FilterStates values, # such that the Rust compiler can compile FilterStates match statements # to a table-switch. freq_count = collections.Counter(nt for s in self.states for nt in s.nonterminals) freq_nt, _ = zip(*freq_count.most_common()) def state_value(s: StateAndTransitions) -> float: value = 0.0 if len(s.epsilon) != 0: return 4.0 if len(s.nonterminals) == 0: return 2.0 i = 1.0 for nt in freq_nt: if nt in s: value += i i /= 2.0 return -value self.states.sort(key=state_value) self.rewrite_reordered_state_indexes() def count_shift_states(self) -> int: return sum(1 for s in self.states if s is not None and len(s.epsilon) == 0) def count_action_states(self) -> int: return sum(1 for s in self.states if s is not None and len(s.epsilon) > 0) def count_action_from_shift_states(self) -> int: def from_shift_states(s: StateAndTransitions) -> bool: return any(not isinstance(e.term, Action) for e in s.backedges) return sum(1 for s in self.states if len(s.epsilon) > 0 and from_shift_states(s)) def prepare_debug_context(self) -> DebugInfo: """To better filter out the traversal of the grammar in debug context, we pre-compute for each state the maximal depth of each state within a production. Therefore, if visiting a state no increases the reducing depth beyind the ability to shrink the shift list to 0, then we can stop going deeper, as we entered a different production. """ depths = collections.defaultdict(lambda: []) for s in self.states: if s is None or not s.epsilon: continue aps = APS.start(s.index) for aps_next in aps.shift_next(self): if not aps_next.reducing: continue for i, edge in enumerate(aps_next.stack): depths[edge.src].append(i + 1) return {s: max(ds) for s, ds in depths.items()} def debug_context( self, state: StateId, split_txt: str = "; ", prefix: str = "" ) -> str: """Reconstruct the grammar production by traversing the parse table.""" if self.debug_info is False: return "" if self.debug_info is True: self.debug_info = self.prepare_debug_context() debug_info = typing.cast(typing.Dict[StateId, int], self.debug_info) record = [] def visit(aps: APS) -> bool: # Stop after reducing once. if aps.history == []: return True last = aps.history[-1].term is_unwind = isinstance(last, Action) and last.update_stack() has_shift_loop = len(aps.shift) != 1 + len(set(zip(aps.shift, aps.shift[1:]))) can_reduce_later = True try: can_reduce_later = debug_info[aps.shift[-1].src] >= len(aps.shift) except KeyError: can_reduce_later = False stop = is_unwind or has_shift_loop or not can_reduce_later # Record state which are reducing at most all the shifted states. save = stop and len(aps.shift) == 1 save = save and is_unwind if save: assert isinstance(last, Action) save = last.update_stack_with().nt in self.states[aps.shift[0].src] if save: record.append(aps) return not stop self.aps_visitor(APS.start(state), visit) context: OrderedSet[str] = OrderedSet() for aps in record: assert aps.history != [] action = aps.history[-1].term assert isinstance(action, Action) assert action.update_stack() stack_diff = action.update_stack_with() replay = stack_diff.replay before = [repr(e.term) for e in aps.stack[:-1]] after = [repr(e.term) for e in aps.history[:-1]] prod = before + ["\N{MIDDLE DOT}"] + after if replay < len(after) and replay > 0: del prod[-replay:] replay = 0 if replay > len(after): replay += 1 if replay > 0: prod = prod[:-replay] + ["[lookahead:"] + prod[-replay:] + ["]"] txt = "{}{} ::= {}".format(prefix, repr(stack_diff.nt), " ".join(prod)) context.add(txt) if split_txt is None: return context return split_txt.join(txt for txt in sorted(context))