WIP/REFACT(net): part 3 of new dag-solver & pin refactoring

+ ALL TCs OK after ~10 fails with #25, #25 bugs, BUT...
- WIP: STILL no PIN on PARALLEL.
+ move check if value in asked outputs before cache-evicting it
  in build-execution-plan time - compute methods
  don't need outputs anymore.
+ test: speed up parallel/multihtread TCs
  by reducing delays & repetitions.
+ refact: network rightfully adopted stray functions
  for parallel processing - they all worke on the net.graph,
+ upd: networkx api by indexing on `dag.nodes` views.
+ enh: add log message when deleting in parallel
  (in par with sequential code).
+ refact: var-renames, if-then-else simplifications, pythonisms.

Author: ankostis

graphkit/network.py

@@ -124,7 +124,7 @@ def add_op(self, operation):
         assert operation.provides is not None, "Operation's 'provides' must be named"
 
         # assert layer is only added once to graph
-        assert operation not in self.graph.nodes(), "Operation may only be added once"
+        assert operation not in self.graph.nodes, "Operation may only be added once"
 
         ## Invalidate old plans.
         self._cached_execution_plans = {}
@@ -152,7 +152,7 @@ def show_layers(self, debug=False, ret=False):
         else:
             print(s)
 
-    def _build_execution_plan(self, dag, inputs):
+    def _build_execution_plan(self, dag, inputs, outputs):
         """
         Create the list of operation-nodes & *instructions* evaluating all
 
@@ -161,6 +161,8 @@ def _build_execution_plan(self, dag, inputs):
 
         :param dag:
             The original dag, pruned; not broken.
+        :param outputs:
+            outp-names to decide whether to add (and which) del-instructions
 
         In the list :class:`DeleteInstructions` steps (DA) are inserted between
         operation nodes to reduce the memory footprint of cached results.
@@ -187,22 +189,29 @@ def _build_execution_plan(self, dag, inputs):
                     plan.append(PinInstruction(node))
 
             elif isinstance(node, Operation):
-
                 plan.append(node)
 
+                # Keep all values in cache if not specific outputs asked.
+                if not outputs:
+                    continue
+
                 # Add instructions to delete predecessors as possible.  A
                 # predecessor may be deleted if it is a data placeholder that
                 # is no longer needed by future Operations.
                 for need in self.graph.pred[node]:
                     if self._debug:
                         print("checking if node %s can be deleted" % need)
                     for future_node in ordered_nodes[i+1:]:
-                        if isinstance(future_node, Operation) and need in future_node.needs:
+                        if (
+                            isinstance(future_node, Operation)
+                            and need in future_node.needs
+                        ):
                             break
                     else:
-                        if self._debug:
-                            print("  adding delete instruction for %s" % need)
-                        plan.append(DeleteInstruction(need))
+                        if need not in outputs:
+                            if self._debug:
+                                print("  adding delete instruction for %s" % need)
+                            plan.append(DeleteInstruction(need))
 
             else:
                 raise AssertionError("Unrecognized network graph node %r" % node)
@@ -317,7 +326,7 @@ def _solve_dag(self, outputs, inputs):
         unsatisfied = self._collect_unsatisfied_operations(broken_dag, inputs)
         pruned_dag = dag.subgraph(broken_dag.nodes - unsatisfied)
 
-        plan = self._build_execution_plan(pruned_dag, inputs)
+        plan = self._build_execution_plan(pruned_dag, inputs, outputs)
 
         return plan
 
@@ -393,7 +402,7 @@ def compute(
                 cache, overwrites_collector, named_inputs)
         else:
             self._compute_sequential_method(
-                cache, overwrites_collector, named_inputs, outputs)
+                cache, overwrites_collector, named_inputs)
 
         if not outputs:
             # Return the whole cache as output, including input and
@@ -432,7 +441,7 @@ def _compute_thread_pool_barrier_method(
 
 
         # this keeps track of all nodes that have already executed
-        has_executed = set()  # unordered, not iterated
+        executed_nodes = set()  # unordered, not iterated
 
         # with each loop iteration, we determine a set of operations that can be
         # scheduled, then schedule them onto a thread pool, then collect their
@@ -443,20 +452,22 @@ def _compute_thread_pool_barrier_method(
             # in the current round of scheduling
             upnext = []
             for node in self.execution_plan:
-                # only delete if all successors for the data node have been executed
                 if isinstance(node, DeleteInstruction):
-                    if ready_to_delete_data_node(node,
-                                                 has_executed,
-                                                 self.graph):
-                        if node in cache:
-                            cache.pop(node)
-
-                # continue if this node is anything but an operation node
-                if not isinstance(node, Operation):
-                    continue
-
-                if ready_to_schedule_operation(node, has_executed, self.graph) \
-                        and node not in has_executed:
+                    # Only delete if all successors for the data node 
+                    # have been executed
+                    # Cache value for an optional may be missing.
+                    if (
+                        node in cache
+                        and self._can_delete_data_node(node, executed_nodes)
+                    ):
+                        if self._debug:
+                            print("removing data '%s' from cache." % node)
+                        del cache[node]
+                elif (
+                    isinstance(node, Operation)
+                    and self._can_schedule_operation(node, executed_nodes)
+                    and node not in executed_nodes
+                ):
                     upnext.append(node)
 
 
@@ -469,10 +480,10 @@ def _compute_thread_pool_barrier_method(
                                 upnext)
             for op, result in done_iterator:
                 cache.update(result)
-                has_executed.add(op)
+                executed_nodes.add(op)
 
 
-    def _compute_sequential_method(self, cache, overwrites, inputs, outputs):
+    def _compute_sequential_method(self, cache, overwrites, inputs):
         """
         This method runs the graph one operation at a time in a single thread
         """
@@ -500,18 +511,12 @@ def _compute_sequential_method(self, cache, overwrites, inputs, outputs):
                 if self._debug:
                     print("step completion time: %s" % t_complete)
 
-            # Process DeleteInstructions by deleting the corresponding data
-            # if possible.
             elif isinstance(step, DeleteInstruction):
-
-                if outputs and step not in outputs:
-                    # Some DeleteInstruction steps may not exist in the cache
-                    # if they come from optional() needs that are not privoded
-                    # as inputs.  Make sure the step exists before deleting.
-                    if step in cache:
-                        if self._debug:
-                            print("removing data '%s' from cache." % step)
-                        cache.pop(step)
+                # Cache value may be missing if it is optional.
+                if step in cache:
+                    if self._debug:
+                        print("removing data '%s' from cache." % step)
+                    del cache[step]
 
             elif isinstance(step, PinInstruction):
                 self._pin_data_in_cache(step, cache, inputs, overwrites)
@@ -550,7 +555,7 @@ def get_node_name(a):
         g = pydot.Dot(graph_type="digraph")
 
         # draw nodes
-        for nx_node in self.graph.nodes():
+        for nx_node in self.graph.nodes:
             if isinstance(nx_node, DataPlaceholderNode):
                 node = pydot.Node(name=nx_node, shape="rect")
             else:
@@ -592,50 +597,45 @@ def get_node_name(a):
         return g
 
 
-def ready_to_schedule_operation(op, has_executed, graph):
-    """
-    Determines if a Operation is ready to be scheduled for execution based on
-    what has already been executed.
+    def _can_schedule_operation(self, op, executed_nodes):
+        """
+        Determines if a Operation is ready to be scheduled for execution
+
+        based on what has already been executed.
 
-    Args:
-        op:
+        :param op:
             The Operation object to check
-        has_executed: set
+        :param set executed_nodes
             A set containing all operations that have been executed so far
-        graph:
-            The networkx graph containing the operations and data nodes
-    Returns:
-        A boolean indicating whether the operation may be scheduled for
-        execution based on what has already been executed.
-    """
-    # unordered, not iterated
-    dependencies = set(filter(lambda v: isinstance(v, Operation),
-                              nx.ancestors(graph, op)))
-    return dependencies.issubset(has_executed)
+        :return:
+            A boolean indicating whether the operation may be scheduled for
+            execution based on what has already been executed.
+        """
+        # unordered, not iterated
+        dependencies = set(n for n in nx.ancestors(self.graph, op)
+                           if isinstance(n, Operation))
+        return dependencies.issubset(executed_nodes)
 
-def ready_to_delete_data_node(name, has_executed, graph):
-    """
-    Determines if a DataPlaceholderNode is ready to be deleted from the
-    cache.
 
