如何计算DAG节点最大并发数？算法优化与性能问题排查

问题背景 #

我有一个由有向无环图（DAG）构成的工作流，每个节点执行时间随机，需要计算该工作流的节点最大并发数。最初采用全遍历所有运行场景的方式求解，但节点/边数量较多时耗时严重，因此转向最大流算法。

参考Juan Lopes的指导，学习了Ford-Fulkerson法、Edmonds Karp算法、Dinic算法及Push label算法，并用Java实现了Ford-Fulkerson和Dinic算法（Push label算法暂未实现）。测试结果准确，但在顺序依赖场景下性能极差，同时发现节点名称长度会影响耗时。

测试结果 #

示例1：201个节点的线性依赖链（1 -> 2 -> 3 -> ... -> 201） #

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个节点的高并发依赖场景 #

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

代码实现 #

Ford-Fulkerson算法实现 #

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;
    }

}

Dinic算法实现 #

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;
    }
}

求助需求 #

请帮忙分析性能问题的原因，并给出具体的优化方案。

内容的提问来源于stack exchange，提问作者hehe