Shuffle files and functions around for more consistent naming and smaller files

src/operation/apply.jl

@@ -0,0 +1,198 @@
+# functions that apply graph operations
+
+# applies all unapplied operations in the DAG
+function apply_all!(graph::DAG)
+    while !isempty(graph.operationsToApply)
+        # get next operation to apply from front of the deque
+        op = popfirst!(graph.operationsToApply)
+
+        # apply it
+        appliedOp = apply_operation!(graph, op)
+
+        # push to the end of the appliedOperations deque
+        push!(graph.appliedOperations, appliedOp)
+    end
+    return nothing
+end
+
+function apply_operation!(graph::DAG, operation::Operation)
+    error("Unknown operation type!")
+end
+
+function apply_operation!(graph::DAG, operation::NodeFusion)
+    diff = node_fusion!(graph, operation.input[1], operation.input[2], operation.input[3])
+    return AppliedNodeFusion(operation, diff)
+end
+
+function apply_operation!(graph::DAG, operation::NodeReduction)
+    diff = node_reduction!(graph, operation.input)
+    return AppliedNodeReduction(operation, diff)
+end
+
+function apply_operation!(graph::DAG, operation::NodeSplit)
+    diff = node_split!(graph, operation.input)
+    return AppliedNodeSplit(operation, diff)
+end
+
+
+function revert_operation!(graph::DAG, operation::AppliedOperation)
+    error("Unknown operation type!")
+end
+
+function revert_operation!(graph::DAG, operation::AppliedNodeFusion)
+    revert_diff!(graph, operation.diff)
+    return operation.operation
+end
+
+function revert_operation!(graph::DAG, operation::AppliedNodeReduction)
+    revert_diff!(graph, operation.diff)
+    return operation.operation
+end
+
+function revert_operation!(graph::DAG, operation::AppliedNodeSplit)
+    revert_diff!(graph, operation.diff)
+    return operation.operation
+end
+
+
+function revert_diff!(graph::DAG, diff::Diff)
+    # add removed nodes, remove added nodes, same for edges
+    # note the order
+    for edge in diff.addedEdges
+        remove_edge!(graph, edge.edge[1], edge.edge[2], false)
+    end
+    for node in diff.addedNodes
+        remove_node!(graph, node, false)
+    end
+
+    for node in diff.removedNodes
+        insert_node!(graph, node, false)
+    end
+    for edge in diff.removedEdges
+        insert_edge!(graph, edge.edge[1], edge.edge[2], false)
+    end
+end
+
+# Fuse nodes n1 -> n2 -> n3 together into one node, return the applied difference to the graph
+function node_fusion!(graph::DAG, n1::ComputeTaskNode, n2::DataTaskNode, n3::ComputeTaskNode)
+    # @assert is_valid_node_fusion_input(graph, n1, n2, n3)
+
+    # clear snapshot
+    get_snapshot_diff(graph)
+
+
+    # save children and parents
+    n1_children = children(n1)
+    n3_parents = parents(n3)
+    n3_children = children(n3)
+
+    # remove the edges and nodes that will be replaced by the fused node
+    remove_edge!(graph, n1, n2)
+    remove_edge!(graph, n2, n3)
+    remove_node!(graph, n1)
+    remove_node!(graph, n2)
+
+    # get n3's children now so it automatically excludes n2
+    n3_children = children(n3)
+    remove_node!(graph, n3)
+
+    # create new node with the fused compute task
+    new_node = ComputeTaskNode(FusedComputeTask{typeof(n1.task),typeof(n3.task)}())
+    insert_node!(graph, new_node)
+
+    # use a set for combined children of n1 and n3 to not get duplicates
+    n1and3_children = Set{Node}()
+
+    # remove edges from n1 children to n1
+    for child in n1_children
+        remove_edge!(graph, child, n1)
+        push!(n1and3_children, child)
+    end
+
+    # remove edges from n3 children to n3
+    for child in n3_children
+        remove_edge!(graph, child, n3)
+        push!(n1and3_children, child)
+    end
+
+    for child in n1and3_children
+        insert_edge!(graph, child, new_node)
+    end
+
+    # "repoint" parents of n3 from new node
+    for parent in n3_parents
+        remove_edge!(graph, n3, parent)
+        insert_edge!(graph, new_node, parent)
+    end
+
+    return get_snapshot_diff(graph)
+end
+
+function node_reduction!(graph::DAG, nodes::Vector{Node})
+    # @assert is_valid_node_reduction_input(graph, nodes)
+
+    # clear snapshot
+    get_snapshot_diff(graph)
+
+    n1 = nodes[1]
+    n1_children = children(n1)
+    
+    n1_parents = Set(n1.parents)
+    new_parents = Set{Node}()
+
+    # remove all of the nodes' parents and children and the nodes themselves (except for first node)
+    for i in 2:length(nodes)
+        n = nodes[i]
+        for child in n1_children
+            remove_edge!(graph, child, n)
+        end
+
+        for parent in parents(n)
+            remove_edge!(graph, n, parent)
+
+            # collect all parents
+            push!(new_parents, parent)
+        end
+
+        remove_node!(graph, n)
+    end
+
+    setdiff!(new_parents, n1_parents)
+
+    for parent in new_parents
+        # now add parents of all input nodes to n1 without duplicates
+        insert_edge!(graph, n1, parent)
+    end
+
+    return get_snapshot_diff(graph)
+end
+
+function node_split!(graph::DAG, n1::Node)
+    # @assert is_valid_node_split_input(graph, n1)
+
+    # clear snapshot
+    get_snapshot_diff(graph)
+
+    n1_parents = parents(n1)
+    n1_children = children(n1)
+
+    for parent in n1_parents
+        remove_edge!(graph, n1, parent)
+    end
+    for child in n1_children
+        remove_edge!(graph, child, n1)
+    end
+    remove_node!(graph, n1)
+
+    for parent in n1_parents
+        n_copy = copy(n1)
+        insert_node!(graph, n_copy)
+        insert_edge!(graph, n_copy, parent)
+
+        for child in n1_children
+            insert_edge!(graph, child, n_copy)
+        end
+    end
+
+    return get_snapshot_diff(graph)
+end

