并行执行有向无环任务图 [英] Parallel execution of directed acyclic graph of tasks

问题描述

我有一个任务列表 [Task-A，Task-B，Task-C，Task-D，...] .
一个任务可以有选择地依赖于其他任务.

例如:
A可以取决于3个任务:B，C和D
B可以取决于两项任务:C和E

基本上是有向无环图，只有在执行相关任务后，才应执行任务.

现在可能会在任何时间出现多个准备执行的任务.在这种情况下，我们可以并行运行它们.

关于如何在具有尽可能多的并行性的同时实现这种执行的任何想法?

  class Task {私有字符串名称；私有列表<任务>dependentTasks;public void run(){//商业逻辑}} 

解决方案

I have a list of tasks [Task-A,Task-B,Task-C,Task-D, ...].
One task can be optionally dependent on other tasks.

For example:
A can be dependent on 3 tasks: B, C and D
B can be dependent on 2 tasks: C and E

It's basically a directed acyclic graph and execution of a task should happen only after the dependent tasks are executed.

Now it might happen that at any point of time, there are multiple tasks that are ready for execution. In such a case, we can run them in parallel.

Any idea on how to implement such an execution while having as much parallelism as possible?

class Task{
     private String name;
     private List<Task> dependentTasks;
     
     public void run(){
     // business logic
     }
}

解决方案

The other answer works fine but is too complicated.

A simpler way is to just execute Kahn's algorithm but in parallel.

The key is to execute all the tasks in parallel for whom all dependencies have been executed.

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;


class DependencyManager {
private final ConcurrentHashMap<String, List<String>> _dependencies = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, List<String>> _reverseDependencies = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, Runnable> _tasks = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, Integer> _numDependenciesExecuted = new ConcurrentHashMap<>();
private final  AtomicInteger _numTasksExecuted = new AtomicInteger(0);
private final ExecutorService _executorService = Executors.newFixedThreadPool(16);

private static Runnable getRunnable(DependencyManager dependencyManager, String taskId){
    return () -> {
    try {
        Thread.sleep(2000);  // A task takes 2 seconds to finish.
        dependencyManager.taskCompleted(taskId);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    };
}

/**
* In case a vertex is disconnected from the rest of the graph.
* @param taskId The task id
*/
public void addVertex(String taskId) {
    _dependencies.putIfAbsent(taskId, new ArrayList<>());
    _reverseDependencies.putIfAbsent(taskId, new ArrayList<>());
    _tasks.putIfAbsent(taskId, getRunnable(this, taskId));
    _numDependenciesExecuted.putIfAbsent(taskId, 0);
}

private void addEdge(String dependentTaskId, String dependeeTaskId) {
    _dependencies.get(dependentTaskId).add(dependeeTaskId);
    _reverseDependencies.get(dependeeTaskId).add(dependentTaskId);
}

public void addDependency(String dependentTaskId, String dependeeTaskId) {
    addVertex(dependentTaskId);
    addVertex(dependeeTaskId);
    addEdge(dependentTaskId, dependeeTaskId);
}

private void taskCompleted(String taskId) {
    System.out.println(String.format("%s:: Task %s done!!", Instant.now(), taskId));
    _numTasksExecuted.incrementAndGet();
    _reverseDependencies.get(taskId).forEach(nextTaskId -> {
    _numDependenciesExecuted.computeIfPresent(nextTaskId, (__, currValue) -> currValue + 1);
    int numDependencies = _dependencies.get(nextTaskId).size();
    int numDependenciesExecuted = _numDependenciesExecuted.get(nextTaskId);
    if (numDependenciesExecuted == numDependencies) {
        // All dependencies have been executed, so we can submit this task to the threadpool. 
        _executorService.submit(_tasks.get(nextTaskId));
    }
    });
    if (_numTasksExecuted.get() == _tasks.size()) {
    topoSortCompleted();
    }
}

private void topoSortCompleted() {
    System.out.println("Topo sort complete!!");
    _executorService.shutdownNow();
}

public void executeTopoSort() {
    System.out.println(String.format("%s:: Topo sort started!!", Instant.now()));
    _dependencies.forEach((taskId, dependencies) -> {
    if (dependencies.isEmpty()) {
        _executorService.submit(_tasks.get(taskId));
    }
    });
}
}

public class TestParallelTopoSort {

public static void main(String[] args) {
    DependencyManager dependencyManager = new DependencyManager();
    dependencyManager.addDependency("8", "5");
    dependencyManager.addDependency("7", "5");
    dependencyManager.addDependency("7", "6");
    dependencyManager.addDependency("6", "3");
    dependencyManager.addDependency("6", "4");
    dependencyManager.addDependency("5", "1");
    dependencyManager.addDependency("5", "2");
    dependencyManager.addDependency("5", "3");
    dependencyManager.addDependency("4", "1");
    dependencyManager.executeTopoSort();
    // Parallel version takes 8 seconds to execute.
    // Serial version would have taken 16 seconds.

}
}

The Directed Acyclic Graph constructed in this example is this:

这篇关于并行执行有向无环任务图的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！