IRDAG implements DAGInterface<IRVertex, IREdge> { public void insert(final RelayVertex relayVertex, final IREdge edgeToStreamize) { assertNonExistence(relayVertex); assertNonControlEdge(edgeToStreamize); final DAGBuilder<IRVertex, IREdge> builder = new DAGBuilder<>(); if (edgeToStreamize.getPropertyValue(MessageIdEdgeProperty.class).isPresent() && !edgeToStreamize.getPropertyValue(MessageIdEdgeProperty.class).get().isEmpty()) { throw new CompileTimeOptimizationException(edgeToStreamize.getId() + " has a MessageId, and cannot be removed"); } if (edgeToStreamize.getDst() instanceof TaskSizeSplitterVertex) { return; } final IRVertex vertexToInsert = wrapSamplingVertexIfNeeded(relayVertex, edgeToStreamize.getSrc()); builder.addVertex(vertexToInsert); edgeToStreamize.getSrc().getPropertyValue(ParallelismProperty.class) .ifPresent(p -> vertexToInsert.setProperty(ParallelismProperty.of(p))); modifiedDAG.topologicalDo(v -> { builder.addVertex(v); for (final IREdge edge : modifiedDAG.getIncomingEdgesOf(v)) { if (edge.equals(edgeToStreamize)) { final IREdge toSV = new IREdge( edgeToStreamize.getPropertyValue(CommunicationPatternProperty.class).get(), edgeToStreamize.getSrc(), vertexToInsert); edgeToStreamize.copyExecutionPropertiesTo(toSV); final IREdge fromSV = new IREdge(CommunicationPatternProperty.Value.ONE_TO_ONE, vertexToInsert, v); fromSV.setProperty(EncoderProperty.of(edgeToStreamize.getPropertyValue(EncoderProperty.class).get())); fromSV.setProperty(DecoderProperty.of(edgeToStreamize.getPropertyValue(DecoderProperty.class).get())); toSV.setPropertyPermanently(DecoderProperty.of(BytesDecoderFactory.of())); toSV.setPropertyPermanently(CompressionProperty.of(CompressionProperty.Value.LZ4)); toSV.setPropertyPermanently(DecompressionProperty.of(CompressionProperty.Value.NONE)); fromSV.setPropertyPermanently(EncoderProperty.of(BytesEncoderFactory.of())); fromSV.setPropertyPermanently(CompressionProperty.of(CompressionProperty.Value.NONE)); fromSV.setPropertyPermanently(DecompressionProperty.of(CompressionProperty.Value.LZ4)); fromSV.setPropertyPermanently(PartitionerProperty.of(PartitionerProperty.Type.DEDICATED_KEY_PER_ELEMENT)); builder.connectVertices(toSV); builder.connectVertices(fromSV); } else { builder.connectVertices(edge); } } }); if (edgeToStreamize.getSrc() instanceof RelayVertex) { streamVertexToOriginalEdge.put(relayVertex, streamVertexToOriginalEdge.get(edgeToStreamize.getSrc())); } else if (edgeToStreamize.getDst() instanceof RelayVertex) { streamVertexToOriginalEdge.put(relayVertex, streamVertexToOriginalEdge.get(edgeToStreamize.getDst())); } else { streamVertexToOriginalEdge.put(relayVertex, edgeToStreamize); } modifiedDAG = builder.build(); } IRDAG(final DAG<IRVertex, IREdge> originalUserApplicationDAG); IRDAGChecker.CheckerResult checkIntegrity(); boolean advanceDAGSnapshot(final BiFunction<IRDAG, IRDAG, Boolean> checker); String irDAGSummary(); Long getInputSize(); void recordExecutorInfo(final List<Pair<Integer, ResourceSpecification>> parsedExecutorInfo); List<Pair<Integer, ResourceSpecification>> getExecutorInfo(); void delete(final IRVertex vertexToDelete); void insert(final RelayVertex relayVertex, final IREdge edgeToStreamize); void insert(final MessageGeneratorVertex messageGeneratorVertex,
                     final MessageAggregatorVertex messageAggregatorVertex,
                     final EncoderProperty triggerOutputEncoder,
                     final DecoderProperty triggerOutputDecoder,
                     final Set<IREdge> edgesToGetStatisticsOf,
                     final Set<IREdge> edgesToOptimize); void insert(final SignalVertex toInsert,
                     final IREdge edgeToOptimize); void insert(final Set<SamplingVertex> toInsert,
                     final Set<IRVertex> executeAfter); void insert(final TaskSizeSplitterVertex toInsert); void reshapeUnsafely(final Function<DAG<IRVertex, IREdge>, DAG<IRVertex, IREdge>> unsafeReshapingFunction); @Override void topologicalDo(final Consumer<IRVertex> function); @Override void dfsTraverse(final Consumer<IRVertex> function, final TraversalOrder traversalOrder); @Override void dfsDo(final IRVertex vertex,
                    final Consumer<IRVertex> vertexConsumer,
                    final TraversalOrder traversalOrder,
                    final Set<IRVertex> visited); @Override Boolean pathExistsBetween(final IRVertex v1, final IRVertex v2); @Override Boolean isCompositeVertex(final IRVertex irVertex); @Override Integer getLoopStackDepthOf(final IRVertex irVertex); @Override LoopVertex getAssignedLoopVertexOf(final IRVertex irVertex); @Override ObjectNode asJsonNode(); @Override void storeJSON(final String directory, final String name, final String description); @Override IRVertex getVertexById(final String id); @Override IREdge getEdgeById(final String id); @Override List<IRVertex> getVertices(); @Override List<IREdge> getEdges(); @Override List<IRVertex> getRootVertices(); @Override List<IREdge> getIncomingEdgesOf(final IRVertex v); @Override List<IREdge> getIncomingEdgesOf(final String vertexId); @Override List<IREdge> getOutgoingEdgesOf(final IRVertex v); @Override List<IREdge> getOutgoingEdgesOf(final String vertexId); @Override List<IRVertex> getParents(final String vertexId); @Override List<IRVertex> getChildren(final String vertexId); @Override IREdge getEdgeBetween(final String srcVertexId,
                               final String dstVertexId); @Override List<IRVertex> getTopologicalSort(); @Override List<IRVertex> getAncestors(final String vertexId); @Override List<IRVertex> getDescendants(final String vertexId); @Override List<IRVertex> filterVertices(final Predicate<IRVertex> condition); @Override String toString();  }