src/operation/clean.jl

@@ -0,0 +1,115 @@
+# functions for "cleaning" nodes, i.e. regenerating the possible operations for a node
+
+# function to find node fusions involving the given node if it's a data node
+# pushes the found fusion everywhere it needs to be and returns nothing
+function find_fusions!(graph::DAG, node::DataTaskNode)
+   # if there is already a fusion here, skip
+   if !ismissing(node.nodeFusion)
+      return nothing
+   end
+
+   if length(node.parents) != 1 || length(node.children) != 1
+      return nothing
+   end
+
+   child_node = first(node.children)
+   parent_node = first(node.parents)
+
+   if !(child_node in graph) || !(parent_node in graph)
+      error("Parents/Children that are not in the graph!!!")
+   end
+
+   if length(child_node.parents) != 1
+      return nothing
+   end
+
+   nf = NodeFusion((child_node, node, parent_node))
+   push!(graph.possibleOperations.nodeFusions, nf)
+   push!(child_node.nodeFusions, nf)
+   node.nodeFusion = nf
+   push!(parent_node.nodeFusions, nf)
+
+   return nothing
+end
+
+
+function find_fusions!(graph::DAG, node::ComputeTaskNode)
+   # just find fusions in neighbouring DataTaskNodes
+   for child in node.children
+      find_fusions!(graph, child)
+   end
+
+   for parent in node.parents
+      find_fusions!(graph, parent)
+   end
+
+   return nothing
+end
+
+function find_reductions!(graph::DAG, node::Node)
+   # there can only be one reduction per node, avoid adding duplicates
+   if !ismissing(node.nodeReduction)
+      return nothing
+   end
+
+   reductionVector = nothing
+   # possible reductions are with nodes that are partners, i.e. parents of children
+   partners_ = partners(node)
+   delete!(partners_, node)
+   for partner in partners_
+      if partner ∉ graph.nodes
+         error("Partner is not part of the graph")
+      end
+
+      if can_reduce(node, partner)
+         if Set(node.children) != Set(partner.children)
+            error("Not equal children")
+         end
+         if reductionVector === nothing
+            # only when there's at least one reduction partner, insert the vector
+            reductionVector = Vector{Node}()
+            push!(reductionVector, node)
+         end
+
+         push!(reductionVector, partner)
+      end
+   end
+
+   if reductionVector !== nothing
+      nr = NodeReduction(reductionVector)
+      push!(graph.possibleOperations.nodeReductions, nr)
+      for node in reductionVector
+         if !ismissing(node.nodeReduction)
+            # it can happen that the dirty node becomes part of an existing NodeReduction and overrides those ones now
+            # this is only a problem insofar the existing NodeReduction has to be deleted and replaced also in the possibleOperations
+            invalidate_caches!(graph, node.nodeReduction)
+         end
+         node.nodeReduction = nr
+      end
+   end
+
+   return nothing
+end
+
+function find_splits!(graph::DAG, node::Node)
+   if !ismissing(node.nodeSplit)
+      return nothing
+   end
+
+   if (can_split(node))
+      ns = NodeSplit(node)
+      push!(graph.possibleOperations.nodeSplits, ns)
+      node.nodeSplit = ns
+   end
+
+   return nothing
+end
+
+# "clean" the operations on a dirty node
+function clean_node!(graph::DAG, node::Node)
+   sort_node!(node)
+   
+   find_fusions!(graph, node)
+   find_reductions!(graph, node)
+   find_splits!(graph, node)
+end

