Как рассчитать максимальный параллелизм узлов в DAG?

У меня есть рабочий процесс, состоящий из направленного ациклического графа (DAG) с несколькими узлами, каждый из которых имеет случайное время выполнения. Я хотел бы определить максимальный параллелизм узлов в этом рабочем процессе. Что я должен делать?

В настоящее время мы тщательно просматриваем все возможные сценарии выполнения, чтобы найти максимальное значение параллелизма. Однако этот подход требует много времени при работе с большим количеством узлов или ребер. Существуют ли алгоритмы меньшей сложности для решения этой проблемы?

Например:

На этой диаграмме я перечислил 5 простых примеров.

В действительности наша группа обеспечения доступности баз данных может иметь несколько сотен или даже тысяч узлов, а также более сложные отношения зависимостей.

Максимальное значение параллелизма относится к максимальному количеству узлов, которые могут выполняться одновременно на основе отношений зависимости в группе обеспечения доступности баз данных, либо от корневого узла, либо при работе вниз. Например, в первом сценарии 1, 2, 5 и 6 могут выполняться одновременно, что представляет собой максимальный параллелизм среди всех возможных параллельных выполнения. Во втором сценарии, если между 2 и 5 вводится зависимость, максимальное одновременное выполнение будет либо 1, 2, 5, либо 1, 5, 6, либо 1 2 4, либо 3 5 6.

@Хуан Лопес ----------------------- 19.08.2024 добавить

Большое спасибо за ваше руководство. Я кратко изучил алгоритм задачи максимального потока и нашел несколько общих алгоритмов, включая метод Форда Фулкерсона, алгоритм Эдмондса Карпа, алгоритм Диника и алгоритм Push label.

Следуя предоставленной вами демонстрации Python, я усердно учился и с помощью )

Я добавил в свой проект несколько сравнительных тестов и обнаружил, что результаты весьма точны.

Однако есть еще одна проблема: две мои реализации алгоритма имеют очень низкую производительность в случае последовательной зависимости.

пример1:

201 узел, зависимость такая: 1 -> 2 -> 3 -> 4 -> 5 -> xxxxxx -> 201

вывод (dag Calcal Max. Эта строка представляет собой реализацию полного обхода + обрезки HashSet):

2024-08-19 16:11:24 CST UTC+08:00 INFO - test DinicSolver alg
2024-08-19 16:11:52 CST UTC+08:00 INFO - test DinicSolver alg return 1, cost time: 28177ms
2024-08-19 16:11:52 CST UTC+08:00 INFO - test FordFulkersonSolver alg
2024-08-19 16:11:53 CST UTC+08:00 INFO - test FordFulkersonSolver alg return 1, cost time: 372ms
2024-08-19 16:11:53 CST UTC+08:00 INFO - dag calculate max concurrency 1 cost time: 7ms

И я обнаружил, что затраты времени не фиксированы, и чем длиннее имя узла, тем больше затраты времени.

используйте номер в качестве имени узла:

используйте одну букву + цифру в качестве имени узла:

используйте две буквы + цифру в качестве имени узла:

используйте три буквы + цифру в качестве имени узла:

пример2:

201 узел, зависимость как на этой картинке:

вывод (dag Calcal Max. Эта строка представляет собой реализацию полного обхода + обрезки HashSet):

2024-08-19 16:11:55 CST UTC+08:00 INFO - test DinicSolver alg
2024-08-19 16:11:55 CST UTC+08:00 INFO - test DinicSolver alg return 201, cost time: 6ms
2024-08-19 16:11:55 CST UTC+08:00 INFO - test FordFulkersonSolver alg
2024-08-19 16:11:55 CST UTC+08:00 INFO - test FordFulkersonSolver alg return 201, cost time: 2ms
2024-08-19 16:11:55 CST UTC+08:00 INFO - dag calculate max concurrency 10 cost time: 660ms

добавить мою реализацию Java-кода

Сообщ. Форда-Фалкерсона:

public class FordFulkersonSolver<T> {

    private final Map<String, Map<String, Integer>> network;
    private final Set<String> visited;

    public FordFulkersonSolver() {
        network = new HashMap<>();
        visited = new HashSet<>();
    }

    private int dfs(String source, String sink, int flow) {
        visited.add(source);
        if (source.equals(sink)) {
            return flow;
        }

        for (Map.Entry<String, Integer> entry : network.get(source).entrySet()) {
            String neighbor = entry.getKey();
            Integer capacity = entry.getValue();
            if (capacity <= 0 || visited.contains(neighbor)) continue;
            int sent = dfs(neighbor, sink, Math.min(flow, capacity));
            if (sent == 0) continue;
            network.get(source).put(neighbor, capacity - sent);
            network.get(neighbor).put(source, network.get(neighbor).getOrDefault(source, 0) + sent);
            return sent;
        }
        return 0;
    }

