sanha commented on a change in pull request #190: [NEMO-328] Refactor IRDAG
URL: https://github.com/apache/incubator-nemo/pull/190#discussion_r252558747
##########
File path:
compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SkewReshapingPass.java
##########
@@ -66,186 +59,66 @@ public SkewReshapingPass() {
}
@Override
- public DAG<IRVertex, IREdge> apply(final DAG<IRVertex, IREdge> dag) {
- int mcCount = 0;
- // destination vertex ID to metric aggregation vertex - ID pair map
- final Map<String, Pair<OperatorVertex, Integer>> dstVtxIdToABV = new
HashMap<>();
- final DAGBuilder<IRVertex, IREdge> builder = new DAGBuilder<>();
-
- for (final IRVertex v : dag.getTopologicalSort()) {
- // We care about OperatorVertices that have shuffle incoming edges with
main output.
- // TODO #210: Data-aware dynamic optimization at run-time
- if (v instanceof OperatorVertex &&
dag.getIncomingEdgesOf(v).stream().anyMatch(irEdge ->
- CommunicationPatternProperty.Value.Shuffle
-
.equals(irEdge.getPropertyValue(CommunicationPatternProperty.class).get()))
- && dag.getIncomingEdgesOf(v).stream().noneMatch(irEdge ->
- irEdge.getPropertyValue(AdditionalOutputTagProperty.class).isPresent()))
{
-
- for (final IREdge edge : dag.getIncomingEdgesOf(v)) {
- if (CommunicationPatternProperty.Value.Shuffle
-
.equals(edge.getPropertyValue(CommunicationPatternProperty.class).get())) {
- final String dstId = edge.getDst().getId();
-
- // Get or generate a metric collection vertex.
- final int metricCollectionId;
- final OperatorVertex abv;
- if (!dstVtxIdToABV.containsKey(dstId)) {
- // There is no metric aggregation vertex for this destination
vertex.
- metricCollectionId = mcCount++;
- abv = generateMetricAggregationVertex();
- builder.addVertex(abv);
-
- abv.setPropertyPermanently(ResourceSlotProperty.of(false));
- dstVtxIdToABV.put(dstId, Pair.of(abv, metricCollectionId));
- } else {
- // There is a metric aggregation vertex for this destination
vertex already.
- final Pair<OperatorVertex, Integer> aggrPair =
dstVtxIdToABV.get(dstId);
- metricCollectionId = aggrPair.right();
- abv = aggrPair.left();
- }
-
- final OperatorVertex mcv = generateMetricCollectVertex(edge);
- builder.addVertex(v);
- builder.addVertex(mcv);
-
- // We then insert the vertex with MetricCollectTransform and
vertex with AggregateMetricTransform
- // between the vertex and incoming vertices.
- final IREdge edgeToMCV = generateEdgeToMCV(edge, mcv);
- final IREdge edgeToABV = generateEdgeToABV(edge, mcv, abv);
- final IREdge edgeToOriginalDstV =
- new
IREdge(edge.getPropertyValue(CommunicationPatternProperty.class).get(),
edge.getSrc(), v);
- edge.copyExecutionPropertiesTo(edgeToOriginalDstV);
-
edgeToOriginalDstV.setPropertyPermanently(MetricCollectionProperty.of(metricCollectionId));
Review comment:
Don't we need to annotate this `MetricCollectionProperty` elsewhere?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