src/operation/find.jl

@@ -0,0 +1,205 @@
+# functions that find operations on the inital graph
+
+using Base.Threads
+
+function insert_operation!(nf::NodeFusion, locks::Dict{ComputeTaskNode, SpinLock})
+   n1 = nf.input[1]; n2 = nf.input[2]; n3 = nf.input[3]
+
+   lock(locks[n1]) do; push!(nf.input[1].nodeFusions, nf); end
+   nf.input[2].nodeFusion = nf
+   lock(locks[n3]) do; push!(nf.input[3].nodeFusions, nf); end
+   return nothing
+end
+
+function insert_operation!(nr::NodeReduction)
+   for n in nr.input
+      n.nodeReduction = nr
+   end
+   return nothing
+end
+
+function insert_operation!(ns::NodeSplit)
+   ns.input.nodeSplit = ns
+   return nothing
+end
+
+function nr_insertion!(operations::PossibleOperations, nodeReductions::Vector{Vector{NodeReduction}})
+   total_len = 0
+   for vec in nodeReductions
+      total_len += length(vec)
+   end
+   sizehint!(operations.nodeReductions, total_len)
+
+   t = @task for vec in nodeReductions
+      union!(operations.nodeReductions, Set(vec))
+   end
+   schedule(t)
+
+   @threads for vec in nodeReductions
+      for op in vec
+         insert_operation!(op)
+      end
+   end
+
+   wait(t)
+
+   return nothing
+end
+
+function nf_insertion!(graph::DAG, operations::PossibleOperations, nodeFusions::Vector{Vector{NodeFusion}})
+   total_len = 0
+   for vec in nodeFusions
+      total_len += length(vec)
+   end
+   sizehint!(operations.nodeFusions, total_len)
+   
+   t = @task for vec in nodeFusions
+      union!(operations.nodeFusions, Set(vec))
+   end
+   schedule(t)
+
+   locks = Dict{ComputeTaskNode, SpinLock}()
+   for n in graph.nodes
+      if (typeof(n) <: ComputeTaskNode)
+         locks[n] = SpinLock()
+      end
+   end
+
+   @threads for vec in nodeFusions 
+      for op in vec
+         insert_operation!(op, locks)
+      end
+   end
+
+   wait(t)
+
+   return nothing
+end
+
+function ns_insertion!(operations::PossibleOperations, nodeSplits::Vector{Vector{NodeSplit}})
+   total_len = 0
+   for vec in nodeSplits
+      total_len += length(vec)
+   end
+   sizehint!(operations.nodeSplits, total_len)
+
+   t = @task for vec in nodeSplits
+      union!(operations.nodeSplits, Set(vec))
+   end
+   schedule(t)
+
+   @threads for vec in nodeSplits
+      for op in vec
+         insert_operation!(op)
+      end
+   end
+
+   wait(t)
+
+   return nothing
+end
+
+# function to generate all possible operations on the graph
+function generate_options(graph::DAG)
+   generatedFusions = [Vector{NodeFusion}() for _ in 1:nthreads()]
+   generatedReductions = [Vector{NodeReduction}() for _ in 1:nthreads()]
+   generatedSplits = [Vector{NodeSplit}() for _ in 1:nthreads()]
+
+   # make sure the graph is fully generated through
+   apply_all!(graph)
+
+   nodeArray = collect(graph.nodes)
+
+   # sort all nodes
+   @threads for node in nodeArray
+      sort_node!(node)
+   end
+
+   checkedNodes = Set{Node}()
+   checkedNodesLock = SpinLock()
+   # --- find possible node reductions ---
+   @threads for node in nodeArray
+      # we're looking for nodes with multiple parents, those parents can then potentially reduce with one another
+      if (length(node.parents) <= 1)
+         continue
+      end
+
+      candidates = node.parents
+
+      # sort into equivalence classes
+      trie = NodeTrie()
+
+      for candidate in candidates
+         # insert into trie
+         insert!(trie, candidate)
+      end
+
+      nodeReductions = collect(trie)
+
+      for nrVec in nodeReductions
+         # parent sets are ordered and any node can only be part of one nodeReduction, so a NodeReduction is uniquely identifiable by its first element
+         # this prevents duplicate nodeReductions being generated
+         lock(checkedNodesLock)
+         if (nrVec[1] in checkedNodes)
+            unlock(checkedNodesLock)
+            continue
+         else
+            push!(checkedNodes, nrVec[1])
+         end
+         unlock(checkedNodesLock)
+
+         push!(generatedReductions[threadid()], NodeReduction(nrVec))
+      end
+   end
+
+   
+   # launch thread for node reduction insertion
+   # remove duplicates
+   nr_task = @task nr_insertion!(graph.possibleOperations, generatedReductions)
+   schedule(nr_task)
+
+   # --- find possible node fusions ---
+   @threads for node in nodeArray
+      if (typeof(node) <: DataTaskNode)
+         if length(node.parents) != 1
+            # data node can only have a single parent
+            continue
+         end
+         parent_node = first(node.parents)
+
+         if length(node.children) != 1
+            # this node is an entry node or has multiple children which should not be possible
+            continue
+         end
+         child_node = first(node.children)
+         if (length(child_node.parents) != 1)
+            continue
+         end
+
+         push!(generatedFusions[threadid()], NodeFusion((child_node, node, parent_node)))
+      end
+   end
+
+   # launch thread for node fusion insertion
+   nf_task = @task nf_insertion!(graph, graph.possibleOperations, generatedFusions)
+   schedule(nf_task)
+
+   # find possible node splits
+   @threads for node in nodeArray
+      if (can_split(node))
+         push!(generatedSplits[threadid()], NodeSplit(node))
+      end
+   end
+
+   # launch thread for node split insertion
+   ns_task = @task ns_insertion!(graph.possibleOperations, generatedSplits)
+   schedule(ns_task)
+
+   empty!(graph.dirtyNodes)
+
+   wait(nr_task)
+   wait(nf_task)
+   wait(ns_task)
+
+   return nothing
+end