@Test public void testThousandRandomConfigurations() { final int thousandConfigs = 1000; for (int i = 0; i < thousandConfigs; i++) { final int numOfTotalMethods = 13; final int methodIndex = random.nextInt(numOfTotalMethods); switch (methodIndex) { case 0: selectRandomVertex().setProperty(randomCSP()); break; case 1: selectRandomVertex().setProperty(randomRLP()); break; case 2: selectRandomVertex().setProperty(randomRPP()); break; case 3: selectRandomVertex().setProperty(randomRSP()); break; case 4: selectRandomEdge().setProperty(randomDFP()); break; case 5: selectRandomEdge().setProperty(randomDPP()); break; case 6: selectRandomEdge().setProperty(randomDSP()); break; case 7: final RelayVertex relayVertex = new RelayVertex(); final IREdge edgeToStreamize = selectRandomEdge(); if (!(edgeToStreamize.getPropertyValue(MessageIdEdgeProperty.class).isPresent() && !edgeToStreamize.getPropertyValue(MessageIdEdgeProperty.class).get().isEmpty())) { irdag.insert(relayVertex, edgeToStreamize); } break; case 8: insertNewTriggerVertex(irdag, selectRandomEdge()); break; case 9: final IRVertex vertexToSample = selectRandomNonUtilityVertex(); final SamplingVertex samplingVertex = new SamplingVertex(vertexToSample, 0.1f); irdag.insert(Sets.newHashSet(samplingVertex), Sets.newHashSet(vertexToSample)); break; case 10: insertNewSignalVertex(irdag, selectRandomEdge()); break; case 11: insertNewSplitterVertex(irdag, selectRandomEdge()); break; case 12: selectRandomUtilityVertex().ifPresent(irdag::delete); break; default: throw new IllegalStateException(String.valueOf(methodIndex)); } if (i % (thousandConfigs / 10) == 0) { } if (methodIndex >= 7) { } mustPass(); } }