使用拓扑排序在有向非循环依赖图中对并发处理进行分组的任务 [英] Group tasks for concurrent processing in directed acyclic dependency graph using topological sorting

问题描述

我有Task类，该类在执行之前依赖于其他任务.我想对可以并行化的任务进行分组并对其进行排序.我决定可以先将其表示为DAG，然后尝试使用JGrapht.首先，我遍历任务的输入列表以获取所有任务及其相关性，并将它们收集在一个列表中.然后，对于每个任务，我都会在图形中创建一个顶点.

I have Task class that has a dependency on other tasks before execution. I want to group tasks that can be parallelized and order them. I decide that it can be represented as DAG first and trying to use JGrapht. First i'm traversing input list of tasks to get all the tasks with their dependancies and collecting them in one list. Then for each task I'm creating a vertex in the graph.

DirectedAcyclicGraph<Task, DefaultEdge> d = new DirectedAcyclicGraph<>(DefaultEdge.class);
Set<Task> all = collectAllTasks(tasks);
all.forEach(d::addVertex);

然后使用相同的列表，我试图在节点之间创建边.

then using same list I'm trying to create edges between nodes.

all.forEach(task -> {
        Set<TaskName> predecessors = task.getPredecessors();

        predecessors.forEach(name -> {
            Task predecessor = taskService.getTaskByName(name);
            d.addEdge(predecessor, task);
        });

    });

然后我要对任务进行分组

Then I'm trying to group tasks

private Set<Set<TaskId>> groupTasks(DirectedAcyclicGraph<TaskId, DefaultEdge> dag) {
    Set<Set<TaskId>> groups = new LinkedHashSet<>();
    Iterator<TaskId> iterator = new TopologicalOrderIterator<>(dag);

    iterator.forEachRemaining(taskId -> {
        //source?
        if (dag.incomingEdgesOf(taskId).isEmpty()) {
            if (groups.isEmpty()) {
                Set<TaskId> set = new HashSet<>();
                set.add(taskId);
                groups.add(set);
            } else {
                groups.iterator().next().add(taskId);
            }
        }

        Set<TaskId> tasks = new HashSet<>(Graphs.successorListOf(dag, taskId));

        //sink?
        if (tasks.isEmpty()) {
            return;
        }

        groups.add(featuresGroup);
    });

    return groups;
}

因此，结果是有序和分组的任务，例如图形

So the outcome is ordered and grouped tasks, for example for graph

结果将为 A，B，{C，D}，E.

但是，当 B 也是 E 的前身(如本例)

However, it completely breaks on the case when B is also the predecessor of E like this example

如何为像以前一样的图实现 A，B，{C，D}，E 的顺序?我是否可以查看任何特定算法，或者如何以更好的方式实现该算法?谢谢.

How can I achieve order of A, B, {C, D}, E for graph like previous? Is there any particular algorithm I can look or how I can achieve that in a better way? Thanks.

推荐答案

可以使用以下过程获取解决方案:

A solution can be obtained using the following procedure:

1. 第一组包含所有没有传入弧的任务:这些任务没有传入依赖性.
2. 从图中删除第一组中的所有任务
3. 第二组由修改后的图中没有传入弧的任务组成:这些任务的所有依赖关系都得到满足.
4. 从修改后的图中删除第二组中的所有任务.继续此过程，直到所有任务都被删除.

使用JGraphT:

public static List<List<String>> getGroups(Graph<String, DefaultEdge> taskGraph){
    List<List<String>> groups = new ArrayList<>();
    //The first group contains all vertices without incoming arcs
    List<String> group = new LinkedList<>();
    for(String task : taskGraph.vertexSet())
        if(taskGraph.inDegreeOf(task) == 0)
            group.add(task);
    //Next we construct all remaining groups. The group k+1 consists of al vertices without incoming arcs if we were
    //to remove all vertices in the previous group k from the graph.
    do {
        groups.add(group);
        List<String> nextGroup = new LinkedList<>();
        for (String task : group) {
            for (String nextTask : Graphs.neighborSetOf(taskGraph, task)) {
                if (taskGraph.inDegreeOf(nextTask) == 1)
                    nextGroup.add(nextTask);
            }
            taskGraph.removeVertex(task); //Removes a vertex and all its edges from the graph
        }
        group=nextGroup;
    }while(!group.isEmpty());
    return groups;
}
public static Graph<String, DefaultEdge> getGraph1(){
    Graph<String, DefaultEdge> taskGraph=new SimpleDirectedGraph<>(DefaultEdge.class);
    Graphs.addAllVertices(taskGraph, Arrays.asList("A","B","C","D","E"));
    taskGraph.addEdge("A","B");
    taskGraph.addEdge("B","C");
    taskGraph.addEdge("B","D");
    taskGraph.addEdge("C","E");
    taskGraph.addEdge("D","E");
    return taskGraph;
}
public static Graph<String, DefaultEdge> getGraph2(){
    Graph<String, DefaultEdge> taskGraph=new SimpleDirectedGraph<>(DefaultEdge.class);
    Graphs.addAllVertices(taskGraph, Arrays.asList("A","B","C","D","E"));
    taskGraph.addEdge("A","B");
    taskGraph.addEdge("B","C");
    taskGraph.addEdge("B","D");
    taskGraph.addEdge("B","E");
    taskGraph.addEdge("C","E");
    taskGraph.addEdge("D","E");
    return taskGraph;
}
public static void main(String[] args){
    System.out.println("Graph1: "+getGroups(getGraph1()));
    System.out.println("Graph2: "+getGroups(getGraph2()));
}

输出:

Graph1: [[A], [B], [C, D], [E]]
Graph2: [[A], [B], [C, D], [E]]

注意:上面的代码假定输入图确实是有效的任务图.您可以建立额外的检查来确定循环依赖关系，例如如果您的序列类似于:A->B->答:

Note: the above code assumes that the input graph is indeed a valid task graph. You could build in an additional check to identify cyclic dependencies, e.g. if you had a sequence like: A -> B -> A.

这篇关于使用拓扑排序在有向非循环依赖图中对并发处理进行分组的任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！