@Test void testSort_graphIsNotConnected() { ExecutionNode node1 = createExecutionNode("node-1", ImmutableList.of("node-2")); VerifyException e = assertThrows( VerifyException.class, () -> ExecutionNodeCompiler.sort(ImmutableList.of(node1))); Assert.assertEquals("workflow graph isn't connected or has a cycle", e.getMessage()); }
static List<ExecutionNode> sort(List<ExecutionNode> nodes) { Map<String, Integer> priorityMap = new HashMap<>(); Map<String, Integer> degreeMap = new HashMap<>(); Map<String, ExecutionNode> lookup = new HashMap<>(); ListMultimap<String, String> downstreamNodeIdsMap = ArrayListMultimap.create(); for (int i = 0; i < nodes.size(); i++) { ExecutionNode node = nodes.get(i); priorityMap.put(node.nodeId(), i); degreeMap.put(node.nodeId(), node.upstreamNodeIds().size()); for (String upstreamNodeId : node.upstreamNodeIds()) { downstreamNodeIdsMap.put(upstreamNodeId, node.nodeId()); } ExecutionNode previous = lookup.put(node.nodeId(), node); Verify.verify(previous == null, "duplicate node id [%s]", node.nodeId()); } Deque<List<String>> deque = new ArrayDeque<>(); Set<String> visitedNodeIds = new HashSet<>(); List<ExecutionNode> topologicallySorted = new ArrayList<>(); deque.add(singletonList(START_NODE_ID)); while (!deque.isEmpty()) { List<String> nodeIds = deque.pollFirst(); List<String> downstreamNodeIds = new ArrayList<>(); for (String nodeId : nodeIds) { if (!nodeId.equals(START_NODE_ID)) { ExecutionNode node = lookup.get(nodeId); Verify.verifyNotNull(node, "node not found [%s]", nodeId); topologicallySorted.add(node); } boolean visited = visitedNodeIds.contains(nodeId); Verify.verify(!visited, "invariant failed"); for (String downstreamNodeId : downstreamNodeIdsMap.get(nodeId)) { int newDegree = degreeMap.get(downstreamNodeId) - 1; if (newDegree == 0) { downstreamNodeIds.add(downstreamNodeId); } degreeMap.put(downstreamNodeId, newDegree); } } visitedNodeIds.addAll(nodeIds); if (!downstreamNodeIds.isEmpty()) { List<String> sortedDownstreamNodeIds = downstreamNodeIds.stream() .sorted(Comparator.comparing(priorityMap::get)) .distinct() .collect(Collectors.toList()); deque.push(sortedDownstreamNodeIds); } } Verify.verify( nodes.size() == topologicallySorted.size(), "workflow graph isn't connected or has a cycle"); return topologicallySorted; }
ExecutionNodeCompiler { static List<ExecutionNode> sort(List<ExecutionNode> nodes) { Map<String, Integer> priorityMap = new HashMap<>(); Map<String, Integer> degreeMap = new HashMap<>(); Map<String, ExecutionNode> lookup = new HashMap<>(); ListMultimap<String, String> downstreamNodeIdsMap = ArrayListMultimap.create(); for (int i = 0; i < nodes.size(); i++) { ExecutionNode node = nodes.get(i); priorityMap.put(node.nodeId(), i); degreeMap.put(node.nodeId(), node.upstreamNodeIds().size()); for (String upstreamNodeId : node.upstreamNodeIds()) { downstreamNodeIdsMap.put(upstreamNodeId, node.nodeId()); } ExecutionNode previous = lookup.put(node.nodeId(), node); Verify.verify(previous == null, "duplicate node id [%s]", node.nodeId()); } Deque<List<String>> deque = new ArrayDeque<>(); Set<String> visitedNodeIds = new HashSet<>(); List<ExecutionNode> topologicallySorted = new ArrayList<>(); deque.add(singletonList(START_NODE_ID)); while (!deque.isEmpty()) { List<String> nodeIds = deque.pollFirst(); List<String> downstreamNodeIds = new ArrayList<>(); for (String nodeId : nodeIds) { if (!nodeId.equals(START_NODE_ID)) { ExecutionNode node = lookup.get(nodeId); Verify.verifyNotNull(node, "node not found [%s]", nodeId); topologicallySorted.add(node); } boolean visited = visitedNodeIds.contains(nodeId); Verify.verify(!visited, "invariant failed"); for (String downstreamNodeId : downstreamNodeIdsMap.get(nodeId)) { int newDegree = degreeMap.get(downstreamNodeId) - 1; if (newDegree == 0) { downstreamNodeIds.add(downstreamNodeId); } degreeMap.put(downstreamNodeId, newDegree); } } visitedNodeIds.addAll(nodeIds); if (!downstreamNodeIds.isEmpty()) { List<String> sortedDownstreamNodeIds = downstreamNodeIds.stream() .sorted(Comparator.comparing(priorityMap::get)) .distinct() .collect(Collectors.toList()); deque.push(sortedDownstreamNodeIds); } } Verify.verify( nodes.size() == topologicallySorted.size(), "workflow graph isn't connected or has a cycle"); return topologicallySorted; } }
ExecutionNodeCompiler { static List<ExecutionNode> sort(List<ExecutionNode> nodes) { Map<String, Integer> priorityMap = new HashMap<>(); Map<String, Integer> degreeMap = new HashMap<>(); Map<String, ExecutionNode> lookup = new HashMap<>(); ListMultimap<String, String> downstreamNodeIdsMap = ArrayListMultimap.create(); for (int i = 0; i < nodes.size(); i++) { ExecutionNode node = nodes.get(i); priorityMap.put(node.nodeId(), i); degreeMap.put(node.nodeId(), node.upstreamNodeIds().size()); for (String upstreamNodeId : node.upstreamNodeIds()) { downstreamNodeIdsMap.put(upstreamNodeId, node.nodeId()); } ExecutionNode previous = lookup.put(node.nodeId(), node); Verify.verify(previous == null, "duplicate node id [%s]", node.nodeId()); } Deque<List<String>> deque = new ArrayDeque<>(); Set<String> visitedNodeIds = new HashSet<>(); List<ExecutionNode> topologicallySorted = new ArrayList<>(); deque.add(singletonList(START_NODE_ID)); while (!deque.isEmpty()) { List<String> nodeIds = deque.pollFirst(); List<String> downstreamNodeIds = new ArrayList<>(); for (String nodeId : nodeIds) { if (!nodeId.equals(START_NODE_ID)) { ExecutionNode node = lookup.get(nodeId); Verify.verifyNotNull(node, "node not found [%s]", nodeId); topologicallySorted.add(node); } boolean visited = visitedNodeIds.contains(nodeId); Verify.verify(!visited, "invariant failed"); for (String downstreamNodeId : downstreamNodeIdsMap.get(nodeId)) { int newDegree = degreeMap.get(downstreamNodeId) - 1; if (newDegree == 0) { downstreamNodeIds.add(downstreamNodeId); } degreeMap.put(downstreamNodeId, newDegree); } } visitedNodeIds.addAll(nodeIds); if (!downstreamNodeIds.isEmpty()) { List<String> sortedDownstreamNodeIds = downstreamNodeIds.stream() .sorted(Comparator.comparing(priorityMap::get)) .distinct() .collect(Collectors.toList()); deque.push(sortedDownstreamNodeIds); } } Verify.verify( nodes.size() == topologicallySorted.size(), "workflow graph isn't connected or has a cycle"); return topologicallySorted; }  }
ExecutionNodeCompiler { static List<ExecutionNode> sort(List<ExecutionNode> nodes) { Map<String, Integer> priorityMap = new HashMap<>(); Map<String, Integer> degreeMap = new HashMap<>(); Map<String, ExecutionNode> lookup = new HashMap<>(); ListMultimap<String, String> downstreamNodeIdsMap = ArrayListMultimap.create(); for (int i = 0; i < nodes.size(); i++) { ExecutionNode node = nodes.get(i); priorityMap.put(node.nodeId(), i); degreeMap.put(node.nodeId(), node.upstreamNodeIds().size()); for (String upstreamNodeId : node.upstreamNodeIds()) { downstreamNodeIdsMap.put(upstreamNodeId, node.nodeId()); } ExecutionNode previous = lookup.put(node.nodeId(), node); Verify.verify(previous == null, "duplicate node id [%s]", node.nodeId()); } Deque<List<String>> deque = new ArrayDeque<>(); Set<String> visitedNodeIds = new HashSet<>(); List<ExecutionNode> topologicallySorted = new ArrayList<>(); deque.add(singletonList(START_NODE_ID)); while (!deque.isEmpty()) { List<String> nodeIds = deque.pollFirst(); List<String> downstreamNodeIds = new ArrayList<>(); for (String nodeId : nodeIds) { if (!nodeId.equals(START_NODE_ID)) { ExecutionNode node = lookup.get(nodeId); Verify.verifyNotNull(node, "node not found [%s]", nodeId); topologicallySorted.add(node); } boolean visited = visitedNodeIds.contains(nodeId); Verify.verify(!visited, "invariant failed"); for (String downstreamNodeId : downstreamNodeIdsMap.get(nodeId)) { int newDegree = degreeMap.get(downstreamNodeId) - 1; if (newDegree == 0) { downstreamNodeIds.add(downstreamNodeId); } degreeMap.put(downstreamNodeId, newDegree); } } visitedNodeIds.addAll(nodeIds); if (!downstreamNodeIds.isEmpty()) { List<String> sortedDownstreamNodeIds = downstreamNodeIds.stream() .sorted(Comparator.comparing(priorityMap::get)) .distinct() .collect(Collectors.toList()); deque.push(sortedDownstreamNodeIds); } } Verify.verify( nodes.size() == topologicallySorted.size(), "workflow graph isn't connected or has a cycle"); return topologicallySorted; }   }
ExecutionNodeCompiler { static List<ExecutionNode> sort(List<ExecutionNode> nodes) { Map<String, Integer> priorityMap = new HashMap<>(); Map<String, Integer> degreeMap = new HashMap<>(); Map<String, ExecutionNode> lookup = new HashMap<>(); ListMultimap<String, String> downstreamNodeIdsMap = ArrayListMultimap.create(); for (int i = 0; i < nodes.size(); i++) { ExecutionNode node = nodes.get(i); priorityMap.put(node.nodeId(), i); degreeMap.put(node.nodeId(), node.upstreamNodeIds().size()); for (String upstreamNodeId : node.upstreamNodeIds()) { downstreamNodeIdsMap.put(upstreamNodeId, node.nodeId()); } ExecutionNode previous = lookup.put(node.nodeId(), node); Verify.verify(previous == null, "duplicate node id [%s]", node.nodeId()); } Deque<List<String>> deque = new ArrayDeque<>(); Set<String> visitedNodeIds = new HashSet<>(); List<ExecutionNode> topologicallySorted = new ArrayList<>(); deque.add(singletonList(START_NODE_ID)); while (!deque.isEmpty()) { List<String> nodeIds = deque.pollFirst(); List<String> downstreamNodeIds = new ArrayList<>(); for (String nodeId : nodeIds) { if (!nodeId.equals(START_NODE_ID)) { ExecutionNode node = lookup.get(nodeId); Verify.verifyNotNull(node, "node not found [%s]", nodeId); topologicallySorted.add(node); } boolean visited = visitedNodeIds.contains(nodeId); Verify.verify(!visited, "invariant failed"); for (String downstreamNodeId : downstreamNodeIdsMap.get(nodeId)) { int newDegree = degreeMap.get(downstreamNodeId) - 1; if (newDegree == 0) { downstreamNodeIds.add(downstreamNodeId); } degreeMap.put(downstreamNodeId, newDegree); } } visitedNodeIds.addAll(nodeIds); if (!downstreamNodeIds.isEmpty()) { List<String> sortedDownstreamNodeIds = downstreamNodeIds.stream() .sorted(Comparator.comparing(priorityMap::get)) .distinct() .collect(Collectors.toList()); deque.push(sortedDownstreamNodeIds); } } Verify.verify( nodes.size() == topologicallySorted.size(), "workflow graph isn't connected or has a cycle"); return topologicallySorted; }    }