src/operation/get.jl

@@ -0,0 +1,18 @@
+# function to return the possible operations of a graph
+
+using Base.Threads
+
+function get_operations(graph::DAG)
+   apply_all!(graph)
+
+   if isempty(graph.possibleOperations)
+      generate_options(graph)
+   end
+
+   for node in graph.dirtyNodes
+      clean_node!(graph, node)
+   end
+   empty!(graph.dirtyNodes)
+
+   return graph.possibleOperations
+end

src/operation/print.jl

@@ -0,0 +1,38 @@
+function show(io::IO, ops::PossibleOperations)
+    print(io, length(ops.nodeFusions))
+    println(io, " Node Fusions: ")
+    for nf in ops.nodeFusions
+        println(io, "  - ", nf)
+    end
+    print(io, length(ops.nodeReductions))
+    println(io, " Node Reductions: ")
+    for nr in ops.nodeReductions
+        println(io, "  - ", nr)
+    end
+    print(io, length(ops.nodeSplits))
+    println(io, " Node Splits: ")
+    for ns in ops.nodeSplits
+        println(io, "  - ", ns)
+    end
+end
+
+function show(io::IO, op::NodeReduction)
+    print(io, "NR: ")
+    print(io, length(op.input))
+    print(io, "x")
+    print(io, op.input[1].task)
+end
+
+function show(io::IO, op::NodeSplit)
+    print(io, "NS: ")
+    print(io, op.input.task)
+end
+
+function show(io::IO, op::NodeFusion)
+    print(io, "NF: ")
+    print(io, op.input[1].task)
+    print(io, "->")
+    print(io, op.input[2].task)
+    print(io, "->")
+    print(io, op.input[3].task)
+end

