REFACT(NET): autocreated SOLUTION with overwrites & executed ...

+ ENH(NETOP): `compute()` return Solutions.
+ refact(net): use special-purpose methods to update Solution
  delegating to ChainMap.
+ drop(base): collect_overwrites consolidated into Solution.
+ DROP(jetsam): drop `executed` arg; now embeded in `solution`.
+ fix(doc): mention forgotten `solution` in jetsams.
+ doc(api): mention new class; add special meths.

Author: ankostis

docs/source/plotting.rst

@@ -87,7 +87,6 @@ with the folllowing properties, as a debug aid:
 ... except ValueError as ex:
 ...     pprint(ex.jetsam)
 {'args': {'args': [None], 'kwargs': {}},
- 'executed': set(),
  'network': Network(
     +--a
     +--FunctionalOperation(name='screamer', needs=['a'], provides=['foo'], fn='scream')
@@ -99,7 +98,7 @@ with the folllowing properties, as a debug aid:
  'provides': None,
  'results_fn': None,
  'results_op': None,
- 'solution':  ChainMap({'a': None})}
+ 'solution': {'a': None}}
 
 
 In interactive *REPL* console you may use this to get the last raised exception::
@@ -137,8 +136,9 @@ The following annotated attributes *might* have meaningfull value on an exceptio
 ``op_results``
     the results, always a dictionary, as matched with operation's `provides`
 