-    Args:
-        name:
+    def _can_delete_data_node(self, name, executed_nodes):
+        """
+        Determines if a DataPlaceholderNode is ready to be deleted from cache.
+
+        :param name:
             The name of the data node to check
-        has_executed: set
+        :param executed_nodes: set
             A set containing all operations that have been executed so far
-        graph:
-            The networkx graph containing the operations and data nodes
-    Returns:
-        A boolean indicating whether the data node can be deleted or not.
-    """
-    data_node = get_data_node(name, graph)
-    return set(graph.successors(data_node)).issubset(has_executed)
+        :return:
+            A boolean indicating whether the data node can be deleted or not.
+        """
+        data_node = self.get_data_node(name)
+        return data_node and set(
+            self.graph.successors(data_node)).issubset(executed_nodes)
 
-def get_data_node(name, graph):
-    """
-    Gets a data node from a graph using its name
-    """
-    for node in graph.nodes():
-        if node == name and isinstance(node, DataPlaceholderNode):
+    def get_data_node(self, name):
+        """
+        Retuen the data node from a graph using its name, or None.
+        """
+        node = self.graph.nodes[name]
+        if isinstance(node, DataPlaceholderNode):
             return node
-    return None

test/test_graphkit.py

@@ -11,6 +11,7 @@
 import graphkit.network as network
 import graphkit.modifiers as modifiers
 from graphkit import operation, compose, Operation
+from graphkit.network import DeleteInstruction
 
 
 def scream(*args, **kwargs):
@@ -226,7 +227,7 @@ def test_pruning_not_overrides_given_intermediate():
     netop.set_overwrites_collector(overwrites)
     assert netop({"a": 5, "overriden": 1, "c": 2}, ["asked"]) == filtdict(exp, "asked")
     assert overwrites == {}  # unjust must have been pruned
-    
+
     overwrites = {}
     netop.set_overwrites_collector(overwrites)
     assert netop({"a": 5, "overriden": 1, "c": 2}) == exp
@@ -413,6 +414,9 @@ def addplusplus(a, b, c=0):
 
 def test_deleteinstructs_vary_with_inputs():
     # Check #21: DeleteInstructions positions vary when inputs change.
+    def count_deletions(steps):
+        return sum(isinstance(n, DeleteInstruction) for n in steps)
+
     netop = compose(name="netop")(
         operation(name="a free without b", needs=["a"], provides=["aa"])(identity),
         operation(name="satisfiable", needs=["a", "b"], provides=["ab"])(add),
@@ -438,27 +442,40 @@ def test_deleteinstructs_vary_with_inputs():
     assert res == filtdict(exp, "asked")  # ok
     steps22 = netop.net.execution_plan
 
-    assert steps11 == steps12
-    assert steps21 == steps22
-    assert steps11 != steps21  # FAILs in v1.2.4 + #18
-    assert steps12 != steps22  # FAILs in v1.2.4 + #18
+    # When no outs, no del-instructs.
+    assert steps11 != steps12
+    assert count_deletions(steps11) == 0
+    assert steps21 != steps22
+    assert count_deletions(steps21) == 0
+
+    # Check steps vary with inputs
+    #
+    # FAILs in v1.2.4 + #18, PASS in #26
+    assert steps11 != steps21
+
+    # Check deletes vary with inputs
+    #
+    # FAILs in v1.2.4 + #18, PASS in #26
+    assert count_deletions(steps12) != count_deletions(steps22)
 
 
 def test_parallel_execution():
     import time
 
+    delay = 0.5
+
     def fn(x):
-        time.sleep(1)
+        time.sleep(delay)
         print("fn %s" % (time.time() - t0))
         return 1 + x
 
     def fn2(a,b):
-        time.sleep(1)
+        time.sleep(delay)
         print("fn2 %s" % (time.time() - t0))
         return a+b
 
     def fn3(z, k=1):
-        time.sleep(1)
+        time.sleep(delay)
         print("fn3 %s" % (time.time() - t0))
         return z + k
 
@@ -527,8 +544,8 @@ def infer(i):
         assert tuple(sorted(results.keys())) == tuple(sorted(outputs)), (outputs, results)
         return results
 
-    N = 100
-    for i in range(20, 200):
+    N = 33
+    for i in range(13, 61):
         pool = Pool(i)
         pool.map(infer, range(N))
         pool.close()