Implement node optimization functions

src/graph_optimizations.jl

@@ -1,41 +1,96 @@
 
 function node_fusion(graph::DAG, n1::ComputeTaskNode, n2::DataTaskNode, n3::ComputeTaskNode)
    if !(n1 in graph) || !(n2 in graph) || !(n3 in graph)
-      error("[Node Fusion] Given nodes are not part of the given graph")
+      error("[Node Fusion] The given nodes are not part of the given graph")
    end
 
-   required_edge1 = Edge(Tuple{ComputeTaskNode, DataTaskNode}(Ref(n1), Ref(n2)))
-   required_edge2 = Edge(Tuple{DataTaskNode, ComputeTaskNode}(Ref(n2), Ref(n3)))
+   required_edge1 = make_edge(n1, n2)
+   required_edge2 = make_edge(n2, n3)
 
    if !(required_edge1 in graph) || !(required_edge2 in graph)
-      error("[Node Fusion] Given nodes are not connected by edges which is required for node fusion")
+      error("[Node Fusion] The given nodes are not connected by edges which is required for node fusion")
    end
 
-   # TODO: Perform node fusion
+   # save children and parents
+   n1_children = children(graph, n1)
+   n3_parents = parents(graph, n3)
+
+   # remove the edges and nodes that will be replaced by the fused node
+   remove_edge(graph, required_edge1)
+   remove_edge(graph, required_edge2)
+   remove_node(graph, n1)
+   remove_node(graph, n2)
+   remove_node(graph, n3)
+
+   # create new node with the fused compute task
+   new_node = ComputeTaskNode(FusedComputeTask{typeof(n1), typeof(n3)}())
+   insert_node(graph, new_node)
+   
+   for child in n1_children
+      insert_edge(graph, make_edge(child, new_node))
+   end
+
+   for parent in n3_parents
+      insert_edge(graph, make_edge(new_node, parent))
+   end
+
+   return nothing
 end
 
 function node_reduction(graph::DAG, n1::Node, n2::Node)
    if !(n1 in graph) || !(n2 in graph)
-      error("[Node Reduction] Given nodes are not part of the given graph")
+      error("[Node Reduction] The given nodes are not part of the given graph")
    end
 
-   prerequisite_nodes = children(graph, n1)
-   if prerequisite_nodes != children(graph, n2)
-      error("[Node Reduction] Given nodes do not have equal prerequisite nodes which is required for node reduction")
+   if typeof(n1) != typeof(n2)
+      error("[Node Reduction] The given nodes are not of the same type")
    end
 
-   # TODO: Perform node reduction
+   # save n2 parents and children
+   n2_children = children(graph, n2)
+   n2_parents = parents(graph, n2)
+
+   if n2_children != children(graph, n1)
+      error("[Node Reduction] The given nodes do not have equal prerequisite nodes which is required for node reduction")
+   end
+
+   # remove n2 and all its parents and children
+   for child in n2_children
+      remove_edge(graph, make_edge(child, n2))
+   end
+   for parent in n2_parents
+      remove_edge(graph, make_edge(n2, parent))
+
+      # add parents of n2 to n1
+      insert_edge(graph, make_edge(n1, parent))
+   end
+   remove_node(graph, n2)
+
+   return nothing
 end
 
 function node_split(graph::DAG, n1::Node)
    if !(n1 in graph)
-      error("[Node Split] Given node is not part of the given graph")
+      error("[Node Split] The given node is not part of the given graph")
    end
 
-   subsequent_nodes = parents(graph, n1)
-   if size(subsequent_nodes) <= 1
-      error("[Node Split] Given node does not have multiple parents which is required for node split")
+   n1_parents = parents(graph, n1)
+   n1_children = children(graph, n1)
+
+   if size(n1_parents) <= 1
+      error("[Node Split] The given node does not have multiple parents which is required for node split")
    end
 
-   # TODO: Perform node split
+   for parent in n1_parents
+      n_copy = copy(n1)
+      insert_node(graph, n_copy)
+      insert_edge(graph, make_edge(n_copy, parent))
+      remove_edge(graph, make_edge(n1, parent))
+      
+      for child in n1_children
+         insert_edge(graph, make_edge(child, n_copy))
+      end
+   end
+
+   return nothing
 end

src/node_functions.jl

@@ -0,0 +1,12 @@
+
+function make_edge(n1::Node, n2::Node)
+    error("Can only create edges from compute to data node or reverse")
+end
+
+function make_edge(n1::ComputeTaskNode, n2::DataTaskNode)
+    return Edge(Tuple{ComputeTaskNode, DataTaskNode}(Ref(n1), Ref(n2)))
+end
+
+function make_edge(n1::DataTaskNode, n2::ComputeTaskNode)
+    return Edge(Tuple{DataTaskNode, ComputeTaskNode}(Ref(n2), Ref(n3)))
+end

src/task_functions.jl

@@ -24,6 +24,10 @@ compute_effort(t::ComputeTaskS2) = 10
 compute_effort(t::ComputeTaskU) = 6
 compute_effort(t::ComputeTaskV) = 20
 compute_effort(t::ComputeTaskP) = 15
+function compute_effort(t::FusedComputeTask)
+   (T1, T2) = collect(typeof(t).parameters)
+   return compute_effort(T1()) + compute_effort(T2())
+end
 
 # actual compute functions for the tasks can stay undefined for now
 # compute(t::ComputeTaskU, data::Any) = mycomputation(data)

src/tasks.jl

@@ -26,3 +26,6 @@ end
 # u task with 1 child
 struct ComputeTaskU <: AbstractComputeTask
 end
+
+struct FusedComputeTask{T1<:AbstractComputeTask, T2<:AbstractComputeTask} <: AbstractComputeTask
+end