Parallel group execution optimization

[Nickqiaoo]

Self Checks

• I have searched for existing issues search for existing issues, including closed ones.
• I confirm that I am using English to submit this report (我已阅读并同意 Language Policy).
• [FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:）
• Please do not modify this template :) and fill in all the required fields.

1. Is this request related to a challenge you're experiencing? Tell me about your story.

#12848

I create a workflow like this, I think 5 should wait for the completion of 1 and 2 before running, but in reality, 5 ran multiple times.
I read the code related to graphs.If the graph is like this:

graph LR;
    START --> 1;
    START --> 2;
    START --> 4;
    1 --> 3;
    2 --> 3;
    3 --> 5;
    5 --> 6;
    4 --> 5;

Loading

Since nodes 1, 2, 3, and 4 are grouped into a parallel group and node 5 is the merge node, in parallel execution, the parallel branches are (1 3), (2 3), and (4). Node 3 is executed twice. I believe the root cause of the problem is that the parallel group division is not reasonable.

2. Additional context or comments

To solve this problem, I propose this solution: if a node's parallel branches are followed by multiple merge nodes, it indicates that some branches have merged prematurely. To be compatible with the existing parallel group algorithm, I will add a segment of processing code. Before these prematurely merged branches, some virtual nodes will be inserted to ensure the normal operation of the existing parallel group algorithm, minimizing code changes.
Just like this:
Before:

graph LR;
    START["START"] --> 1;
    START --> 2;
    START --> 3;
    START --> 4;
    1 --> 5;
    2 --> 5;
    5 --> 6;
    3 --> 6;
    6 --> 7;
    4 --> 7;

Loading

After:

graph LR;
    START --> v2;
    START --> 4;
    v1 --> 1;
    v1 --> 2;
    v2 --> v1;
    v2 --> 3;
    1 --> 5;
    2 --> 5;
    3 --> 6;
    5 --> 6;
    4 --> 7;
    6 --> 7;

Loading

I wrote a piece of test code in Go, referencing the implementation of recursivelyAddParallels in the graph, reused some of the code, and implemented this logic.

func recursivelyAddVirtualNode(
	edgeMapping map[string][]*GraphEdge,
	reverseEdgeMapping map[string][]*GraphEdge,
	startNodeID string,
) {
	targetNodeEdges := edgeMapping[startNodeID]

	if len(targetNodeEdges) > 1 {
		parallelBranchNodeIDs := make(map[string][]string)
		conditionEdgeMappings := make(map[string][]*GraphEdge)
		for _, graphEdge := range targetNodeEdges {
			// If the condition is empty, add the target node ID to the branch node array,
			// indicating that multiple edges come out from a single node and can be executed in parallel
			if graphEdge.RunCondition == nil {
				if _, ok := parallelBranchNodeIDs["default"]; !ok {
					parallelBranchNodeIDs["default"] = []string{}
				}
				parallelBranchNodeIDs["default"] = append(parallelBranchNodeIDs["default"], graphEdge.TargetNodeID)
			} else {
				// If the condition is not empty, add the condition and target node ID to the conditional branch node array
				conditionHash := fmt.Sprintf("%v", graphEdge.RunCondition) // Simple hash
				conditionEdgeMappings[conditionHash] = append(conditionEdgeMappings[conditionHash], graphEdge)
			}
		}

		for conditionHash, graphEdges := range conditionEdgeMappings {
			// If the number of edges in the conditional branch is greater than 1, 
			// the target node IDs of the edges should also be added to the branch node array
			if len(graphEdges) > 1 {
				for _, graphEdge := range graphEdges {
					parallelBranchNodeIDs[conditionHash] = append(parallelBranchNodeIDs[conditionHash], graphEdge.TargetNodeID)
				}
			}
		}

		for _, conditionParallelBranchNodeIDs := range parallelBranchNodeIDs {
			if len(conditionParallelBranchNodeIDs) > 0 {
				addVirtualNodeInParallel(startNodeID, edgeMapping, reverseEdgeMapping, conditionParallelBranchNodeIDs)
			}
		}

		if len(conditionEdgeMappings) > 0 {
			for _, graphEdges := range conditionEdgeMappings {
				for _, graphEdge := range graphEdges {
					recursivelyAddVirtualNode(edgeMapping, reverseEdgeMapping, graphEdge.TargetNodeID)
				}
			}
		} else {
			for _, graphEdge := range edgeMapping[startNodeID] {
				recursivelyAddVirtualNode(edgeMapping, reverseEdgeMapping, graphEdge.TargetNodeID)
			}
		}
	} else {
		for _, graphEdge := range targetNodeEdges {
			recursivelyAddVirtualNode(edgeMapping, reverseEdgeMapping, graphEdge.TargetNodeID)
		}
	}
}