-``executed```
-    a set with the operation nodes & instructions executed till the error happened.
+``solution``
+    an instance of :class:`.Solution`, contains `inputs` & `outputs` till the error happened;
+    note that :attr:`.Solution.executed` contain the list of executed `operations` so far.
 
 Ofcourse you may use many of the above "jetsam" values when plotting.
 

docs/source/reference.rst

@@ -40,11 +40,16 @@ Module: `network`
 .. autoclass:: graphtik.network.Network
      :members:
      :private-members:
+     :special-members:
      :undoc-members:
 .. autoclass:: graphtik.network.ExecutionPlan
      :members:
      :private-members:
+     :special-members:
      :undoc-members:
+.. autoclass:: graphtik.network.Solution
+     :members:
+     :special-members:
 
 Module: `plot`
 ===============

graphtik/__init__.py

@@ -20,4 +20,3 @@
 )
 from .netop import compose
 from .op import operation
-from .base import collect_overwrites

graphtik/base.py

@@ -45,22 +45,6 @@ def astuple(i, argname, allowed_types=tuple):
     return i
 
 
-def collect_overwrites(maps) -> Mapping[Any, List]:
-    """
-    Collect items in the maps that exist more than once.
-
-    :return:
-        a dictionary with keys only those items that existed in more than one map,
-        an values, all those values, in the order of given `maps`
-    """
-    dd = defaultdict(list)
-    for d in maps:
-        for k, v in d.items():
-            dd[k].append(v)
-
-    return {k: v for k, v in dd.items() if len(v) > 1}
-
-
 def jetsam(ex, locs, *salvage_vars: str, annotation="jetsam", **salvage_mappings):
     """
     Annotate exception with salvaged values from locals() and raise!

graphtik/netop.py

@@ -4,15 +4,15 @@
 
 import logging
 import re
-from collections import ChainMap, abc
+from collections import abc
 from typing import Any, Callable, Mapping
 
 import networkx as nx
 from boltons.setutils import IndexedSet as iset
 
 from .base import Items, Plotter, aslist, astuple, jetsam
 from .modifiers import optional, sideffect
-from .network import Network, yield_ops
+from .network import Network, Solution, yield_ops
 from .op import FunctionalOperation, Operation, reparse_operation_data
 
 log = logging.getLogger(__name__)
@@ -161,7 +161,7 @@ def _build_pydot(self, **kws):
         plotter = self.last_plan or self.net
         return plotter._build_pydot(**kws)
 
-    def compute(self, named_inputs, outputs=None, solution: ChainMap = None) -> dict:
+    def compute(self, named_inputs, outputs=None) -> Solution:
         """
         Solve & execute the graph, sequentially or parallel.
 
@@ -176,18 +176,10 @@ def compute(self, named_inputs, outputs=None, solution: ChainMap = None) -> dict
             a string or a list of strings with all data asked to compute.
             If you set this variable to ``None``, all data nodes will be kept
             and returned at runtime.
-        :param solution:
-            If not None, it must be a :class:`collections.ChainMap`, which will
-            collect all results in a separate dictionary for each operation execution.
-            The 1st dictionary in its maplist will collect the inputs, but will endup
-            to the be last one when execution finishes.
-
-            See :term:`solution`
 
         :return:
-            the chained-map `solution` "compressed" as a plain dictionary;
-            if you you want to acccess all intermediate values provide your own
-            ``ChainMap`` instance in this method.
+            The :term:`solution` which contains the results of each operation executed
+            +1 for inputs in separate dictionaries.
 
         :raises ValueError:
             - If `outputs` asked do not exist in network, with msg:
@@ -209,23 +201,16 @@ def compute(self, named_inputs, outputs=None, solution: ChainMap = None) -> dict
         try:
             net = self.net  # jetsam
 
-            if solution is not None and not isinstance(solution, ChainMap):
-                raise ValueError(
-                    f"Solution was not ChainMap, but {type(solution)}!\n  solution; {solution}"
-                )
-
             # Build the execution plan.
             self.last_plan = plan = net.compile(named_inputs.keys(), outputs)
 
-            solution = plan.execute(
-                named_inputs, outputs, solution=solution, method=self.execution_method
-            )
+            solution = plan.execute(named_inputs, outputs, method=self.execution_method)
 
-            return dict(solution)  # Convert ChainMap --> plain dict.
+            return solution
         except Exception as ex:
             jetsam(ex, locals(), "plan", "solution", "outputs", network="net")
 
-    def __call__(self, **input_kwargs) -> dict:
+    def __call__(self, **input_kwargs) -> Solution:
         """
         Delegates to :meth:`compute()`, respecting any narrowed `outputs`.
         """

graphtik/network.py

@@ -63,10 +63,52 @@ def is_skip_evictions():
     return _execution_configs.get()["skip_evictions"]
 
 
-def _del_chain_key(amap: ChainMap, key):
-    log.debug("removing data '%s' from solution.", key)
-    for d in amap.maps:
-        d.pop(key, None)
+class Solution(ChainMap):
+    """Collects outputs from operations, preserving :term:`overwrites`."""
+
+    def __init__(self, plan, *args, **kw):
+        super().__init__(*args, **kw)
+        self.executed = iset()
+        self.finished = False
+        self.plan = plan
+
+    def __repr__(self):
+        items = ", ".join(f"{k!r}: {v!r}" for k, v in self.items())
+        return f"{{{items}}}"
+
+    def operation_executed(self, op, outputs):
+        """invoked once per operation, with its results"""
+        assert not self.finished, f"Cannot reuse solution: {self}"
+        self.maps.append(outputs)
+        self.executed.add(op)
+
+    def finish(self):
+        """invoked only once, after all ops have been executed"""
+        # Invert solution so that last value wins
+        if not self.finished:
+            self.maps = self.maps[::-1]
+            self.finised = True
+
+    def __delitem__(self, key):
+        log.debug("removing data '%s' from solution.", key)
+        for d in self.maps:
+            d.pop(key, None)
+
+    def overwrites(self) -> Mapping[Any, List]:
+        """
+        Collect items in the maps that exist more than once.
+
+        :return:
+            a dictionary with keys only those items that existed in more than one map,
+            an values, all those values, in the order of given `maps`
+        """
+        maps = self.maps
+        dd = defaultdict(list)
+        for d in maps:
+            for k, v in d.items():
+                dd[k].append(v)
+
+        return {k: v for k, v in dd.items() if len(v) > 1}
 
 
 class _DataNode(str):
@@ -241,14 +283,14 @@ def _call_operation(self, op, solution):
         try:
             return op.compute(solution)
         except Exception as ex:
-            jetsam(ex, locals(), plan="self")
+            jetsam(ex, locals(), "solution", plan="self")
         finally:
             # record execution time
             t_complete = round(time.time() - t0, 5)
             self.times[op.name] = t_complete
             log.debug("...step completion time: %s", t_complete)
 
-    def _execute_thread_pool_barrier_method(self, solution: ChainMap, executed):
+    def _execute_thread_pool_barrier_method(self, solution: Solution):
         """
         This method runs the graph using a parallel pool of thread executors.
         You may achieve lower total latency if your graph is sufficiently
@@ -263,7 +305,7 @@ def _execute_thread_pool_barrier_method(self, solution: ChainMap, executed):
         # scheduled, then schedule them onto a thread pool, then collect their
         # results onto a memory solution for use upon the next iteration.
         while True:
-            self._check_if_aborted(executed)
+            self._check_if_aborted(solution.executed)
 
             # the upnext list contains a list of operations for scheduling
             # in the current round of scheduling
@@ -273,14 +315,14 @@ def _execute_thread_pool_barrier_method(self, solution: ChainMap, executed):
                 #  based on what has already been executed.
                 if (
                     isinstance(node, Operation)
-                    and node not in executed
+                    and node not in solution.executed
                     #  Use `broken_dag` to allow executing operations from given inputs
                     #  regardless of whether their producers have yet to re-calc them.
                     and set(
                         n
                         for n in nx.ancestors(self.broken_dag, node)
                         if isinstance(n, Operation)
-                    ).issubset(executed)
+                    ).issubset(solution.executed)
                 ):
                     upnext.append(node)
                 elif isinstance(node, _EvictInstruction):
@@ -294,10 +336,12 @@ def _execute_thread_pool_barrier_method(self, solution: ChainMap, executed):
                             node not in self.dag.nodes
                             # Scan node's successors in `broken_dag`, not to block
                             # an op waiting for calced data already given as input.
-                            or set(self.broken_dag.successors(node)).issubset(executed)
+                            or set(self.broken_dag.successors(node)).issubset(
+                                solution.executed
+                            )
                         )
                     ):
-                        _del_chain_key(solution, node)
+                        del solution[node]
 
             # stop if no nodes left to schedule, exit out of the loop
             if not upnext:
@@ -308,36 +352,33 @@ def _execute_thread_pool_barrier_method(self, solution: ChainMap, executed):
             )
 
             for op, outputs in done_iterator:
-                solution.maps.append(outputs)
-                executed.add(op)
+                solution.operation_executed(op, outputs)
 
-    def _execute_sequential_method(self, solution: ChainMap, executed):
+    def _execute_sequential_method(self, solution: Solution):
         """
         This method runs the graph one operation at a time in a single thread
 
         :param solution:
             must contain the input values only, gets modified
         """
         for step in self.steps:
-            self._check_if_aborted(executed)
+            self._check_if_aborted(solution.executed)
 
             if isinstance(step, Operation):
                 log.debug("%sexecuting step: %s", "-" * 32, step.name)
 
                 outputs = self._call_operation(step, solution)
-                solution.maps.append(outputs)
-                executed.add(step)
+                solution.operation_executed(step, outputs)
+
             elif isinstance(step, _EvictInstruction):
                 # Cache value may be missing if it is optional.
                 if step in solution:
-                    _del_chain_key(solution, step)
+                    del solution[step]
 
             else:
                 raise AssertionError(f"Unrecognized instruction.{step}")
 
-    def execute(
-        self, named_inputs, outputs=None, *, solution: ChainMap = None, method=None
-    ):
+    def execute(self, named_inputs, outputs=None, *, method=None) -> Solution:
         """
         :param named_inputs:
             A maping of names --> values that must contain at least
@@ -347,16 +388,10 @@ def execute(
         :param outputs:
             If not None, they are just checked if possible, based on :attr:`provides`,
             and scream if not.
-        :param solution:
-            If not None, it must be a :class:`collections.ChainMap`, which will
-            collect all results in a separate dictionary for each operation execution.
-            The 1st dictionary in its maplist will collect the inputs, but will endup
-            to the be last one when execution finishes.
-
-            See :term:`solution`
 
         :return:
-            the populates `solution`
+            The :term:`solution` which contains the results of each operation executed
+            +1 for inputs in separate dictionaries.
 
         :raises ValueError:
             - If plan does not contain any operations, with msg:
@@ -382,25 +417,20 @@ def execute(
                 else self._execute_sequential_method
             )
 
-            if solution is None:
-                solution = ChainMap()
-            preload_layers = len(solution.maps)  # TODO: move to solution
-
             # If certain outputs asked, put relevant-only inputs in solution,
             # otherwise, keep'em all.
             #
             # Note: clone and keep original `inputs` in the 1st chained-map.
-            solution.update(
+            solution = Solution(
+                self,
                 {k: v for k, v in named_inputs.items() if k in self.dag.nodes}
                 if self.evict
-                else named_inputs
+                else named_inputs,
             )
-            executed = set()
-            executor(solution, executed)
-
-            # Invert solution so that last value wins
-            # TODO: move to solution
-            solution.maps = solution.maps[::-1]
+            try:
+                executor(solution)
+            finally:
+                solution.finish()
 
             # Validate eviction was perfect
             #
@@ -411,20 +441,9 @@ def execute(
                 or set(solution).issubset(self.provides)
             ), f"Evictions left more data{list(iset(solution) - set(self.provides))} than {self}!"
 
-            # Validate solution layers match operations executed + 1(inputs)
-            # TODO: move to solution
-            #
-            assert len(solution.maps) - preload_layers == sum(
-                1 for i in yield_ops(self.steps)
-            ), (
-                f"Solution layers({len(solution.maps)}, preloaded: {preload_layers}) mismatched "
-                f"operations executed({sum(1 for i in yield_ops(self.dag))})!"
-                f"\n  {self}\n  solution: {solution}"
-            )
-
             return solution
         except Exception as ex:
-            jetsam(ex, locals(), "solution", "executed")
+            jetsam(ex, locals(), "solution")
 
 
 class Network(Plotter):