    private Set<String> reach(Map<T, Set<T>> graph, T t, Set<String> visited) {
        Queue<T> queue = new LinkedList<>();
        queue.add(t);

        while (!queue.isEmpty()) {
            T current = queue.poll();
            String currentKey = "A" + current.toString();
            visited.add(currentKey);
            for (T neighbor : graph.get(current)) {
                String neighborKey = "B" + neighbor.toString();
                if (!visited.contains(neighborKey)) {
                    queue.add(neighbor);
                    visited.add(neighborKey);
                }
            }
        }

        return visited;
    }

    private void addEdge(String from, String to, int capacity) {
        network.computeIfAbsent(from, k -> new HashMap<>()).put(to, capacity);
        network.computeIfAbsent(to, k -> new HashMap<>()).put(from, 0);
    }

    public int solve(Map<T, Set<T>> graph) {
        for (T t : graph.keySet()) {
            addEdge("src", "A" + t.toString(), 1);
            addEdge("B" + t, "sink", 1);
            // Corrected here to pass the correct generic type for visited
            Set<String> visitedSubset = new HashSet<>();
            for (String u : reach(graph, t, visitedSubset)) {
                addEdge("A" + t, u, 1);
            }
        }

        int total = 0;
        while (true) {
            this.visited.clear(); // Clear visited set for each iteration
            int sent = dfs("src", "sink", Integer.MAX_VALUE);
            if (sent == 0) break;
            total += sent;
        }

        return graph.size() - total;
    }

}

Динические условия:

public class DinicSolver<T>  {

    private final Map<String, Map<String, Integer>> network;
    private List<String> nodes;
    private int[] level;

    public DinicSolver() {
        network = new HashMap<>();
        nodes = new ArrayList<>();
        nodes.add("src");
        nodes.add("sink");
    }

    private void bfs(String source) {
        level = new int[nodes.size()];
        Arrays.fill(level, -1);
        level[nodes.indexOf(source)] = 0;

        Queue<String> queue = new LinkedList<>();
        queue.offer(source);

        while (!queue.isEmpty()) {
            String u = queue.poll();
            for (Map.Entry<String, Integer> entry : network.get(u).entrySet()) {
                String v = entry.getKey();
                int capacity = entry.getValue();
                if (capacity > 0 && level[nodes.indexOf(v)] == -1) {
                    level[nodes.indexOf(v)] = level[nodes.indexOf(u)] + 1;
                    queue.offer(v);
                }
            }
        }
    }

    private int dfs(String u, int flow, String sink) {
        if (u.equals(sink)) {
            return flow;
        }

        for (Map.Entry<String, Integer> entry : network.get(u).entrySet()) {
            String v = entry.getKey();
            int capacity = entry.getValue();
            if (capacity > 0 && level[nodes.indexOf(u)] < level[nodes.indexOf(v)]) {
                int sent = dfs(v, Math.min(flow, capacity), sink);
                if (sent > 0) {
                    network.get(u).put(v, capacity - sent);
                    network.get(v).put(u, network.get(v).getOrDefault(u, 0) + sent);
                    return sent;
                }
            }
        }
        return 0;
    }

    private void addEdge(String from, String to, int capacity) {
        network.computeIfAbsent(from, k -> new HashMap<>()).put(to, capacity);
        network.computeIfAbsent(to, k -> new HashMap<>()).put(from, 0);
        if (!nodes.contains(from)) nodes.add(from);
        if (!nodes.contains(to)) nodes.add(to);
    }

    private Set<String> reach(Map<T, Set<T>> graph, T t, Set<String> visited) {
        Queue<T> queue = new LinkedList<>();
        queue.add(t);

        while (!queue.isEmpty()) {
            T current = queue.poll();
            String currentKey = "A" + current.toString();
            visited.add(currentKey);
            for (T neighbor : graph.get(current)) {
                String neighborKey = "B" + neighbor.toString();
                if (!visited.contains(neighborKey)) {
                    queue.add(neighbor);
                    visited.add(neighborKey);
                }
            }
        }

        return visited;
    }

    public int solve(Map<T, Set<T>> graph) {
        for (T t : graph.keySet()) {
            addEdge("src", "A" + t.toString(), 1);
            addEdge("B" + t, "sink", 1);
            Set<String> visitedSubset = new HashSet<>();
            for (String u : reach(graph, t, visitedSubset)) {
                addEdge("A" + t, u, 1);
            }
        }

        int maxFlow = 0;
        while (true) {
            bfs("src");
            if (level[nodes.indexOf("sink")] == -1) break;

            int flow;
            while ((flow = dfs("src", Integer.MAX_VALUE, "sink")) > 0) {
                maxFlow += flow;
            }
        }

        return graph.size() - maxFlow;
    }
}