src/operation/type.jl

@@ -0,0 +1,34 @@
+# An abstract base class for operations
+# an operation can be applied to a DAG
+abstract type Operation end
+
+# An abstract base class for already applied operations
+# an applied operation can be reversed iff it is the last applied operation on the DAG
+abstract type AppliedOperation end
+
+struct NodeFusion <: Operation
+   input::Tuple{ComputeTaskNode, DataTaskNode, ComputeTaskNode}
+end
+
+struct AppliedNodeFusion <: AppliedOperation
+   operation::NodeFusion
+   diff::Diff
+end
+
+struct NodeReduction <: Operation
+   input::Vector{Node}
+end
+
+struct AppliedNodeReduction <: AppliedOperation
+   operation::NodeReduction
+   diff::Diff
+end
+
+struct NodeSplit <: Operation
+   input::Node
+end
+
+struct AppliedNodeSplit <: AppliedOperation
+   operation::NodeSplit
+   diff::Diff
+end

src/operation/utility.jl

@@ -0,0 +1,107 @@
+
+function isempty(operations::PossibleOperations)
+   return isempty(operations.nodeFusions) && 
+          isempty(operations.nodeReductions) && 
+          isempty(operations.nodeSplits)
+end
+
+function length(operations::PossibleOperations)
+   return (nodeFusions = length(operations.nodeFusions),
+           nodeReductions = length(operations.nodeReductions),
+           nodeSplits = length(operations.nodeSplits))
+end
+
+function delete!(operations::PossibleOperations, op::NodeFusion)
+   delete!(operations.nodeFusions, op)
+   return operations
+end
+
+function delete!(operations::PossibleOperations, op::NodeReduction)
+   delete!(operations.nodeReductions, op)
+   return operations
+end
+
+function delete!(operations::PossibleOperations, op::NodeSplit)
+   delete!(operations.nodeSplits, op)
+   return operations
+end
+
+
+function can_fuse(n1::ComputeTaskNode, n2::DataTaskNode, n3::ComputeTaskNode)
+   if !is_child(n1, n2) || !is_child(n2, n3)
+      # the checks are redundant but maybe a good sanity check
+      return false
+   end
+
+   if length(n2.parents) != 1 || length(n2.children) != 1 || length(n1.parents) != 1
+      return false
+   end
+
+   return true
+end
+
+function can_reduce(n1::Node, n2::Node)
+   if (n1.task != n2.task)
+      return false
+   end
+   
+   n1_length = length(n1.children)
+   n2_length = length(n2.children)
+
+   if (n1_length != n2_length)
+      return false
+   end
+
+   # this seems to be the most common case so do this first
+   # doing it manually is a lot faster than using the sets for a general solution
+   if (n1_length == 2)
+      if (n1.children[1] != n2.children[1])
+         if (n1.children[1] != n2.children[2])
+            return false
+         end
+         # 1_1 == 2_2
+         if (n1.children[2] != n2.children[1])
+            return false
+         end
+         return true
+      end
+
+      # 1_1 == 2_1
+      if (n1.children[2] != n2.children[2])
+         return false
+      end
+      return true
+   end
+
+   # this is simple
+   if (n1_length == 1)
+      return n1.children[1] == n2.children[1]
+   end
+   
+   # this takes a long time
+   return Set(n1.children) == Set(n2.children)
+end
+
+function can_split(n::Node)
+   return length(parents(n)) > 1
+end
+
+function ==(op1::Operation, op2::Operation)
+   return false
+end
+
+function ==(op1::NodeFusion, op2::NodeFusion)
+   # there can only be one node fusion on a given data task, so if the data task is the same, the fusion is the same
+   return op1.input[2] == op2.input[2]
+end
+
+function ==(op1::NodeReduction, op2::NodeReduction)
+   # node reductions are equal exactly if their first input is the same
+   return op1.input[1].id == op2.input[1].id
+end
+
+function ==(op1::NodeSplit, op2::NodeSplit)
+   return op1.input == op2.input
+end
+
+copy(id::UUID) = UUID(id.value)

