CloneAmbiguousFlattensFunction implements Function<MutableNetwork<Node, Edge>, MutableNetwork<Node, Edge>> { @Override public MutableNetwork<Node, Edge> apply(MutableNetwork<Node, Edge> network) { Set<Node> sortedNodesSet = Networks.topologicalOrder(network); Node[] sortedNodes = sortedNodesSet.toArray(new Node[sortedNodesSet.size()]); for (int i = sortedNodes.length - 1; i >= 0; i--) { Node node = sortedNodes[i]; if (node instanceof ParallelInstructionNode && ((ParallelInstructionNode) node).getParallelInstruction().getFlatten() != null && ((ParallelInstructionNode) node).getExecutionLocation() == ExecutionLocation.AMBIGUOUS) { cloneFlatten(node, network); } } return network; }  @Override MutableNetwork<Node, Edge> apply(MutableNetwork<Node, Edge> network);  }
@Test public void testEmptyNetwork() throws Exception { assertEquals( createEmptyNetwork(), new CloneAmbiguousFlattensFunction().apply(createEmptyNetwork())); }
@Test public void testSingleFlatten() throws Exception { MutableNetwork<Node, Edge> network = createEmptyNetwork(); Node sdkPredecessor = createSdkNode("sdk_predecessor"); Node runnerPredecessor = createRunnerNode("runner_predecessor"); Node sdkPredecessorOutput = createPCollection("sdk_predecessor.out"); Node runnerPredecessorOutput = createPCollection("runner_predecessor.out"); Node ambiguousFlatten = createFlatten("ambiguous_flatten", ExecutionLocation.AMBIGUOUS); Node ambiguousFlattenOutput = createPCollection("ambiguous_flatten.out"); Node sdkSuccessor = createSdkNode("sdk_successor"); Node runnerSuccessor = createRunnerNode("runner_successor"); Node noLocationSuccessor = createNoLocationNode(); Node sdkSuccessorOutput = createPCollection("sdk_successor.out"); Node runnerSuccessorOutput = createPCollection("runner_successor.out"); Node noLocationSuccessorOutput = createPCollection("no_location_successor.out"); network.addNode(sdkPredecessor); network.addNode(runnerPredecessor); network.addNode(sdkPredecessorOutput); network.addNode(runnerPredecessorOutput); network.addNode(ambiguousFlatten); network.addNode(ambiguousFlattenOutput); network.addNode(sdkSuccessor); network.addNode(runnerSuccessor); network.addNode(noLocationSuccessor); network.addNode(sdkSuccessorOutput); network.addNode(runnerSuccessorOutput); network.addNode(noLocationSuccessorOutput); network.addEdge(sdkPredecessor, sdkPredecessorOutput, DefaultEdge.create()); network.addEdge(runnerPredecessor, runnerPredecessorOutput, DefaultEdge.create()); network.addEdge(sdkPredecessorOutput, ambiguousFlatten, DefaultEdge.create()); network.addEdge(runnerPredecessorOutput, ambiguousFlatten, DefaultEdge.create()); network.addEdge(ambiguousFlatten, ambiguousFlattenOutput, DefaultEdge.create()); network.addEdge(ambiguousFlattenOutput, sdkSuccessor, DefaultEdge.create()); network.addEdge(ambiguousFlattenOutput, runnerSuccessor, DefaultEdge.create()); network.addEdge(ambiguousFlattenOutput, noLocationSuccessor, DefaultEdge.create()); network.addEdge(sdkSuccessor, sdkSuccessorOutput, DefaultEdge.create()); network.addEdge(runnerSuccessor, runnerSuccessorOutput, DefaultEdge.create()); network.addEdge(noLocationSuccessor, noLocationSuccessorOutput, DefaultEdge.create()); List<List<Node>> originalPaths = Networks.allPathsFromRootsToLeaves(network); network = new CloneAmbiguousFlattensFunction().apply(network); ParallelInstructionNode sdkFlatten = null; ParallelInstructionNode runnerFlatten = null; for (Node node : network.nodes()) { if (node instanceof ParallelInstructionNode && ((ParallelInstructionNode) node).getParallelInstruction().getFlatten() != null) { ParallelInstructionNode castNode = ((ParallelInstructionNode) node); if (castNode.getExecutionLocation() == ExecutionLocation.SDK_HARNESS) { sdkFlatten = castNode; } else if (castNode.getExecutionLocation() == ExecutionLocation.RUNNER_HARNESS) { runnerFlatten = castNode; } else { assertTrue( "Ambiguous flatten not removed from network.", castNode.getExecutionLocation() != ExecutionLocation.AMBIGUOUS); } } } assertNotNull("Ambiguous flatten was not cloned into sdk flatten.", sdkFlatten); assertNotNull("Ambiguous flatten was not cloned into runner flatten.", runnerFlatten); Node sdkFlattenOutput = Iterables.getOnlyElement(network.successors(sdkFlatten)); Node runnerFlattenOutput = Iterables.getOnlyElement(network.successors(runnerFlatten)); assertEquals(2, network.predecessors(sdkFlatten).size()); assertEquals(2, network.predecessors(runnerFlatten).size()); assertEquals(1, network.successors(sdkFlattenOutput).size()); assertEquals(2, network.successors(runnerFlattenOutput).size()); assertSame(sdkSuccessor, Iterables.getOnlyElement(network.successors(sdkFlattenOutput))); assertThat( network.successors(runnerFlattenOutput), hasItems(runnerSuccessor, noLocationSuccessor)); assertEquals(originalPaths.size(), Networks.allPathsFromRootsToLeaves(network).size()); }
@Test public void testNonAmbiguousFlattens() throws Exception { MutableNetwork<Node, Edge> network = createEmptyNetwork(); Node sdk1 = createSdkNode("sdk1"); Node sdk2 = createSdkNode("sdk2"); Node sdk3 = createSdkNode("sdk3"); Node sdk1Out = createPCollection("sdk1.out"); Node sdk2Out = createPCollection("sdk2.out"); Node sdk3Out = createPCollection("sdk3.out"); Node runner1 = createRunnerNode("runner1"); Node runner2 = createRunnerNode("runner2"); Node runner3 = createRunnerNode("runner3"); Node runner4 = createRunnerNode("runner4"); Node runner1Out = createPCollection("runner1.out"); Node runner2Out = createPCollection("runner2.out"); Node runner3Out = createPCollection("runner3.out"); Node runner4Out = createPCollection("runner4.out"); Node ambiguousFlatten = createFlatten("ambiguous_flatten", ExecutionLocation.AMBIGUOUS); Node ambiguousFlattenOut = createPCollection("ambiguous_flatten.out"); Node sdkFlatten = createFlatten("sdk_flatten", ExecutionLocation.SDK_HARNESS); Node sdkFlattenOut = createPCollection("sdk_flatten.out"); Node runnerFlatten = createFlatten("runner_flatten", ExecutionLocation.RUNNER_HARNESS); Node runnerFlattenOut = createPCollection("runner_flatten.out"); network.addNode(sdk1); network.addNode(sdk2); network.addNode(sdk3); network.addNode(sdk1Out); network.addNode(sdk2Out); network.addNode(sdk3Out); network.addNode(runner1); network.addNode(runner2); network.addNode(runner3); network.addNode(runner4); network.addNode(runner1Out); network.addNode(runner2Out); network.addNode(runner3Out); network.addNode(runner4Out); network.addNode(ambiguousFlatten); network.addNode(ambiguousFlattenOut); network.addNode(sdkFlatten); network.addNode(sdkFlattenOut); network.addNode(runnerFlatten); network.addNode(runnerFlattenOut); network.addEdge(sdk1, sdk1Out, DefaultEdge.create()); network.addEdge(sdk2, sdk2Out, DefaultEdge.create()); network.addEdge(sdk3, sdk3Out, DefaultEdge.create()); network.addEdge(runner1, runner1Out, DefaultEdge.create()); network.addEdge(runner2, runner2Out, DefaultEdge.create()); network.addEdge(runner3, runner3Out, DefaultEdge.create()); network.addEdge(runner4, runner4Out, DefaultEdge.create()); network.addEdge(ambiguousFlatten, ambiguousFlattenOut, DefaultEdge.create()); network.addEdge(sdkFlatten, sdkFlattenOut, DefaultEdge.create()); network.addEdge(runnerFlatten, runnerFlattenOut, DefaultEdge.create()); network.addEdge(sdk1Out, ambiguousFlatten, DefaultEdge.create()); network.addEdge(runner1Out, ambiguousFlatten, DefaultEdge.create()); network.addEdge(ambiguousFlattenOut, sdkFlatten, DefaultEdge.create()); network.addEdge(sdk2Out, sdkFlatten, DefaultEdge.create()); network.addEdge(sdkFlattenOut, sdk3, DefaultEdge.create()); network.addEdge(ambiguousFlattenOut, runner2, DefaultEdge.create()); network.addEdge(runner2Out, runnerFlatten, DefaultEdge.create()); network.addEdge(runner3Out, runnerFlatten, DefaultEdge.create()); network.addEdge(runnerFlattenOut, runner4, DefaultEdge.create()); List<List<Node>> originalPaths = Networks.allPathsFromRootsToLeaves(network); network = new CloneAmbiguousFlattensFunction().apply(network); for (Node node : network.nodes()) { if (node instanceof ParallelInstructionNode && ((ParallelInstructionNode) node).getParallelInstruction().getFlatten() != null) { ParallelInstructionNode castNode = ((ParallelInstructionNode) node); assertTrue( "Ambiguous flatten not removed from network.", castNode.getExecutionLocation() != ExecutionLocation.AMBIGUOUS); if ("sdk_flatten".equals(castNode.getParallelInstruction().getName())) { assertSame("SDK flatten has been incorrectly modified.", sdkFlatten, castNode); } else if ("runner_flatten".equals(castNode.getParallelInstruction().getName())) { assertSame("Runner flatten has been incorrectly modified.", runnerFlatten, castNode); } } } assertEquals(originalPaths.size(), Networks.allPathsFromRootsToLeaves(network).size()); }
@Test public void testConnectedFlattens() throws Exception { MutableNetwork<Node, Edge> network = createEmptyNetwork(); Node sdk1 = createSdkNode("sdk1"); Node sdk2 = createSdkNode("sdk2"); Node sdk3 = createSdkNode("sdk3"); Node sdk4 = createSdkNode("sdk4"); Node sdk1Out = createPCollection("sdk1.out"); Node sdk2Out = createPCollection("sdk2.out"); Node sdk3Out = createPCollection("sdk3.out"); Node sdk4Out = createPCollection("sdk4.out"); Node runner1 = createRunnerNode("runner1"); Node runner2 = createRunnerNode("runner2"); Node runner3 = createRunnerNode("runner3"); Node runner4 = createRunnerNode("runner4"); Node runner1Out = createPCollection("runner1.out"); Node runner2Out = createPCollection("runner2.out"); Node runner3Out = createPCollection("runner3.out"); Node runner4Out = createPCollection("runner4.out"); Node ambiguousFlatten1 = createFlatten("ambiguous_flatten1", ExecutionLocation.AMBIGUOUS); Node ambiguousFlatten2 = createFlatten("ambiguous_flatten2", ExecutionLocation.AMBIGUOUS); Node ambiguousFlatten3 = createFlatten("ambiguous_flatten3", ExecutionLocation.AMBIGUOUS); Node ambiguousFlatten1Out = createPCollection("ambiguous_flatten1.out"); Node ambiguousFlatten2Out = createPCollection("ambiguous_flatten2.out"); Node ambiguousFlatten3Out = createPCollection("ambiguous_flatten3.out"); network.addNode(sdk1); network.addNode(sdk2); network.addNode(sdk3); network.addNode(sdk4); network.addNode(sdk1Out); network.addNode(sdk2Out); network.addNode(sdk3Out); network.addNode(sdk4Out); network.addNode(runner1); network.addNode(runner2); network.addNode(runner3); network.addNode(runner4); network.addNode(runner1Out); network.addNode(runner2Out); network.addNode(runner3Out); network.addNode(runner4Out); network.addNode(ambiguousFlatten1); network.addNode(ambiguousFlatten2); network.addNode(ambiguousFlatten3); network.addNode(ambiguousFlatten1Out); network.addNode(ambiguousFlatten2Out); network.addNode(ambiguousFlatten3Out); network.addEdge(sdk1, sdk1Out, DefaultEdge.create()); network.addEdge(sdk2, sdk2Out, DefaultEdge.create()); network.addEdge(sdk3, sdk3Out, DefaultEdge.create()); network.addEdge(sdk4, sdk4Out, DefaultEdge.create()); network.addEdge(runner1, runner1Out, DefaultEdge.create()); network.addEdge(runner2, runner2Out, DefaultEdge.create()); network.addEdge(runner3, runner3Out, DefaultEdge.create()); network.addEdge(runner4, runner4Out, DefaultEdge.create()); network.addEdge(ambiguousFlatten1, ambiguousFlatten1Out, DefaultEdge.create()); network.addEdge(ambiguousFlatten2, ambiguousFlatten2Out, DefaultEdge.create()); network.addEdge(ambiguousFlatten3, ambiguousFlatten3Out, DefaultEdge.create()); network.addEdge(sdk1Out, ambiguousFlatten1, DefaultEdge.create()); network.addEdge(runner1Out, ambiguousFlatten1, DefaultEdge.create()); network.addEdge(sdk2Out, ambiguousFlatten2, DefaultEdge.create()); network.addEdge(runner2Out, ambiguousFlatten2, DefaultEdge.create()); network.addEdge(ambiguousFlatten1Out, sdk3, DefaultEdge.create()); network.addEdge(ambiguousFlatten1Out, ambiguousFlatten3, DefaultEdge.create()); network.addEdge(ambiguousFlatten2Out, ambiguousFlatten3, DefaultEdge.create()); network.addEdge(ambiguousFlatten2Out, runner3, DefaultEdge.create()); network.addEdge(ambiguousFlatten3Out, sdk4, DefaultEdge.create()); network.addEdge(ambiguousFlatten3Out, runner4, DefaultEdge.create()); List<List<Node>> originalPaths = Networks.allPathsFromRootsToLeaves(network); network = new CloneAmbiguousFlattensFunction().apply(network); for (Node node : network.nodes()) { if (node instanceof ParallelInstructionNode && ((ParallelInstructionNode) node).getParallelInstruction().getFlatten() != null) { ParallelInstructionNode castNode = ((ParallelInstructionNode) node); assertTrue( "Ambiguous flatten not removed from network.", castNode.getExecutionLocation() != ExecutionLocation.AMBIGUOUS); } } assertEquals(originalPaths.size(), Networks.allPathsFromRootsToLeaves(network).size()); }