func addVirtualNodeInParallel(
	startNode string,
	edgeMapping map[string][]*GraphEdge,
	reverseEdgeMapping map[string][]*GraphEdge,
	parallelBranchNodeIDs []string,
) {
	routesNodeIDs := make(map[string][]string)
	// For multiple nodes coming out from the same node, get all reachable nodes from each branch
	for _, parallelBranchNodeID := range parallelBranchNodeIDs {
		routesNodeIDs[parallelBranchNodeID] = []string{parallelBranchNodeID}
		// Get all nodes reachable from a branch node
		routesNodeIDs[parallelBranchNodeID] = recursivelyFetchRoutes(edgeMapping, parallelBranchNodeID, routesNodeIDs[parallelBranchNodeID])
	}

	leafNodeIDs := make(map[string][]string)
	mergeBranchNodeIDs := make(map[string][]string)

	// Iterate through all reachable nodes from each branch, identifying leaf nodes and merge nodes
	for branchNodeID, nodeIDs := range routesNodeIDs {
		for _, nodeID := range nodeIDs {
			// If a node has no outgoing edges, it is a leaf node (the last node)
			if _, hasOutgoing := edgeMapping[nodeID]; !hasOutgoing || len(edgeMapping[nodeID]) == 0 {
				leafNodeIDs[branchNodeID] = append(leafNodeIDs[branchNodeID], nodeID)
			}
			// If a node has an in-degree greater than 1 and appears on other branch paths, it is a merge node
			for branchNodeID2, innerRoute2 := range routesNodeIDs {
				if branchNodeID != branchNodeID2 && slices.Contains(innerRoute2, nodeID) && len(reverseEdgeMapping[nodeID]) > 1 {
					//&& isNodeInRoutes(reverseEdgeMapping, nodeID, routesNodeIDs) {
					if !slices.Contains(mergeBranchNodeIDs[nodeID], branchNodeID2) {
						// mergeBranchNodeIDs records which branches the merge node appears in
						mergeBranchNodeIDs[nodeID] = append(mergeBranchNodeIDs[nodeID], branchNodeID2)
					}
				}
			}
		}
	}
	type mergeNodeEntry struct {
		nodeID      string
		branchNodes []string
	}

	// Convert map to slice for sorting
	var sortedEntries []mergeNodeEntry
	for nodeID, branchNodes := range mergeBranchNodeIDs {
		sortedEntries = append(sortedEntries, mergeNodeEntry{
			nodeID:      nodeID,
			branchNodes: branchNodes,
		})
	}

	// Sort by the number of branch nodes in ascending order
	slices.SortFunc(sortedEntries, func(a, b mergeNodeEntry) int {
		return len(a.branchNodes) - len(b.branchNodes)
	})

	duplicateEndNodeIDs := make(map[[2]string][]string)
	for _, entry := range sortedEntries {
		nodeID := entry.nodeID
		branchNodeIDs := entry.branchNodes
		for _, entry2 := range sortedEntries {
			nodeID2 := entry2.nodeID
			branchNodeIDs2 := entry2.branchNodes
			if nodeID != nodeID2 && slices.Equal(branchNodeIDs, branchNodeIDs2) {
				// If two merge nodes share the exact same branches, keep the earlier node
				if _, found := duplicateEndNodeIDs[[2]string{nodeID, nodeID2}]; !found {
					if _, found := duplicateEndNodeIDs[[2]string{nodeID2, nodeID}]; !found {
						duplicateEndNodeIDs[[2]string{nodeID, nodeID2}] = branchNodeIDs
					}
				}
			}
		}
	}
	// When two merge nodes share the same branches, keep the earlier node
	for nodes := range duplicateEndNodeIDs {
		nodeID := nodes[0]
		nodeID2 := nodes[1]
		if isNode2AfterNode1(nodeID, nodeID2, edgeMapping) {
			delete(mergeBranchNodeIDs, nodeID2)
		} else if isNode2AfterNode1(nodeID2, nodeID, edgeMapping) {
			delete(mergeBranchNodeIDs, nodeID)
		}
	}

	// If there is only one merge node, return immediately
	if len(mergeBranchNodeIDs) <= 1{
		return
	}

	// Rebuild sortedEntries, only including remaining merge nodes
	sortedEntries = nil
	for nodeID, branchNodes := range mergeBranchNodeIDs {
		sortedEntries = append(sortedEntries, mergeNodeEntry{
			nodeID:      nodeID,
			branchNodes: branchNodes,
		})
	}

	// Re-sort by the number of branch nodes in ascending order
	slices.SortFunc(sortedEntries, func(a, b mergeNodeEntry) int {
		return len(a.branchNodes) - len(b.branchNodes)
	})

	// Record each branch's corresponding merge node
	branchesMergeNodeIDs := make(map[string]string)
	// Virtual node corresponding to each merge node
	mergeVNode := make(map[string]string)
	// Virtual node corresponding to each branch
	branchVNode := make(map[string]string)

	var latestVNode string

	for _, entry := range sortedEntries {
		mergeNodeID := entry.nodeID
		branchNodeIDs := entry.branchNodes

		if len(branchNodeIDs) <= 1 {
			continue
		}

		for _, branchNodeID := range branchNodeIDs {
			if existingMergeNodeID, found := branchesMergeNodeIDs[branchNodeID]; !found {
				branchesMergeNodeIDs[branchNodeID] = mergeNodeID
			} else {
				// When an existing merge node is found in branchesMergeNodeIDs, it means a new merge node is encountered, so a virtual node needs to be created
				vnode, exist := mergeVNode[existingMergeNodeID]
				if !exist {
					vnode = uuid.New().String()
					mergeVNode[existingMergeNodeID] = vnode
					latestVNode = vnode
				}
				// Connect the branch node to the virtual node
				sourceNode := branchNodeID
				// If there is already a virtual node before the branch node, the connection should be to the virtual node
				if v, ex := branchVNode[branchNodeID]; ex {
					sourceNode = v
				}
				addEdgeToMappings(edgeMapping,reverseEdgeMapping, &GraphEdge{SourceNodeID: vnode, TargetNodeID: sourceNode})
				branchVNode[branchNodeID] = vnode
				// After adding a virtual node, the merge node in the branch becomes the new merge node
				branchesMergeNodeIDs[branchNodeID] = mergeNodeID
			}
		}
	}

	if len(branchVNode) > 0 {
				newEdges := make([]*GraphEdge, 0, len(edgeMapping[startNode]))
		for _, edge := range edgeMapping[startNode] {
			if _, isBranch := branchVNode[edge.TargetNodeID]; !isBranch {
				newEdges = append(newEdges, edge)
			}
		}
		edgeMapping[startNode] = newEdges
	}

	addEdgeToMappings(edgeMapping, reverseEdgeMapping, &GraphEdge{SourceNodeID: startNode, TargetNodeID: latestVNode})
}

func addEdgeToMappings(edgeMapping, reverseEdgeMapping map[string][]*GraphEdge, edge *GraphEdge) {
	// Check forward mapping
	forwardEdges := edgeMapping[edge.SourceNodeID]
	for _, e := range forwardEdges {
		if e.SourceNodeID == edge.SourceNodeID && e.TargetNodeID == edge.TargetNodeID {
			return 
		}
	}
	
	// Add to forward mapping
	edgeMapping[edge.SourceNodeID] = append(edgeMapping[edge.SourceNodeID], edge)
	
	// Add to reverse mapping
	reverseEdgeMapping[edge.TargetNodeID] = append(reverseEdgeMapping[edge.TargetNodeID], edge)
}

The recursivelyAddVirtualNode function needs to be called before the recursivelyAddParallels function. It will modify the graph by inserting virtual nodes to match the existing parallel group execution logic. Modifying the way parallel groups are constructed could also solve the problem, but this would require more extensive code changes to both the graph and graph_engine, which is prone to errors. I believe that using the method of inserting virtual nodes is simpler to implement and only requires minor modifications to the graph_engine to handle virtual nodes specially when encountered.

3. Can you help us with this feature?

• I am interested in contributing to this feature.