src/operation/validate.jl

@@ -0,0 +1,79 @@
+# functions to throw assertion errors for inconsistent or wrong node operations
+# should be called with @assert
+# the functions throw their own errors though, to still have helpful error messages
+
+function is_valid_node_fusion_input(graph::DAG, n1::ComputeTaskNode, n2::DataTaskNode, n3::ComputeTaskNode)
+    if !(n1 in graph) || !(n2 in graph) || !(n3 in graph)
+        throw(AssertionError("[Node Fusion] The given nodes are not part of the given graph"))
+    end
+
+    if !is_child(n1, n2) || !is_child(n2, n3) || !is_parent(n3, n2) || !is_parent(n2, n1)
+        throw(AssertionError("[Node Fusion] The given nodes are not connected by edges which is required for node fusion"))
+    end
+
+    if length(n2.parents) > 1
+        throw(AssertionError("[Node Fusion] The given data node has more than one parent"))
+    end
+    if length(n2.children) > 1
+        throw(AssertionError("[Node Fusion] The given data node has more than one child"))
+    end
+    if length(n1.parents) > 1
+        throw(AssertionError("[Node Fusion] The given n1 has more than one parent"))
+    end
+
+    return true
+end
+
+function is_valid_node_reduction_input(graph::DAG, nodes::Vector{Node})
+    for n in nodes
+        if n ∉ graph
+            throw(AssertionError("[Node Reduction] The given nodes are not part of the given graph"))
+        end
+    end
+
+    t = typeof(nodes[1].task)
+    for n in nodes
+        if typeof(n.task) != t
+            throw(AssertionError("[Node Reduction] The given nodes are not of the same type"))
+        end
+    end
+
+    n1_children = nodes[1].children
+    for n in nodes
+        if Set(n1_children) != Set(n.children)
+            throw(AssertionError("[Node Reduction] The given nodes do not have equal prerequisite nodes which is required for node reduction"))
+        end
+    end
+
+    return true
+end
+
+function is_valid_node_split_input(graph::DAG, n1::Node)
+    if n1 ∉ graph
+        throw(AssertionError("[Node Split] The given node is not part of the given graph"))
+    end
+
+    if length(n1.parents) <= 1
+        throw(AssertionError("[Node Split] The given node does not have multiple parents which is required for node split"))
+    end
+
+    return true
+end
+
+function is_valid(graph::DAG, nr::NodeReduction)
+    @assert is_valid_node_reduction_input(graph, nr.input)
+    @assert nr in graph.possibleOperations.nodeReductions "NodeReduction is not part of the graph's possible operations!"
+    return true
+end
+
+function is_valid(graph::DAG, ns::NodeSplit)
+    @assert is_valid_node_split_input(graph, ns.input)
+    @assert ns in graph.possibleOperations.nodeSplits "NodeSplit is not part of the graph's possible operations!"
+    return true
+end
+
+function is_valid(graph::DAG, nf::NodeFusion)
+    @assert is_valid_node_fusion_input(graph, nf.input[1], nf.input[2], nf.input[3])
+    @assert nf in graph.possibleOperations.nodeFusions "NodeFusion is not part of the graph's possible operations!"
+    return true
+end